Improve consistency between *Pools (#894)

Also does some cleanup on the PowerDistributor tests.

Author: shsms

RELEASE_NOTES.md

@@ -10,6 +10,10 @@
 
 - The minimum required `frequenz-channels` version is now [`v1.0.0-rc1`](https://github.com/frequenz-floss/frequenz-channels-python/releases/tag/v1.0.0-rc.1).
 
+- The set of battery IDs managed by a battery pool are now available through `BatteryPool.component_ids`, and no longer through `BatteryPool.battery_ids`.  This is done to have a consistent interface with other `*Pool`s.
+
+- The `maxsize` parameter in calls to `BatteryPool.{soc/capacity/temperature}.new_receiver()` methods have now been renamed to `limit`, to be consistent with the channels repository.
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

benchmarks/timeseries/benchmark_ringbuffer.py

@@ -174,14 +174,14 @@ def main() -> None:
         "Time to fill 29 days with data:\n\t"
         + f"Array: {array_times['fill']} seconds\n\t"
         + f"List:  {list_times['fill']} seconds\n\t"
-        + f"Diff:  {array_times['fill'] -  list_times['fill']}"
+        + f"Diff:  {array_times['fill'] - list_times['fill']}"
     )
 
     print(
         "Day-Slices into 29 days with data:\n\t"
         + f"Array: {array_times['test']/num_runs} seconds\n\t"
         + f"List:  {list_times['test']/num_runs} seconds\n\t"
-        + f"Diff:  {array_times['test']/num_runs -  list_times['test']/num_runs}"
+        + f"Diff:  {array_times['test']/num_runs - list_times['test']/num_runs}"
     )
 
     print(f"       {''.join(['='] * (num_runs + 1))}")
@@ -195,15 +195,15 @@ def main() -> None:
         "Avg of windows of 29 days and running average & mean on every day:\n\t"
         + f"Array: {slicing_array_times['avg']/num_runs} seconds\n\t"
         + f"List:  {slicing_list_times['avg']/num_runs} seconds\n\t"
-        + f"Diff:  {slicing_array_times['avg']/num_runs -  slicing_list_times['avg']/num_runs}"
+        + f"Diff:  {slicing_array_times['avg']/num_runs - slicing_list_times['avg']/num_runs}"
     )
 
     print(
         "Median of windows of 29 days and running average & mean on every day:\n\t"
         + f"Array: {slicing_array_times['median']/num_runs} seconds\n\t"
         + f"List:  {slicing_list_times['median']/num_runs} seconds\n\t"
         + "Diff:  "
-        + f"{slicing_array_times['median']/num_runs -  slicing_list_times['median']/num_runs}"
+        + f"{slicing_array_times['median']/num_runs - slicing_list_times['median']/num_runs}"
     )
 
 

examples/battery_pool.py

@@ -30,10 +30,10 @@ async def main() -> None:
 
     battery_pool = microgrid.battery_pool()
     receivers = [
-        battery_pool.soc.new_receiver(maxsize=1),
-        battery_pool.capacity.new_receiver(maxsize=1),
+        battery_pool.soc.new_receiver(limit=1),
+        battery_pool.capacity.new_receiver(limit=1),
         # pylint: disable=protected-access
-        battery_pool._system_power_bounds.new_receiver(maxsize=1),
+        battery_pool._system_power_bounds.new_receiver(limit=1),
         # pylint: enable=protected-access
     ]
 

src/frequenz/sdk/_internal/_channels.py

@@ -16,11 +16,11 @@ class ReceiverFetcher(typing.Generic[T_co], typing.Protocol):
     """An interface that just exposes a `new_receiver` method."""
 
     @abc.abstractmethod
