Rename Status → ComponentStatusEnum

And move it to the `_component_status` module.

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

Author: shsms

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

@@ -9,7 +9,6 @@
 from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from enum import Enum
 
 # pylint: disable=no-name-in-module
 from frequenz.api.microgrid.battery_pb2 import ComponentState as BatteryComponentState
@@ -29,25 +28,11 @@
     ComponentData,
     InverterData,
 )
+from ._component_status import ComponentStatusEnum
 
 _logger = logging.getLogger(__name__)
 
 
-class Status(Enum):
-    """Tells if battery is can be used."""
-
-    NOT_WORKING = 0
-    """Component is not working and should not be used"""
-
-    UNCERTAIN = 1
-    """Component should work, but it failed in last request. It is blocked for few
-    seconds and it is not recommended to use it unless it is necessary.
-    """
-
-    WORKING = 2
-    """Component is working"""
-
-
 @dataclass
 class SetPowerResult:
     """Information what batteries succeed or failed the last request."""
@@ -188,7 +173,7 @@ def __init__(  # pylint: disable=too-many-arguments
         battery_id: int,
         max_data_age_sec: float,
         max_blocking_duration_sec: float,
-        status_sender: Sender[Status],
+        status_sender: Sender[ComponentStatusEnum],
         set_power_result_receiver: Receiver[SetPowerResult],
     ) -> None:
         """Create class instance.
@@ -211,7 +196,7 @@ def __init__(  # pylint: disable=too-many-arguments
         self._max_data_age = max_data_age_sec
         # First battery is considered as not working.
         # Change status after first messages are received.
-        self._last_status: Status = Status.NOT_WORKING
+        self._last_status: ComponentStatusEnum = ComponentStatusEnum.NOT_WORKING
         self._blocking_status: _BlockingStatus = _BlockingStatus(
             1.0, max_blocking_duration_sec
         )
@@ -272,7 +257,8 @@ def _handle_status_set_power_result(self, result: SetPowerResult) -> None:
             self._blocking_status.unblock()
 
         elif (
-            self.battery_id in result.failed and self._last_status != Status.NOT_WORKING
+            self.battery_id in result.failed
+            and self._last_status != ComponentStatusEnum.NOT_WORKING
         ):
             duration = self._blocking_status.block()
 
@@ -301,7 +287,7 @@ def _handle_status_inverter_timer(self) -> None:
                 self._inverter.last_msg_timestamp,
             )
 
-    def _get_new_status_if_changed(self) -> Status | None:
+    def _get_new_status_if_changed(self) -> ComponentStatusEnum | None:
         current_status = self._get_current_status()
         if self._last_status != current_status:
             self._last_status = current_status
@@ -315,7 +301,7 @@ def _get_new_status_if_changed(self) -> Status | None:
 
     async def _run(
         self,
-        status_sender: Sender[Status],
+        status_sender: Sender[ComponentStatusEnum],
         set_power_result_receiver: Receiver[SetPowerResult],
     ) -> None:
         """Process data from the components and set_power_result_receiver.
