Add functionality to force power requests

A power request might need to be forced to implement
safety mechanisms, even when some components might be
seemingly failing (i.e. when there is not proper consumption
information, the user wants to slowly discharge batteries to
prevent potential peak breaches).

Signed-off-by: Daniel Zullo <daniel.zullo@frequenz.com>

RELEASE_NOTES.md

@@ -12,6 +12,8 @@
 
 * The `LogicalMeter` no longer takes a `component_graph` parameter.
 
+* A power request can now be forced by setting the `force` attribute. This is helpful as a safety mechanism when some components might be seemingly failing, for instance, there is not proper battery metrics information.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

src/frequenz/sdk/actor/power_distributing/power_distributing.py

@@ -195,6 +195,10 @@ def __init__(
             max_data_age_sec=10.0,
         )
 
+        self._cached_metrics: dict[int, InvBatPair | None] = {
+            bat_id: None for bat_id, _ in self._bat_inv_map.items()
+        }
+
     def _create_users_tasks(self) -> List[asyncio.Task[None]]:
         """For each user create a task to wait for request.
 
@@ -208,37 +212,39 @@ def _create_users_tasks(self) -> List[asyncio.Task[None]]:
             )
         return tasks
 
-    def _get_upper_bound(self, batteries: Set[int]) -> float:
+    def _get_upper_bound(self, batteries: Set[int], use_all: bool) -> float:
         """Get total upper bound of power to be set for given batteries.
 
         Note, output of that function doesn't guarantee that this bound will be
         the same when the request is processed.
 
         Args:
             batteries: List of batteries
+            use_all: flag whether all batteries must be used for the power request.
 
         Returns:
             Upper bound for `set_power` operation.
         """
-        pairs_data: List[InvBatPair] = self._get_components_data(batteries)
+        pairs_data: List[InvBatPair] = self._get_components_data(batteries, use_all)
         return sum(
             min(battery.power_upper_bound, inverter.active_power_upper_bound)
             for battery, inverter in pairs_data
         )
 
-    def _get_lower_bound(self, batteries: Set[int]) -> float:
+    def _get_lower_bound(self, batteries: Set[int], use_all: bool) -> float:
         """Get total lower bound of power to be set for given batteries.
 
         Note, output of that function doesn't guarantee that this bound will be
         the same when the request is processed.
 
         Args:
             batteries: List of batteries
+            use_all: flag whether all batteries must be used for the power request.
 
         Returns:
             Lower bound for `set_power` operation.
         """
-        pairs_data: List[InvBatPair] = self._get_components_data(batteries)
+        pairs_data: List[InvBatPair] = self._get_components_data(batteries, use_all)
         return sum(
             max(battery.power_lower_bound, inverter.active_power_lower_bound)
             for battery, inverter in pairs_data
@@ -266,7 +272,7 @@ async def run(self) -> None:
 
             try:
                 pairs_data: List[InvBatPair] = self._get_components_data(
-                    request.batteries
+                    request.batteries, request.force
                 )
             except KeyError as err:
                 await user.channel.send(Error(request=request, msg=str(err)))
@@ -373,7 +379,7 @@ def _check_request(self, request: Request) -> Optional[Result]:
             Result for the user if the request is wrong, None otherwise.
         """
         for battery in request.batteries:
