Stream exclusion bounds from the battery pool

RELEASE_NOTES.md

@@ -8,6 +8,8 @@
 
 - Upgrade to microgrid API v0.15.1.  If you're using any of the lower level microgrid interfaces, you will need to upgrade your code.
 
+- The `BatteryPool.power_bounds` method now streams inclusion/exclusion bounds.  The bounds are now represented by `Power` objects and not `float`s.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

src/frequenz/sdk/actor/_data_sourcing/microgrid_api_source.py

@@ -85,6 +85,12 @@ def get_channel_name(self) -> str:
     ComponentMetricId.POWER_INCLUSION_LOWER_BOUND: lambda msg: (
         msg.power_inclusion_lower_bound
     ),
+    ComponentMetricId.POWER_EXCLUSION_LOWER_BOUND: lambda msg: (
+        msg.power_exclusion_lower_bound
+    ),
+    ComponentMetricId.POWER_EXCLUSION_UPPER_BOUND: lambda msg: (
+        msg.power_exclusion_upper_bound
+    ),
     ComponentMetricId.POWER_INCLUSION_UPPER_BOUND: lambda msg: (
         msg.power_inclusion_upper_bound
     ),
@@ -96,6 +102,12 @@ def get_channel_name(self) -> str:
     ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND: lambda msg: (
         msg.active_power_inclusion_lower_bound
     ),
+    ComponentMetricId.ACTIVE_POWER_EXCLUSION_LOWER_BOUND: lambda msg: (
+        msg.active_power_exclusion_lower_bound
+    ),
+    ComponentMetricId.ACTIVE_POWER_EXCLUSION_UPPER_BOUND: lambda msg: (
+        msg.active_power_exclusion_upper_bound
+    ),
     ComponentMetricId.ACTIVE_POWER_INCLUSION_UPPER_BOUND: lambda msg: (
         msg.active_power_inclusion_upper_bound
     ),

src/frequenz/sdk/microgrid/component/_component.py

@@ -145,9 +145,13 @@ class ComponentMetricId(Enum):
     CAPACITY = "capacity"
 
     POWER_INCLUSION_LOWER_BOUND = "power_inclusion_lower_bound"
+    POWER_EXCLUSION_LOWER_BOUND = "power_exclusion_lower_bound"
+    POWER_EXCLUSION_UPPER_BOUND = "power_exclusion_upper_bound"
     POWER_INCLUSION_UPPER_BOUND = "power_inclusion_upper_bound"
 
     ACTIVE_POWER_INCLUSION_LOWER_BOUND = "active_power_inclusion_lower_bound"
+    ACTIVE_POWER_EXCLUSION_LOWER_BOUND = "active_power_exclusion_lower_bound"
+    ACTIVE_POWER_EXCLUSION_UPPER_BOUND = "active_power_exclusion_upper_bound"
     ACTIVE_POWER_INCLUSION_UPPER_BOUND = "active_power_inclusion_upper_bound"
 
     TEMPERATURE = "temperature"

src/frequenz/sdk/timeseries/battery_pool/__init__.py

@@ -3,11 +3,11 @@
 
 """Manage a pool of batteries."""
 
-from ._result_types import Bound, PowerMetrics
+from ._result_types import Bounds, PowerMetrics
 from .battery_pool import BatteryPool
 
 __all__ = [
     "BatteryPool",
     "PowerMetrics",
-    "Bound",
+    "Bounds",
 ]

src/frequenz/sdk/timeseries/battery_pool/_metric_calculator.py

@@ -12,9 +12,10 @@
 
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory, ComponentMetricId, InverterType
-from ...timeseries import Energy, Percentage, Sample, Temperature
+from ...timeseries import Sample
+from .._quantities import Energy, Percentage, Power, Temperature
 from ._component_metrics import ComponentMetricsData
-from ._result_types import Bound, PowerMetrics
+from ._result_types import Bounds, PowerMetrics
 
 _logger = logging.getLogger(__name__)
 _MIN_TIMESTAMP = datetime.min.replace(tzinfo=timezone.utc)