@@ -391,7 +377,7 @@ async def _run(
             except Exception as err:  # pylint: disable=broad-except
                 _logger.exception("BatteryStatusTracker crashed with error: %s", err)
 
-    def _get_current_status(self) -> Status:
+    def _get_current_status(self) -> ComponentStatusEnum:
         """Get current battery status.
 
         Returns:
@@ -402,15 +388,15 @@ def _get_current_status(self) -> Status:
         )
 
         if not is_msg_correct:
-            return Status.NOT_WORKING
-        if self._last_status == Status.NOT_WORKING:
+            return ComponentStatusEnum.NOT_WORKING
+        if self._last_status == ComponentStatusEnum.NOT_WORKING:
             # If message just become correct, then try to use it
             self._blocking_status.unblock()
-            return Status.WORKING
+            return ComponentStatusEnum.WORKING
         if self._blocking_status.is_blocked():
-            return Status.UNCERTAIN
+            return ComponentStatusEnum.UNCERTAIN
 
-        return Status.WORKING
+        return ComponentStatusEnum.WORKING
 
     def _is_capacity_present(self, msg: BatteryData) -> bool:
         """Check whether the battery capacity is NaN or not.
@@ -424,7 +410,7 @@ def _is_capacity_present(self, msg: BatteryData) -> bool:
             True if battery capacity is present, false otherwise.
         """
         if math.isnan(msg.capacity):
-            if self._last_status == Status.WORKING:
+            if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
                     "Battery %d capacity is NaN",
                     msg.component_id,
@@ -445,7 +431,7 @@ def _no_critical_error(self, msg: BatteryData | InverterData) -> bool:
         # pylint: disable=protected-access
         critical_err = next((err for err in msg._errors if err.level == critical), None)
         if critical_err is not None:
-            if self._last_status == Status.WORKING:
+            if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
                     "Component %d has critical error: %s",
                     msg.component_id,
@@ -467,7 +453,7 @@ def _is_inverter_state_correct(self, msg: InverterData) -> bool:
         # pylint: disable=protected-access
         state = msg._component_state
         if state not in BatteryStatusTracker._inverter_valid_state:
-            if self._last_status == Status.WORKING:
+            if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
                     "Inverter %d has invalid state: %s",
                     msg.component_id,
@@ -489,7 +475,7 @@ def _is_battery_state_correct(self, msg: BatteryData) -> bool:
         # pylint: disable=protected-access
         state = msg._component_state
         if state not in BatteryStatusTracker._battery_valid_state:
-            if self._last_status == Status.WORKING:
+            if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
                     "Battery %d has invalid state: %s",
                     self.battery_id,
@@ -501,7 +487,7 @@ def _is_battery_state_correct(self, msg: BatteryData) -> bool:
         # pylint: disable=protected-access
         relay_state = msg._relay_state
         if relay_state not in BatteryStatusTracker._battery_valid_relay:
-            if self._last_status == Status.WORKING:
+            if self._last_status == ComponentStatusEnum.WORKING:
                 _logger.warning(
                     "Battery %d has invalid relay state: %s",
                     self.battery_id,
@@ -534,7 +520,7 @@ def _is_message_reliable(self, message: ComponentData) -> bool:
         """
         is_outdated = self._is_timestamp_outdated(message.timestamp)
 
-        if is_outdated and self._last_status == Status.WORKING:
+        if is_outdated and self._last_status == ComponentStatusEnum.WORKING:
             _logger.warning(
                 "Component %d stopped sending data. Last timestamp: %s.",
                 message.component_id,

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

@@ -13,8 +13,8 @@
 from frequenz.channels.util import MergeNamed
 
 from ..._internal._asyncio import cancel_and_await
-from ._battery_status import BatteryStatusTracker, SetPowerResult, Status
-from ._component_status import ComponentPoolStatus
+from ._battery_status import BatteryStatusTracker, SetPowerResult
+from ._component_status import ComponentPoolStatus, ComponentStatusEnum
 
 _logger = logging.getLogger(__name__)
 
@@ -32,7 +32,7 @@ class _ComponentStatusChannelHelper:
 
     def __post_init__(self) -> None:
         self.name: str = f"component-{self.component_id}-status"
-        channel = Broadcast[Status](self.name)
+        channel = Broadcast[ComponentStatusEnum](self.name)
 
         receiver_name = f"{self.name}-receiver"
         self.receiver = channel.new_receiver(name=receiver_name, maxsize=1)
@@ -82,7 +82,7 @@ def __init__(  # noqa: DOC502 (RuntimeError raised from BatteryStatusTracker)
 
         # Receivers for individual components statuses are needed to create a
         # `MergeNamed` object.
-        receivers: dict[str, Receiver[Status]] = {}
+        receivers: dict[str, Receiver[ComponentStatusEnum]] = {}
 
         for battery_id in component_ids:
             channel = _ComponentStatusChannelHelper(battery_id)
@@ -98,7 +98,7 @@ def __init__(  # noqa: DOC502 (RuntimeError raised from BatteryStatusTracker)
                 ),
             )
 
-        self._component_status_channel = MergeNamed[Status](
+        self._component_status_channel = MergeNamed[ComponentStatusEnum](
             **receivers,
         )
 
@@ -148,13 +148,13 @@ async def _update_status(
         """
         async for channel_name, status in self._component_status_channel:
             component_id = self._batteries[channel_name].battery_id
-            if status == Status.WORKING:
+            if status == ComponentStatusEnum.WORKING:
                 self._current_status.working.add(component_id)
                 self._current_status.uncertain.discard(component_id)
-            elif status == Status.UNCERTAIN:
+            elif status == ComponentStatusEnum.UNCERTAIN:
                 self._current_status.working.discard(component_id)
                 self._current_status.uncertain.add(component_id)
-            elif status == Status.NOT_WORKING:
+            elif status == ComponentStatusEnum.NOT_WORKING:
                 self._current_status.working.discard(component_id)
                 self._current_status.uncertain.discard(component_id)
 

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

@@ -4,6 +4,7 @@
 """Classes to track the status of components in the microgrid."""
 
 
+import enum
 from collections import abc
 from dataclasses import dataclass
 
@@ -31,3 +32,20 @@ def get_working_components(self, components: abc.Set[int]) -> set[int]:
         if len(working) > 0:
             return working
         return self.uncertain.intersection(components)
+
+
+class ComponentStatusEnum(enum.Enum):
+    """Enum for component status."""
+
+    NOT_WORKING = 0
+    """Component is not working and should not be used."""
+
+    UNCERTAIN = 1
+    """Component should work, although the last request to it failed.
+
+    It is blocked for few seconds and it is not recommended to use it unless it is
+    necessary.
+    """
+
+    WORKING = 2
+    """Component is working"""