Implement Reduce operator and examples; enhance platformio.ini for new Scan example

Author: rzeldent

examples/Reduce/Reduce.cpp

@@ -13,24 +13,42 @@ using namespace Reactive;
 int values[] = { 0, 1, 4, 6, 2, 5, 7, 3, 5, 8 };
 int valuesLength = sizeof(values) / sizeof(values[0]);
 
+// Reduce function: Start with a float value of 10.0 and add 1.5 times the integer value
+float accumulateValues(float acc, int value) {
+    return acc + value * 1.5f;
+}
+
+// Action to print the final reduced value
+void printFinalResult(float x) {
+    Serial.print("Final reduced result: ");
+    Serial.println(x);
+}
+
+// Action when reduction is complete
+void printComplete() {
+    Serial.println("Reduction complete!");
+}
+
 void setup()
 {
 	Serial.begin(115200);
 	while (!Serial) delay(1);
+	
+	Serial.println("=== Reduce Operator Example ===");
+	Serial.println("Values: 0, 1, 4, 6, 2, 5, 7, 3, 5, 8");
+	Serial.println("Formula: acc + value * 1.5, starting with 10.0");
+	Serial.println();
 }
 
-// Start with a float value of 10.0 (optional) and add 1.5 times the (integer) value
-
 void loop()
 {
+	Serial.println("Running reduction...");
+	
+	// Using ReduceToFinal - only emits the final accumulated result
 	FromArray(values, valuesLength)
-    .Reduce<float>([](float acc, int value) {
-        return acc + value * 1.5f;
-    }, 10.0f)
-	.DoAndFinally(
-		[](float x) { Serial.println(x); },
-		[]() { Serial.println("No more items"); }
-	);
-
-	delay(2000);
+    .ReduceToFinal<float>(accumulateValues, 10.0f)
+	.DoAndFinally(printFinalResult, printComplete);
+
+	Serial.println();
+	delay(3000);
 }

examples/Scan/Scan.cpp

@@ -0,0 +1,108 @@
+/***************************************************
+Copyright (c) 2019 Luis Llamas
+(www.luisllamas.es)
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License
+ ****************************************************/
+ 
+#include "ReactiveArduinoLib.h"
+using namespace Reactive;
+
+int values[] = { 1, 2, 3, 4, 5 };
+int valuesLength = sizeof(values) / sizeof(values[0]);
+
+// Accumulator function for sum calculation
+int addValues(int acc, int value) {
+    return acc + value;
+}
+
+// Accumulator function for product calculation
+int multiplyValues(int acc, int value) {
+    return acc * value;
+}
+
+// Action to print intermediate scan results
+void printScanResult(int x) {
+    Serial.print("Scan result: ");
+    Serial.println(x);
+}
+
+// Action when scan is complete
+void printScanComplete() {
+    Serial.println("Scan complete!");
+    Serial.println();
+}
+
+// Example showing running average calculation
+float runningAverageValues[] = { 10.0, 20.0, 30.0, 40.0, 50.0 };
+int avgValuesLength = sizeof(runningAverageValues) / sizeof(runningAverageValues[0]);
+int valueCount = 0;
+
+// Running average accumulator
+float calculateRunningAverage(float acc, float value) {
+    valueCount++;
+    return ((acc * (valueCount - 1)) + value) / valueCount;
+}
+
+// Action to print running average
+void printRunningAverage(float x) {
+    Serial.print("Running average: ");
+    Serial.println(x, 2);
+}
+
+// Action when running average is complete
+void printAverageComplete() {
+    Serial.println("Running average complete!");
+    Serial.println();
+    valueCount = 0; // Reset for next iteration
+}
+
+void setup()
+{
+    Serial.begin(115200);
+    while (!Serial) delay(1);
+    
+    Serial.println("=== Scan Operator Example ===");
+    Serial.println("Scan emits intermediate accumulated results");
+    Serial.println("Unlike Reduce which only emits the final result");
+    Serial.println();
+}
+
+void loop()
+{
+    Serial.println("Example 1: Running Sum with Scan");
+    Serial.println("Values: 1, 2, 3, 4, 5 (starting with 0)");
+    
+    // Scan emits: 1, 3, 6, 10, 15
+    FromArray(values, valuesLength)
+    .Scan<int>(0, addValues)  // Start with 0, add each value
+    .DoAndFinally(printScanResult, printScanComplete);
+    
+    delay(1000);
+    
+    Serial.println("Example 2: Running Product with Scan");
+    Serial.println("Values: 1, 2, 3, 4, 5 (starting with 1)");
+    
+    // Scan emits: 1, 2, 6, 24, 120
+    FromArray(values, valuesLength)
+    .Scan<int>(1, multiplyValues)  // Start with 1, multiply each value
+    .DoAndFinally(printScanResult, printScanComplete);
+    
+    delay(1000);
+    
+    Serial.println("Example 3: Running Average with Scan");
+    Serial.println("Values: 10.0, 20.0, 30.0, 40.0, 50.0");
+    
+    // Reset counter for this example
+    valueCount = 0;
+    
+    // Scan emits running averages: 10.0, 15.0, 20.0, 25.0, 30.0
+    FromArray(runningAverageValues, avgValuesLength)
+    .Scan<float>(0.0f, calculateRunningAverage)
+    .DoAndFinally(printRunningAverage, printAverageComplete);
+    
+    delay(3000);
+    Serial.println("========================================");
+}

