diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 70f35876b..c0fdec885 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -6,7 +6,11 @@ This release raises the minimum `frequenz-microgrid-component-graph` version to ## Upgrading - +- The minimum `frequenz-channels` version is now 1.13.0. + +- The minimum `typing-extensions` version is now 4.16.0. `frequenz-channels` also requires `frequenz-core >= 1.6.0, < 2`. + +- The SDK no longer uses the channels APIs that will be removed in channels v2.0.0. The SDK's public API, timer behavior, and channel closing are unchanged. `Broadcast` and the SDK's `new_receiver(limit=...)` arguments stay as they are. ## New Features diff --git a/pyproject.toml b/pyproject.toml index 05c547225..b9d503897 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,10 +32,10 @@ dependencies = [ "frequenz-client-microgrid >= 0.18.3, < 0.19.0", "frequenz-microgrid-component-graph >= 0.6.1, < 0.7", "frequenz-client-common >= 0.3.6, < 0.5.0", - "frequenz-channels >= 1.6.1, < 2.0.0", + "frequenz-channels @ git+https://github.com/frequenz-floss/frequenz-channels-python@refs/pull/595/head", "frequenz-quantities[marshmallow] >= 1.0.0, < 2.0.0", "numpy >= 2.1.0, < 3", - "typing_extensions >= 4.14.1, < 5", + "typing_extensions >= 4.16.0, < 5", "marshmallow >= 3.19.0, < 5", "marshmallow_dataclass >= 8.7.1, < 9", "more-itertools >= 10.8.0, < 12", diff --git a/src/frequenz/sdk/_internal/_channels.py b/src/frequenz/sdk/_internal/_channels.py index 74b2f6929..4c75a216c 100644 --- a/src/frequenz/sdk/_internal/_channels.py +++ b/src/frequenz/sdk/_internal/_channels.py @@ -182,7 +182,7 @@ async def close_and_remove(self, key: str) -> None: entry = self._channels.pop(key, None) if entry is None: raise KeyError(f"No channel for key {key!r} exists.") - await entry.channel.close() + await entry.channel.aclose() @dataclasses.dataclass(frozen=True) diff --git a/src/frequenz/sdk/actor/__init__.py b/src/frequenz/sdk/actor/__init__.py index cb799b6d8..166c490aa 100644 --- a/src/frequenz/sdk/actor/__init__.py +++ b/src/frequenz/sdk/actor/__init__.py @@ -517,7 +517,7 @@ async def main() -> None: # (6)! await run(echo_actor) # (9)! - await echo_channel.close() # (16)! + await echo_channel.aclose() # (16)! async for message in echo_receiver: # (17)! print(f"Received {message=}") diff --git a/src/frequenz/sdk/config/_manager.py b/src/frequenz/sdk/config/_manager.py index f6ec08467..cde7432bd 100644 --- a/src/frequenz/sdk/config/_manager.py +++ b/src/frequenz/sdk/config/_manager.py @@ -12,8 +12,7 @@ from typing import Any, Final, Literal, TypeGuard, overload import marshmallow -from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError -from frequenz.channels.experimental import WithPrevious +from frequenz.channels import Broadcast, Receiver, ReceiverStoppedError, WithPrevious from marshmallow import Schema, ValidationError from typing_extensions import override diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py index c93511356..09656bab8 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_battery_status_tracker.py @@ -22,7 +22,7 @@ from datetime import datetime, timedelta, timezone from frequenz.channels import Receiver, Sender, select, selected_from -from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.channels.timer import MonotonicTimer, SkipMissedAndDrift from frequenz.client.common.microgrid.components import ComponentId from frequenz.client.microgrid.component import ComponentStateCode, Inverter from typing_extensions import override @@ -48,7 +48,7 @@ class _ComponentStreamStatus: component_id: ComponentId """Component id.""" - data_recv_timer: Timer + data_recv_timer: MonotonicTimer """Timer that is set when no component data has been received for some time.""" last_msg_timestamp: datetime = datetime.now(tz=timezone.utc) @@ -144,11 +144,11 @@ def __init__( # pylint: disable=too-many-arguments self._battery: _ComponentStreamStatus = _ComponentStreamStatus( component_id, - data_recv_timer=Timer(max_data_age, SkipMissedAndDrift()), + data_recv_timer=MonotonicTimer(max_data_age, SkipMissedAndDrift()), ) self._inverter: _ComponentStreamStatus = _ComponentStreamStatus( inverter_id, - data_recv_timer=Timer(max_data_age, SkipMissedAndDrift()), + data_recv_timer=MonotonicTimer(max_data_age, SkipMissedAndDrift()), ) # Select needs receivers that can be get in async way only. diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_ev_charger_status_tracker.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_ev_charger_status_tracker.py index 415478be3..d771c6760 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_ev_charger_status_tracker.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_ev_charger_status_tracker.py @@ -8,7 +8,7 @@ from datetime import datetime, timedelta, timezone from frequenz.channels import Receiver, Sender, select, selected_from -from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.channels.timer import MonotonicTimer, SkipMissedAndDrift from frequenz.client.common.microgrid.components import ComponentId from frequenz.client.microgrid.component import ComponentStateCode from typing_extensions import override @@ -149,7 +149,7 @@ async def _run(self) -> None: connection_manager.get().api_client, self._component_id ) set_power_result_rx = self._set_power_result_receiver - missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift()) + missing_data_timer = MonotonicTimer(self._max_data_age, SkipMissedAndDrift()) # Send initial status await self._status_sender.send( diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_pv_inverter_status_tracker.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_pv_inverter_status_tracker.py index c9a879620..283824254 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_pv_inverter_status_tracker.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_pv_inverter_status_tracker.py @@ -8,7 +8,7 @@ from datetime import datetime, timedelta, timezone from frequenz.channels import Receiver, Sender, select, selected_from -from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.channels.timer import MonotonicTimer, SkipMissedAndDrift from frequenz.client.common.microgrid.components import ComponentId from frequenz.client.microgrid.component import ComponentStateCode from typing_extensions import override @@ -149,7 +149,7 @@ async def _run(self) -> None: connection_manager.get().api_client, self._component_id ) set_power_result_rx = self._set_power_result_receiver - missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift()) + missing_data_timer = MonotonicTimer(self._max_data_age, SkipMissedAndDrift()) # Send initial status await self._status_sender.send( diff --git a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py index f12049480..eb80ae139 100644 --- a/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py +++ b/src/frequenz/sdk/microgrid/_power_distributing/_component_status/_steam_boiler_status_tracker.py @@ -8,7 +8,7 @@ from datetime import datetime, timedelta, timezone from frequenz.channels import Receiver, Sender, select, selected_from -from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.channels.timer import MonotonicTimer, SkipMissedAndDrift from frequenz.client.common.microgrid.components import ComponentId from frequenz.client.microgrid.component import ComponentStateCode from typing_extensions import override @@ -147,7 +147,7 @@ async def _run(self) -> None: connection_manager.get().api_client, self._component_id ) set_power_result_rx = self._set_power_result_receiver - missing_data_timer = Timer(self._max_data_age, SkipMissedAndDrift()) + missing_data_timer = MonotonicTimer(self._max_data_age, SkipMissedAndDrift()) # Send initial status await self._status_sender.send( diff --git a/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py b/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py index 7051e0275..d39e0e6f6 100644 --- a/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py +++ b/src/frequenz/sdk/microgrid/_power_managing/_power_managing_actor.py @@ -12,7 +12,7 @@ from typing import assert_never from frequenz.channels import Receiver, Sender, select, selected_from -from frequenz.channels.timer import SkipMissedAndDrift, Timer +from frequenz.channels.timer import MonotonicTimer, SkipMissedAndDrift from frequenz.client.common.microgrid.components import ComponentId from frequenz.client.microgrid.component import ( Battery, @@ -211,7 +211,9 @@ async def _send_updated_target_power( async def _run(self) -> None: """Run the power managing actor.""" last_result_partial_failure = False - drop_old_proposals_timer = Timer(timedelta(seconds=1.0), SkipMissedAndDrift()) + drop_old_proposals_timer = MonotonicTimer( + timedelta(seconds=1.0), SkipMissedAndDrift() + ) async for selected in select( self._proposals_receiver, self._bounds_subscription_receiver, diff --git a/src/frequenz/sdk/timeseries/_resampling/_resampler.py b/src/frequenz/sdk/timeseries/_resampling/_resampler.py index 7ac3b72a4..d0d150df6 100644 --- a/src/frequenz/sdk/timeseries/_resampling/_resampler.py +++ b/src/frequenz/sdk/timeseries/_resampling/_resampler.py @@ -14,7 +14,7 @@ from datetime import datetime, timedelta, timezone from typing import assert_never -from frequenz.channels.timer import Timer, TriggerAllMissed, _to_microseconds +from frequenz.channels.timer import MonotonicTimer, TriggerAllMissed, _to_microseconds from frequenz.quantities import Quantity from ..._internal._asyncio import cancel_and_await @@ -49,7 +49,7 @@ def __init__(self, config: ResamplerConfig) -> None: config: The configuration for the resampler. If a `ResamplerConfig2` is provided, the resampler will use a [`WallClockTimer`][frequenz.sdk.timeseries.WallClockTimer] instead of a - [`Timer`][frequenz.channels.timer.Timer]. + [`MonotonicTimer`][frequenz.channels.timer.MonotonicTimer]. """ self._config = config """The configuration for this resampler.""" @@ -57,7 +57,7 @@ def __init__(self, config: ResamplerConfig) -> None: self._resamplers: dict[Source, _StreamingHelper] = {} """A mapping between sources and the streaming helper handling that source.""" - self._timer: Timer | WallClockTimer + self._timer: MonotonicTimer | WallClockTimer """The timer used to trigger the resampling windows.""" if isinstance(config, ResamplerConfig2): @@ -78,10 +78,10 @@ def __init__(self, config: ResamplerConfig) -> None: the window end is deterministic. """ - self._timer = Timer(config.resampling_period, TriggerAllMissed()) + self._timer = MonotonicTimer(config.resampling_period, TriggerAllMissed()) """The timer used to trigger the resampling windows.""" - # Hack to align the timer, this should be implemented in the Timer class + # Align the timer; MonotonicTimer should align its own ticks. self._timer._next_tick_time = _to_microseconds( timedelta(seconds=asyncio.get_running_loop().time()) + config.resampling_period @@ -180,7 +180,7 @@ async def resample(self, *, one_shot: bool = False) -> None: case TickInfo(): # WallClockTimer next_tick_time = tick_info.expected_tick_time - case timedelta() as drift: # Timer (monotonic) + case timedelta() as drift: # MonotonicTimer next_tick_time = self._window_end if drift > tolerance: diff --git a/src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py b/src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py index 3de0d6db4..8d2881b53 100644 --- a/src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py +++ b/src/frequenz/sdk/timeseries/formulas/_formula_3_phase_evaluator.py @@ -76,7 +76,7 @@ async def _run(self) -> None: _logger.debug( "input streams closed; stopping three-phase formula evaluator." ) - await self._output_channel.close() + await self._output_channel.aclose() return if ( @@ -87,7 +87,7 @@ async def _run(self) -> None: _logger.debug( "One of the three phase samples is None, stopping the evaluator." ) - await self._output_channel.close() + await self._output_channel.aclose() return if not ( diff --git a/src/frequenz/sdk/timeseries/formulas/_formula_evaluator.py b/src/frequenz/sdk/timeseries/formulas/_formula_evaluator.py index 404b9fdbb..440c84afd 100644 --- a/src/frequenz/sdk/timeseries/formulas/_formula_evaluator.py +++ b/src/frequenz/sdk/timeseries/formulas/_formula_evaluator.py @@ -63,9 +63,9 @@ async def _run(self) -> None: "No more input samples available; stopping formula evaluator. (%s)", self._root, ) - await self._output_channel.close() + await self._output_channel.aclose() return except Exception: # pylint: disable=broad-except _logger.exception("Error evaluating formula %s", self._root) - await self._output_channel.close() + await self._output_channel.aclose() return diff --git a/tests/timeseries/_pv_pool/test_pv_pool.py b/tests/timeseries/_pv_pool/test_pv_pool.py index 5139a73e0..4beff9a12 100644 --- a/tests/timeseries/_pv_pool/test_pv_pool.py +++ b/tests/timeseries/_pv_pool/test_pv_pool.py @@ -161,7 +161,7 @@ async def test_bounds_tracker_tracks_unreachable_power_again_on_restart( await feeder.wait_for_formulas(1) # An error ends the run, the way `run_forever` sees it. - await data_channel.close() + await data_channel.aclose() with pytest.raises(ReceiverStoppedError): await run data_channel = Broadcast[InverterData](name="data") diff --git a/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py index b0e11b6f5..e8aff9807 100644 --- a/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py +++ b/tests/timeseries/_steam_boiler_pool/test_steam_boiler_pool.py @@ -228,7 +228,7 @@ async def test_bounds_tracker_tracks_unreachable_power_again_on_restart( await feeder.wait_for_formulas(1) # An error ends the run, the way `run_forever` sees it. - await data_channel.close() + await data_channel.aclose() with pytest.raises(ReceiverStoppedError): await run data_channel = Broadcast[SteamBoilerData](name="data") diff --git a/tests/timeseries/test_resampling.py b/tests/timeseries/test_resampling.py index 6d94a68bc..bbcecb887 100644 --- a/tests/timeseries/test_resampling.py +++ b/tests/timeseries/test_resampling.py @@ -53,7 +53,7 @@ async def source_chan() -> AsyncIterator[Broadcast[Sample[Quantity]]]: """Create a broadcast channel of samples.""" chan = Broadcast[Sample[Quantity]](name="test") yield chan - await chan.close() + await chan.aclose() def as_float_tuple(sample: Sample[Quantity]) -> tuple[datetime, float]: @@ -1512,7 +1512,7 @@ async def test_receiving_stopped_resampling_error( resampling_fun_mock.reset_mock() # Close channel, try to resample again - await source_chan.close() + await source_chan.aclose() with pytest.raises(SenderError): await source_sender.send(sample0s) await _advance_time(fake_time, resampling_period_s) diff --git a/tests/utils/mock_microgrid_client.py b/tests/utils/mock_microgrid_client.py index a41ec24e4..c0d26c53c 100644 --- a/tests/utils/mock_microgrid_client.py +++ b/tests/utils/mock_microgrid_client.py @@ -143,7 +143,7 @@ async def close_channels(self, cid: ComponentId) -> None: """ for key, channel in self._component_data_channels.items(): if key.component_id == cid: - await channel.close() + await channel.aclose() def _create_mock_api(self) -> MagicMock: """Create mock of MicrogridApiClient.