<commit_before>(ns lambdaci.dsl)

(def initial-pipeline-state {:running [] :finished []})
(def pipeline-state (atom initial-pipeline-state))

(defn get-pipeline-state []
  @pipeline-state)

(defn reset-pipeline-state []
  (reset! pipeline-state initial-pipeline-state))

(defn set-running! [step-id]
  (swap! pipeline-state #(assoc %1 :running (cons step-id (:running %1)))))

(defn set-finished! [step-id] ;; TODO: this should also remove it from the running-list. at the moment, css magic makes appear ok
  (swap! pipeline-state #(assoc %1 :finished (cons step-id (:finished %1)))))

(defn- range-from [from len] (range (inc from) (+ (inc from) len)))

(defn execute-step [step args step-id]
  (set-running! step-id)
  (let [step-result (step args step-id)]
    (println (str "executing step " step-id step-result))
    (set-finished! step-id)
    {:outputs { step-id step-result}} ))


(defn steps-with-ids [steps prev-id]
  (let [significant-part (first prev-id)
        rest-part (rest prev-id)
        significant-ids (range-from significant-part (count steps))
        ids (map #(cons %1 rest-part) significant-ids)]
    (map vector ids steps)))

(defn merge-step-results [r1 r2]
  (merge-with merge r1 r2))

(defn execute-step-foo [args [step-id step]]
  (execute-step step args step-id))


(defn execute-steps-internal [step-result-producer steps args step-id]
  (let [step-results (step-result-producer args (steps-with-ids steps step-id))]
    (reduce merge-step-results args step-results)))


(defn serial-step-result-producer [args s-with-id]
  (map (partial execute-step-foo args) s-with-id))

(defn execute-steps [steps args step-id]
  (execute-steps-internal serial-step-result-producer steps args step-id))

(defn parallel-step-result-producer [args s-with-id]
  (pmap (partial execute-step-foo args) s-with-id))

(defn execute-steps-in-parallel [steps args step-id]
  (execute-steps-internal parallel-step-result-producer steps args step-id))

(defn in-parallel [& steps]
  (fn [args step-id]
    (execute-steps-in-parallel steps args (cons 0 step-id))))

(defn in-cwd [cwd & steps]
  (fn [args step-id]
    (execute-steps steps (assoc args :cwd cwd) (cons 0 step-id))))


(defn run [pipeline]<commit_msg>moving functions around for clarity<commit_after>(ns lambdaci.dsl)

(def initial-pipeline-state {:running [] :finished []})
(def pipeline-state (atom initial-pipeline-state))

(defn get-pipeline-state []
  @pipeline-state)

(defn reset-pipeline-state []
  (reset! pipeline-state initial-pipeline-state))

(defn set-running! [step-id]
  (swap! pipeline-state #(assoc %1 :running (cons step-id (:running %1)))))

(defn set-finished! [step-id] ;; TODO: this should also remove it from the running-list. at the moment, css magic makes appear ok
  (swap! pipeline-state #(assoc %1 :finished (cons step-id (:finished %1)))))

(defn- range-from [from len] (range (inc from) (+ (inc from) len)))

(defn execute-step [step args step-id]
  (set-running! step-id)
  (let [step-result (step args step-id)]
    (println (str "executing step " step-id step-result))
    (set-finished! step-id)
    {:outputs { step-id step-result}} ))


(defn execute-step-foo [args [step-id step]]
  (execute-step step args step-id))


(defn merge-step-results [r1 r2]
  (merge-with merge r1 r2))

(defn steps-with-ids [steps prev-id]
  (let [significant-part (first prev-id)
        rest-part (rest prev-id)
        significant-ids (range-from significant-part (count steps))
        ids (map #(cons %1 rest-part) significant-ids)]
    (map vector ids steps)))

(defn execute-steps-internal [step-result-producer steps args step-id]
  (let [step-results (step-result-producer args (steps-with-ids steps step-id))]
    (reduce merge-step-results args step-results)))


(defn serial-step-result-producer [args s-with-id]
  (map (partial execute-step-foo args) s-with-id))

(defn execute-steps [steps args step-id]
  (execute-steps-internal serial-step-result-producer steps args step-id))

(defn parallel-step-result-producer [args s-with-id]
  (pmap (partial execute-step-foo args) s-with-id))

(defn execute-steps-in-parallel [steps args step-id]
  (execute-steps-internal parallel-step-result-producer steps args step-id))

(defn in-parallel [& steps]
  (fn [args step-id]
    (execute-steps-in-parallel steps args (cons 0 step-id))))

(defn in-cwd [cwd & steps]
  (fn [args step-id]
    (execute-steps steps (assoc args :cwd cwd) (cons 0 step-id))))


(defn run [pipeline]
{'fraction_non_alphanumeric': 0.15353625783348254, 'fraction_numerical': 0.004923903312444047, 'mean_word_length': 3.884153005464481, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4292697', 'n_tokens_mistral': 1567, 'n_tokens_neox': 1553, 'n_words': 445}