Temperature streaming from the BatteryPool (#552)

Closes #466

Author: shsms

RELEASE_NOTES.md

@@ -22,6 +22,7 @@
   * Add a new quantity class `Frequency` for frequency values.
 
 - `FormulaEngine` arithmetics now supports scalar multiplication with floats and addition with Quantities
+- Add a new method for streaming average temperature values for the battery pool.
 
 ## Bug Fixes
 

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

@@ -84,6 +84,7 @@ def get_channel_name(self) -> str:
     ComponentMetricId.CAPACITY: lambda msg: msg.capacity,
     ComponentMetricId.POWER_LOWER_BOUND: lambda msg: msg.power_lower_bound,
     ComponentMetricId.POWER_UPPER_BOUND: lambda msg: msg.power_upper_bound,
+    ComponentMetricId.TEMPERATURE: lambda msg: msg.temperature,
 }
 
 _InverterDataMethods: Dict[ComponentMetricId, Callable[[InverterData], float]] = {

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

@@ -139,3 +139,5 @@ class ComponentMetricId(Enum):
 
     ACTIVE_POWER_LOWER_BOUND = "active_power_lower_bound"
     ACTIVE_POWER_UPPER_BOUND = "active_power_upper_bound"
+
+    TEMPERATURE = "temperature"

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

@@ -141,8 +141,8 @@ class BatteryData(ComponentData):
         This will be a positive number, or zero if no charging is possible.
     """
 
-    temperature_max: float
-    """The maximum temperature of all the blocks in a battery, in Celcius (°C)."""
+    temperature: float
+    """The (average) temperature reported by the battery, in Celcius (°C)."""
 
     _relay_state: battery_pb.RelayState.ValueType
     """State of the battery relay."""
@@ -172,7 +172,7 @@ def from_proto(cls, raw: microgrid_pb.ComponentData) -> BatteryData:
             capacity=raw.battery.properties.capacity,
             power_lower_bound=raw.battery.data.dc.power.system_bounds.lower,
             power_upper_bound=raw.battery.data.dc.power.system_bounds.upper,
-            temperature_max=raw.battery.data.temperature.max,
+            temperature=raw.battery.data.temperature.avg,
             _relay_state=raw.battery.state.relay_state,
             _component_state=raw.battery.state.component_state,
             _errors=list(raw.battery.errors),

src/frequenz/sdk/timeseries/__init__.py

@@ -45,6 +45,7 @@
     Percentage,
     Power,
     Quantity,
+    Temperature,
     Voltage,
 )
 from ._resampling import ResamplerConfig
@@ -63,6 +64,7 @@
     "Current",
     "Energy",
     "Power",
+    "Temperature",
     "Voltage",
     "Frequency",
     "Percentage",

src/frequenz/sdk/timeseries/_quantities.py

@@ -20,6 +20,7 @@
     "Energy",
     "Frequency",
     "Percentage",
+    "Temperature",
 )
 
 
@@ -386,6 +387,38 @@ def __call__(cls, *_args: Any, **_kwargs: Any) -> NoReturn:
         )
 
 
+class Temperature(
+    Quantity,
+    metaclass=_NoDefaultConstructible,
+    exponent_unit_map={
+        0: "°C",
+    },
+):
+    """A temperature quantity (in degrees Celsius)."""
+
+    @classmethod
+    def from_celsius(cls, value: float) -> Self:
+        """Initialize a new temperature quantity.
+
+        Args:
+            value: The temperature in degrees Celsius.
+
+        Returns:
+            A new temperature quantity.
+        """
+        power = cls.__new__(cls)
+        power._base_value = value
+        return power
+
+    def as_celsius(self) -> float:
+        """Return the temperature in degrees Celsius.
+
+        Returns:
+            The temperature in degrees Celsius.
+        """
+        return self._base_value
+
+
 class Power(
     Quantity,
     metaclass=_NoDefaultConstructible,

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

@@ -12,7 +12,7 @@
 
 from ...microgrid import connection_manager
 from ...microgrid.component import ComponentCategory, ComponentMetricId, InverterType
-from ...timeseries import Energy, Percentage, Sample
+from ...timeseries import Energy, Percentage, Sample, Temperature
 from ._component_metrics import ComponentMetricsData
 from ._result_types import Bound, PowerMetrics
 
@@ -59,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", Sample[Percentage], Sample[Energy], PowerMetrics)
+T = TypeVar("T", Sample[Percentage], Sample[Energy], PowerMetrics, Sample[Temperature])
 
 
 class MetricCalculator(ABC, Generic[T]):
@@ -234,6 +234,93 @@ def calculate(
         )
 
 
+class TemperatureCalculator(MetricCalculator[Sample[Temperature]]):
+    """Define how to calculate temperature metrics."""
+
+    def __init__(self, batteries: Set[int]) -> None:
+        """Create class instance.
+
+        Args:
+            batteries: What batteries should be used for calculation.
+        """
+        super().__init__(batteries)
+
+        self._metrics = [
+            ComponentMetricId.TEMPERATURE,
+        ]
+
+    @classmethod
+    def name(cls) -> str:
+        """Return name of the calculator.
+
+        Returns:
+            Name of the calculator
+        """
+        return "temperature"
+
+    @property
+    def battery_metrics(self) -> Mapping[int, list[ComponentMetricId]]:
+        """Return what metrics are needed for each battery.
+
+        Returns:
+            Map between battery id and set of required metrics id.
+        """
+        return {bid: self._metrics for bid in self._batteries}
+
+    @property
+    def inverter_metrics(self) -> Mapping[int, list[ComponentMetricId]]:
+        """Return what metrics are needed for each inverter.
+
+        Returns:
+            Map between inverter id and set of required metrics id.
+        """
+        return {}
+
+    def calculate(
+        self,
+        metrics_data: dict[int, ComponentMetricsData],
+        working_batteries: set[int],
+    ) -> Sample[Temperature] | None:
+        """Aggregate the metrics_data and calculate high level metric for temperature.
+
+        Missing components will be ignored. Formula will be calculated for all
+        working batteries that are in metrics_data.
+
+        Args:
+            metrics_data: Components metrics data, that should be used to calculate the
+                result.
+            working_batteries: working batteries. These batteries will be used
+                to calculate the result. It should be subset of the batteries given in a
+                constructor.
+
+        Returns:
+            High level metric calculated from the given metrics.
+            Return None if there are no component metrics.
+        """
+        timestamp = _MIN_TIMESTAMP
+        temperature_sum: float = 0.0
+        temperature_count: int = 0
+        for battery_id in working_batteries:
+            if battery_id not in metrics_data:
+                continue
+            metrics = metrics_data[battery_id]
+            temperature = metrics.get(ComponentMetricId.TEMPERATURE)
+            if temperature is None:
+                continue
+            timestamp = max(timestamp, metrics.timestamp)
+            temperature_sum += temperature
+            temperature_count += 1
+        if timestamp == _MIN_TIMESTAMP:
+            return None
+
+        temperature_avg = temperature_sum / temperature_count
+
+        return Sample[Temperature](
+            timestamp=timestamp,
+            value=Temperature.from_celsius(value=temperature_avg),
+        )
+
+
 class SoCCalculator(MetricCalculator[Sample[Percentage]]):
     """Define how to calculate SoC metrics."""
 

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

@@ -27,9 +27,14 @@
     FormulaGeneratorConfig,
     FormulaType,
 )
-from .._quantities import Energy, Percentage, Power
+from .._quantities import Energy, Percentage, Power, Temperature
 from ._methods import MetricAggregator, SendOnUpdate
-from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
+from ._metric_calculator import (
+    CapacityCalculator,
+    PowerBoundsCalculator,
+    SoCCalculator,
+    TemperatureCalculator,
+)
 from ._result_types import PowerMetrics
 
 
@@ -382,6 +387,24 @@ def soc(self) -> MetricAggregator[Sample[Percentage]]:
 
         return self._active_methods[method_name]
 
+    @property
+    def temperature(self) -> MetricAggregator[Sample[Temperature]]:
+        """Fetch the average temperature of the batteries in the pool.
+
+        Returns:
+            A MetricAggregator that will calculate and stream the average temperature
+                of all batteries in the pool.
+        """
+        method_name = SendOnUpdate.name() + "_" + TemperatureCalculator.name()
+        if method_name not in self._active_methods:
+            calculator = TemperatureCalculator(self._batteries)
+            self._active_methods[method_name] = SendOnUpdate(
+                metric_calculator=calculator,
+                working_batteries=self._working_batteries,
+                min_update_interval=self._min_update_interval,
+            )
+        return self._active_methods[method_name]
+
     @property
     def capacity(self) -> MetricAggregator[Sample[Energy]]:
         """Get receiver to receive new capacity metrics when they change.