Add a EVChargerPool.set_bounds method for setting current bounds

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.

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

Author: shsms

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
 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,20 @@ 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: Current bound value to set for the EV Charger.
+        """
+        if not self._bounds_setter:
+            self._bounds_setter = BoundsSetter(self._repeat_interval)
+        await self._bounds_setter.set(component_id, max_amps)
+
     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,122 @@
+# 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
+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: Current bound value to set for the EV Charger.
+        """
+        await self._bounds_tx.send(ComponentCurrentLimit(component_id, max_amps))
+
+    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,
+                    )