<commit_before>(ns ophion.protograph
  (:require
   [cheshire.core :as json]
   [taoensso.timbre :as log]
   [ophion.kafka :as kafka])
  (:import
   [protograph Protograph ProtographEmitter]))

(defn load
  [path]
  (Protograph/loadProtograph path))

(defn emitter
  [emit-vertex emit-edge]
  (reify ProtographEmitter
    (emitVertex [_ vertex] (emit-vertex vertex))
    (emitEdge [_ edge] (emit-edge edge))))

(defn process
  [protograph emit label data]
  (.processMessage protograph emit label data))

(defn kafka-emitter
  [producer vertex-topic edge-topic]
  (emitter
   (fn [vertex]
     (kafka/send producer vertex-topic (Protograph/writeJSON vertex)))
   (fn [edge]
     (kafka/send producer edge-topic (Protograph/writeJSON edge)))))

(defn transform-message
  [protograph emit message]
  (let [label (kafka/topic->label (.topic message))
        data (Protograph/readJSON (.value message))]
    (log/info label data)
    (process protograph emit label data)))

(defn transform-kafka
  [config protograph consumer producer]
  (let [vertex-topic (str (get-in config [:protograph :output]) ".Vertex")
        edge-topic (str (get-in config [:protograph :output]) ".Edge")
        emit (kafka-emitter producer vertex-topic edge-topic)]
    (kafka/consume
     consumer
     (partial transform-message protograph emit))))

(defn transform-topics
  [config protograph topics]
  (let [host (get-in config [:kafka :host])
        group-id (get-in config [:kafka :consumer :group-id])
        consumer (kafka/consumer host group-id topics)
        producer (kafka/producer host)]
    (transform-kafka config protograph consumer producer)))

(def default-config
  {:protograph
   {:path "../gaia-bmeg/bmeg.protograph.yml"
    :output "protograph.bmeg"}
   :kafka kafka/default-config})

(def bmeg-topics
  {:ccle ["ccle.ga4gh.VariantAnnotation" "ccle.ga4gh.CallSet" "ccle.ResponseCurve" "ccle.Biosample" "ccle.GeneExpression" "ccle.Cohort" "ccle.ga4gh.Variant"]
   :cna ["ccle.bmeg.cna.CNASegment" "ccle.bmeg.cna.CNACallSet"]
   :ctdd ["ctdd.json.bmeg.phenotype.ResponseCurve" "ctdd.json.bmeg.phenotype.Compound"]
   :gdc ["gdc.Cohort"]
   :hugo ["hugo.GeneSynonym" "hugo.Pubmed" "hugo.GeneFamily" "hugo.Gene" "hugo.GeneDatabase"]
   :mc3 ["mc3.ga4gh.VariantAnnotation" "mc3.ga4gh.Variant" "mc3.ga4gh.CallSet"]
   :tcga ["tcga.IndividualCohort" "tcga.Biosample" "tcga.GeneExpression" "tcga.Individual"]})

(defn -main
  [& args]
  (let [config default-config
        protograph (load (get-in config [:protograph :path]))]
    (transform-topics config protograph args)))<commit_msg>running each protograph ingestor in a separate thread<commit_after>(ns ophion.protograph
  (:require
   [cheshire.core :as json]
   [taoensso.timbre :as log]
   [ophion.kafka :as kafka])
  (:import
   [protograph Protograph ProtographEmitter]))

(defn load
  [path]
  (Protograph/loadProtograph path))

(defn emitter
  [emit-vertex emit-edge]
  (reify ProtographEmitter
    (emitVertex [_ vertex] (emit-vertex vertex))
    (emitEdge [_ edge] (emit-edge edge))))

(defn process
  [protograph emit label data]
  (.processMessage protograph emit label data))

(defn kafka-emitter
  [producer vertex-topic edge-topic]
  (emitter
   (fn [vertex]
     (kafka/send producer vertex-topic (Protograph/writeJSON vertex)))
   (fn [edge]
     (kafka/send producer edge-topic (Protograph/writeJSON edge)))))

(defn transform-message
  [protograph emit message]
  (let [label (kafka/topic->label (.topic message))
        data (Protograph/readJSON (.value message))]
    (process protograph emit label data)))

(defn transform-kafka
  [config protograph consumer producer]
  (let [vertex-topic (str (get-in config [:protograph :output]) ".Vertex")
        edge-topic (str (get-in config [:protograph :output]) ".Edge")
        emit (kafka-emitter producer vertex-topic edge-topic)]
    (kafka/consume
     consumer
     (partial transform-message protograph emit))))

(defn transform-topics
  [config protograph topics]
  (let [host (get-in config [:kafka :host])
        group-id (get-in config [:kafka :consumer :group-id])
        consumer (kafka/consumer host group-id topics)
        producer (kafka/producer host)]
    (transform-kafka config protograph consumer producer)))

(def default-config
  {:protograph
   {:path "../gaia-bmeg/bmeg.protograph.yml"
    :output "protograph.bmeg"}
   :kafka kafka/default-config})

(def bmeg-topics
  {:ccle ["ccle.ga4gh.VariantAnnotation" "ccle.ga4gh.CallSet" "ccle.ResponseCurve" "ccle.Biosample" "ccle.GeneExpression" "ccle.Cohort" "ccle.ga4gh.Variant"]
   :cna ["ccle.bmeg.cna.CNASegment" "ccle.bmeg.cna.CNACallSet"]
   :ctdd ["ctdd.json.bmeg.phenotype.ResponseCurve" "ctdd.json.bmeg.phenotype.Compound"]
   :gdc ["gdc.Cohort"]
   :hugo ["hugo.GeneSynonym" "hugo.Pubmed" "hugo.GeneFamily" "hugo.Gene" "hugo.GeneDatabase"]
   :mc3 ["mc3.ga4gh.VariantAnnotation" "mc3.ga4gh.Variant" "mc3.ga4gh.CallSet"]
   :tcga ["tcga.IndividualCohort" "tcga.Biosample" "tcga.GeneExpression" "tcga.Individual"]})

(defn -main
  [& args]
  (let [config default-config
        protograph (load (get-in config [:protograph :path]))
        devourers (mapv
                   (fn [[key topics]]
                     (log/info "devouring" (name key))
                     (future
                       (transform-topics config protograph topics)))
                   bmeg-topics)]
    (doseq [devourer devourers]
      (deref devourer))))
{'fraction_non_alphanumeric': 0.13822933922326522, 'fraction_numerical': 0.003681207436039021, 'mean_word_length': 3.6804478897502153, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7026711', 'n_tokens_mistral': 1948, 'n_tokens_neox': 1874, 'n_words': 435}