Update the SDK codebase to work with the latest channel version

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

benchmarks/data_ingestion/benchmark_microgrid_data.py

@@ -17,7 +17,8 @@
 from frequenz.api.microgrid.inverter_pb2 import Inverter
 from frequenz.api.microgrid.meter_pb2 import Data as PbMeterData
 from frequenz.api.microgrid.meter_pb2 import Meter
-from frequenz.channels import Broadcast, MergeNamed, Receiver, Select, Sender
+from frequenz.channels import Broadcast, Receiver, Sender
+from frequenz.channels.util import MergeNamed, Select
 from google.protobuf.timestamp_pb2 import Timestamp  # pylint:disable=no-name-in-module
 
 import frequenz.sdk.microgrid.graph as gr
@@ -237,15 +238,15 @@ async def meter_data(
                 data.
         """
         if component_id in self._component_streams:
-            return self._component_streams[component_id].get_receiver()
+            return self._component_streams[component_id].new_receiver()
         chan: Broadcast[MeterData] = Broadcast(f"{component_id=}")
         component_category, meter_connection = self.categories_from_id(component_id)
         if component_category == ComponentCategory.METER:
             if meter_connection is None:
-                asyncio.create_task(market_data_sender(component_id, chan.get_sender()))
+                asyncio.create_task(market_data_sender(component_id, chan.new_sender()))
             if meter_connection == ComponentCategory.PV_ARRAY:
-                asyncio.create_task(pv_data_sender(component_id, chan.get_sender()))
-        return chan.get_receiver()
+                asyncio.create_task(pv_data_sender(component_id, chan.new_sender()))
+        return chan.new_receiver()
 
     async def battery_data(
         self,
@@ -261,10 +262,10 @@ async def battery_data(
                 battery data.
         """
         if component_id in self._component_streams:
-            return self._component_streams[component_id].get_receiver()
+            return self._component_streams[component_id].new_receiver()
         chan: Broadcast[BatteryData] = Broadcast(f"{component_id=}")
-        asyncio.create_task(battery_data_sender(component_id, chan.get_sender()))
-        return chan.get_receiver()
+        asyncio.create_task(battery_data_sender(component_id, chan.new_sender()))
+        return chan.new_receiver()
 
     async def inverter_data(
         self,
@@ -280,10 +281,10 @@ async def inverter_data(
                 inverter data.
         """
         if component_id in self._component_streams:
-            return self._component_streams[component_id].get_receiver()
+            return self._component_streams[component_id].new_receiver()
         chan: Broadcast[InverterData] = Broadcast(f"{component_id=}")
-        asyncio.create_task(inverter_data_sender(component_id, chan.get_sender()))
-        return chan.get_receiver()
+        asyncio.create_task(inverter_data_sender(component_id, chan.new_sender()))
+        return chan.new_receiver()
 
 
 # pylint: disable=too-many-locals
@@ -339,20 +340,20 @@ async def benchmark_multiple_batteries(n_bat: int = 10, n_msg: int = 100) -> Non
         microgrid_client,
         component_graph,
         {
-            "client_load": client_load_chan.get_sender(),
-            "grid_load": grid_load_chan.get_sender(),
-            "pv_prod": pv_prod_chan.get_sender(),
-            "ev_chargers_consumption": ev_chargers_consumption_chan.get_sender(),
-            "batteries_remaining_energy": batteries_remaining_energy_chan.get_sender(),
-            "batteries_active_power": batteries_active_power_chan.get_sender(),
-            "batteries_active_power_bounds": batteries_active_power_bounds_chan.get_sender(),
+            "client_load": client_load_chan.new_sender(),
+            "grid_load": grid_load_chan.new_sender(),
+            "pv_prod": pv_prod_chan.new_sender(),
+            "ev_chargers_consumption": ev_chargers_consumption_chan.new_sender(),
+            "batteries_remaining_energy": batteries_remaining_energy_chan.new_sender(),
+            "batteries_active_power": batteries_active_power_chan.new_sender(),
+            "batteries_active_power_bounds": batteries_active_power_bounds_chan.new_sender(),
         },
         formula_calculator,
     )
     select = Select(
-        batteries_remaining_energy=batteries_remaining_energy_chan.get_receiver(),
-        batteries_active_power=batteries_active_power_chan.get_receiver(),
-        batteries_active_power_bounds=batteries_active_power_bounds_chan.get_receiver(),
+        batteries_remaining_energy=batteries_remaining_energy_chan.new_receiver(),
+        batteries_active_power=batteries_active_power_chan.new_receiver(),
+        batteries_active_power_bounds=batteries_active_power_bounds_chan.new_receiver(),
     )
     start = time.time()
     while await select.ready():
@@ -431,21 +432,21 @@ async def benchmark_multiple_meters(n_meter: int = 5, n_msg: int = 100) -> None:
         microgrid_client,
         component_graph,
         {
-            "client_load": client_load_chan.get_sender(),
-            "grid_load": grid_load_chan.get_sender(),
-            "pv_prod": pv_prod_chan.get_sender(),
-            "ev_chargers_consumption": ev_chargers_consumption_chan.get_sender(),
-            "batteries_remaining_energy": batteries_remaining_energy_chan.get_sender(),
-            "batteries_active_power": batteries_active_power_chan.get_sender(),
-            "batteries_active_power_bounds": batteries_active_power_bounds_chan.get_sender(),
+            "client_load": client_load_chan.new_sender(),
+            "grid_load": grid_load_chan.new_sender(),
+            "pv_prod": pv_prod_chan.new_sender(),
+            "ev_chargers_consumption": ev_chargers_consumption_chan.new_sender(),
+            "batteries_remaining_energy": batteries_remaining_energy_chan.new_sender(),
+            "batteries_active_power": batteries_active_power_chan.new_sender(),
+            "batteries_active_power_bounds": batteries_active_power_bounds_chan.new_sender(),
         },
         formula_calculator,
     )
     n_client_load_msg = 0
     n_pv_prod_msg = 0
     select = Select(
-        client_load=client_load_chan.get_receiver(),
-        pv_prod=pv_prod_chan.get_receiver(),
+        client_load=client_load_chan.new_receiver(),
+        pv_prod=pv_prod_chan.new_receiver(),
     )
     start = time.time()
     while await select.ready():
@@ -529,19 +530,19 @@ async def benchmark_multiple_formulas(n_formula: int = 100, n_msg: int = 100) ->
         for index in range(n_formula)
     }
     senders: Dict[str, Sender[TimeSeriesEntry[Any]]] = {
-        "client_load": client_load_chan.get_sender(),
-        "grid_load": grid_load_chan.get_sender(),
-        "pv_prod": pv_prod_chan.get_sender(),
-        "ev_chargers_consumption": ev_chargers_consumption_chan.get_sender(),
-        "batteries_remaining_energy": batteries_remaining_energy_chan.get_sender(),
-        "batteries_active_power": batteries_active_power_chan.get_sender(),
-        "batteries_active_power_bounds": batteries_active_power_bounds_chan.get_sender(),
+        "client_load": client_load_chan.new_sender(),
+        "grid_load": grid_load_chan.new_sender(),
+        "pv_prod": pv_prod_chan.new_sender(),
+        "ev_chargers_consumption": ev_chargers_consumption_chan.new_sender(),
+        "batteries_remaining_energy": batteries_remaining_energy_chan.new_sender(),
+        "batteries_active_power": batteries_active_power_chan.new_sender(),
+        "batteries_active_power_bounds": batteries_active_power_bounds_chan.new_sender(),
     }
     for index in range(n_formula):
         locals()["f" + str(index) + "_chan"] = Broadcast[TimeSeriesEntry[Any]](
             "f" + str(index)
         )
-        senders["f" + str(index)] = locals()["f" + str(index) + "_chan"].get_sender()
+        senders["f" + str(index)] = locals()["f" + str(index) + "_chan"].new_sender()
     formula_calculator = FormulaCalculator(
         component_graph, additional_formulas=microgrid_formulas
     )
@@ -550,7 +551,7 @@ async def benchmark_multiple_formulas(n_formula: int = 100, n_msg: int = 100) ->
     )
     recv = {}
     for index in range(n_formula):
-        recv["f" + str(index)] = locals()["f" + str(index) + "_chan"].get_receiver()
+        recv["f" + str(index)] = locals()["f" + str(index) + "_chan"].new_receiver()
 
     recv_merged = MergeNamed(**recv)
     n_formula_msg = {"f" + str(fi): 0 for fi in range(n_formula)}

benchmarks/power_distribution/power_distributor.py

@@ -11,7 +11,7 @@
 from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
 
 import grpc.aio as grpcaio
-from frequenz.channels import Bidirectional, BidirectionalHandle
+from frequenz.channels import Bidirectional
 
 from frequenz.sdk.microgrid.client import MicrogridApiClient, MicrogridGrpcClient
 from frequenz.sdk.microgrid.component import Component, ComponentCategory
@@ -27,7 +27,7 @@ class User:
     """User definition."""
 
     user_id: str
-    channel: BidirectionalHandle[Request, Result]
+    channel: Bidirectional.Handle[Request, Result]
 
 
 async def run_user(user: User, batteries: Set[int], request_num: int) -> List[Result]:

examples/sdk_resampling_example.py

@@ -5,7 +5,8 @@
 
 import asyncio
 
-from frequenz.channels import Broadcast, MergeNamed
+from frequenz.channels import Broadcast
+from frequenz.channels.util import MergeNamed
 
 from frequenz.sdk.actor import ChannelRegistry
 from frequenz.sdk.actor.data_sourcing import DataSourcingActor
@@ -30,14 +31,14 @@ async def run() -> None:
     data_source_request_channel = Broadcast[ComponentMetricRequest](
         "Data Source Request Channel"
     )
-    data_source_request_sender = data_source_request_channel.get_sender()
-    data_source_request_receiver = data_source_request_channel.get_receiver()
+    data_source_request_sender = data_source_request_channel.new_sender()
+    data_source_request_receiver = data_source_request_channel.new_receiver()
 
     resampling_actor_request_channel = Broadcast[ComponentMetricRequest](
         "Resampling Actor Request Channel"
     )
-    resampling_actor_request_sender = resampling_actor_request_channel.get_sender()
-    resampling_actor_request_receiver = resampling_actor_request_channel.get_receiver()
+    resampling_actor_request_sender = resampling_actor_request_channel.new_sender()
+    resampling_actor_request_receiver = resampling_actor_request_channel.new_receiver()
 
     # Instantiate a data sourcing actor
     _data_sourcing_actor = DataSourcingActor(
@@ -81,7 +82,7 @@ async def run() -> None:
     # Store sample receivers for each subscription
     sample_receiver = MergeNamed(
         **{
-            channel_name: channel_registry.get_receiver(channel_name)
+            channel_name: channel_registry.new_receiver(channel_name)
             for channel_name in map(
                 lambda req: req.get_channel_name(), subscription_requests
             )

examples/sdk_usage_example.py

@@ -16,13 +16,7 @@
 from queue import Queue
 from typing import Any, List, Optional, Set
 
-from frequenz.channels import (
-    Bidirectional,
-    BidirectionalHandle,
-    Broadcast,
-    Receiver,
-    Sender,
-)
+from frequenz.channels import Bidirectional, Broadcast, Receiver, Sender
 
 from frequenz.sdk.actor import actor
 from frequenz.sdk.data_handling import TimeSeriesEntry
@@ -48,7 +42,7 @@ class DecisionMakingActor:
     def __init__(  # pylint: disable=too-many-arguments
         self,
         power_channel: Receiver[List[float]],
-        power_distributor_handle: BidirectionalHandle[Request, Result],
+        power_distributor_handle: Bidirectional.Handle[Request, Result],
         batteries: Set[int],
     ) -> None:
         """Create actor instance.
@@ -177,7 +171,7 @@ async def run() -> None:
         # microgrid_client=microgrid_api.microgrid_api,  # in v0.8.0
         component_graph=api.component_graph,
         outputs={
-            key: channel.get_sender()
+            key: channel.new_sender()
             for key, channel in microgrid_data_channels.items()
         },
         formula_calculator=formula_calculator,
@@ -212,18 +206,18 @@ async def run() -> None:
     )
 
     service_actor = DecisionMakingActor(
-        power_channel=request_channel.get_receiver(),
+        power_channel=request_channel.new_receiver(),
         power_distributor_handle=power_distributor_channels[
             sending_actor_id
         ].client_handle,
         batteries={battery.component_id for battery in batteries},
     )
 
     client_actor = DataCollectingActor(
-        request_channel=request_channel.get_sender(),
+        request_channel=request_channel.new_sender(),
         active_power_data=microgrid_data_channels[
             "batteries_active_power"
-        ].get_receiver(name="DecisionMakingActor"),
+        ].new_receiver(name="DecisionMakingActor"),
     )
 
     # pylint: disable=no-member

src/frequenz/sdk/actor/channel_registry.py

@@ -24,7 +24,7 @@ def __init__(self, *, name: str) -> None:
         self._name = name
         self._channels: Dict[str, Broadcast[Any]] = {}
 
-    def get_sender(self, key: str) -> Sender[Any]:
+    def new_sender(self, key: str) -> Sender[Any]:
         """Get a sender to a dynamically created channel with the given key.
 
         Args:
@@ -35,9 +35,9 @@ def get_sender(self, key: str) -> Sender[Any]:
         """
         if key not in self._channels:
             self._channels[key] = Broadcast(f"{self._name}-{key}")
-        return self._channels[key].get_sender()
+        return self._channels[key].new_sender()
 
-    def get_receiver(self, key: str) -> Receiver[Any]:
+    def new_receiver(self, key: str) -> Receiver[Any]:
         """Get a receiver to a dynamically created channel with the given key.
 
         Args:
@@ -48,4 +48,4 @@ def get_receiver(self, key: str) -> Receiver[Any]:
         """
         if key not in self._channels:
             self._channels[key] = Broadcast(f"{self._name}-{key}")
-        return self._channels[key].get_receiver()
+        return self._channels[key].new_receiver()

src/frequenz/sdk/actor/data_sourcing/microgrid_api_source.py

@@ -275,7 +275,7 @@ def _get_metric_senders(
             (
                 self._get_data_extraction_method(category, metric),
                 [
-                    self._registry.get_sender(request.get_channel_name())
+                    self._registry.new_sender(request.get_channel_name())
                     for request in reqlist
                 ],
             )

src/frequenz/sdk/actor/decorator.py

@@ -106,13 +106,13 @@ async def run(self) -> None:
 
     echo_actor = EchoActor(
         "EchoActor",
-        recv1=input_chan_1.get_receiver(),
-        recv2=input_chan_2.get_receiver(),
-        output=echo_chan.get_sender(),
+        recv1=input_chan_1.new_receiver(),
+        recv2=input_chan_2.new_receiver(),
+        output=echo_chan.new_sender(),
     )
-    echo_rx = echo_chan.get_receiver()
+    echo_rx = echo_chan.new_receiver()
 
-    await input_chan_2.get_sender().send(True)
+    await input_chan_2.new_sender().send(True)
     msg = await echo_rx.receive()
     ```
 
@@ -157,18 +157,18 @@ async def run(self) -> None:
     a2_chan: Broadcast[bool] = Broadcast["A2 stream"]
     a1 = Actor1(
         name="ActorOne",
-        recv=input_chan.get_receiver(),
-        output=a1_chan.get_sender(),
+        recv=input_chan.new_receiver(),
+        output=a1_chan.new_sender(),
     )
     a2 = Actor2(
         name="ActorTwo",
-        recv=a1_chan.get_receiver(),
-        output=a2_chan.get_sender(),
+        recv=a1_chan.new_receiver(),
+        output=a2_chan.new_sender(),
     )
 
-    a2_rx = a2_chan.get_receiver()
+    a2_rx = a2_chan.new_receiver()
 
-    await input_chan.get_sender().send(True)
+    await input_chan.new_sender().send(True)
     msg = await a2_rx.receive()
     ```