Add supporting for configuring formula generators

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

Author: shsms

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

@@ -11,6 +11,10 @@
 
 from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid.component import ComponentMetricId
+from ._formula_generators._formula_generator import (
+    FormulaGenerator,
+    FormulaGeneratorConfig,
+)
 from ._resampled_formula_builder import ResampledFormulaBuilder
 
 if TYPE_CHECKING:
@@ -21,7 +25,6 @@
         _GenericEngine,
         _GenericFormulaReceiver,
     )
-    from ._formula_generators import FormulaGenerator
 
 
 class FormulaEnginePool:
@@ -88,12 +91,14 @@ async def from_generator(
         self,
         channel_key: str,
         generator: "Type[FormulaGenerator[_GenericEngine]]",
+        config: FormulaGeneratorConfig = FormulaGeneratorConfig(),
     ) -> "_GenericFormulaReceiver":
         """Get a receiver for a formula from a generator.
 
         Args:
             channel_key: A string to uniquely identify the formula.
             generator: A formula generator.
+            config: config to initialize the formula generator with.
 
         Returns:
             A FormulaReceiver or a FormulaReceiver3Phase instance based on what the
@@ -103,7 +108,10 @@ async def from_generator(
             return self._engines[channel_key].new_receiver()
 
         engine = await generator(
-            self._namespace, self._channel_registry, self._resampler_subscription_sender
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            config,
         ).generate()
         self._engines[channel_key] = engine
         return engine.new_receiver()

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py

@@ -11,6 +11,7 @@
     ComponentNotFound,
     FormulaGenerationError,
     FormulaGenerator,
+    FormulaGeneratorConfig,
 )
 from ._grid_current_formula import GridCurrentFormula
 from ._grid_power_formula import GridPowerFormula
@@ -21,6 +22,7 @@
     # Base class
     #
     "FormulaGenerator",
+    "FormulaGeneratorConfig",
     #
     # Power Formula generators
     #

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_formula_generator.py

@@ -3,8 +3,11 @@
 
 """Base class for formula generators that use the component graphs."""
 
+from __future__ import annotations
+
 import sys
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Generic
 
 from frequenz.channels import Sender
@@ -26,6 +29,13 @@ class ComponentNotFound(FormulaGenerationError):
 NON_EXISTING_COMPONENT_ID = sys.maxsize
 
 
+@dataclass(frozen=True)
+class FormulaGeneratorConfig:
+    """Config for formula generators."""
+
+    component_ids: set[int] | None = None
+
+
 class FormulaGenerator(ABC, Generic[_GenericEngine]):
     """A class for generating formulas from the component graph."""
 
@@ -34,6 +44,7 @@ def __init__(
         namespace: str,
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
+        config: FormulaGeneratorConfig,
     ) -> None:
         """Create a `FormulaGenerator` instance.
 
@@ -43,10 +54,12 @@ def __init__(
                 actor.
             resampler_subscription_sender: A sender for sending metric requests to the
                 resampling actor.
+            config: configs for the formula generator.
         """
         self._channel_registry = channel_registry
         self._resampler_subscription_sender = resampler_subscription_sender
         self._namespace = namespace
+        self._config = config
 
     def _get_builder(
         self, name: str, component_metric_id: ComponentMetricId