Handle proposals from power shifting actors in the PowerManager

This is done by having an additional `Matryoshka` instance for
resolving requests from shifting actors.  And the target power thus
obtained is added to the target power from the regular actors, which
in effect, shifts the target power of the regular actors by the target
power of the shifting actors.

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

Author: shsms

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

@@ -16,7 +16,8 @@
 from frequenz.client.microgrid import ComponentCategory, ComponentType, InverterType
 from typing_extensions import override
 
-from ...timeseries._base_types import SystemBounds
+from ...timeseries import Power
+from ...timeseries._base_types import Bounds, SystemBounds
 from .._actor import Actor
 from .._channel_registry import ChannelRegistry
 from ._base_classes import Algorithm, BaseAlgorithm, Proposal, ReportRequest, _Report
@@ -28,7 +29,7 @@
     from .. import power_distributing
 
 
-class PowerManagingActor(Actor):
+class PowerManagingActor(Actor):  # pylint: disable=too-many-instance-attributes
     """The power manager."""
 
     def __init__(  # pylint: disable=too-many-arguments
@@ -84,10 +85,18 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._system_bounds: dict[frozenset[int], SystemBounds] = {}
         self._bound_tracker_tasks: dict[frozenset[int], asyncio.Task[None]] = {}
-        self._subscriptions: dict[frozenset[int], dict[int, Sender[_Report]]] = {}
+        self._non_shifting_subscriptions: dict[
+            frozenset[int], dict[int, Sender[_Report]]
+        ] = {}
+        self._shifting_subscriptions: dict[
+            frozenset[int], dict[int, Sender[_Report]]
+        ] = {}
         self._distribution_results: dict[frozenset[int], power_distributing.Result] = {}
 
-        self._algorithm: BaseAlgorithm = Matryoshka(
+        self._non_shifting_group: BaseAlgorithm = Matryoshka(
+            max_proposal_age=timedelta(seconds=60.0)
+        )
+        self._shifting_group: BaseAlgorithm = Matryoshka(
             max_proposal_age=timedelta(seconds=60.0)
         )
 
@@ -104,14 +113,29 @@ async def _send_reports(self, component_ids: frozenset[int]) -> None:
         if bounds is None:
             _logger.warning("PowerManagingActor: No bounds for %s", component_ids)
             return
-        for priority, sender in self._subscriptions.get(component_ids, {}).items():
-            status = self._algorithm.get_status(
+        for priority, sender in self._shifting_subscriptions.get(
+            component_ids, {}
+        ).items():
+            status = self._shifting_group.get_status(
                 component_ids,
                 priority,
                 bounds,
                 self._distribution_results.get(component_ids),
             )
             await sender.send(status)
+        for priority, sender in self._non_shifting_subscriptions.get(
+            component_ids, {}
+        ).items():
+            status = self._non_shifting_group.get_status(
+                component_ids,
+                priority,
+                self._calculate_remaining_bounds(
+                    bounds,
+                    self._shifting_group.get_target_power(component_ids),
+                ),
+                self._distribution_results.get(component_ids),
+            )
+            await sender.send(status)
 
     async def _bounds_tracker(
         self,
@@ -184,6 +208,116 @@ def _add_system_bounds_tracker(self, component_ids: frozenset[int]) -> None:
             self._bounds_tracker(component_ids, bounds_receiver)
         )
 
+    def _calculate_remaining_bounds(
+        self, bounds: SystemBounds, target_power: Power | None
+    ) -> SystemBounds:
+        """Calculate the remaining bounds after a target power is applied.
+
+        This is the values that can be added to the target power and still remain within
+        the actual system bounds.
+
+          | system bounds |     shifting |    shifted |
+          |               | target power |     bounds |
+          |---------------+--------------+------------|
+          | -100 to 100   |           70 | -170 to 30 |
+          | -100 to 100   |          -50 | -50 to 150 |
+
+        Args:
+            bounds: The bounds to calculate the remaining bounds from.
+            target_power: The target power to apply.
+
+        Returns:
+            The remaining bounds.
+        """
+        if target_power is None:
+            return bounds
+
+        inclusion_bounds: Bounds[Power] | None = None
+        if bounds.inclusion_bounds is not None:
+            inclusion_bounds = Bounds(
+                bounds.inclusion_bounds.lower - target_power,
+                bounds.inclusion_bounds.upper - target_power,
+            )
+        return SystemBounds(
+            timestamp=bounds.timestamp,
+            inclusion_bounds=inclusion_bounds,
+            exclusion_bounds=bounds.exclusion_bounds,
+        )
+
+    def _calculate_target_power(
+        self,
+        component_ids: frozenset[int],
+        proposal: Proposal | None,
+        must_send: bool = False,
+    ) -> Power | None:
+        """Calculate the target power for a set of components.
+
+        This is the power from the non-shifting group, shifted by the power from the
+        shifting group.
+
+        Args:
+            component_ids: The component IDs for which to calculate the target power.
+            proposal: The proposal to calculate the target power for.
+            must_send: If `True`, a new request will be sent to the PowerDistributor,
+                even if there's no change in power.
+
+        Returns:
+            The target power.
+        """
+        tgt_power_shift: Power | None = None
+        tgt_power_no_shift: Power | None = None
+        if proposal is not None:
+            if proposal.in_shifting_group:
+                tgt_power_shift = self._shifting_group.calculate_target_power(
+                    component_ids,
+                    proposal,
+                    self._system_bounds[component_ids],
+                    must_send,
+                )
+                tgt_power_no_shift = self._non_shifting_group.calculate_target_power(
+                    component_ids,
+                    None,
+                    self._calculate_remaining_bounds(
+                        self._system_bounds[component_ids], tgt_power_shift
+                    ),
+                    must_send,
+                )
+            else:
+                tgt_power_no_shift = self._non_shifting_group.calculate_target_power(
+                    component_ids,
+                    proposal,
+                    self._system_bounds[component_ids],
+                    must_send,
+                )
+                tgt_power_shift = self._shifting_group.calculate_target_power(
+                    component_ids,
+                    None,
+                    self._calculate_remaining_bounds(
+                        self._system_bounds[component_ids], tgt_power_no_shift
+                    ),
+                    must_send,
+                )
+        else:
+            tgt_power_no_shift = self._non_shifting_group.calculate_target_power(
+                component_ids,
+                None,
+                self._system_bounds[component_ids],
+                must_send,
+            )
+            tgt_power_shift = self._shifting_group.calculate_target_power(
+                component_ids,
+                None,
+                self._calculate_remaining_bounds(
+                    self._system_bounds[component_ids], tgt_power_no_shift
+                ),
+                must_send,
+            )
+        if tgt_power_shift is not None and tgt_power_no_shift is not None:
+            return tgt_power_shift + tgt_power_no_shift
+        if tgt_power_shift is not None:
+            return tgt_power_shift
+        return tgt_power_no_shift
+
     async def _send_updated_target_power(
         self,
         component_ids: frozenset[int],
@@ -192,10 +326,9 @@ async def _send_updated_target_power(
     ) -> None:
         from .. import power_distributing  # pylint: disable=import-outside-toplevel
 
-        target_power = self._algorithm.calculate_target_power(
+        target_power = self._calculate_target_power(
             component_ids,
             proposal,
-            self._system_bounds[component_ids],
             must_send,
         )
         request_timeout = (
@@ -245,22 +378,29 @@ async def _run(self) -> None:
                 sub = selected.message
                 component_ids = sub.component_ids
                 priority = sub.priority
+                in_shifting_group = sub.in_shifting_group
+
+                subs_set = (
+                    self._shifting_subscriptions
+                    if in_shifting_group
+                    else self._non_shifting_subscriptions
+                )
 
-                if component_ids not in self._subscriptions:
-                    self._subscriptions[component_ids] = {
+                if component_ids not in subs_set:
+                    subs_set[component_ids] = {
                         priority: self._channel_registry.get_or_create(
                             _Report, sub.get_channel_name()
                         ).new_sender()
                     }
-                elif priority not in self._subscriptions[component_ids]:
-                    self._subscriptions[component_ids][priority] = (
+                elif priority not in subs_set[component_ids]:
+                    subs_set[component_ids][priority] = (
                         self._channel_registry.get_or_create(
                             _Report, sub.get_channel_name()
                         ).new_sender()
                     )
 
-                if sub.component_ids not in self._bound_tracker_tasks:
-                    self._add_system_bounds_tracker(sub.component_ids)
+                if component_ids not in self._bound_tracker_tasks:
+                    self._add_system_bounds_tracker(component_ids)
 
             elif selected_from(selected, self._power_distributing_results_receiver):
                 from .. import (  # pylint: disable=import-outside-toplevel
@@ -287,4 +427,7 @@ async def _run(self) -> None:
                 await self._send_reports(frozenset(result.request.component_ids))
 
             elif selected_from(selected, drop_old_proposals_timer):
-                self._algorithm.drop_old_proposals(asyncio.get_event_loop().time())
+                self._non_shifting_group.drop_old_proposals(
+                    asyncio.get_event_loop().time()
+                )
+                self._shifting_group.drop_old_proposals(asyncio.get_event_loop().time())