Cleanup the actor package (#1031)

The `frequenz.sdk.actor` package was shared by the `Actor`
implementation and by a number of core actors of the SDK and other
utilities.

This PR moves all the core actors and utilities into more appropriate
locations.

- `frequenz.sdk.`
  + `actor.ChannelRegistry` → `_internal._channels.ChannelRegistry`
  + `actor.ConfigManagingActor` → `config.ConfigManagingActor`
  + `actor._power_managing` → `microgrid._power_managing`
  + `actor.power_distributing` → `microgrid._power_distributing`
  + `actor._resampling` → `microgrid._resampling`
  + `actor._data_sourcing` → `microgrid._data_sourcing`

This is part of #575

Author: shsms

RELEASE_NOTES.md

@@ -23,6 +23,21 @@
 
 - The `ConfigManagingActor` now uses `collections.abc.Mapping` as the output sender type. This change indicates that the broadcasted configuration is intended to be read-only.
 
+- The `ConfigManagingActor` has moved from `frequenz.sdk.actor` to `frequenz.sdk.config`.
+
+- The following core actors are no longer part of the public API:
+  - `PowerDistributingActor`
+  - `ComponentMetricsResamplingActor`
+  - `DataSourcingActor`
+
+- The following two types which are used for communicating with the data sourcing and resampling actors are also no longer part of the public API:
+  - `ComponentMetricId`
+  - `ComponentMetricRequest`
+
+- The `ChannelRegistry` is no longer part of the public API.
+
+- The `Result` types for the power distribution results are now exposed through the `frequenz.sdk.microgrid.battery_pool` module.
+
 ## New Features
 
 - Classes `Bounds` and `SystemBounds` now implement the `__contains__` method, allowing the use of the `in` operator to check whether a value falls within the bounds or not.

benchmarks/power_distribution/power_distributor.py

@@ -16,7 +16,8 @@
 
 from frequenz.sdk import microgrid
 from frequenz.sdk.actor import ResamplerConfig
-from frequenz.sdk.actor.power_distributing import (
+from frequenz.sdk.microgrid import connection_manager
+from frequenz.sdk.microgrid._power_distributing import (
     ComponentPoolStatus,
     Error,
     OutOfBounds,
@@ -26,7 +27,6 @@
     Result,
     Success,
 )
-from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.timeseries._quantities import Power
 
 HOST = "microgrid.sandbox.api.frequenz.io"

benchmarks/timeseries/benchmark_datasourcing.py

@@ -20,8 +20,8 @@
 from frequenz.client.microgrid import ComponentMetricId
 
 from frequenz.sdk import microgrid
-from frequenz.sdk.actor import (
-    ChannelRegistry,
+from frequenz.sdk._internal._channels import ChannelRegistry
+from frequenz.sdk.microgrid._data_sourcing import (
     ComponentMetricRequest,
     DataSourcingActor,
 )

src/frequenz/sdk/_internal/_channels.py

@@ -4,12 +4,18 @@
 """General purpose classes for use with channels."""
 
 import abc
+import dataclasses
+import logging
+import traceback
 import typing
 
-from frequenz.channels import Receiver
+from frequenz.channels import Broadcast, Receiver
+
+_logger = logging.getLogger(__name__)
 
 T_co = typing.TypeVar("T_co", covariant=True)
 U_co = typing.TypeVar("U_co", covariant=True)
+T = typing.TypeVar("T")
 
 
 class ReceiverFetcher(typing.Generic[T_co], typing.Protocol):
@@ -55,3 +61,141 @@ def new_receiver(self, *, limit: int = 50) -> Receiver[U_co]:
             A receiver instance.
         """
         return self._mapping_function(self._fetcher.new_receiver(limit=limit))
+
+
+class ChannelRegistry:
+    """Dynamically creates, own and provide access to broadcast channels.
+
+    It can be used by actors to dynamically establish a communication channel
+    between each other.
+
+    The registry is responsible for creating channels when they are first requested via
+    the [`get_or_create()`][frequenz.sdk.actor.ChannelRegistry.get_or_create] method.
+
+    The registry also stores type information to make sure that the same channel is not
+    used for different message types.
+
+    Since the registry owns the channels, it is also responsible for closing them when
+    they are no longer needed. There is no way to remove a channel without closing it.
+
+    Note:
+        This registry stores [`Broadcast`][frequenz.channels.Broadcast] channels.
+    """
+
+    def __init__(self, *, name: str) -> None:
+        """Initialize this registry.
+
+        Args:
+            name: A name to identify the registry in the logs. This name is also used as
+                a prefix for the channel names.
+        """
+        self._name = name
+        self._channels: dict[str, _Entry] = {}
+
+    @property
+    def name(self) -> str:
+        """The name of this registry."""
+        return self._name
+
+    def message_type(self, key: str) -> type:
+        """Get the message type of the channel for the given key.
+
+        Args:
+            key: The key to identify the channel.
+
+        Returns:
+            The message type of the channel.
+
+        Raises:
+            KeyError: If the channel does not exist.
+        """
+        entry = self._channels.get(key)
+        if entry is None:
+            raise KeyError(f"No channel for key {key!r} exists.")
+        return entry.message_type
+
+    def __contains__(self, key: str) -> bool:
+        """Check whether the channel for the given `key` exists."""
+        return key in self._channels
+
+    def get_or_create(self, message_type: type[T], key: str) -> Broadcast[T]:
+        """Get or create a channel for the given key.
+
+        If a channel for the given key already exists, the message type of the existing
+        channel is checked against the requested message type. If they do not match,
+        a `ValueError` is raised.
+
+        Note:
+            The types have to match exactly, it doesn't do a subtype check due to
+            technical limitations. In the future subtype checks might be supported.
+
+        Args:
+            message_type: The type of the message that is sent through the channel.
+            key: The key to identify the channel.
+
+        Returns:
+            The channel for the given key.
+
+        Raises:
+            ValueError: If the channel exists and the message type does not match.
+        """
+        if key not in self._channels:
+            if _logger.isEnabledFor(logging.DEBUG):
+                _logger.debug(
+                    "Creating a new channel for key %r with type %s at:\n%s",
+                    key,
+                    message_type,
+                    "".join(traceback.format_stack(limit=10)[:9]),
+                )
+            self._channels[key] = _Entry(
+                message_type, Broadcast(name=f"{self._name}-{key}")
+            )
+
+        entry = self._channels[key]
+        if entry.message_type is not message_type:
+            exception = ValueError(
+                f"Type mismatch, a channel for key {key!r} exists and the requested "
+                f"message type {message_type} is not the same as the existing "
+                f"message type {entry.message_type}."
+            )
+            if _logger.isEnabledFor(logging.DEBUG):
+                _logger.debug(
+                    "%s at:\n%s",
+                    str(exception),
+                    # We skip the last frame because it's this method, and limit the
+                    # stack to 9 frames to avoid adding too much noise.
+                    "".join(traceback.format_stack(limit=10)[:9]),
+                )
+            raise exception
+
+        return typing.cast(Broadcast[T], entry.channel)
+
+    async def close_and_remove(self, key: str) -> None:
+        """Remove the channel for the given key.
+
+        Args:
+            key: The key to identify the channel.
+
+        Raises:
+            KeyError: If the channel does not exist.
+        """
+        entry = self._channels.pop(key, None)
+        if entry is None:
+            raise KeyError(f"No channel for key {key!r} exists.")
+        await entry.channel.close()
+
+
+@dataclasses.dataclass(frozen=True)
+class _Entry:
+    """An entry in a channel registry."""
+
+    message_type: type
+    """The type of the message that is sent through the channel in this entry."""
+
+    # We use object instead of Any to minimize the chances of hindering type checking.
+    # If for some reason the channel is not casted to the proper underlaying type, when
+    # using object at least accessing any member that's not part of the object base
+    # class will yield a type error, while if we used Any, it would not and the issue
+    # would be much harder to find.
+    channel: Broadcast[object]
+    """The channel in this entry."""

src/frequenz/sdk/actor/__init__.py

@@ -599,21 +599,11 @@ async def main() -> None:  # (6)!
 from ..timeseries._resampling import ResamplerConfig
 from ._actor import Actor
 from ._background_service import BackgroundService
-from ._channel_registry import ChannelRegistry
-from ._config_managing import ConfigManagingActor
-from ._data_sourcing import ComponentMetricId, ComponentMetricRequest, DataSourcingActor
-from ._resampling import ComponentMetricsResamplingActor
 from ._run_utils import run
 
 __all__ = [
     "Actor",
     "BackgroundService",
-    "ChannelRegistry",
-    "ComponentMetricId",
-    "ComponentMetricRequest",
-    "ComponentMetricsResamplingActor",
-    "ConfigManagingActor",
-    "DataSourcingActor",
     "ResamplerConfig",
     "run",
 ]

src/frequenz/sdk/actor/_channel_registry.py

@@ -0,0 +1,8 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Read and update config variables."""
+
+from ._config_managing import ConfigManagingActor
+
+__all__ = ["ConfigManagingActor"]