DataPipeline: Fix resampling/ds/power actors not started

If any function using the resampling actor or datasourcing actor
was used after the call to `_DataPipeline._start()` the according actors
weer never started as they were only created once we requested them
through functions like `logical_meter()` and similar.

Additionally, the PowerDistributingActor was never started at all.

This commit fixes both those issues.

Signed-off-by: Mathias L. Baumann <[email protected]>

# Conflicts:
#	src/frequenz/sdk/microgrid/_data_pipeline.py

Author: Marenz

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -215,6 +215,7 @@ def _start_power_distributing_actor(self) -> None:
             channel_registry=self._channel_registry,
             battery_status_sender=self._battery_status_channel.new_sender(),
         )
+        self._power_distributing_actor.start()
 
     def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
         """Return a Sender for sending requests to the data sourcing actor.
@@ -237,6 +238,7 @@ def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
                 registry=self._channel_registry,
             )
             self._data_sourcing_actor = _ActorInfo(actor, channel)
+            self._data_sourcing_actor.actor.start()
         return self._data_sourcing_actor.channel.new_sender()
 
     def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
@@ -262,17 +264,9 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
                 config=self._resampler_config,
             )
             self._resampling_actor = _ActorInfo(actor, channel)
+            self._resampling_actor.actor.start()
         return self._resampling_actor.channel.new_sender()
 
-    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()
-        # The power distributing actor is started lazily when the first battery pool is
-        # created.
-
     async def _stop(self) -> None:
         """Stop the data pipeline actors."""
         if self._data_sourcing_actor:
@@ -300,7 +294,6 @@ 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:

tests/microgrid/test_datapipeline.py

@@ -0,0 +1,75 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Basic tests for the DataPipeline."""
+
+import asyncio
+from datetime import timedelta
+from typing import Iterator
+
+import async_solipsism
+import pytest
+from pytest_mock import MockerFixture
+
+from frequenz.sdk.microgrid._data_pipeline import _DataPipeline
+from frequenz.sdk.microgrid.client import Connection
+from frequenz.sdk.microgrid.component import Component, ComponentCategory, InverterType
+from frequenz.sdk.timeseries._resampling import ResamplerConfig
+
+from ..utils.mock_microgrid_client import MockMicrogridClient
+
+
+@pytest.fixture
+def event_loop() -> Iterator[async_solipsism.EventLoop]:
+    """Replace the loop with one that doesn't interact with the outside world."""
+    loop = async_solipsism.EventLoop()
+    asyncio.set_event_loop(loop)  # Set the loop as default
+    yield loop
+    loop.close()
+
+
+async def test_actors_started(mocker: MockerFixture) -> None:
+    """Test that the datasourcing, resampling and power distributing actors are started."""
+
+    datapipeline = _DataPipeline(
+        resampler_config=ResamplerConfig(resampling_period=timedelta(seconds=1))
+    )
+    await asyncio.sleep(1)
+
+    # pylint: disable=protected-access
+    assert datapipeline._data_sourcing_actor is None
+    assert datapipeline._resampling_actor is None
+    assert datapipeline._power_distributing_actor is None
+
+    datapipeline.logical_meter()
+
+    assert datapipeline._data_sourcing_actor is not None
+    assert datapipeline._data_sourcing_actor.actor is not None
+    await asyncio.sleep(1)
+    assert datapipeline._data_sourcing_actor.actor.is_running
+
+    assert datapipeline._resampling_actor is not None
+    assert datapipeline._resampling_actor.actor is not None
+    assert datapipeline._resampling_actor.actor.is_running
+
+    assert datapipeline._power_distributing_actor is None
+
+    mock_client = MockMicrogridClient(
+        set(
+            [
+                Component(1, ComponentCategory.GRID),
+                Component(4, ComponentCategory.INVERTER, InverterType.BATTERY),
+                Component(15, ComponentCategory.BATTERY),
+            ]
+        ),
+        connections=set([Connection(1, 4), Connection(4, 15)]),
+    )
+    mock_client.initialize(mocker)
+
+    datapipeline.battery_pool()
+
+    assert datapipeline._power_distributing_actor is not None
+    await asyncio.sleep(1)
+    assert datapipeline._power_distributing_actor.is_running
+
+    await datapipeline._stop()