fix aggregations

README.md

@@ -43,7 +43,7 @@
 `Flinkrunner 3` is available on maven central, built against Flink 1.11 with Scala 2.12 and JDK 11.
 
 ```sbtshell
-libraryDependencies += "io.epiphanous" %% "flinkrunner" % "3.0.5"
+libraryDependencies += "io.epiphanous" %% "flinkrunner" % "3.0.6"
 ```
 
 ## What is FlinkRunner?

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/Count.scala

@@ -2,7 +2,7 @@ package io.epiphanous.flinkrunner.model.aggregate
 import java.time.Instant
 
 import io.epiphanous.flinkrunner.model.UnitMapper
-import squants.Each
+import squants.{Dimensionless, Each, Quantity}
 
 final case class Count(
   dimension: String,
@@ -19,11 +19,11 @@ final case class Count(
 
   override def outUnit: String = Each.symbol
 
-  override def update(
-    value: Double,
-    unit: String,
-    aggLU: Instant,
-    unitMapper: UnitMapper = UnitMapper.defaultUnitMapper
-  ) =
-    Some(copy(value = this.value + 1, unit = outUnit, count = count + 1, aggregatedLastUpdated = aggLU))
+  override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) =
+    current + current.unit(1)
+
+}
+
+object Count {
+  def apply(): Count = new Count(Dimensionless.name, Each.symbol)
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingAverage.scala

@@ -17,7 +17,7 @@ final case class ExponentialMovingAverage(
   def alpha: Double = params.getOrElse("alpha", ExponentialMovingAverage.defaultAlpha).toDouble
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) =
-    current * (1 - alpha) + quantity * alpha
+    if (count == 0) quantity else current * (1 - alpha) + quantity * alpha
 
 }
 

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingStandardDeviation.scala

@@ -21,8 +21,10 @@ final case class ExponentialMovingStandardDeviation(
   }
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) = {
-    val updatedEmv = depAggs("ExponentialMovingVariance")
-    current.unit(Math.sqrt(updatedEmv.value))
+    if (count == 0) current.unit(0d) else {
+      val updatedEmv = depAggs("ExponentialMovingVariance")
+      current.unit(Math.sqrt(updatedEmv.value))
+    }
   }
 
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingVariance.scala

@@ -23,10 +23,12 @@ final case class ExponentialMovingVariance(
   }
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) = {
-    val currentEma = getDependents("ExponentialMovingAverage")
-    val q = quantity in current.unit
-    val delta = q - current.unit(currentEma.value)
-    (1 - alpha) * (current + delta * delta.value * alpha)
+    if (count == 0) quantity.unit(0d) else {
+      val currentEma = getDependents("ExponentialMovingAverage")
+      val q          = quantity in current.unit
+      val delta      = q - current.unit(currentEma.value)
+      (1 - alpha) * (current + delta * delta.value * alpha)
+    }
   }
 
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/Max.scala

@@ -15,6 +15,6 @@ final case class Max(
     extends Aggregate {
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) =
-    current.max(quantity)
+    if (count == 0) quantity else current.max(quantity)
 
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/Min.scala

@@ -15,6 +15,6 @@ final case class Min(
     extends Aggregate {
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) =
-    current.min(quantity)
+    if (count == 0) quantity else current.min(quantity)
 
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/Range.scala

@@ -21,5 +21,5 @@ final case class Range(
   }
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) =
-    current.unit(depAggs("Max").value - depAggs("Min").value)
+    if (count == 0) current else current.unit(depAggs("Max").value - depAggs("Min").value)
 }

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/StandardDeviation.scala

@@ -16,8 +16,10 @@ final case class StandardDeviation(
     extends Aggregate {
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) = {
-    val updatedVariance = depAggs("Variance")
-    current.unit(Math.sqrt(updatedVariance.value))
+    if (count == 0) current else {
+      val updatedVariance = depAggs("Variance")
+      current.unit(Math.sqrt(updatedVariance.value))
+    }
   }
 
   override def getDependents = {

src/main/scala/io/epiphanous/flinkrunner/model/aggregate/Variance.scala

@@ -21,8 +21,10 @@ final case class Variance(
   }
 
   override def updateQuantity[A <: Quantity[A]](current: A, quantity: A, depAggs: Map[String, Aggregate]) = {
-    val k = count.doubleValue()
-    val s = current.unit(depAggs("SumOfSquaredDeviations").value)
-    s / (k - 1)
+    if (count < 2) current else {
+      val k = count.doubleValue()
+      val s = current.unit(depAggs("SumOfSquaredDeviations").value)
+      s / k
+    }
   }
 }

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/CountSpec.scala

@@ -0,0 +1,22 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import io.epiphanous.flinkrunner.model.UnitMapper
+import squants.Each
+
+class CountSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val c = Count()
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      c1 <- c.update(Each(1),t,u)
+      c2 <- c1.update(Each(1), t, u)
+    } yield c2.value
+    q.value shouldBe(2)
+  }
+
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingAverageSpec.scala

