Add consumtion and production operations to Formula Engine (#746)

In order to simplify usability but don't loose functionality we decided
to remove the consumption and production power formulas, i.e. the
formulas that are having a production or consumption attached to their
name, and replace them with operators that can be called on a formula
engine.

The consumption operator returns either the identity if the power value
is
positive or 0 and the production operator returns either the identity if
the power value is
negative or 0.

As an example we can now create the following formula:

```python
grid_consumption = logical_meter.grid_power().consumption().build()
```

This would return the consumption part of the grid_power stream.

Removing the old consumption and production power formulas will happen
in the follow up PR
#697.

Author: matthias-wende-frequenz

RELEASE_NOTES.md

@@ -62,6 +62,8 @@ This version ships an experimental version of the **Power Manager**, adds prelim
 
 - Allow configuration of the `resend_latest` flag in channels owned by the `ChannelRegistry`.
 
+- Add consumption and production operators that will replace the logical meters production and consumption function variants.
+
 ## Bug Fixes
 
 - Fix rendering of diagrams in the documentation.

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

@@ -25,13 +25,15 @@
     Adder,
     Clipper,
     ConstantValue,
+    Consumption,
     Divider,
     FormulaStep,
     Maximizer,
     MetricFetcher,
     Minimizer,
     Multiplier,
     OpenParen,
+    Production,
     Subtractor,
 )
 from ._tokenizer import TokenType
@@ -41,12 +43,14 @@
 _operator_precedence = {
     "max": 0,
     "min": 1,
-    "(": 2,
-    "/": 3,
-    "*": 4,
-    "-": 5,
-    "+": 6,
-    ")": 7,
+    "consumption": 2,
+    "production": 3,
+    "(": 4,
+    "/": 5,
+    "*": 6,
+    "-": 7,
+    "+": 8,
+    ")": 9,
 }
 """The dictionary of operator precedence for the shunting yard algorithm."""
 
