Update distribution algorithm to support exclusion bounds

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

Author: shsms

src/frequenz/sdk/power/_distribution_algorithm.py

@@ -277,8 +277,11 @@ def _total_capacity(self, components: List[InvBatPair]) -> float:
         return total_capacity
 
     def _compute_battery_availability_ratio(
-        self, components: List[InvBatPair], available_soc: Dict[int, float]
-    ) -> Tuple[List[Tuple[InvBatPair, float]], float]:
+        self,
+        components: List[InvBatPair],
+        available_soc: Dict[int, float],
+        excl_bounds: Dict[int, float],
+    ) -> Tuple[List[Tuple[InvBatPair, float, float]], float]:
         r"""Compute battery ratio and the total sum of all of them.
 
         battery_availability_ratio = capacity_ratio[i] * available_soc[i]
@@ -291,6 +294,7 @@ def _compute_battery_availability_ratio(
             available_soc: How much SoC remained to reach
                 * SoC upper bound - if need to distribute consumption power
                 * SoC lower bound - if need to distribute supply power
+            excl_bounds: Exclusion bounds for each inverter
 
         Returns:
             Tuple where first argument is battery availability ratio for each
@@ -299,30 +303,35 @@ def _compute_battery_availability_ratio(
                 of all battery ratios in the list.
         """
         total_capacity = self._total_capacity(components)
-        battery_availability_ratio: List[Tuple[InvBatPair, float]] = []
+        battery_availability_ratio: List[Tuple[InvBatPair, float, float]] = []
         total_battery_availability_ratio: float = 0.0
 
         for pair in components:
-            battery = pair[0]
+            battery, inverter = pair
             capacity_ratio = battery.capacity / total_capacity
             soc_factor = pow(
                 available_soc[battery.component_id], self._distributor_exponent
             )
 
             ratio = capacity_ratio * soc_factor
-            battery_availability_ratio.append((pair, ratio))
+            battery_availability_ratio.append(
+                (pair, excl_bounds[inverter.component_id], ratio)
+            )
             total_battery_availability_ratio += ratio
 
-        battery_availability_ratio.sort(key=lambda item: item[1], reverse=True)
+        battery_availability_ratio.sort(
+            key=lambda item: (item[1], item[2]), reverse=True
+        )
 
         return battery_availability_ratio, total_battery_availability_ratio
 
-    def _distribute_power(
+    def _distribute_power(  # pylint: disable=too-many-arguments
         self,
         components: List[InvBatPair],
         power_w: float,
         available_soc: Dict[int, float],
-        upper_bounds: Dict[int, float],
+        incl_bounds: Dict[int, float],
+        excl_bounds: Dict[int, float],
     ) -> DistributionResult:
         # pylint: disable=too-many-locals
         """Distribute power between given components.
@@ -336,17 +345,18 @@ def _distribute_power(
             available_soc: how much SoC remained to reach:
                 * SoC upper bound - if need to distribute consumption power
                 * SoC lower bound - if need to distribute supply power
-            upper_bounds: Min between upper bound of each pair in the components list:
-                * supply upper bound - if need to distribute consumption power
-                * consumption lower bound - if need to distribute supply power
+            incl_bounds: Inclusion bounds for each inverter
+            excl_bounds: Exclusion bounds for each inverter
 
         Returns:
             Distribution result.
         """
         (
             battery_availability_ratio,
             sum_ratio,
-        ) = self._compute_battery_availability_ratio(components, available_soc)
+        ) = self._compute_battery_availability_ratio(
+            components, available_soc, excl_bounds
+        )
 
         distribution: Dict[int, float] = {}
 
