Add DataSourcingActor

noxfile.py

@@ -68,8 +68,9 @@ def mypy(session: nox.Session, install_deps: bool = True) -> None:
     # Since we use other packages in the frequenz namespace, we need to run the
     # checks for frequenz.sdk from the installed package instead of the src
     # directory.
-    mypy_paths = [path for path in source_file_paths(session)
-            if not path.startswith("src")]
+    mypy_paths = [
+        path for path in source_file_paths(session) if not path.startswith("src")
+    ]
 
     mypy_cmd = [
         "mypy",
@@ -115,10 +116,7 @@ def docstrings(session: nox.Session, install_deps: bool = True) -> None:
     # This is needed only for the `src` dir, so we exclude the other top level
     # dirs that contain code.
     darglint_paths = filter(
-        lambda path: not (
-            path.startswith("tests")
-            or path.startswith("benchmarks")
-        ),
+        lambda path: not (path.startswith("tests") or path.startswith("benchmarks")),
         source_file_paths(session),
     )
     session.run(

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

@@ -0,0 +1,15 @@
+"""
+The DataSourcingActor.
+
+Copyright
+Copyright © 2021 Frequenz Energy-as-a-Service GmbH
+
+License
+MIT
+"""
+
+from .data_sourcing import DataSourcingActor
+
+__all__ = [
+    "DataSourcingActor",
+]

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

@@ -0,0 +1,41 @@
+"""The DataSourcing Actor.
+
+Copyright
+Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+License
+MIT
+"""
+
+from frequenz.channels import Receiver
+
+from frequenz.sdk.actor import actor
+from frequenz.sdk.actor.channel_registry import ChannelRegistry
+from frequenz.sdk.data_pipeline import ComponentMetricRequest
+
+from .microgrid_api_source import MicrogridApiSource
+
+
+@actor
+class DataSourcingActor:
+    """An actor that provides data streams of metrics as time series."""
+
+    def __init__(
+        self,
+        request_receiver: Receiver[ComponentMetricRequest],
+        registry: ChannelRegistry,
+    ) -> None:
+        """Create a `DataSourcingActor` instance.
+
+        Args:
+            request_receiver: A channel receiver to accept metric requests from.
+            registry: A channel registry.  To be replaced by a singleton
+                instance.
+        """
+        self._request_receiver = request_receiver
+        self._microgrid_api_source = MicrogridApiSource(registry)
+
+    async def run(self) -> None:
+        """Run the actor."""
+        async for request in self._request_receiver:
+            await self._microgrid_api_source.add_metric(request)