Resend request if PowerDistributor returns PartialFailure

- When there is an issue with any of the batteries, the power
distributor would mark the battery as temporarily broken and update
the battery status stream.

- Changes to the battery status stream would trigger big changes to
the system bounds calculated in the battery pool.

- Changes to the system bounds would not affect the target power
unless it was very big to begin with, so we need to resend the target
power to the power distributor, whenever a partial failure is received
from the power distributor.

- In the next run, the power distributor won't use the broken battery
and will return `Success`.

- If all batteries are broken and the power distributor returns an
`Error` instead, then the bounds will go down to zero, and it will
force a target power to get sent out already.

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

Author: shsms

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

@@ -130,6 +130,7 @@ def get_target_power(
         battery_ids: frozenset[int],
         proposal: Proposal | None,
         system_bounds: PowerMetrics,
+        must_return_power: bool = False,
     ) -> Power | None:
         """Calculate and return the target power for the given batteries.
 
@@ -138,6 +139,8 @@ def get_target_power(
             proposal: If given, the proposal to added to the bucket, before the target
                 power is calculated.
             system_bounds: The system bounds for the batteries in the proposal.
+            must_return_power: If `True`, the algorithm must return a target power,
+                even if it hasn't changed since the last call.
 
         Returns:
             The new target power for the batteries, or `None` if the target power

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

@@ -80,6 +80,7 @@ def get_target_power(
         battery_ids: frozenset[int],
         proposal: Proposal | None,
         system_bounds: PowerMetrics,
+        must_return_power: bool = False,
     ) -> Power | None:
         """Calculate and return the target power for the given batteries.
 
