Use Sample[Quantity] to stream BatteryPool.soc values

A custom type is no longer necessary because there are no additional
fields like bounds that need to be streamed with SoC values.

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

Author: shsms

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 Quantity, Sample
 from ._component_metrics import ComponentMetricsData
-from ._result_types import Bound, CapacityMetrics, PowerMetrics, SoCMetrics
+from ._result_types import Bound, CapacityMetrics, 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], CapacityMetrics, PowerMetrics)
 
 
 class MetricCalculator(ABC, Generic[T]):
@@ -228,7 +229,7 @@ def calculate(
         return None if result.timestamp == _MIN_TIMESTAMP else result
 
 
-class SoCCalculator(MetricCalculator[SoCMetrics]):
+class SoCCalculator(MetricCalculator[Sample[Quantity]]):
     """Define how to calculate SoC metrics."""
 
     def __init__(self, batteries: Set[int]) -> None:
@@ -277,7 +278,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
@@ -341,13 +342,13 @@ def calculate(
 
         # To avoid zero division error
         if total_capacity_x100 == 0:
-            return SoCMetrics(
+            return Sample(
                 timestamp=timestamp,
-                average_soc=0,
+                value=Quantity(0.0),
             )
-        return SoCMetrics(
+        return Sample(
             timestamp=timestamp,
-            average_soc=used_capacity_x100 / total_capacity_x100,
+            value=Quantity(used_capacity_x100 / total_capacity_x100),
         )
 
 

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

@@ -20,6 +20,7 @@
 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,
@@ -29,7 +30,7 @@
 from .._quantities import Power
 from ._methods import MetricAggregator, SendOnUpdate
 from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
-from ._result_types import CapacityMetrics, PowerMetrics, SoCMetrics
+from ._result_types import CapacityMetrics, PowerMetrics
 
 
 class BatteryPool:
@@ -329,7 +330,7 @@ def consumption_power(self) -> FormulaEngine[Power]:
         return engine
 
     @property
-    def soc(self) -> MetricAggregator[SoCMetrics]:
+    def soc(self) -> MetricAggregator[Sample[Quantity]]:
         """Fetch the normalized average weighted-by-capacity SoC values for the pool.
 
         The formulas for calculating this metric are described

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -24,13 +24,13 @@
 from frequenz.sdk.actor import ResamplerConfig
 from frequenz.sdk.actor.power_distributing import BatteryStatus
 from frequenz.sdk.microgrid.component import ComponentCategory
+from frequenz.sdk.timeseries import Quantity, Sample
 from frequenz.sdk.timeseries._quantities import Power
 from frequenz.sdk.timeseries.battery_pool import (
     BatteryPool,
     Bound,
     CapacityMetrics,
     PowerMetrics,
-    SoCMetrics,
 )
 from frequenz.sdk.timeseries.battery_pool._metric_calculator import (
     battery_inverter_mapping,
@@ -670,31 +670,31 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         receiver.receive(), timeout=WAIT_FOR_COMPONENT_DATA_SEC + 0.2
     )
     now = datetime.now(tz=timezone.utc)
-    expected = SoCMetrics(
+    expected = Sample(
         timestamp=now,
-        average_soc=10.0,
+        value=Quantity(10.0),
     )
     compare_messages(msg, expected, WAIT_FOR_COMPONENT_DATA_SEC + 0.2)
 
     batteries_in_pool = list(battery_pool.battery_ids)
-    scenarios: list[Scenario[SoCMetrics]] = [
+    scenarios: list[Scenario[Sample[Quantity]]] = [
         Scenario(
             batteries_in_pool[0],
             {"capacity": 150, "soc": 10},
-            SoCMetrics(now, 2.5),
+            Sample(now, Quantity(2.5)),
         ),
         Scenario(
             batteries_in_pool[0],
             {
                 "soc_lower_bound": 0.0,
             },
-            SoCMetrics(now, 12.727272727272727),
+            Sample(now, Quantity(12.727272727272727)),
         ),
         # If NaN, then not include that battery in the metric.
         Scenario(
             batteries_in_pool[0],
             {"soc_upper_bound": float("NaN")},
-            SoCMetrics(now, 10),
+            Sample(now, Quantity(10.0)),
         ),
         # All batteries are sending NaN, can't calculate SoC so we should send None
         Scenario(
@@ -705,7 +705,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[1],
             {"soc": 30},
-            SoCMetrics(now, 10.0),
+            Sample(now, Quantity(10.0)),
         ),
         # Final metric didn't change, so nothing should be received.
         Scenario(
@@ -718,17 +718,17 @@ async def run_soc_test(setup_args: SetupArgs) -> None:
         Scenario(
             batteries_in_pool[1],
             {"capacity": 0},
-            SoCMetrics(now, 0),
+            Sample(now, Quantity(0.0)),
         ),
         Scenario(
             batteries_in_pool[0],
             {"capacity": 50, "soc": 55.0},
-            SoCMetrics(now, 50.0),
+            Sample(now, Quantity(50.0)),
         ),
         Scenario(
             batteries_in_pool[1],
             {"capacity": 150},
-            SoCMetrics(now, 25.0),
+            Sample(now, Quantity(25.0)),
         ),
     ]
 
@@ -741,15 +741,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, 25.0),
-        only_first_battery_result=SoCMetrics(now, 50.0),
+        all_pool_result=Sample(now, Quantity(25.0)),
+        only_first_battery_result=Sample(now, Quantity(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, 50.0), 0.2)
+    compare_messages(msg, Sample(now, Quantity(50.0)), 0.2)
 
     # All batteries stopped sending data.
     await streamer.stop_streaming(batteries_in_pool[0])
@@ -761,7 +761,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, 50.0), 0.2)
+    compare_messages(msg, Sample(now, Quantity(50.0)), 0.2)
 
 
 async def run_power_bounds_test(  # pylint: disable=too-many-locals