Expose bounds from PowerManager through the BatteryPool

Also update the battery pool control method tests to use the new
bounds streams.

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

Author: shsms

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

@@ -3,11 +3,12 @@
 
 """A power manager implementation."""
 
-from ._base_classes import Algorithm, Proposal, Report, ReportRequest
+from ._base_classes import Algorithm, Bounds, Proposal, Report, ReportRequest
 from ._power_managing_actor import PowerManagingActor
 
 __all__ = [
     "Algorithm",
+    "Bounds",
     "PowerManagingActor",
     "Proposal",
     "Report",

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -218,6 +218,9 @@ def battery_pool(
                 power_manager_requests_sender=(
                     self._power_management_proposals_channel.new_sender()
                 ),
+                power_manager_bounds_subscription_sender=(
+                    self._power_manager_bounds_subscription_channel.new_sender()
+                ),
                 power_manager_results_receiver=(
                     self._power_distribution_results_channel.new_receiver()
                 ),

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

@@ -14,8 +14,7 @@
 
 from ..._internal._asyncio import cancel_and_await
 from ..._internal._channels import ReceiverFetcher
-from ...actor import ChannelRegistry, ComponentMetricRequest
-from ...actor._power_managing import Proposal
+from ...actor import ComponentMetricRequest, _channel_registry, _power_managing
 from ...actor.power_distributing._battery_pool_status import BatteryStatus
 from ...actor.power_distributing.result import Result
 from ...microgrid import connection_manager
@@ -38,7 +37,7 @@
 from ._result_types import PowerMetrics
 
 
-class BatteryPool:
+class BatteryPool:  # pylint: disable=too-many-instance-attributes
     """Calculate high level metrics for a pool of the batteries.
 
     BatterPool accepts subset of the battery ids and provides methods methods for
@@ -47,10 +46,11 @@ class BatteryPool:
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
-        channel_registry: ChannelRegistry,
+        channel_registry: _channel_registry.ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
         batteries_status_receiver: Receiver[BatteryStatus],
-        power_manager_requests_sender: Sender[Proposal],
+        power_manager_requests_sender: Sender[_power_managing.Proposal],
+        power_manager_bounds_subscription_sender: Sender[_power_managing.ReportRequest],
         power_manager_results_receiver: Receiver[Result],
         min_update_interval: timedelta,
         batteries_id: Set[int] | None = None,
@@ -70,6 +70,8 @@ def __init__(  # pylint: disable=too-many-arguments
                 battery.
             power_manager_requests_sender: A Channel sender for sending power
                 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_manager_results_receiver: A Channel receiver for receiving results
                 from the power managing actor.
             min_update_interval: Some metrics in BatteryPool are send only when they
@@ -101,14 +103,19 @@ def __init__(  # pylint: disable=too-many-arguments
         self._min_update_interval = min_update_interval
 
         self._power_manager_requests_sender = power_manager_requests_sender
+        self._power_manager_bounds_subscription_sender = (
+            power_manager_bounds_subscription_sender
+        )
         self._power_manager_results_receiver = power_manager_results_receiver
 
         self._active_methods: dict[str, MetricAggregator[Any]] = {}
+        self._power_bounds_subs: dict[str, asyncio.Task[None]] = {}
         self._namespace: str = f"battery-pool-{self._batteries}-{uuid.uuid4()}"
         self._power_distributing_namespace: str = f"power-distributor-{self._namespace}"
+        self._channel_registry = channel_registry
         self._formula_pool: FormulaEnginePool = FormulaEnginePool(
             self._namespace,
-            channel_registry,
+            self._channel_registry,
             resampler_subscription_sender,
         )
 
@@ -148,7 +155,7 @@ async def set_power(
             _priority: The priority of the actor making the request.
         """
         await self._power_manager_requests_sender.send(
-            Proposal(
+            _power_managing.Proposal(
                 source_id=self._namespace,
                 preferred_power=preferred_power,
                 bounds=_bounds,
@@ -191,7 +198,7 @@ async def charge(
         if power < Power.zero():
             raise ValueError("Charge power must be positive.")
         await self._power_manager_requests_sender.send(
-            Proposal(
+            _power_managing.Proposal(
                 source_id=self._namespace,
                 preferred_power=power,
                 bounds=None,
@@ -234,7 +241,7 @@ async def discharge(
         if power < Power.zero():
             raise ValueError("Discharge power must be positive.")
         await self._power_manager_requests_sender.send(
-            Proposal(
+            _power_managing.Proposal(
                 source_id=self._namespace,
                 preferred_power=power,
                 bounds=None,
@@ -447,6 +454,28 @@ def capacity(self) -> ReceiverFetcher[Sample[Energy]]:
 
         return self._active_methods[method_name]
 
+    def power_bounds(
+        self, priority: int = 0
+    ) -> ReceiverFetcher[_power_managing.Report]:
+        """Get a receiver to receive new power bounds when they change.
+
+        These bounds are the bounds specified the power manager for actors with the
+        given priority.
+
+        Args:
+            priority: The priority of the actor to get the power bounds for.
+
+        Returns:
+            A receiver that will receive the power bounds for the given priority.
+        """
+        sub = _power_managing.ReportRequest(
+            source_id=self._namespace, priority=priority, battery_ids=self._batteries
+        )
+        self._power_bounds_subs[sub.get_channel_name()] = asyncio.create_task(
+            self._power_manager_bounds_subscription_sender.send(sub)
+        )
+        return self._channel_registry.new_receiver_fetcher(sub.get_channel_name())
+
     @property
     def _system_power_bounds(self) -> ReceiverFetcher[PowerMetrics]:
         """Get receiver to receive new power bounds when they change.

tests/timeseries/_battery_pool/test_battery_pool_control_methods.py

@@ -14,7 +14,7 @@
 from pytest_mock import MockerFixture
 
 from frequenz.sdk import microgrid
-from frequenz.sdk.actor import ResamplerConfig
+from frequenz.sdk.actor import ResamplerConfig, _power_managing
 from frequenz.sdk.actor.power_distributing import BatteryStatus
 from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryPoolStatus
 from frequenz.sdk.timeseries import Power
@@ -136,6 +136,17 @@ async def _init_data_for_inverters(self, mocks: Mocks) -> None:
                 0.05,
             )
 
+    def _make_report(
+        self, *, power: float, lower: float, upper: float
+    ) -> _power_managing.Report:
+        return _power_managing.Report(
+            target_power=Power.from_watts(power),
+            available_bounds=_power_managing.Bounds(
+                lower=Power.from_watts(lower),
+                upper=Power.from_watts(upper),
+            ),
+        )
+
     async def test_case_1(
         self,
         mocks: Mocks,
@@ -161,10 +172,17 @@ async def test_case_1(
         #
         # It will be replaced by a reporting streaming from the PowerManager in a
         # subsequent commit.
-        results_rx = battery_pool.power_distribution_results()
+        bounds_rx = battery_pool.power_bounds().new_receiver()
+
+        assert await bounds_rx.receive() == self._make_report(
+            power=0.0, lower=-4000.0, upper=4000.0
+        )
 
         await battery_pool.set_power(Power.from_watts(1000.0))
-        await results_rx.receive()
+
+        assert await bounds_rx.receive() == self._make_report(
+            power=1000.0, lower=-4000.0, upper=4000.0
+        )
 
         assert set_power.call_count == 4
         assert set_power.call_args_list == [
@@ -187,11 +205,20 @@ async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         await self._init_data_for_inverters(mocks)
 
         battery_pool_1 = microgrid.battery_pool(set(mocks.microgrid.battery_ids[:2]))
+        bounds_1_rx = battery_pool_1.power_bounds().new_receiver()
         battery_pool_2 = microgrid.battery_pool(set(mocks.microgrid.battery_ids[2:]))
+        bounds_2_rx = battery_pool_2.power_bounds().new_receiver()
 
-        results_rx = battery_pool_1.power_distribution_results()
+        assert await bounds_1_rx.receive() == self._make_report(
+            power=0.0, lower=-2000.0, upper=2000.0
+        )
+        assert await bounds_2_rx.receive() == self._make_report(
+            power=0.0, lower=-2000.0, upper=2000.0
+        )
         await battery_pool_1.set_power(Power.from_watts(1000.0))
-        await results_rx.receive()
+        assert await bounds_1_rx.receive() == self._make_report(
+            power=1000.0, lower=-2000.0, upper=2000.0
+        )
         assert set_power.call_count == 2
         assert set_power.call_args_list == [
             mocker.call(inv_id, 500.0)
@@ -200,7 +227,9 @@ async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         set_power.reset_mock()
 
         await battery_pool_2.set_power(Power.from_watts(1000.0))
-        await results_rx.receive()
+        assert await bounds_2_rx.receive() == self._make_report(
+            power=1000.0, lower=-2000.0, upper=2000.0
+        )
         assert set_power.call_count == 2
         assert set_power.call_args_list == [
             mocker.call(inv_id, 500.0)
@@ -222,15 +251,28 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
         await self._init_data_for_inverters(mocks)
 
         battery_pool_1 = microgrid.battery_pool()
+        bounds_1_rx = battery_pool_1.power_bounds(2).new_receiver()
         battery_pool_2 = microgrid.battery_pool()
+        bounds_2_rx = battery_pool_2.power_bounds(1).new_receiver()
 
-        results_rx = battery_pool_1.power_distribution_results()
+        assert await bounds_1_rx.receive() == self._make_report(
+            power=0.0, lower=-4000.0, upper=4000.0
+        )
+        assert await bounds_2_rx.receive() == self._make_report(
+            power=0.0, lower=-4000.0, upper=4000.0
+        )
         await battery_pool_1.set_power(
             Power.from_watts(-1000.0),
             _priority=2,
             _bounds=(Power.from_watts(-1000.0), Power.from_watts(0.0)),
         )
-        await results_rx.receive()
+        assert await bounds_1_rx.receive() == self._make_report(
+            power=-1000.0, lower=-4000.0, upper=4000.0
+        )
+        assert await bounds_2_rx.receive() == self._make_report(
+            power=-1000.0, lower=-1000.0, upper=0.0
+        )
+
         assert set_power.call_count == 4
         assert set_power.call_args_list == [
             mocker.call(inv_id, -250.0)
@@ -243,7 +285,13 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
             _priority=1,
             _bounds=(Power.from_watts(0.0), Power.from_watts(1000.0)),
         )
-        await results_rx.receive()
+        assert await bounds_1_rx.receive() == self._make_report(
+            power=0.0, lower=-4000.0, upper=4000.0
+        )
+        assert await bounds_2_rx.receive() == self._make_report(
+            power=0.0, lower=-1000.0, upper=0.0
+        )
+
         assert set_power.call_count == 4
         assert set_power.call_args_list == [
             mocker.call(inv_id, 0.0) for inv_id in mocks.microgrid.battery_inverter_ids