Upgrade channel version to 1.0.0-rc1 (#893)

This version of `frequenz-channels` includes a number of improvements
and a fix for a timer bug:
https://github.com/frequenz-floss/frequenz-channels-python/releases/tag/v1.0.0-rc.1

Author: shsms

RELEASE_NOTES.md

@@ -8,6 +8,8 @@
 
 - The SDK is now using the microgrid API client from [`frequenz-client-microgrid`](https://github.com/frequenz-floss/frequenz-client-microgrid-python/). You should update your code if you are using the microgrid API client directly.
 
+- 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).
+
 ## New Features
 
 <!-- Here goes the main new features and examples or instructions on how to use them -->

benchmarks/power_distribution/power_distributor.py

@@ -111,9 +111,9 @@ async def run_test(  # pylint: disable=too-many-locals
     """
     start = timeit.default_timer()
 
-    power_request_channel = Broadcast[Request]("power-request")
-    battery_status_channel = Broadcast[ComponentPoolStatus]("battery-status")
-    power_result_channel = Broadcast[Result]("power-result")
+    power_request_channel = Broadcast[Request](name="power-request")
+    battery_status_channel = Broadcast[ComponentPoolStatus](name="battery-status")
+    power_result_channel = Broadcast[Result](name="power-result")
     async with PowerDistributingActor(
         requests_receiver=power_request_channel.new_receiver(),
         results_sender=power_result_channel.new_sender(),

benchmarks/timeseries/benchmark_datasourcing.py

@@ -77,12 +77,13 @@ async def benchmark_data_sourcing(
     mock_grid.start_mock_client(enable_mock_client)
 
     request_channel = Broadcast[ComponentMetricRequest](
-        "DataSourcingActor Request Channel"
+        name="DataSourcingActor Request Channel"
     )
 
     channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
     request_receiver = request_channel.new_receiver(
-        "datasourcing-benchmark", maxsize=(num_ev_chargers * len(COMPONENT_METRIC_IDS))
+        name="datasourcing-benchmark",
+        limit=(num_ev_chargers * len(COMPONENT_METRIC_IDS)),
     )
     request_sender = request_channel.new_sender()
 

benchmarks/timeseries/periodic_feature_extractor.py

@@ -33,7 +33,7 @@ async def init_feature_extractor(
 ) -> collections.abc.AsyncIterator[PeriodicFeatureExtractor]:
     """Initialize the PeriodicFeatureExtractor class."""
     # We only need the moving window to initialize the PeriodicFeatureExtractor class.
-    lm_chan = Broadcast[Sample[Quantity]]("lm_net_power")
+    lm_chan = Broadcast[Sample[Quantity]](name="lm_net_power")
     async with MovingWindow(
         timedelta(seconds=1), lm_chan.new_receiver(), timedelta(seconds=1)
     ) as moving_window:

examples/battery_pool.py

@@ -7,10 +7,8 @@
 import asyncio
 import logging
 from datetime import timedelta
-from typing import Any
 
-from frequenz.channels import Receiver
-from frequenz.channels.util import MergeNamed
+from frequenz.channels import merge
 
 from frequenz.sdk import microgrid
 from frequenz.sdk.actor import ResamplerConfig
@@ -31,17 +29,16 @@ async def main() -> None:
     )
 
     battery_pool = microgrid.battery_pool()
-    receivers: dict[str, Receiver[Any]] = {
-        "soc": battery_pool.soc.new_receiver(maxsize=1),
-        "capacity": battery_pool.capacity.new_receiver(maxsize=1),
+    receivers = [
+        battery_pool.soc.new_receiver(maxsize=1),
+        battery_pool.capacity.new_receiver(maxsize=1),
         # pylint: disable=protected-access
-        "power_bounds": battery_pool._system_power_bounds.new_receiver(maxsize=1),
+        battery_pool._system_power_bounds.new_receiver(maxsize=1),
         # pylint: enable=protected-access
-    }
+    ]
 
-    merged_channel = MergeNamed[Any](**receivers)
-    async for metric_name, metric in merged_channel:
-        print(f"Received new {metric_name}: {metric}")
+    async for metric in merge(*receivers):
+        print(f"Received new metric: {metric}")
 
 
 asyncio.run(main())

pyproject.toml

@@ -30,8 +30,8 @@ dependencies = [
   # Make sure to update the mkdocs.yml file when
   # changing the version
   # (plugins.mkdocstrings.handlers.python.import)
-  "frequenz-channels == 1.0.0b2",
-  "frequenz-client-microgrid >= 0.1.2, < 0.2.0",
+  "frequenz-channels == 1.0.0-rc1",
+  "frequenz-client-microgrid >= 0.2.0, < 0.3.0",
   "google-api-python-client >= 2.71, < 3",
   "grpcio >= 1.54.2, < 2",
   "grpcio-tools >= 1.54.2, < 2",

src/frequenz/sdk/_internal/_channels.py

@@ -9,14 +9,14 @@
 
 from frequenz.channels import Receiver
 
-T = typing.TypeVar("T")
+T_co = typing.TypeVar("T_co", covariant=True)
 
 
-class ReceiverFetcher(typing.Generic[T], typing.Protocol):
+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]:
+    def new_receiver(self, *, maxsize: int = 50) -> Receiver[T_co]:
         """Get a receiver from the channel.
 
         Args:
@@ -31,20 +31,20 @@ class _Sentinel:
     """A sentinel to denote that no value has been received yet."""
 
 
-class LatestValueCache(typing.Generic[T]):
+class LatestValueCache(typing.Generic[T_co]):
     """A cache that stores the latest value in a receiver."""
 
-    def __init__(self, receiver: Receiver[T]) -> None:
+    def __init__(self, receiver: Receiver[T_co]) -> None:
         """Create a new cache.
 
         Args:
             receiver: The receiver to cache.
         """
         self._receiver = receiver
-        self._latest_value: T | _Sentinel = _Sentinel()
+        self._latest_value: T_co | _Sentinel = _Sentinel()
         self._task = asyncio.create_task(self._run())
 
-    def get(self) -> T:
+    def get(self) -> T_co:
         """Return the latest value that has been received.
 
         This raises a `ValueError` if no value has been received yet. Use `has_value` to

src/frequenz/sdk/actor/_channel_registry.py

@@ -98,7 +98,9 @@ def get_or_create(self, message_type: type[_T], key: str) -> Broadcast[_T]:
                     message_type,
                     "".join(traceback.format_stack(limit=10)[:9]),
                 )
-            self._channels[key] = _Entry(message_type, Broadcast(f"{self._name}-{key}"))
+            self._channels[key] = _Entry(
+                message_type, Broadcast(name=f"{self._name}-{key}")
+            )
 
         entry = self._channels[key]
         if entry.message_type is not message_type:

src/frequenz/sdk/actor/_config_managing.py

@@ -10,7 +10,7 @@
 from typing import Any, assert_never
 
 from frequenz.channels import Sender
-from frequenz.channels.util import FileWatcher
+from frequenz.channels.file_watcher import EventType, FileWatcher
 
 from ..actor._actor import Actor
 from ..config import Config
@@ -34,7 +34,7 @@ def __init__(
         self,
         config_path: pathlib.Path | str,
         output: Sender[Config],
-        event_types: abc.Set[FileWatcher.EventType] = frozenset(FileWatcher.EventType),
+        event_types: abc.Set[EventType] = frozenset(EventType),
         *,
         name: str | None = None,
     ) -> None:
@@ -99,21 +99,21 @@ async def _run(self) -> None:
                 continue
 
             match event.type:
-                case FileWatcher.EventType.CREATE:
+                case EventType.CREATE:
                     _logger.info(
                         "%s: The configuration file %s was created, sending new config...",
                         self,
                         self._config_path,
                     )
                     await self.send_config()
-                case FileWatcher.EventType.MODIFY:
+                case EventType.MODIFY:
                     _logger.info(
                         "%s: The configuration file %s was modified, sending update...",
                         self,
                         self._config_path,
                     )
                     await self.send_config()
-                case FileWatcher.EventType.DELETE:
+                case EventType.DELETE:
                     _logger.info(
                         "%s: The configuration file %s was deleted, ignoring...",
                         self,

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -10,8 +10,8 @@
 import typing
 from datetime import datetime, timedelta, timezone
 
-from frequenz.channels import Receiver, Sender
-from frequenz.channels.util import SkipMissedAndDrift, Timer, select, selected_from
+from frequenz.channels import Receiver, Sender, select, selected_from
+from frequenz.channels.timer import SkipMissedAndDrift, Timer
 from frequenz.client.microgrid import ComponentCategory
 from typing_extensions import override
 
@@ -195,7 +195,7 @@ async def _run(self) -> None:
             drop_old_proposals_timer,
         ):
             if selected_from(selected, self._proposals_receiver):
-                proposal = selected.value
+                proposal = selected.message
                 if proposal.component_ids not in self._bound_tracker_tasks:
                     self._add_bounds_tracker(proposal.component_ids)
 
@@ -214,7 +214,7 @@ async def _run(self) -> None:
                 await self._send_reports(proposal.component_ids)
 
             elif selected_from(selected, self._bounds_subscription_receiver):
-                sub = selected.value
+                sub = selected.message
                 component_ids = sub.component_ids
                 priority = sub.priority
 
@@ -239,7 +239,7 @@ async def _run(self) -> None:
                     power_distributing,
                 )
 
-                result = selected.value
+                result = selected.message
                 self._distribution_results[frozenset(result.request.component_ids)] = (
                     result
                 )