Add system bounds tracking for components in an EVChargerPool

This is exposed through the private `_system_power_bounds` method in
the EVChargerPool.

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

Author: shsms

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -92,6 +92,9 @@ def __init__(
         self._battery_power_wrapper = PowerWrapper(
             ComponentCategory.BATTERY, self._channel_registry
         )
+        self._ev_power_wrapper = PowerWrapper(
+            ComponentCategory.EV_CHARGER, self._channel_registry
+        )
 
         self._logical_meter: LogicalMeter | None = None
         self._consumer: Consumer | None = None
@@ -173,6 +176,9 @@ def ev_charger_pool(
         """
         from ..timeseries.ev_charger_pool import EVChargerPool
 
+        if not self._ev_power_wrapper.started:
+            self._ev_power_wrapper.start()
+
         # We use frozenset to make a hashable key from the input set.
         key: frozenset[int] = frozenset()
         if ev_charger_ids is not None:
@@ -182,6 +188,9 @@ def ev_charger_pool(
             self._ev_charger_pools[key] = EVChargerPool(
                 channel_registry=self._channel_registry,
                 resampler_subscription_sender=self._resampling_request_sender(),
+                status_receiver=self._ev_power_wrapper.status_channel.new_receiver(
+                    limit=1
+                ),
                 component_ids=ev_charger_ids,
             )
         return self._ev_charger_pools[key]

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

@@ -7,11 +7,14 @@
 import uuid
 from collections import abc
 
-from frequenz.channels import Sender
+from frequenz.channels import Broadcast, Receiver, Sender
 from frequenz.client.microgrid import ComponentCategory
 
+from ..._internal._channels import ReceiverFetcher
 from ...actor import ChannelRegistry, ComponentMetricRequest
+from ...actor.power_distributing import ComponentPoolStatus
 from ...microgrid import connection_manager
+from .._base_types import SystemBounds
 from .._quantities import Current, Power
 from ..formula_engine import FormulaEngine, FormulaEngine3Phase
 from ..formula_engine._formula_engine_pool import FormulaEnginePool
@@ -20,6 +23,7 @@
     EVChargerPowerFormula,
     FormulaGeneratorConfig,
 )
+from ._system_bounds_tracker import EVCSystemBoundsTracker
 
 
 class EVChargerPoolError(Exception):
@@ -41,10 +45,11 @@ class EVChargerPool:
         measurements of the EV Chargers in the pool.
     """
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
+        status_receiver: Receiver[ComponentPoolStatus],
         component_ids: abc.Set[int] | None = None,
     ) -> None:
         """Create an `EVChargerPool` instance.
@@ -59,6 +64,8 @@ def __init__(
                 actor.
             resampler_subscription_sender: A sender for sending metric requests to the
                 resampling actor.
+            status_receiver: A receiver that streams the status of the EV Chargers in
+                the pool.
             component_ids: An optional list of component_ids belonging to this pool.  If
                 not specified, IDs of all EV Chargers in the microgrid will be fetched
                 from the component graph.
@@ -67,6 +74,7 @@ def __init__(
         self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
             resampler_subscription_sender
         )
+        self._status_receiver: Receiver[ComponentPoolStatus] = status_receiver
         self._component_ids: abc.Set[int] = set()
         if component_ids is not None:
             self._component_ids = component_ids
@@ -85,6 +93,14 @@ def __init__(
             self._resampler_subscription_sender,
         )
 
+        self._bounds_channel: Broadcast[SystemBounds] = Broadcast(
+            name=f"System Bounds for EV Chargers: {component_ids}"
+        )
+        self._bounds_tracker: EVCSystemBoundsTracker = EVCSystemBoundsTracker(
+            self.component_ids, self._status_receiver, self._bounds_channel.new_sender()
+        )
+        self._bounds_tracker.start()
+
     @property
     def component_ids(self) -> abc.Set[int]:
         """Return component IDs of all EV Chargers managed by this EVChargerPool.
@@ -147,3 +163,8 @@ def power(self) -> FormulaEngine[Power]:
     async def stop(self) -> None:
         """Stop all tasks and channels owned by the EVChargerPool."""
         await self._formula_pool.stop()
+
+    @property
+    def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
+        """Return a receiver for the system power bounds."""
+        return self._bounds_channel