Handle NaN values in FormulaEngine (#99)

While you are at this, it should probably be a good idea to also treat
*NaN*'s specially (probably the same as `None`). You can check using
`math.isnan()`.

(same for all steps)

**OR**, maybe replace `None` with *NaN* and it should have the same
effect of propagating `None` manually as you are doing here, but it will
be done automatically. Then you only need to convert the final *NaN* in
the stack to `None` and you are set. This option is easier to write (as
you don't have to touch every step) but it might be harder read (to
figure out what's going on).

_Originally posted by @leandro-lucarella-frequenz in
#95 (comment)

Author: daniel-zullo-frequenz

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

@@ -10,6 +10,7 @@
 import weakref
 from collections import deque
 from datetime import datetime
+from math import isinf, isnan
 from typing import Dict, List, Optional, Set, Tuple
 from uuid import UUID, uuid4
 
@@ -154,7 +155,11 @@ async def _apply(self) -> Sample:
         if len(eval_stack) != 1:
             raise RuntimeError(f"Formula application failed: {self._name}")
 
-        return Sample(metric_ts, eval_stack[0])
+        res = eval_stack.pop()
+        if isnan(res) or isinf(res):
+            res = None
+
+        return Sample(metric_ts, res)
 
     async def _run(self) -> None:
         sender = self._channel.new_sender()

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

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from math import isinf, isnan
 from typing import List, Optional
 
 from frequenz.channels import Receiver
@@ -57,10 +58,7 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        if val1 is None or val2 is None:
-            res = None
-        else:
-            res = val1 + val2
+        res = val1 + val2
         eval_stack.append(res)
 
 
@@ -83,10 +81,7 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        if val1 is None or val2 is None:
-            res = None
-        else:
-            res = val1 - val2
+        res = val1 - val2
         eval_stack.append(res)
 
 
@@ -109,10 +104,7 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        if val1 is None or val2 is None:
-            res = None
-        else:
-            res = val1 * val2
+        res = val1 * val2
         eval_stack.append(res)
 
 
@@ -135,10 +127,7 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
         """
         val2 = eval_stack.pop()
         val1 = eval_stack.pop()
-        if val1 is None or val2 is None:
-            res = None
-        else:
-            res = val1 / val2
+        res = val1 / val2
         eval_stack.append(res)
 
 
@@ -265,7 +254,12 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
         """
         if self._next_value is None:
             raise RuntimeError("No next value available to append.")
-        if self._next_value.value is None and self._nones_are_zeros:
-            eval_stack.append(0.0)
+
+        next_value = self._next_value.value
+        if next_value is None or isnan(next_value) or isinf(next_value):
+            if self._nones_are_zeros:
+                eval_stack.append(0.0)
+            else:
+                eval_stack.append(float("NaN"))
         else:
             eval_stack.append(self._next_value.value)

tests/timeseries/test_formula_engine.py

@@ -541,3 +541,67 @@ async def test_nones_are_not_zeros(self) -> None:
             ],
             False,
         )
+
+
+class TestFormulaAverager:
+    """Tests for the formula step for calculating average."""
+
+    async def run_test(
+        self,
+        components: List[str],
+        io_pairs: List[Tuple[List[Optional[float]], Optional[float]]],
+    ) -> None:
+        """Run a formula test."""
+        channels: Dict[str, Broadcast[Sample]] = {}
+        streams: List[Tuple[str, Receiver[Sample], bool]] = []
+        builder = FormulaBuilder("test_averager")
+        for comp_id in components:
+            if comp_id not in channels:
+                channels[comp_id] = Broadcast(comp_id)
+                streams.append((f"{comp_id}", channels[comp_id].new_receiver(), False))
+
+        builder.push_average(streams)
+        engine = builder.build()
+
+        now = datetime.now()
+        tests_passed = 0
+        for io_pair in io_pairs:
+            io_input, io_output = io_pair
+            assert all(
+                await asyncio.gather(
+                    *[
+                        chan.new_sender().send(Sample(now, value))
+                        for chan, value in zip(channels.values(), io_input)
+                    ]
+                )
+            )
+            next_val = await engine._apply()  # pylint: disable=protected-access
+            assert (next_val).value == io_output
+            tests_passed += 1
+        await engine._stop()  # pylint: disable=protected-access
+        assert tests_passed == len(io_pairs)
+
+    async def test_simple(self) -> None:
+        """Test simple formulas."""
+        await self.run_test(
+            ["#2", "#4", "#5"],
+            [
+                ([10.0, 12.0, 14.0], 12.0),
+                ([15.0, 17.0, 19.0], 17.0),
+                ([11.1, 11.1, 11.1], 11.1),
+            ],
+        )
+
+    async def test_nones_are_skipped(self) -> None:
+        """Test that `None`s are skipped for computing the average."""
+        await self.run_test(
+            ["#2", "#4", "#5"],
+            [
+                ([11.0, 13.0, 15.0], 13.0),
+                ([None, 13.0, 19.0], 16.0),
+                ([12.2, None, 22.2], 17.2),
+                ([16.5, 19.5, None], 18.0),
+                ([None, 13.0, None], 13.0),
+                ([None, None, None], 0.0),
+            ],
+        )