Dispatch python interface change suggestion

Still a WIP, but already provided here for early feedback.

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

Author: Marenz

src/frequenz/dispatch/_dispatcher.py

@@ -10,7 +10,7 @@
 from frequenz.channels import Broadcast, Receiver
 from frequenz.client.dispatch.types import Dispatch
 
-from frequenz.dispatch._event import DispatchEvent
+from frequenz.dispatch._event import DispatchCommandRequest, DispatchEvent
 from frequenz.dispatch.actor import DispatchingActor
 
 ReceivedT = TypeVar("ReceivedT")
@@ -107,7 +107,7 @@ def __init__(
             grpc_channel: The gRPC channel.
             svc_addr: The service address.
         """
-        self._ready_channel = Broadcast[Dispatch]("ready_dispatches")
+        self._ready_channel = Broadcast[DispatchCommandRequest]("ready_dispatches")
         self._updated_channel = Broadcast[DispatchEvent]("new_dispatches")
         self._actor = DispatchingActor(
             microgrid_id,

src/frequenz/dispatch/_event.py

@@ -38,3 +38,57 @@ class Deleted:
 This type is used to send dispatches that were created, updated or deleted
 over the channel.
 """
+
+
+@dataclass(frozen=True)
+class Start:
+    """A request to start a dispatch.
+
+    Sent when a dispatch starts as scheduled or was
+    modified to be active after.
+
+    Possible modifications that could lead to this:
+     * Change of start_time to one now() - 5 seconds
+     * Change of active to True
+     * Change of duration to be longer
+    """
+
+    dispatch: Dispatch
+    """The dispatch that should start."""
+
+
+@dataclass(frozen=False)
+class Stop:
+    """A request to stop a running dispatch.
+
+    Sent when a dispatch naturally ends according to its duration,
+    but also when it was deleted or modified to be no longer active.
+    """
+
+    dispatch: Dispatch
+    """The dispatch that is being stopped."""
+
+
+@dataclass(frozen=True)
+class DryRunStart:
+    """A request to start a dispatch in dry run mode."""
+
+    dispatch: Dispatch
+    """The dispatch that should start in dry run mode."""
+
+
+@dataclass(frozen=True)
+class DryRunStop:
+    """A request to stop a running dispatch in dry run mode."""
+
+    dispatch: Dispatch
+    """The dispatch that is being stopped in dry run mode."""
+
+
+DispatchCommandRequest = Start | Stop | DryRunStart | DryRunStop
+"""Type that is sent over the channel for dispatch command requests.
+
+This is sent whenever a dispatch is ready to run or finished running.
+Other reasons include the deletion of dispatches as well as modification
+updates to the start time, duration or active flag, among others.
+"""

src/frequenz/dispatch/actor.py

@@ -16,7 +16,17 @@
 from frequenz.client.dispatch.types import Dispatch, Frequency, Weekday
 from frequenz.sdk.actor import Actor
 
-from frequenz.dispatch._event import Created, Deleted, DispatchEvent, Updated
+from frequenz.dispatch._event import (
+    Created,
+    Deleted,
+    DispatchCommandRequest,
+    DispatchEvent,
+    DryRunStart,
+    DryRunStop,
+    Start,
+    Stop,
+    Updated,
+)
 
 _MAX_AHEAD_SCHEDULE = timedelta(hours=5)
 """The maximum time ahead to schedule a dispatch.
@@ -36,27 +46,6 @@
 """The logger for this module."""
 
 
-_RRULE_FREQ_MAP = {
-    Frequency.MINUTELY: rrule.MINUTELY,
-    Frequency.HOURLY: rrule.HOURLY,
-    Frequency.DAILY: rrule.DAILY,
-    Frequency.WEEKLY: rrule.WEEKLY,
-    Frequency.MONTHLY: rrule.MONTHLY,
-}
-"""To map from our Frequency enum to the dateutil library enum."""
-
-_RRULE_WEEKDAY_MAP = {
-    Weekday.MONDAY: rrule.MO,
-    Weekday.TUESDAY: rrule.TU,
-    Weekday.WEDNESDAY: rrule.WE,
-    Weekday.THURSDAY: rrule.TH,
-    Weekday.FRIDAY: rrule.FR,
-    Weekday.SATURDAY: rrule.SA,
-    Weekday.SUNDAY: rrule.SU,
-}
-"""To map from our Weekday enum to the dateutil library enum."""
-
-
 class DispatchingActor(Actor):
     """Dispatch actor.
 
@@ -73,7 +62,7 @@ def __init__(
         grpc_channel: grpc.aio.Channel,
         svc_addr: str,
         updated_dispatch_sender: Sender[DispatchEvent],
-        ready_dispatch_sender: Sender[Dispatch],
+        dispatch_cmd_sender: Sender[DispatchCommandRequest],
         poll_interval: timedelta = _DEFAULT_POLL_INTERVAL,
     ) -> None:
         """Initialize the actor.
@@ -83,7 +72,7 @@ def __init__(
             grpc_channel: The gRPC channel to use for communication with the API.
             svc_addr: Address of the service to connect to.
             updated_dispatch_sender: A sender for new or updated dispatches.
-            ready_dispatch_sender: A sender for ready dispatches.
+            dispatch_cmd_sender: A sender for start/stop dispatch requests
             poll_interval: The interval to poll the API for dispatche changes.
         """
         super().__init__(name="dispatch")
@@ -93,7 +82,7 @@ def __init__(
         self._scheduled: dict[int, asyncio.Task[None]] = {}
         self._microgrid_id = microgrid_id
         self._updated_dispatch_sender = updated_dispatch_sender
-        self._ready_dispatch_sender = ready_dispatch_sender
+        self._cmd_sender = dispatch_cmd_sender
         self._poll_timer = Timer.timeout(poll_interval)
 
     async def _run(self) -> None:
@@ -196,7 +185,7 @@ def next_run_info() -> tuple[datetime, datetime] | None:
             await asyncio.sleep((next_time - now).total_seconds())
 
             _logger.info("Dispatch ready: %s", dispatch)
-            await self._ready_dispatch_sender.send(dispatch)
+            await self._cmd_sender.send(Start(dispatch))
 
         _logger.info("Dispatch finished: %s", dispatch)
         self._scheduled.pop(dispatch.id)

tests/test_frequenz_dispatch.py

@@ -88,7 +88,7 @@ async def send(self, msg: T) -> None:
         grpc_channel=MagicMock(),
         svc_addr="localhost",
         updated_dispatch_sender=YieldingSender(updated_dispatches),
-        ready_dispatch_sender=YieldingSender(ready_dispatches),
+        dispatch_cmd_sender=YieldingSender(ready_dispatches),
     )
 
     client = FakeClient()