Add EV power and current streams to EVChargerPool (#201)

These formulas were getting generated from the component graph in the
logical meter.

Those have now been removed, and the formulas are now generated for the
ev chargers that are part of the ev charger pool.

Additional methods have been added to fetch the power and current values
for individual ev chargers.

Author: sahas-subramanian-frequenz

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

@@ -0,0 +1,25 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""A formula engine for applying formulas."""
+from ._formula_engine import (
+    FormulaEngine,
+    FormulaEngine3Phase,
+    FormulaReceiver,
+    FormulaReceiver3Phase,
+    _GenericEngine,
+    _GenericFormulaReceiver,
+)
+from ._formula_engine_pool import FormulaEnginePool
+from ._resampled_formula_builder import ResampledFormulaBuilder
+
+__all__ = [
+    "FormulaEngine",
+    "FormulaEngine3Phase",
+    "FormulaReceiver",
+    "FormulaReceiver3Phase",
+    "FormulaEnginePool",
+    "_GenericEngine",
+    "_GenericFormulaReceiver",
+    "ResampledFormulaBuilder",
+]

…/timeseries/logical_meter/_exceptions.py …imeseries/_formula_engine/_exceptions.pysrc/frequenz/sdk/timeseries/logical_meter/_exceptions.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_exceptions.py src/frequenz/sdk/timeseries/logical_meter/_exceptions.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_exceptions.py

@@ -0,0 +1,117 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""A formula pool for helping with tracking running formula engines."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Type
+
+from frequenz.channels import Sender
+
+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:
+    # Break circular import by enclosing these type hints in a `TYPE_CHECKING` block.
+    from .._formula_engine import (
+        FormulaReceiver,
+        FormulaReceiver3Phase,
+        _GenericEngine,
+        _GenericFormulaReceiver,
+    )
+
+
+class FormulaEnginePool:
+    """Creates and owns formula engines from string formulas, or formula generators.
+
+    If an engine already exists with a given name, it is reused instead.
+    """
+
+    def __init__(
+        self,
+        namespace: str,
+        channel_registry: ChannelRegistry,
+        resampler_subscription_sender: Sender[ComponentMetricRequest],
+    ) -> None:
+        """Create a new instance.
+
+        Args:
+            namespace: namespace to use with the data pipeline.
+            channel_registry: A channel registry instance shared with the resampling
+                actor.
+            resampler_subscription_sender: A sender for sending metric requests to the
+                resampling actor.
+        """
+        self._namespace = namespace
+        self._channel_registry = channel_registry
+        self._resampler_subscription_sender = resampler_subscription_sender
+        self._engines: dict[str, "FormulaReceiver|FormulaReceiver3Phase"] = {}
+
+    async def from_string(
+        self,
+        formula: str,
+        component_metric_id: ComponentMetricId,
+        nones_are_zeros: bool = False,
+    ) -> "FormulaReceiver":
+        """Get a receiver for a manual formula.
+
+        Args:
+            formula: formula to execute.
+            component_metric_id: The metric ID to use when fetching receivers from the
+                resampling actor.
+            nones_are_zeros: Whether to treat None values from the stream as 0s.  If
+                False, the returned value will be a None.
+
+        Returns:
+            A FormulaReceiver that streams values with the formulas applied.
+        """
+        channel_key = formula + component_metric_id.value
+        if channel_key in self._engines:
+            return self._engines[channel_key].new_receiver()
+
+        builder = ResampledFormulaBuilder(
+            self._namespace,
+            formula,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            component_metric_id,
+        )
+        formula_engine = await builder.from_string(formula, nones_are_zeros)
+        self._engines[channel_key] = formula_engine
+
+        return formula_engine.new_receiver()
+
+    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
+                FormulaGenerator returns.
+        """
+        if channel_key in self._engines:
+            return self._engines[channel_key].new_receiver()
+
+        engine = await generator(
+            self._namespace,
+            self._channel_registry,
+            self._resampler_subscription_sender,
+            config,
+        ).generate()
+        self._engines[channel_key] = engine
+        return engine.new_receiver()

…eseries/logical_meter/_formula_engine.py …eries/_formula_engine/_formula_engine.pysrc/frequenz/sdk/timeseries/logical_meter/_formula_engine.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.py src/frequenz/sdk/timeseries/logical_meter/_formula_engine.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_engine.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_engine_pool.py

@@ -7,7 +7,7 @@
 
 from .....sdk import microgrid
 from ....microgrid.component import ComponentCategory, ComponentMetricId, InverterType
-from .._formula_engine import FormulaEngine
+from ..._formula_engine import FormulaEngine
 from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
 
 logger = logging.getLogger(__name__)

…al_meter/_formula_generators/__init__.py …a_engine/_formula_generators/__init__.pysrc/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py src/frequenz/sdk/timeseries/logical_meter/_formula_generators/__init__.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/__init__.py

@@ -3,11 +3,11 @@
 
 """Formula generator from component graph for 3-phase Grid Current."""
 
+from __future__ import annotations
+
 import logging
-from typing import List
 
-from .....sdk import microgrid
-from ....microgrid.component import Component, ComponentCategory, ComponentMetricId
+from ....microgrid.component import ComponentMetricId
 from .._formula_engine import FormulaEngine, FormulaEngine3Phase
 from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
 
@@ -18,21 +18,16 @@ class EVChargerCurrentFormula(FormulaGenerator):
     """Create a formula engine from the component graph for calculating grid current."""
 
     async def generate(self) -> FormulaEngine3Phase:
-        """Generate a formula for calculating total ev current from the component graph.
+        """Generate a formula for calculating total EV current for given component ids.
 
         Returns:
-            A formula engine that calculates total 3-phase ev charger current values.
+            A formula engine that calculates total 3-phase EV Charger current values.
         """
-        component_graph = microgrid.get().component_graph
-        ev_chargers = [
-            comp
-            for comp in component_graph.components()
-            if comp.category == ComponentCategory.EV_CHARGER
-        ]
+        component_ids = self._config.component_ids
 
-        if not ev_chargers:
+        if not component_ids:
             logger.warning(
-                "Unable to find any EV Chargers in the component graph. "
+                "No EV Charger component IDs specified. "
                 "Subscribing to the resampling actor with a non-existing "
                 "component id, so that `0` values are sent from the formula."
             )
@@ -54,34 +49,34 @@ async def generate(self) -> FormulaEngine3Phase:
             (
                 (
                     await self._gen_phase_formula(
-                        ev_chargers, ComponentMetricId.CURRENT_PHASE_1
+                        component_ids, ComponentMetricId.CURRENT_PHASE_1
                     )
                 ).new_receiver(),
                 (
                     await self._gen_phase_formula(
-                        ev_chargers, ComponentMetricId.CURRENT_PHASE_2
+                        component_ids, ComponentMetricId.CURRENT_PHASE_2
                     )
                 ).new_receiver(),
                 (
                     await self._gen_phase_formula(
-                        ev_chargers, ComponentMetricId.CURRENT_PHASE_3
+                        component_ids, ComponentMetricId.CURRENT_PHASE_3
                     )
                 ).new_receiver(),
             ),
         )
 
     async def _gen_phase_formula(
         self,
-        ev_chargers: List[Component],
+        component_ids: set[int],
         metric_id: ComponentMetricId,
     ) -> FormulaEngine:
         builder = self._get_builder("ev-current", metric_id)
 
-        # generate a formula that just adds values from all ev-chargers.
-        for idx, comp in enumerate(ev_chargers):
+        # generate a formula that just adds values from all EV Chargers.
+        for idx, component_id in enumerate(component_ids):
             if idx > 0:
                 builder.push_oper("+")
 
-            await builder.push_component_metric(comp.component_id, nones_are_zeros=True)
+            await builder.push_component_metric(component_id, nones_are_zeros=True)
 
         return builder.build()

…ula_generators/_battery_power_formula.py …ula_generators/_battery_power_formula.pysrc/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_power_formula.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_power_formula.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_power_formula.py

@@ -5,8 +5,7 @@
 
 import logging
 
-from .....sdk import microgrid
-from ....microgrid.component import ComponentCategory, ComponentMetricId
+from ....microgrid.component import ComponentMetricId
 from .._formula_engine import FormulaEngine
 from ._formula_generator import NON_EXISTING_COMPONENT_ID, FormulaGenerator
 
@@ -17,22 +16,17 @@ class EVChargerPowerFormula(FormulaGenerator):
     """Create a formula engine from the component graph for calculating grid power."""
 
     async def generate(self) -> FormulaEngine:
-        """Generate a formula for calculating total EV power from the component graph.
+        """Generate a formula for calculating total EV power for given component ids.
 
         Returns:
-            A formula engine that calculates total EV charger power values.
+            A formula engine that calculates total EV Charger power values.
         """
         builder = self._get_builder("ev-power", ComponentMetricId.ACTIVE_POWER)
-        component_graph = microgrid.get().component_graph
-        ev_chargers = [
-            comp
-            for comp in component_graph.components()
-            if comp.category == ComponentCategory.EV_CHARGER
-        ]
-
-        if not ev_chargers:
+
+        component_ids = self._config.component_ids
+        if not component_ids:
             logger.warning(
-                "Unable to find any EV Chargers in the component graph. "
+                "No EV Charger component IDs specified. "
                 "Subscribing to the resampling actor with a non-existing "
                 "component id, so that `0` values are sent from the formula."
             )
@@ -44,10 +38,10 @@ async def generate(self) -> FormulaEngine:
             )
             return builder.build()
 
-        for idx, comp in enumerate(ev_chargers):
+        for idx, component_id in enumerate(component_ids):
             if idx > 0:
                 builder.push_oper("+")
 
-            await builder.push_component_metric(comp.component_id, nones_are_zeros=True)
+            await builder.push_component_metric(component_id, nones_are_zeros=True)
 
         return builder.build()

…rmula_generators/_battery_soc_formula.py …rmula_generators/_battery_soc_formula.pysrc/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_soc_formula.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_soc_formula.py src/frequenz/sdk/timeseries/logical_meter/_formula_generators/_battery_soc_formula.py renamed to src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_battery_soc_formula.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