Expose power distribution results directly from the battery pool

This would stream results only for the batteries that are part of the
pool.

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

Author: shsms

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -420,6 +420,9 @@ def new_battery_pool(
                     power_manager_bounds_subscription_sender=(
                         self._battery_power_wrapper.bounds_subscription_channel.new_sender()
                     ),
+                    power_distribution_results_fetcher=(
+                        self._battery_power_wrapper.distribution_results_fetcher()
+                    ),
                     min_update_interval=self._resampler_config.resampling_period,
                     batteries_id=component_ids,
                 )

src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

@@ -13,8 +13,8 @@
 from collections import abc
 
 from ... import timeseries
-from ..._internal._channels import ReceiverFetcher
-from ...actor import _power_managing
+from ..._internal._channels import ReceiverFetcher, ReceiverFetcherWith
+from ...actor import _power_managing, power_distributing
 from ...timeseries import Energy, Percentage, Power, Sample, Temperature
 from .._base_types import SystemBounds
 from ..formula_engine import FormulaEngine
@@ -384,6 +384,21 @@ def power_status(self) -> ReceiverFetcher[BatteryPoolReport]:
 
         return channel
 
+    @property
+    def power_distribution_results(self) -> ReceiverFetcher[power_distributing.Result]:
+        """Get a receiver to receive power distribution results.
+
+        Returns:
+            A receiver that will stream power distribution results for the pool's set of
+            batteries.
+        """
+        return ReceiverFetcherWith(
+            self._pool_ref_store._power_dist_results_fetcher,
+            lambda recv: recv.filter(
+                lambda x: x.request.component_ids == self._pool_ref_store._batteries
+            ),
+        )
+
     @property
     def _system_power_bounds(self) -> ReceiverFetcher[SystemBounds]:
         """Get receiver to receive new power bounds when they change.

src/frequenz/sdk/timeseries/battery_pool/_battery_pool_reference_store.py

@@ -14,9 +14,11 @@
 from frequenz.client.microgrid import ComponentCategory
 
 from ..._internal._asyncio import cancel_and_await
+from ..._internal._channels import ReceiverFetcher
 from ...actor._channel_registry import ChannelRegistry
 from ...actor._data_sourcing._component_metric_request import ComponentMetricRequest
 from ...actor._power_managing._base_classes import Proposal, ReportRequest
+from ...actor.power_distributing import Result
 from ...actor.power_distributing._component_status import ComponentPoolStatus
 from ...microgrid import connection_manager
 from ..formula_engine._formula_engine_pool import FormulaEnginePool
@@ -43,6 +45,7 @@ def __init__(  # pylint: disable=too-many-arguments
         batteries_status_receiver: Receiver[ComponentPoolStatus],
         power_manager_requests_sender: Sender[Proposal],
         power_manager_bounds_subscription_sender: Sender[ReportRequest],
+        power_distribution_results_fetcher: ReceiverFetcher[Result],
         min_update_interval: timedelta,
         batteries_id: Set[int] | None = None,
     ) -> None:
@@ -63,6 +66,8 @@ def __init__(  # pylint: disable=too-many-arguments
                 requests to the power managing actor.
             power_manager_bounds_subscription_sender: A Channel sender for sending
                 power bounds requests to the power managing actor.
+            power_distribution_results_fetcher: A ReceiverFetcher for the results from
+                the power distributing actor.
             min_update_interval: Some metrics in BatteryPool are send only when they
                 change. For these metrics min_update_interval is the minimum time
                 interval between the following messages.
@@ -105,6 +110,9 @@ def __init__(  # pylint: disable=too-many-arguments
         self._namespace: str = f"battery-pool-{self._batteries}-{uuid.uuid4()}"
         self._power_distributing_namespace: str = f"power-distributor-{self._namespace}"
         self._channel_registry: ChannelRegistry = channel_registry
+        self._power_dist_results_fetcher: ReceiverFetcher[Result] = (
+            power_distribution_results_fetcher
+        )
         self._formula_pool: FormulaEnginePool = FormulaEnginePool(
             self._namespace,
             self._channel_registry,