Split the formula execution logic from the formula building logic (#96)

sahas-subramanian-frequenz · web-flow · commit 7c8f2cee25b4 · 2022-12-06T11:25:32.000+01:00
Splits the `FormulaEngine` class into: - `FormulaBuilder` - `FormulaEngine` The `FormulaBuilder` class is used to build formulas, and has a `build` method that returns a `FormulaEngine` instance. Closes #82
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_formula_engine.py b/src/frequenz/sdk/timeseries/logical_meter/_formula_engine.py
@@ -35,94 +35,23 @@
 class FormulaEngine:
     """A post-fix formula engine that operates on `Sample` receivers.
 
-    Operators and metrics need to be pushed into the engine in in-fix order, and they
-    get rearranged into post-fix in the engine.  This is done using the [Shunting yard
-    algorithm](https://en.wikipedia.org/wiki/Shunting_yard_algorithm).
-
-    Example:
-        To create an engine that adds the latest entries from two receivers, the
-        following calls need to be made:
-
-        ```python
-        engine = FormulaEngine()
-        engine.push_metric("metric_1", receiver_1)
-        engine.push_oper("+")
-        engine.push_metric("metric_2", receiver_2)
-        engine.finalize()
-        ```
-
-        and then every call to `engine.apply()` would fetch a value from each receiver,
-        add the values and return the result.
+    Use the `FormulaBuilder` to create `FormulaEngine` instances.
     """
 
     def __init__(
-        self,
-    ) -> None:
-        """Create a `FormulaEngine` instance."""
-        self._steps: List[FormulaStep] = []
-        self._build_stack: List[FormulaStep] = []
-        self._metric_fetchers: Dict[str, MetricFetcher] = {}
-        self._first_run = True
-
-    def push_oper(self, oper: str) -> None:
-        """Push an operator into the engine.
-
-        Args:
-            oper: One of these strings - "+", "-", "*", "/", "(", ")"
-        """
-        if self._build_stack and oper != "(":
-            op_prec = _operator_precedence[oper]
-            while self._build_stack:
-                prev_step = self._build_stack[-1]
-                if op_prec <= _operator_precedence[repr(prev_step)]:
-                    break
-                if oper == ")" and repr(prev_step) == "(":
-                    self._build_stack.pop()
-                    break
-                if repr(prev_step) == "(":
-                    break
-                self._steps.append(prev_step)
-                self._build_stack.pop()
-
-        if oper == "+":
-            self._build_stack.append(Adder())
-        elif oper == "-":
-            self._build_stack.append(Subtractor())
-        elif oper == "*":
-            self._build_stack.append(Multiplier())
-        elif oper == "/":
-            self._build_stack.append(Divider())
-        elif oper == "(":
-            self._build_stack.append(OpenParen())
-
-    def push_metric(
-        self,
-        name: str,
-        data_stream: Receiver[Sample],
-        nones_are_zeros: bool,
+        self, steps: List[FormulaStep], metric_fetchers: Dict[str, MetricFetcher]
     ) -> None:
-        """Push a metric receiver into the engine.
+        """Create a `FormulaEngine` instance.
 
         Args:
-            name: A name for the metric.
-            data_stream: A receiver to fetch this metric from.
-            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
-                False, the returned value will be a None.
-        """
-        fetcher = self._metric_fetchers.setdefault(
-            name, MetricFetcher(name, data_stream, nones_are_zeros)
-        )
-        self._steps.append(fetcher)
-
-    def finalize(self) -> None:
-        """Finalize the formula engine.
-
-        This function must be called before calls to `apply` can be made.
+            steps: Steps for the engine to execute, in post-fix order.
+            metric_fetchers: Fetchers for each metric stream the formula depends on.
         """
-        while self._build_stack:
-            self._steps.append(self._build_stack.pop())
+        self._steps = steps
+        self._metric_fetchers = metric_fetchers
+        self._first_run = True
 
-    async def synchronize_metric_timestamps(
+    async def _synchronize_metric_timestamps(
         self, metrics: Set[asyncio.Task[Optional[Sample]]]
     ) -> datetime:
         """Synchronize the metric streams.
@@ -191,7 +120,7 @@ async def apply(self) -> Sample:
             raise RuntimeError("Some resampled metrics didn't arrive")
 
         if self._first_run:
-            metric_ts = await self.synchronize_metric_timestamps(ready_metrics)
+            metric_ts = await self._synchronize_metric_timestamps(ready_metrics)
         else:
             res = next(iter(ready_metrics)).result()
             assert res is not None
@@ -206,3 +135,96 @@ async def apply(self) -> Sample:
             raise RuntimeError("Formula application failed.")
 
         return Sample(metric_ts, eval_stack[0])
+
+
+class FormulaBuilder:
+    """Builds a post-fix formula engine that operates on `Sample` receivers.
+
+    Operators and metrics need to be pushed in in-fix order, and they get rearranged
+    into post-fix order.  This is done using the [Shunting yard
+    algorithm](https://en.wikipedia.org/wiki/Shunting_yard_algorithm).
+
+    Example:
+        To create an engine that adds the latest entries from two receivers, the
+        following calls need to be made:
+
+        ```python
+        builder = FormulaBuilder()
+        builder.push_metric("metric_1", receiver_1)
+        builder.push_oper("+")
+        builder.push_metric("metric_2", receiver_2)
+        engine = builder.build()
+        ```
+
+        and then every call to `engine.apply()` would fetch a value from each receiver,
+        add the values and return the result.
+    """
+
+    def __init__(
+        self,
+    ) -> None:
+        """Create a `FormulaBuilder` instance."""
+        self._build_stack: List[FormulaStep] = []
+        self._steps: List[FormulaStep] = []
+        self._metric_fetchers: Dict[str, MetricFetcher] = {}
+
+    def push_oper(self, oper: str) -> None:
+        """Push an operator into the engine.
+
+        Args:
+            oper: One of these strings - "+", "-", "*", "/", "(", ")"
+        """
+        if self._build_stack and oper != "(":
+            op_prec = _operator_precedence[oper]
+            while self._build_stack:
+                prev_step = self._build_stack[-1]
+                if op_prec <= _operator_precedence[repr(prev_step)]:
+                    break
+                if oper == ")" and repr(prev_step) == "(":
+                    self._build_stack.pop()
+                    break
+                if repr(prev_step) == "(":
+                    break
+                self._steps.append(prev_step)
+                self._build_stack.pop()
+
+        if oper == "+":
+            self._build_stack.append(Adder())
+        elif oper == "-":
+            self._build_stack.append(Subtractor())
+        elif oper == "*":
+            self._build_stack.append(Multiplier())
+        elif oper == "/":
+            self._build_stack.append(Divider())
+        elif oper == "(":
+            self._build_stack.append(OpenParen())
+
+    def push_metric(
+        self,
+        name: str,
+        data_stream: Receiver[Sample],
+        nones_are_zeros: bool,
+    ) -> None:
+        """Push a metric receiver into the engine.
+
+        Args:
+            name: A name for the metric.
+            data_stream: A receiver to fetch this metric from.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
+                False, the returned value will be a None.
+        """
+        fetcher = self._metric_fetchers.setdefault(
+            name, MetricFetcher(name, data_stream, nones_are_zeros)
+        )
+        self._steps.append(fetcher)
+
+    def build(self) -> FormulaEngine:
+        """Finalize and build the formula engine.
+
+        Returns:
+            A `FormulaEngine` instance.
+        """
+        while self._build_stack:
+            self._steps.append(self._build_stack.pop())
+
+        return FormulaEngine(self._steps, self._metric_fetchers)
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py b/src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py
@@ -12,8 +12,8 @@
 from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid.component import ComponentMetricId
 from .._sample import Sample
-from ._formula_builder import FormulaBuilder
 from ._formula_engine import FormulaEngine
+from ._resampled_formula_builder import ResampledFormulaBuilder
 
 
 class LogicalMeter:
@@ -54,7 +54,7 @@ def __init__(
     async def _engine_from_formula_string(
         self, formula: str, metric_id: ComponentMetricId, nones_are_zeros: bool
     ) -> FormulaEngine:
-        builder = FormulaBuilder(
+        builder = ResampledFormulaBuilder(
             self._namespace,
             self._channel_registry,
             self._resampler_subscription_sender,
diff --git a/src/frequenz/sdk/timeseries/logical_meter/_resampled_formula_builder.py b/src/frequenz/sdk/timeseries/logical_meter/_resampled_formula_builder.py
@@ -9,11 +9,11 @@
 from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid.component import ComponentMetricId
 from .._sample import Sample
-from ._formula_engine import FormulaEngine
+from ._formula_engine import FormulaBuilder, FormulaEngine
 from ._tokenizer import Tokenizer, TokenType
 
 
-class FormulaBuilder:
+class ResampledFormulaBuilder:
     """Provides a way to build a FormulaEngine from resampled data streams."""
 
     def __init__(
@@ -23,7 +23,7 @@ def __init__(
         resampler_subscription_sender: Sender[ComponentMetricRequest],
         metric_id: ComponentMetricId,
     ) -> None:
-        """Create a `FormulaBuilder` instance.
+        """Create a `ResampledFormulaBuilder` instance.
 
         Args:
             namespace: The unique namespace to allow reuse of streams in the data
@@ -37,14 +37,14 @@ def __init__(
         self._channel_registry = channel_registry
         self._resampler_subscription_sender = resampler_subscription_sender
         self._namespace = namespace
-        self._formula = FormulaEngine()
+        self._formula = FormulaBuilder()
         self._metric_id = metric_id
 
     async def _get_resampled_receiver(self, component_id: int) -> Receiver[Sample]:
         """Get a receiver with the resampled data for the given component id.
 
         This receiver would contain data for the `metric_id` specified when creating the
-        `FormulaBuilder` instance.
+        `ResampledFormulaBuilder` instance.
 
         Args:
             component_id: The component id for which to get a resampled data receiver.
@@ -105,5 +105,4 @@ async def from_string(
             else:
                 raise ValueError(f"Unknown token type: {token}")
 
-        self._formula.finalize()
-        return self._formula
+        return self._formula.build()
diff --git a/tests/timeseries/test_formula_engine.py b/tests/timeseries/test_formula_engine.py
@@ -9,7 +9,7 @@
 from frequenz.channels import Broadcast
 
 from frequenz.sdk.timeseries import Sample
-from frequenz.sdk.timeseries.logical_meter._formula_engine import FormulaEngine
+from frequenz.sdk.timeseries.logical_meter._formula_engine import FormulaBuilder
 from frequenz.sdk.timeseries.logical_meter._tokenizer import Token, Tokenizer, TokenType
 
 
@@ -63,19 +63,19 @@ async def run_test(
         nones_are_zeros: bool = False,
     ) -> None:
         channels: Dict[str, Broadcast[Sample]] = {}
-        engine = FormulaEngine()
+        builder = FormulaBuilder()
         for token in Tokenizer(formula):
             if token.type == TokenType.COMPONENT_METRIC:
                 if token.value not in channels:
                     channels[token.value] = Broadcast(token.value)
-                engine.push_metric(
+                builder.push_metric(
                     f"#{token.value}",
                     channels[token.value].new_receiver(),
                     nones_are_zeros,
                 )
             elif token.type == TokenType.OPER:
-                engine.push_oper(token.value)
-        engine.finalize()
+                builder.push_oper(token.value)
+        engine = builder.build()
 
         assert repr(engine._steps) == postfix
 
diff --git a/tests/timeseries/test_logical_meter.py b/tests/timeseries/test_logical_meter.py
@@ -25,7 +25,9 @@
 )
 from frequenz.sdk.microgrid.component import ComponentMetricId
 from frequenz.sdk.timeseries.logical_meter import LogicalMeter
-from frequenz.sdk.timeseries.logical_meter._formula_builder import FormulaBuilder
+from frequenz.sdk.timeseries.logical_meter._resampled_formula_builder import (
+    ResampledFormulaBuilder,
+)
 from tests.microgrid import mock_api
 
 
@@ -149,7 +151,7 @@ async def test_1(self, mocker: MockerFixture) -> None:
         # `_get_resampled_receiver` function implementation.
 
         # pylint: disable=protected-access
-        builder = FormulaBuilder(
+        builder = ResampledFormulaBuilder(
             logical_meter._namespace,
             channel_registry,
             request_sender,
