Add component_data method to EVChargerPool (#266)

This PR:
1. introduces the `component_data` method to the EVChargerPool, which
ties the resampled 3-phase current values for an ev charger with the
component state, which comes directly from the microgrid, streaming
objects of type:

   ``` python
   @DataClass(frozen=True)
   class EVChargerData:
"""Data for an EV Charger, including the 3-phase current and the
component state."""

       component_id: int
       current: Sample3Phase
       state: EVChargerState
   ```

This is done so that the state-tracking code can piggy-back on the
resampler to detect if a component is missing data, and switch its state
to `MISSING` whenever that is the case.


3. adds these methods for `Sample3Phase`: `__iter__, min, max, map`.

Author: sahas-subramanian-frequenz

src/frequenz/sdk/timeseries/_base_types.py

@@ -3,9 +3,12 @@
 
 """Timeseries basic types."""
 
+from __future__ import annotations
+
+import functools
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Optional
+from typing import Callable, Iterator, Optional, overload
 
 
 # Ordering by timestamp is a bit arbitrary, and it is not always what might be
@@ -48,3 +51,85 @@ class Sample3Phase:
 
     value_p3: Optional[float]
     """The value of the 3rd phase in this sample."""
+
+    def __iter__(self) -> Iterator[float | None]:
+        """Return an iterator that yields values from each of the phases.
+
+        Yields:
+            Per-phase measurements one-by-one.
+        """
+        yield self.value_p1
+        yield self.value_p2
+        yield self.value_p3
+
+    @overload
+    def max(self, default: float) -> float:
+        ...
+
+    @overload
+    def max(self, default: None = None) -> float | None:
+        ...
+
+    def max(self, default: float | None = None) -> float | None:
+        """Return the max value among all phases, or default if they are all `None`.
+
+        Args:
+            default: value to return if all phases are `None`.
+
+        Returns:
+            Max value among all phases, if available, default value otherwise.
+        """
+        if not any(self):
+            return default
+        value: float = functools.reduce(
+            lambda x, y: x if x > y else y,
+            filter(None, self),
+        )
+        return value
+
+    @overload
+    def min(self, default: float) -> float:
+        ...
+
+    @overload
+    def min(self, default: None = None) -> float | None:
+        ...
+
+    def min(self, default: float | None = None) -> float | None:
+        """Return the min value among all phases, or default if they are all `None`.
+
+        Args:
+            default: value to return if all phases are `None`.
+
+        Returns:
+            Min value among all phases, if available, default value otherwise.
+        """
+        if not any(self):
+            return default
+        value: float = functools.reduce(
+            lambda x, y: x if x < y else y,
+            filter(None, self),
+        )
+        return value
+
+    def map(
+        self, function: Callable[[float], float], default: float | None = None
+    ) -> Sample3Phase:
+        """Apply the given function on each of the phase values and return the result.
+
+        If a phase value is `None`, replace it with `default` instead.
+
+        Args:
+            function: The function to apply on each of the phase values.
+            default: The value to apply if a phase value is `None`.
+
+        Returns:
+            A new `Sample3Phase` instance, with the given function applied on values
+                for each of the phases.
+        """
+        return Sample3Phase(
+            timestamp=self.timestamp,
+            value_p1=default if self.value_p1 is None else function(self.value_p1),
+            value_p2=default if self.value_p2 is None else function(self.value_p2),
+            value_p3=default if self.value_p3 is None else function(self.value_p3),
+        )

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

@@ -3,8 +3,12 @@
 
 """Interactions with EV Chargers."""
 
-from ._ev_charger_pool import EVChargerPool
+from ._ev_charger_pool import EVChargerData, EVChargerPool, EVChargerPoolError
+from ._state_tracker import EVChargerState
 
 __all__ = [
     "EVChargerPool",
+    "EVChargerData",
+    "EVChargerPoolError",
+    "EVChargerState",
 ]

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py

@@ -5,20 +5,26 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import uuid
+from asyncio import Task
+from collections import abc
+from dataclasses import dataclass
 
-from frequenz.channels import Sender
+from frequenz.channels import Broadcast, ChannelClosedError, Receiver, Sender
 
 from ...actor import ChannelRegistry, ComponentMetricRequest
 from ...microgrid import connection_manager
-from ...microgrid.component import ComponentCategory
+from ...microgrid.component import ComponentCategory, ComponentMetricId
+from .. import Sample, Sample3Phase
 from .._formula_engine import FormulaEnginePool, FormulaReceiver, FormulaReceiver3Phase
 from .._formula_engine._formula_generators import (
     EVChargerCurrentFormula,
     EVChargerPowerFormula,
     FormulaGeneratorConfig,
 )
+from ._state_tracker import EVChargerState, StateTracker
 
 logger = logging.getLogger(__name__)
 
@@ -27,6 +33,15 @@ class EVChargerPoolError(Exception):
     """An error that occurred in any of the EVChargerPool methods."""
 
 
+@dataclass(frozen=True)
+class EVChargerData:
+    """Data for an EV Charger, including the 3-phase current and the component state."""
+
+    component_id: int
+    current: Sample3Phase
+    state: EVChargerState
+
+
 class EVChargerPool:
     """Interactions with EV Chargers."""
 
@@ -62,14 +77,27 @@ def __init__(
                     component_category={ComponentCategory.EV_CHARGER}
                 )
             }
