Update PowerDistributor to use use frequenz.sdk.microgrid

...instead of taking them in constructor.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

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(