<commit_before>(ns desdemona.utils
  (:require [clojure.test :refer [is]]
            [clojure.core.async :refer [chan sliding-buffer >!!]]
            [clojure.java.io :refer [resource]]
            [onyx.plugin.core-async :refer [take-segments!]]))

;;;; Test utils ;;;;

(def zk-address "127.0.0.1")

(def zk-port 2188)

(def zk-str (str zk-address ":" zk-port))

(defn only [coll]
  (assert (not (next coll)))
  (if-let [result (first coll)]
    result
    (assert false)))

(defn find-task
  "Finds the catalog entry where the :onyx/name key equals task-name"
  [catalog task-name]
  (let [matches (filter #(= task-name (:onyx/name %)) catalog)]
    (when-not (seq matches)
      (throw (ex-info (format "Couldn't find task %s in catalog" task-name)
                      {:catalog catalog :task-name task-name})))
    (first matches)))

(defn update-task
  "Finds the catalog entry with :onyx/name task-name
  and applies f to it, returning the full catalog with the
  transformed catalog entry"
  [catalog task-name f]
  (mapv (fn [entry]
          (if (= task-name (:onyx/name entry))
            (f entry)
            entry))
        catalog))

(defn add-to-job
  "Adds to the catalog and lifecycles of a job in form
  {:workflow ...
   :catalog ...
   :lifecycles ...}"
  [job {:keys [catalog lifecycles]}]
  (-> job
      (update :catalog into catalog)
      (update :lifecycles into lifecycles)))

(defn n-peers
  "Takes a workflow and catalog, returns the minimum number of peers
   needed to execute this job."
  [catalog workflow]
  (let [task-set (into #{} (apply concat workflow))]
    (reduce<commit_msg>into #{} -> set<commit_after>(ns desdemona.utils
  (:require [clojure.test :refer [is]]
            [clojure.core.async :refer [chan sliding-buffer >!!]]
            [clojure.java.io :refer [resource]]
            [onyx.plugin.core-async :refer [take-segments!]]))

;;;; Test utils ;;;;

(def zk-address "127.0.0.1")

(def zk-port 2188)

(def zk-str (str zk-address ":" zk-port))

(defn only [coll]
  (assert (not (next coll)))
  (if-let [result (first coll)]
    result
    (assert false)))

(defn find-task
  "Finds the catalog entry where the :onyx/name key equals task-name"
  [catalog task-name]
  (let [matches (filter #(= task-name (:onyx/name %)) catalog)]
    (when-not (seq matches)
      (throw (ex-info (format "Couldn't find task %s in catalog" task-name)
                      {:catalog catalog :task-name task-name})))
    (first matches)))

(defn update-task
  "Finds the catalog entry with :onyx/name task-name
  and applies f to it, returning the full catalog with the
  transformed catalog entry"
  [catalog task-name f]
  (mapv (fn [entry]
          (if (= task-name (:onyx/name entry))
            (f entry)
            entry))
        catalog))

(defn add-to-job
  "Adds to the catalog and lifecycles of a job in form
  {:workflow ...
   :catalog ...
   :lifecycles ...}"
  [job {:keys [catalog lifecycles]}]
  (-> job
      (update :catalog into catalog)
      (update :lifecycles into lifecycles)))

(defn n-peers
  "Takes a workflow and catalog, returns the minimum number of peers
   needed to execute this job."
  [catalog workflow]
  (let [task-set (set (apply concat workflow))]
    (reduce
{'fraction_non_alphanumeric': 0.14254859611231102, 'fraction_numerical': 0.006170934896636841, 'mean_word_length': 2.752314814814815, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '1324006', 'n_tokens_mistral': 1125, 'n_tokens_neox': 1102, 'n_words': 350}