From 002ddbee48f5612ce0e2af0e62a6aaab2d1d1654 Mon Sep 17 00:00:00 2001
From: Alexander Nozik <altavir@gmail.com>
Date: Sat, 2 Feb 2019 19:56:45 +0300
Subject: [PATCH] Sequential operations

---
 .../kmath/sequential/Accumulators.kt          | 37 ++++++++++++++++---
 .../scientifik/kmath/sequential/Reducers.kt   |  2 +-
 .../kmath/sequential/AccumulatorsExt.kt       | 28 ++++++++++++++
 .../kmath/{chains => sequential}/ChainExt.kt  |  3 +-
 4 files changed, 61 insertions(+), 9 deletions(-)
 create mode 100644 kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/AccumulatorsExt.kt
 rename kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/{chains => sequential}/ChainExt.kt (93%)

diff --git a/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Accumulators.kt b/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Accumulators.kt
index 283db0809..94ac0ac2e 100644
--- a/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Accumulators.kt
+++ b/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Accumulators.kt
@@ -1,25 +1,50 @@
 package scientifik.kmath.sequential
 
 import kotlinx.atomicfu.atomic
+import kotlinx.atomicfu.atomicArrayOfNulls
 import kotlinx.atomicfu.getAndUpdate
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
 import scientifik.kmath.operations.Space
 
 /**
  * An object with a state that accumulates incoming elements
  */
 interface Accumulator<in T> {
-    //PENDING use suspend operations?
+    /**
+     * Push a value to accumulator. Blocks if [Accumulator] can't access any more elements at that time
+     */
     fun push(value: T)
-}
 
-fun <T> Accumulator<T>.pushAll(values: Iterable<T>) {
-    values.forEach { push(it) }
+    /**
+     * Does the same as [push], but suspends instead of blocking if accumulator is full
+     */
+    suspend fun send(value: T) = push(value)
 }
 
 /**
- * Generic thread-safe summator
+ * Push all elements to accumulator
  */
-class GenericSum<T : Any>(val context: Space<T>) : Accumulator<T> {
+fun <T> Accumulator<T>.pushAll(values: Iterable<T>) {
+    for (value in values) {
+        push(value)
+    }
+}
+
+/**
+ * Offer all elements from channel to accumulator
+ */
+suspend fun <T> Accumulator<T>.offerAll(channel: ReceiveChannel<T>) {
+    for (value in channel) {
+        send(value)
+    }
+}
+
+/**
+ * Generic thread-safe average
+ */
+class GenericMean<T : Any>(val context: Space<T>) : Accumulator<T> {
     //TODO add guard against overflow
     val counter = atomic(0)
     val sum = atomic(context.zero)
diff --git a/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Reducers.kt b/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Reducers.kt
index 65dea5ac0..cf0601135 100644
--- a/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Reducers.kt
+++ b/kmath-sequential/src/commonMain/kotlin/scientifik/kmath/sequential/Reducers.kt
@@ -16,6 +16,6 @@ inline fun <T, C, R> Array<T>.reduce(context: C, crossinline reducer: Reducer<T,
 
 object Reducers {
     fun <T : Any> mean(): Reducer<T, Space<T>, T> = { context, data ->
-        data.fold(GenericSum(context)) { sum, value -> sum.apply { push(value) } }.value
+        data.fold(GenericMean(context)) { sum, value -> sum.apply { push(value) } }.value
     }
 }
\ No newline at end of file
diff --git a/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/AccumulatorsExt.kt b/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/AccumulatorsExt.kt
new file mode 100644
index 000000000..cc9a07668
--- /dev/null
+++ b/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/AccumulatorsExt.kt
@@ -0,0 +1,28 @@
+package scientifik.kmath.sequential
+
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ReceiveChannel
+import scientifik.kmath.operations.Space
+import scientifik.kmath.structures.runBlocking
+import java.util.*
+
+/**
+ * A moving average with fixed window
+ */
+class MovingAverage<T : Any>(val window: Int, val context: Space<T>) : Accumulator<T> {
+    private val outputChannel = Channel<T>()
+    private val queue = ArrayDeque<T>(window)
+
+    override suspend fun send(value: T) {
+        queue.add(value)
+        if (queue.size == window) {
+            val sum = queue.fold(context.zero) { a, b -> context.run { a + b } }
+            outputChannel.send(context.run { sum / window })
+            queue.pop()
+        }
+    }
+
+    override fun push(value: T) = runBlocking { send(value) }
+
+    val output: ReceiveChannel<T> = outputChannel
+}
\ No newline at end of file
diff --git a/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/chains/ChainExt.kt b/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/ChainExt.kt
similarity index 93%
rename from kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/chains/ChainExt.kt
rename to kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/ChainExt.kt
index af61a5059..74cb6bc6d 100644
--- a/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/chains/ChainExt.kt
+++ b/kmath-sequential/src/jvmMain/kotlin/scientifik/kmath/sequential/ChainExt.kt
@@ -1,7 +1,6 @@
-package scientifik.kmath.chains
+package scientifik.kmath.sequential
 
 import kotlinx.coroutines.runBlocking
-import scientifik.kmath.sequential.Chain
 import kotlin.sequences.Sequence
 
 /**