Normalize average SoC values from BatteryPool to 0-100% range

This allows us to hide actual battery bounds, and allows end users to
not have to worry about SoC bounds, and just deal with normalized
values instead.

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

Author: shsms

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

@@ -299,9 +299,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 +323,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:
+        if total_capacity_x100 == 0:
             return SoCMetrics(
                 timestamp=timestamp,
                 average_soc=0,
-                bound=Bound(0, 0),
             )
         return SoCMetrics(
             timestamp=timestamp,
-            average_soc=used_capacity / total_capacity,
-            bound=Bound(
-                lower=capacity_bound.lower / total_capacity,
-                upper=capacity_bound.upper / total_capacity,
-            ),
+            average_soc=used_capacity_x100 / total_capacity_x100,
         )
 
 

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

@@ -59,35 +59,27 @@ class SoCMetrics:
     """Timestamp of the metrics."""
 
     average_soc: float
-    """Average soc.
+    """Average SoC of working batteries in the pool, weighted by usable capacity.
+
+    The values are normalized to the 0-100% range.
 
     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)
+    battery.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
     ```
     """
 
-    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:

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

@@ -330,10 +330,11 @@ def consumption_power(self) -> FormulaEngine[Power]:
 
     @property
     def soc(self) -> MetricAggregator[SoCMetrics]:
-        """Get receiver to receive new soc metrics when they change.
+        """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 formulas for calculating this metric are described
+        [here][frequenz.sdk.timeseries.battery_pool.SoCMetrics].  `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.

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -628,8 +628,8 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
                 timestamp=datetime.now(tz=timezone.utc),
                 capacity=50,
                 soc=30,
-                soc_lower_bound=20,
-                soc_upper_bound=80,
+                soc_lower_bound=25,
+                soc_upper_bound=75,
             ),
             sampling_rate=0.05,
         )
@@ -643,8 +643,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
     now = datetime.now(tz=timezone.utc)
     expected = SoCMetrics(
         timestamp=now,
-        average_soc=30,
-        bound=Bound(lower=20, upper=80),
+        average_soc=10.0,
     )
     compare_messages(msg, expected, WAIT_FOR_COMPONENT_DATA_SEC + 0.2)
 
@@ -653,18 +652,20 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[0],
             {"capacity": 150, "soc": 10},
-            SoCMetrics(now, 15, Bound(20, 80)),
+            SoCMetrics(now, 2.5),
         ),
         Scenario(
             batteries_in_pool[0],
-            {"soc_lower_bound": 0},
-            SoCMetrics(now, 15, Bound(5, 80)),
+            {
+                "soc_lower_bound": 0.0,
+            },
+            SoCMetrics(now, 12.727272727272727),
         ),
         # If NaN, then not include that battery in the metric.
         Scenario(
             batteries_in_pool[0],
             {"soc_upper_bound": float("NaN")},
-            SoCMetrics(now, 30, Bound(20, 80)),
+            SoCMetrics(now, 10),
         ),
         # All batteries are sending NaN, can't calculate SoC so we should send None
         Scenario(
@@ -675,34 +676,34 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[1],
             {"soc": 30},
-            SoCMetrics(now, 30, Bound(20, 80)),
+            SoCMetrics(now, 10.0),
         ),
         # Final metric didn't change, so nothing should be received.
         Scenario(
             batteries_in_pool[0],
-            {"capacity": 0, "soc_lower_bound": 10, "soc_upper_bound": 100},
+            {"capacity": 0, "soc_lower_bound": 10.0, "soc_upper_bound": 100.0},
             None,
             wait_for_result=False,
         ),
         # Test zero division error
         Scenario(
             batteries_in_pool[1],
             {"capacity": 0},
-            SoCMetrics(now, 0, Bound(0, 0)),
+            SoCMetrics(now, 0),
         ),
         Scenario(
             batteries_in_pool[0],
-            {"capacity": 50},
-            SoCMetrics(now, 10, Bound(10, 100)),
+            {"capacity": 50, "soc": 55.0},
+            SoCMetrics(now, 50.0),
         ),
         Scenario(
             batteries_in_pool[1],
-            {"capacity": 50},
-            SoCMetrics(now, 20, Bound(15, 90)),
+            {"capacity": 150},
+            SoCMetrics(now, 25.0),
         ),
     ]
 
-    waiting_time_sec = setup_args.min_update_interval + 0.02
+    waiting_time_sec = setup_args.min_update_interval + 0.2
     await run_scenarios(scenarios, streamer, receiver, waiting_time_sec)
 
     await run_test_battery_status_channel(
@@ -711,15 +712,15 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         all_batteries=all_batteries,
         batteries_in_pool=batteries_in_pool,
         waiting_time_sec=waiting_time_sec,
-        all_pool_result=SoCMetrics(now, 20, Bound(15, 90)),
-        only_first_battery_result=SoCMetrics(now, 10, Bound(10, 100)),
+        all_pool_result=SoCMetrics(now, 25.0),
+        only_first_battery_result=SoCMetrics(now, 50.0),
     )
 
     # One battery stopped sending data.
     await streamer.stop_streaming(batteries_in_pool[1])
     await asyncio.sleep(MAX_BATTERY_DATA_AGE_SEC + 0.2)
     msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
-    compare_messages(msg, SoCMetrics(now, 10, Bound(10, 100)), 0.2)
+    compare_messages(msg, SoCMetrics(now, 50.0), 0.2)
 
     # All batteries stopped sending data.
     await streamer.stop_streaming(batteries_in_pool[0])
@@ -731,7 +732,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
     latest_data = streamer.get_current_component_data(batteries_in_pool[0])
     streamer.start_streaming(latest_data, sampling_rate=0.1)
     msg = await asyncio.wait_for(receiver.receive(), timeout=waiting_time_sec)
-    compare_messages(msg, SoCMetrics(now, 10, Bound(10, 100)), 0.2)
+    compare_messages(msg, SoCMetrics(now, 50.0), 0.2)
 
 
 async def run_power_bounds_test(  # pylint: disable=too-many-locals