Use enum for the values returned by is_running()

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

src/frequenz/dispatch/__init__.py

@@ -3,7 +3,7 @@
 
 """A highlevel interface for the dispatch API."""
 
-from ._dispatch import Dispatch
+from ._dispatch import Dispatch, RunningState
 from ._dispatcher import Dispatcher, ReceiverFetcher
 from ._event import Created, Deleted, DispatchEvent, Updated
 
@@ -15,4 +15,5 @@
     "ReceiverFetcher",
     "Updated",
     "Dispatch",
+    "RunningState",
 ]

src/frequenz/dispatch/_dispatch.py

@@ -7,6 +7,7 @@
 import logging
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from enum import Enum
 from typing import Iterator, cast
 
 from dateutil import rrule
@@ -37,6 +38,19 @@
 """To map from our Weekday enum to the dateutil library enum."""
 
 
+class RunningState(Enum):
+    """The running state of a dispatch."""
+
+    RUNNING = "RUNNING"
+    """The dispatch is running."""
+
+    STOPPED = "STOPPED"
+    """The dispatch is stopped."""
+
+    DIFFERENT_TYPE = "DIFFERENT_TYPE"
+    """The dispatch is for a different type."""
+
+
 @dataclass(frozen=True)
 class Dispatch(BaseDispatch):
     """Dispatch type with extra functionality."""
@@ -85,26 +99,28 @@ def set_running_status_notified(self) -> None:
         """Mark the latest running state change notification as sent."""
         object.__setattr__(self, "running_state_change_synced", self.update_time)
 
-    def running(self, type_: str) -> bool:
+    def running(self, type_: str) -> RunningState:
         """Check if the dispatch is currently supposed to be running.
 
         Args:
             type_: The type of the dispatch that should be running.
 
         Returns:
-            True if the dispatch is currently meant to be running, False otherwise.
+            RUNNING if the dispatch is running,
+            STOPPED if it is stopped,
+            DIFFERENT_TYPE if it is for a different type.
         """
         if self.type != type_:
-            return False
+            return RunningState.DIFFERENT_TYPE
 
         if not self.active or self.deleted:
-            return False
+            return RunningState.STOPPED
 
         now = datetime.now(tz=timezone.utc)
         if until := self._until(now):
-            return now < until
+            return RunningState.RUNNING if now < until else RunningState.STOPPED
 
-        return False
+        return RunningState.STOPPED
 
     @property
     def until(self) -> datetime | None:

src/frequenz/dispatch/_dispatcher.py

@@ -50,46 +50,51 @@ class Dispatcher:
     reconfigure itself.
 
     Example: Processing running state change dispatches
-        ```python
-        import os
-        import grpc.aio
-        from unittest.mock import MagicMock
 
-        async def run():
-            host = os.getenv("DISPATCH_API_HOST", "localhost")
-            port = os.getenv("DISPATCH_API_PORT", "50051")
+    ```python
+    import os
+    import grpc.aio
+    from unittest.mock import MagicMock
 
-            service_address = f"{host}:{port}"
-            grpc_channel = grpc.aio.insecure_channel(service_address)
-            microgrid_id = 1
-            dispatcher = Dispatcher(microgrid_id, grpc_channel, service_address)
-            await dispatcher.start()
+    async def run():
+        host = os.getenv("DISPATCH_API_HOST", "localhost")
+        port = os.getenv("DISPATCH_API_PORT", "50051")
 
-            actor = MagicMock() # replace with your actor
+        service_address = f"{host}:{port}"
+        grpc_channel = grpc.aio.insecure_channel(service_address)
+        microgrid_id = 1
+        dispatcher = Dispatcher(microgrid_id, grpc_channel, service_address)
+        await dispatcher.start()
 
-            changed_running_status_rx = dispatcher.running_status_change.new_receiver()
+        actor = MagicMock() # replace with your actor
 
-            async for dispatch in changed_running_status_rx:
-                if dispatch.type != "DEMO_TYPE":
-                    continue
+        changed_running_status_rx = dispatcher.running_status_change.new_receiver()
 
-                print(f"Executing dispatch {dispatch.id}, due on {dispatch.start_time}")
-                if dispatch.running:
+        async for dispatch in changed_running_status_rx:
+            match dispatch.running("DEMO_TYPE"):
+                case RunningState.RUNNING:
+                    print(f"Executing dispatch {dispatch.id}, due on {dispatch.start_time}")
                     if actor.is_running:
                         actor.reconfigure(
+                            components=dispatch.selector,
+                            run_parameters=dispatch.payload, # custom actor parameters
+                            dry_run=dispatch.dry_run,
+                            until=dispatch.until,
                         )  # this will reconfigure the actor
                     else:
-                        # this will start a new or reconfigure a running actor
+                        # this will start a new actor with the given components
                         # and run it for the duration of the dispatch
-                        actor.start_or_reconfigure(
+                        actor.start(
                             components=dispatch.selector,
                             run_parameters=dispatch.payload, # custom actor parameters
                             dry_run=dispatch.dry_run,
                             until=dispatch.until,
                         )
-                else:
+                case RunningState.STOPPED:
                     actor.stop()  # this will stop the actor
-        ```
+                case RunningState.DIFFERENT_TYPE:
+                    pass  # dispatch not for this type
+    ```
 
     Example: Getting notification about dispatch lifecycle events
         ```python

src/frequenz/dispatch/actor.py

@@ -13,7 +13,7 @@
 from frequenz.client.dispatch import Client
 from frequenz.sdk.actor import Actor
 
-from ._dispatch import Dispatch
+from ._dispatch import Dispatch, RunningState
 from ._event import Created, Deleted, DispatchEvent, Updated
 
 _MAX_AHEAD_SCHEDULE = timedelta(hours=5)
@@ -220,7 +220,10 @@ def _running_state_change(
         # Deleted dispatch
         if updated_dispatch is None:
             assert previous_dispatch is not None
-            return previous_dispatch.running(previous_dispatch.type)
+            return (
+                previous_dispatch.running(previous_dispatch.type)
+                == RunningState.RUNNING
+            )
 
         # If any of the runtime attributes changed, we need to send a message
         runtime_state_attributes = [