Update channel dependancy to 0.16.0

Signed-off-by: Jack <[email protected]>

Author: jack-herrmann

RELEASE_NOTES.md

@@ -8,6 +8,8 @@
 
 <!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with --> 
 
+- `Channels` has been upgraded to version 0.16.0, for information on how to upgrade visit https://github.com/frequenz-floss/frequenz-channels-python/releases/tag/v0.16.0
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

pyproject.toml

@@ -30,7 +30,7 @@ dependencies = [
   # Make sure to update the mkdocs.yml file when
   # changing the version
   # (plugins.mkdocstrings.handlers.python.import)
-  "frequenz-channels >= 0.14.0, < 0.15.0",
+  "frequenz-channels >= 0.16.0, < 0.17.0",
   "google-api-python-client >= 2.71, < 3",
   "grpcio >= 1.54.2, < 2",
   "grpcio-tools >= 1.54.2, < 2",

src/frequenz/sdk/actor/_config_managing.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import tomllib
+from collections import abc
 from typing import Any, Dict, Optional, Set
 
 from frequenz.channels import Sender
@@ -31,7 +32,7 @@ def __init__(
         self,
         conf_file: str,
         output: Sender[Config],
-        event_types: Optional[Set[FileWatcher.EventType]] = None,
+        event_types: abc.Set[FileWatcher.EventType] = frozenset(FileWatcher.EventType),
     ) -> None:
         """Read config variables from the file.
 
@@ -79,8 +80,8 @@ async def run(self) -> None:
         """
         await self.send_config()
 
-        async for path in self._file_watcher:
-            if str(path) == self._conf_file:
+        async for event in self._file_watcher:
+            if str(event.path) == self._conf_file:
                 _logger.info(
                     "Update configs, because file %s was modified.",
                     self._conf_file,

src/frequenz/sdk/actor/_decorator.py

@@ -82,7 +82,7 @@ def actor(cls: Callable[_P, _R]) -> Type[_R]:
     Example (one actor receiving from two receivers):
     ```python
     from frequenz.channels import Broadcast, Receiver, Sender
-    from frequenz.channels.util import Select
+    from frequenz.channels.util import select, selected_from
     @actor
     class EchoActor:
         def __init__(
@@ -99,13 +99,9 @@ def __init__(
             self._output = output
 
         async def run(self) -> None:
-            select = Select(channel_1=self._recv1, channel_2=self._recv2)
-            while await select.ready():
-                if msg := select.channel_1:
-                    await self._output.send(msg.inner)
-                elif msg := select.channel_2:
-                    await self._output.send(msg.inner)
-
+            async for selected in select(self._recv1, self._recv2):
+                if selected_from(selected, self._recv1):
+                    await self._output.send(selected.value)
 
     input_chan_1: Broadcast[bool] = Broadcast("input_chan_1")
     input_chan_2: Broadcast[bool] = Broadcast("input_chan_2")

src/frequenz/sdk/actor/power_distributing/_battery_status.py

@@ -17,7 +17,7 @@
 from frequenz.api.microgrid.common_pb2 import ErrorLevel
 from frequenz.api.microgrid.inverter_pb2 import ComponentState as InverterComponentState
 from frequenz.channels import Receiver, Sender
-from frequenz.channels.util import Select, Timer
+from frequenz.channels.util import Timer, select, selected_from
 
 from frequenz.sdk._internal._asyncio import cancel_and_await
 from frequenz.sdk.microgrid import connection_manager
@@ -197,14 +197,15 @@ def __init__(  # pylint: disable=too-many-arguments
             raise RuntimeError(f"Can't find inverter adjacent to battery: {battery_id}")
 
         self._battery: _ComponentStreamStatus = _ComponentStreamStatus(
-            battery_id, data_recv_timer=Timer(max_data_age_sec)
+            battery_id,
+            data_recv_timer=Timer.timeout(timedelta(max_data_age_sec)),
         )
         self._inverter: _ComponentStreamStatus = _ComponentStreamStatus(
-            inverter_id, data_recv_timer=Timer(max_data_age_sec)
+            inverter_id,
+            data_recv_timer=Timer.timeout(timedelta(max_data_age_sec)),
         )
 
         # Select needs receivers that can be get in async way only.
-        self._select: Select | None = None
 
         self._task: asyncio.Task[None] = asyncio.create_task(
             self._run(status_sender, set_power_result_receiver)
@@ -223,8 +224,70 @@ async def stop(self) -> None:
         """Stop tracking battery status."""
         await cancel_and_await(self._task)
 
-        if self._select is not None:
-            await self._select.stop()
+    def _handle_status_battery(self, bat_data: BatteryData) -> None:
+        self._battery.last_msg_correct = (
+            self._is_message_reliable(bat_data)
+            and self._is_battery_state_correct(bat_data)
+            and self._no_critical_error(bat_data)
+            and self._is_capacity_present(bat_data)
+        )
+        self._battery.last_msg_timestamp = bat_data.timestamp
+        self._battery.data_recv_timer.reset()
+
+    def _handle_status_inverter(self, inv_data: InverterData) -> None:
+        self._inverter.last_msg_correct = (
+            self._is_message_reliable(inv_data)
+            and self._is_inverter_state_correct(inv_data)
+            and self._no_critical_error(inv_data)
+        )
+        self._inverter.last_msg_timestamp = inv_data.timestamp
+        self._inverter.data_recv_timer.reset()
+
+    def _handle_status_set_power_result(self, result: SetPowerResult) -> None:
+        if self.battery_id in result.succeed:
+            self._blocking_status.unblock()
+
+        elif (
+            self.battery_id in result.failed and self._last_status != Status.NOT_WORKING
+        ):
+            duration = self._blocking_status.block()
+
+            if duration > 0:
+                _logger.warning(
+                    "battery %d failed last response. block it for %f sec",
+                    self.battery_id,
+                    duration,
+                )
+
+    def _handle_status_battery_timer(self) -> None:
+        if self._battery.last_msg_correct:
+            self._battery.last_msg_correct = False
+            _logger.warning(
+                "Battery %d stopped sending data, last timestamp: %s",
+                self._battery.component_id,
+                self._battery.last_msg_timestamp,
+            )
+
+    def _handle_status_inverter_timer(self) -> None:
+        if self._inverter.last_msg_correct:
+            self._inverter.last_msg_correct = False
+            _logger.warning(
+                "Inverter %d stopped sending data, last timestamp: %s",
+                self._inverter.component_id,
+                self._inverter.last_msg_timestamp,
+            )
+
+    def _get_new_status_if_changed(self) -> Optional[Status]:
+        current_status = self._get_current_status()
+        if self._last_status != current_status:
+            self._last_status = current_status
+            _logger.info(
+                "battery %d changed status %s",
+                self.battery_id,
+                str(self._last_status),
+            )
+            return current_status
+        return None
 
     async def _run(
         self,
@@ -245,93 +308,48 @@ async def _run(
         battery_receiver = await api_client.battery_data(self._battery.component_id)
         inverter_receiver = await api_client.inverter_data(self._inverter.component_id)
 
-        self._select = Select(
-            battery=battery_receiver,
-            battery_timer=self._battery.data_recv_timer,
-            inverter_timer=self._inverter.data_recv_timer,
-            inverter=inverter_receiver,
-            set_power_result=set_power_result_receiver,
-        )
+        battery = battery_receiver
+        battery_timer = self._battery.data_recv_timer
+        inverter_timer = self._inverter.data_recv_timer
+        inverter = inverter_receiver
+        set_power_result = set_power_result_receiver
 
         while True:
             try:
-                while await self._select.ready():
-                    new_status = self._update_status(self._select)
+                async for selected in select(
+                    battery,
+                    battery_timer,
+                    inverter_timer,
+                    inverter,
+                    set_power_result,
+                ):
+                    new_status = None
 
-                    if new_status is not None:
-                        await status_sender.send(new_status)
+                    if selected_from(selected, battery):
+                        self._handle_status_battery(selected.value)
 
-            except Exception as err:  # pylint: disable=broad-except
-                _logger.exception("BatteryStatusTracker crashed with error: %s", err)
+                    elif selected_from(selected, inverter):
+                        self._handle_status_inverter(selected.value)
 
-    def _update_status(self, select: Select) -> Optional[Status]:
-        if msg := select.battery:
-            self._battery.last_msg_correct = (
-                self._is_message_reliable(msg.inner)
-                and self._is_battery_state_correct(msg.inner)
-                and self._no_critical_error(msg.inner)
-                and self._is_capacity_present(msg.inner)
-            )
-            self._battery.last_msg_timestamp = msg.inner.timestamp
-            self._battery.data_recv_timer.reset()
-
-        elif msg := select.inverter:
-            self._inverter.last_msg_correct = (
-                self._is_message_reliable(msg.inner)
-                and self._is_inverter_state_correct(msg.inner)
-                and self._no_critical_error(msg.inner)
-            )
-            self._inverter.last_msg_timestamp = msg.inner.timestamp
-            self._inverter.data_recv_timer.reset()
-
-        elif msg := select.set_power_result:
-            result: SetPowerResult = msg.inner
-            if self.battery_id in result.succeed:
-                self._blocking_status.unblock()
-
-            elif (
-                self.battery_id in result.failed
-                and self._last_status != Status.NOT_WORKING
-            ):
-                duration = self._blocking_status.block()
-
-                if duration > 0:
-                    _logger.warning(
-                        "battery %d failed last response. block it for %f sec",
-                        self.battery_id,
-                        duration,
-                    )
-        elif msg := select.battery_timer:
-            if self._battery.last_msg_correct:
-                self._battery.last_msg_correct = False
-                _logger.warning(
-                    "Battery %d stopped sending data, last timestamp: %s",
-                    self._battery.component_id,
-                    self._battery.last_msg_timestamp,
-                )
-        elif msg := select.inverter_timer:
-            if self._inverter.last_msg_correct:
-                self._inverter.last_msg_correct = False
-                _logger.warning(
-                    "Inverter %d stopped sending data, last timestamp: %s",
-                    self._inverter.component_id,
-                    self._inverter.last_msg_timestamp,
-                )
+                    elif selected_from(selected, set_power_result):
+                        self._handle_status_set_power_result(selected.value)
 
-        else:
-            _logger.error("Unknown message returned from select")
+                    elif selected_from(selected, battery_timer):
+                        self._handle_status_battery_timer()
 
-        current_status = self._get_current_status()
-        if self._last_status != current_status:
-            self._last_status = current_status
-            _logger.info(
-                "battery %d changed status %s",
-                self.battery_id,
-                str(self._last_status),
-            )
-            return current_status
+                    elif selected_from(selected, inverter_timer):
+                        self._handle_status_inverter_timer()
 
-        return None
+                    else:
+                        _logger.error("Unknown message returned from select")
+
+                    new_status = self._get_new_status_if_changed()
+
+                    if new_status is not None:
+                        await status_sender.send(new_status)
+
+            except Exception as err:  # pylint: disable=broad-except
+                _logger.exception("BatteryStatusTracker crashed with error: %s", err)
 
     def _get_current_status(self) -> Status:
         """Get current battery status.

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

@@ -10,7 +10,7 @@
 from typing import Dict
 
 from frequenz.channels import Broadcast, Sender
-from frequenz.channels.util import Select, Timer
+from frequenz.channels.util import Timer, select, selected_from
 
 from ..._internal._asyncio import cancel_and_await
 from ...microgrid import connection_manager
@@ -95,17 +95,16 @@ async def _run(self) -> None:
         ).into_peekable()
         latest_bound: Dict[int, ComponentCurrentLimit] = {}
 
-        select = Select(
-            bound_chan=self._bounds_rx,
-            timer=Timer(self._repeat_interval.total_seconds()),
-        )
-        while await select.ready():
+        bound_chan = self._bounds_rx
+        timer = Timer.timeout(timedelta(self._repeat_interval.total_seconds()))
+
+        async for selected in select(bound_chan, timer):
             meter = meter_data.peek()
             if meter is None:
                 raise ValueError("Meter channel closed.")
 
-            if msg := select.bound_chan:
-                bound: ComponentCurrentLimit = msg.inner
+            if selected_from(selected, bound_chan):
+                bound: ComponentCurrentLimit = selected.value
                 if (
                     bound.component_id in latest_bound
                     and latest_bound[bound.component_id] == bound
@@ -119,7 +118,7 @@ async def _run(self) -> None:
                     0,
                     bound.max_amps * min_voltage * self._NUM_PHASES,
                 )
-            elif msg := select.timer:
+            elif selected_from(selected, timer):
                 for bound in latest_bound.values():
                     min_voltage = min(meter.voltage_per_phase)
                     logging.debug("resending bounds: %s", bound)