Apply suggestions from review comments

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

Author: shsms

src/frequenz/sdk/timeseries/formulas/_ast.py

@@ -310,9 +310,7 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None:
                     value=left.value * right.value.base_value,
                 )
             case Quantity(), Quantity():
-                return left.__class__._new(  # pylint: disable=protected-access
-                    left.base_value * right.base_value
-                )
+                return left * right.base_value
             case (Sample(), Quantity()):
                 return (
                     left
@@ -378,12 +376,12 @@ async def evaluate(self) -> Sample[QuantityT] | QuantityT | None:
         match left, right:
             case Sample(), Sample():
                 if left.value is None:
-                    return None
+                    return left
                 if right.value is None:
-                    return None
+                    return right
                 if is_close_to_zero(right.value.base_value):
                     _logger.warning("Division by zero encountered in formula.")
-                    return None
+                    return Sample(left.timestamp, None)
                 return Sample(left.timestamp, left.value / right.value.base_value)
             case Quantity(), Quantity():
                 if is_close_to_zero(right.base_value):

src/frequenz/sdk/timeseries/formulas/_base_ast_node.py

@@ -17,6 +17,8 @@
 
 _logger = logging.getLogger(__name__)
 
+_MAX_SYNC_RETRIES = 10
+
 
 @dataclass(kw_only=True)
 class AstNode(abc.ABC, Generic[QuantityT]):
@@ -103,7 +105,7 @@ async def evaluate(
 
             return await self._synchronize_to_timestamp(values, nodes, target_timestamp)
 
-        return [await node.evaluate() for node in nodes]
+        return await asyncio.gather(*(node.evaluate() for node in nodes))
 
     async def _synchronize_to_timestamp(
         self,
@@ -114,17 +116,18 @@ async def _synchronize_to_timestamp(
         for i, value in enumerate(values):
             if isinstance(value, Sample):
                 ctr = 0
-                while ctr < 10 and value.timestamp < target_timestamp:
+                while ctr < _MAX_SYNC_RETRIES and value.timestamp < target_timestamp:
                     value = await nodes[i].evaluate()
                     if not isinstance(value, Sample):
                         raise RuntimeError(
                             "Subsequent AST node evaluation did not return a Sample"
                         )
                     values[i] = value
                     ctr += 1
-                if ctr >= 10 and value.timestamp < target_timestamp:
+                if ctr >= _MAX_SYNC_RETRIES and value.timestamp < target_timestamp:
                     raise RuntimeError(
-                        "Could not synchronize AST node evaluations after 10 tries"
+                        "Could not synchronize AST node evaluations after "
+                        + f"{_MAX_SYNC_RETRIES} tries"
                     )
                 if value.timestamp > target_timestamp:
                     self._latest_values[id(nodes[i])] = value

src/frequenz/sdk/timeseries/formulas/_formula.py

@@ -32,9 +32,9 @@ def metric_fetcher(
 ) -> Callable[[], Coroutine[None, None, Receiver[Sample[QuantityT]]]]:
     """Fetch a receiver for the formula's output samples."""
 
-    async def fetcher(formula: Formula[QuantityT]) -> Receiver[Sample[QuantityT]]:
-        formula.start()
-        return formula.new_receiver()
+    async def fetcher(f: Formula[QuantityT]) -> Receiver[Sample[QuantityT]]:
+        f.start()
+        return f.new_receiver()
 
     return lambda: fetcher(formula)
 

src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py

@@ -97,9 +97,7 @@ async def _run(self) -> None:
             ):
                 # This should never happen because the components are Formula
                 # instances
-                raise RuntimeError(
-                    "Expected all phase samples to be of type Sample3Phase"
-                )
+                raise RuntimeError("Expected all phase samples to be of type Sample")
 
             sample_3phase = Sample3Phase(
                 timestamp=phase_1_sample.timestamp,

src/frequenz/sdk/timeseries/formulas/_functions.py

@@ -100,6 +100,7 @@ def from_string(
                 raise ValueError(f"Unknown function name: {name}")
 
 
+@dataclass
 class Coalesce(Function[QuantityT]):
     """A function that returns the first non-None argument."""
 
@@ -117,7 +118,7 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None:
         ts: datetime | None = None
 
         if self.num_subscribed == 0:
-            await self._subscribe_next()
+            await self._subscribe_next_param()
 
         args = await self._synchronizer.evaluate(
             self.params[: self.num_subscribed], sync_to_first_node=True
@@ -128,13 +129,13 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None:
                     if value is not None:
                         # Found a non-None value, unsubscribe from subsequent params
                         if ctr < self.num_subscribed:
-                            await self._unsubscribe_after(ctr)
+                            await self._unsubscribe_all_params_after(ctr)
                         return arg
                     ts = timestamp
                 case Quantity():
                     # Found a non-None value, unsubscribe from subsequent params
                     if ctr < self.num_subscribed:
-                        await self._unsubscribe_after(ctr)
+                        await self._unsubscribe_all_params_after(ctr)
                     if ts is not None:
                         return Sample(timestamp=ts, value=arg)
                     return arg
@@ -143,7 +144,7 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None:
         # Don't have a non-None value yet, subscribe to the next parameter for
         # next time and return None for now, unless the next value is a constant.
         next_value: Sample[QuantityT] | QuantityT | None = None
-        await self._subscribe_next()
+        await self._subscribe_next_param()
 
         if isinstance(self.params[self.num_subscribed - 1], Constant):
             next_value = await self.params[self.num_subscribed - 1].evaluate()
@@ -158,9 +159,9 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None:
     async def subscribe(self) -> None:
         """Subscribe to the first parameter if not already subscribed."""
         if self.num_subscribed == 0:
-            await self._subscribe_next()
+            await self._subscribe_next_param()
 
-    async def _subscribe_next(self) -> None:
+    async def _subscribe_next_param(self) -> None:
         """Subscribe to the next parameter."""
         if self.num_subscribed < len(self.params):
             _logger.debug(
@@ -171,7 +172,7 @@ async def _subscribe_next(self) -> None:
             await self.params[self.num_subscribed].subscribe()
             self.num_subscribed += 1
 
-    async def _unsubscribe_after(self, index: int) -> None:
+    async def _unsubscribe_all_params_after(self, index: int) -> None:
         """Unsubscribe from parameters after the given index."""
         for param in self.params[index:]:
             _logger.debug(
@@ -182,6 +183,7 @@ async def _unsubscribe_after(self, index: int) -> None:
         self.num_subscribed = index
 
 
+@dataclass
 class Max(Function[QuantityT]):
     """A function that returns the maximum of the arguments."""
 
@@ -216,6 +218,7 @@ async def __call__(self) -> Sample[QuantityT] | QuantityT | None:
         return max_value
 
 
+@dataclass
 class Min(Function[QuantityT]):
     """A function that returns the minimum of the arguments."""
 

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

@@ -94,12 +94,12 @@ def start_formula(
     ) -> Formula[Quantity]:
         """Start execution of the given formula.
 
-        Formulas can have Component IDs that are preceeded by a pound symbol("#"),
-        constant values and these operators: +, -, *, /, (, ).
+        Formulas can have Component IDs that are preceded by a pound symbol(`#`),
+        constant values and these operators: `+`, `-`, `*`, `/`, `(`, `)`.
 
         These functions are also supported: `COALESCE`, `MAX`, `MIN`.
 
-        For example, the input string: "#20 + #5" is a formula for adding metrics from
+        For example, the input string: `#20 + #5` is a formula for adding metrics from
         two components with ids 20 and 5.
 
         A more detailed description of the formula syntax with examples can be found

tests/timeseries/_formulas/test_formulas.py

@@ -367,8 +367,6 @@ async def run_test(  # pylint: disable=too-many-locals
 
         def stream_recv(comp_id: int) -> Receiver[Sample[Quantity]]:
             comp_id = int(comp_id)
-            if comp_id not in channels:
-                channels[comp_id] = Broadcast(name=f"chan-#{comp_id}")
             return channels[comp_id].new_receiver()
 
         telem_fetcher = MagicMock(spec=ResampledStreamFetcher)