From 698a95fecfad4a88381145f3c5f27bb72e6f779f Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Mon, 8 May 2023 17:29:55 +0200 Subject: [PATCH 1/4] Rename `BatteryPool.AggregateMethod` to `MetricAggregator` Signed-off-by: Sahas Subramanian --- src/frequenz/sdk/timeseries/battery_pool/_methods.py | 4 ++-- src/frequenz/sdk/timeseries/battery_pool/battery_pool.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/frequenz/sdk/timeseries/battery_pool/_methods.py b/src/frequenz/sdk/timeseries/battery_pool/_methods.py index ec1685a6b..ea831c08f 100644 --- a/src/frequenz/sdk/timeseries/battery_pool/_methods.py +++ b/src/frequenz/sdk/timeseries/battery_pool/_methods.py @@ -25,7 +25,7 @@ _logger = logging.getLogger(__name__) -class AggregateMethod(Generic[T], ABC): +class MetricAggregator(Generic[T], ABC): """Interface to control how the component data should be aggregated and send.""" @abstractmethod @@ -61,7 +61,7 @@ def name(cls) -> str: """ -class SendOnUpdate(AggregateMethod[T]): +class SendOnUpdate(MetricAggregator[T]): """Wait for the change of the components metrics and send updated result. This method will cache the component metrics. When any metric change it will diff --git a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py index 09cef0ca0..32bf9ac2d 100644 --- a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py +++ b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py @@ -25,7 +25,7 @@ FormulaGeneratorConfig, FormulaType, ) -from ._methods import AggregateMethod, SendOnUpdate +from ._methods import MetricAggregator, SendOnUpdate from ._metric_calculator import CapacityCalculator, PowerBoundsCalculator, SoCCalculator from ._result_types import CapacityMetrics, PowerMetrics, SoCMetrics @@ -85,7 +85,7 @@ def __init__( # pylint: disable=too-many-arguments ) self._min_update_interval = min_update_interval - self._active_methods: dict[str, AggregateMethod[Any]] = {} + self._active_methods: dict[str, MetricAggregator[Any]] = {} self._namespace: str = f"battery-pool-{self._batteries}-{uuid.uuid4()}" self._formula_pool: FormulaEnginePool = FormulaEnginePool( From 0cdd6b2b9abe3a700c4839c28e29ce4f1dc37045 Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Mon, 8 May 2023 17:30:47 +0200 Subject: [PATCH 2/4] Add default value for `maxsize` parameter of `new_receiver` method Signed-off-by: Sahas Subramanian --- src/frequenz/sdk/timeseries/battery_pool/_methods.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/frequenz/sdk/timeseries/battery_pool/_methods.py b/src/frequenz/sdk/timeseries/battery_pool/_methods.py index ea831c08f..3161e8251 100644 --- a/src/frequenz/sdk/timeseries/battery_pool/_methods.py +++ b/src/frequenz/sdk/timeseries/battery_pool/_methods.py @@ -13,7 +13,7 @@ from frequenz.channels import Broadcast, Receiver from ..._internal._asyncio import cancel_and_await -from ..._internal._constants import WAIT_FOR_COMPONENT_DATA_SEC +from ..._internal._constants import RECEIVER_MAX_SIZE, WAIT_FOR_COMPONENT_DATA_SEC from ._component_metric_fetcher import ( ComponentMetricFetcher, LatestBatteryMetricsFetcher, @@ -37,7 +37,9 @@ def update_working_batteries(self, new_working_batteries: set[int]) -> None: """ @abstractmethod - def new_receiver(self, maxsize: int | None) -> Receiver[T | None]: + def new_receiver( + self, maxsize: int | None = RECEIVER_MAX_SIZE + ) -> Receiver[T | None]: """Return new receiver for the aggregated metric results. Args: @@ -110,7 +112,9 @@ def name(cls) -> str: """ return "SendOnUpdate" - def new_receiver(self, maxsize: int | None) -> Receiver[T | None]: + def new_receiver( + self, maxsize: int | None = RECEIVER_MAX_SIZE + ) -> Receiver[T | None]: """Return new receiver for the aggregated metric results. Args: From b194674ba8e646c7d13421cab14240ca05b66195 Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Mon, 8 May 2023 17:35:12 +0200 Subject: [PATCH 3/4] Return `MetricAggregator` instances from `BatteryPool` metrics methods This makes their interface to be more consistent with the power and current methods. soc_recv = battery_pool.soc.new_receiver() instead of: soc_recv = battery_pool.soc() Signed-off-by: Sahas Subramanian --- examples/battery_pool.py | 6 +-- .../timeseries/battery_pool/battery_pool.py | 54 +++++++++---------- .../_battery_pool/test_battery_pool.py | 6 +-- 3 files changed, 31 insertions(+), 35 deletions(-) diff --git a/examples/battery_pool.py b/examples/battery_pool.py index d49be7e46..ada0e0b31 100644 --- a/examples/battery_pool.py +++ b/examples/battery_pool.py @@ -33,9 +33,9 @@ async def main() -> None: battery_pool = microgrid.battery_pool() receivers: Dict[str, Receiver[Any]] = { - "soc": await battery_pool.soc(maxsize=1), - "capacity": await battery_pool.capacity(maxsize=1), - "power_bounds": await battery_pool.power_bounds(maxsize=1), + "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), } merged_channel = MergeNamed[Any](**receivers) diff --git a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py index 32bf9ac2d..de54446b6 100644 --- a/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py +++ b/src/frequenz/sdk/timeseries/battery_pool/battery_pool.py @@ -14,7 +14,6 @@ from frequenz.channels import Receiver, Sender from ..._internal._asyncio import cancel_and_await -from ..._internal._constants import RECEIVER_MAX_SIZE from ...actor import ChannelRegistry, ComponentMetricRequest from ...actor.power_distributing._battery_pool_status import BatteryStatus from ...microgrid import connection_manager @@ -184,19 +183,19 @@ def consumption_power(self) -> FormulaEngine: assert isinstance(engine, FormulaEngine) return engine - async def soc( - self, maxsize: int | None = RECEIVER_MAX_SIZE - ) -> Receiver[SoCMetrics | None]: + @property + def soc(self) -> MetricAggregator[SoCMetrics]: """Get receiver to receive new soc metrics when they change. - Soc formulas are described in the receiver return type. - None will be send if there is no component to calculate metric. + Soc formulas are described in the receiver return type. None will be send if + there is no component to calculate metric. - Args: - maxsize: Maxsize of the receiver channel. + A receiver from the MetricAggregator can be obtained by calling the + `new_receiver` method. Returns: - Receiver for this metric. + A MetricAggregator that will calculate and stream the aggregate soc of + all batteries in the pool. """ method_name = SendOnUpdate.name() + "_" + SoCCalculator.name() @@ -208,22 +207,21 @@ async def soc( min_update_interval=self._min_update_interval, ) - running_method = self._active_methods[method_name] - return running_method.new_receiver(maxsize) + return self._active_methods[method_name] - async def capacity( - self, maxsize: int | None = RECEIVER_MAX_SIZE - ) -> Receiver[CapacityMetrics | None]: + @property + def capacity(self) -> MetricAggregator[CapacityMetrics]: """Get receiver to receive new capacity metrics when they change. - Capacity formulas are described in the receiver return type. - None will be send if there is no component to calculate metrics. + Capacity formulas are described in the receiver return type. None will be send + if there is no component to calculate metrics. - Args: - maxsize: Maxsize of the receiver channel. + A receiver from the MetricAggregator can be obtained by calling the + `new_receiver` method. Returns: - Receiver for this metric. + A MetricAggregator that will calculate and stream the capacity of all + batteries in the pool. """ method_name = SendOnUpdate.name() + "_" + CapacityCalculator.name() @@ -235,22 +233,21 @@ async def capacity( min_update_interval=self._min_update_interval, ) - running_method = self._active_methods[method_name] - return running_method.new_receiver(maxsize) + return self._active_methods[method_name] - async def power_bounds( - self, maxsize: int | None = RECEIVER_MAX_SIZE - ) -> Receiver[PowerMetrics | None]: + @property + def power_bounds(self) -> MetricAggregator[PowerMetrics]: """Get receiver to receive new power bounds when they change. Power bounds formulas are described in the receiver return type. None will be send if there is no component to calculate metrics. - Args: - maxsize: Maxsize of the receivers channel. + A receiver from the MetricAggregator can be obtained by calling the + `new_receiver` method. Returns: - Receiver for this metric. + A MetricAggregator that will calculate and stream the power bounds + of all batteries in the pool. """ method_name = SendOnUpdate.name() + "_" + PowerBoundsCalculator.name() @@ -262,8 +259,7 @@ async def power_bounds( min_update_interval=self._min_update_interval, ) - running_method = self._active_methods[method_name] - return running_method.new_receiver(maxsize) + return self._active_methods[method_name] async def stop(self) -> None: """Stop all pending async tasks.""" diff --git a/tests/timeseries/_battery_pool/test_battery_pool.py b/tests/timeseries/_battery_pool/test_battery_pool.py index 673540368..94b9d9556 100644 --- a/tests/timeseries/_battery_pool/test_battery_pool.py +++ b/tests/timeseries/_battery_pool/test_battery_pool.py @@ -500,7 +500,7 @@ async def run_capacity_test(setup_args: SetupArgs) -> None: sampling_rate=0.05, ) - capacity_receiver = await battery_pool.capacity(maxsize=50) + capacity_receiver = battery_pool.capacity.new_receiver(maxsize=50) # First metrics delivers slower because of the startup delay in the pool. msg = await asyncio.wait_for( @@ -633,7 +633,7 @@ async def run_soc_test(setup_args: SetupArgs) -> None: sampling_rate=0.05, ) - receiver = await battery_pool.soc(maxsize=50) + receiver = battery_pool.soc.new_receiver(maxsize=50) # First metrics delivers slower because of the startup delay in the pool. msg = await asyncio.wait_for( @@ -775,7 +775,7 @@ async def run_power_bounds_test( # pylint: disable=too-many-locals sampling_rate=0.1, ) - receiver = await battery_pool.power_bounds(maxsize=50) + receiver = battery_pool.power_bounds.new_receiver(maxsize=50) # First metrics delivers slower because of the startup delay in the pool. msg = await asyncio.wait_for( From 7e77a18bd75e88a8d70dfb51ac5c3edc27a55cfa Mon Sep 17 00:00:00 2001 From: Sahas Subramanian Date: Thu, 25 May 2023 12:14:30 +0200 Subject: [PATCH 4/4] Update RELEASE_NOTES.md about BatteryPool interface changes Signed-off-by: Sahas Subramanian --- RELEASE_NOTES.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 4fee8cb94..e40e73b2e 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -16,6 +16,14 @@ This release drops support for Python versions older than 3.11. * Now `frequenz.sdk.timeseries.Sample` uses a more sensible comparison. Before this release `Sample`s were compared only based on the `timestamp`. This was due to a limitation in Python versions earlier than 3.10. Now that the minimum supported version is 3.11 this hack is not needed anymore and `Sample`s are compared using both `timestamp` and `value` as most people probably expects. +* `BatteryPool` metric streaming interfaces have changed for `soc`, `capacity` and `power_bounds`: + + ```python + soc_rx = battery_pool.soc() # old + + soc_rx = battery_pool.soc.new_receiver() # new + ``` + ## New Features