@@ -88,6 +89,8 @@ def get_target_power(
             proposal: If given, the proposal to added to the bucket, before the target
                 power is calculated.
             system_bounds: The system bounds for the batteries in the proposal.
+            must_return_power: If `True`, the algorithm must return a target power,
+                even if it hasn't changed since the last call.
 
         Returns:
             The new target power for the batteries, or `None` if the target power
@@ -132,7 +135,8 @@ def get_target_power(
         target_power = self._calc_target_power(proposals, system_bounds)
 
         if (
-            battery_ids not in self._target_power
+            must_return_power
+            or battery_ids not in self._target_power
             or self._target_power[battery_ids] != target_power
         ):
             self._target_power[battery_ids] = target_power

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

@@ -142,14 +142,18 @@ def _add_bounds_tracker(self, battery_ids: frozenset[int]) -> None:
         )
 
     async def _send_updated_target_power(
-        self, battery_ids: frozenset[int], proposal: Proposal | None
+        self,
+        battery_ids: frozenset[int],
+        proposal: Proposal | None,
+        must_send: bool = False,
     ) -> None:
         from .. import power_distributing  # pylint: disable=import-outside-toplevel
 
         target_power = self._algorithm.get_target_power(
             battery_ids,
             proposal,
             self._system_bounds[battery_ids],
+            must_send,
         )
         request_timeout = (
             proposal.request_timeout if proposal else timedelta(seconds=5.0)
@@ -203,5 +207,15 @@ async def _run(self) -> None:
                     self._add_bounds_tracker(sub.battery_ids)
 
             elif selected_from(selected, self._power_distributing_results_receiver):
+                from .. import (  # pylint: disable=import-outside-toplevel
+                    power_distributing,
+                )
+
                 result = selected.value
                 self._distribution_results[frozenset(result.request.batteries)] = result
+                match result:
+                    case power_distributing.PartialFailure(request):
+                        await self._send_updated_target_power(
+                            frozenset(request.batteries), None, must_send=True
+                        )
+                await self._send_reports(frozenset(result.request.batteries))

tests/timeseries/_battery_pool/test_battery_pool_control_methods.py

@@ -13,8 +13,8 @@
 from frequenz.channels import Sender
 from pytest_mock import MockerFixture
 
-from frequenz.sdk import microgrid, timeseries
-from frequenz.sdk.actor import ResamplerConfig, _power_managing
+from frequenz.sdk import microgrid
+from frequenz.sdk.actor import ResamplerConfig, _power_managing, power_distributing
 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,21 +136,29 @@ async def _init_data_for_inverters(self, mocks: Mocks) -> None:
                 0.05,
             )
 
-    def _make_report(
-        self, *, power: float | None, lower: float, upper: float
-    ) -> _power_managing.Report:
-        return _power_managing.Report(
-            target_power=Power.from_watts(power) if power is not None else None,
-            inclusion_bounds=timeseries.Bounds(
-                lower=Power.from_watts(lower),
-                upper=Power.from_watts(upper),
-            ),
-            exclusion_bounds=timeseries.Bounds(
-                lower=Power.from_watts(0.0),
-                upper=Power.from_watts(0.0),
-            ),
-            distribution_result=None,
+    def _assert_report(
+        self,
+        report: _power_managing.Report,
+        *,
+        power: float | None,
+        lower: float,
+        upper: float,
+        expected_result_pred: typing.Callable[[power_distributing.Result], bool]
+        | None = None,
+    ) -> None:
+        assert report.target_power == (
+            Power.from_watts(power) if power is not None else None
         )
+        assert (
+            report.inclusion_bounds is not None and report.exclusion_bounds is not None
+        )
+        assert report.inclusion_bounds.lower == Power.from_watts(lower)
+        assert report.inclusion_bounds.upper == Power.from_watts(upper)
+        assert report.exclusion_bounds.lower == Power.from_watts(0.0)
+        assert report.exclusion_bounds.upper == Power.from_watts(0.0)
+        if expected_result_pred is not None:
+            assert report.distribution_result is not None
+            assert expected_result_pred(report.distribution_result)
 
     async def test_case_1(
         self,
@@ -160,7 +168,7 @@ async def test_case_1(
         """Test case 1.
 
         - single battery pool with all batteries.
-        - all batteries are working.
+        - all batteries are working, then one battery stops working.
         """
         set_power = typing.cast(
             AsyncMock, microgrid.connection_manager.get().api_client.set_power
@@ -179,21 +187,82 @@ async def test_case_1(
         # subsequent commit.
         bounds_rx = battery_pool.power_bounds().new_receiver()
 
-        assert await bounds_rx.receive() == self._make_report(
-            power=None, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_rx.receive(), power=None, lower=-4000.0, upper=4000.0
         )
 
         await battery_pool.set_power(Power.from_watts(1000.0))
 
-        assert await bounds_rx.receive() == self._make_report(
-            power=1000.0, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_rx.receive(), power=1000.0, lower=-4000.0, upper=4000.0
         )
 
         assert set_power.call_count == 4
         assert set_power.call_args_list == [
             mocker.call(inv_id, 250.0)
             for inv_id in mocks.microgrid.battery_inverter_ids
         ]
+        self._assert_report(
+            await bounds_rx.receive(),
+            power=1000.0,
+            lower=-4000.0,
+            upper=4000.0,
+            expected_result_pred=lambda result: isinstance(
+                result, power_distributing.Success
+            ),
+        )
+
+        set_power.reset_mock()
+
+        # First battery stops working (aka set_power never returns for it, times out).
+        async def side_effect(inv_id: int, _: float) -> None:
+            if inv_id == mocks.microgrid.battery_inverter_ids[0]:
+                await asyncio.sleep(1000.0)
+
+        set_power.side_effect = side_effect
+        await battery_pool.set_power(
+            Power.from_watts(100.0), request_timeout=timedelta(seconds=0.1)
+        )
+        self._assert_report(
+            await bounds_rx.receive(),
+            power=100.0,
+            lower=-4000.0,
+            upper=4000.0,
+            expected_result_pred=lambda result: isinstance(
+                result, power_distributing.Success
+            ),
+        )
+        assert set_power.call_count == 4
+        assert set_power.call_args_list == [
+            mocker.call(inv_id, 25.0) for inv_id in mocks.microgrid.battery_inverter_ids
+        ]
+        set_power.reset_mock()
+        self._assert_report(
+            await bounds_rx.receive(),
+            power=100.0,
+            lower=-4000.0,
+            upper=4000.0,
+            expected_result_pred=lambda result: isinstance(
+                result, power_distributing.PartialFailure
+            )
+            and result.failed_batteries == {mocks.microgrid.battery_ids[0]},
+        )
+
+        # There should be an automatic retry.
+        set_power.side_effect = None
+        assert set_power.call_count == 4
+        assert set_power.call_args_list == [
+            mocker.call(inv_id, 25.0) for inv_id in mocks.microgrid.battery_inverter_ids
+        ]
+        self._assert_report(
+            await bounds_rx.receive(),
+            power=100.0,
+            lower=-4000.0,
+            upper=4000.0,
+            expected_result_pred=lambda result: isinstance(
+                result, power_distributing.Success
+            ),
+        )
 
     async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         """Test case 2.
@@ -214,15 +283,15 @@ async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         battery_pool_2 = microgrid.battery_pool(set(mocks.microgrid.battery_ids[2:]))
         bounds_2_rx = battery_pool_2.power_bounds().new_receiver()
 
-        assert await bounds_1_rx.receive() == self._make_report(
-            power=None, lower=-2000.0, upper=2000.0
+        self._assert_report(
+            await bounds_1_rx.receive(), power=None, lower=-2000.0, upper=2000.0
         )
-        assert await bounds_2_rx.receive() == self._make_report(
-            power=None, lower=-2000.0, upper=2000.0
+        self._assert_report(
+            await bounds_2_rx.receive(), power=None, lower=-2000.0, upper=2000.0
         )
         await battery_pool_1.set_power(Power.from_watts(1000.0))
-        assert await bounds_1_rx.receive() == self._make_report(
-            power=1000.0, lower=-2000.0, upper=2000.0
+        self._assert_report(
+            await bounds_1_rx.receive(), power=1000.0, lower=-2000.0, upper=2000.0
         )
         assert set_power.call_count == 2
         assert set_power.call_args_list == [
@@ -232,8 +301,8 @@ 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))
-        assert await bounds_2_rx.receive() == self._make_report(
-            power=1000.0, lower=-2000.0, upper=2000.0
+        self._assert_report(
+            await bounds_2_rx.receive(), power=1000.0, lower=-2000.0, upper=2000.0
         )
         assert set_power.call_count == 2
         assert set_power.call_args_list == [
@@ -260,22 +329,22 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
         battery_pool_2 = microgrid.battery_pool()
         bounds_2_rx = battery_pool_2.power_bounds(1).new_receiver()
 
-        assert await bounds_1_rx.receive() == self._make_report(
-            power=None, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_1_rx.receive(), power=None, lower=-4000.0, upper=4000.0
         )
-        assert await bounds_2_rx.receive() == self._make_report(
-            power=None, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_2_rx.receive(), power=None, 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)),
         )
-        assert await bounds_1_rx.receive() == self._make_report(
-            power=-1000.0, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_1_rx.receive(), 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
+        self._assert_report(
+            await bounds_2_rx.receive(), power=-1000.0, lower=-1000.0, upper=0.0
         )
 
         assert set_power.call_count == 4
@@ -290,11 +359,11 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
             _priority=1,
             _bounds=(Power.from_watts(0.0), Power.from_watts(1000.0)),
         )
-        assert await bounds_1_rx.receive() == self._make_report(
-            power=0.0, lower=-4000.0, upper=4000.0
+        self._assert_report(
+            await bounds_1_rx.receive(), 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
+        self._assert_report(
+            await bounds_2_rx.receive(), power=0.0, lower=-1000.0, upper=0.0
         )
 
         assert set_power.call_count == 4