Add a EVChargerPool.set_bounds method for setting current bounds (#297)

This method would send the given current bound for the given EV Charger
component ID, to the Microgrid API.

It would also repeat the last bound to the microgrid, every configured
interval, so that our requests don't expire.

Bounds are used to limit the max current drawn by an EV, although the
exact value will be determined by the EV.

Author: sahas-subramanian-frequenz

src/frequenz/sdk/_internal/asyncio.py

@@ -13,11 +13,15 @@
 async def cancel_and_await(task: asyncio.Task[Any]) -> None:
     """Cancel a task and wait for it to finish.
 
+    Exits immediately if the task is already done.
+
     The `CancelledError` is suppresed, but any other exception will be propagated.
 
     Args:
         task: The task to be cancelled and waited for.
     """
+    if task.done():
+        return
     task.cancel()
     try:
         await task

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

@@ -4,9 +4,11 @@
 """Interactions with EV Chargers."""
 
 from ._ev_charger_pool import EVChargerData, EVChargerPool, EVChargerPoolError
+from ._set_current_bounds import ComponentCurrentLimit
 from ._state_tracker import EVChargerState
 
 __all__ = [
+    "ComponentCurrentLimit",
     "EVChargerPool",
     "EVChargerData",
     "EVChargerPoolError",

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

@@ -11,6 +11,7 @@
 from asyncio import Task
 from collections import abc
 from dataclasses import dataclass
+from datetime import timedelta
 
 from frequenz.channels import Broadcast, ChannelClosedError, Receiver, Sender
 
@@ -24,6 +25,7 @@
     EVChargerPowerFormula,
     FormulaGeneratorConfig,
 )
+from ._set_current_bounds import BoundsSetter, ComponentCurrentLimit
 from ._state_tracker import EVChargerState, StateTracker
 
 logger = logging.getLogger(__name__)
@@ -50,6 +52,7 @@ def __init__(
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
         component_ids: set[int] | None = None,
+        repeat_interval: timedelta = timedelta(seconds=3.0),
     ) -> None:
         """Create an `EVChargerPool` instance.
 
@@ -61,8 +64,11 @@ def __init__(
             component_ids: An optional list of component_ids belonging to this pool.  If
                 not specified, IDs of all EV Chargers in the microgrid will be fetched
                 from the component graph.
+            repeat_interval: Interval after which to repeat the last set bounds to the
+                microgrid API, if no new calls to `set_bounds` have been made.
         """
         self._channel_registry: ChannelRegistry = channel_registry
+        self._repeat_interval = repeat_interval
         self._resampler_subscription_sender: Sender[
             ComponentMetricRequest
         ] = resampler_subscription_sender
@@ -87,6 +93,7 @@ def __init__(
             self._channel_registry,
             self._resampler_subscription_sender,
         )
+        self._bounds_setter: BoundsSetter | None = None
 
     @property
     def component_ids(self) -> abc.Set[int]:
@@ -158,6 +165,40 @@ async def component_data(self, component_id: int) -> Receiver[EVChargerData]:
 
         return output_chan.new_receiver()
 
+    async def set_bounds(self, component_id: int, max_amps: float) -> None:
+        """Send given max current bound for the given EV Charger to the microgrid API.
+
+        Bounds are used to limit the max current drawn by an EV, although the exact
+        value will be determined by the EV.
+
+        Args:
+            component_id: ID of EV Charger to set the current bounds to.
+            max_amps: maximum current in amps, that an EV can draw from this EV Charger.
+        """
+        if not self._bounds_setter:
+            self._bounds_setter = BoundsSetter(self._repeat_interval)
+        await self._bounds_setter.set(component_id, max_amps)
+
+    def new_bounds_sender(self) -> Sender[ComponentCurrentLimit]:
+        """Return a `Sender` for setting EV Charger current bounds with.
+
+        Bounds are used to limit the max current drawn by an EV, although the exact
+        value will be determined by the EV.
+
+        Returns:
+            A new `Sender`.
+        """
+        if not self._bounds_setter:
+            self._bounds_setter = BoundsSetter(self._repeat_interval)
+        return self._bounds_setter.new_bounds_sender()
+
+    async def stop(self) -> None:
+        """Stop all tasks and channels owned by the EVChargerPool."""
+        if self._bounds_setter:
+            await self._bounds_setter.stop()
+        if self._state_tracker:
+            await self._state_tracker.stop()
+
     async def _get_current_streams(
         self, component_id: int
     ) -> tuple[Receiver[Sample], Receiver[Sample], Receiver[Sample]]:

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

@@ -0,0 +1,130 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""A task for sending EV Charger power bounds to the microgrid API."""
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Dict
+
+from frequenz.channels import Broadcast, Sender
+from frequenz.channels.util import Select, Timer
+
+from ..._internal.asyncio import cancel_and_await
+from ...microgrid import connection_manager
+from ...microgrid.component import ComponentCategory
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ComponentCurrentLimit:
+    """A current limit, to be sent to the EV Charger."""
+
+    component_id: int
+    max_amps: float
+
+
+class BoundsSetter:
+    """A task for sending EV Charger power bounds to the microgrid API.
+
+    Also, periodically resends the last set bounds to the microgrid API, if no new
+    bounds have been set.
+    """
+
+    _NUM_PHASES = 3
+
+    def __init__(self, repeat_interval: timedelta) -> None:
+        """Create a `BoundsSetter` instance.
+
+        Args:
+            repeat_interval: Interval after which to repeat the last set bounds to the
+                microgrid API, if no new calls to `set_bounds` have been made.
+        """
+        self._repeat_interval = repeat_interval
+
+        self._task: asyncio.Task[None] = asyncio.create_task(self._run())
+        self._bounds_chan: Broadcast[ComponentCurrentLimit] = Broadcast("BoundsSetter")
+        self._bounds_rx = self._bounds_chan.new_receiver()
+        self._bounds_tx = self._bounds_chan.new_sender()
+
+    async def set(self, component_id: int, max_amps: float) -> None:
+        """Send the given current limit to the microgrid for the given component id.
+
+        Args:
+            component_id: ID of EV Charger to set the current bounds to.
+            max_amps: maximum current in amps, that an EV can draw from this EV Charger.
+        """
+        await self._bounds_tx.send(ComponentCurrentLimit(component_id, max_amps))
+
+    def new_bounds_sender(self) -> Sender[ComponentCurrentLimit]:
+        """Return a `Sender` for setting EV Charger current bounds with.
+
+        Returns:
+            A new `Sender`.
+        """
+        return self._bounds_chan.new_sender()
+
+    async def stop(self) -> None:
+        """Stop the BoundsSetter."""
+        await self._bounds_chan.close()
+        await cancel_and_await(self._task)
+
+    async def _run(self) -> None:
+        """Wait for new bounds and forward them to the microgrid API.
+
+        Also, periodically resend the last set bounds to the microgrid API, if no new
+        bounds have been set.
+
+        Raises:
+            RuntimeError: If no meters are found in the component graph.
+            ValueError: If the meter channel is closed.
+        """
+        api_client = connection_manager.get().api_client
+        graph = connection_manager.get().component_graph
+        meters = graph.components(component_category={ComponentCategory.METER})
+        if not meters:
+            err = "No meters found in the component graph."
+            logger.error(err)
+            raise RuntimeError(err)
+
+        meter_data = (
+            await api_client.meter_data(next(iter(meters)).component_id)
+        ).into_peekable()
+        latest_bound: Dict[int, ComponentCurrentLimit] = {}
+
+        select = Select(
+            bound_chan=self._bounds_rx,
+            timer=Timer(self._repeat_interval.total_seconds()),
+        )
+        while await select.ready():
+            meter = meter_data.peek()
+            if meter is None:
+                raise ValueError("Meter channel closed.")
+
+            if msg := select.bound_chan:
+                bound: ComponentCurrentLimit = msg.inner
+                if (
+                    bound.component_id in latest_bound
+                    and latest_bound[bound.component_id] == bound
+                ):
+                    continue
+                latest_bound[bound.component_id] = bound
+                min_voltage = min(meter.voltage_per_phase)
+                logging.info("sending new bounds: %s", bound)
+                await api_client.set_bounds(
+                    bound.component_id,
+                    0,
+                    bound.max_amps * min_voltage * self._NUM_PHASES,
+                )
+            elif msg := select.timer:
+                for bound in latest_bound.values():
+                    min_voltage = min(meter.voltage_per_phase)
+                    logging.debug("resending bounds: %s", bound)
+                    await api_client.set_bounds(
+                        bound.component_id,
+                        0,
+                        bound.max_amps * min_voltage * self._NUM_PHASES,
+                    )