Use generic ComponentStatusTracker in ComponentPoolStatusTracker

This allows us to completely decouple `ComponentPoolStatusTracker`
from anything specific to batteries.

And when a `ComponentPoolStatusTracker` instance is created, users can
pass a specialized class that implements `ComponentStatusTracker`,
like the `BatteryStatusTracker`.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

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

@@ -18,6 +18,7 @@
 # pylint: enable=no-name-in-module
 from frequenz.channels import Receiver, Sender
 from frequenz.channels.util import Timer, select, selected_from
+from typing_extensions import override
 
 from ..._internal._asyncio import cancel_and_await
 from ...microgrid import connection_manager
@@ -27,7 +28,12 @@
     ComponentData,
     InverterData,
 )
-from ._component_status import ComponentStatus, ComponentStatusEnum, SetPowerResult
+from ._component_status import (
+    ComponentStatus,
+    ComponentStatusEnum,
+    ComponentStatusTracker,
+    SetPowerResult,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -121,7 +127,7 @@ def is_blocked(self) -> bool:
         return self.blocked_until > datetime.now(tz=timezone.utc)
 
 
-class BatteryStatusTracker:
+class BatteryStatusTracker(ComponentStatusTracker):
     """Class for tracking if battery is working.
 
     Status updates are sent out only when there is a status change.
@@ -156,6 +162,7 @@ class BatteryStatusTracker:
     A working inverter in any other inverter state will be reported as failing.
     """
 
+    @override
     def __init__(  # pylint: disable=too-many-arguments
         self,
         component_id: int,

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

@@ -12,11 +12,11 @@
 from frequenz.channels.util import Merge
 
 from ..._internal._asyncio import cancel_and_await
-from ._battery_status_tracker import BatteryStatusTracker
 from ._component_status import (
     ComponentPoolStatus,
     ComponentStatus,
     ComponentStatusEnum,
+    ComponentStatusTracker,
     SetPowerResult,
 )
 
@@ -30,12 +30,13 @@ class ComponentPoolStatusTracker:
     components changes.
     """
 
-    def __init__(  # noqa: DOC502 (RuntimeError raised from BatteryStatusTracker)
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         component_ids: set[int],
         component_status_sender: Sender[ComponentPoolStatus],
         max_data_age_sec: float,
         max_blocking_duration_sec: float,
+        component_status_tracker_type: type[ComponentStatusTracker],
     ) -> None:
         """Create ComponentPoolStatusTracker instance.
 
@@ -49,15 +50,14 @@ def __init__(  # noqa: DOC502 (RuntimeError raised from BatteryStatusTracker)
                 sending data.
             max_blocking_duration_sec: This value tell what should be the maximum
                 timeout used for blocking failing component.
-
-        Raises:
-            RuntimeError: When managing batteries, if any battery has no adjacent
-                inverter.
+            component_status_tracker_type: component status tracker to use
+                for tracking the status of the components.
         """
         self._component_ids = component_ids
         self._max_data_age_sec = max_data_age_sec
         self._max_blocking_duration_sec = max_blocking_duration_sec
         self._component_status_sender = component_status_sender
+        self._component_status_tracker_type = component_status_tracker_type
 
         # At first no component is working, we will get notification when they start
         # working.
@@ -68,7 +68,7 @@ def __init__(  # noqa: DOC502 (RuntimeError raised from BatteryStatusTracker)
             "component_request_status"
         )
         self._set_power_result_sender = self._set_power_result_channel.new_sender()
-        self._component_status_trackers: list[BatteryStatusTracker] = []
+        self._component_status_trackers: list[ComponentStatusTracker] = []
         self._merged_status_receiver = self._make_merged_status_receiver()
 
         self._task = asyncio.create_task(self._run())
@@ -97,7 +97,7 @@ def _make_merged_status_receiver(
             channel: Broadcast[ComponentStatus] = Broadcast(
                 f"component_{component_id}_status"
             )
-            tracker = BatteryStatusTracker(
+            tracker = self._component_status_tracker_type(
                 component_id=component_id,
                 max_data_age_sec=self._max_data_age_sec,
                 max_blocking_duration_sec=self._max_blocking_duration_sec,

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

@@ -6,9 +6,12 @@
 
 import enum
 import typing
+from abc import ABC, abstractmethod
 from collections import abc
 from dataclasses import dataclass
 
+from frequenz.channels import Receiver, Sender
+
 
 @dataclass
 class ComponentPoolStatus:
@@ -72,3 +75,35 @@ class SetPowerResult:
 
     failed: typing.Iterable[int]
     """Component IDs for which the last set power command failed."""
+
+
+class ComponentStatusTracker(ABC):
+    """Interface for specialized component status trackers to implement."""
+
+    @abstractmethod
+    def __init__(  # pylint: disable=too-many-arguments
+        self,
+        component_id: int,
+        max_data_age_sec: float,
+        max_blocking_duration_sec: float,
+        status_sender: Sender[ComponentStatus],
+        set_power_result_receiver: Receiver[SetPowerResult],
+    ) -> None:
+        """Create class instance.
+
+        Args:
+            component_id: Id of this battery
+            max_data_age_sec: If component stopped sending data, then
+                this is the maximum time when its last message should be considered as
+                valid. After that time, component won't be used until it starts sending
+                data.
+            max_blocking_duration_sec: This value tell what should be the maximum
+                timeout used for blocking failing component.
+            status_sender: Channel to send status updates.
+            set_power_result_receiver: Channel to receive results of the requests to the
+                components.
+        """
+
+    @abstractmethod
+    async def stop(self) -> None:
+        """Stop tracking battery status."""

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

@@ -36,6 +36,7 @@
     ComponentCategory,
     InverterData,
 )
+from ._battery_status_tracker import BatteryStatusTracker
 from ._component_pool_status_tracker import ComponentPoolStatusTracker
 from ._component_status import ComponentPoolStatus
 from ._distribution_algorithm import (
@@ -223,6 +224,7 @@ def __init__(
             component_status_sender=battery_status_sender,
             max_blocking_duration_sec=30.0,
             max_data_age_sec=10.0,
+            component_status_tracker_type=BatteryStatusTracker,
         )
 
     def _get_bounds(

tests/actor/test_battery_pool_status.py

@@ -7,6 +7,9 @@
 from frequenz.channels import Broadcast
 from pytest_mock import MockerFixture
 
+from frequenz.sdk.actor.power_distributing._battery_status_tracker import (
+    BatteryStatusTracker,
+)
 from frequenz.sdk.actor.power_distributing._component_pool_status_tracker import (
     ComponentPoolStatusTracker,
 )
@@ -46,6 +49,7 @@ async def test_batteries_status(self, mocker: MockerFixture) -> None:
             component_status_sender=battery_status_channel.new_sender(),
             max_data_age_sec=5,
             max_blocking_duration_sec=30,
+            component_status_tracker_type=BatteryStatusTracker,
         )
         await asyncio.sleep(0.1)