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/_dispatch.py

@@ -0,0 +1,182 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Dispatch type with support for next_run calculation."""
+
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Generator, cast
+
+from dateutil import rrule
+from frequenz.client.dispatch.types import Dispatch as BaseDispatch
+from frequenz.client.dispatch.types import Frequency, Weekday
+
+_logger = logging.getLogger(__name__)
+"""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."""
+
+
+@dataclass
+class Dispatch(BaseDispatch):
+    """Dispatch type with extra functionality."""
+
+    deleted: bool = False
+    """Whether the dispatch is deleted."""
+
+    last_sent_running_state_change_notification: datetime | None = None
+    """The last time a message was sent about the running state change."""
+
+    def __init__(self, **kwargs):
+        """Initialize the dispatch.
+
+        Args:
+            **kwargs: The dispatch attributes to set.
+        """
+        super().__init__(**kwargs)
+
+        self.deleted = kwargs.get("deleted", False)
+
+    @property
+    def running(self) -> bool:
+        """Check if the dispatch is currently supposed to be running.
+
+        Returns:
+            True if the dispatch is currently meant to be running, False otherwise.
+        """
+        if not self.active or self.deleted:
+            return False
+
+        now = datetime.now(tz=timezone.utc)
+        if until := self._until(now):
+            return now < until
+
+        return False
+
+    @property
+    def until(self) -> datetime | None:
+        """Time when the dispatch should end.
+
+        Returns the time that a running dispatch should end.
+        If the dispatch is not running, None is returned.
+
+        Returns:
+            The time when the dispatch should end or None if the dispatch is not running.
+        """
+        if not self.active or self.deleted:
+            return None
+
+        now = datetime.now(tz=timezone.utc)
+        return self._until(now)
+
+    def missed_runs(self) -> Generator[datetime, None, None]:
+        """Yield all missed runs of a dispatch.
+
+        Yields all missed runs of a dispatch.
+
+        If a running state change notification was not sent in time
+        due to connection issues, this method will yield all missed runs
+        since the last sent notification.
+
+        Returns:
+            A generator that yields all missed runs of a dispatch.
+        """
+        if self.update_time == self.last_sent_running_state_change_notification:
+            return
+
+        next_run = self.update_time
+        now = datetime.now(tz=timezone.utc)
+
+        while (next_run := self.next_run(next_run)) and next_run < now:
+            yield next_run
+
+    def next_run(self, _from: datetime) -> datetime | None:
+        """Calculate the next run of a dispatch.
+
+        Args:
+            _from: The time to calculate the next run from.
+
+        Returns:
+            The next run of the dispatch or None if the dispatch is finished.
+        """
+        if (
+            not self.recurrence.frequency
+            or self.recurrence.frequency == Frequency.UNSPECIFIED
+        ):
+            if _from > self.start_time:
+                return None
+            return self.start_time
+
+        # Make sure no weekday is UNSPECIFIED
+        if Weekday.UNSPECIFIED in self.recurrence.byweekdays:
+            _logger.warning("Dispatch %s has UNSPECIFIED weekday, ignoring...", self.id)
+            return None
+
+        # No type information for rrule, so we need to cast
+        return cast(datetime | None, self._prepare_rrule().after(_from, inc=True))
+
+    def _prepare_rrule(self) -> rrule.rrule:
+        """Prepare the rrule object.
+
+        Returns:
+            The rrule object.
+        """
+        count, until = (None, None)
+        if end := self.recurrence.end_criteria:
+            count = end.count
+            until = end.until
+
+        rrule_obj = rrule.rrule(
+            freq=_RRULE_FREQ_MAP[self.recurrence.frequency],
+            dtstart=self.start_time,
+            count=count,
+            until=until,
+            byminute=self.recurrence.byminutes,
+            byhour=self.recurrence.byhours,
+            byweekday=[
+                _RRULE_WEEKDAY_MAP[weekday] for weekday in self.recurrence.byweekdays
+            ],
+            bymonthday=self.recurrence.bymonthdays,
+            bymonth=self.recurrence.bymonths,
+            interval=self.recurrence.interval,
+        )
+
+        return rrule_obj
+
+    def _until(self, now: datetime) -> datetime | None:
+        """Calculate the time when the dispatch should end.
+
+        If no previous run is found, None is returned.
+
+        Args:
+            now: The current time.
+
+        Returns:
+            The time when the dispatch should end or None if the dispatch is not running.
+        """
+        latest_past_start = self._prepare_rrule().before(now, inc=True)
+
+        if not latest_past_start:
+            return None
+
+        return latest_past_start + self.duration

src/frequenz/dispatch/_dispatcher.py

@@ -8,10 +8,10 @@
 
 import grpc.aio
 from frequenz.channels import Broadcast, Receiver
-from frequenz.client.dispatch.types import Dispatch
 
-from frequenz.dispatch._event import DispatchEvent
-from frequenz.dispatch.actor import DispatchingActor
+from ._dispatch import Dispatch
+from ._event import DispatchEvent
+from .actor import DispatchingActor
 
 ReceivedT = TypeVar("ReceivedT")
 """The type being received."""
@@ -48,7 +48,7 @@ class Dispatcher:
 
     allows to receive new dispatches and ready dispatches.
 
-    Example: Processing ready-to-execute dispatches
+    Example: Processing running state change dispatches
         ```python
         import grpc.aio
 
@@ -58,13 +58,18 @@ async def run():
             service_address = "localhost:50051"
 
             dispatcher = Dispatcher(microgrid_id, grpc_channel, service_address)
-            dispatcher.start()  # this will start the actor
 
-            ready_receiver = dispatcher.ready_to_execute.new_receiver()
+            changed_running_status_rx= dispatcher.running_status_change.new_receiver()
 
-            async for dispatch in ready_receiver:
+            async for dispatch in changed_running_status_rx:
                 print(f"Executing dispatch {dispatch.id}, due on {dispatch.start_time}")
-                # execute the dispatch
+                if dispatch.running:
+                    # this will start the actor
+                    # and run it for the duration of the dispatch
+                    dispatcher.start(duration=dispatch.duration, dry_run=dispatch.dry_run)
+                else:
+                    dispatcher.stop()  # this will stop the actor
+
         ```
 
     Example: Getting notification about dispatch lifecycle events
@@ -107,14 +112,14 @@ def __init__(
             grpc_channel: The gRPC channel.
             svc_addr: The service address.
         """
