[Bugfix] Only probes collected during stable MAD count

For calculating the final RPM, only those probes that are sent/received
during the stable MAD should count.

Signed-off-by: Will Hawkins <[email protected]>

1 files changed, 30 insertions, 2 deletions

diff --git a/stabilizer/algorithm.go b/stabilizer/algorithm.go
index 86b4bb6..42dc0c6 100644
--- a/stabilizer/algorithm.go
+++ b/stabilizer/algorithm.go
@@ -24,7 +24,7 @@ import (
 	"golang.org/x/exp/constraints"
 )
 
-type MeasurementStablizer[Data constraints.Float | constraints.Integer, Bucket constraints.Ordered] struct {
+type MeasurementStablizer[Data constraints.Float | constraints.Integer, Bucket utilities.Number] struct {
 	// The number of instantaneous measurements in the current interval could be infinite (Forever).
 	instantaneousses series.WindowSeries[Data, Bucket]
 	// There are a fixed, finite number of aggregates (WindowOnly).
@@ -58,7 +58,7 @@ type MeasurementStablizer[Data constraints.Float | constraints.Integer, Bucket c
 // If the calculated standard deviation of *those* values is less than SDT, we declare
 // stability.
 
-func NewStabilizer[Data constraints.Float | constraints.Integer, Bucket constraints.Ordered](
+func NewStabilizer[Data constraints.Float | constraints.Integer, Bucket utilities.Number](
 	mad int,
 	sdt float64,
 	trimmingLevel uint,
@@ -199,6 +199,7 @@ func (r3 *MeasurementStablizer[Data, Bucket]) IsStable() bool {
 	r3.aggregates.ForEach(func(b int, md *utilities.Optional[series.WindowSeries[Data, Bucket]]) {
 		if utilities.IsSome[series.WindowSeries[Data, Bucket]](*md) {
 			md := utilities.GetSome[series.WindowSeries[Data, Bucket]](*md)
+
 			_, average := series.CalculateAverage(md)
 			averages = append(averages, average)
 		}
@@ -214,3 +215,30 @@ func (r3 *MeasurementStablizer[Data, Bucket]) IsStable() bool {
 
 	return isStable
 }
+
+func (r3 *MeasurementStablizer[Data, Bucket]) GetBounds() (Bucket, Bucket) {
+	r3.m.Lock()
+	defer r3.m.Unlock()
+
+	haveMinimum := false
+
+	lowerBound := Bucket(0)
+	upperBound := Bucket(0)
+
+	r3.aggregates.ForEach(func(b int, md *utilities.Optional[series.WindowSeries[Data, Bucket]]) {
+		if utilities.IsSome[series.WindowSeries[Data, Bucket]](*md) {
+			md := utilities.GetSome[series.WindowSeries[Data, Bucket]](*md)
+			currentAggregateLowerBound, currentAggregateUpperBound := md.GetBucketBounds()
+
+			if !haveMinimum {
+				lowerBound = currentAggregateLowerBound
+				haveMinimum = true
+			} else {
+				lowerBound = min(lowerBound, currentAggregateLowerBound)
+			}
+			upperBound = max(upperBound, currentAggregateUpperBound)
+		}
+	})
+
+	return lowerBound, upperBound
+}