@@ -479,11 +480,15 @@ def __init__(
         super().__init__(used_batteries)
         self._battery_metrics = [
             ComponentMetricId.POWER_INCLUSION_LOWER_BOUND,
+            ComponentMetricId.POWER_EXCLUSION_LOWER_BOUND,
+            ComponentMetricId.POWER_EXCLUSION_UPPER_BOUND,
             ComponentMetricId.POWER_INCLUSION_UPPER_BOUND,
         ]
 
         self._inverter_metrics = [
             ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND,
+            ComponentMetricId.ACTIVE_POWER_EXCLUSION_LOWER_BOUND,
+            ComponentMetricId.ACTIVE_POWER_EXCLUSION_UPPER_BOUND,
             ComponentMetricId.ACTIVE_POWER_INCLUSION_UPPER_BOUND,
         ]
 
@@ -514,6 +519,84 @@ def inverter_metrics(self) -> Mapping[int, list[ComponentMetricId]]:
         """
         return {cid: self._inverter_metrics for cid in set(self._bat_inv_map.values())}
 
+    def _fetch_inclusion_bounds(
+        self,
+        battery_id: int,
+        inverter_id: int,
+        metrics_data: dict[int, ComponentMetricsData],
+    ) -> tuple[datetime, list[float], list[float]]:
+        timestamp = _MIN_TIMESTAMP
+        inclusion_lower_bounds: list[float] = []
+        inclusion_upper_bounds: list[float] = []
+
+        # Inclusion upper and lower bounds are not related.
+        # If one is missing, then we can still use the other.
+        if battery_id in metrics_data:
+            data = metrics_data[battery_id]
+            value = data.get(ComponentMetricId.POWER_INCLUSION_UPPER_BOUND)
+            if value is not None:
+                timestamp = max(timestamp, data.timestamp)
+                inclusion_upper_bounds.append(value)
+
+            value = data.get(ComponentMetricId.POWER_INCLUSION_LOWER_BOUND)
+            if value is not None:
+                timestamp = max(timestamp, data.timestamp)
+                inclusion_lower_bounds.append(value)
+
+        if inverter_id in metrics_data:
+            data = metrics_data[inverter_id]
+
+            value = data.get(ComponentMetricId.ACTIVE_POWER_INCLUSION_UPPER_BOUND)
+            if value is not None:
+                timestamp = max(data.timestamp, timestamp)
+                inclusion_upper_bounds.append(value)
+
+            value = data.get(ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND)
+            if value is not None:
+                timestamp = max(data.timestamp, timestamp)
+                inclusion_lower_bounds.append(value)
+
+        return (timestamp, inclusion_lower_bounds, inclusion_upper_bounds)
+
+    def _fetch_exclusion_bounds(
+        self,
+        battery_id: int,
+        inverter_id: int,
+        metrics_data: dict[int, ComponentMetricsData],
+    ) -> tuple[datetime, list[float], list[float]]:
+        timestamp = _MIN_TIMESTAMP
+        exclusion_lower_bounds: list[float] = []
+        exclusion_upper_bounds: list[float] = []
+
+        # Exclusion upper and lower bounds are not related.
+        # If one is missing, then we can still use the other.
+        if battery_id in metrics_data:
+            data = metrics_data[battery_id]
+            value = data.get(ComponentMetricId.POWER_EXCLUSION_UPPER_BOUND)
+            if value is not None:
+                timestamp = max(timestamp, data.timestamp)
+                exclusion_upper_bounds.append(value)
+
+            value = data.get(ComponentMetricId.POWER_EXCLUSION_LOWER_BOUND)
+            if value is not None:
+                timestamp = max(timestamp, data.timestamp)
+                exclusion_lower_bounds.append(value)
+
+        if inverter_id in metrics_data:
+            data = metrics_data[inverter_id]
+
+            value = data.get(ComponentMetricId.ACTIVE_POWER_EXCLUSION_UPPER_BOUND)
+            if value is not None:
+                timestamp = max(data.timestamp, timestamp)
+                exclusion_upper_bounds.append(value)
+
+            value = data.get(ComponentMetricId.ACTIVE_POWER_EXCLUSION_LOWER_BOUND)
+            if value is not None:
+                timestamp = max(data.timestamp, timestamp)
+                exclusion_lower_bounds.append(value)
+
+        return (timestamp, exclusion_lower_bounds, exclusion_upper_bounds)
+
     def calculate(
         self,
         metrics_data: dict[int, ComponentMetricsData],
@@ -533,53 +616,45 @@ def calculate(
             High level metric calculated from the given metrics.
             Return None if there are no component metrics.
         """
-        # In the future we will have lower bound, too.
-
-        result = PowerMetrics(
-            timestamp=_MIN_TIMESTAMP,
-            supply_bound=Bound(0, 0),
-            consume_bound=Bound(0, 0),
-        )
+        timestamp = _MIN_TIMESTAMP
+        inclusion_bounds_lower = 0.0
+        inclusion_bounds_upper = 0.0
+        exclusion_bounds_lower = 0.0
+        exclusion_bounds_upper = 0.0
 
         for battery_id in working_batteries:
-            supply_upper_bounds: list[float] = []
-            consume_upper_bounds: list[float] = []
-
-            if battery_id in metrics_data:
-                data = metrics_data[battery_id]
-
-                # Consume and supply bounds are not related.
-                # If one is missing, then we can still use the other.
-                value = data.get(ComponentMetricId.POWER_INCLUSION_UPPER_BOUND)
-                if value is not None:
-                    result.timestamp = max(result.timestamp, data.timestamp)
-                    consume_upper_bounds.append(value)
-
-                value = data.get(ComponentMetricId.POWER_INCLUSION_LOWER_BOUND)
-                if value is not None:
-                    result.timestamp = max(result.timestamp, data.timestamp)
-                    supply_upper_bounds.append(value)
-
             inverter_id = self._bat_inv_map[battery_id]
-            if inverter_id in metrics_data:
-                data = metrics_data[inverter_id]
-
-                value = data.get(ComponentMetricId.ACTIVE_POWER_INCLUSION_UPPER_BOUND)
-                if value is not None:
-                    result.timestamp = max(data.timestamp, result.timestamp)
-                    consume_upper_bounds.append(value)
-
-                value = data.get(ComponentMetricId.ACTIVE_POWER_INCLUSION_LOWER_BOUND)
-                if value is not None:
-                    result.timestamp = max(data.timestamp, result.timestamp)
-                    supply_upper_bounds.append(value)
-
-            if len(consume_upper_bounds) > 0:
-                result.consume_bound.upper += min(consume_upper_bounds)
-            if len(supply_upper_bounds) > 0:
-                result.supply_bound.lower += max(supply_upper_bounds)
+            (
+                _ts,
+                inclusion_lower_bounds,
+                inclusion_upper_bounds,
+            ) = self._fetch_inclusion_bounds(battery_id, inverter_id, metrics_data)
+            timestamp = max(timestamp, _ts)
+            (
+                _ts,
+                exclusion_lower_bounds,
+                exclusion_upper_bounds,
+            ) = self._fetch_exclusion_bounds(battery_id, inverter_id, metrics_data)
+            if len(inclusion_upper_bounds) > 0:
+                inclusion_bounds_upper += min(inclusion_upper_bounds)
+            if len(inclusion_lower_bounds) > 0:
+                inclusion_bounds_lower += max(inclusion_lower_bounds)
+            if len(exclusion_upper_bounds) > 0:
+                exclusion_bounds_upper += max(exclusion_upper_bounds)
+            if len(exclusion_lower_bounds) > 0:
+                exclusion_bounds_lower += min(exclusion_lower_bounds)
 
-        if result.timestamp == _MIN_TIMESTAMP:
+        if timestamp == _MIN_TIMESTAMP:
             return None
 
-        return result
+        return PowerMetrics(
+            timestamp=timestamp,
+            inclusion_bounds=Bounds(
+                Power.from_watts(inclusion_bounds_lower),
+                Power.from_watts(inclusion_bounds_upper),
+            ),
+            exclusion_bounds=Bounds(
+                Power.from_watts(exclusion_bounds_lower),
+                Power.from_watts(exclusion_bounds_upper),
+            ),
+        )

src/frequenz/sdk/timeseries/battery_pool/_result_types.py

@@ -6,15 +6,17 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 
+from .._quantities import Power
+
 
 @dataclass
-class Bound:
+class Bounds:
     """Lower and upper bound values."""
 
-    lower: float
+    lower: Power
     """Lower bound."""
 
-    upper: float
+    upper: Power
     """Upper bound."""
 
 
@@ -26,38 +28,24 @@ class PowerMetrics:
     timestamp: datetime = field(compare=False)
     """Timestamp of the metrics."""
 
-    supply_bound: Bound
-    """Supply power bounds.
-
-    Upper bound is always 0 and will be supported later.
-    Lower bound is negative number calculated with with the formula:
-    ```python
-    working_pairs: Set[BatteryData, InverterData] # working batteries from the battery
-        pool and adjacent inverters
-
-    supply_bound.lower = sum(
-        max(
-            battery.power_inclusion_lower_bound, inverter.active_power_inclusion_lower_bound)
-            for each working battery in battery pool
-            )
-        )
-    ```
+    # pylint: disable=line-too-long
+    inclusion_bounds: Bounds
+    """Inclusion power bounds for all batteries in the battery pool instance.
+
+    This is the range within which power requests are allowed by the battery pool.
+
+    When exclusion bounds are present, they will exclude a subset of the inclusion
+    bounds.
+
+    More details [here](https://github.com/frequenz-floss/frequenz-api-common/blob/v0.3.0/proto/frequenz/api/common/metrics.proto#L37-L91).
     """
 
-    consume_bound: Bound
-    """Consume power bounds.
-
-    Lower bound is always 0 and will be supported later.
-    Upper bound is positive number calculated with with the formula:
-    ```python
-    working_pairs: Set[BatteryData, InverterData] # working batteries from the battery
-        pool and adjacent inverters
-
-    consume_bound.upper = sum(
-        min(
-            battery.power_inclusion_upper_bound, inverter.active_power_inclusion_upper_bound)
-            for each working battery in battery pool
-            )
-        )
-    ```
+    exclusion_bounds: Bounds
+    """Exclusion power bounds for all batteries in the battery pool instance.
+
+    This is the range within which power requests are NOT allowed by the battery pool.
+    If present, they will be a subset of the inclusion bounds.
+
+    More details [here](https://github.com/frequenz-floss/frequenz-api-common/blob/v0.3.0/proto/frequenz/api/common/metrics.proto#L37-L91).
     """
+    # pylint: enable=line-too-long