-        self._ready_channel = Broadcast[Dispatch]("ready_dispatches")
-        self._updated_channel = Broadcast[DispatchEvent]("new_dispatches")
+        self._running_state_channel = Broadcast[Dispatch]("ready_dispatches")
+        self._lifecycle_events_channel = Broadcast[DispatchEvent]("new_dispatches")
         self._actor = DispatchingActor(
             microgrid_id,
             grpc_channel,
             svc_addr,
-            self._updated_channel.new_sender(),
-            self._ready_channel.new_sender(),
+            self._lifecycle_events_channel.new_sender(),
+            self._running_state_channel.new_sender(),
         )
 
     async def start(self) -> None:
@@ -123,18 +128,46 @@ async def start(self) -> None:
 
     @property
     def lifecycle_events(self) -> ReceiverFetcher[DispatchEvent]:
-        """Return new, updated or deleted dispatches receiver.
+        """Return new, updated or deleted dispatches receiver fetcher.
 
         Returns:
             A new receiver for new dispatches.
         """
-        return self._updated_channel
+        return self._lifecycle_events_channel
 
     @property
-    def ready_to_execute(self) -> ReceiverFetcher[Dispatch]:
-        """Return ready dispatches receiver.
+    def running_status_change(self) -> ReceiverFetcher[Dispatch]:
+        """Return running status change receiver fetcher.
+
+        This receiver will receive a message whenever the current running
+        status of a dispatch changes.
+
+        Usually, one message per scheduled run is to be expected.
+        However, things get complicated when a dispatch was modified:
+
+        If it was currently running and the modification now says
+        it should not be running or running with different parameters,
+        then a message will be sent.
+
+        In other words: Any change that is expected to make an actor start, stop
+        or reconfigure itself with new parameters causes a message to be
+        sent.
+
+        A non-exhaustive list of possible changes that will cause a message to be sent:
+         - The normal scheduled start_time has been reached
+         - The duration of the dispatch has been modified
+         - The start_time has been modified to be in the future
+         - The component selection changed
+         - The active status changed
+         - The dry_run status changed
+         - The payload changed
+         - The dispatch was deleted
+
+        Note: Reaching the end time (start_time + duration) will not
+        send a message, except when it was reached by modifying the duration.
+
 
         Returns:
-            A new receiver for ready dispatches.
+            A new receiver for dispatches whose running status changed.
         """
-        return self._ready_channel
+        return self._running_state_channel