Rename resampling actor arguments

This is to make them easier to understand. Also require to use keyword
arguments to avoid mixing up `resampling_period_s` and
`max_data_age_in_periods`.

Also remove the example from the init documentation as it was outdated
and we already provide an example in `examples/` that at least gets
actually checked for syntax and linted.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

examples/resampling.py

@@ -32,9 +32,9 @@ async def run() -> None:  # pylint: disable=too-many-locals
     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]("resample")
-    resampling_actor_request_sender = resampling_actor_request_channel.new_sender()
-    resampling_actor_request_receiver = resampling_actor_request_channel.new_receiver()
+    resampling_request_channel = Broadcast[ComponentMetricRequest]("resample")
+    resampling_request_sender = resampling_request_channel.new_sender()
+    resampling_request_receiver = resampling_request_channel.new_receiver()
 
     # Instantiate a data sourcing actor
     _data_sourcing_actor = DataSourcingActor(
@@ -44,8 +44,8 @@ async def run() -> None:  # pylint: disable=too-many-locals
     # Instantiate a resampling actor
     _resampling_actor = ComponentMetricsResamplingActor(
         channel_registry=channel_registry,
-        subscription_sender=data_source_request_sender,
-        subscription_receiver=resampling_actor_request_receiver,
+        data_sourcing_request_sender=data_source_request_sender,
+        resampling_request_receiver=resampling_request_receiver,
         resampling_period_s=1.0,
     )
 
@@ -69,10 +69,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
 
     # Send the subscription requests
     await asyncio.gather(
-        *[
-            resampling_actor_request_sender.send(request)
-            for request in subscription_requests
-        ]
+        *[resampling_request_sender.send(request) for request in subscription_requests]
     )
 
     # Merge sample receivers for each subscription into one receiver

src/frequenz/sdk/actor/_resampling.py

@@ -45,103 +45,43 @@ class ComponentMetricsResamplingActor:
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
+        *,
         channel_registry: ChannelRegistry,
-        subscription_sender: Sender[ComponentMetricRequest],
-        subscription_receiver: Receiver[ComponentMetricRequest],
+        data_sourcing_request_sender: Sender[ComponentMetricRequest],
+        resampling_request_receiver: Receiver[ComponentMetricRequest],
         resampling_period_s: float = 0.2,
         max_data_age_in_periods: float = 3.0,
         resampling_function: ResamplingFunction = average,
     ) -> None:
         """Initialize the ComponentMetricsResamplingActor.
 
         Args:
-            channel_registry: global channel registry used for receiving component
-                data from DataSource and for sending resampled samples downstream
-            subscription_sender: channel for sending component metric requests to the
-                DataSourcing actor
-            subscription_receiver: channel for receiving component metric requests
-            resampling_period_s: value describing how often resampling should be
-                performed, in seconds
-            max_data_age_in_periods: max age that samples shouldn't exceed in order
-                to be used in the resampling function
-            resampling_function: function to be applied to a sequence of samples within
-                a resampling period to produce a single output sample
-
-        Example:
-            ```python
-            async def run() -> None:
-                await microgrid_api.initialize(HOST, PORT)
-
-                channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
-
-                data_source_request_channel = Broadcast[ComponentMetricRequest](
-                    "Data Source Request Channel"
-                )
-                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.new_sender()
-                resampling_actor_request_receiver = resampling_actor_request_channel.new_receiver()
-
-                _data_sourcing_actor = DataSourcingActor(
-                    request_receiver=data_source_request_receiver, registry=channel_registry
-                )
-
-                _resampling_actor = ComponentMetricsResamplingActor(
-                    channel_registry=channel_registry,
-                    subscription_sender=data_source_request_sender,
-                    subscription_receiver=resampling_actor_request_receiver,
-                    resampling_period_s=1.0,
-                )
-
-                components = await microgrid_api.get().microgrid_api_client.components()
-                battery_ids = [
-                    comp.component_id
-                    for comp in components
-                    if comp.category == ComponentCategory.BATTERY
-                ]
-
-                subscription_requests = [
-                    ComponentMetricRequest(
-                        namespace="Resampling",
-                        component_id=component_id,
-                        metric_id=ComponentMetricId.SOC,
-                        start_time=None,
-                    )
-                    for component_id in battery_ids
-                ]
-
-                await asyncio.gather(
-                    *[
-                        resampling_actor_request_sender.send(request)
-                        for request in subscription_requests
-                    ]
-                )
-
-                sample_receiver = MergeNamed(
-                    **{
-                        channel_name: channel_registry.new_receiver(channel_name)
-                        for channel_name in map(
-                            lambda req: req.get_channel_name(), subscription_requests
-                        )
-                    }
-                )
-
-                async for channel_name, msg in sample_receiver:
-                    print(msg)
-
-            asyncio.run(run())
-            ```
+            channel_registry: The channel registry used to get senders and
+                receivers for data sourcing subscriptions.
+            data_sourcing_request_sender: The sender used to send requests to
+                the [`DataSourcingActor`][frequenz.sdk.actor.DataSourcingActor]
+                to subscribe to component metrics.
+            resampling_request_receiver: The receiver to use to receive new
+                resampmling subscription requests.
+            resampling_period_s: The time it passes between resampled data
+                should be calculated (in seconds).
+            max_data_age_in_periods: The maximum age a sample can have to be
+                considered *relevant* for resampling purposes, expressed in the
+                number of resampling periods. For exapmle is
+                `resampling_period_s` is 3 and `max_data_age_in_periods` is 2,
+                then data older than `3*2 = 6` secods will be discarded when
+                creating a new sample and never passed to the resampling
+                function.
+            resampling_function: The function to be applied to the sequence of
+                *relevant* samples at a given time. The result of the function
+                is what is sent as the resampled data.
         """
         self._channel_registry = channel_registry
