Add min and max operations to formula engine (#561)

Author: matthias-wende-frequenz

RELEASE_NOTES.md

@@ -102,6 +102,12 @@ This release replaces the `@actor` decorator with a new `Actor` class.
 
 - `Actor`: This new class inherits from `BackgroundService` and it replaces the `@actor` decorator.
 
+- Newly added `min` and `max` functions for Formulas. They can be used as follows:
+
+    ```python
+    formula1.min(formula2)
+    ```
+
 ## Bug Fixes
 
 - Fixes a bug in the ring buffer updating the end timestamp of gaps when they are outdated.

src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py

@@ -35,7 +35,9 @@
     ConstantValue,
     Divider,
     FormulaStep,
+    Maximizer,
     MetricFetcher,
+    Minimizer,
     Multiplier,
     OpenParen,
     Subtractor,
@@ -45,12 +47,14 @@
 _logger = logging.Logger(__name__)
 
 _operator_precedence = {
-    "(": 0,
-    "/": 1,
-    "*": 2,
-    "-": 3,
-    "+": 4,
-    ")": 5,
+    "max": 0,
+    "min": 1,
+    "(": 2,
+    "/": 3,
+    "*": 4,
+    "-": 5,
+    "+": 6,
+    ")": 7,
 }
 
 
@@ -168,6 +172,36 @@ def __truediv__(
         """
         return self._higher_order_builder(self, self._create_method) / other  # type: ignore
 
+    def _max(
+        self, other: _GenericEngine | _GenericHigherOrderBuilder | QuantityT
+    ) -> _GenericHigherOrderBuilder:
+        """Return a formula engine that outputs the maximum of `self` and `other`.
+
+        Args:
+            other: A formula receiver, a formula builder or a QuantityT instance
+                corresponding to a sub-expression.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        return self._higher_order_builder(self, self._create_method).max(other)  # type: ignore
+
+    def _min(
+        self, other: _GenericEngine | _GenericHigherOrderBuilder | QuantityT
+    ) -> _GenericHigherOrderBuilder:
+        """Return a formula engine that outputs the minimum of `self` and `other`.
+
+        Args:
+            other: A formula receiver, a formula builder or a QuantityT instance
+                corresponding to a sub-expression.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        return self._higher_order_builder(self, self._create_method).min(other)  # type: ignore
+
 
 class FormulaEngine(
     Generic[QuantityT],
@@ -467,6 +501,10 @@ def push_oper(self, oper: str) -> None:
             self._build_stack.append(Divider())
         elif oper == "(":
             self._build_stack.append(OpenParen())
+        elif oper == "max":
+            self._build_stack.append(Maximizer())
+        elif oper == "min":
+            self._build_stack.append(Minimizer())
 
     def push_metric(
         self,
@@ -653,15 +691,15 @@ def _push(
         self._steps.append((TokenType.OPER, ")"))
         self._steps.append((TokenType.OPER, oper))
 
-        # pylint: disable=protected-access
         if isinstance(other, (FormulaEngine, FormulaEngine3Phase)):
             self._steps.append((TokenType.COMPONENT_METRIC, other))
-        elif isinstance(other, (Quantity, float)):
+        elif isinstance(other, (Quantity, float, int)):
             match oper:
-                case "+" | "-":
+                case "+" | "-" | "max" | "min":
                     if not isinstance(other, Quantity):
                         raise RuntimeError(
-                            f"A Quantity must be provided for addition or subtraction to {other}"
+                            "A Quantity must be provided for addition,"
+                            f" subtraction, min or max to {other}"
                         )
                 case "*" | "/":
                     if not isinstance(other, (float, int)):
@@ -671,9 +709,8 @@ def _push(
             self._steps.append((TokenType.CONSTANT, other))
         elif isinstance(other, _BaseHOFormulaBuilder):
             self._steps.append((TokenType.OPER, "("))
-            self._steps.extend(other._steps)
+            self._steps.extend(other._steps)  # pylint: disable=protected-access
             self._steps.append((TokenType.OPER, ")"))
-        # pylint: enable=protected-access
         else:
             raise RuntimeError(f"Can't build a formula from: {other}")
         assert isinstance(
@@ -804,6 +841,66 @@ def __truediv__(
         """
         return self._push("/", other)
 
+    @overload
+    def max(
+        self, other: _CompositionType1Phase
+    ) -> HigherOrderFormulaBuilder[QuantityT]:
+        ...
+
+    @overload
+    def max(
+        self, other: _CompositionType3Phase | QuantityT
+    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
+        ...
+
+    def max(
+        self, other: _CompositionType | QuantityT
+    ) -> (
+        HigherOrderFormulaBuilder[QuantityT]
+        | HigherOrderFormulaBuilder3Phase[QuantityT]
+    ):
+        """Return a formula builder that calculates the maximum of `self` and `other`.
+
+        Args:
+            other: A formula receiver, or a formula builder instance corresponding to a
+                sub-expression.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        return self._push("max", other)
+
+    @overload
+    def min(
+        self, other: _CompositionType1Phase
+    ) -> HigherOrderFormulaBuilder[QuantityT]:
+        ...
+
+    @overload
+    def min(
+        self, other: _CompositionType3Phase | QuantityT
+    ) -> HigherOrderFormulaBuilder3Phase[QuantityT]:
+        ...
+
+    def min(
+        self, other: _CompositionType | QuantityT
+    ) -> (
+        HigherOrderFormulaBuilder[QuantityT]
+        | HigherOrderFormulaBuilder3Phase[QuantityT]
+    ):
+        """Return a formula builder that calculates the minimum of `self` and `other`.
+
+        Args:
+            other: A formula receiver, or a formula builder instance corresponding to a
+                sub-expression.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        return self._push("min", other)
+
 
 class HigherOrderFormulaBuilder(Generic[QuantityT], _BaseHOFormulaBuilder[QuantityT]):
     """A specialization of the _BaseHOFormulaBuilder for `FormulaReceiver`."""

src/frequenz/sdk/timeseries/_formula_engine/_formula_steps.py

@@ -132,6 +132,52 @@ def apply(self, eval_stack: List[float]) -> None:
         eval_stack.append(res)
 
 
+class Maximizer(FormulaStep):
+    """A formula step that represents the max function."""
+
+    def __repr__(self) -> str:
+        """Return a string representation of the step.
+
+        Returns:
+            A string representation of the step.
+        """
+        return "max"
+
+    def apply(self, eval_stack: List[float]) -> None:
+        """Extract two values from the stack and pushes back the maximum.
+
+        Args:
+            eval_stack: An evaluation stack, to apply the formula step on.
+        """
+        val2 = eval_stack.pop()
+        val1 = eval_stack.pop()
+        res = max(val1, val2)
+        eval_stack.append(res)
+
+
+class Minimizer(FormulaStep):
+    """A formula step that represents the min function."""
+
+    def __repr__(self) -> str:
+        """Return a string representation of the step.
+
+        Returns:
+            A string representation of the step.
+        """
+        return "min"
+
+    def apply(self, eval_stack: List[float]) -> None:
+        """Extract two values from the stack and pushes back the minimum.
+
+        Args:
+            eval_stack: An evaluation stack, to apply the formula step on.
+        """
+        val2 = eval_stack.pop()
+        val1 = eval_stack.pop()
+        res = min(val1, val2)
+        eval_stack.append(res)
+
+
 class OpenParen(FormulaStep):
     """A no-op formula step used while building a prefix formula engine.
 

tests/timeseries/_formula_engine/test_formula_composition.py

@@ -170,6 +170,51 @@ async def test_formula_composition_missing_bat(self, mocker: MockerFixture) -> N
 
         assert count == 10
 
+    async def test_formula_composition_min_max(self, mocker: MockerFixture) -> None:
+        """Test the composition of formulas with min/max values."""
+        mockgrid = MockMicrogrid(grid_meter=True)
+        await mockgrid.start(mocker)
+
+        logical_meter = microgrid.logical_meter()
+        engine_min = logical_meter.grid_power._min(  # pylint: disable=protected-access
+            Power.zero()
+        ).build("grid_power_min")
+        engine_min_rx = engine_min.new_receiver()
+        engine_max = logical_meter.grid_power._max(  # pylint: disable=protected-access
+            Power.zero()
+        ).build("grid_power_max")
+        engine_max_rx = engine_max.new_receiver()
+
+        await mockgrid.mock_resampler.send_meter_power([100.0])
+
+        # Test min
+        min_pow = await engine_min_rx.receive()
+        assert min_pow and min_pow.value and min_pow.value.isclose(Power.zero())
+
+        # Test max
+        max_pow = await engine_max_rx.receive()
+        assert (
+            max_pow and max_pow.value and max_pow.value.isclose(Power.from_watts(100.0))
+        )
+
+        await mockgrid.mock_resampler.send_meter_power([-100.0])
+
+        # Test min
+        min_pow = await engine_min_rx.receive()
+        assert (
+            min_pow
+            and min_pow.value
+            and min_pow.value.isclose(Power.from_watts(-100.0))
+        )
+
+        # Test max
+        max_pow = await engine_max_rx.receive()
+        assert max_pow and max_pow.value and max_pow.value.isclose(Power.zero())
+
+        await engine_min._stop()  # pylint: disable=protected-access
+        await mockgrid.cleanup()
+        await logical_meter.stop()
+
     async def test_formula_composition_constant(self, mocker: MockerFixture) -> None:
         """Test the composition of formulas with constant values."""
         mockgrid = MockMicrogrid(grid_meter=True)