Merge draft implementation from christianparpart

From #485

Also resolve conflicts.

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

Author: shsms

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_MAX: lambda msg: msg.temperature_max,
 }
 
 _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_MAX = "temperature_max"

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",
 )
 
 
@@ -401,6 +402,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/__init__.py

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

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

@@ -12,9 +12,9 @@
 
 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
+from ._result_types import Bound, PowerMetrics, TemperatureMetrics
 
 _logger = logging.getLogger(__name__)
 _MIN_TIMESTAMP = datetime.min.replace(tzinfo=timezone.utc)
@@ -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, TemperatureMetrics)
 
 
 class MetricCalculator(ABC, Generic[T]):
@@ -234,6 +234,99 @@ def calculate(
         )
 
 
+class TemperatureCalculator(MetricCalculator[TemperatureMetrics]):
+    """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_MAX,
+        ]
+
+    @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],
+    ) -> TemperatureMetrics | 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
+        temperature_min: float = float("inf")
+        temperature_max: float = float("-inf")
+        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_MAX)
+            if temperature is None:
+                continue
+            timestamp = max(timestamp, metrics.timestamp)
+            temperature_sum += temperature
+            temperature_min = min(temperature_min, temperature)
+            temperature_max = max(temperature_max, temperature)
+            temperature_count += 1
+        if timestamp == _MIN_TIMESTAMP:
+            return None
+
+        temperature_avg = temperature_sum / temperature_count
+
+        return TemperatureMetrics(
+            timestamp=timestamp,
+            min=Temperature.from_celsius(value=temperature_min),
+            avg=Temperature.from_celsius(value=temperature_avg),
+            max=Temperature.from_celsius(value=temperature_max),
+        )
+
+
 class SoCCalculator(MetricCalculator[Sample[Percentage]]):
     """Define how to calculate SoC metrics."""
 

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

@@ -6,6 +6,8 @@
 from dataclasses import dataclass, field
 from datetime import datetime
 
+from ...timeseries import Temperature
+
 
 @dataclass
 class Bound:
@@ -61,3 +63,20 @@ class PowerMetrics:
         )
     ```
     """
+
+
+@dataclass
+class TemperatureMetrics:
+    """Container for temperature metrics."""
+
+    timestamp: datetime
+    """Timestamp of the metrics."""
+
+    max: Temperature
+    """Maximum temperature of the collection of temperatures."""
+
+    min: Temperature
+    """Minimum temperature of the collection of temperatures."""
+
+    avg: Temperature
+    """Average temperature of the collection of temperatures."""

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

@@ -29,8 +29,13 @@
 )
 from .._quantities import Energy, Percentage, Power
 from ._methods import MetricAggregator, SendOnUpdate
-from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator
-from ._result_types import PowerMetrics
+from ._metric_calculator import (
+    CapacityCalculator,
+    PowerBoundsCalculator,
+    SoCCalculator,
+    TemperatureCalculator,
+)
+from ._result_types import PowerMetrics, TemperatureMetrics
 
 
 class BatteryPool:
@@ -382,6 +387,24 @@ def soc(self) -> MetricAggregator[Sample[Percentage]]:
 
         return self._active_methods[method_name]
 
+    @property
+    def temperature(self) -> MetricAggregator[TemperatureMetrics]:
+        """Fetch the maximum temperature of the batteries in the pool.
+
+        Returns:
+            A MetricAggregator that will calculate and stream the maximum 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.