Add component_data method to EVChargerPool

This method ties the resampled 3-phase current values for an ev
charger with the component state, which comes directly from the
microgrid.

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`.

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

Author: shsms

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,25 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import uuid
+from asyncio import Task
+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 +32,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,6 +76,10 @@ 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,
@@ -148,3 +166,115 @@ async def power(self, component_id: int) -> FormulaReceiver:
             EVChargerPowerFormula,
             FormulaGeneratorConfig(component_ids={component_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 for which data is requested.
+
+        Returns:
+            A receiver that streams objects containing 3-phase current and state of
+                an EV Charger.
+        """
+        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}"
+            )
+
+        task = asyncio.create_task(
+            self._stream_component_data(component_id, output_chan.new_sender())
+        )
+
+        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 EV Charger for which current streams are being fetched.
+
+        Returns:
+            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:
+            ChannelClosedError: If the channels from the resampler are closed.
+        """
+        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,
+                )
+            )

tests/timeseries/test_ev_charger_pool.py

@@ -6,8 +6,11 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import datetime
 from math import isclose
+from typing import Any
 
+from frequenz.channels import Broadcast, Receiver
 from pytest_mock import MockerFixture
 
 from frequenz.sdk import microgrid
@@ -16,6 +19,7 @@
     EVChargerCableState,
     EVChargerComponentState,
 )
+from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries.ev_charger_pool._state_tracker import (
     EVChargerState,
     StateTracker,
@@ -112,3 +116,99 @@ async def test_ev_power(  # pylint: disable=too-many-locals
 
         await mockgrid.cleanup()
         assert len(ev_results) == 10
+
+    async def test_ev_component_data(self, mocker: MockerFixture) -> None:
+        """Test the component_data method of EVChargerPool."""
+        mockgrid = MockMicrogrid(grid_side_meter=False)
+        mockgrid.add_ev_chargers(1)
+        await mockgrid.start(mocker)
+        evc_id = mockgrid.evc_ids[0]
+
+        ev_pool = microgrid.ev_charger_pool()
+
+        resampled_p1_channel = Broadcast[Sample]("resampled-current-phase-1")
+        resampled_p2_channel = Broadcast[Sample]("resampled-current-phase-2")
+        resampled_p3_channel = Broadcast[Sample]("resampled-current-phase-3")
+
+        async def send_resampled_current(
+            phase_1: float | None, phase_2: float | None, phase_3: float | None
+        ) -> None:
+            sender_p1 = resampled_p1_channel.new_sender()
+            sender_p2 = resampled_p2_channel.new_sender()
+            sender_p3 = resampled_p3_channel.new_sender()
+
+            now = datetime.now()
+            asyncio.gather(
+                sender_p1.send(Sample(now, phase_1)),
+                sender_p2.send(Sample(now, phase_2)),
+                sender_p3.send(Sample(now, phase_3)),
+            )
+
+        async def mock_current_streams(
+            _1: Any, _2: int
+        ) -> tuple[Receiver[Sample], Receiver[Sample], Receiver[Sample]]:
+            return (
+                resampled_p1_channel.new_receiver(),
+                resampled_p2_channel.new_receiver(),
+                resampled_p3_channel.new_receiver(),
+            )
+
+        mocker.patch(
+            "frequenz.sdk.timeseries.ev_charger_pool.EVChargerPool._get_current_streams",
+            mock_current_streams,
+        )
+
+        recv = await ev_pool.component_data(evc_id)
+
+        await send_resampled_current(2, 3, 5)
+        await asyncio.sleep(0.02)
+        status = await recv.receive()
+        assert (
+            status.current.value_p1,
+            status.current.value_p2,
+            status.current.value_p3,
+        ) == (2, 3, 5)
+        assert status.state == EVChargerState.MISSING
+
+        await send_resampled_current(2, 3, None)
+        await asyncio.sleep(0.02)
+        status = await recv.receive()
+        assert (
+            status.current.value_p1,
+            status.current.value_p2,
+            status.current.value_p3,
+        ) == (2, 3, None)
+        assert status.state == EVChargerState.IDLE
+
+        await send_resampled_current(None, None, None)
+        await asyncio.sleep(0.02)
+        status = await recv.receive()
+        assert (
+            status.current.value_p1,
+            status.current.value_p2,
+            status.current.value_p3,
+        ) == (None, None, None)
+        assert status.state == EVChargerState.MISSING
+
+        await send_resampled_current(None, None, None)
+        mockgrid.evc_cable_states[evc_id] = EVChargerCableState.EV_PLUGGED
+        await asyncio.sleep(0.02)
+        status = await recv.receive()
+        assert (
+            status.current.value_p1,
+            status.current.value_p2,
+            status.current.value_p3,
+        ) == (None, None, None)
+        assert status.state == EVChargerState.MISSING
+
+        await send_resampled_current(4, None, None)
+        await asyncio.sleep(0.02)
+        status = await recv.receive()
+        assert (
+            status.current.value_p1,
+            status.current.value_p2,
+            status.current.value_p3,
+        ) == (4, None, None)
+        assert status.state == EVChargerState.EV_PLUGGED
+
+        await mockgrid.cleanup()