-            if battery not in self._battery_receivers:
+            if battery not in self._battery_receivers and request.force is False:
                 msg = (
                     f"No battery {battery}, available batteries: "
                     f"{list(self._battery_receivers.keys())}"
@@ -382,11 +388,11 @@ def _check_request(self, request: Request) -> Optional[Result]:
 
         if not request.adjust_power:
             if request.power < 0:
-                bound = self._get_lower_bound(request.batteries)
+                bound = self._get_lower_bound(request.batteries, request.force)
                 if request.power < bound:
                     return OutOfBound(request=request, bound=bound)
             else:
-                bound = self._get_upper_bound(request.batteries)
+                bound = self._get_upper_bound(request.batteries, request.force)
                 if request.power > bound:
                     return OutOfBound(request=request, bound=bound)
 
@@ -535,11 +541,14 @@ def _get_components_pairs(
 
         return bat_inv_map, inv_bat_map
 
-    def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
+    def _get_components_data(
+        self, batteries: Set[int], use_all: bool
+    ) -> List[InvBatPair]:
         """Get data for the given batteries and adjacent inverters.
 
         Args:
             batteries: Batteries that needs data.
+            use_all: flag whether all batteries must be used for the power request.
 
         Raises:
             KeyError: If any battery in the given list doesn't exists in microgrid.
@@ -549,11 +558,13 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
         """
         pairs_data: List[InvBatPair] = []
         working_batteries = (
-            self._all_battery_status.get_working_batteries(batteries) or batteries
+            batteries
+            if use_all
+            else self._all_battery_status.get_working_batteries(batteries) or batteries
         )
 
         for battery_id in working_batteries:
-            if battery_id not in self._battery_receivers:
+            if battery_id not in self._battery_receivers and use_all is False:
                 raise KeyError(
                     f"No battery {battery_id}, "
                     f"available batteries: {list(self._battery_receivers.keys())}"
@@ -562,6 +573,8 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
             inverter_id: int = self._bat_inv_map[battery_id]
 
             data = self._get_battery_inverter_data(battery_id, inverter_id)
+            if data is None and use_all is True:
+                data = self._cached_metrics[battery_id]
             if data is None:
                 _logger.warning(
                     "Skipping battery %d because its message isn't correct.",
@@ -629,7 +642,8 @@ def _get_battery_inverter_data(
 
         # If all values are ok then return them.
         if not any(map(isnan, replaceable_metrics)):
-            return InvBatPair(battery_data, inverter_data)
+            self._cached_metrics[battery_id] = InvBatPair(battery_data, inverter_data)
+            return self._cached_metrics[battery_id]
 
         # Replace NaN with the corresponding value in the adjacent component.
         # If both metrics are None, return None to ignore this battery.
@@ -651,10 +665,11 @@ def _get_battery_inverter_data(
             elif isnan(inv_bound):
                 inverter_new_metrics[inv_attr] = bat_bound
 
-        return InvBatPair(
+        self._cached_metrics[battery_id] = InvBatPair(
             replace(battery_data, **battery_new_metrics),
             replace(inverter_data, **inverter_new_metrics),
         )
+        return self._cached_metrics[battery_id]
 
     async def _create_channels(self) -> None:
         """Create channels to get data of components in microgrid."""

src/frequenz/sdk/actor/power_distributing/request.py

@@ -29,3 +29,6 @@ class Request:
     If `False` and the power is outside the batteries' bounds, the request will
     fail and be replied to with an `OutOfBound` result.
     """
+
+    force: bool = False
+    """Whether to force the power request regardless the status of components."""

tests/actor/test_power_distributing.py

@@ -765,10 +765,61 @@ async def test_use_all_batteries_none_is_working(
             battery_status_sender=battery_status_channel.new_sender(),
         )
 
+        for force in (False, True):
+            request = Request(
+                power=1200.0,
+                batteries={106, 206},
+                request_timeout_sec=SAFETY_TIMEOUT,
+                force=force,
+            )
+
+            await channel.client_handle.send(request)
+
+            done, pending = await asyncio.wait(
+                [asyncio.create_task(channel.client_handle.receive())],
+                timeout=SAFETY_TIMEOUT,
+            )
+
+            assert len(pending) == 0
+            assert len(done) == 1
+
+            result = done.pop().result()
+            assert isinstance(result, Success)
+            assert result.excess_power == approx(200.0)
+            assert result.succeeded_power == approx(1000.0)
+            assert result.request == request
+
+        await distributor._stop_actor()
+
+    async def test_force_request_a_battery_is_not_working(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Test force request when a battery is not working."""
+        await self.init_mock_microgrid(mocker)
+
+        channel = Bidirectional[Request, Result]("user1", "power_distributor")
+
+        batteries = {106, 206}
+
+        attrs = {"get_working_batteries.return_value": batteries - {106}}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        mocker.patch("asyncio.sleep", new_callable=AsyncMock)
+
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        distributor = PowerDistributingActor(
+            users_channels={"user1": channel.service_handle},
+            battery_status_sender=battery_status_channel.new_sender(),
+        )
+
         request = Request(
             power=1200.0,
-            batteries={106, 206},
+            batteries=batteries,
             request_timeout_sec=SAFETY_TIMEOUT,
+            force=True,
         )
 
         await channel.client_handle.send(request)
@@ -788,3 +839,78 @@ async def test_use_all_batteries_none_is_working(
         assert result.request == request
 
         await distributor._stop_actor()
+
+    async def test_battery_force_request_nan(self, mocker: MockerFixture) -> None:
+        """Test battery with NaN in SoC, capacity or power is used if request is forced."""
+        mock_microgrid = await self.init_mock_microgrid(mocker)
+
+        channel = Bidirectional[Request, Result]("user1", "power_distributor")
+
+        batteries = {106, 206, 306}
+
+        attrs = {"get_working_batteries.return_value": batteries}
+        mocker.patch(
+            "frequenz.sdk.actor.power_distributing.power_distributing.BatteryPoolStatus",
+            return_value=MagicMock(spec=BatteryPoolStatus, **attrs),
+        )
+
+        mocker.patch("asyncio.sleep", new_callable=AsyncMock)
+        battery_status_channel = Broadcast[BatteryStatus]("battery_status")
+        distributor = PowerDistributingActor(
+            {"user1": channel.service_handle},
+            battery_status_sender=battery_status_channel.new_sender(),
+        )
+
+        # The initial request is needed to set the battery metrics cache
+        request = Request(
+            power=1200.0,
+            batteries=batteries,
+            request_timeout_sec=SAFETY_TIMEOUT,
+            force=True,
+        )
+
+        await channel.client_handle.send(request)
+
+        async def test_result() -> None:
+            done, pending = await asyncio.wait(
+                [asyncio.create_task(channel.client_handle.receive())],
+                timeout=SAFETY_TIMEOUT,
+            )
+            assert len(pending) == 0
+            assert len(done) == 1
+            result: Result = done.pop().result()
+            assert isinstance(result, Success)
+            assert result.succeeded_batteries == {106, 206, 306}
+            assert result.succeeded_power == approx(1199.9999)
+            assert result.excess_power == approx(0.0)
+            assert result.request == request
+
+        await test_result()
+
+        batteries_data = (
+            battery_msg(
+                106,
+                soc=Metric(float("NaN"), Bound(20, 80)),
+                capacity=Metric(98000),
+                power=Bound(-1000, 1000),
+            ),
+            battery_msg(
+                206,
+                soc=Metric(40, Bound(20, 80)),
+                capacity=Metric(float("NaN")),
+                power=Bound(-1000, 1000),
+            ),
+            battery_msg(
+                306,
+                soc=Metric(40, Bound(20, 80)),
+                capacity=Metric(float(98000)),
+                power=Bound(float("NaN"), float("NaN")),
+            ),
+        )
+
+        for battery in batteries_data:
+            await mock_microgrid.send(battery)
+            await channel.client_handle.send(request)
+            await test_result()
+
+        await distributor._stop_actor()