Update power distributor to use frequenz.sdk.microgrid (#156)

PowerDistributingActor should not take microgrid_api and component_graph
as constructor argument.
Instead it should get them from frequenz.sdk.microgrid.

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -21,9 +21,11 @@
 - Add a formula generator for SoC in the LogicalMeter
   https://github.com/frequenz-floss/frequenz-sdk-python/pull/137
 
-
 ## Bug Fixes
 
 - Formulas with repeated operators like `#1 - #2 - #3` were getting
   calculated incorrectly as `#1 - (#2 - #3)`.  This has been fixed in
   https://github.com/frequenz-floss/frequenz-sdk-python/pull/141
+
+- Remove `microgrid_api` and `component_graph` arguments from PowerDistributingActor constructor
+  https://github.com/frequenz-floss/frequenz-sdk-python/pull/156

benchmarks/power_distribution/power_distributor.py

@@ -10,9 +10,9 @@
 from dataclasses import dataclass
 from typing import Any, Coroutine, Dict, List, Set  # pylint: disable=unused-import
 
-import grpc.aio as grpcaio
 from frequenz.channels import Bidirectional
 
+from frequenz.sdk import microgrid
 from frequenz.sdk.actor.power_distributing import (
     Error,
     Ignored,
@@ -23,12 +23,9 @@
     Result,
     Success,
 )
-from frequenz.sdk.microgrid import ComponentGraph
-from frequenz.sdk.microgrid._graph import _MicrogridComponentGraph
-from frequenz.sdk.microgrid.client import MicrogridApiClient, MicrogridGrpcClient
 from frequenz.sdk.microgrid.component import Component, ComponentCategory
 
-HOST = "157.90.243.180"
+HOST = "microgrid.sandbox.api.frequenz.io"
 PORT = 61060
 
 
@@ -104,16 +101,13 @@ async def run_test(  # pylint: disable=too-many-locals
     users_num: int,
     requests_per_user: int,
     batteries: Set[int],
-    api: MicrogridApiClient,
-    graph: ComponentGraph,
 ) -> Dict[str, Any]:
     """Run test.
 
     Args:
         users_num: Number of users to register
         requests_per_user: How many request user should send.
         batteries: Set of batteries for each request.
-        distributor: PowerDistributingActor instance.
 
     Returns:
         Dictionary with statistics.
@@ -129,7 +123,7 @@ async def run_test(  # pylint: disable=too-many-locals
         user_id: channel.service_handle for user_id, channel in channels.items()
     }
 
-    distributor = PowerDistributingActor(api, graph, service_channels)
+    distributor = PowerDistributingActor(service_channels)
 
     tasks: List[Coroutine[Any, Any, List[Result]]] = []
     for user_id, channel in channels.items():
@@ -152,25 +146,22 @@ async def run_test(  # pylint: disable=too-many-locals
 async def run() -> None:
     """Create microgrid api and run tests."""
     # pylint: disable=protected-access
-    grpc_channel = grpcaio.insecure_channel(f"{HOST}:{PORT}")
-    api = MicrogridGrpcClient(grpc_channel, target=f"{HOST}:{PORT}")
 
-    graph = _MicrogridComponentGraph()
-    await graph.refresh_from_api(api)
+    await microgrid.initialize(HOST, PORT)
 
-    all_batteries: Set[Component] = graph.components(
+    all_batteries: Set[Component] = microgrid.get().component_graph.components(
         component_category={ComponentCategory.BATTERY}
     )
     batteries_ids = {c.component_id for c in all_batteries}
     # Take some time to get data from components
     await asyncio.sleep(4)
     with open("/dev/stdout", "w", encoding="utf-8") as csvfile:
-        fields = await run_test(0, 0, batteries_ids, api, graph)
+        fields = await run_test(0, 0, batteries_ids)
         out = csv.DictWriter(csvfile, fields.keys())
         out.writeheader()
-        out.writerow(await run_test(1, 1, batteries_ids, api, graph))
-        out.writerow(await run_test(1, 10, batteries_ids, api, graph))
-        out.writerow(await run_test(10, 10, batteries_ids, api, graph))
+        out.writerow(await run_test(1, 1, batteries_ids))
+        out.writerow(await run_test(1, 10, batteries_ids))
+        out.writerow(await run_test(10, 10, batteries_ids))
 
 
 async def main() -> None:

examples/power_distribution.py

@@ -184,9 +184,6 @@ async def run() -> None:
     }
 
     power_distributor = PowerDistributingActor(
-        microgrid_api=microgrid.get().api_client,
-        # microgrid_api=microgrid_api.microgrid_api, in v0.8.0
-        component_graph=microgrid.get().component_graph,
         users_channels={
             key: channel.service_handle
             for key, channel in power_distributor_channels.items()

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

@@ -32,6 +32,7 @@
 from frequenz.channels import Bidirectional, Peekable, Receiver
 from google.protobuf.empty_pb2 import Empty  # pylint: disable=no-name-in-module
 
+from ... import microgrid
 from ...actor._decorator import actor
 from ...microgrid import ComponentGraph
 from ...microgrid.client import MicrogridApiClient
@@ -41,7 +42,7 @@
     ComponentCategory,
     InverterData,
 )
-from ...power import DistributionAlgorithm, InvBatPair
+from ...power import DistributionAlgorithm, DistributionResult, InvBatPair
 from ._battery_pool_status import BatteryPoolStatus
 from .request import Request
 from .result import Error, Ignored, OutOfBound, PartialFailure, Result, Success
@@ -140,22 +141,17 @@ class PowerDistributingActor:
 
     def __init__(
         self,
-        microgrid_api: MicrogridApiClient,
-        component_graph: ComponentGraph,
         users_channels: Dict[str, Bidirectional.Handle[Result, Request]],
         wait_for_data_sec: float = 2,
     ) -> None:
         """Create class instance.
 
         Args:
-            microgrid_api: api for sending the requests.
-            component_graph: component graph of the given microgrid api.
             users_channels: BidirectionalHandle for each user. Key should be
                 user id and value should be BidirectionalHandle.
             wait_for_data_sec: How long actor should wait before processing first
                 request. It is a time needed to collect first components data.
         """
-        self._api = microgrid_api
         self._wait_for_data_sec = wait_for_data_sec
 
         # NOTE: power_distributor_exponent should be received from ConfigManager
@@ -164,19 +160,16 @@ def __init__(
             self.power_distributor_exponent
         )
 
-        batteries = component_graph.components(
-            component_category={ComponentCategory.BATTERY}
-        )
+        graph = microgrid.get().component_graph
+        batteries = graph.components(component_category={ComponentCategory.BATTERY})
 
         self._battery_pool = BatteryPoolStatus(
             battery_ids={battery.component_id for battery in batteries},
             max_blocking_duration_sec=30.0,
             max_data_age_sec=10.0,
         )
 
-        self._bat_inv_map, self._inv_bat_map = self._get_components_pairs(
-            component_graph
-        )
+        self._bat_inv_map, self._inv_bat_map = self._get_components_pairs(graph)
         self._battery_receivers: Dict[int, Peekable[BatteryData]] = {}
         self._inverter_receivers: Dict[int, Peekable[InverterData]] = {}
 
@@ -250,6 +243,7 @@ async def run(self) -> None:
         """
         await self._create_channels()
         await self._battery_pool.async_init()
+        api = microgrid.get().api_client
 
         # Wait few seconds to get data from the channels created above.
         await asyncio.sleep(self._wait_for_data_sec)
@@ -291,20 +285,10 @@ async def run(self) -> None:
                 str(battery_distribution),
             )
 
