Implement the PowerManagingActor (#692)

This PR add a PowerManagingActor implementation with the Matryoshka
algorithm, that uses actor priorities to resolve conflicting power
requests when multiple actors are trying to use the same batteries.

The Matryoshka algorithm requires actors to specify a desired power
value, and optionally, bounds to limit the range of power values
available for lower priority actors.

Some of the other changes in this PR are:

- updates the `microgrid.battery_pool()` method to accept a priority
  value, and return a new battery pool instance with each call, while
  using a shared reference store class internally, that helps reuse
  formulas, metric aggregators and battery status tracking tasks.

- update BatteryPool's control methods to send requests to the
  PowerManagingActor rather than the PowerDistributingActor.

- Rename the control methods in the BatteryPool from
  `set_power/charge/discharge` -> `propose_{power/charge/discharge}`,
  and update them to accept optional bounds to apply on lower priority
  actors.  The new methods also don't accept a `adjust_power` flag,
  because the `PowerManagingActor` will always adjust power to fit
  within the available bounds.

- Rename `BatteryPool.power_bounds` to a private method called
  `_system_power_bounds`, because it will be used only by the
  PowerManager as the top level bounds.

- Add `BatteryPool.power_status`, which streams
  - bounds adjusted to an actors priority,
  - the latest target power for a set of batteries,
  - the result from the power distributor for the last call.

- Update the PowerDistributingActor to have a single input and output
  channel, because the PowerManagingActor will be the only component
  that uses the PowerDistributingActor from now on.

Closes #161

Author: llucax

RELEASE_NOTES.md

@@ -6,7 +6,21 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+- `microgrid.battery_pool()` method now accepts a priority value.
+
+- `BatteryPool`'s control methods
+
+  * Original methods `{set_power/charge/discharge}` are now replaced by `propose_{power/charge/discharge}`
+  * The `propose_*` methods send power proposals to the `PowerManagingActor`, where it can be overridden by proposals from other actors.
+  * They no longer have the `adjust_power` flag, because the `PowerManagingActor` will always adjust power to fit within the available bounds.
+
+- `BatteryPool`'s reporting methods
+
+  * `power_bounds` is replaced by `power_status`
+  * The `power_status` method streams objects containing:
+    + bounds adjusted to the actor's priorities
+    + the latest target power for the set of batteries
+    + the results from the power distributor for the last request
 
 ## New Features
 
@@ -31,12 +45,11 @@
   - Add `fill_value` option to window method to impute missing values. By default missing values are imputed with `NaN`.
 - Add `at` method to `MovingWindow` to access a single element and use it in `__getitem__` magic to fully support single element access.
 
-
-
 - The PowerDistributingActor now supports n:m relations between inverters and
   batteries.
   This means that one or more inverters can be connected to one or more batteries.
 
+- A `PowerManagingActor` implementation
 
 ## Bug Fixes
 

benchmarks/power_distribution/power_distributor.py

@@ -14,7 +14,7 @@
 from frequenz.channels import Broadcast
 
 from frequenz.sdk import microgrid
-from frequenz.sdk.actor import ChannelRegistry, ResamplerConfig
+from frequenz.sdk.actor import ResamplerConfig
 from frequenz.sdk.actor.power_distributing import (
     BatteryStatus,
     Error,
@@ -33,6 +33,10 @@
 PORT = 61060
 
 
+# TODO: this send_requests function uses the battery pool to # pylint: disable=fixme
+# send requests, and those no longer go directly to the power distributing actor, but
+# instead through the power managing actor.  So the below function needs to be updated
+# to use the PowerDistributingActor directly.
 async def send_requests(batteries: set[int], request_num: int) -> list[Result]:
     """Send requests to the PowerDistributingActor and wait for the response.
 
@@ -47,10 +51,12 @@ async def send_requests(batteries: set[int], request_num: int) -> list[Result]:
         List of the results from the PowerDistributingActor.
     """
     battery_pool = microgrid.battery_pool(batteries)
-    results_rx = battery_pool.power_distribution_results()
-    result: list[Result] = []
+    results_rx = battery_pool.power_status.new_receiver()
+    result: list[Any] = []
     for _ in range(request_num):
-        await battery_pool.set_power(Power(float(random.randrange(100000, 1000000))))
+        await battery_pool.propose_power(
+            Power(float(random.randrange(100000, 1000000)))
+        )
         try:
             output = await asyncio.wait_for(results_rx.receive(), timeout=3)
             if output is None:
@@ -107,10 +113,10 @@ async def run_test(  # pylint: disable=too-many-locals
 
     power_request_channel = Broadcast[Request]("power-request")
     battery_status_channel = Broadcast[BatteryStatus]("battery-status")
-    channel_registry = ChannelRegistry(name="power_distributor")
+    power_result_channel = Broadcast[Result]("power-result")
     async with PowerDistributingActor(
-        channel_registry=channel_registry,
         requests_receiver=power_request_channel.new_receiver(),
+        results_sender=power_result_channel.new_sender(),
         battery_status_sender=battery_status_channel.new_sender(),
     ):
         tasks: list[Coroutine[Any, Any, list[Result]]] = []

examples/battery_pool.py

@@ -34,7 +34,9 @@ async def main() -> None:
     receivers: dict[str, Receiver[Any]] = {
         "soc": battery_pool.soc.new_receiver(maxsize=1),
         "capacity": battery_pool.capacity.new_receiver(maxsize=1),
-        "power_bounds": battery_pool.power_bounds.new_receiver(maxsize=1),
+        # pylint: disable=protected-access
+        "power_bounds": battery_pool._system_power_bounds.new_receiver(maxsize=1),
+        # pylint: enable=protected-access
     }
 
     merged_channel = MergeNamed[Any](**receivers)

src/frequenz/sdk/_internal/_channels.py

@@ -0,0 +1,26 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""General purpose classes for use with channels."""
+
+import abc
+import typing
+
+from frequenz.channels import Receiver
+
+T = typing.TypeVar("T")
+
+
+class ReceiverFetcher(typing.Generic[T], typing.Protocol):
+    """An interface that just exposes a `new_receiver` method."""
+
+    @abc.abstractmethod
+    def new_receiver(self, maxsize: int = 50) -> Receiver[T]:
+        """Get a receiver from the channel.
+
+        Args:
+            maxsize: The maximum size of the receiver.
+
+        Returns:
+            A receiver instance.
+        """

src/frequenz/sdk/actor/_channel_registry.py

@@ -3,10 +3,14 @@
 
 """A class that would dynamically create, own and provide access to channels."""
 
-from typing import Any
+from __future__ import annotations
+
+import typing
 
 from frequenz.channels import Broadcast, Receiver, Sender
 
+from .._internal._channels import ReceiverFetcher
+
 
 class ChannelRegistry:
     """Dynamically creates, own and provide access to channels.
@@ -22,9 +26,9 @@ def __init__(self, *, name: str) -> None:
             name: A unique name for the registry.
         """
         self._name = name
-        self._channels: dict[str, Broadcast[Any]] = {}
+        self._channels: dict[str, Broadcast[typing.Any]] = {}
 
-    def new_sender(self, key: str) -> Sender[Any]:
+    def new_sender(self, key: str) -> Sender[typing.Any]:
         """Get a sender to a dynamically created channel with the given key.
 
         Args:
@@ -37,18 +41,32 @@ def new_sender(self, key: str) -> Sender[Any]:
             self._channels[key] = Broadcast(f"{self._name}-{key}")
         return self._channels[key].new_sender()
 
-    def new_receiver(self, key: str) -> Receiver[Any]:
+    def new_receiver(self, key: str, maxsize: int = 50) -> Receiver[typing.Any]:
         """Get a receiver to a dynamically created channel with the given key.
 
         Args:
             key: A key to identify the channel.
+            maxsize: The maximum size of the receiver.
 
         Returns:
             A receiver for a dynamically created channel with the given key.
         """
         if key not in self._channels:
             self._channels[key] = Broadcast(f"{self._name}-{key}")
-        return self._channels[key].new_receiver()
+        return self._channels[key].new_receiver(maxsize=maxsize)
+
+    def new_receiver_fetcher(self, key: str) -> ReceiverFetcher[typing.Any]:
+        """Get a receiver fetcher to a dynamically created channel with the given key.
+
+        Args:
+            key: A key to identify the channel.
+
+        Returns:
+            A receiver fetcher for a dynamically created channel with the given key.
+        """
+        if key not in self._channels:
+            self._channels[key] = Broadcast(f"{self._name}-{key}")
+        return _RegistryReceiverFetcher(self, key)
 
     async def _close_channel(self, key: str) -> None:
         """Close a channel with the given key.
@@ -61,3 +79,35 @@ async def _close_channel(self, key: str) -> None:
         if key in self._channels:
             if channel := self._channels.pop(key, None):
                 await channel.close()
+
+
+T = typing.TypeVar("T")
+
+
+class _RegistryReceiverFetcher(typing.Generic[T]):
+    """A receiver fetcher that is bound to a channel registry and a key."""
+
+    def __init__(
+        self,
+        registry: ChannelRegistry,
+        key: str,
+    ) -> None:
+        """Create a new instance of a receiver fetcher.
+
+        Args:
+            registry: The channel registry.
+            key: A key to identify the channel.
+        """
+        self._registry = registry
+        self._key = key
+
+    def new_receiver(self, maxsize: int = 50) -> Receiver[T]:
+        """Get a receiver from the channel.
+
+        Args:
+            maxsize: The maximum size of the receiver.
+
+        Returns:
+            A receiver instance.
+        """
+        return self._registry.new_receiver(self._key, maxsize)

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

@@ -0,0 +1,15 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""A power manager implementation."""
+
+from ._base_classes import Algorithm, Proposal, Report, ReportRequest
+from ._power_managing_actor import PowerManagingActor
+
+__all__ = [
+    "Algorithm",
+    "PowerManagingActor",
+    "Proposal",
+    "Report",
+    "ReportRequest",
+]