Allow optionally forcing 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

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

@@ -211,6 +211,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.
 
@@ -224,37 +228,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: whether all batteries in the power request must be used.
 
         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: whether all batteries in the power request must be used.
 
         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
@@ -282,21 +288,19 @@ 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)))
                 continue
 
-            if len(pairs_data) == 0:
+            if len(pairs_data) == 0 and request.force is False:
                 error_msg = f"No data for the given batteries {str(request.batteries)}"
                 await user.channel.send(Error(request=request, msg=str(error_msg)))
                 continue
 
             try:
-                distribution = self.distribution_algorithm.distribute_power(
-                    request.power, pairs_data
-                )
+                distribution = self._get_power_distribution(request, pairs_data)
             except ValueError as err:
                 error_msg = f"Couldn't distribute power, error: {str(err)}"
                 await user.channel.send(Error(request=request, msg=str(error_msg)))
@@ -379,6 +383,34 @@ async def _set_distributed_power(
 
         return self._parse_result(tasks, distribution.distribution, timeout_sec)
 
+    def _get_power_distribution(
+        self, request: Request, inv_vat_pairs: List[InvBatPair]
+    ) -> DistributionResult:
+        """Get power distribution result for the batteries in the request.
+
+        Args:
+            request: the power request to process.
+            inv_vat_pairs: the battery and adjacent inverter data pairs.
+
+        Returns:
+            the power distribution result.
+        """
+        if request.force and len(request.batteries) != len(inv_vat_pairs):
+            # Distribute power equally for each battery in the request if
+            # there are missing components metrics
+            power_per_battery = request.power / len(request.batteries)
+            return DistributionResult(
+                distribution={
+                    self._bat_inv_map[battery_id]: power_per_battery
+                    for battery_id in request.batteries
+                },
+                remaining_power=0.0,
+            )
+
+        return self.distribution_algorithm.distribute_power(
+            request.power, inv_vat_pairs
+        )
+
     def _check_request(self, request: Request) -> Optional[Result]:
         """Check whether the given request if correct.
 
@@ -401,11 +433,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)
 
@@ -554,11 +586,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: whether all batteries in the power request must be used.
 
         Raises:
             KeyError: If any battery in the given list doesn't exists in microgrid.
@@ -568,7 +603,9 @@ 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:
@@ -581,6 +618,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.",
@@ -648,7 +687,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.
@@ -670,10 +710,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."""