-            tasks = {
-                inverter_id: asyncio.create_task(
-                    self._api.set_power(inverter_id, power)
-                )
-                for inverter_id, power in distribution.distribution.items()
-            }
-
-            _, pending = await asyncio.wait(
-                tasks.values(),
-                timeout=request.request_timeout_sec,
-                return_when=ALL_COMPLETED,
+            tasks = await self._set_distributed_power(
+                api, distribution, request.request_timeout_sec
             )
 
-            await self._cancel_tasks(pending)
             failed_power, failed_batteries = self._parse_result(
                 tasks, distribution.distribution, request.request_timeout_sec
             )
@@ -330,6 +314,36 @@ async def run(self) -> None:
             self._battery_pool.update_last_request_status(response)
             await user.channel.send(response)
 
+    async def _set_distributed_power(
+        self,
+        api: MicrogridApiClient,
+        distribution: DistributionResult,
+        timeout_sec: float,
+    ) -> Dict[int, asyncio.Task[Empty]]:
+        """Send distributed power to the inverters.
+
+        Args:
+            api: Microgrid api client
+            distribution: Distribution result
+            timeout_sec: How long wait for the response
+
+        Returns:
+            Dict with finished or cancelled task for each inverter.
+        """
+        tasks = {
+            inverter_id: asyncio.create_task(api.set_power(inverter_id, power))
+            for inverter_id, power in distribution.distribution.items()
+        }
+
+        _, pending = await asyncio.wait(
+            tasks.values(),
+            timeout=timeout_sec,
+            return_when=ALL_COMPLETED,
+        )
+
+        await self._cancel_tasks(pending)
+        return tasks
+
     def _check_request(self, request: Request) -> Optional[Result]:
         """Check whether the given request if correct.
 
@@ -546,13 +560,12 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
 
     async def _create_channels(self) -> None:
         """Create channels to get data of components in microgrid."""
+        api = microgrid.get().api_client
         for battery_id, inverter_id in self._bat_inv_map.items():
-            bat_recv: Receiver[BatteryData] = await self._api.battery_data(battery_id)
+            bat_recv: Receiver[BatteryData] = await api.battery_data(battery_id)
             self._battery_receivers[battery_id] = bat_recv.into_peekable()
 
-            inv_recv: Receiver[InverterData] = await self._api.inverter_data(
-                inverter_id
-            )
+            inv_recv: Receiver[InverterData] = await api.inverter_data(inverter_id)
             self._inverter_receivers[inverter_id] = inv_recv.into_peekable()
 
     def _parse_result(

src/frequenz/sdk/timeseries/_resampling.py

@@ -284,7 +284,7 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 period.
 
         Raises:
-            ResamplingError: If some timseries source or sink encounters any
+            ResamplingError: If some timeseries source or sink encounters any
                 errors while receiving or sending samples. In this case the
                 timer still runs and the timeseries will keep receiving data.
                 The user should remove (and re-add if desired) the faulty