Cleanup and modularization of Power Manager, Power Distributor and DataPipeline (#881)

Author: shsms

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -14,7 +14,8 @@
 from frequenz.channels.util import SkipMissedAndDrift, Timer, select, selected_from
 from typing_extensions import override
 
-from ...timeseries._base_types import PoolType, SystemBounds
+from ...microgrid.component import ComponentCategory
+from ...timeseries._base_types import SystemBounds
 from .._actor import Actor
 from .._channel_registry import ChannelRegistry
 from ._base_classes import Algorithm, BaseAlgorithm, Proposal, ReportRequest, _Report
@@ -31,7 +32,7 @@ class PowerManagingActor(Actor):
 
     def __init__(  # pylint: disable=too-many-arguments
         self,
-        pool_type: PoolType,
+        component_category: ComponentCategory,
         proposals_receiver: Receiver[Proposal],
         bounds_subscription_receiver: Receiver[ReportRequest],
         power_distributing_requests_sender: Sender[power_distributing.Request],
@@ -44,8 +45,8 @@ def __init__(  # pylint: disable=too-many-arguments
         """Create a new instance of the power manager.
 
         Args:
-            pool_type: The type of the component pool this power manager instance is
-                going to support.
+            component_category: The category of the component this power manager
+                instance is going to support.
             proposals_receiver: The receiver for proposals.
             bounds_subscription_receiver: The receiver for bounds subscriptions.
             power_distributing_requests_sender: The sender for power distribution
@@ -63,7 +64,7 @@ def __init__(  # pylint: disable=too-many-arguments
                 f"PowerManagingActor: Unknown algorithm: {algorithm}"
             )
 
-        self._pool_type = pool_type
+        self._component_category = component_category
         self._bounds_subscription_receiver = bounds_subscription_receiver
         self._power_distributing_requests_sender = power_distributing_requests_sender
         self._power_distributing_results_receiver = power_distributing_results_receiver
@@ -132,8 +133,11 @@ def _add_bounds_tracker(self, component_ids: frozenset[int]) -> None:
             microgrid,
         )
 
-        if self._pool_type is not PoolType.BATTERY_POOL:
-            err = f"PowerManagingActor: Unsupported pool type: {self._pool_type}"
+        if self._component_category is not ComponentCategory.BATTERY:
+            err = (
+                "PowerManagingActor: Unsupported component category: "
+                f"{self._component_category}"
+            )
             _logger.error(err)
             raise NotImplementedError(err)
         battery_pool = microgrid.battery_pool(component_ids)

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

@@ -12,6 +12,7 @@
 
 import grpc
 from frequenz.channels import Receiver, Sender
+from typing_extensions import override
 
 from .... import microgrid
 from ...._internal._channels import LatestValueCache
@@ -122,6 +123,7 @@ def _add(key: str, value: dict[int, set[int]] | None) -> None:
 class BatteryManager(ComponentManager):
     """Class to manage the data streams for batteries."""
 
+    @override
     def __init__(
         self,
         component_pool_status_sender: Sender[ComponentPoolStatus],
@@ -163,18 +165,22 @@ def __init__(
         )
         """The distribution algorithm used to distribute power between batteries."""
 
+    @override
     def component_ids(self) -> collections.abc.Set[int]:
         """Return the set of component ids."""
         return self._battery_ids
 
+    @override
     async def start(self) -> None:
         """Start the battery data manager."""
         await self._create_channels()
 
+    @override
     async def stop(self) -> None:
         """Stop the battery data manager."""
         await self._component_pool_status_tracker.stop()
 
+    @override
     async def distribute_power(self, request: Request) -> Result:
         """Distribute the requested power to the components.
 

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

@@ -1,7 +1,19 @@
 # License: MIT
 # Copyright © 2022 Frequenz Energy-as-a-Service GmbH
-"""Class to return battery status."""
 
+"""Background service that tracks the status of a battery.
+
+A battery is consider to be WORKING if both the battery and the adjacent inverter are
+sending data that shows that they are working.
+
+If either of them stops sending data, or if the data shows that they are not working,
+then the battery is considered to be NOT_WORKING.
+
+If a battery and its adjacent inverter are WORKING, but the last request to the battery
+failed, then the battery's status is considered to be UNCERTAIN. In this case, the
+battery is blocked for a short time, and it is not recommended to use it unless it is
+necessary.
+"""
 
 import asyncio
 import logging
@@ -28,6 +40,7 @@
     InverterData,
 )
 from ..._background_service import BackgroundService
+from ._blocking_status import BlockingStatus
 from ._component_status import (
     ComponentStatus,
     ComponentStatusEnum,
@@ -53,79 +66,6 @@ class _ComponentStreamStatus:
     """Flag whether last message was correct or not."""
 
 
-@dataclass
-class _BlockingStatus:
-    min_duration: timedelta
-    """The minimum blocking duration."""
-
-    max_duration: timedelta
-    """The maximum blocking duration."""
-
-    last_blocking_duration: timedelta = timedelta(seconds=0.0)
-    """Last blocking duration."""
-
-    blocked_until: datetime | None = None
-    """Until when the battery is blocked."""
-
-    def __post_init__(self) -> None:
-        assert self.min_duration <= self.max_duration, (
-            f"Minimum blocking duration ({self.min_duration}) cannot be greater "
-            f"than maximum blocking duration ({self.max_duration})"
-        )
-        self.last_blocking_duration = self.min_duration
-        self._timedelta_zero = timedelta(seconds=0.0)
-
-    def block(self) -> timedelta:
-        """Block battery.
-
-        Battery can be unblocked using `self.unblock()` method.
-
-        Returns:
-            The duration for which the battery is blocked.
-        """
-        now = datetime.now(tz=timezone.utc)
-
-        # If is not blocked
-        if self.blocked_until is None:
-            self.last_blocking_duration = self.min_duration
-            self.blocked_until = now + self.last_blocking_duration
-            return self.last_blocking_duration
-
-        # If still blocked, then do nothing
-        if self.blocked_until > now:
-            return self._timedelta_zero
-
-        # If previous blocking time expired, then blocked it once again.
-        # Increase last blocking time, unless it reach the maximum.
-        self.last_blocking_duration = min(
-            2 * self.last_blocking_duration, self.max_duration
-        )
-        self.blocked_until = now + self.last_blocking_duration
-
-        return self.last_blocking_duration
-
-    def unblock(self) -> None:
-        """Unblock battery.
-
-        This will reset duration of the next blocking timeout.
-
-        Battery can be blocked using `self.block()` method.
-        """
-        self.blocked_until = None
-
-    def is_blocked(self) -> bool:
-        """Return if battery is blocked.
-
-        Battery can be blocked if last request for that battery failed.
-
-        Returns:
-            True if battery is blocked, False otherwise.
-        """
-        if self.blocked_until is None:
-            return False
-        return self.blocked_until > datetime.now(tz=timezone.utc)
-
-
 class BatteryStatusTracker(ComponentStatusTracker, BackgroundService):
     """Class for tracking if battery is working.
 
@@ -194,8 +134,8 @@ def __init__(  # pylint: disable=too-many-arguments
         # First battery is considered as not working.
         # Change status after first messages are received.
         self._last_status: ComponentStatusEnum = ComponentStatusEnum.NOT_WORKING
-        self._blocking_status: _BlockingStatus = _BlockingStatus(
-            timedelta(seconds=1.0), max_blocking_duration
+        self._blocking_status: BlockingStatus = BlockingStatus(
+            min_duration=timedelta(seconds=1.0), max_duration=max_blocking_duration
         )
         self._timedelta_zero = timedelta(seconds=0.0)
 

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

@@ -0,0 +1,74 @@
+# License: MIT
+# Copyright © 2022 Frequenz Energy-as-a-Service GmbH
+
+"""Tracking of the blocking status of a component."""
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+
+@dataclass(kw_only=True)
+class BlockingStatus:
+    """Tracking of the blocking status of a component."""
+
+    min_duration: timedelta
+    """The minimum blocking duration."""
+
+    max_duration: timedelta
+    """The maximum blocking duration."""
+
+    last_blocking_duration: timedelta = timedelta(seconds=0.0)
+    """Last blocking duration."""
+
+    blocked_until: datetime | None = None
+    """Time until which the component is blocked."""
+
+    def __post_init__(self) -> None:
+        """Validate the blocking duration."""
+        assert self.min_duration <= self.max_duration, (
+            f"Minimum blocking duration ({self.min_duration}) cannot be greater "
+            f"than maximum blocking duration ({self.max_duration})"
+        )
+        self.last_blocking_duration = self.min_duration
+        self._timedelta_zero = timedelta(seconds=0.0)
+
+    def block(self) -> timedelta:
+        """Set the component as blocked.
+
+        Returns:
+            The duration for which the component is blocked.
+        """
+        now = datetime.now(tz=timezone.utc)
+
+        # If is not blocked
+        if self.blocked_until is None:
+            self.last_blocking_duration = self.min_duration
+            self.blocked_until = now + self.last_blocking_duration
+            return self.last_blocking_duration
+
+        # If still blocked, then do nothing
+        if self.blocked_until > now:
+            return self._timedelta_zero
+
+        # If previous blocking time expired, then blocked it once again.
+        # Increase last blocking time, unless it reach the maximum.
+        self.last_blocking_duration = min(
+            2 * self.last_blocking_duration, self.max_duration
+        )
+        self.blocked_until = now + self.last_blocking_duration
+
+        return self.last_blocking_duration
+
+    def unblock(self) -> None:
+        """Set the component as unblocked."""
+        self.blocked_until = None
+
+    def is_blocked(self) -> bool:
+        """Check if the component is blocked.
+
+        Returns:
+            True if battery is blocked, False otherwise.
+        """
+        if self.blocked_until is None:
+            return False
+        return self.blocked_until > datetime.now(tz=timezone.utc)