Add pretty-printing FormulaFormatter

Signed-off-by: Christian Parpart <[email protected]>

Author: christianparpart

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

@@ -4,11 +4,13 @@
 """A formula engine for applying formulas."""
 from ._formula_engine import FormulaEngine, FormulaEngine3Phase
 from ._formula_engine_pool import FormulaEnginePool
+from ._formula_formatter import FormulaFormatter
 from ._resampled_formula_builder import ResampledFormulaBuilder
 
 __all__ = [
     "FormulaEngine",
     "FormulaEngine3Phase",
     "FormulaEnginePool",
+    "FormulaFormatter",
     "ResampledFormulaBuilder",
 ]

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

@@ -1,6 +1,8 @@
 # License: MIT
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 
+# pylint: disable=too-many-lines
+
 """A formula engine that can apply formulas on streaming data."""
 
 from __future__ import annotations
@@ -28,6 +30,7 @@
 from .. import Sample, Sample3Phase
 from .._quantities import Quantity, QuantityT
 from ._formula_evaluator import FormulaEvaluator
+from ._formula_formatter import FormulaFormatter
 from ._formula_steps import (
     Adder,
     Averager,
@@ -314,6 +317,19 @@ async def _run(self) -> None:
             else:
                 await sender.send(msg)
 
+    def __str__(self) -> str:
+        """Return a string representation of the formula.
+
+        Returns:
+            A string representation of the formula.
+        """
+        steps = (
+            self._builder._build_stack
+            if len(self._builder._build_stack) > 0
+            else self._builder._steps
+        )
+        return FormulaFormatter.format(steps)
+
     def new_receiver(
         self, name: Optional[str] = None, max_size: int = 50
     ) -> Receiver[Sample[QuantityT]]:
@@ -629,6 +645,15 @@ def finalize(
 
         return self._steps, self._metric_fetchers
 
+    def __str__(self) -> str:
+        """Return a string representation of the formula.
+
+        Returns:
+            A string representation of the formula.
+        """
+        steps = self._steps if len(self._steps) > 0 else self._build_stack
+        return FormulaFormatter.format(steps)
+
     def build(self) -> FormulaEngine[QuantityT]:
         """Create a formula engine with the steps and fetchers that have been pushed.
 

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

@@ -0,0 +1,239 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Formatter for the formula."""
+
+from __future__ import annotations
+
+import enum
+
+from ._formula_steps import (
+    Adder,
+    Averager,
+    Clipper,
+    ConstantValue,
+    Divider,
+    FormulaStep,
+    Maximizer,
+    MetricFetcher,
+    Minimizer,
+    Multiplier,
+    OpenParen,
+    Subtractor,
+)
+
+
+class OperatorPrecedence(enum.Enum):
+    """The precedence of an operator."""
+
+    SUBTRACTION = 1
+    ADDITION = 2
+    DIVISION = 3
+    MULTIPLICATION = 4
+    PRIMARY = 9
+
+    def __lt__(self, other: OperatorPrecedence) -> bool:
+        """Test the precedence of this operator is less than the precedence of the other operator.
+
+        Args:
+            other: The other operator (on the right-hand side).
+
+        Returns:
+            Whether the precedence of this operator is less than the other operator.
+        """
+        return self.value < other.value
+
+    def __str__(self) -> str:
+        """Return the string representation of the operator precedence.
+
+        Returns:
+            The string representation of the operator precedence.
+        """
+        match self:
+            case OperatorPrecedence.SUBTRACTION:
+                return "-"
+            case OperatorPrecedence.ADDITION:
+                return "+"
+            case OperatorPrecedence.DIVISION:
+                return "/"
+            case OperatorPrecedence.MULTIPLICATION:
+                return "*"
+            case OperatorPrecedence.PRIMARY:
+                return "primary"
+
+
+class StackItem:
+    """Stack item for the formula formatter."""
+
+    def __init__(self, value: str, precedence: OperatorPrecedence):
+        """Initialize the StackItem.
+
+        Args:
+            value: The value of the stack item.
+            precedence: The precedence of the stack item.
+        """
+        self.value = value
+        self.precedence = precedence
+
+    def __str__(self) -> str:
+        """Return the string representation of the stack item.
+
+        This is used for debugging purposes.
+
+        Returns:
+            str: The string representation of the stack item.
+        """
+        return f'("{self.value}", {self.precedence})'
+
+    def as_left_value(self, outer_precedence: OperatorPrecedence) -> str:
+        """Return the value of the stack item with parentheses if necessary.
+
+        Args:
+            outer_precedence: The precedence of the outer stack item.
+
+        Returns:
+            str: The value of the stack item with parentheses if necessary.
+        """
+        return f"({self.value})" if self.precedence < outer_precedence else self.value
+
+    def as_right_value(self, outer_precedence: OperatorPrecedence) -> str:
+        """Return the value of the stack item with parentheses if necessary.
+
+        Args:
+            outer_precedence: The precedence of the outer stack item.
+
+        Returns:
+            str: The value of the stack item with parentheses if necessary.
+        """
+        return f"({self.value})" if self.precedence < outer_precedence else self.value
+
+    @staticmethod
+    def create_binary(
+        lhs: StackItem, operator: OperatorPrecedence, rhs: StackItem
+    ) -> StackItem:
+        """Create a binary stack item.
+
+        Args:
+            lhs: The left-hand side of the binary operation.
+            operator: The operator of the binary operation.
+            rhs: The right-hand side of the binary operation.
+
+        Returns:
+            StackItem: The binary stack item.
+        """
+        pred = OperatorPrecedence(operator)
+        return StackItem(
+            f"{lhs.as_left_value(pred)} {operator} {rhs.as_right_value(pred)}",
+            pred,
+        )
+
+    @staticmethod
+    def create_primary(value: float) -> StackItem:
+        """Create a stack item for literal values or function calls (primary expressions).
+
+        Args:
+            value: The value of the literal.
+
+        Returns:
+            StackItem: The literal stack item.
+        """
+        return StackItem(str(value), OperatorPrecedence.PRIMARY)
+
+
+class FormulaFormatter:
+    """Formats a formula into a human readable string in infix-notation."""
+
+    def __init__(self) -> None:
+        """Initialize the FormulaFormatter."""
+        self._stack = list[StackItem]()
+
+    @classmethod
+    def format(cls, postfix_expr: list[FormulaStep]) -> str:
+        """Return the formula as a string in infix notation.
+
+        Args:
+            postfix_expr: The steps of the formula in postfix notation order.
+
+        Returns:
+            str: The formula in infix notation.
+        """
+        formatter = FormulaFormatter()
+        return formatter._format(postfix_expr)
+
+    def _format(self, postfix_expr: list[FormulaStep]) -> str:
+        """Format the postfix expression to infix notation.
+
+        Args:
+            postfix_expr: The steps of the formula in postfix notation order.
+
+        Returns:
+            str: The formula in infix notation.
+        """
+        for step in postfix_expr:
+            match step:
+                case ConstantValue():
+                    self._stack.append(StackItem.create_primary(step.value))
+                case Adder():
+                    self._format_binary(OperatorPrecedence.ADDITION)
+                case Subtractor():
+                    self._format_binary(OperatorPrecedence.SUBTRACTION)
+                case Multiplier():
+                    self._format_binary(OperatorPrecedence.MULTIPLICATION)
+                case Divider():
+                    self._format_binary(OperatorPrecedence.DIVISION)
+                case Averager():
+                    value = (
+                        # pylint: disable=protected-access
+                        f"avg({', '.join(self._format([f]) for f in step.fetchers)})"
+                    )
+                    self._stack.append(StackItem(value, OperatorPrecedence.PRIMARY))
+                case Clipper():
+                    the_value = self._stack.pop()
+                    min_value = step.min_value if step.min_value is not None else "-inf"
+                    max_value = step.max_value if step.max_value is not None else "inf"
+                    value = f"clip({min_value}, {the_value.value}, {max_value})"
+                    self._stack.append(StackItem(value, OperatorPrecedence.PRIMARY))
+                case Maximizer():
+                    left, right = self._pop_two_from_stack()
+                    value = f"max({left.value}, {right.value})"
+                    self._stack.append(StackItem(value, OperatorPrecedence.PRIMARY))
+                case Minimizer():
+                    left, right = self._pop_two_from_stack()
+                    value = f"min({left.value}, {right.value})"
+                    self._stack.append(StackItem(value, OperatorPrecedence.PRIMARY))
+                case MetricFetcher():
+                    metric_fetcher = step
+                    if engine_reference := getattr(
+                        metric_fetcher.stream, "_engine_reference", None
+                    ):
+                        value = str(engine_reference)
+                    else:
+                        value = metric_fetcher._name  # pylint: disable=protected-access
+                    self._stack.append(StackItem(value, OperatorPrecedence.PRIMARY))
+                case OpenParen():
+                    pass  # We gently ignore this one.
+
+        assert len(self._stack) == 1
+        return self._stack[0].value
+
+    def _format_binary(self, operator: OperatorPrecedence) -> None:
+        """Format a binary operation.
+
+        Pops the arguments of the binary expression from the stack
+        and pushes the string representation of the binary operation to the stack.
+
+        Args:
+            operator: The operator of the binary operation.
+        """
+        left, right = self._pop_two_from_stack()
+        self._stack.append(StackItem.create_binary(left, operator, right))
+
+    def _pop_two_from_stack(self) -> tuple[StackItem, StackItem]:
+        """Pop two items from the stack.
+
+        Returns:
+            The two items popped from the stack.
+        """
+        right = self._stack.pop()
+        left = self._stack.pop()
+        return left, right

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

@@ -7,7 +7,7 @@
 
 import math
 from abc import ABC, abstractmethod
-from typing import Generic, List, Optional
+from typing import Generic, List, Optional, Sequence
 
 from frequenz.channels import Receiver
 
@@ -207,6 +207,15 @@ def __init__(self, fetchers: List[MetricFetcher[QuantityT]]) -> None:
         """
         self._fetchers: list[MetricFetcher[QuantityT]] = fetchers
 
+    @property
+    def fetchers(self) -> Sequence[MetricFetcher[QuantityT]]:
+        """Return the metric fetchers.
+
+        Returns:
+            The metric fetchers.
+        """
+        return self._fetchers
+
     def __repr__(self) -> str:
         """Return a string representation of the step.
 
@@ -255,6 +264,15 @@ def __init__(self, value: float) -> None:
         """
         self._value = value
 
+    @property
+    def value(self) -> float:
+        """Return the constant value.
+
+        Returns:
+            The constant value.
+        """
+        return self._value
+
     def __repr__(self) -> str:
         """Return a string representation of the step.
 
@@ -285,6 +303,24 @@ def __init__(self, min_val: float | None, max_val: float | None) -> None:
         self._min_val = min_val
         self._max_val = max_val
 
+    @property
+    def min_value(self) -> float | None:
+        """Return the minimum value.
+
+        Returns:
+            The minimum value.
+        """
+        return self._min_val
+
+    @property
+    def max_value(self) -> float | None:
+        """Return the maximum value.
+
+        Returns:
+            The maximum value.
+        """
+        return self._max_val
+
     def __repr__(self) -> str:
         """Return a string representation of the step.
 
@@ -329,6 +365,23 @@ def __init__(
         self._next_value: Optional[Sample[QuantityT]] = None
         self._nones_are_zeros = nones_are_zeros
 
+    @property
+    def stream(self) -> Receiver[Sample[QuantityT]]:
+        """Return the stream from which to fetch values.
+
+        Returns:
+            The stream from which to fetch values.
+        """
+        return self._stream
+
+    def stream_name(self) -> str:
+        """Return the name of the stream.
+
+        Returns:
+            The name of the stream.
+        """
+        return str(self._stream.__doc__)
+
     async def fetch_next(self) -> Optional[Sample[QuantityT]]:
         """Fetch the next value from the stream.