Support handling None values in LogicalMeter/FormulaEngine (#95)

The Resampling actor sends `None` values out when there is no data from
a component. Currently the logical meter raises an exception if it sees
a `None` value from any of its streams, which kills the LogicalMeter.
This is incorrect.

`None` values should instead be treated either as `None` or `0` based on
context.

To this end, a new parameter `nones_are_zeros` is introduced in the
LogicalMeter and FormulaEngine.

If this field is False, then when there are None values in any of the
input streams, the calculated values will be None.

When it is True, the None values will be treated as 0.0 when the formula
is evaluated.

Author: sahas-subramanian-frequenz

src/frequenz/sdk/timeseries/logical_meter/_formula_builder.py

@@ -58,16 +58,24 @@ async def _get_resampled_receiver(self, component_id: int) -> Receiver[Sample]:
         await self._resampler_subscription_sender.send(request)
         return self._channel_registry.new_receiver(request.get_channel_name())
 
-    async def push_component_metric(self, component_id: int) -> None:
+    async def push_component_metric(
+        self, component_id: int, nones_are_zeros: bool
+    ) -> None:
         """Push a resampled component metric stream to the formula engine.
 
         Args:
             component_id: The component id for which to push a metric fetcher.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
+                False, the returned value will be a None.
         """
         receiver = await self._get_resampled_receiver(component_id)
-        self._formula.push_metric(f"#{component_id}", receiver)
+        self._formula.push_metric(f"#{component_id}", receiver, nones_are_zeros)
 
-    async def from_string(self, formula: str) -> FormulaEngine:
+    async def from_string(
+        self,
+        formula: str,
+        nones_are_zeros: bool,
+    ) -> FormulaEngine:
         """Construct a `FormulaEngine` from the given formula string.
 
         Formulas can have Component IDs that are preceeded by a pound symbol("#"), and
@@ -78,6 +86,8 @@ async def from_string(self, formula: str) -> FormulaEngine:
 
         Args:
             formula: A string formula.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
+                False, the returned value will be a None.
 
         Returns:
             A FormulaEngine instance corresponding to the given formula.
@@ -89,7 +99,7 @@ async def from_string(self, formula: str) -> FormulaEngine:
 
         for token in tokenizer:
             if token.type == TokenType.COMPONENT_METRIC:
-                await self.push_component_metric(int(token.value))
+                await self.push_component_metric(int(token.value), nones_are_zeros)
             elif token.type == TokenType.OPER:
                 self._formula.push_oper(token.value)
             else:

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

@@ -95,15 +95,22 @@ def push_oper(self, oper: str) -> None:
         elif oper == "(":
             self._build_stack.append(OpenParen())
 
-    def push_metric(self, name: str, data_stream: Receiver[Sample]) -> None:
+    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)
+            name, MetricFetcher(name, data_stream, nones_are_zeros)
         )
         self._steps.append(fetcher)
 
@@ -171,7 +178,7 @@ async def apply(self) -> Sample:
             RuntimeError: if some samples didn't arrive, or if formula application
                 failed.
         """
-        eval_stack: List[float] = []
+        eval_stack: List[Optional[float]] = []
         ready_metrics, pending = await asyncio.wait(
             [
                 asyncio.create_task(fetcher.fetch_next(), name=name)

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

@@ -27,7 +27,7 @@ def __repr__(self) -> str:
         """
 
     @abstractmethod
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Apply a formula operation on the eval_stack.
 
         Args:
@@ -46,15 +46,19 @@ def __repr__(self) -> str:
         """
         return "+"
 
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Extract two values from the stack, add them, push the result back in.
 
         Args:
             eval_stack: An evaluation stack, to apply the formula step on.
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        eval_stack.append(val1 + val2)
+        if val1 is None or val2 is None:
+            res = None
+        else:
+            res = val1 + val2
+        eval_stack.append(res)
 
 
 class Subtractor(FormulaStep):
@@ -68,15 +72,19 @@ def __repr__(self) -> str:
         """
         return "-"
 
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Extract two values from the stack, subtract them, push the result back in.
 
         Args:
             eval_stack: An evaluation stack, to apply the formula step on.
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        eval_stack.append(val1 - val2)
+        if val1 is None or val2 is None:
+            res = None
+        else:
+            res = val1 - val2
+        eval_stack.append(res)
 
 
 class Multiplier(FormulaStep):
@@ -90,15 +98,19 @@ def __repr__(self) -> str:
         """
         return "*"
 
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Extract two values from the stack, multiply them, push the result back in.
 
         Args:
             eval_stack: An evaluation stack, to apply the formula step on.
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        eval_stack.append(val1 * val2)
+        if val1 is None or val2 is None:
+            res = None
+        else:
+            res = val1 * val2
+        eval_stack.append(res)
 
 
 class Divider(FormulaStep):
@@ -112,15 +124,19 @@ def __repr__(self) -> str:
         """
         return "/"
 
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Extract two values from the stack, divide them, push the result back in.
 
         Args:
             eval_stack: An evaluation stack, to apply the formula step on.
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        eval_stack.append(val1 / val2)
+        if val1 is None or val2 is None:
+            res = None
+        else:
+            res = val1 / val2
+        eval_stack.append(res)
 
 
 class OpenParen(FormulaStep):
@@ -137,23 +153,27 @@ def __repr__(self) -> str:
         """
         return "("
 
-    def apply(self, _: List[float]) -> None:
+    def apply(self, _: List[Optional[float]]) -> None:
         """No-op."""
 
 
 class MetricFetcher(FormulaStep):
     """A formula step for fetching a value from a metric Receiver."""
 
-    def __init__(self, name: str, stream: Receiver[Sample]) -> None:
+    def __init__(
+        self, name: str, stream: Receiver[Sample], nones_are_zeros: bool
+    ) -> None:
         """Create a `MetricFetcher` instance.
 
         Args:
             name: The name of the metric.
             stream: A channel receiver from which to fetch samples.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.
         """
         self._name = name
         self._stream = stream
         self._next_value: Optional[Sample] = None
+        self._nones_are_zeros = nones_are_zeros
 
     async def fetch_next(self) -> Optional[Sample]:
         """Fetch the next value from the stream.
@@ -174,7 +194,7 @@ def __repr__(self) -> str:
         """
         return self._name
 
-    def apply(self, eval_stack: List[float]) -> None:
+    def apply(self, eval_stack: List[Optional[float]]) -> None:
         """Push the latest value from the stream into the evaluation stack.
 
         Args:
@@ -183,6 +203,9 @@ def apply(self, eval_stack: List[float]) -> None:
         Raises:
             RuntimeError: No next value available to append.
         """
-        if self._next_value is None or self._next_value.value is None:
+        if self._next_value is None:
             raise RuntimeError("No next value available to append.")
-        eval_stack.append(self._next_value.value)
+        if self._next_value.value is None and self._nones_are_zeros:
+            eval_stack.append(0.0)
+        else:
+            eval_stack.append(self._next_value.value)

src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py

@@ -52,15 +52,15 @@ def __init__(
         self._tasks: List[asyncio.Task[None]] = []
 
     async def _engine_from_formula_string(
-        self, formula: str, metric_id: ComponentMetricId
+        self, formula: str, metric_id: ComponentMetricId, nones_are_zeros: bool
     ) -> FormulaEngine:
         builder = FormulaBuilder(
             self._namespace,
             self._channel_registry,
             self._resampler_subscription_sender,
             metric_id,
         )
-        return await builder.from_string(formula)
+        return await builder.from_string(formula, nones_are_zeros)
 
     async def _run_formula(
         self, formula: FormulaEngine, sender: Sender[Sample]
@@ -78,13 +78,16 @@ async def start_formula(
         self,
         formula: str,
         component_metric_id: ComponentMetricId,
+        nones_are_zeros: bool = False,
     ) -> Receiver[Sample]:
         """Start execution of the given formula name.
 
         Args:
             formula: formula to execute.
             component_metric_id: The metric ID to use when fetching receivers from the
                 resampling actor.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
+                False, the returned value will be a None.
 
         Returns:
             A Receiver that streams values with the formulas applied.
@@ -94,8 +97,7 @@ async def start_formula(
             return self._output_channels[channel_key].new_receiver()
 
         formula_engine = await self._engine_from_formula_string(
-            formula,
-            component_metric_id,
+            formula, component_metric_id, nones_are_zeros
         )
         out_chan = Broadcast[Sample](channel_key)
         self._output_channels[channel_key] = out_chan

tests/timeseries/test_formula_engine.py

@@ -4,7 +4,7 @@
 """Tests for the FormulaEngine and the Tokenizer."""
 
 from datetime import datetime
-from typing import Dict, List, Tuple
+from typing import Dict, List, Optional, Tuple
 
 from frequenz.channels import Broadcast
 
@@ -56,7 +56,11 @@ def setup(self) -> None:
         # pylint: enable=attribute-defined-outside-init
 
     async def run_test(
-        self, formula: str, postfix: str, io_pairs: List[Tuple[List[float], float]]
+        self,
+        formula: str,
+        postfix: str,
+        io_pairs: List[Tuple[List[Optional[float]], Optional[float]]],
+        nones_are_zeros: bool = False,
     ) -> None:
         channels: Dict[str, Broadcast[Sample]] = {}
         engine = FormulaEngine()
@@ -65,7 +69,9 @@ async def run_test(
                 if token.value not in channels:
                     channels[token.value] = Broadcast(token.value)
                 engine.push_metric(
-                    f"#{token.value}", channels[token.value].new_receiver()
+                    f"#{token.value}",
+                    channels[token.value].new_receiver(),
+                    nones_are_zeros,
                 )
             elif token.type == TokenType.OPER:
                 engine.push_oper(token.value)
@@ -170,3 +176,59 @@ async def test_compound(self) -> None:
                 ([15.0, 17.0, 20.0, 5.0], 28.0),
             ],
         )
+
+    async def test_nones_are_zeros(self) -> None:
+        """Test that `None`s are treated as zeros when configured."""
+        await self.run_test(
+            "#2 - #4 + #5",
+            "[#2, #4, -, #5, +]",
+            [
+                ([10.0, 12.0, 15.0], 13.0),
+                ([None, 12.0, 15.0], 3.0),
+                ([10.0, None, 15.0], 25.0),
+                ([15.0, 17.0, 20.0], 18.0),
+                ([15.0, None, None], 15.0),
+            ],
+            True,
+        )
+
+        await self.run_test(
+            "#2 + #4 - (#5 * #6)",
+            "[#2, #4, #5, #6, *, -, +]",
+            [
+                ([10.0, 12.0, 15.0, 2.0], -8.0),
+                ([10.0, 12.0, 15.0, None], 22.0),
+                ([10.0, None, 15.0, 2.0], -20.0),
+                ([15.0, 17.0, 20.0, 5.0], -68.0),
+                ([15.0, 17.0, None, 5.0], 32.0),
+            ],
+            True,
+        )
+
+    async def test_nones_are_not_zeros(self) -> None:
+        """Test that calculated values are `None` on input `None`s."""
+        await self.run_test(
+            "#2 - #4 + #5",
+            "[#2, #4, -, #5, +]",
+            [
+                ([10.0, 12.0, 15.0], 13.0),
+                ([None, 12.0, 15.0], None),
+                ([10.0, None, 15.0], None),
+                ([15.0, 17.0, 20.0], 18.0),
+                ([15.0, None, None], None),
+            ],
+            False,
+        )
+
+        await self.run_test(
+            "#2 + #4 - (#5 * #6)",
+            "[#2, #4, #5, #6, *, -, +]",
+            [
+                ([10.0, 12.0, 15.0, 2.0], -8.0),
+                ([10.0, 12.0, 15.0, None], None),
+                ([10.0, None, 15.0, 2.0], None),
+                ([15.0, 17.0, 20.0, 5.0], -68.0),
+                ([15.0, 17.0, None, 5.0], None),
+            ],
+            False,
+        )