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 <[email protected]>

Author: daniel-zullo-frequenz

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,38 +212,40 @@ def _create_users_tasks(self) -> List[asyncio.Task[None]]:
             )
         return tasks
 
-    def _get_upper_bound(self, batteries: Set[int]) -> int:
+    def _get_upper_bound(self, batteries: Set[int], use_all: bool) -> int:
         """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)
         bound = sum(
             min(battery.power_upper_bound, inverter.active_power_upper_bound)
             for battery, inverter in pairs_data
         )
         return floor(bound)
 
-    def _get_lower_bound(self, batteries: Set[int]) -> int:
+    def _get_lower_bound(self, batteries: Set[int], use_all: bool) -> int:
         """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)
         bound = sum(
             max(battery.power_lower_bound, inverter.active_power_lower_bound)
             for battery, inverter in pairs_data
@@ -268,7 +274,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)))
@@ -375,7 +381,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())}"
@@ -384,11 +390,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)
 
@@ -537,11 +543,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.
@@ -551,11 +560,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())}"
@@ -564,6 +575,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.",
@@ -631,7 +644,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.
@@ -653,10 +667,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."""