Migrate DataSourcingActor to the Actor class

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

Author: llucax

benchmarks/timeseries/benchmark_datasourcing.py

@@ -113,23 +113,22 @@ async def consume(channel: Receiver[Any]) -> None:
             await request_sender.send(request)
             consume_tasks.append(asyncio.create_task(consume(recv_channel)))
 
-    DataSourcingActor(request_receiver, channel_registry)
-
-    await asyncio.gather(*consume_tasks)
-
-    time_taken = perf_counter() - start_time
-
-    await mock_grid.cleanup()
-
-    print(f"Samples Sent: {samples_sent}, time taken: {time_taken}")
-    print(f"Samples per second: {samples_sent / time_taken}")
-    print(
-        "Expected samples: "
-        f"{num_expected_messages}, missing: {num_expected_messages - samples_sent}"
-    )
-    print(
-        f"Missing per EVC: {(num_expected_messages - samples_sent) / num_ev_chargers}"
-    )
+    async with DataSourcingActor(request_receiver, channel_registry):
+        await asyncio.gather(*consume_tasks)
+
+        time_taken = perf_counter() - start_time
+
+        await mock_grid.cleanup()
+
+        print(f"Samples Sent: {samples_sent}, time taken: {time_taken}")
+        print(f"Samples per second: {samples_sent / time_taken}")
+        print(
+            "Expected samples: "
+            f"{num_expected_messages}, missing: {num_expected_messages - samples_sent}"
+        )
+        print(
+            f"Missing per EVC: {(num_expected_messages - samples_sent) / num_ev_chargers}"
+        )
 
 
 def parse_args() -> Tuple[int, int, bool]:

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

@@ -5,13 +5,12 @@
 
 from frequenz.channels import Receiver
 
+from .._actor import Actor
 from .._channel_registry import ChannelRegistry
-from .._decorator import actor
 from .microgrid_api_source import ComponentMetricRequest, MicrogridApiSource
 
 
-@actor
-class DataSourcingActor:
+class DataSourcingActor(Actor):
     """An actor that provides data streams of metrics as time series."""
 
     def __init__(
@@ -26,10 +25,11 @@ def __init__(
             registry: A channel registry.  To be replaced by a singleton
                 instance.
         """
+        super().__init__()
         self._request_receiver = request_receiver
         self._microgrid_api_source = MicrogridApiSource(registry)
 
-    async def run(self) -> None:
+    async def _run(self) -> None:
         """Run the actor."""
         async for request in self._request_receiver:
             await self._microgrid_api_source.add_metric(request)

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -266,17 +266,17 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
 
     async def _start(self) -> None:
         """Start the data pipeline actors."""
+        if self._data_sourcing_actor:
+            await self._data_sourcing_actor.actor.start()
         if self._resampling_actor:
             await self._resampling_actor.actor.start()
 
     async def _stop(self) -> None:
         """Stop the data pipeline actors."""
-        # pylint: disable=protected-access
         if self._data_sourcing_actor:
-            await self._data_sourcing_actor.actor._stop()  # type: ignore
+            await self._data_sourcing_actor.actor.stop()
         if self._resampling_actor:
             await self._resampling_actor.actor.stop()
-        # pylint: enable=protected-access
 
 
 _DATA_PIPELINE: _DataPipeline | None = None

tests/actor/test_data_sourcing.py

@@ -57,41 +57,41 @@ async def test_data_sourcing_actor(self) -> None:
 
         registry = ChannelRegistry(name="test-registry")
 
-        DataSourcingActor(req_chan.new_receiver(), registry)
-        active_power_request = ComponentMetricRequest(
-            "test-namespace", 4, ComponentMetricId.ACTIVE_POWER, None
-        )
-        active_power_recv = registry.new_receiver(
-            active_power_request.get_channel_name()
-        )
-        await req_sender.send(active_power_request)
-
-        soc_request = ComponentMetricRequest(
-            "test-namespace", 9, ComponentMetricId.SOC, None
-        )
-        soc_recv = registry.new_receiver(soc_request.get_channel_name())
-        await req_sender.send(soc_request)
-
-        soc2_request = ComponentMetricRequest(
-            "test-namespace", 9, ComponentMetricId.SOC, None
-        )
-        soc2_recv = registry.new_receiver(soc2_request.get_channel_name())
-        await req_sender.send(soc2_request)
-
-        for _ in range(3):
-            sample = await soc_recv.receive()
-            assert sample is not None
-            assert 9.0 == sample.value.base_value
-
-            sample = await soc2_recv.receive()
-            assert sample is not None
-            assert 9.0 == sample.value.base_value
-
-            sample = await active_power_recv.receive()
-            assert sample is not None
-            assert 100.0 == sample.value.base_value
-
-        assert await server.graceful_shutdown()
-        connection_manager._CONNECTION_MANAGER = (  # pylint: disable=protected-access
-            None
-        )
+        async with DataSourcingActor(req_chan.new_receiver(), registry):
+            active_power_request = ComponentMetricRequest(
+                "test-namespace", 4, ComponentMetricId.ACTIVE_POWER, None
+            )
+            active_power_recv = registry.new_receiver(
+                active_power_request.get_channel_name()
+            )
+            await req_sender.send(active_power_request)
+
+            soc_request = ComponentMetricRequest(
+                "test-namespace", 9, ComponentMetricId.SOC, None
+            )
+            soc_recv = registry.new_receiver(soc_request.get_channel_name())
+            await req_sender.send(soc_request)
+
+            soc2_request = ComponentMetricRequest(
+                "test-namespace", 9, ComponentMetricId.SOC, None
+            )
+            soc2_recv = registry.new_receiver(soc2_request.get_channel_name())
+            await req_sender.send(soc2_request)
+
+            for _ in range(3):
+                sample = await soc_recv.receive()
+                assert sample is not None
+                assert 9.0 == sample.value.base_value
+
+                sample = await soc2_recv.receive()
+                assert sample is not None
+                assert 9.0 == sample.value.base_value
+
+                sample = await active_power_recv.receive()
+                assert sample is not None
+                assert 100.0 == sample.value.base_value
+
+            assert await server.graceful_shutdown()
+            connection_manager._CONNECTION_MANAGER = (  # pylint: disable=protected-access
+                None
+            )