Migrate ComponentMetricsResamplingActor to the Actor class

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

Author: llucax

src/frequenz/sdk/actor/_resampling.py

@@ -15,15 +15,14 @@
 from ..timeseries import Sample
 from ..timeseries._quantities import Quantity
 from ..timeseries._resampling import Resampler, ResamplerConfig, ResamplingError
+from ._actor import Actor
 from ._channel_registry import ChannelRegistry
 from ._data_sourcing import ComponentMetricRequest
-from ._decorator import actor
 
 _logger = logging.getLogger(__name__)
 
 
-@actor
-class ComponentMetricsResamplingActor:
+class ComponentMetricsResamplingActor(Actor):
     """An actor to resample microgrid component metrics."""
 
     def __init__(  # pylint: disable=too-many-arguments
@@ -46,6 +45,7 @@ def __init__(  # pylint: disable=too-many-arguments
                 resampmling subscription requests.
             config: The configuration for the resampler.
         """
+        super().__init__()
         self._channel_registry: ChannelRegistry = channel_registry
         self._data_sourcing_request_sender: Sender[
             ComponentMetricRequest
@@ -91,7 +91,7 @@ async def _process_resampling_requests(self) -> None:
         async for request in self._resampling_request_receiver:
             await self._subscribe(request)
 
-    async def run(self) -> None:
+    async def _run(self) -> None:
         """Resample known component metrics and process resampling requests.
 
         If there is a resampling error while resampling some component metric,

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -264,6 +264,11 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
             self._resampling_actor = _ActorInfo(actor, channel)
         return self._resampling_actor.channel.new_sender()
 
+    async def _start(self) -> None:
+        """Start the data pipeline actors."""
+        if self._resampling_actor:
+            await self._resampling_actor.actor.start()
+
     async def _stop(self) -> None:
         """Stop the data pipeline actors."""
         # pylint: disable=protected-access
@@ -291,6 +296,7 @@ async def initialize(resampler_config: ResamplerConfig) -> None:
     if _DATA_PIPELINE is not None:
         raise RuntimeError("DataPipeline is already initialized.")
     _DATA_PIPELINE = _DataPipeline(resampler_config)
+    await _DATA_PIPELINE._start()  # pylint: disable=protected-access
 
 
 def logical_meter() -> LogicalMeter:

src/frequenz/sdk/timeseries/logical_meter/_logical_meter.py

@@ -57,41 +57,40 @@ class LogicalMeter:
         )
 
         # Instantiate a resampling actor
-        _resampling_actor = ComponentMetricsResamplingActor(
+        async with ComponentMetricsResamplingActor(
             channel_registry=channel_registry,
             data_sourcing_request_sender=data_source_request_sender,
             resampling_request_receiver=resampling_request_receiver,
             config=ResamplerConfig(resampling_period=timedelta(seconds=1)),
-        )
-
-        await initialize(
-            "127.0.0.1",
-            50051,
-            ResamplerConfig(resampling_period=timedelta(seconds=1))
-        )
-
-        # Create a logical meter instance
-        logical_meter = LogicalMeter(
-            channel_registry,
-            resampling_request_sender,
-        )
+        ):
+            await initialize(
+                "127.0.0.1",
+                50051,
+                ResamplerConfig(resampling_period=timedelta(seconds=1))
+            )
 
-        # Get a receiver for a builtin formula
-        grid_power_recv = logical_meter.grid_power.new_receiver()
-        for grid_power_sample in grid_power_recv:
-            print(grid_power_sample)
+            # Create a logical meter instance
+            logical_meter = LogicalMeter(
+                channel_registry,
+                resampling_request_sender,
+            )
 
-        # or compose formula receivers to create a new formula
-        net_power_recv = (
-            (
-                logical_meter.grid_power
-                - logical_meter.pv_power
+            # Get a receiver for a builtin formula
+            grid_power_recv = logical_meter.grid_power.new_receiver()
+            for grid_power_sample in grid_power_recv:
+                print(grid_power_sample)
+
+            # or compose formula receivers to create a new formula
+            net_power_recv = (
+                (
+                    logical_meter.grid_power
+                    - logical_meter.pv_power
+                )
+                .build("net_power")
+                .new_receiver()
             )
-            .build("net_power")
-            .new_receiver()
-        )
-        for net_power_sample in net_power_recv:
-            print(net_power_sample)
+            for net_power_sample in net_power_recv:
+                print(net_power_sample)
         ```
     """
 

tests/actor/test_resampling.py

@@ -122,39 +122,37 @@ async def test_single_request(
     resampling_req_chan = Broadcast[ComponentMetricRequest]("resample-req")
     resampling_req_sender = resampling_req_chan.new_sender()
 
-    resampling_actor = ComponentMetricsResamplingActor(
+    async with ComponentMetricsResamplingActor(
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_req_chan.new_sender(),
         resampling_request_receiver=resampling_req_chan.new_receiver(),
         config=ResamplerConfig(
             resampling_period=timedelta(seconds=0.2),
             max_data_age_in_periods=2,
         ),
-    )
-
-    subs_req = ComponentMetricRequest(
-        namespace="Resampling",
-        component_id=9,
-        metric_id=ComponentMetricId.SOC,
-        start_time=None,
-    )
-
-    await resampling_req_sender.send(subs_req)
-    data_source_req = await data_source_req_recv.receive()
-    assert data_source_req is not None
-    assert data_source_req == dataclasses.replace(
-        subs_req, namespace="Resampling:Source"
-    )
-
-    await _assert_resampling_works(
-        channel_registry,
-        fake_time,
-        resampling_chan_name=subs_req.get_channel_name(),
-        data_source_chan_name=data_source_req.get_channel_name(),
-    )
-
-    await resampling_actor._stop()  # type: ignore # pylint: disable=no-member,protected-access
-    await resampling_actor._resampler.stop()  # pylint: disable=protected-access
+    ) as resampling_actor:
+        subs_req = ComponentMetricRequest(
+            namespace="Resampling",
+            component_id=9,
+            metric_id=ComponentMetricId.SOC,
+            start_time=None,
+        )
+
+        await resampling_req_sender.send(subs_req)
+        data_source_req = await data_source_req_recv.receive()
+        assert data_source_req is not None
+        assert data_source_req == dataclasses.replace(
+            subs_req, namespace="Resampling:Source"
+        )
+
+        await _assert_resampling_works(
+            channel_registry,
+            fake_time,
+            resampling_chan_name=subs_req.get_channel_name(),
+            data_source_chan_name=data_source_req.get_channel_name(),
+        )
+
+        await resampling_actor._resampler.stop()  # pylint: disable=protected-access
 
 
 async def test_duplicate_request(
@@ -168,37 +166,35 @@ async def test_duplicate_request(
     resampling_req_chan = Broadcast[ComponentMetricRequest]("resample-req")
     resampling_req_sender = resampling_req_chan.new_sender()
 
-    resampling_actor = ComponentMetricsResamplingActor(
+    async with ComponentMetricsResamplingActor(
         channel_registry=channel_registry,
         data_sourcing_request_sender=data_source_req_chan.new_sender(),
         resampling_request_receiver=resampling_req_chan.new_receiver(),
         config=ResamplerConfig(
             resampling_period=timedelta(seconds=0.2),
             max_data_age_in_periods=2,
         ),
-    )
-
-    subs_req = ComponentMetricRequest(
-        namespace="Resampling",
-        component_id=9,
-        metric_id=ComponentMetricId.SOC,
-        start_time=None,
-    )
-
-    await resampling_req_sender.send(subs_req)
-    data_source_req = await data_source_req_recv.receive()
-
-    # Send duplicate request
-    await resampling_req_sender.send(subs_req)
-    with pytest.raises(asyncio.TimeoutError):
-        await asyncio.wait_for(data_source_req_recv.receive(), timeout=0.1)
-
-    await _assert_resampling_works(
-        channel_registry,
-        fake_time,
-        resampling_chan_name=subs_req.get_channel_name(),
-        data_source_chan_name=data_source_req.get_channel_name(),
-    )
-
-    await resampling_actor._stop()  # type: ignore # pylint: disable=no-member,protected-access
-    await resampling_actor._resampler.stop()  # pylint: disable=protected-access
+    ) as resampling_actor:
+        subs_req = ComponentMetricRequest(
+            namespace="Resampling",
+            component_id=9,
+            metric_id=ComponentMetricId.SOC,
+            start_time=None,
+        )
+
+        await resampling_req_sender.send(subs_req)
+        data_source_req = await data_source_req_recv.receive()
+
+        # Send duplicate request
+        await resampling_req_sender.send(subs_req)
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(data_source_req_recv.receive(), timeout=0.1)
+
+        await _assert_resampling_works(
+            channel_registry,
+            fake_time,
+            resampling_chan_name=subs_req.get_channel_name(),
+            data_source_chan_name=data_source_req.get_channel_name(),
+        )
+
+        await resampling_actor._resampler.stop()  # pylint: disable=protected-access