Remove unused telemetry stream tracking in formula evaluator

The nodes themselves have become responsible to progressing the input
streams, so they don't have to be handled by the formula evaluator
anymore.

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

Author: shsms

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

@@ -48,7 +48,6 @@ def __init__(  # pylint: disable=too-many-arguments
         name: str,
         root: AstNode[QuantityT],
         create_method: Callable[[float], QuantityT],
-        streams: list[_ast.TelemetryStream[QuantityT]],
         sub_formulas: list[Formula[QuantityT]] | None = None,
         metric_fetcher: ResampledStreamFetcher | None = None,
     ) -> None:
@@ -60,15 +59,13 @@ def __init__(  # pylint: disable=too-many-arguments
             create_method: A method to generate the output values with.  If the
                 formula is for generating power values, this would be
                 `Power.from_watts`, for example.
-            streams: The telemetry streams that the formula depends on.
             sub_formulas: Any sub-formulas that this formula depends on.
             metric_fetcher: An optional metric fetcher that needs to be started
                 before the formula can be evaluated.
         """
         BackgroundService.__init__(self)
         self._name: str = name
         self._root: AstNode[QuantityT] = root
-        self._components: list[_ast.TelemetryStream[QuantityT]] = streams
         self._create_method: Callable[[float], QuantityT] = create_method
         self._sub_formulas: list[Formula[QuantityT]] = sub_formulas or []
 
@@ -78,7 +75,6 @@ def __init__(  # pylint: disable=too-many-arguments
         )
         self._evaluator: FormulaEvaluatingActor[QuantityT] = FormulaEvaluatingActor(
             root=self._root,
-            components=self._components,
             output_channel=self._channel,
             metric_fetcher=metric_fetcher,
         )
@@ -383,6 +379,5 @@ def build(
             name=name,
             root=self.root,
             create_method=self._create_method,
-            streams=self._streams,
             sub_formulas=self._sub_formulas,
         )

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

@@ -11,7 +11,6 @@
 
 from ...actor import Actor
 from .._base_types import QuantityT, Sample
-from . import _ast
 from ._base_ast_node import AstNode
 from ._resampled_stream_fetcher import ResampledStreamFetcher
 
@@ -25,23 +24,20 @@ def __init__(  # pylint: disable=too-many-arguments
         self,
         *,
         root: AstNode[QuantityT],
-        components: list[_ast.TelemetryStream[QuantityT]],
         output_channel: Broadcast[Sample[QuantityT]],
         metric_fetcher: ResampledStreamFetcher | None = None,
     ) -> None:
         """Create a `FormulaEvaluatingActor` instance.
 
         Args:
             root: The root node of the formula AST.
-            components: The telemetry streams that the formula depends on.
             output_channel: The channel to send evaluated samples to.
             metric_fetcher: An optional metric fetcher that needs to be started
                 before the formula can be evaluated.
         """
         super().__init__()
 
         self._root: AstNode[QuantityT] = root
-        self._components: list[_ast.TelemetryStream[QuantityT]] = components
         self._metric_fetcher: ResampledStreamFetcher | None = metric_fetcher
         self._output_channel: Broadcast[Sample[QuantityT]] = output_channel
 
@@ -56,21 +52,11 @@ async def _run(self) -> None:
                 if isinstance(res, Sample):
                     next_sample = res
                 else:
-                    timestamp = next(
-                        (
-                            comp.latest_sample.timestamp
-                            for comp in self._components
-                            if comp.latest_sample is not None
-                        ),
-                        None,
+                    _logger.debug(
+                        "No input samples available; stopping formula evaluator. (%s)",
+                        self._root,
                     )
-                    if timestamp is None:
-                        _logger.debug(
-                            "No input samples available; stopping formula evaluator. (%s)",
-                            self._root,
-                        )
-                        return
-                    next_sample = Sample(timestamp, res)
+                    return
                 await self._output_sender.send(next_sample)
             except (StopAsyncIteration, ReceiverStoppedError):
                 _logger.debug(

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

@@ -67,7 +67,6 @@ def __init__(
         self._name: str = name
         self._lexer: Peekable[_token.Token] = Peekable(Lexer(formula))
         self._telemetry_fetcher: ResampledStreamFetcher = telemetry_fetcher
-        self._components: list[_ast.TelemetryStream[QuantityT]] = []
         self._create_method: Callable[[float], QuantityT] = create_method
 
     def _parse_term(self) -> AstNode[QuantityT] | None:
@@ -202,7 +201,6 @@ def make_component_stream_fetcher(
                 ),
                 create_method=self._create_method,
             )
-            self._components.append(comp)
             return comp
 
         if isinstance(token, _token.Number):
@@ -227,6 +225,5 @@ def parse(self) -> Formula[QuantityT]:
             name=self._name,
             root=expr,
             create_method=self._create_method,
-            streams=self._components,
             metric_fetcher=self._telemetry_fetcher,
         )