async flow last updates

project.clj

@@ -10,6 +10,7 @@
                  [org.clojure/data.priority-map "0.0.7"]
                  [org.clojure/core.match "0.3.0-alpha4"]
                  [org.clojure/core.unify "0.5.5"]
+                 [org.clojure/core.async "0.2.374"]
                  [com.taoensso/carmine "2.12.2"]
                  [mount "0.1.10"]]
   :main ^:skip-aot narjure.core

src/nal/core.clj

@@ -1,6 +1,7 @@
 (ns nal.core
   (:require [nal.deriver.truth :as t]
-            [nal.deriver :refer [generate-conclusions]]))
+            [nal.deriver :refer [generate-conclusions]]
+            [nal.rules :as r]))
 
 (defn choice [[f1 c1] [f2 c2]]
   (if (>= c1 c2) [f1 c1] [f2 c2]))
@@ -10,3 +11,8 @@
   (generate-conclusions (rules task-type) task belief))
 
 (def revision t/revision)
+
+(comment
+  :shift-occurrence-forward                                 ;pre
+  :shift-occurrence-backward                                ;pre
+  :linkage-temporal)

src/nal/deriver/matching.clj

@@ -23,7 +23,7 @@
     `n/reduce-int-dif `n/reduce-and `n/reduce-ext-dif `n/reduce-image
     `n/reduce-int-inter `n/reduce-neg `n/reduce-or `nil? `not `or `abs
     `implications-and-equivalences `get-terms `empty? `intersection
-    `n/reduce-seq-conj})
+    `n/reduce-seq-conj `clojure.core.match/match})
 
 (defn operators->placeholders
   [statement]

src/narjure/control/buffer.clj

@@ -0,0 +1,31 @@
+(ns narjure.control.buffer
+  (:require [clojure.core.async.impl.protocols :as impl])
+  (:import [java.util LinkedList]
+           [clojure.lang Fn Counted]))
+
+(deftype PanickingSlidingBuffer
+  [^LinkedList buf ^long n ^Fn warning-callback ^long warning-n]
+  impl/UnblockingBuffer
+  impl/Buffer
+  (full? [this]
+    false)
+  (remove! [this]
+    (.removeLast buf))
+  (add!* [this itm]
+    (let [size (.size buf)]
+      (when (>= size warning-n)
+        (warning-callback size)
+        (when (= size n)
+          (impl/remove! this))))
+    (.addFirst buf itm)
+    this)
+  (close-buf! [this])
+  Counted
+  (count [this]
+    (.size buf)))
+
+(defn panicking-sliding-buffer
+  ([n callback]
+   (panicking-sliding-buffer n callback (Math/round (* 0.9 n))))
+  ([n callback warning-n]
+   (PanickingSlidingBuffer. (LinkedList.) n callback warning-n)))

src/narjure/control/flow.clj

@@ -0,0 +1,110 @@
+(ns narjure.control.flow
+  (:require [clojure.core.async :refer [go-loop <! >! chan]]
+            [clojure.set :as set]))
+
+(defn check-element-in-map
+  "Checks if elements exist in map, if not
+  assocs elements to map with default value."
+  ([s m] (check-element-in-map s 0 m))
+  ([s default m]
+   (reduce (fn [ac k]
+             (if (ac k)
+               ac
+               (assoc ac k default)))
+           m s)))
+
+(defn all
+  "Returns set aff all functions from workflow."
+  [wf]
+  (set (flatten wf)))
+
+(defn kw->fn
+  "Transform function's keyword to var."
+  [kw]
+  (->> (str kw)
+       (drop 1)
+       (apply str)
+       symbol
+       resolve))
+
+(defn vertex
+  "Creates vertex of flow graph. Arguments:
+   - functions: collection of collections, where first element is function
+     and second (optional) is output port
+   - inputs: ports (edges) that should be listened by vertex
+   - p: number of parallelism"
+  [functions inputs p]
+  (doseq [in inputs
+          _ (range (* p (count functions)))]
+    (go-loop []
+      (when-let [val (<! in)]
+        (doseq [[f out] functions]
+          (let [results (f val)]
+            (when out
+              (doseq [res (if (map? results)
+                            [results]
+                            results)]
+                (>! out res)))))
+        (recur)))))
+
+(defn check-output
+  "Creates output port for function if it is necessary."
+  [buffer [function output-cnt]]
+  [function (when (pos? output-cnt) (chan buffer))])
+
+(defn fn-outputs
+  "Generates map where keys are functions and values are ports which
+  will be used to send result of execution of functions."
+  [workflow buffer]
+  (->> (group-by first workflow)
+       (map (fn [[n t]] [n (count t)]))
+       (into {})
+       (check-element-in-map (all workflow))
+       (map #(check-output buffer %))
+       (into {})))
+
+(defn fn-inputs [workflow]
+  "Groups functions to identify vertexes and edges that they should listen.
+  Returns map {vertexes edges ...}
+
+  in: [[:a :b]
+       [:a :c]
+       [:c :d]
+       [:b :d]]
+
+  out: {[:c :b] [:a], [:d] [:b :c]}"
+  (->> workflow
+       (reduce (fn [ac [k v]]
+                 (update ac k conj v))
+               {})
+       (reduce (fn [ac [k v]]
+                 (update ac v conj k))
+               {})))
+
+(defn generate-flow
+  "Generates flow of functions which is discribed by pairs of functions,
+  where result of fisrt function will be sent to input of the second function.
+  Optionally map of configuration params can be passed.
+  Possible configs:
+   - parallelism: map where keys are functions and values
+     are parallelization numbers
+   - default-p: defaulp parallelization number
+   - buffer: capacity of fixed buffer for all channels"
+  ;TODO configuration for custom buffers
+  ([workflow] (generate-flow workflow {}))
+  ([workflow {:keys [parallelism default-p buffer]
+              :or   {parallelism {}
+                     default-p   1
+                     buffer      100}}]
+   (let [in (chan buffer)
+         outputs (assoc (fn-outputs workflow buffer) :in in)
+         inputs (fn-inputs workflow)
+         all-inputs (set (mapcat key inputs))
+         input-tasks (set/difference (all workflow) all-inputs)
+         it2 (assoc inputs input-tasks [:in])]
+     (doseq [[tasks from] it2]
+       (vertex
+         (map (fn [f] [(kw->fn f) (outputs f)]) tasks)
+         (map outputs from)
+         (apply max (map #(parallelism % default-p) tasks))))
+     in)))

src/narjure/control/general_inference.clj

@@ -0,0 +1,69 @@
+(ns narjure.control.general-inference
+  (:require [narjure.system :refer [memory inference]]
+            [narjure.control.flow :as f]
+            [narjure.memory.api :as m]))
+
+;;  select-concepts
+;;     | | |     |
+;;     v v v ... v
+;;  select-task-link --> update-tasklink-budget
+;;         |
+;;         v
+;;  select-term-link --> update-termlink-budget
+;;         |
+;;         v
+;;   do-inference
+(def workflow
+  [[::select-concepts ::select-task-link]
+
+   [::select-task-link ::update-tasklink-budget]
+   [::select-task-link ::select-term-link]
+
+   [::select-term-link ::update-termlink-budget]
+   [::select-term-link ::do-inference]])
+
+
+(defn select-concept [_]
+  (m/pull-activated-concepts memory))
+
+(defn select-task-link [concept]
+  (let [{task-id :task} (m/select-tasklink memory concept)
+        task (m/task memory task-id)]
+    {:task    task
+     :concept concept}))
+
+;(defn update-concept-budget [data] data)
+
+(defn select-term-link [{:keys [concept task] :as data}]
+  (let [{linked-concept :concept} (m/select-termlink memory concept)
+        occurrence (:occurrence task)
+        truth (m/select-truth memory linked-concept occurrence)]
+    (assoc data :truth truth
+                :term (m/term memory linked-concept))))
+;(defn update-tasklink-budget [data] data)
+;(defn update-termlink-budget [data] data)
+
+(defn do-inference [{:keys [task truth term]}]
+  (let [{statement :term
+         :keys     [frequency confidence plausibility
+                    desirability task-type occurrence]}
+        task
+
+        task {:statement  statement
+              :desire     [plausibility desirability]
+              :truth      [frequency confidence]
+              :task-type  task-type
+              :occurrence occurrence}
+
+        truth (assoc truth :statement term)
+        results (inference task truth)]
+    (doseq [res results]
+      (m/push-task memory res))))
+
+(def default-parallelism {::do-inference 4})
+
+(defn general-inference-flow
+  [{:keys [parallelism]
+    :or   {parallelism default-parallelism}
+    :as   options}]
+  (f/generate-flow workflow options))

src/narjure/control/local_inference.clj

@@ -0,0 +1,37 @@
+(ns narjure.control.local-inference)
+
+(def workflow
+  [[:read-task :answer-yn-question]
+   [:read-task :answer-general-question]
+   [:answer-yn-question :out-answers]
+   [:answer-general-question :out-answers]
+
+   [:read-task :find-related-concepts]
+   [:find-related-concepts :out-update-tasklinks]
+   [:find-related-concepts :out-check-tasklinks-capacity]
+
+   [:read-task :belief-revision]
+   [:belief-revision :out-update-beliefs]
+
+   [:read-task :goal-revision]
+   [:goal-revision :out-update-goals]])
+
+(defn answer-yn-question [{:keys [task] :as segment}]
+  (println :yn)
+  {})
+
+(defn answer-general-question [{:keys [task] :as segment}]
+  (println :general)
+  {})
+
+(defn find-related-concepts [{:keys [task] :as segment}]
+  (println :rel)
+  {})
+
+(defn belief-revision [{:keys [task] :as segment}]
+  (println :bel)
+  {})
+
+(defn goal-revision [{:keys [task] :as segment}]
+  (println :goal)
+  {})

test/nal/test/underiver.clj

@@ -0,0 +1,64 @@
+(ns nal.test.underiver
+  (:require nal.reader
+            [nal.deriver.rules :as r]
+            [nal.deriver.matching :as m]
+            [nal.deriver.utils :as u]
+            [clojure.core.unify :as un]
+            [clojure.core.match :as omg]
+            [clojure.set :as cs]
+            [nal.core :as c]))
+
+(r/defrules rls
+  #R[(P ==> M) (S ==> M) |- (S ==> P) :post (:t/induction :allow-backward) :pre ((:!= S P))])
+
+(def compiled (r/compile-rules rls))
+(def r-map (first (r/rule (first rls))))
+
+(defn sym-map [m p1 p2]
+  (let [vals (set (vals m))
+        all (cs/difference (set (remove u/operator? (flatten [p1 p2])))
+                           vals)]
+    (->> all
+         (map (fn [el] [`(quote ~el) el]))
+         (into {})
+         (merge m))))
+
+(defn underiver [{:keys [p1 p2 conclusions]}]
+  (let [concl (vec (:conclusion (first conclusions)))
+        [m pattern] (m/find-and-replace-symbols concl "x")
+        m (sym-map m p1 p2)
+        p1 (m/replace-symbols p1 m)
+        p2 (m/replace-symbols p2 m)]
+    (eval (m/quote-operators
+            `(fn [xn#] (omg/match xn#
+                         ~pattern [~p1 ~p2]
+                         :else []))))))
+
+(defn deriver [rls]
+  (let [compiled (r/compile-rules rls)]
+    (fn [[p1 p2]]
+      (let [t {:statement  p1
+               :desire     [1 0.9]
+               :task-type  :judgement
+               :occurrence 1}
+            b {:statement  p2
+               :truth      [1 0.9]
+               :occurrence 0}]
+        (c/inference compiled t b)))))
+
+(comment
+  ((underiver r-map) '[==> wut ahh?])
+
+  => [[==> ahh? M] [==> wut M]]
+
+  ((deriver rls) '[[==> ahh? M] [==> wut M]])
+  (let [[{st :statement}] (c/inference compiled
+                                       {:statement  '[==> ahh? M]
+                                        :truth      [1 0.9]
+                                        :task-type  :judgement
+                                        :occurrence 1}
+                                       {:statement  '[==> wut M]
+                                        :truth      [1 0.9]
+                                        :occurrence 0})]
+    ((underiver r-map) st))
+  )

test/narjure/test/control/buffer.clj

@@ -0,0 +1,41 @@
+(ns narjure.test.control.buffer
+  (:require
+    [clojure.test :refer :all]
+    [narjure.control.buffer :refer :all]
+    [clojure.core.async.impl.protocols :refer [full? add! remove! close-buf!]]))
+
+(defmacro throws? [expr]
+  `(try
+     ~expr
+     false
+     (catch Throwable _# true)))
+
+(def warn-cnt (atom 0))
+
+(defn panick! [_]
+  (swap! warn-cnt inc))
+
+(deftest sliding-buffer-tests
+  (let [fb (panicking-sliding-buffer 2 panick! 1)]
+    (reset! warn-cnt 0)
+    (is (= 0 (count fb)))
+
+    (add! fb :1)
+    (is (= 1 (count fb)))
+
+    (add! fb :2)
+    (is (= 2 (count fb)))
+    (is (= 1 @warn-cnt))
+
+    (is (not (full? fb)))
+    (is (not (throws? (add! fb :3))))
+    (is (= 2 (count fb)))
+
+    (is (= :2 (remove! fb)))
+    (is (not (full? fb)))
+
+    (is (= 1 (count fb)))
+    (is (= :3 (remove! fb)))
+
+    (is (= 0 (count fb)))
+    (is (throws? (remove! fb)))))

test/narjure/test/control/flow.clj

@@ -0,0 +1,65 @@
+(ns narjure.test.control.flow
+  (:require
+    [clojure.test :refer :all]
+    [narjure.control.flow :refer :all]
+    [clojure.core.async :as as]))
+
+(deftest test-check-element-in-map
+  (let [m {:a 1 :b 2}]
+    (is (= 0 (:c (check-element-in-map [:c] m))))
+    (is (= [] (:c (check-element-in-map [:c] [] m))))
+    (is (= nil (:k (check-element-in-map [:c] [] m))))))
+
+(deftest test-all
+  (is (= #{:a :b :c :d :e}
+         (all [[:a :b]
+               [:a :c]
+               [:c :d]
+               [:c :e]]))))
+
+(defn some-var [])
+(deftest test-kw->var
+  (is (var? (kw->fn :narjure.control.flow/all)))
+  (is (var? (kw->fn ::some-var)))
+  (is (nil? (kw->fn ::wrong-var))))
+
+(def wf [[:a :b]
+         [:a :c]
+         [:a :d]
+         [:b :d]])
+
+(deftest test-fn-outputs
+  (let [outputs (fn-outputs wf 1)]
+    (is (= 4 (count (keys outputs))))
+    (is (every? nil? (map outputs [:c :d])))))
+
+(deftest test-fn-inputs
+  (is (= {[:d :c :b] [:a]
+          [:d]       [:b]}
+         (fn-inputs wf)))
+  (is (= {[:d :c :b] [:a]
+          [:k :d]    [:c :b]}
+         (fn-inputs (concat wf [[:c :d]
+                                [:c :k]
+                                [:b :k]])))))
+
+(def out (as/chan 2))
+(defn first-fn [data] (assoc data :fn1 :ok))
+(defn second-fn [data] (assoc data :fn2 :ok))
+(defn third-fn [data]
+  (as/>!! out (assoc data :fn3 :ok)))
+
+(def test-flow [[::first-fn ::second-fn]
+                [::second-fn ::third-fn]
+                [::first-fn ::third-fn]])
+
+(deftest test-generate-flow
+  (let [in (generate-flow test-flow {:buffer 2})]
+    (as/>!! in {})
+    (let [res [(as/<!! out) (as/<!! out)]]
+      (is (= #{{:fn1 :ok
+                :fn3 :ok}
+               {:fn1 :ok
+                :fn2 :ok
+                :fn3 :ok}}
+             (set res))))))