Add power_status method to EVChargerPool

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

Author: shsms

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -202,6 +202,9 @@ def ev_charger_pool(
                 status_receiver=self._ev_power_wrapper.status_channel.new_receiver(
                     maxsize=1
                 ),
+                power_manager_bounds_subs_sender=(
+                    self._ev_power_wrapper.bounds_subscription_channel.new_sender()
+                ),
                 component_ids=ev_charger_ids,
             )
         return EVChargerPool(self._ev_charger_pools[key], name, priority)

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

@@ -4,8 +4,10 @@
 """Interactions with EV Chargers."""
 
 from ._ev_charger_pool import EVChargerPool, EVChargerPoolError
+from ._result_types import EVChargerPoolReport
 
 __all__ = [
     "EVChargerPool",
     "EVChargerPoolError",
+    "EVChargerPoolReport",
 ]

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

@@ -4,10 +4,13 @@
 """Interactions with pools of EV Chargers."""
 
 
+import asyncio
+import typing
 import uuid
 from collections import abc
 
 from ..._internal._channels import ReceiverFetcher
+from ...actor import _power_managing
 from .._base_types import SystemBounds
 from .._quantities import Current, Power
 from ..formula_engine import FormulaEngine, FormulaEngine3Phase
@@ -17,6 +20,7 @@
     FormulaGeneratorConfig,
 )
 from ._ev_charger_pool_reference_store import EVChargerPoolReferenceStore
+from ._result_types import EVChargerPoolReport
 
 
 class EVChargerPoolError(Exception):
@@ -58,7 +62,7 @@ def __init__(  # pylint: disable=too-many-arguments
             priority: The priority of the actor using this wrapper.
         """
         self._ev_charger_pool = ev_charger_pool_ref
-        unique_id = uuid.uuid4()
+        unique_id = str(uuid.uuid4())
         self._source_id = unique_id if name is None else f"{name}-{unique_id}"
         self._priority = priority
 
@@ -125,6 +129,38 @@ def power(self) -> FormulaEngine[Power]:
         assert isinstance(engine, FormulaEngine)
         return engine
 
+    @property
+    def power_status(self) -> ReceiverFetcher[EVChargerPoolReport]:
+        """Get a receiver to receive new power status reports when they change.
+
+        These include
+          - the current inclusion/exclusion bounds available for the pool's priority,
+          - the current target power for the pool's set of batteries,
+          - the result of the last distribution request for the pool's set of batteries.
+
+        Returns:
+            A receiver that will stream power status reports for the pool's priority.
+        """
+        sub = _power_managing.ReportRequest(
+            source_id=self._source_id,
+            priority=self._priority,
+            component_ids=self._ev_charger_pool.component_ids,
+        )
+        self._ev_charger_pool.power_bounds_subs[sub.get_channel_name()] = (
+            asyncio.create_task(
+                self._ev_charger_pool.power_manager_bounds_subs_sender.send(sub)
+            )
+        )
+        channel = self._ev_charger_pool.channel_registry.get_or_create(
+            _power_managing._Report,  # pylint: disable=protected-access
+            sub.get_channel_name(),
+        )
+        channel.resend_latest = True
+
+        # More details on why the cast is needed here:
+        # https://github.com/frequenz-floss/frequenz-sdk-python/issues/823
+        return typing.cast(ReceiverFetcher[EVChargerPoolReport], channel)
+
     async def stop(self) -> None:
         """Stop all tasks and channels owned by the EVChargerPool."""
         await self._ev_charger_pool.stop()

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

@@ -4,12 +4,14 @@
 """Manages shared state/tasks for a set of EV chargers."""
 
 
+import asyncio
 import uuid
 from collections import abc
 
 from frequenz.channels import Broadcast, Receiver, Sender
 
 from ...actor import ChannelRegistry, ComponentMetricRequest
+from ...actor._power_managing import ReportRequest
 from ...actor.power_distributing import ComponentPoolStatus
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory
@@ -31,11 +33,12 @@ class EVChargerPoolReferenceStore:
     They are exposed through the EVChargerPool class.
     """
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
         status_receiver: Receiver[ComponentPoolStatus],
+        power_manager_bounds_subs_sender: Sender[ReportRequest],
         component_ids: abc.Set[int] | None = None,
     ):
         """Create an instance of the class.
@@ -47,13 +50,16 @@ def __init__(
                 resampling actor.
             status_receiver: A receiver that streams the status of the EV Chargers in
                 the pool.
+            power_manager_bounds_subs_sender: A Channel sender for sending power bounds
+                subscription requests to the power managing actor.
             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.
         """
         self.channel_registry = channel_registry
         self.resampler_subscription_sender = resampler_subscription_sender
         self.status_receiver = status_receiver
+        self.power_manager_bounds_subs_sender = power_manager_bounds_subs_sender
 
         if component_ids is not None:
             self.component_ids: frozenset[int] = frozenset(component_ids)
@@ -68,6 +74,8 @@ def __init__(
                 }
             )
 
+        self.power_bounds_subs: dict[str, asyncio.Task[None]] = {}
+
         self.namespace: str = f"ev-charger-pool-{uuid.uuid4()}"
         self.formula_pool = FormulaEnginePool(
             self.namespace,

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

@@ -0,0 +1,36 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Types for exposing EV charger pool reports."""
+
+import typing
+
+from ...actor import power_distributing
+from .._base_types import Bounds
+from .._quantities import Power
+
+
+class EVChargerPoolReport(typing.Protocol):
+    """A status report for an EV chargers pool."""
+
+    target_power: Power | None
+    """The currently set power for the batteries."""
+
+    distribution_result: power_distributing.Result | None
+    """The result of the last power distribution.
+
+    This is `None` if no power distribution has been performed yet.
+    """
+
+    @property
+    def bounds(self) -> Bounds[Power] | None:
+        """The usable bounds for the batteries.
+
+        These bounds are adjusted to any restrictions placed by actors with higher
+        priorities.
+
+        There might be exclusion zones within these bounds. If necessary, the
+        [`adjust_to_bounds`][frequenz.sdk.timeseries.battery_pool.BatteryPoolReport.adjust_to_bounds]
+        method may be used to check if a desired power value fits the bounds, or to get
+        the closest possible power values that do fit the bounds.
+        """