impl inprogress

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

Author: shsms

src/frequenz/sdk/microgrid/_power_managing/_shifting_matryoshka.py

@@ -26,6 +26,8 @@
 from frequenz.quantities import Power
 from typing_extensions import override
 
+from frequenz.sdk.timeseries._base_types import Bounds
+
 from ... import timeseries
 from . import _bounds
 from ._base_classes import BaseAlgorithm, Proposal, _Report
@@ -37,24 +39,26 @@
 
 
 class ShiftingMatryoshka(BaseAlgorithm):
-    """The matryoshka algorithm."""
+    """The ShiftingMatryoshka algorithm."""
 
     def __init__(self, max_proposal_age: timedelta) -> None:
         """Create a new instance of the matryoshka algorithm."""
         self._max_proposal_age_sec = max_proposal_age.total_seconds()
         self._component_buckets: dict[frozenset[int], set[Proposal]] = {}
         self._target_power: dict[frozenset[int], Power] = {}
 
-    def _calc_target_power(
+    def _calc_targets(
         self,
         proposals: set[Proposal],
         system_bounds: SystemBounds,
-    ) -> Power:
-        """Calculate the target power for the given components.
+        priority: int | None = None,
+    ) -> tuple[Power, Bounds[Power]]:
+        """Calculate the target power and bounds for the given components.
 
         Args:
             proposals: The proposals for the given components.
             system_bounds: The system bounds for the components in the proposal.
+            priority: The priority of the actor for which the target power is calculated.
 
         Returns:
             The new target power for the components.
@@ -78,64 +82,73 @@ def _calc_target_power(
             system_bounds.exclusion_bounds.lower != Power.zero()
             or system_bounds.exclusion_bounds.upper != Power.zero()
         ):
-            exclusion_bounds = system_bounds.exclusion_bounds
+            exclusion_bounds = Bounds(
+                system_bounds.exclusion_bounds.lower,
+                system_bounds.exclusion_bounds.upper,
+            )
 
-        unshifted_power = Power.zero()
         target_power = Power.zero()
         for next_proposal in sorted(proposals, reverse=True):
+            unshifted_power = Power.zero()
+            if priority is not None and next_proposal.priority <= priority:
+                break
+
             if upper_bound < lower_bound:
                 break
-            if next_proposal.preferred_power:
-                match _bounds.clamp_to_bounds(
-                    next_proposal.preferred_power,
+
+            proposal_lower = next_proposal.bounds.lower or lower_bound
+            proposal_upper = next_proposal.bounds.upper or upper_bound
+            proposal_power = next_proposal.preferred_power
+
+            if proposal_upper < proposal_lower:
+                continue
+
+            if proposal_power and (
+                proposal_power < proposal_lower or proposal_power > proposal_upper
+            ):
+                continue
+
+            if proposal_lower >= upper_bound and proposal_power:
+                proposal_power = upper_bound
+            elif proposal_upper <= lower_bound and proposal_power:
+                proposal_power = lower_bound
+            else:
+                lower_bound = max(lower_bound, proposal_lower)
+                upper_bound = min(upper_bound, proposal_upper)
+
+            if proposal_power:
+                clamped = _bounds.clamp_to_bounds(
+                    proposal_power,
                     lower_bound,
                     upper_bound,
                     exclusion_bounds,
-                ):
+                )
+                match clamped:
                     case (None, power) | (power, None) if power:
                         unshifted_power = power
                     case (power_low, power_high) if power_low and power_high:
-                        if (
-                            power_high - next_proposal.preferred_power
-                            < next_proposal.preferred_power - power_low
-                        ):
+                        if power_high - proposal_power < proposal_power - power_low:
                             unshifted_power = power_high
                         else:
                             unshifted_power = power_low
                     case _:
                         pass
 
-            # Shift the bounds if no clamping bounds are specified.
-            if (
-                next_proposal.bounds.lower is None
-                and next_proposal.bounds.upper is None
-                and next_proposal.preferred_power is not None
-            ):
-                lower_bound = lower_bound - unshifted_power
-                upper_bound = upper_bound - unshifted_power
-                target_power += unshifted_power
-                unshifted_power = Power.zero()
-                continue
-
-            proposal_lower = next_proposal.bounds.lower or lower_bound
-            proposal_upper = next_proposal.bounds.upper or upper_bound
-            # If the bounds from the current proposal are fully within the exclusion
-            # bounds, then don't use them to narrow the bounds further. This allows
-            # subsequent proposals to not be blocked by the current proposal.
-            match _bounds.check_exclusion_bounds_overlap(
-                proposal_lower, proposal_upper, exclusion_bounds
-            ):
-                case (True, True):
-                    continue
-            lower_bound = max(lower_bound, proposal_lower)
-            upper_bound = min(upper_bound, proposal_upper)
             lower_bound, upper_bound = _bounds.adjust_exclusion_bounds(
                 lower_bound, upper_bound, exclusion_bounds
             )
 
-        target_power += unshifted_power
+            lower_bound = lower_bound - unshifted_power
+            upper_bound = upper_bound - unshifted_power
+            target_power += unshifted_power
+
+            if exclusion_bounds is not None:
+                exclusion_bounds = Bounds[Power](
+                    exclusion_bounds.lower - unshifted_power,
+                    exclusion_bounds.upper - unshifted_power,
+                )
 
-        return target_power
+        return target_power, Bounds[Power](lower=lower_bound, upper=upper_bound)
 
     def _validate_component_ids(
         self,
@@ -217,7 +230,7 @@ def calculate_target_power(
         if proposals is None:
             return None
 
-        target_power = self._calc_target_power(proposals, system_bounds)
+        target_power, _ = self._calc_targets(proposals, system_bounds)
 
         if (
             must_return_power
@@ -247,77 +260,13 @@ def get_status(  # pylint: disable=too-many-locals
                 the given priority.
         """
         target_power = self._target_power.get(component_ids)
-        if system_bounds.inclusion_bounds is None:
-            return _Report(
-                target_power=target_power,
-                _inclusion_bounds=None,
-                _exclusion_bounds=system_bounds.exclusion_bounds,
-            )
-
-        lower_bound = system_bounds.inclusion_bounds.lower
-        upper_bound = system_bounds.inclusion_bounds.upper
-
-        exclusion_bounds = None
-        if system_bounds.exclusion_bounds is not None and (
-            system_bounds.exclusion_bounds.lower != Power.zero()
-            or system_bounds.exclusion_bounds.upper != Power.zero()
-        ):
-            exclusion_bounds = system_bounds.exclusion_bounds
-
-        limited_power = Power.zero()
-        for next_proposal in sorted(
-            self._component_buckets.get(component_ids, []), reverse=True
-        ):
-            if next_proposal.priority <= priority:
-                break
-            if next_proposal.preferred_power:
-                match _bounds.clamp_to_bounds(
-                    next_proposal.preferred_power,
-                    lower_bound,
-                    upper_bound,
-                    exclusion_bounds,
-                ):
-                    case (None, power) | (power, None) if power:
-                        limited_power = power
-                    case (power_low, power_high) if power_low and power_high:
-                        if (
-                            power_high - next_proposal.preferred_power
-                            < next_proposal.preferred_power - power_low
-                        ):
-                            limited_power = power_high
-                        else:
-                            limited_power = power_low
-                    case _:
-                        pass
-            # Shift the bounds if no clamping bounds are specified.
-            if (
-                next_proposal.bounds.lower is None
-                and next_proposal.bounds.upper is None
-                and next_proposal.preferred_power is not None
-            ):
-                lower_bound = lower_bound - limited_power
-                upper_bound = upper_bound - limited_power
-                limited_power = Power.zero()
-                continue
-            proposal_lower = next_proposal.bounds.lower or lower_bound
-            proposal_upper = next_proposal.bounds.upper or upper_bound
-            match _bounds.check_exclusion_bounds_overlap(
-                proposal_lower, proposal_upper, exclusion_bounds
-            ):
-                case (True, True):
-                    continue
-            calc_lower_bound = max(lower_bound, proposal_lower)
-            calc_upper_bound = min(upper_bound, proposal_upper)
-            if calc_lower_bound <= calc_upper_bound:
-                lower_bound, upper_bound = _bounds.adjust_exclusion_bounds(
-                    calc_lower_bound, calc_upper_bound, exclusion_bounds
-                )
-            else:
-                break
+        _, bounds = self._calc_targets(
+            self._component_buckets.get(component_ids, set()), system_bounds, priority
+        )
         return _Report(
             target_power=target_power,
             _inclusion_bounds=timeseries.Bounds[Power](
-                lower=lower_bound, upper=upper_bound
+                lower=bounds.lower, upper=bounds.upper
             ),
             _exclusion_bounds=system_bounds.exclusion_bounds,
         )