Use more precise channel sender and receiver types

Signed-off-by: Simon Völcker <simon.voelcker@frequenz.com>

Author: simonvoelcker

benchmarks/timeseries/benchmark_datasourcing.py

@@ -17,6 +17,7 @@
 from typing import Any
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, ReceiverStoppedError
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Quantity
 
@@ -106,7 +107,7 @@ async def consume(channel: Receiver[Any]) -> None:
     for evc_id in mock_grid.evc_ids:
         for component_metric_id in COMPONENT_METRIC_IDS:
             telem_stream_sender, telem_stream_receiver = OneshotChannel[
-                Receiver[Sample[Quantity]]
+                BroadcastReceiver[Sample[Quantity]]
             ]()
             request = ComponentMetricRequest(
                 namespace="current_phase_requests",

src/frequenz/sdk/microgrid/_data_sourcing/_component_metric_request.py

@@ -6,7 +6,8 @@
 from dataclasses import dataclass
 from datetime import datetime
 
-from frequenz.channels import Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
+from frequenz.channels._oneshot import OneshotSender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Quantity
@@ -55,7 +56,7 @@ class ComponentMetricRequest:
     If None, only live data is streamed.
     """
 
-    telem_stream_sender: Sender[Receiver[Sample[Quantity]]]
+    telem_stream_sender: OneshotSender[BroadcastReceiver[Sample[Quantity]]]
     """Sender of a oneshot channel used to send the data stream back to the requester."""
 
     def get_channel_name(self) -> str:

src/frequenz/sdk/microgrid/_power_managing/_base_classes.py

@@ -10,7 +10,8 @@
 import enum
 import typing
 
-from frequenz.channels import Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
+from frequenz.channels._oneshot import OneshotSender
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Power
 
@@ -34,7 +35,7 @@ class ReportRequest:
     priority: int
     """The priority of the actor."""
 
-    report_stream_sender: Sender[Receiver[_Report]]
+    report_stream_sender: OneshotSender[BroadcastReceiver[_Report]]
     """Oneshot sender to transmit the report receiver."""
 
     def get_channel_name(self) -> str:

src/frequenz/sdk/timeseries/_grid_frequency.py

@@ -9,6 +9,7 @@
 import logging
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.channels.experimental import Pipe
 from frequenz.client.microgrid.component import Component, EvCharger, Inverter, Meter
 from frequenz.client.microgrid.metrics import Metric
@@ -50,7 +51,7 @@ def __init__(
 
         # Microgrid API source will send the stream through a oneshot channel
         telem_stream_sender, self._telem_stream_receiver = OneshotChannel[
-            Receiver[Sample[Quantity]]
+            BroadcastReceiver[Sample[Quantity]]
         ]()
 
         self._component_metric_request = ComponentMetricRequest(
@@ -116,7 +117,7 @@ def new_receiver(self) -> Receiver[Sample[Frequency]]:
     async def subscribe(self) -> Receiver[Sample[Frequency]]:
         """Create a receiver for grid frequency."""
         telem_stream_sender, telem_stream_receiver = OneshotChannel[
-            Receiver[Sample[Quantity]]
+            BroadcastReceiver[Sample[Quantity]]
         ]()
         component_metric_request = ComponentMetricRequest(
             "grid-frequency",

src/frequenz/sdk/timeseries/_voltage_streamer.py

@@ -14,6 +14,7 @@
 from typing import TYPE_CHECKING
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.client.microgrid.component import Component, EvCharger, Inverter, Meter
 from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Quantity, Voltage
@@ -145,7 +146,7 @@ async def _send_request(self) -> None:
         phases_rx: list[Receiver[Sample[Quantity]]] = []
         for metric in metrics:
             telem_stream_sender, telem_stream_receiver = OneshotChannel[
-                Receiver[Sample[Quantity]]
+                BroadcastReceiver[Sample[Quantity]]
             ]()
             req = ComponentMetricRequest(
                 namespace=self._namespace,

src/frequenz/sdk/timeseries/battery_pool/_battery_pool.py

@@ -13,6 +13,7 @@
 from collections import abc
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.channels.experimental import Pipe
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Energy, Percentage, Power, Temperature
@@ -336,7 +337,7 @@ def power_status(self) -> ReceiverFetcher[BatteryPoolReport]:
             A receiver that will stream power status reports for the pool's priority.
         """
         report_stream_sender, self._report_stream_receiver = OneshotChannel[
-            Receiver[_Report]
+            BroadcastReceiver[_Report]
         ]()
         request = _power_managing.ReportRequest(
             source_id=self._source_id,

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

@@ -9,6 +9,7 @@
 from collections import abc
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.channels.experimental import Pipe
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Current, Power
@@ -178,7 +179,7 @@ def power_status(self) -> ReceiverFetcher[EVChargerPoolReport]:
             A receiver that will stream power status reports for the pool's priority.
         """
         report_stream_sender, self._report_stream_receiver = OneshotChannel[
-            Receiver[_Report]
+            BroadcastReceiver[_Report]
         ]()
         request = _power_managing.ReportRequest(
             source_id=self._source_id,

src/frequenz/sdk/timeseries/formulas/_resampled_stream_fetcher.py

@@ -4,6 +4,7 @@
 """Fetches telemetry streams for components."""
 
 from frequenz.channels import OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Quantity
 
@@ -50,7 +51,7 @@ async def fetch_stream(
             A receiver to stream resampled data for the given component id.
         """
         telem_stream_sender, telem_stream_receiver = OneshotChannel[
-            Receiver[Sample[Quantity]]
+            BroadcastReceiver[Sample[Quantity]]
         ]()
 
         request = ComponentMetricRequest(

src/frequenz/sdk/timeseries/pv_pool/_pv_pool.py

@@ -8,6 +8,7 @@
 from collections import abc
 
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.channels.experimental import Pipe
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.quantities import Power
@@ -150,7 +151,7 @@ def power_status(self) -> ReceiverFetcher[PVPoolReport]:
             A receiver that will stream power status reports for the pool's priority.
         """
         report_stream_sender, self._report_stream_receiver = OneshotChannel[
-            Receiver[_Report]
+            BroadcastReceiver[_Report]
         ]()
         request = _power_managing.ReportRequest(
             source_id=self._source_id,

tests/actor/test_resampling.py

@@ -9,6 +9,7 @@
 import pytest
 import time_machine
 from frequenz.channels import Broadcast, OneshotChannel, Receiver, Sender
+from frequenz.channels._broadcast import BroadcastReceiver
 from frequenz.client.common.microgrid.components import ComponentId
 from frequenz.client.microgrid.metrics import Metric
 from frequenz.quantities import Quantity
@@ -106,7 +107,7 @@ async def test_single_request(
         ),
     ) as resampling_actor:
         telem_stream_sender, telem_stream_receiver = OneshotChannel[
-            Receiver[Sample[Quantity]]
+            BroadcastReceiver[Sample[Quantity]]
         ]()
         subs_req = ComponentMetricRequest(
             namespace="Resampling",
@@ -157,7 +158,7 @@ async def test_duplicate_request(
         ),
     ) as resampling_actor:
         telem_stream_sender, telem_stream_receiver = OneshotChannel[
-            Receiver[Sample[Quantity]]
+            BroadcastReceiver[Sample[Quantity]]
         ]()
         subs_req = ComponentMetricRequest(
             namespace="Resampling",
@@ -211,7 +212,7 @@ async def test_resubscribe(fake_time: time_machine.Coordinates) -> None:
 
         async def send_metric_request() -> Receiver[Receiver[Sample[Quantity]]]:
             telem_stream_sender, telem_stream_receiver = OneshotChannel[
-                Receiver[Sample[Quantity]]
+                BroadcastReceiver[Sample[Quantity]]
             ]()
             subs_req = ComponentMetricRequest(
                 namespace="Resampling",