@@ -359,7 +369,7 @@ def _distribute_power(
         power_to_distribute: float = power_w
         used_ratio: float = 0.0
         ratio = sum_ratio
-        for pair, battery_ratio in battery_availability_ratio:
+        for pair, excl_bound, battery_ratio in battery_availability_ratio:
             inverter = pair[1]
             # ratio = 0, means all remaining batteries reach max SoC lvl or have no
             # capacity
@@ -375,10 +385,17 @@ def _distribute_power(
 
             # If the power allocated for that inverter is out of bound,
             # then we need to distribute more power over all remaining batteries.
-            upper_bound = upper_bounds[inverter.component_id]
-            if distribution[inverter.component_id] > upper_bound:
-                distribution[inverter.component_id] = upper_bound
-                distributed_power += upper_bound
+            incl_bound = incl_bounds[inverter.component_id]
+            if distribution[inverter.component_id] > incl_bound:
+                distribution[inverter.component_id] = incl_bound
+                distributed_power += incl_bound
+                # Distribute only the remaining power.
+                power_to_distribute = power_w - distributed_power
+                # Distribute between remaining batteries
+                ratio = sum_ratio - used_ratio
+            elif distribution[inverter.component_id] < excl_bound:
+                distribution[inverter.component_id] = excl_bound
+                distributed_power += excl_bound
                 # Distribute only the remaining power.
                 power_to_distribute = power_w - distributed_power
                 # Distribute between remaining batteries
@@ -487,19 +504,25 @@ def _distribute_consume_power(
                 0.0, battery.soc_upper_bound - battery.soc
             )
 
-        bounds: Dict[int, float] = {}
+        incl_bounds: Dict[int, float] = {}
+        excl_bounds: Dict[int, float] = {}
         for battery, inverter in components:
             # We can supply/consume with int only
-            inverter_bound = inverter.active_power_inclusion_upper_bound
-            battery_bound = battery.power_inclusion_upper_bound
-            bounds[inverter.component_id] = min(inverter_bound, battery_bound)
+            incl_bounds[inverter.component_id] = min(
+                inverter.active_power_inclusion_upper_bound,
+                battery.power_inclusion_upper_bound,
+            )
+            excl_bounds[inverter.component_id] = max(
+                inverter.active_power_exclusion_upper_bound,
+                battery.power_exclusion_upper_bound,
+            )
 
         result: DistributionResult = self._distribute_power(
-            components, power_w, available_soc, bounds
+            components, power_w, available_soc, incl_bounds, excl_bounds
         )
 
         return self._greedy_distribute_remaining_power(
-            result.distribution, bounds, result.remaining_power
+            result.distribution, incl_bounds, result.remaining_power
         )
 
     def _distribute_supply_power(
@@ -525,19 +548,24 @@ def _distribute_supply_power(
                 0.0, battery.soc - battery.soc_lower_bound
             )
 
-        bounds: Dict[int, float] = {}
+        incl_bounds: Dict[int, float] = {}
+        excl_bounds: Dict[int, float] = {}
         for battery, inverter in components:
-            # We can consume with int only
-            inverter_bound = inverter.active_power_inclusion_lower_bound
-            battery_bound = battery.power_inclusion_lower_bound
-            bounds[inverter.component_id] = -1 * max(inverter_bound, battery_bound)
+            incl_bounds[inverter.component_id] = -1 * max(
+                inverter.active_power_inclusion_lower_bound,
+                battery.power_inclusion_lower_bound,
+            )
+            excl_bounds[inverter.component_id] = -1 * min(
+                inverter.active_power_exclusion_lower_bound,
+                battery.power_exclusion_lower_bound,
+            )
 
         result: DistributionResult = self._distribute_power(
-            components, -1 * power_w, available_soc, bounds
+            components, -1 * power_w, available_soc, incl_bounds, excl_bounds
         )
 
         result = self._greedy_distribute_remaining_power(
-            result.distribution, bounds, result.remaining_power
+            result.distribution, incl_bounds, result.remaining_power
         )
 
         for inverter_id in result.distribution.keys():

tests/power/test_distribution_algorithm.py

@@ -140,11 +140,12 @@ def test_distribute_power_one_battery(self) -> None:
         components = self.create_components_with_capacity(1, capacity)
 
         available_soc: Dict[int, float] = {0: 40}
-        upper_bounds: Dict[int, float] = {1: 500}
+        incl_bounds: Dict[int, float] = {1: 500}
+        excl_bounds: Dict[int, float] = {1: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 650, available_soc, upper_bounds
+            components, 650, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 500})
@@ -160,11 +161,12 @@ def test_distribute_power_two_batteries_1(self) -> None:
         components = self.create_components_with_capacity(2, capacity)
 
         available_soc: Dict[int, float] = {0: 40, 2: 20}
-        upper_bounds: Dict[int, float] = {1: 500, 3: 500}
+        incl_bounds: Dict[int, float] = {1: 500, 3: 500}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 600, available_soc, upper_bounds
+            components, 600, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 400, 3: 200})
@@ -180,11 +182,12 @@ def test_distribute_power_two_batteries_2(self) -> None:
         components = self.create_components_with_capacity(2, capacity)
 
         available_soc: Dict[int, float] = {0: 20, 2: 20}
-        upper_bounds: Dict[int, float] = {1: 500, 3: 500}
+        incl_bounds: Dict[int, float] = {1: 500, 3: 500}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 600, available_soc, upper_bounds
+            components, 600, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 200, 3: 400})
@@ -201,11 +204,12 @@ def test_distribute_power_two_batteries_bounds(self) -> None:
         components = self.create_components_with_capacity(2, capacity)
 
         available_soc: Dict[int, float] = {0: 40, 2: 20}
-        upper_bounds: Dict[int, float] = {1: 250, 3: 330}
+        incl_bounds: Dict[int, float] = {1: 250, 3: 330}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 600, available_soc, upper_bounds
+            components, 600, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 250, 3: 330})
@@ -217,11 +221,12 @@ def test_distribute_power_three_batteries(self) -> None:
         components = self.create_components_with_capacity(3, capacity)
 
         available_soc: Dict[int, float] = {0: 40, 2: 20, 4: 20}
-        upper_bounds: Dict[int, float] = {1: 1000, 3: 3400, 5: 3550}
+        incl_bounds: Dict[int, float] = {1: 1000, 3: 3400, 5: 3550}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0, 5: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 1000, available_soc, upper_bounds
+            components, 1000, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 400, 3: 400, 5: 200})
@@ -233,11 +238,12 @@ def test_distribute_power_three_batteries_2(self) -> None:
         components = self.create_components_with_capacity(3, capacity)
 
         available_soc: Dict[int, float] = {0: 80, 2: 10, 4: 20}
-        upper_bounds: Dict[int, float] = {1: 400, 3: 3400, 5: 300}
+        incl_bounds: Dict[int, float] = {1: 400, 3: 3400, 5: 300}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0, 5: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 1000, available_soc, upper_bounds
+            components, 1000, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 400, 3: 300, 5: 300})
@@ -249,11 +255,12 @@ def test_distribute_power_three_batteries_3(self) -> None:
         components = self.create_components_with_capacity(3, capacity)
 
         available_soc: Dict[int, float] = {0: 80, 2: 10, 4: 20}
-        upper_bounds: Dict[int, float] = {1: 500, 3: 300, 5: 300}
+        incl_bounds: Dict[int, float] = {1: 500, 3: 300, 5: 300}
+        excl_bounds: Dict[int, float] = {1: 0, 3: 0, 5: 0}
 
         algorithm = DistributionAlgorithm(distributor_exponent=1)
         result = algorithm._distribute_power(  # pylint: disable=protected-access
-            components, 1000, available_soc, upper_bounds
+            components, 1000, available_soc, incl_bounds, excl_bounds
         )
 
         assert result.distribution == approx({1: 0, 3: 300, 5: 0})