Define way to create object with async constructor (#145)

Author: ela-kotulska-frequenz

CONTRIBUTING.md

@@ -21,7 +21,7 @@ python -m pip install -e .
 You can also use `nox` to run the tests and other checks:
 
 ```sh
-python -m pip install nox
+python -m pip install nox toml tomli
 nox
 ```
 

src/frequenz/sdk/_internal/asyncio.py

@@ -4,6 +4,7 @@
 """General purpose async tools."""
 
 import asyncio
+from abc import ABC
 
 
 async def cancel_and_await(task: asyncio.Task) -> None:
@@ -19,3 +20,22 @@ async def cancel_and_await(task: asyncio.Task) -> None:
         await task
     except asyncio.CancelledError:
         pass
+
+
+class NotSyncConstructible(AssertionError):
+    """Raised when object with async constructor is created in sync way."""
+
+
+class AsyncConstructible(ABC):
+    """Parent class for classes where part of the constructor is async."""
+
+    def __init__(self) -> None:
+        """Raise error when object is created in sync way.
+
+        Raises:
+            NotSyncConstructible: If this method is called.
+        """
+        raise NotSyncConstructible(
+            "This object shouldn't be created with default constructor. ",
+            "Check class documentation for more information.",
+        )

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

@@ -2,30 +2,39 @@
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
 """Class that stores pool of batteries and manage them."""
 
+from __future__ import annotations
+
 import asyncio
 import logging
-from typing import Set
+from typing import Dict, Set
 
+from ..._internal.asyncio import AsyncConstructible
 from ...microgrid._battery import BatteryStatus, StatusTracker
 from .result import PartialFailure, Result, Success
 
 _logger = logging.getLogger(__name__)
 
 
-class BatteryPoolStatus:
-    """Holds pool of batteries and returns data from them."""
+class BatteryPoolStatus(AsyncConstructible):
+    """Return status of batteries in the pool.
+
+    To create an instance of this class you should use `async_new` class method.
+    Standard constructor (__init__) is not supported and using it will raise
+    `NotSyncConstructible` error.
+    """
 
-    def __init__(
-        self,
+    # This is instance attribute.
+    # Don't assign default value, because then it becomes class attribute.
+    _batteries: Dict[int, StatusTracker]
+
+    @classmethod
+    async def async_new(
+        cls,
         battery_ids: Set[int],
         max_data_age_sec: float,
         max_blocking_duration_sec: float,
-    ) -> None:
-        """Create partially initialized object instance.
-
-        Note:
-            Please call `async_init` method to fully initialize BatteryPoolStatus. Otherwise
-            it is not possible to use BatteryPoolStatus.
+    ) -> BatteryPoolStatus:
+        """Create BatteryPoolStatus instance.
 
         Args:
             battery_ids: set of batteries ids that should be stored in pool.
@@ -35,20 +44,24 @@ def __init__(
                 data.
             max_blocking_duration_sec: This value tell what should be the maximum
                 timeout used for blocking failing component.
+
+        Raises:
+            RuntimeError: If any battery has no adjacent inverter.
+
+        Returns:
+            New instance of this class.
         """
-        self._batteries = {
-            id: StatusTracker(id, max_data_age_sec, max_blocking_duration_sec)
+        self: BatteryPoolStatus = BatteryPoolStatus.__new__(cls)
+
+        tasks = [
+            StatusTracker.async_new(id, max_data_age_sec, max_blocking_duration_sec)
             for id in battery_ids
-        }
-        self._init_method_called: bool = False
-
-    async def async_init(self) -> None:
-        """Init battery pool."""
-        await asyncio.gather(
-            *[bat.async_init() for bat in self._batteries.values()],
-            return_exceptions=True,
-        )
-        self._init_method_called = True
+        ]
+
+        trackers = await asyncio.gather(*tasks)
+        self._batteries = {tracker.battery_id: tracker for tracker in trackers}
+
+        return self
 
     def get_working_batteries(self, battery_ids: Set[int]) -> Set[int]:
         """Get subset of battery_ids with working batteries.
@@ -64,11 +77,6 @@ def get_working_batteries(self, battery_ids: Set[int]) -> Set[int]:
         Returns:
             Subset of given batteries with working batteries.
         """
-        if not self._init_method_called:
-            raise RuntimeError(
-                "`async_init` method not called or not awaited. Run it before first use"
-            )
-
         working: Set[int] = set()
         uncertain: Set[int] = set()
         for bat_id in battery_ids:
@@ -101,11 +109,6 @@ def update_last_request_status(self, result: Result):
             RuntimeError: If `async_init` method was not called at the beginning to
                 initialize object.
         """
-        if not self._init_method_called:
-            raise RuntimeError(
-                "`async_init` method not called or not awaited. Run it before first use"
-            )
-
         if isinstance(result, Success):
             for bat_id in result.used_batteries:
                 self._batteries[bat_id].unblock()

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

@@ -161,16 +161,9 @@ def __init__(
             self.power_distributor_exponent
         )
 
-        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(
+            microgrid.get().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]] = {}
 
@@ -251,18 +244,24 @@ async def run(self) -> None:
         as broken for some time.
         """
         await self._create_channels()
-        await self._battery_pool.async_init()
+
         api = microgrid.get().api_client
+        battery_pool = await BatteryPoolStatus.async_new(
+            battery_ids=set(self._bat_inv_map.keys()),
+            max_blocking_duration_sec=30.0,
+            max_data_age_sec=10.0,
+        )
 
         # Wait few seconds to get data from the channels created above.
         await asyncio.sleep(self._wait_for_data_sec)
+
         self._started.set()
         while True:
             request, user = await self._request_queue.get()
 
             try:
                 pairs_data: List[InvBatPair] = self._get_components_data(
-                    request.batteries
+                    battery_pool.get_working_batteries(request.batteries)
                 )
             except KeyError as err:
                 await user.channel.send(Error(request, str(err)))
@@ -294,14 +293,10 @@ async def run(self) -> None:
                 str(battery_distribution),
             )
 
-            tasks = await self._set_distributed_power(
+            failed_power, failed_batteries = await self._set_distributed_power(
                 api, distribution, request.request_timeout_sec
             )
 
-            failed_power, failed_batteries = self._parse_result(
-                tasks, distribution.distribution, request.request_timeout_sec
-            )
-
             if len(failed_batteries) > 0:
                 succeed_batteries = set(battery_distribution.keys()) - failed_batteries
                 response = PartialFailure(
@@ -320,15 +315,15 @@ async def run(self) -> None:
                     excess_power=distribution.remaining_power,
                 )
 
-            self._battery_pool.update_last_request_status(response)
+            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]]:
+    ) -> Tuple[int, Set[int]]:
         """Send distributed power to the inverters.
 
         Args:
@@ -337,7 +332,8 @@ async def _set_distributed_power(
             timeout_sec: How long wait for the response
 
         Returns:
-            Dict with finished or cancelled task for each inverter.
+            Tuple where first element is total failed power, and the second element
+            set of batteries that failed.
         """
         tasks = {
             inverter_id: asyncio.create_task(api.set_power(inverter_id, power))
@@ -351,7 +347,8 @@ async def _set_distributed_power(
         )
 
         await self._cancel_tasks(pending)
-        return tasks
+
+        return self._parse_result(tasks, distribution.distribution, timeout_sec)
 
     def _check_request(self, request: Request) -> Optional[Result]:
         """Check whether the given request if correct.
@@ -539,7 +536,7 @@ def _get_components_data(self, batteries: Set[int]) -> List[InvBatPair]:
         """
         pairs_data: List[InvBatPair] = []
 
-        for battery_id in self._battery_pool.get_working_batteries(batteries):
+        for battery_id in batteries:
             if battery_id not in self._battery_receivers:
                 raise KeyError(
                     f"No battery {battery_id}, "