platformio.ini

@@ -53,6 +53,9 @@ build_src_filter = +<../examples/Reduce/>
 [env:rotary_encoder]
 build_src_filter = +<../examples/RotaryEncoder/>
 
+[env:scan]
+build_src_filter = +<../examples/Scan/>
+
 [env:throttle]
 build_src_filter = +<../examples/Throttle/>
 

src/Operators/OperatorReduce.h

@@ -0,0 +1,69 @@
+/***************************************************
+Copyright (c) 2019 Luis Llamas
+(www.luisllamas.es)
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License
+ ****************************************************/
+
+#ifndef _REACTIVEOPERATORREDUCE_h
+#define _REACTIVEOPERATORREDUCE_h
+
+template <typename Torig, typename Tdest>
+class OperatorReduce : public Operator<Torig, Tdest>
+{
+public:
+	typedef Tdest(*ReactiveReduceFunction)(Tdest, Torig);
+	
+	OperatorReduce(ReactiveReduceFunction function, Tdest initialValue);
+
+	void OnNext(Torig value) override;
+	void OnComplete() override;
+	void Reset() override;
+
+private:
+	ReactiveReduceFunction _function;
+	Tdest _accumulator;
+	Tdest _initialValue;
+	bool _hasValue;
+};
+
+template <typename Torig, typename Tdest>
+OperatorReduce<Torig, Tdest>::OperatorReduce(ReactiveReduceFunction function, Tdest initialValue)
+{
+	_function = function;
+	_initialValue = initialValue;
+	_accumulator = initialValue;
+	_hasValue = false;
+}
+
+template <typename Torig, typename Tdest>
+void OperatorReduce<Torig, Tdest>::OnNext(Torig value)
+{
+	// Accumulate the value but don't emit yet
+	_accumulator = _function(_accumulator, value);
+	_hasValue = true;
+}
+
+template <typename Torig, typename Tdest>
+void OperatorReduce<Torig, Tdest>::OnComplete()
+{
+	// Only emit the final accumulated result when the observable completes
+	if (_hasValue || _accumulator != _initialValue)
+	{
+		this->_childObservers.OnNext(_accumulator);
+	}
+	
+	// Forward the completion signal
+	Operator<Torig, Tdest>::OnComplete();
+}
+
+template <typename Torig, typename Tdest>
+void OperatorReduce<Torig, Tdest>::Reset()
+{
+	_accumulator = _initialValue;
+	_hasValue = false;
+}
+
+#endif

src/Operators/Operators.h

@@ -47,5 +47,6 @@ Unless required by applicable law or agreed to in writing, software distributed
 #include "OperatorStartWith.h"
 #include "OperatorDebounce.h"
 #include "OperatorSelect.h"
+#include "OperatorReduce.h"
 
 #endif

src/ReactiveArduinoCore.h

@@ -57,6 +57,7 @@ template <typename T> class OperatorWhere;
 template <typename T> class OperatorDistinct;
 template <typename T> class OperatorDistinctUntilChanged;
 template <typename Torig, typename Tdest> class OperatorSelect;
+template <typename Torig, typename Tdest> class OperatorReduce;
 template <typename T> class OperatorLast;
 template <typename T> class OperatorFirst;
 template <typename T> class OperatorTake;
@@ -201,6 +202,7 @@ class Observable : IObservable<T>, IResetable<T>
 	template <class Tdest> OperatorSelect<T, Tdest>& SelectTo(ReactiveMap<T, Tdest> selector);
 	template <class Tdest> TransformationMap<T, Tdest>& Map(ReactiveMap<T, Tdest> map);
 	template <class Tdest> TransformationReduce<T, Tdest>& Reduce(ReactiveReduce<T, Tdest> function, Tdest init = Tdest());
+	template <class Tdest> OperatorReduce<T, Tdest>& ReduceToFinal(ReactiveReduce<T, Tdest> function, Tdest init = Tdest());
 	TransformationUpperLimit<T>& LimitUpper(T upperLimit);
 	TransformationLowerLimit<T>& LimitLower(T lowerLimit);
 	TransformationLimit<T>& Limit(T lowerLimit, T upperLimit);
@@ -528,6 +530,15 @@ auto Observable<T>::Reduce(ReactiveReduce<T, Tdest> function, Tdest init) -> Tra
 	return *newOp;
 }
 
+template <typename T>
+template <typename Tdest>
+auto Observable<T>::ReduceToFinal(ReactiveReduce<T, Tdest> function, Tdest init) -> OperatorReduce<T, Tdest>&
+{
+	auto newOp = new OperatorReduce<T, Tdest>(function, init);
+	Compound(*this, *newOp);
+	return *newOp;
+}
+
 
 template <typename T>
 auto Observable<T>::LimitUpper(T upperLimit) -> TransformationUpperLimit<T>&

src/ReactiveArduinoLib.h

@@ -373,6 +373,12 @@ namespace Reactive
 	{
 		return *(new OperatorRepeat<T>(N));
 	}
+
+	template <typename Torig, typename Tdest>
+	OperatorReduce<Torig, Tdest>& ReduceToFinal(typename OperatorReduce<Torig, Tdest>::ReactiveReduceFunction function, Tdest init)
+	{
+		return *(new OperatorReduce<Torig, Tdest>(function, init));
+	}
 // #pragma endregion