Stop using channels APIs removed in v2

RELEASE_NOTES.md

@@ -6,7 +6,11 @@ This release raises the minimum `frequenz-microgrid-component-graph` version to
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+- 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
 

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",

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)

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=}")

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
 

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.

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(

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(

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(

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,

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,15 +49,15 @@ 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."""
 
         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:

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 (

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

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")

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")

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)

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.