+        self._state_tracker: StateTracker | None = None
+        self._status_streams: dict[
+            int, tuple[Task[None], Broadcast[EVChargerData]]
+        ] = {}
         self._namespace: str = f"ev-charger-pool-{uuid.uuid4()}"
         self._formula_pool: FormulaEnginePool = FormulaEnginePool(
             self._namespace,
             self._channel_registry,
             self._resampler_subscription_sender,
         )
 
-    async def total_current(self) -> FormulaReceiver3Phase:
+    @property
+    def component_ids(self) -> abc.Set[int]:
+        """Return component IDs of all EV Chargers managed by this EVChargerPool.
+
+        Returns:
+            Set of managed component IDs.
+        """
+        return self._component_ids
+
+    async def current(self) -> FormulaReceiver3Phase:
         """Fetch the total current for the EV Chargers in the pool.
 
         If a formula engine to calculate EV Charger current is not already running, it
@@ -85,7 +113,7 @@ async def total_current(self) -> FormulaReceiver3Phase:
             FormulaGeneratorConfig(component_ids=self._component_ids),
         )
 
-    async def total_power(self) -> FormulaReceiver:
+    async def power(self) -> FormulaReceiver:
         """Fetch the total power for the EV Chargers in the pool.
 
         If a formula engine to calculate EV Charger power is not already running, it
@@ -102,49 +130,114 @@ async def total_power(self) -> FormulaReceiver:
             FormulaGeneratorConfig(component_ids=self._component_ids),
         )
 
-    async def current(self, component_id: int) -> FormulaReceiver3Phase:
-        """Fetch the 3-phase current for the given EV Charger id.
+    async def component_data(self, component_id: int) -> Receiver[EVChargerData]:
+        """Stream 3-phase current values and state of an EV Charger.
 
         Args:
-            component_id: id of the EV Charger to stream current values for.
+            component_id: id of the EV Charger for which data is requested.
 
         Returns:
-            A *new* receiver that will stream 3-phase current values for the given
-                EV Charger.
-
-        Raises:
-            EVChargerPoolError: if the given component_id is not part of the pool.
+            A receiver that streams objects containing 3-phase current and state of
+                an EV Charger.
         """
-        if component_id not in self._component_ids:
-            raise EVChargerPoolError(
-                f"{component_id=} is not part of the EVChargerPool"
-                f" (with ids={self._component_ids})"
+        if recv := self._status_streams.get(component_id, None):
+            task, output_chan = recv
+            if not task.done():
+                return output_chan.new_receiver()
+            logger.warning("Restarting component_status for id: %s", component_id)
+        else:
+            output_chan = Broadcast[EVChargerData](
+                f"evpool-component_status-{component_id}"
             )
-        return await self._formula_pool.from_generator(
-            f"ev_charger_current_{component_id}",
-            EVChargerCurrentFormula,
-            FormulaGeneratorConfig(component_ids={component_id}),
+
+        task = asyncio.create_task(
+            self._stream_component_data(component_id, output_chan.new_sender())
         )
 
-    async def power(self, component_id: int) -> FormulaReceiver:
-        """Fetch the power for the given EV Charger id.
+        self._status_streams[component_id] = (task, output_chan)
+
+        return output_chan.new_receiver()
+
+    async def _get_current_streams(
+        self, component_id: int
+    ) -> tuple[Receiver[Sample], Receiver[Sample], Receiver[Sample]]:
+        """Fetch current streams from the resampler for each phase.
 
         Args:
-            component_id: id of the EV Charger to stream power values for.
+            component_id: id of EV Charger for which current streams are being fetched.
 
         Returns:
-            A *new* receiver that will stream power values for the given EV Charger.
+            A tuple of 3 receivers stream resampled current values for the given
+                component id, one for each phase.
+        """
+
+        async def resampler_subscribe(metric_id: ComponentMetricId) -> Receiver[Sample]:
+            request = ComponentMetricRequest(
+                namespace="ev-pool",
+                component_id=component_id,
+                metric_id=metric_id,
+                start_time=None,
+            )
+            await self._resampler_subscription_sender.send(request)
+            return self._channel_registry.new_receiver(request.get_channel_name())
+
+        return (
+            await resampler_subscribe(ComponentMetricId.CURRENT_PHASE_1),
+            await resampler_subscribe(ComponentMetricId.CURRENT_PHASE_2),
+            await resampler_subscribe(ComponentMetricId.CURRENT_PHASE_3),
+        )
+
+    async def _stream_component_data(
+        self,
+        component_id: int,
+        sender: Sender[EVChargerData],
+    ) -> None:
+        """Stream 3-phase current values and state of an EV Charger.
+
+        Args:
+            component_id: id of the EV Charger for which data is requested.
+            sender: A sender to stream EV Charger data to.
 
         Raises:
-            EVChargerPoolError: if the given component_id is not part of the pool.
+            ChannelClosedError: If the channels from the resampler are closed.
         """
-        if component_id not in self._component_ids:
-            raise EVChargerPoolError(
-                f"{component_id=} is not part of the EVChargerPool"
-                f" (with ids={self._component_ids})"
-            )
-        return await self._formula_pool.from_generator(
-            f"ev_charger_current_{component_id}",
-            EVChargerPowerFormula,
-            FormulaGeneratorConfig(component_ids={component_id}),
+        if not self._state_tracker:
+            self._state_tracker = StateTracker(self._component_ids)
+
+        (phase_1_rx, phase_2_rx, phase_3_rx) = await self._get_current_streams(
+            component_id
         )
+        while True:
+            try:
+                (phase_1, phase_2, phase_3) = (
+                    await phase_1_rx.receive(),
+                    await phase_2_rx.receive(),
+                    await phase_3_rx.receive(),
+                )
+            except ChannelClosedError:
+                logger.exception("Streams closed for component_id=%s.", component_id)
+                raise
+
+            sample = Sample3Phase(
+                timestamp=phase_1.timestamp,
+                value_p1=phase_1.value,
+                value_p2=phase_2.value,
+                value_p3=phase_3.value,
+            )
+
+            if (
+                phase_1.value is None
+                and phase_2.value is None
+                and phase_3.value is None
+            ):
+                state = EVChargerState.MISSING
+            else:
+                state = self._state_tracker.get(component_id)
+
+            await sender.send(
+                EVChargerData(
+                    component_id=component_id,
+                    current=sample,
+                    state=state,
+                )
+            )