@@ -0,0 +1,22 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import io.epiphanous.flinkrunner.model.UnitMapper
+import squants.Kilograms
+import squants.mass.Mass
+
+class ExponentialMovingAverageSpec extends BasePropSpec {
+  property("updateQuantity property") {
+    val a = ExponentialMovingAverage(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      a1 <- a.update(Kilograms(10),t,u)
+      a2 <- a1.update(Kilograms(20), t, u)
+      a3 <- a2.update(Kilograms(30), t, u)
+    } yield a3.value
+    q.value shouldBe(26.1)
+  }
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingStandardDeviationSpec.scala

@@ -0,0 +1,26 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import io.epiphanous.flinkrunner.model.UnitMapper
+import org.scalactic.{Equality, TolerantNumerics}
+import squants.Kilograms
+import squants.mass.Mass
+
+class ExponentialMovingStandardDeviationSpec extends BasePropSpec {
+
+  implicit val tol: Equality[Double] = TolerantNumerics.tolerantDoubleEquality(1e-4)
+
+  property("updateQuantity property") {
+    val v = ExponentialMovingStandardDeviation(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      v1 <- v.update(Kilograms(10),t,u)
+      v2 <- v1.update(Kilograms(20), t, u)
+      v3 <- v2.update(Kilograms(30), t, u)
+    } yield v3.value
+    q.value shouldEqual(Math.sqrt(41.79))
+  }
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/ExponentialMovingVarianceSpec.scala

@@ -0,0 +1,26 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+import org.scalactic.{Equality, TolerantNumerics}
+
+class ExponentialMovingVarianceSpec extends BasePropSpec {
+
+  implicit val tol: Equality[Double] = TolerantNumerics.tolerantDoubleEquality(1e-4)
+
+  property("updateQuantity property") {
+    val v = ExponentialMovingVariance(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      v1 <- v.update(Kilograms(10),t,u)
+      v2 <- v1.update(Kilograms(20), t, u)
+      v3 <- v2.update(Kilograms(30), t, u)
+    } yield v3.value
+    q.value shouldEqual(41.79)
+  }
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/MaxSpec.scala

@@ -0,0 +1,21 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class MaxSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val m = Max(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      m1 <- m.update(Kilograms(10), t, u)
+      m2 <- m1.update(Kilograms(16), t, u)
+    } yield m2.value
+    q.value shouldBe(16)
+  }}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/MeanSpec.scala

@@ -0,0 +1,24 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class MeanSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val m = Mean(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      m1 <- m.update(Kilograms(10), t, u)
+      m2 <- m1.update(Kilograms(20), t, u)
+      m3 <- m2.update(Kilograms(75), t, u)
+    } yield m3.value
+    q.value shouldBe(35)
+  }
+
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/MinSpec.scala

@@ -0,0 +1,21 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class MinSpec extends BasePropSpec {
+  property("updateQuantity property") {
+    val m = Min(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      m1 <- m.update(Kilograms(10), t, u)
+      m2 <- m1.update(Kilograms(8), t, u)
+    } yield m2.value
+    q.value shouldBe(8)
+  }
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/RangeSpec.scala

@@ -0,0 +1,22 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class RangeSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val q = for {
+      r <- Some(Range(Mass.name, Kilograms.symbol))
+      r2 <- r.update(Kilograms(10), Instant.now(), UnitMapper.defaultUnitMapper)
+      r3 <- r2.update(Kilograms(30), Instant.now(), UnitMapper.defaultUnitMapper)
+      r4 <- r2.update(Kilograms(37), Instant.now(), UnitMapper.defaultUnitMapper)
+    } yield r4.labeledValue
+    q.value shouldBe("Range: 27.000000 kg")
+  }
+
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/StandardDeviationSpec.scala

@@ -0,0 +1,25 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class StandardDeviationSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val s = StandardDeviation(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      s1 <- s.update(Kilograms(10), t, u)
+      s2 <- s1.update(Kilograms(20), t, u)
+      s3 <- s2.update(Kilograms(30), t, u)
+      s4 <- s3.update(Kilograms(40), t, u)
+    } yield s4.value
+    q.value shouldBe(Math.sqrt(166 + 2d/3))
+  }
+
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/SumOfSquaredDeviationsSpec.scala

@@ -0,0 +1,22 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import squants.Kilograms
+import squants.mass.Mass
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.model.UnitMapper
+
+class SumOfSquaredDeviationsSpec extends BasePropSpec {
+  property("updateQuantity property") {
+    val s = SumOfSquaredDeviations(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      s1 <- s.update(Kilograms(10), t, u)
+      s2 <- s1.update(Kilograms(20), t, u)
+      s3 <- s2.update(Kilograms(30), t, u)
+    } yield s3.value
+    q.value shouldBe(200)
+  }
+}

src/test/scala/io/epiphanous/flinkrunner/model/aggregate/VarianceSpec.scala

@@ -0,0 +1,25 @@
+package io.epiphanous.flinkrunner.model.aggregate
+
+import java.time.Instant
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import io.epiphanous.flinkrunner.model.UnitMapper
+import squants.Kilograms
+import squants.mass.Mass
+
+class VarianceSpec extends BasePropSpec {
+
+  property("updateQuantity property") {
+    val v = Variance(Mass.name, Kilograms.symbol)
+    val t = Instant.now()
+    val u = UnitMapper.defaultUnitMapper
+    val q = for {
+      v1 <- v.update(Kilograms(10), t, u)
+      v2 <- v1.update(Kilograms(20), t, u)
+      v3 <- v2.update(Kilograms(30), t, u)
+      v4 <- v3.update(Kilograms(40), t, u)
+    } yield v4.value
+    q.value shouldBe(166 + 2d/3)
+  }
+
+}