Update EV formula generators to use component_ids from the ev pool

... instead of from the component graph.

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

Author: shsms

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_current_formula.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.
         """
-        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):
+        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()

src/frequenz/sdk/timeseries/_formula_engine/_formula_generators/_ev_charger_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.
         """
         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()

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py

@@ -17,6 +17,7 @@
 from .._formula_engine._formula_generators import (
     EVChargerCurrentFormula,
     EVChargerPowerFormula,
+    FormulaGeneratorConfig,
 )
 from ._state_tracker import EVChargerPoolStates, StateTracker
 
@@ -90,7 +91,9 @@ async def total_current(self) -> FormulaReceiver3Phase:
             A *new* receiver that will stream ev_charger current values.
         """
         return await self._formula_pool.from_generator(
-            "ev_charger_total_current", EVChargerCurrentFormula
+            "ev_charger_total_current",
+            EVChargerCurrentFormula,
+            FormulaGeneratorConfig(component_ids=self._component_ids),
         )
 
     async def total_power(self) -> FormulaReceiver:
@@ -104,5 +107,7 @@ async def total_power(self) -> FormulaReceiver:
             A *new* receiver that will stream ev_charger power values.
         """
         return await self._formula_pool.from_generator(
-            "ev_charger_total_power", EVChargerPowerFormula
+            "ev_charger_total_power",
+            EVChargerPowerFormula,
+            FormulaGeneratorConfig(component_ids=self._component_ids),
         )