Dispatch python interface changes

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

Author: Marenz

RELEASE_NOTES.md

@@ -6,11 +6,11 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+* `Dispatcher.ready_to_execute()` was renamed to `Dispatcher.running_status_change()`
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* Introduced new class `Dispatch` (based on the client class) that contains useful functions and extended information about the received dispatch.
 
 ## Bug Fixes
 

src/frequenz/dispatch/__init__.py

@@ -6,11 +6,14 @@
 from frequenz.dispatch._dispatcher import Dispatcher, ReceiverFetcher
 from frequenz.dispatch._event import Created, Deleted, DispatchEvent, Updated
 
+from ._dispatch import Dispatch
+
 __all__ = [
     "Created",
     "Deleted",
     "DispatchEvent",
     "Dispatcher",
     "ReceiverFetcher",
     "Updated",
+    "Dispatch",
 ]

src/frequenz/dispatch/_dispatch.py

@@ -0,0 +1,226 @@
+# 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 Iterator, 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(frozen=True)
+class Dispatch(BaseDispatch):
+    """Dispatch type with extra functionality."""
+
+    deleted: bool = False
+    """Whether the dispatch is deleted."""
+
+    running_state_change_synced: datetime | None = None
+    """The last time a message was sent about the running state change."""
+
+    def __init__(
+        self,
+        client_dispatch: BaseDispatch,
+        deleted: bool = False,
+        running_state_change_synced: datetime | None = None,
+    ):
+        """Initialize the dispatch.
+
+        Args:
+            client_dispatch: The client dispatch.
+            deleted: Whether the dispatch is deleted.
+            running_state_change_synced: Timestamp of the last running state change message.
+        """
+        super().__init__(**client_dispatch.__dict__)
+        # Work around frozen to set deleted
+        object.__setattr__(self, "deleted", deleted)
+        object.__setattr__(
+            self,
+            "running_state_change_synced",
+            running_state_change_synced,
+        )
+
+    def set_deleted(self) -> None:
+        """Mark the dispatch as deleted."""
+        object.__setattr__(self, "deleted", True)
+
+    def running_status_notified(self) -> bool:
+        """Check that the latest running state change notification was sent.
+
+        Returns:
+            True if the latest running state change notification was sent, False otherwise.
+        """
+        return self.running_state_change_synced == self.update_time
+
+    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:
+        """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.
+        """
+        if self.type != type_:
+            return False
+
+        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)
+
+    @property
+    def missed_runs(self) -> Iterator[datetime]:  # noqa: DOC405
+        """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.running_state_change_synced:
+            return
+
+        from_time = self.update_time
+        now = datetime.now(tz=timezone.utc)
+
+        while (next_run := self.next_run_after(from_time)) and next_run < now:
+            yield next_run
+            from_time = next_run
+
+    @property
+    def next_run(self) -> datetime | None:
+        """Calculate the next run of a dispatch.
+
+        Returns:
+            The next run of the dispatch or None if the dispatch is finished.
+        """
+        return self.next_run_after(datetime.now(tz=timezone.utc))
+
+    def next_run_after(self, after: datetime) -> datetime | None:
+        """Calculate the next run of a dispatch.
+
+        Args:
+            after: 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 after > 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(after, 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: datetime | None = 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_co = TypeVar("ReceivedT_co", covariant=True)
 """The type being received."""
@@ -44,27 +44,39 @@ class Dispatcher:
     One that sends a dispatch event message whenever a dispatch is created, updated or deleted.
 
     The other sends a dispatch message whenever a dispatch is ready to be
-    executed according to the schedule.
+    executed according to the schedule or the running status of the dispatch
+    changed in a way that could potentially require the actor to start, stop or
+    reconfigure itself.
 
-    allows to receive new dispatches and ready dispatches.
-
-    Example: Processing ready-to-execute dispatches
+    Example: Processing running state change dispatches
         ```python
         import grpc.aio
+        from unittest.mock import MagicMock
 
         async def run():
             grpc_channel = grpc.aio.insecure_channel("localhost:50051")
             microgrid_id = 1
             service_address = "localhost:50051"
 
             dispatcher = Dispatcher(microgrid_id, grpc_channel, service_address)
-            dispatcher.start()  # this will start the actor
+            actor = MagicMock() # replace with your 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:
+                    if actor.is_running:
+                        actor.reconfigure(
+                            components=dispatch.selector,
+                            run_parameters=dispatch.payload
+                        )  # this will reconfigure the actor
+                    else:
+                        # this will start the actor
+                        # and run it for the duration of the dispatch
+                        actor.start(duration=dispatch.duration, dry_run=dispatch.dry_run)
+                else:
+                    actor.stop()  # this will stop the actor
         ```
 
     Example: Getting notification about dispatch lifecycle events
@@ -107,14 +119,16 @@ def __init__(
             grpc_channel: The gRPC channel.
             svc_addr: The service address.
         """
-        self._ready_channel = Broadcast[Dispatch](name="ready_dispatches")
-        self._updated_channel = Broadcast[DispatchEvent](name="new_dispatches")
+        self._running_state_channel = Broadcast[Dispatch](name="running_state_change")
+        self._lifecycle_events_channel = Broadcast[DispatchEvent](
+            name="lifecycle_events"
+        )
         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 +137,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