@@ -188,12 +192,31 @@ def min(
             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
 
+    def consumption(self) -> _GenericHigherOrderBuilder:
+        """
+        Return a formula builder that applies the consumption operator on `self`.
+
+        The consumption operator returns either the identity if the power value is
+        positive or 0.
+        """
+        return self._higher_order_builder(self, self._create_method).consumption()  # type: ignore
+
+    def production(self) -> _GenericHigherOrderBuilder:
+        """
+        Return a formula builder that applies the production operator on `self`.
+
+        The production operator returns either the absolute value if the power value is
+        negative or 0.
+        """
+        return self._higher_order_builder(self, self._create_method).production()  # type: ignore
+
 
 class FormulaEngine(
     Generic[QuantityT],
@@ -574,7 +597,7 @@ def __init__(self, name: str, create_method: Callable[[float], QuantityT]) -> No
         self._steps: list[FormulaStep] = []
         self._metric_fetchers: dict[str, MetricFetcher[QuantityT]] = {}
 
-    def push_oper(self, oper: str) -> None:
+    def push_oper(self, oper: str) -> None:  # pylint: disable=too-many-branches
         """Push an operator into the engine.
 
         Args:
@@ -608,6 +631,10 @@ def push_oper(self, oper: str) -> None:
             self._build_stack.append(Maximizer())
         elif oper == "min":
             self._build_stack.append(Minimizer())
+        elif oper == "consumption":
+            self._build_stack.append(Consumption())
+        elif oper == "production":
+            self._build_stack.append(Production())
 
     def push_metric(
         self,
@@ -996,6 +1023,52 @@ def min(
         """
         return self._push("min", other)
 
+    def consumption(
+        self,
+    ) -> (
+        HigherOrderFormulaBuilder[QuantityT]
+        | HigherOrderFormulaBuilder3Phase[QuantityT]
+    ):
+        """Apply the Consumption Operator.
+
+        The consumption operator returns either the identity if the power value is
+        positive or 0.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        self._steps.appendleft((TokenType.OPER, "("))
+        self._steps.append((TokenType.OPER, ")"))
+        self._steps.append((TokenType.OPER, "consumption"))
+        assert isinstance(
+            self, (HigherOrderFormulaBuilder, HigherOrderFormulaBuilder3Phase)
+        )
+        return self
+
+    def production(
+        self,
+    ) -> (
+        HigherOrderFormulaBuilder[QuantityT]
+        | HigherOrderFormulaBuilder3Phase[QuantityT]
+    ):
+        """Apply the Production Operator.
+
+        The production operator returns either the absolute value if the power value is
+        negative or 0.
+
+        Returns:
+            A formula builder that can take further expressions, or can be built
+                into a formula engine.
+        """
+        self._steps.appendleft((TokenType.OPER, "("))
+        self._steps.append((TokenType.OPER, ")"))
+        self._steps.append((TokenType.OPER, "production"))
+        assert isinstance(
+            self, (HigherOrderFormulaBuilder, HigherOrderFormulaBuilder3Phase)
+        )
+        return self
+
 
 class HigherOrderFormulaBuilder(Generic[QuantityT], _BaseHOFormulaBuilder[QuantityT]):
     """A specialization of the _BaseHOFormulaBuilder for `FormulaReceiver`."""

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

@@ -177,6 +177,64 @@ def apply(self, eval_stack: list[float]) -> None:
         eval_stack.append(res)
 
 
+class Consumption(FormulaStep):
+    """A formula step that represents the consumption operator.
+
+    The consumption operator is the maximum of the value on top
+    of the evaluation stack and 0.
+    """
+
+    def __repr__(self) -> str:
+        """Return a string representation of the step.
+
+        Returns:
+            A string representation of the step.
+        """
+        return "consumption"
+
+    def apply(self, eval_stack: list[float]) -> None:
+        """
+        Apply the consumption formula.
+
+        Replace the top of the eval eval_stack with the same value if the value
+        is positive or 0.
+
+        Args:
+            eval_stack: An evaluation stack, to apply the formula step on.
+        """
+        val = eval_stack.pop()
+        eval_stack.append(max(val, 0))
+
+
+class Production(FormulaStep):
+    """A formula step that represents the production operator.
+
+    The production operator is the maximum of the value times minus one on top
+    of the evaluation stack and 0.
+    """
+
+    def __repr__(self) -> str:
+        """Return a string representation of the step.
+
+        Returns:
+            A string representation of the step.
+        """
+        return "production"
+
+    def apply(self, eval_stack: list[float]) -> None:
+        """
+        Apply the production formula.
+
+        Replace the top of the eval eval_stack with its absolute value if the
+        value is negative or 0.
+
+        Args:
+            eval_stack: An evaluation stack, to apply the formula step on.
+        """
+        val = eval_stack.pop()
+        eval_stack.append(max(-val, 0))
+
+
 class OpenParen(FormulaStep):
     """A no-op formula step used while building a prefix formula engine.
 

tests/timeseries/test_formula_engine.py

@@ -318,6 +318,19 @@ async def run_test(  # pylint: disable=too-many-locals
         num_items: int,
         make_builder: (
             Callable[
+                [
+                    FormulaEngine[Quantity],
+                ],
+                HigherOrderFormulaBuilder[Quantity],
+            ]
+            | Callable[
+                [
+                    FormulaEngine[Quantity],
+                    FormulaEngine[Quantity],
+                ],
+                HigherOrderFormulaBuilder[Quantity],
+            ]
+            | Callable[
                 [
                     FormulaEngine[Quantity],
                     FormulaEngine[Quantity],
@@ -586,6 +599,95 @@ async def test_compound(self) -> None:
             ],
         )
 
+    async def test_consumption(self) -> None:
+        """Test the consumption operator."""
+        await self.run_test(
+            1,
+            lambda c1: c1.consumption(),
+            [
+                ([10.0], 10.0),
+                ([-10.0], 0.0),
+            ],
+        )
+
+    async def test_production(self) -> None:
+        """Test the production operator."""
+        await self.run_test(
+            1,
+            lambda c1: c1.production(),
+            [
+                ([10.0], 0.0),
+                ([-10.0], 10.0),
+            ],
+        )
+
+    async def test_consumption_production(self) -> None:
+        """Test the consumption and production operator combined."""
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.consumption() + c2.production(),
+            [
+                ([10.0, 12.0], 10.0),
+                ([-12.0, -10.0], 10.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.consumption() + c2.consumption(),
+            [
+                ([10.0, -12.0], 10.0),
+                ([-10.0, 12.0], 12.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.production() + c2.production(),
+            [
+                ([10.0, -12.0], 12.0),
+                ([-10.0, 12.0], 10.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.min(c2).consumption(),
+            [
+                ([10.0, -12.0], 0.0),
+                ([10.0, 12.0], 10.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.max(c2).consumption(),
+            [
+                ([10.0, -12.0], 10.0),
+                ([-10.0, -12.0], 0.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.min(c2).production(),
+            [
+                ([10.0, -12.0], 12.0),
+                ([10.0, 12.0], 0.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.max(c2).production(),
+            [
+                ([10.0, -12.0], 0.0),
+                ([-10.0, -12.0], 10.0),
+            ],
+        )
+        await self.run_test(
+            2,
+            lambda c1, c2: c1.production() + c2,
+            [
+                ([10.0, -12.0], -12.0),
+                ([-10.0, -12.0], -2.0),
+            ],
+        )
+
     async def test_nones_are_zeros(self) -> None:
         """Test that `None`s are treated as zeros when configured."""
         await self.run_test(