Make _data_pipeline.initialize() async

We'll need to start actors in this function in the future, so we need it
to be async to be able to `await actor.start()`.

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

Author: llucax

src/frequenz/sdk/microgrid/__init__.py

@@ -22,7 +22,7 @@ async def initialize(host: str, port: int, resampler_config: ResamplerConfig) ->
         resampler_config: Configuration for the resampling actor.
     """
     await connection_manager.initialize(host, port)
-    _data_pipeline.initialize(resampler_config)
+    await _data_pipeline.initialize(resampler_config)
 
 
 __all__ = [

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -270,14 +270,14 @@ async def _stop(self) -> None:
         if self._data_sourcing_actor:
             await self._data_sourcing_actor.actor._stop()  # type: ignore
         if self._resampling_actor:
-            await self._resampling_actor.actor._stop()  # type: ignore
+            await self._resampling_actor.actor.stop()
         # pylint: enable=protected-access
 
 
 _DATA_PIPELINE: _DataPipeline | None = None
 
 
-def initialize(resampler_config: ResamplerConfig) -> None:
+async def initialize(resampler_config: ResamplerConfig) -> None:
     """Initialize a `DataPipeline` instance.
 
     Args:

tests/timeseries/_battery_pool/test_battery_pool.py

@@ -136,7 +136,7 @@ async def setup_all_batteries(mocker: MockerFixture) -> AsyncIterator[SetupArgs]
     min_update_interval: float = 0.2
     # pylint: disable=protected-access
     microgrid._data_pipeline._DATA_PIPELINE = None
-    microgrid._data_pipeline.initialize(
+    await microgrid._data_pipeline.initialize(
         ResamplerConfig(resampling_period=timedelta(seconds=min_update_interval))
     )
     streamer = MockComponentDataStreamer(mock_microgrid)
@@ -188,7 +188,7 @@ async def setup_batteries_pool(mocker: MockerFixture) -> AsyncIterator[SetupArgs
     min_update_interval: float = 0.2
     # pylint: disable=protected-access
     microgrid._data_pipeline._DATA_PIPELINE = None
-    microgrid._data_pipeline.initialize(
+    await microgrid._data_pipeline.initialize(
         ResamplerConfig(resampling_period=timedelta(seconds=min_update_interval))
     )