Control methods for EVChargerPool using PowerManager and PowerDistributor (#900)

This PR adds new `propose_power` and `power_status` methods to the
`EVChargerPool` similar to the `BatteryPool`. These method interface
with the `PowerManager` and `PowerDistributor`, which currently uses a
first-come-first-serve algorithm to distribute power to EVs.

The following changes were made:
- An `EVChargerStatusTracker` implementation which tracks the health of
an EV charger and whether an EV is connected.
- `PowerDistributor` and `PowerManager` are now modular to support
multiple component types.
- Add an `EVChargerManager` that plugs into the `PowerDistributor`
- Connect the `PowerDistributor` and `PowerManager` to the
`EVChargerPool`.

Fixes #872.

Author: shsms

RELEASE_NOTES.md

@@ -16,6 +16,8 @@
 
 - Support for per-component interaction in `EVChargerPool` has been removed.
 
+- New `propose_power` and `power_status` methods have been added to the `EVChargerPool` similar to the `BatteryPool`.  These method interface with the `PowerManager` and `PowerDistributor`, which currently uses a first-come-first-serve algorithm to distribute power to EVs.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

benchmarks/power_distribution/power_distributor.py

@@ -119,6 +119,7 @@ async def run_test(  # pylint: disable=too-many-locals
         requests_receiver=power_request_channel.new_receiver(),
         results_sender=power_result_channel.new_sender(),
         component_pool_status_sender=battery_status_channel.new_sender(),
+        wait_for_data_sec=2.0,
     ):
         tasks: list[Coroutine[Any, Any, list[Result]]] = []
         tasks.append(send_requests(batteries, num_requests))

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -133,17 +133,22 @@ def _add_bounds_tracker(self, component_ids: frozenset[int]) -> None:
             microgrid,
         )
 
-        if self._component_category is not ComponentCategory.BATTERY:
+        bounds_receiver: Receiver[SystemBounds]
+        # pylint: disable=protected-access
+        if self._component_category is ComponentCategory.BATTERY:
+            battery_pool = microgrid.battery_pool(component_ids)
+            bounds_receiver = battery_pool._system_power_bounds.new_receiver()
+        elif self._component_category is ComponentCategory.EV_CHARGER:
+            ev_charger_pool = microgrid.ev_charger_pool(component_ids)
+            bounds_receiver = ev_charger_pool._system_power_bounds.new_receiver()
+        # pylint: enable=protected-access
+        else:
             err = (
                 "PowerManagingActor: Unsupported component category: "
                 f"{self._component_category}"
             )
             _logger.error(err)
             raise NotImplementedError(err)
-        battery_pool = microgrid.battery_pool(component_ids)
-        # pylint: disable=protected-access
-        bounds_receiver = battery_pool._system_power_bounds.new_receiver()
-        # pylint: enable=protected-access
 
         self._system_bounds[component_ids] = SystemBounds(
             timestamp=datetime.now(tz=timezone.utc),

src/frequenz/sdk/actor/power_distributing/_component_managers/__init__.py

@@ -5,8 +5,10 @@
 
 from ._battery_manager import BatteryManager
 from ._component_manager import ComponentManager
+from ._ev_charger_manager import EVChargerManager
 
 __all__ = [
     "BatteryManager",
     "ComponentManager",
+    "EVChargerManager",
 ]

src/frequenz/sdk/actor/power_distributing/_component_managers/_battery_manager.py

@@ -127,8 +127,18 @@ class BatteryManager(ComponentManager):
     def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
+        results_sender: Sender[Result],
     ):
-        """Initialize the battery data manager."""
+        """Initialize this instance.
+
+        Args:
+            component_pool_status_sender: Channel sender to send the status of the
+                battery pool to.  This status is used by the battery pool metric
+                streams, to dynamically adjust the values based on the health of the
+                individual batteries.
+            results_sender: Channel sender to send the power distribution results to.
+        """
+        self._results_sender = results_sender
         self._batteries = connection_manager.get().component_graph.components(
             component_categories={ComponentCategory.BATTERY}
         )
@@ -181,20 +191,18 @@ async def stop(self) -> None:
         await self._component_pool_status_tracker.stop()
 
     @override
-    async def distribute_power(self, request: Request) -> Result:
+    async def distribute_power(self, request: Request) -> None:
         """Distribute the requested power to the components.
 
         Args:
             request: Request to get the distribution for.
-
-        Returns:
-            Result of the distribution.
         """
         distribution_result = await self._get_distribution(request)
         if not isinstance(distribution_result, DistributionResult):
-            return distribution_result
-        result = await self._distribute_power(request, distribution_result)
-        return result
+            result = distribution_result
+        else:
+            result = await self._distribute_power(request, distribution_result)
+        await self._results_sender.send(result)
 
     async def _get_distribution(self, request: Request) -> DistributionResult | Result:
         """Get the distribution of the batteries.

src/frequenz/sdk/actor/power_distributing/_component_managers/_component_manager.py

@@ -20,12 +20,14 @@ class ComponentManager(abc.ABC):
     def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
+        results_sender: Sender[Result],
     ):
         """Initialize the component data manager.
 
         Args:
             component_pool_status_sender: Channel for sending information about which
                 components are expected to be working.
+            results_sender: Channel for sending the results of power distribution.
         """
 
     @abc.abstractmethod
@@ -37,7 +39,7 @@ async def start(self) -> None:
         """Start the component data manager."""
 
     @abc.abstractmethod
-    async def distribute_power(self, request: Request) -> Result:
+    async def distribute_power(self, request: Request) -> None:
         """Distribute the requested power to the components.
 
         Args:

src/frequenz/sdk/actor/power_distributing/_component_managers/_ev_charger_manager/__init__.py

@@ -0,0 +1,10 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Manage ev chargers for the power distributor."""
+
+from ._ev_charger_manager import EVChargerManager
+
+__all__ = [
+    "EVChargerManager",
+]

src/frequenz/sdk/actor/power_distributing/_component_managers/_ev_charger_manager/_config.py

@@ -0,0 +1,27 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Configuration for the power distributor's EV charger manager."""
+
+from collections import abc
+from dataclasses import dataclass, field
+from datetime import timedelta
+
+from .....timeseries import Current
+
+
+@dataclass(frozen=True)
+class EVDistributionConfig:
+    """Configuration for the power distributor's EV charger manager."""
+
+    component_ids: abc.Set[int]
+    """The component ids of the EV chargers."""
+
+    min_current: Current = field(default_factory=lambda: Current.from_amperes(6.0))
+    """The minimum current that can be allocated to an EV charger."""
+
+    initial_current: Current = field(default_factory=lambda: Current.from_amperes(10.0))
+    """The initial current that can be allocated to an EV charger."""
+
+    increase_power_interval: timedelta = timedelta(seconds=60)
+    """The interval at which the power can be increased for an EV charger."""