Make PowerManagingActor more modular and generic (#773)

This PR updates the `PowerManagingActor` to

- take type of components we are controlling. Currently only batteries
  are supported.
- Remove all other references to batteries from the
  `PowerManagingActor`, except in the interactions with the
  `PowerDistributingActor`.

The `PowerDistributingActor` needs to be modularized as well, before
the either of these actors can be used with other types of components.

All changes in this PR are internal and don't affect the external
interface of the `BatteryPool`.

Author: shsms

RELEASE_NOTES.md

@@ -6,7 +6,7 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+- The `BatteryPool.power_status` method now streams objects of type `BatteryPoolReport`.  They're identical to the previous `Report` objects, except for the name of the class.
 
 ## New Features
 

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

@@ -3,13 +3,13 @@
 
 """A power manager implementation."""
 
-from ._base_classes import Algorithm, Proposal, Report, ReportRequest
+from ._base_classes import Algorithm, Proposal, ReportRequest, _Report
 from ._power_managing_actor import PowerManagingActor
 
 __all__ = [
     "Algorithm",
     "PowerManagingActor",
     "Proposal",
-    "Report",
+    "_Report",
     "ReportRequest",
 ]

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

@@ -16,7 +16,7 @@
 from . import _bounds
 
 if typing.TYPE_CHECKING:
-    from ...timeseries.battery_pool import PowerMetrics
+    from ...timeseries._base_types import SystemBounds
     from .. import power_distributing
 
 
@@ -27,8 +27,8 @@ class ReportRequest:
     source_id: str
     """The source ID of the actor sending the request."""
 
-    battery_ids: frozenset[int]
-    """The battery IDs to report on."""
+    component_ids: frozenset[int]
+    """The component IDs to report on."""
 
     priority: int
     """The priority of the actor ."""
@@ -40,15 +40,67 @@ def get_channel_name(self) -> str:
             The channel name to use to identify the corresponding report channel
                 from the channel registry.
         """
-        return f"power_manager.report.{self.battery_ids=}.{self.priority=}"
+        return f"power_manager.report.{self.component_ids=}.{self.priority=}"
+
+
+class Report(typing.Protocol):
+    """Current PowerManager report for a set of components.
+
+    This protocol can be specialized by different component pools to provide more
+    specific details and documentation for the reports.
+    """
+
+    @property
+    def bounds(self) -> timeseries.Bounds[Power] | None:
+        """The bounds for the components.
+
+        These bounds are adjusted to any restrictions placed by actors with higher
+        priorities.
+
+        There might be exclusion zones within these bounds. If necessary, the
+        `adjust_to_bounds` method may be used to check if a desired power value fits the
+        bounds, or to get the closest possible power values that do fit the bounds.
+        """
+
+    @abc.abstractmethod
+    def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
+        """Adjust a power value to the bounds.
+
+        This method can be used to adjust a desired power value to the power bounds
+        available to the actor.
+
+        If the given power value falls within the usable bounds, it will be returned
+        unchanged.
+
+        If it falls outside the usable bounds, the closest possible value on the
+        corresponding side will be returned.  For example, if the given power is lower
+        than the lowest usable power, only the lowest usable power will be returned, and
+        similarly for the highest usable power.
+
+        If the given power falls within an exclusion zone that's contained within the
+        usable bounds, the closest possible power values on both sides will be returned.
+
+        !!! note
+            It is completely optional to use this method to adjust power values before
+            proposing them, because the PowerManager will do this automatically.  This
+            method is provided for convenience, and for granular control when there are
+            two possible power values, both of which fall within the available bounds.
+
+        Args:
+            power: The power value to adjust.
+
+        Returns:
+            A tuple of the closest power values to the desired power that fall within
+                the available bounds for the actor.
+        """
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
-class Report:
-    """Current PowerManager report for a set of batteries."""
+class _Report(Report):
+    """Current PowerManager report for a set of components."""
 
     target_power: Power | None
-    """The currently set power for the batteries."""
+    """The currently set power for the components."""
 
     distribution_result: power_distributing.Result | None
     """The result of the last power distribution.
@@ -57,14 +109,14 @@ class Report:
     """
 
     _inclusion_bounds: timeseries.Bounds[Power] | None
-    """The available inclusion bounds for the batteries, for the actor's priority.
+    """The available inclusion bounds for the components, for the actor's priority.
 
     These bounds are adjusted to any restrictions placed by actors with higher
     priorities.
     """
 
     _exclusion_bounds: timeseries.Bounds[Power] | None
-    """The exclusion bounds for the batteries.
+    """The exclusion bounds for the components.
 
     The power manager doesn't manage exclusion bounds, so these are aggregations of
     values reported by the microgrid API.
@@ -75,15 +127,14 @@ class Report:
 
     @property
     def bounds(self) -> timeseries.Bounds[Power] | None:
-        """The bounds for the batteries.
+        """The bounds for the components.
 
         These bounds are adjusted to any restrictions placed by actors with higher
         priorities.
 
         There might be exclusion zones within these bounds. If necessary, the
-        [`adjust_to_bounds`][frequenz.sdk.timeseries.battery_pool.Report.adjust_to_bounds]
-        method may be used to check if a desired power value fits the bounds, or to get
-        the closest possible power values that do fit the bounds.
+        `adjust_to_bounds` method may be used to check if a desired power value fits the
+        bounds, or to get the closest possible power values that do fit the bounds.
         """
         return self._inclusion_bounds
 
@@ -106,29 +157,9 @@ def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
 
         !!! note
             It is completely optional to use this method to adjust power values before
-            proposing them through the battery pool, because the battery pool will do
-            this automatically.  This method is provided for convenience, and for
-            granular control when there are two possible power values, both of which
-            fall within the available bounds.
-
-        Example:
-            ```python
-            from frequenz.sdk import microgrid
-
-            power_status_rx = microgrid.battery_pool().power_status.new_receiver()
-            power_status = await power_status_rx.receive()
-            desired_power = Power.from_watts(1000.0)
-
-            match power_status.adjust_to_bounds(desired_power):
-                case (power, _) if power == desired_power:
-                    print("Desired power is available.")
-                case (None, power) | (power, None) if power:
-                    print(f"Closest available power is {power}.")
-                case (lower, upper) if lower and upper:
-                    print(f"Two options {lower}, {upper} to propose to battery pool.")
-                case (None, None):
-                    print("No available power")
-            ```
+            proposing them, because the PowerManager will do this automatically.  This
+            method is provided for convenience, and for granular control when there are
+            two possible power values, both of which fall within the available bounds.
 
         Args:
             power: The power value to adjust.
@@ -150,13 +181,13 @@ def adjust_to_bounds(self, power: Power) -> tuple[Power | None, Power | None]:
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
 class Proposal:
-    """A proposal for a battery to be charged or discharged."""
+    """A proposal for a set of components to be charged or discharged."""
 
     source_id: str
     """The source ID of the actor sending the request."""
 
     preferred_power: Power | None
-    """The preferred power to be distributed to the batteries.
+    """The preferred power to be distributed to the components.
 
     If `None`, the preferred power of higher priority actors will get precedence.
     """
@@ -166,12 +197,12 @@ class Proposal:
 
     These bounds will apply to actors with a lower priority, and can be overridden by
     bounds from actors with a higher priority.  If None, the power bounds will be set to
-    the maximum power of the batteries in the pool.  This is currently an experimental
+    the maximum power of the components in the pool.  This is currently an experimental
     feature.
     """
 
-    battery_ids: frozenset[int]
-    """The battery IDs to distribute the power to."""
+    component_ids: frozenset[int]
+    """The component IDs to distribute the power to."""
 
     priority: int
     """The priority of the actor sending the proposal."""
@@ -207,23 +238,23 @@ class BaseAlgorithm(abc.ABC):
     @abc.abstractmethod
     def calculate_target_power(
         self,
-        battery_ids: frozenset[int],
+        component_ids: frozenset[int],
         proposal: Proposal | None,
-        system_bounds: PowerMetrics,
+        system_bounds: SystemBounds,
         must_return_power: bool = False,
     ) -> Power | None:
-        """Calculate and return the target power for the given batteries.
+        """Calculate and return the target power for the given components.
 
         Args:
-            battery_ids: The battery IDs to calculate the target power for.
+            component_ids: The component IDs to calculate the target power for.
             proposal: If given, the proposal to added to the bucket, before the target
                 power is calculated.
-            system_bounds: The system bounds for the batteries in the proposal.
+            system_bounds: The system bounds for the components in the proposal.
             must_return_power: If `True`, the algorithm must return a target power,
                 even if it hasn't changed since the last call.
 
         Returns:
-            The new target power for the batteries, or `None` if the target power
+            The new target power for the components, or `None` if the target power
                 didn't change.
         """
 
@@ -232,19 +263,19 @@ def calculate_target_power(
     @abc.abstractmethod
     def get_status(
         self,
-        battery_ids: frozenset[int],
+        component_ids: frozenset[int],
         priority: int,
-        system_bounds: PowerMetrics,
+        system_bounds: SystemBounds,
         distribution_result: power_distributing.Result | None,
-    ) -> Report:
-        """Get the bounds for a set of batteries, for the given priority.
+    ) -> _Report:
+        """Get the bounds for a set of components, for the given priority.
 
         Args:
-            battery_ids: The IDs of the batteries to get the bounds for.
+            component_ids: The IDs of the components to get the bounds for.
             priority: The priority of the actor for which the bounds are requested.
-            system_bounds: The system bounds for the batteries.
+            system_bounds: The system bounds for the components.
             distribution_result: The result of the last power distribution.
 
         Returns:
-            The bounds for the batteries.
+            The bounds for the components.
         """