Use stricter typing for eval_stack used by the FormulaEngine (#162)

... now that `None` values are getting converted to `NaN` before they
get stored in the `eval_stack`.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: sahas-subramanian-frequenz

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

@@ -126,7 +126,7 @@ async def _apply(self) -> Sample:
             RuntimeError: if some samples didn't arrive, or if formula application
                 failed.
         """
-        eval_stack: List[Optional[float]] = []
+        eval_stack: List[float] = []
         ready_metrics, pending = await asyncio.wait(
             [
                 asyncio.create_task(fetcher.fetch_next(), name=name)

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

@@ -31,7 +31,7 @@ def __repr__(self) -> str:
         """
 
     @abstractmethod
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Apply a formula operation on the eval_stack.
 
         Args:
@@ -50,7 +50,7 @@ def __repr__(self) -> str:
         """
         return "+"
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Extract two values from the stack, add them, push the result back in.
 
         Args:
@@ -73,7 +73,7 @@ def __repr__(self) -> str:
         """
         return "-"
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Extract two values from the stack, subtract them, push the result back in.
 
         Args:
@@ -96,7 +96,7 @@ def __repr__(self) -> str:
         """
         return "*"
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Extract two values from the stack, multiply them, push the result back in.
 
         Args:
@@ -119,7 +119,7 @@ def __repr__(self) -> str:
         """
         return "/"
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Extract two values from the stack, divide them, push the result back in.
 
         Args:
@@ -168,7 +168,7 @@ def __repr__(self) -> str:
         """
         return f"avg({', '.join(repr(f) for f in self._fetchers)})"
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Calculate average of given metrics, push the average to the eval_stack.
 
         Args:
@@ -243,7 +243,7 @@ def __repr__(self) -> str:
         """
         return self._name
 
-    def apply(self, eval_stack: List[Optional[float]]) -> None:
+    def apply(self, eval_stack: List[float]) -> None:
         """Push the latest value from the stream into the evaluation stack.
 
         Args:
@@ -262,4 +262,4 @@ def apply(self, eval_stack: List[Optional[float]]) -> None:
             else:
                 eval_stack.append(float("NaN"))
         else:
-            eval_stack.append(self._next_value.value)
+            eval_stack.append(next_value)