Implement power_status reporting from the PV pool

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

Author: shsms

src/frequenz/sdk/timeseries/pv_pool/_pv_pool.py

@@ -3,11 +3,15 @@
 
 """Interactions with pools of PV inverters."""
 
+import asyncio
+import typing
 import uuid
 
 from ..._internal._channels import ReceiverFetcher
+from ...actor import _power_managing
 from .._base_types import SystemBounds
 from ._pv_pool_reference_store import PVPoolReferenceStore
+from ._result_types import PVPoolReport
 
 
 class PVPoolError(Exception):
@@ -47,9 +51,41 @@ def __init__(  # pylint: disable=too-many-arguments
         """
         self._pv_pool_ref = pv_pool_ref
         unique_id = uuid.uuid4()
-        self._source_id = unique_id if name is None else f"{name}-{unique_id}"
+        self._source_id = str(unique_id) if name is None else f"{name}-{unique_id}"
         self._priority = priority
 
+    @property
+    def power_status(self) -> ReceiverFetcher[PVPoolReport]:
+        """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._pv_pool_ref.component_ids,
+        )
+        self._pv_pool_ref.power_bounds_subs[sub.get_channel_name()] = (
+            asyncio.create_task(
+                self._pv_pool_ref.power_manager_bounds_subs_sender.send(sub)
+            )
+        )
+        channel = self._pv_pool_ref.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[PVPoolReport], channel)
+
     @property
     def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
         """Return a receiver fetcher for the system power bounds."""

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

@@ -0,0 +1,31 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Types for exposing PV pool reports."""
+
+import typing
+
+from ...actor import power_distributing
+from .._base_types import Bounds
+from .._quantities import Power
+
+
+class PVPoolReport(typing.Protocol):
+    """A status report for a PV pool."""
+
+    target_power: Power | None
+    """The currently set power for the PV inverters."""
+
+    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 PV inverters.
+
+        These bounds are adjusted to any restrictions placed by actors with higher
+        priorities.
+        """