-    def new_receiver(self, *, maxsize: int = 50) -> Receiver[T_co]:
+    def new_receiver(self, *, limit: int = 50) -> Receiver[T_co]:
         """Get a receiver from the channel.
 
         Args:
-            maxsize: The maximum size of the receiver.
+            limit: The maximum size of the receiver.
 
         Returns:
             A receiver instance.

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -154,7 +154,7 @@ def producer(self) -> Producer:
 
     def ev_charger_pool(
         self,
-        ev_charger_ids: set[int] | None = None,
+        ev_charger_ids: abc.Set[int] | None = None,
     ) -> EVChargerPool:
         """Return the corresponding EVChargerPool instance for the given ids.
 
@@ -350,7 +350,7 @@ def producer() -> Producer:
     return _get().producer()
 
 
-def ev_charger_pool(ev_charger_ids: set[int] | None = None) -> EVChargerPool:
+def ev_charger_pool(ev_charger_ids: abc.Set[int] | None = None) -> EVChargerPool:
     """Return the corresponding EVChargerPool instance for the given ids.
 
     If an EVChargerPool instance for the given ids doesn't exist, a new one is

src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

@@ -230,7 +230,7 @@ async def propose_discharge(
         )
 
     @property
-    def battery_ids(self) -> abc.Set[int]:
+    def component_ids(self) -> abc.Set[int]:
         """Return ids of the batteries in the pool.
 
         Returns:

src/frequenz/sdk/timeseries/battery_pool/_methods.py

@@ -40,11 +40,11 @@ def update_working_batteries(self, new_working_batteries: set[int]) -> None:
         """
 
     @abstractmethod
-    def new_receiver(self, maxsize: int | None = RECEIVER_MAX_SIZE) -> Receiver[T]:
+    def new_receiver(self, limit: int | None = RECEIVER_MAX_SIZE) -> Receiver[T]:
         """Return new receiver for the aggregated metric results.
 
         Args:
-            maxsize: Buffer size of the receiver
+            limit: Buffer size of the receiver
 
         Returns:
             Receiver for the metric results.
@@ -119,18 +119,18 @@ def name(cls) -> str:
         """
         return "SendOnUpdate"
 
-    def new_receiver(self, maxsize: int | None = RECEIVER_MAX_SIZE) -> Receiver[T]:
+    def new_receiver(self, limit: int | None = RECEIVER_MAX_SIZE) -> Receiver[T]:
         """Return new receiver for the aggregated metric results.
 
         Args:
-            maxsize: Buffer size of the receiver
+            limit: Buffer size of the receiver
 
         Returns:
             Receiver for the metric results.
         """
-        if maxsize is None:
+        if limit is None:
             return self._result_channel.new_receiver()
-        return self._result_channel.new_receiver(limit=maxsize)
+        return self._result_channel.new_receiver(limit=limit)
 
     def update_working_batteries(self, new_working_batteries: set[int]) -> None:
         """Update set of the working batteries.

src/frequenz/sdk/timeseries/ev_charger_pool/_ev_charger_pool.py

@@ -77,7 +77,7 @@ def __init__(
         self,
         channel_registry: ChannelRegistry,
         resampler_subscription_sender: Sender[ComponentMetricRequest],
-        component_ids: set[int] | None = None,
+        component_ids: abc.Set[int] | None = None,
         repeat_interval: timedelta = timedelta(seconds=3.0),
     ) -> None:
         """Create an `EVChargerPool` instance.
@@ -103,7 +103,7 @@ def __init__(
         self._resampler_subscription_sender: Sender[ComponentMetricRequest] = (
             resampler_subscription_sender
         )
-        self._component_ids: set[int] = set()
+        self._component_ids: abc.Set[int] = set()
         if component_ids is not None:
             self._component_ids = component_ids
         else:

src/frequenz/sdk/timeseries/ev_charger_pool/_state_tracker.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections import abc
 from enum import Enum
 
 from frequenz.channels import Merger, Receiver, merge
@@ -78,7 +79,7 @@ def is_ev_connected(self) -> bool:
 class StateTracker:
     """A class for keeping track of the states of all EV Chargers in a pool."""
 
-    def __init__(self, component_ids: set[int]) -> None:
+    def __init__(self, component_ids: abc.Set[int]) -> None:
         """Create a `_StateTracker` instance.
 
         Args: