Expose power distribution results through PowerManager reporting streams

Failure messages need to trigger a resending of target power to the
PowerDistributor.  This is done in a subsequent commit.

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

Author: shsms

src/frequenz/sdk/actor/_power_managing/_base_classes.py

@@ -16,6 +16,7 @@
 
 if typing.TYPE_CHECKING:
     from ...timeseries.battery_pool import PowerMetrics
+    from .. import power_distributing
 
 
 @dataclasses.dataclass(frozen=True)
@@ -65,6 +66,12 @@ class Report:
     priorities.
     """
 
+    distribution_result: power_distributing.Result | None
+    """The result of the last power distribution.
+
+    This is `None` if no power distribution has been performed yet.
+    """
+
 
 @dataclasses.dataclass(frozen=True)
 class Proposal:
@@ -141,14 +148,19 @@ def get_target_power(
     # It can be loosened up when more algorithms are added.
     @abc.abstractmethod
     def get_status(
-        self, battery_ids: frozenset[int], priority: int, system_bounds: PowerMetrics
+        self,
+        battery_ids: frozenset[int],
+        priority: int,
+        system_bounds: PowerMetrics,
+        distribution_result: power_distributing.Result | None,
     ) -> Report:
         """Get the bounds for a set of batteries, for the given priority.
 
         Args:
             battery_ids: The IDs of the batteries to get the bounds for.
             priority: The priority of the actor for which the bounds are requested.
             system_bounds: The system bounds for the batteries.
+            distribution_result: The result of the last power distribution.
 
         Returns:
             The bounds for the batteries.

src/frequenz/sdk/actor/_power_managing/_matryoshka.py

@@ -17,6 +17,7 @@
 
 if typing.TYPE_CHECKING:
     from ...timeseries.battery_pool import PowerMetrics
+    from .. import power_distributing
 
 _logger = logging.getLogger(__name__)
 
@@ -140,22 +141,29 @@ def get_target_power(
 
     @override
     def get_status(
-        self, battery_ids: frozenset[int], priority: int, system_bounds: PowerMetrics
+        self,
+        battery_ids: frozenset[int],
+        priority: int,
+        system_bounds: PowerMetrics,
+        distribution_result: power_distributing.Result | None,
     ) -> Report:
         """Get the bounds for the algorithm.
 
         Args:
             battery_ids: The IDs of the batteries to get the bounds for.
             priority: The priority of the actor for which the bounds are requested.
             system_bounds: The system bounds for the batteries.
+            distribution_result: The result of the last power distribution.
 
         Returns:
             The target power and the available bounds for the given batteries, for
                 the given priority.
         """
         target_power = self._target_power.get(battery_ids)
         if system_bounds.inclusion_bounds is None:
-            return Report(target_power, None, system_bounds.exclusion_bounds)
+            return Report(
+                target_power, None, system_bounds.exclusion_bounds, distribution_result
+            )
 
         lower_bound = system_bounds.inclusion_bounds.lower
         upper_bound = system_bounds.inclusion_bounds.upper
@@ -178,4 +186,5 @@ def get_status(
                 lower=lower_bound, upper=upper_bound
             ),
             exclusion_bounds=system_bounds.exclusion_bounds,
+            distribution_result=distribution_result,
         )

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -34,6 +34,7 @@ def __init__(  # pylint: disable=too-many-arguments
         proposals_receiver: Receiver[Proposal],
         bounds_subscription_receiver: Receiver[ReportRequest],
         power_distributing_requests_sender: Sender[power_distributing.Request],
+        power_distributing_results_receiver: Receiver[power_distributing.Result],
         channel_registry: ChannelRegistry,
         algorithm: Algorithm = Algorithm.MATRYOSHKA,
     ):
@@ -44,6 +45,8 @@ def __init__(  # pylint: disable=too-many-arguments
             bounds_subscription_receiver: The receiver for bounds subscriptions.
             power_distributing_requests_sender: The sender for power distribution
                 requests.
+            power_distributing_results_receiver: The receiver for power distribution
+                results.
             channel_registry: The channel registry.
             algorithm: The power management algorithm to use.
 
@@ -58,12 +61,14 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._bounds_subscription_receiver = bounds_subscription_receiver
         self._power_distributing_requests_sender = power_distributing_requests_sender
+        self._power_distributing_results_receiver = power_distributing_results_receiver
         self._channel_registry = channel_registry
         self._proposals_receiver = proposals_receiver
 
         self._system_bounds: dict[frozenset[int], PowerMetrics] = {}
         self._bound_tracker_tasks: dict[frozenset[int], asyncio.Task[None]] = {}
         self._subscriptions: dict[frozenset[int], dict[int, Sender[Report]]] = {}
+        self._distribution_results: dict[frozenset[int], power_distributing.Result] = {}
 
         self._algorithm: BaseAlgorithm = Matryoshka()
 
@@ -80,7 +85,13 @@ async def _send_reports(self, battery_ids: frozenset[int]) -> None:
             _logger.warning("PowerManagingActor: No bounds for %s", battery_ids)
             return
         for priority, sender in self._subscriptions.get(battery_ids, {}).items():
-            await sender.send(self._algorithm.get_status(battery_ids, priority, bounds))
+            status = self._algorithm.get_status(
+                battery_ids,
+                priority,
+                bounds,
+                self._distribution_results.get(battery_ids),
+            )
+            await sender.send(status)
 
     async def _bounds_tracker(
         self,
@@ -161,7 +172,9 @@ async def _send_updated_target_power(
     async def _run(self) -> None:
         """Run the power managing actor."""
         async for selected in select(
-            self._proposals_receiver, self._bounds_subscription_receiver
+            self._proposals_receiver,
+            self._bounds_subscription_receiver,
+            self._power_distributing_results_receiver,
         ):
             if selected_from(selected, self._proposals_receiver):
                 proposal = selected.value
@@ -188,3 +201,7 @@ async def _run(self) -> None:
 
                 if sub.battery_ids not in self._bound_tracker_tasks:
                     self._add_bounds_tracker(sub.battery_ids)
+
+            elif selected_from(selected, self._power_distributing_results_receiver):
+                result = selected.value
+                self._distribution_results[frozenset(result.request.batteries)] = result

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -257,6 +257,9 @@ def _start_power_managing_actor(self) -> None:
             power_distributing_requests_sender=(
                 self._power_distribution_requests_channel.new_sender()
             ),
+            power_distributing_results_receiver=(
+                self._power_distribution_results_channel.new_receiver()
+            ),
             channel_registry=self._channel_registry,
         )
         self._power_managing_actor.start()

tests/actor/_power_managing/test_matryoshka.py

@@ -54,7 +54,7 @@ def test_tgt_power(
     def test_bounds(
         priority: int, expected_power: float, expected_bounds: tuple[float, float]
     ) -> None:
-        report = algorithm.get_status(batteries, priority, system_bounds)
+        report = algorithm.get_status(batteries, priority, system_bounds, None)
         assert report.target_power is not None and report.inclusion_bounds is not None
         assert report.target_power.as_watts() == expected_power
         assert report.inclusion_bounds.lower.as_watts() == expected_bounds[0]

tests/timeseries/_battery_pool/test_battery_pool_control_methods.py

@@ -149,6 +149,7 @@ def _make_report(
                 lower=Power.from_watts(0.0),
                 upper=Power.from_watts(0.0),
             ),
+            distribution_result=None,
         )
 
     async def test_case_1(