luciano/xforms
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Makes x/for to support kvs in and out, remove tag and map-kv (replaced by for)

Browse source
This commit is contained in:
Christophe Grand 2016-06-01 10:28:08 +02:00
parent 07225a92a4
commit 219e7d2aac
1 changed files with 86 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

171
src/net/cgrand/xforms.clj
View file

@ -6,25 +6,42 @@
(defmacro for (defmacro for
"Like clojure.core/for with the first expression being replaced by % (or _). Returns a transducer." "Like clojure.core/for with the first expression being replaced by % (or _). Returns a transducer."
[[binding %or_ & seq-exprs] body] [[binding %or_ & seq-exprs] body-expr]
(assert (and (symbol? %or_) (#{"%" "_"} (name %or_))) (assert (and (symbol? %or_) (#{"%" "_"} (name %or_)))
"The second element of the comprehension vector must be % or _.") "The second element of the comprehension vector must be % or _.")
(let [rf (gensym 'rf) (let [rf (gensym 'rf)
acc (gensym 'acc) acc (gensym 'acc)
body pair? #(and (vector? %) (= 2 (clj/count %)))
(clj/reduce (fn [body [expr binding]] build (fn [init]
(clj/reduce (fn [body [expr binding]]
(case binding (case binding
:let `(let ~expr ~body) :let `(let ~expr ~body)
:when `(if ~expr ~body ~acc) :when `(if ~expr ~body ~acc)
:while `(if ~expr ~body (reduced ~acc)) :while `(if ~expr ~body (reduced ~acc))
`(clj/reduce (fn [~acc ~binding] ~body) ~acc ~expr))) `(clj/reduce (fn [~acc ~binding] ~body) ~acc ~expr)))
`(~rf ~acc ~body) init (clj/partition 2 (rseq (vec seq-exprs)))))
(clj/partition 2 (rseq (vec seq-exprs))))] body (build `(~rf ~acc ~body-expr))
`(fn [~rf] kvbody (when (pair? body-expr) (build `(~rf ~acc ~@body-expr)))
(fn fnsym (if (and (pair? binding) (not (some keyword? binding)) (not (some #{"&"} (filter symbol? binding)))) `kvrf `fn)]
([] (~rf)) (if kvbody
([~acc] (~rf ~acc)) `(fn [~rf]
([~acc ~binding] ~body))))) (if-some [~rf (some-kvrf ~rf)]
(~fnsym
([] (~rf))
([~acc] (~rf ~acc))
([~acc ~binding] ~kvbody)
~@(when (= fnsym `kvrf) [`([~acc ~@binding] ~kvbody)]))
(~fnsym
([] (~rf))
([~acc] (~rf ~acc))
([~acc ~binding] ~body)
~@(when (= fnsym `kvrf) [`([~acc ~@binding] ~body)]))))
`(fn [~rf]
(~fnsym
([] (~rf))
([~acc] (~rf ~acc))
([~acc ~binding] ~body)
~@(when (= fnsym `kvrf) [`([~acc ~@binding] ~body)]))))))
(defprotocol KvRfable "Protocol for reducing fns that takes key and val as separate arguments." (defprotocol KvRfable "Protocol for reducing fns that takes key and val as separate arguments."
(some-kvrf [f] "Returns a kvrf or nil")) (some-kvrf [f] "Returns a kvrf or nil"))
@ -45,6 +62,14 @@
~@(clj/for [[args & body] fn-bodies] ~@(clj/for [[args & body] fn-bodies]
`(invoke [~name ~@args] ~@body))))) `(invoke [~name ~@args] ~@body)))))
(defn ensure-kvrf [rf]
(or (some-kvrf rf)
(kvrf
([] (rf))
([acc] (rf acc))
([acc x] (rf acc x))
([acc k v] (rf acc (clojure.lang.MapEntry. k v))))))
(defn reduce (defn reduce
"A transducer that reduces a collection to a 1-item collection consisting of only the reduced result. "A transducer that reduces a collection to a 1-item collection consisting of only the reduced result.
Unlike reduce but like transduce it does call the completing arity (1) of the reducing fn." Unlike reduce but like transduce it does call the completing arity (1) of the reducing fn."
@ -151,9 +176,8 @@
(let [make-rf (cond (let [make-rf (cond
(nil? pair) (constantly (multiplexable rf)) (nil? pair) (constantly (multiplexable rf))
(= ::default pair) (= ::default pair)
(if-some [rf (some-kvrf rf)] (let [rf (ensure-kvrf rf)]
(fn [k] (fn ([acc] acc) ([acc v] (rf acc k v)))) (fn [k] (fn ([acc] acc) ([acc v] (rf acc k v)))))
(fn [k] (fn ([acc] acc) ([acc v] (rf acc (vector k v))))))
:else (fn [k] (fn ([acc] acc) ([acc v] (rf acc (pair k v)))))) :else (fn [k] (fn ([acc] acc) ([acc v] (rf acc (pair k v))))))
m (volatile! (transient {}))] m (volatile! (transient {}))]
(if (and (nil? kfn) (nil? vfn)) (if (and (nil? kfn) (nil? vfn))
@ -290,48 +314,55 @@
n is the integral number of steps by which the window slides. With a 1-hour window, 4 means that the window slides every 15 minutes. n is the integral number of steps by which the window slides. With a 1-hour window, 4 means that the window slides every 15 minutes.
f and invf work like in #'window." f and invf work like in #'window."
[timef n f invf] ([timef n f]
(let [timef (fn [x] (long (Math/floor (* n (timef x)))))] (window-by-time timef n
(fn [rf] (fn
(let [dq (java.util.ArrayDeque.) ([] clojure.lang.PersistentQueue/EMPTY)
vwacc (volatile! (f)) ([q] (f (clj/reduce f (f) q)))
flush! ([q x] (conj q x)))
(fn [acc ^long from-ts ^long to-ts] (fn [q _] (pop q))))
(loop [ts from-ts acc acc wacc @vwacc] ([timef n f invf]
(let [x (.peekFirst dq)] (let [timef (fn [x] (long (Math/floor (* n (timef x)))))]
(cond (fn [rf]
(= ts (timef x)) (let [dq (java.util.ArrayDeque.)
(do vwacc (volatile! (f))
(.pollFirst dq) flush!
(recur ts acc (invf wacc x))) (fn [acc ^long from-ts ^long to-ts]
(= ts to-ts) (loop [ts from-ts acc acc wacc @vwacc]
(do (let [x (.peekFirst dq)]
(vreset! vwacc wacc) (cond
acc) (= ts (timef x))
:else (do
(let [acc (rf acc (f wacc))] (.pollFirst dq)
(if (reduced? acc) (recur ts acc (invf wacc x)))
(do (= ts to-ts)
(vreset! vwacc wacc) (do
acc) (vreset! vwacc wacc)
(recur (inc ts) acc wacc)))))))] acc)
(fn :else
([] (rf)) (let [acc (rf acc (f wacc))]
([acc] (if (reduced? acc)
(let [acc (if-not (.isEmpty dq) (do
(unreduced (rf acc (f @vwacc))) (vreset! vwacc wacc)
acc)] acc)
(rf acc))) (recur (inc ts) acc wacc)))))))]
([acc x] (fn
(let [limit (- (timef x) n) ([] (rf))
prev-limit (if-some [prev-x (.peekLast dq)] ([acc]
(- (timef prev-x) n) (let [acc (if-not (.isEmpty dq)
limit) (unreduced (rf acc (f @vwacc)))
_ (.addLast dq x) ; so dq is never empty for flush! acc)]
acc (flush! acc prev-limit limit)] (rf acc)))
(when-not (reduced? acc) ([acc x]
(vswap! vwacc f x)) (let [limit (- (timef x) n)
acc))))))) prev-limit (if-some [prev-x (.peekLast dq)]
(- (timef prev-x) n)
limit)
_ (.addLast dq x) ; so dq is never empty for flush!
acc (flush! acc prev-limit limit)]
(when-not (reduced? acc)
(vswap! vwacc f x))
acc))))))))
(defn count ([] 0) ([n] n) ([n _] (inc n))) (defn count ([] 0) ([n] n) ([n _] (inc n)))
@ -386,33 +417,3 @@
(fn (fn
))) )))
) )
(defn tag
"Like (map #(vector tag %)) but potentially more efficient."
[tag]
(fn [rf]
(if-some [rf (some-kvrf rf)]
(fn
([] (rf))
([acc] (rf acc))
([acc v] (rf acc tag v)))
(fn
([] (rf))
([acc] (rf acc))
([acc v] (rf acc [tag v]))))))
(defn map-kv
"Like (map (fn [[k v]] [(kf k v) (vf k v)])) but potentially more efficient."
[kf vf]
(fn [rf]
(if-some [rf (some-kvrf rf)]
(kvrf
([] (rf))
([acc] (rf acc))
([acc [k v]] (rf acc (kf k v) (vf k v)))
([acc k v] (rf acc (kf k v) (vf k v))))
(kvrf
([] (rf))
([acc] (rf acc))
([acc [k v]] (rf acc [(kf k v) (vf k v)]))
([acc k v] (rf acc [(kf k v) (vf k v)]))))))