Add a DataPipeline implementation (#270)

... that abstracts away the creation of the data sourcing and resampling
actors, and provides an easy way to access the logical meter or one of
the pools.

Currently only provides two methods: `logical_meter` and
`ev_charger_pool`.

`battery_pool` probably needs some consensus on how args should be pass
and interaction between `PowerDistributingActor`, which can be done in a
subsequent PR.

This allows for the following interface:

``` python
async def run():
    await microgrid.initialize(
        host=HOST, port=PORT, resampler_config=ResamplerConfig(resampling_period_s=1.0)
    )
    grid_power = microgrid.logical_meter().grid_power()
```

This PR also renames:
1. `microgrid._microgrid` to `microgrid.connection_manager`
2. `tests/utils/mock_microgrid.py` to `mock_microgrid_client.py`

Author: sahas-subramanian-frequenz

RELEASE_NOTES.md

@@ -11,7 +11,17 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* Automatic creation of core data-pipeline actors, to eliminate a lot
+  of boiler plate code.  This makes it much simpler to deploy apps
+  (#270).  For example:
+
+  ``` python
+  async def run():
+      await microgrid.initialize(
+          host=HOST, port=PORT, resampler_config=ResamplerConfig(resampling_period_s=1.0)
+      )
+      grid_power = microgrid.logical_meter().grid_power()
+  ```
 
 ## Bug Fixes
 

benchmarks/power_distribution/power_distributor.py

@@ -13,6 +13,7 @@
 from frequenz.channels import Bidirectional, Broadcast
 
 from frequenz.sdk import microgrid
+from frequenz.sdk.actor import ResamplerConfig
 from frequenz.sdk.actor.power_distributing import (
     BatteryStatus,
     Error,
@@ -24,6 +25,7 @@
     Result,
     Success,
 )
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import Component, ComponentCategory
 
 HOST = "microgrid.sandbox.api.frequenz.io"
@@ -152,9 +154,9 @@ async def run() -> None:
     """Create microgrid api and run tests."""
     # pylint: disable=protected-access
 
-    await microgrid.initialize(HOST, PORT)
+    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
 
-    all_batteries: Set[Component] = microgrid.get().component_graph.components(
+    all_batteries: Set[Component] = connection_manager.get().component_graph.components(
         component_category={ComponentCategory.BATTERY}
     )
     batteries_ids = {c.component_id for c in all_batteries}

examples/battery_pool.py

@@ -18,9 +18,11 @@
     ChannelRegistry,
     ComponentMetricRequest,
     DataSourcingActor,
+    ResamplerConfig,
 )
 from frequenz.sdk.actor.power_distributing import PowerDistributingActor
 from frequenz.sdk.actor.power_distributing._battery_pool_status import BatteryStatus
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import ComponentCategory
 from frequenz.sdk.timeseries.battery_pool import BatteryPool
 
@@ -57,7 +59,7 @@ def create_battery_pool() -> BatteryPool:
         battery_status_sender=battery_status_channel.new_sender(),
     )
 
-    batteries = microgrid.get().component_graph.components(
+    batteries = connection_manager.get().component_graph.components(
         component_category={ComponentCategory.BATTERY}
     )
 
@@ -75,7 +77,9 @@ async def main() -> None:
     logging.basicConfig(
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
-    await microgrid.initialize(host=HOST, port=PORT)
+    await microgrid.initialize(
+        host=HOST, port=PORT, resampler_config=ResamplerConfig(resampling_period_s=1.0)
+    )
 
     battery_pool = create_battery_pool()
     receivers: Dict[str, Receiver[Any]] = {

examples/battery_status.py

@@ -13,10 +13,12 @@
 from frequenz.channels import Broadcast
 
 from frequenz.sdk import microgrid
+from frequenz.sdk.actor import ResamplerConfig
 from frequenz.sdk.actor.power_distributing._battery_pool_status import (
     BatteryPoolStatus,
     BatteryStatus,
 )
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import ComponentCategory
 
 _logger = logging.getLogger(__name__)
@@ -29,10 +31,10 @@ async def main() -> None:
     logging.basicConfig(
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
-    await microgrid.initialize(HOST, PORT)
+    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
     batteries = {
         bat.component_id
-        for bat in microgrid.get().component_graph.components(
+        for bat in connection_manager.get().component_graph.components(
             component_category={ComponentCategory.BATTERY}
         )
     }

examples/power_distribution.py

@@ -33,6 +33,7 @@
     Result,
     Success,
 )
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import Component, ComponentCategory
 from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries.logical_meter import LogicalMeter
@@ -158,7 +159,7 @@ async def run() -> None:
     logging.basicConfig(
         level=logging.DEBUG, format="%(asctime)s %(name)s %(levelname)s:%(message)s"
     )
-    await microgrid.initialize(HOST, PORT)
+    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=1.0))
 
     channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
 
@@ -185,7 +186,7 @@ async def run() -> None:
     logical_meter = LogicalMeter(
         channel_registry,
         resampling_actor_request_channel.new_sender(),
-        microgrid.get().component_graph,
+        connection_manager.get().component_graph,
     )
     sending_actor_id: str = "SendingActor"
     # Bidirectional channel is used for one sender - one receiver communication
@@ -212,7 +213,7 @@ async def run() -> None:
 
     # You should get components from ComponentGraph, not from the api.
     # It is faster and and non blocking approach.
-    batteries: Set[Component] = microgrid.get().component_graph.components(
+    batteries: Set[Component] = connection_manager.get().component_graph.components(
         # component_type=set(ComponentType.BATTERY) in v0.8.0
         component_category={ComponentCategory.BATTERY}
     )

examples/resampling.py

@@ -16,6 +16,7 @@
     ComponentMetricsResamplingActor,
     DataSourcingActor,
 )
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId
 from frequenz.sdk.timeseries import Sample
 from frequenz.sdk.timeseries._resampling import Resampler, ResamplerConfig, Sink, Source
@@ -42,7 +43,7 @@ async def _print_sample(sample: Sample) -> None:
 
 async def run() -> None:  # pylint: disable=too-many-locals
     """Run main functions that initializes and creates everything."""
-    await microgrid.initialize(HOST, PORT)
+    await microgrid.initialize(HOST, PORT, ResamplerConfig(resampling_period_s=0.2))
 
     channel_registry = ChannelRegistry(name="data-registry")
 
@@ -68,7 +69,7 @@ async def run() -> None:  # pylint: disable=too-many-locals
         config=ResamplerConfig(resampling_period_s=1),
     )
 
-    components = await microgrid.get().api_client.components()
+    components = await connection_manager.get().api_client.components()
     battery_ids = [
         comp.component_id
         for comp in components

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

@@ -11,7 +11,7 @@
 
 from frequenz.channels import Receiver, Sender
 
-from ... import microgrid
+from ...microgrid import connection_manager
 from ...microgrid.component import (
     BatteryData,
     ComponentCategory,
@@ -138,7 +138,7 @@ async def _get_component_category(
         if comp_id in self._comp_categories_cache:
             return self._comp_categories_cache[comp_id]
 
-        api = microgrid.get().api_client
+        api = connection_manager.get().api_client
         for comp in await api.components():
             self._comp_categories_cache[comp.component_id] = comp.category
 
@@ -168,7 +168,7 @@ async def _check_battery_request(
         if comp_id not in self.comp_data_receivers:
             self.comp_data_receivers[
                 comp_id
-            ] = await microgrid.get().api_client.battery_data(comp_id)
+            ] = await connection_manager.get().api_client.battery_data(comp_id)
 
     async def _check_ev_charger_request(
         self,
@@ -191,7 +191,7 @@ async def _check_ev_charger_request(
         if comp_id not in self.comp_data_receivers:
             self.comp_data_receivers[
                 comp_id
-            ] = await microgrid.get().api_client.ev_charger_data(comp_id)
+            ] = await connection_manager.get().api_client.ev_charger_data(comp_id)
 
     async def _check_inverter_request(
         self,
@@ -214,7 +214,7 @@ async def _check_inverter_request(
         if comp_id not in self.comp_data_receivers:
             self.comp_data_receivers[
                 comp_id
-            ] = await microgrid.get().api_client.inverter_data(comp_id)
+            ] = await connection_manager.get().api_client.inverter_data(comp_id)
 
     async def _check_meter_request(
         self,
@@ -237,7 +237,7 @@ async def _check_meter_request(
         if comp_id not in self.comp_data_receivers:
             self.comp_data_receivers[
                 comp_id
-            ] = await microgrid.get().api_client.meter_data(comp_id)
+            ] = await connection_manager.get().api_client.meter_data(comp_id)
 
     async def _check_requested_component_and_metrics(
         self,

src/frequenz/sdk/actor/power_distributing/_battery_status.py

@@ -20,7 +20,7 @@
 from frequenz.channels.util import Select, Timer
 
 from frequenz.sdk._internal.asyncio import cancel_and_await
-from frequenz.sdk.microgrid import get as get_microgrid
+from frequenz.sdk.microgrid import connection_manager
 from frequenz.sdk.microgrid.component import (
     BatteryData,
     ComponentCategory,
@@ -240,7 +240,7 @@ async def _run(
             set_power_result_receiver: Channel to receive results of the requests to the
                 components.
         """
-        api_client = get_microgrid().api_client
+        api_client = connection_manager.get().api_client
 
         battery_receiver = await api_client.battery_data(self._battery.component_id)
         inverter_receiver = await api_client.inverter_data(self._inverter.component_id)
@@ -494,7 +494,7 @@ def _find_adjacent_inverter_id(self, battery_id: int) -> Optional[int]:
         Returns:
             Id of the inverter. If battery hasn't adjacent inverter, then return None.
         """
-        graph = get_microgrid().component_graph
+        graph = connection_manager.get().component_graph
         return next(
             (
                 comp.component_id

src/frequenz/sdk/actor/power_distributing/power_distributing.py

@@ -32,10 +32,9 @@
 from frequenz.channels import Bidirectional, Peekable, Receiver, Sender
 from google.protobuf.empty_pb2 import Empty  # pylint: disable=no-name-in-module
 
-from ... import microgrid
 from ..._internal.asyncio import cancel_and_await
 from ...actor._decorator import actor
-from ...microgrid import ComponentGraph
+from ...microgrid import ComponentGraph, connection_manager
 from ...microgrid.client import MicrogridApiClient
 from ...microgrid.component import (
     BatteryData,
@@ -168,7 +167,7 @@ def __init__(
         )
 
         self._bat_inv_map, self._inv_bat_map = self._get_components_pairs(
-            microgrid.get().component_graph
+            connection_manager.get().component_graph
         )
         self._battery_receivers: Dict[int, Peekable[BatteryData]] = {}
         self._inverter_receivers: Dict[int, Peekable[InverterData]] = {}
@@ -258,7 +257,7 @@ async def run(self) -> None:
         """
         await self._create_channels()
 
-        api = microgrid.get().api_client
+        api = connection_manager.get().api_client
 
         # Wait few seconds to get data from the channels created above.
         await asyncio.sleep(self._wait_for_data_sec)
@@ -677,7 +676,7 @@ def _get_battery_inverter_data(
 
     async def _create_channels(self) -> None:
         """Create channels to get data of components in microgrid."""
-        api = microgrid.get().api_client
+        api = connection_manager.get().api_client
         for battery_id, inverter_id in self._bat_inv_map.items():
             bat_recv: Receiver[BatteryData] = await api.battery_data(battery_id)
             self._battery_receivers[battery_id] = bat_recv.into_peekable()

src/frequenz/sdk/microgrid/__init__.py

@@ -7,15 +7,29 @@
 for monitoring and adjusting the state of a microgrid.
 """
 
-from . import client, component
+from ..actor import ResamplerConfig
+from . import _data_pipeline, client, component, connection_manager
+from ._data_pipeline import ev_charger_pool, logical_meter
 from ._graph import ComponentGraph
-from ._microgrid import ConnectionManager, get, initialize
+
+
+async def initialize(host: str, port: int, resampler_config: ResamplerConfig) -> None:
+    """Initialize the microgrid connection manager and the data pipeline.
+
+    Args:
+        host: Host to connect to, to reach the microgrid API.
+        port: port to connect to.
+        resampler_config: Configuration for the resampling actor.
+    """
+    await connection_manager.initialize(host, port)
+    _data_pipeline.initialize(resampler_config)
+
 
 __all__ = [
     "ComponentGraph",
-    "ConnectionManager",
-    "get",
     "initialize",
     "client",
     "component",
+    "ev_charger_pool",
+    "logical_meter",
 ]