From 6e7d8bc8628af2a6744f2c41e7e94ae7368b4d84 Mon Sep 17 00:00:00 2001
From: Alexander Zobnin <alexanderzobnin@gmail.com>
Date: Thu, 27 May 2021 16:55:00 +0300
Subject: [PATCH] Implement removeAboveValue/removeBelowValue

---
 pkg/datasource/functions.go           | 34 +++++++++++---
 pkg/timeseries/transform_functions.go | 68 ++++++++++++++-------------
 2 files changed, 63 insertions(+), 39 deletions(-)

diff --git a/pkg/datasource/functions.go b/pkg/datasource/functions.go
index 2af7a98..a6164cb 100644
--- a/pkg/datasource/functions.go
+++ b/pkg/datasource/functions.go
@@ -57,12 +57,14 @@ var frontendFuncMap map[string]bool
 
 func init() {
 	seriesFuncMap = map[string]DataProcessingFunc{
-		"groupBy":       applyGroupBy,
-		"scale":         applyScale,
-		"offset":        applyOffset,
-		"transformNull": applyTransformNull,
-		"percentile":    applyPercentile,
-		"timeShift":     applyTimeShiftPost,
+		"groupBy":          applyGroupBy,
+		"scale":            applyScale,
+		"offset":           applyOffset,
+		"removeAboveValue": applyRemoveAboveValue,
+		"removeBelowValue": applyRemoveBelowValue,
+		"transformNull":    applyTransformNull,
+		"percentile":       applyPercentile,
+		"timeShift":        applyTimeShiftPost,
 	}
 
 	aggFuncMap = map[string]AggDataProcessingFunc{
@@ -203,6 +205,26 @@ func applyOffset(series timeseries.TimeSeries, params ...interface{}) (timeserie
 	return series.Transform(transformFunc), nil
 }
 
+func applyRemoveAboveValue(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
+	threshold, err := MustFloat64(params[0])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	transformFunc := timeseries.TransformRemoveAboveValue(threshold)
+	return series.Transform(transformFunc), nil
+}
+
+func applyRemoveBelowValue(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
+	threshold, err := MustFloat64(params[0])
+	if err != nil {
+		return nil, errParsingFunctionParam(err)
+	}
+
+	transformFunc := timeseries.TransformRemoveBelowValue(threshold)
+	return series.Transform(transformFunc), nil
+}
+
 func applyTransformNull(series timeseries.TimeSeries, params ...interface{}) (timeseries.TimeSeries, error) {
 	nullValue, err := MustFloat64(params[0])
 	if err != nil {
diff --git a/pkg/timeseries/transform_functions.go b/pkg/timeseries/transform_functions.go
index f7c0eb6..7d0ad35 100644
--- a/pkg/timeseries/transform_functions.go
+++ b/pkg/timeseries/transform_functions.go
@@ -4,53 +4,55 @@ import "time"
 
 func TransformScale(factor float64) TransformFunc {
 	return func(point TimePoint) TimePoint {
-		return transformScale(point, factor)
+		if point.Value != nil {
+			newValue := *point.Value * factor
+			point.Value = &newValue
+		}
+		return point
 	}
 }
 
 func TransformOffset(offset float64) TransformFunc {
 	return func(point TimePoint) TimePoint {
-		return transformOffset(point, offset)
+		if point.Value != nil {
+			newValue := *point.Value + offset
+			point.Value = &newValue
+		}
+		return point
 	}
 }
 
 func TransformNull(nullValue float64) TransformFunc {
 	return func(point TimePoint) TimePoint {
-		return transformNull(point, nullValue)
+		if point.Value == nil {
+			point.Value = &nullValue
+		}
+		return point
+	}
+}
+
+func TransformRemoveAboveValue(threshold float64) TransformFunc {
+	return func(point TimePoint) TimePoint {
+		if *point.Value > threshold {
+			point.Value = nil
+		}
+		return point
+	}
+}
+
+func TransformRemoveBelowValue(threshold float64) TransformFunc {
+	return func(point TimePoint) TimePoint {
+		if *point.Value < threshold {
+			point.Value = nil
+		}
+		return point
 	}
 }
 
 func TransformShiftTime(interval time.Duration) TransformFunc {
 	return func(point TimePoint) TimePoint {
-		return transformShiftTime(point, interval)
+		shiftedTime := point.Time.Add(interval)
+		point.Time = shiftedTime
+		return point
 	}
 }
-
-func transformScale(point TimePoint, factor float64) TimePoint {
-	if point.Value != nil {
-		newValue := *point.Value * factor
-		point.Value = &newValue
-	}
-	return point
-}
-
-func transformOffset(point TimePoint, offset float64) TimePoint {
-	if point.Value != nil {
-		newValue := *point.Value + offset
-		point.Value = &newValue
-	}
-	return point
-}
-
-func transformShiftTime(point TimePoint, interval time.Duration) TimePoint {
-	shiftedTime := point.Time.Add(interval)
-	point.Time = shiftedTime
-	return point
-}
-
-func transformNull(point TimePoint, nullValue float64) TimePoint {
-	if point.Value == nil {
-		point.Value = &nullValue
-	}
-	return point
-}