-        self._subscription_sender = subscription_sender
-        self._subscription_receiver = subscription_receiver
         self._resampling_period_s = resampling_period_s
         self._max_data_age_in_periods: float = max_data_age_in_periods
         self._resampling_function: ResamplingFunction = resampling_function
+        self._data_sourcing_request_sender = data_sourcing_request_sender
+        self._resampling_request_receiver = resampling_request_receiver
         self._resampler = Resampler(
             resampling_period_s=resampling_period_s,
             max_data_age_in_periods=max_data_age_in_periods,
@@ -158,7 +98,7 @@ async def _subscribe(self, request: ComponentMetricRequest) -> None:
             request, namespace=request.namespace + ":Source"
         )
         data_source_channel_name = data_source_request.get_channel_name()
-        await self._subscription_sender.send(data_source_request)
+        await self._data_sourcing_request_sender.send(data_source_request)
         receiver = self._channel_registry.new_receiver(data_source_channel_name)
 
         # This is a temporary hack until the Sender implementation uses
@@ -173,7 +113,7 @@ async def sink_adapter(sample: Sample) -> None:
 
     async def _process_resampling_requests(self) -> None:
         """Process resampling data requests."""
-        async for request in self._subscription_receiver:
+        async for request in self._resampling_request_receiver:
             await self._subscribe(request)
 
     async def run(self) -> None:

tests/actor/test_resampling.py

@@ -58,8 +58,8 @@ async def test_component_metrics_resampling_actor(  # pylint: disable=too-many-s
 
     resampling_actor = ComponentMetricsResamplingActor(
         channel_registry=channel_registry,
-        subscription_sender=data_source_req_chan.new_sender(),
-        subscription_receiver=resampling_req_chan.new_receiver(),
+        data_sourcing_request_sender=data_source_req_chan.new_sender(),
+        resampling_request_receiver=resampling_req_chan.new_receiver(),
         resampling_period_s=0.2,
         max_data_age_in_periods=2,
     )

tests/timeseries/test_logical_meter.py

@@ -123,8 +123,8 @@ def next_msg(value: float) -> ComponentData:
 
         ComponentMetricsResamplingActor(
             channel_registry=channel_registry,
-            subscription_sender=data_source_request_sender,
-            subscription_receiver=resampling_actor_request_receiver,
+            data_sourcing_request_sender=data_source_request_sender,
+            resampling_request_receiver=resampling_actor_request_receiver,
             resampling_period_s=0.1,
         )