Expose only usable SoC and Capacity through the BatteryPool (#459)

With this PR,
- soc values from the battery pool are always normalized to the 0-100%
range, and they are now streamed as `Sample[Quantity]` objects.
- only the usable capacity is streamed, eliminating the need for
capacity bounds, so they are now streamed as `Sample[Energy]` objects.

Closes #417

Author: shsms

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

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

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

@@ -12,8 +12,9 @@
 
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory, ComponentMetricId, InverterType
+from ...timeseries import Energy, Quantity, Sample
 from ._component_metrics import ComponentMetricsData
-from ._result_types import Bound, CapacityMetrics, PowerMetrics, SoCMetrics
+from ._result_types import Bound, PowerMetrics
 
 _logger = logging.getLogger(__name__)
 _MIN_TIMESTAMP = datetime.min.replace(tzinfo=timezone.utc)
@@ -58,7 +59,7 @@ def battery_inverter_mapping(batteries: Iterable[int]) -> dict[int, int]:
 
 # Formula output types class have no common interface
 # Print all possible types here.
-T = TypeVar("T", SoCMetrics, CapacityMetrics, PowerMetrics)
+T = TypeVar("T", Sample[Quantity], Sample[Energy], PowerMetrics)
 
 
 class MetricCalculator(ABC, Generic[T]):
@@ -141,7 +142,7 @@ def calculate(
         """
 
 
-class CapacityCalculator(MetricCalculator[CapacityMetrics]):
+class CapacityCalculator(MetricCalculator[Sample[Energy]]):
     """Define how to calculate Capacity metrics."""
 
     def __init__(self, batteries: Set[int]) -> None:
@@ -189,7 +190,7 @@ def calculate(
         self,
         metrics_data: dict[int, ComponentMetricsData],
         working_batteries: set[int],
-    ) -> CapacityMetrics | None:
+    ) -> Sample[Energy] | None:
         """Aggregate the metrics_data and calculate high level metric.
 
         Missing components will be ignored. Formula will be calculated for all
@@ -206,9 +207,8 @@ def calculate(
             High level metric calculated from the given metrics.
             Return None if there are no component metrics.
         """
-        result = CapacityMetrics(
-            timestamp=_MIN_TIMESTAMP, total_capacity=0, bound=Bound(lower=0, upper=0)
-        )
+        timestamp = _MIN_TIMESTAMP
+        total_capacity = 0.0
 
         for battery_id in working_batteries:
             if battery_id not in metrics_data:
@@ -223,16 +223,18 @@ def calculate(
             # All metrics are related so if any is missing then we skip the component.
             if capacity is None or soc_lower_bound is None or soc_upper_bound is None:
                 continue
+            usable_capacity = capacity * (soc_upper_bound - soc_lower_bound) / 100
+            timestamp = max(timestamp, metrics.timestamp)
+            total_capacity += usable_capacity
 
-            result.timestamp = max(result.timestamp, metrics.timestamp)
-            result.total_capacity += capacity
-            result.bound.upper += capacity * soc_upper_bound
-            result.bound.lower += capacity * soc_lower_bound
-
-        return None if result.timestamp == _MIN_TIMESTAMP else result
+        return (
+            None
+            if timestamp == _MIN_TIMESTAMP
+            else Sample[Energy](timestamp, Energy.from_watt_hours(total_capacity))
+        )
 
 
-class SoCCalculator(MetricCalculator[SoCMetrics]):
+class SoCCalculator(MetricCalculator[Sample[Quantity]]):
     """Define how to calculate SoC metrics."""
 
     def __init__(self, batteries: Set[int]) -> None:
@@ -281,7 +283,7 @@ def calculate(
         self,
         metrics_data: dict[int, ComponentMetricsData],
         working_batteries: set[int],
-    ) -> SoCMetrics | None:
+    ) -> Sample[Quantity] | None:
         """Aggregate the metrics_data and calculate high level metric.
 
         Missing components will be ignored. Formula will be calculated for all
@@ -299,9 +301,9 @@ def calculate(
             Return None if there are no component metrics.
         """
         timestamp = _MIN_TIMESTAMP
-        used_capacity: float = 0
-        total_capacity: float = 0
-        capacity_bound = Bound(0, 0)
+        usable_capacity_x100: float = 0
+        used_capacity_x100: float = 0
+        total_capacity_x100: float = 0
 
         for battery_id in working_batteries:
             if battery_id not in metrics_data:
@@ -323,29 +325,35 @@ def calculate(
             ):
                 continue
 
+            # The SoC bounds are in the 0-100 range, so to get the actual usable
+            # capacity, we need to divide by 100.
+            #
+            # We only want to calculate the SoC, and the usable capacity calculation is
+            # just an intermediate step, so don't have to divide by 100 here, because it
+            # gets cancelled out later.
+            #
+            # Therefore, the variables are named with a `_x100` suffix.
+            usable_capacity_x100 = capacity * (soc_upper_bound - soc_lower_bound)
+            soc_scaled = (
+                (soc - soc_lower_bound) / (soc_upper_bound - soc_lower_bound) * 100
+            )
+            soc_scaled = max(soc_scaled, 0)
             timestamp = max(timestamp, metrics.timestamp)
-            used_capacity += capacity * soc
-            total_capacity += capacity
-            capacity_bound.upper += capacity * soc_upper_bound
-            capacity_bound.lower += capacity * soc_lower_bound
+            used_capacity_x100 += usable_capacity_x100 * soc_scaled
+            total_capacity_x100 += usable_capacity_x100
 
         if timestamp == _MIN_TIMESTAMP:
             return None
 
         # To avoid zero division error
-        if total_capacity == 0:
-            return SoCMetrics(
+        if total_capacity_x100 == 0:
+            return Sample(
                 timestamp=timestamp,
-                average_soc=0,
-                bound=Bound(0, 0),
+                value=Quantity(0.0),
             )
-        return SoCMetrics(
+        return Sample(
             timestamp=timestamp,
-            average_soc=used_capacity / total_capacity,
-            bound=Bound(
-                lower=capacity_bound.lower / total_capacity,
-                upper=capacity_bound.upper / total_capacity,
-            ),
+            value=Quantity(used_capacity_x100 / total_capacity_x100),
         )
 
 

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

@@ -18,77 +18,6 @@ class Bound:
     """Upper bound."""
 
 
-@dataclass
-class CapacityMetrics:
-    """Capacity metrics."""
-
-    # compare = False tells the dataclass to not use name for comparison methods
-    timestamp: datetime = field(compare=False)
-    """Timestamp of the metrics,"""
-
-    total_capacity: float
-    """Total batteries capacity.
-
-    Calculated with the formula:
-    ```python
-    working_batteries: Set[BatteryData] # working batteries from the battery pool
-    total_capacity = sum(battery.capacity for battery in working_batteries)
-    ```
-    """
-    bound: Bound
-    """Capacity bounds.
-
-    Bounds are calculated with the formula:
-    ```python
-    working_batteries: Set[BatteryData] # working batteries from the battery
-    bound.lower = sum(
-        battery.capacity * battery.soc_lower_bound for battery in working_batteries)
-
-    bound.upper = sum(
-        battery.capacity * battery.soc_upper_bound for battery in working_batteries)
-    ```
-    """
-
-
-@dataclass
-class SoCMetrics:
-    """Soc metrics."""
-
-    # compare = False tells the dataclass to not use name for comparison methods
-    timestamp: datetime = field(compare=False)
-    """Timestamp of the metrics."""
-
-    average_soc: float
-    """Average soc.
-
-    Average soc is calculated with the formula:
-    ```python
-    working_batteries: Set[BatteryData] # working batteries from the battery pool
-
-    used_capacity = sum(battery.capacity * battery.soc for battery in working_batteries)
-    total_capacity = sum(battery.capacity for battery in working_batteries)
-    average_soc = used_capacity/total_capacity
-    ```
-    """
-
-    bound: Bound
-    """SoC bounds weighted by capacity.
-
-    Bounds are calculated with the formula:
-    capacity_lower_bound = sum(
-        battery.capacity * battery.soc_lower_bound for battery in working_batteries)
-
-    capacity_upper_bound = sum(
-        battery.capacity * battery.soc_upper_bound for battery in working_batteries)
-
-    total_capacity = sum(battery.capacity for battery in working_batteries)
-
-    bound.lower = capacity_lower_bound/total_capacity
-    bound.upper = capacity_upper_bound/total_capacity
-
-    """
-
-
 @dataclass
 class PowerMetrics:
     """Power bounds metrics."""

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

@@ -20,16 +20,17 @@
 from ...actor.power_distributing.result import Result
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory
+from ...timeseries import Quantity, Sample
 from .._formula_engine import FormulaEngine, FormulaEnginePool
 from .._formula_engine._formula_generators import (
     BatteryPowerFormula,
     FormulaGeneratorConfig,
     FormulaType,
 )
-from .._quantities import Power
+from .._quantities import Energy, Power
 from ._methods import MetricAggregator, SendOnUpdate
 from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
-from ._result_types import CapacityMetrics, PowerMetrics, SoCMetrics
+from ._result_types import PowerMetrics
 
 
 class BatteryPool:
@@ -329,11 +330,29 @@ def consumption_power(self) -> FormulaEngine[Power]:
         return engine
 
     @property
-    def soc(self) -> MetricAggregator[SoCMetrics]:
-        """Get receiver to receive new soc metrics when they change.
+    def soc(self) -> MetricAggregator[Sample[Quantity]]:
+        """Fetch the normalized average weighted-by-capacity SoC values for the pool.
 
-        Soc formulas are described in the receiver return type.  None will be send if
-        there is no component to calculate metric.
+        The values are normalized to the 0-100% range.
+
+        Average soc is calculated with the formula:
+        ```
+        working_batteries: Set[BatteryData] # working batteries from the battery pool
+
+        soc_scaled = max(
+            0,
+            (soc - soc_lower_bound) / (soc_upper_bound - soc_lower_bound) * 100,
+        )
+        used_capacity = sum(
+            battery.usable_capacity * battery.soc_scaled
+            for battery in working_batteries
+        )
+        total_capacity = sum(battery.usable_capacity for battery in working_batteries)
+        average_soc = used_capacity/total_capacity
+        ```
+
+        `None` values will be sent if there are no components to calculate the metric
+        with.
 
         A receiver from the MetricAggregator can be obtained by calling the
         `new_receiver` method.
@@ -355,11 +374,19 @@ def soc(self) -> MetricAggregator[SoCMetrics]:
         return self._active_methods[method_name]
 
     @property
-    def capacity(self) -> MetricAggregator[CapacityMetrics]:
+    def capacity(self) -> MetricAggregator[Sample[Energy]]:
         """Get receiver to receive new capacity metrics when they change.
 
-        Capacity formulas are described in the receiver return type.  None will be send
-        if there is no component to calculate metrics.
+        Calculated with the formula:
+        ```
+        working_batteries: Set[BatteryData] # working batteries from the battery pool
+        total_capacity = sum(
+            battery.capacity * (soc_upper_bound - soc_lower_bound) / 100
+            for battery in working_batteries
+        )
+        ```
+
+        None will be send if there is no component to calculate metrics.
 
         A receiver from the MetricAggregator can be obtained by calling the
         `new_receiver` method.