Add initial structure

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

Author: Marenz

pyproject.toml

@@ -170,7 +170,11 @@ packages = ["frequenz.dispatch"]
 strict = true
 
 [[tool.mypy.overrides]]
-module = ["mkdocs_macros.*"]
+module = [
+  "mkdocs_macros.*",
+  "async_solipsism",
+  "async_solipsism.*",
+]
 ignore_missing_imports = true
 
 [tool.setuptools_scm]

src/frequenz/dispatch/__init__.py

@@ -2,3 +2,70 @@
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
 """A highlevel interface for the dispatch API."""
+
+import grpc.aio
+from frequenz.channels import Broadcast, Receiver
+from frequenz.client.dispatch.types import Dispatch
+
+from frequenz.dispatch.actor import DispatchActor
+
+__all__ = ["Dispatcher"]
+
+
+class Dispatcher:
+    """A highlevel interface for the dispatch API.
+
+    This class provides a highlevel interface to the dispatch API. It
+    allows to receive new dispatches and ready dispatches.
+
+    Example:
+        ```python
+        from frequenz.dispatch import Dispatcher
+
+        async def run():
+            dispatcher = Dispatcher(API_CONNECTION_INFO)
+            dispatcher.start()  # this will start the actor
+            dispatch_arrived = dispatcher.new_dispatches().new_receiver()
+            dispatch_ready = dispatcher.ready_dispatches().new_receiver()
+        ```
+    """
+
+    def __init__(
+        self, microgrid_id: int, grpc_channel: grpc.aio.Channel, svc_addr: str
+    ):
+        """Initialize the dispatcher.
+
+        Args:
+            microgrid_id: The microgrid id.
+            grpc_channel: The gRPC channel.
+            svc_addr: The service address.
+        """
+        self._ready_channel = Broadcast[Dispatch]("ready_dispatches")
+        self._new_channel = Broadcast[Dispatch]("new_dispatches")
+        self._actor = DispatchActor(
+            microgrid_id,
+            grpc_channel,
+            svc_addr,
+            self._new_channel.new_sender(),
+            self._ready_channel.new_sender(),
+        )
+
+    async def start(self) -> None:
+        """Start the actor."""
+        self._actor.start()
+
+    def new_dispatches(self) -> Receiver[Dispatch]:
+        """Return new dispatches receiver.
+
+        Returns:
+            A new receiver for new dispatches.
+        """
+        return self._new_channel.new_receiver()
+
+    def ready_dispatches(self) -> Receiver[Dispatch]:
+        """Return ready dispatches receiver.
+
+        Returns:
+            A new receiver for ready dispatches.
+        """
+        return self._ready_channel.new_receiver()

src/frequenz/dispatch/actor.py

@@ -0,0 +1,199 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""The dispatch actor."""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+import grpc.aio
+from frequenz.channels import Sender
+from frequenz.client.dispatch import Client
+from frequenz.client.dispatch.types import Dispatch, Frequency
+from frequenz.sdk.actor import Actor
+
+_FREQUENCY_MAP = {
+    Frequency.MINUTELY: timedelta(minutes=1),
+    Frequency.HOURLY: timedelta(hours=1),
+    Frequency.DAILY: timedelta(days=1),
+    Frequency.WEEKLY: timedelta(weeks=1),
+}
+
+_MAX_AHEAD_SCHEDULE = timedelta(hours=5)
+
+
+class DispatchActor(Actor):
+    """Dispatch actor.
+
+    This actor is responsible for handling dispatches for a microgrid.
+
+    This means staying in sync with the API and scheduling
+    dispatches as necessary.
+    """
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        microgrid_id: int,
+        grpc_channel: grpc.aio.Channel,
+        svc_addr: str,
+        updated_dispatch_sender: Sender[Dispatch],
+        ready_dispatch_sender: Sender[Dispatch],
+    ) -> None:
+        """Initialize the actor.
+
+        Args:
+            microgrid_id: The microgrid ID to handle dispatches for.
+            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.
+        """
+        super().__init__(name="dispatch")
+
+        self._client = Client(grpc_channel, svc_addr)
+        self._dispatches: dict[int, Dispatch] = {}
+        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
+
+    async def _run(self) -> None:
+        """Run the actor."""
+        try:
+            while True:
+                # for _ in range(30):
+                await self._fetch()
+                await asyncio.sleep(timedelta(hours=5).total_seconds())
+        except asyncio.CancelledError:
+            for task in self._scheduled.values():
+                task.cancel()
+            raise
+
+    async def _fetch(self) -> None:
+        """Fetch all relevant dispatches."""
+        old_dispatches = self._dispatches
+        self._dispatches = {}
+
+        try:
+            logging.info("Fetching dispatches for microgrid %s", self._microgrid_id)
+            async for dispatch in self._client.list(microgrid_id=self._microgrid_id):
+                self._dispatches[dispatch.id] = dispatch
+
+                old_dispatch = old_dispatches.pop(dispatch.id, None)
+                if not old_dispatch:
+                    self._update_dispatch_schedule(dispatch, None)
+                    logging.info("New dispatch: %s", dispatch)
+                    await self._updated_dispatch_sender.send(dispatch)
+                elif dispatch.update_time != old_dispatch.update_time:
+                    self._update_dispatch_schedule(dispatch, old_dispatch)
+                    logging.info("Updated dispatch: %s", dispatch)
+                    await self._updated_dispatch_sender.send(dispatch)
+
+        except grpc.aio.AioRpcError as error:
+            logging.error("Error fetching dispatches: %s", error)
+            self._dispatches = old_dispatches
+            return
+
+        for dispatch in old_dispatches.values():
+            logging.info("Deleted dispatch: %s", dispatch)
+            if task := self._scheduled.pop(dispatch.id, None):
+                task.cancel()
+
+    def _update_dispatch_schedule(
+        self, dispatch: Dispatch, old_dispatch: Dispatch | None
+    ) -> None:
+        """Update the schedule for a dispatch.
+
+        Schedules, reschedules or cancels the dispatch based on the start_time
+        and active status.
+
+        Args:
+            dispatch: The dispatch to update the schedule for.
+            old_dispatch: The old dispatch, if available.
+        """
+        if (
+            old_dispatch
+            and old_dispatch.active
+            and old_dispatch.start_time != dispatch.start_time
+        ):
+            if task := self._scheduled.pop(dispatch.id, None):
+                task.cancel()
+
+        if dispatch.active and dispatch.id not in self._scheduled:
+            self._scheduled[dispatch.id] = asyncio.create_task(
+                self._schedule_task(dispatch)
+            )
+
+    async def _schedule_task(self, dispatch: Dispatch) -> None:
+        """Wait for a dispatch to become ready.
+
+        Waits for the dispatches next run and then notifies that it is ready.
+
+        Args:
+            dispatch: The dispatch to schedule.
+        """
+        while True:
+            next_time = self.calculate_next_run(
+                dispatch, datetime.now().replace(tzinfo=timezone.utc)
+            )
+            if next_time is None:
+                logging.info("Dispatch finished: %s", dispatch)
+                self._scheduled.pop(dispatch.id)
+                return
+
+            if (
+                next_time - datetime.now().replace(tzinfo=timezone.utc)
+                > _MAX_AHEAD_SCHEDULE
+            ):
+                await asyncio.sleep(_MAX_AHEAD_SCHEDULE.total_seconds())
+                continue
+
+            logging.info("Dispatch %s scheduled for %s", dispatch.id, next_time)
+            logging.info(
+                "Sleeping for %s",
+                next_time - datetime.now().replace(tzinfo=timezone.utc),
+            )
+            await asyncio.sleep(
+                (
+                    next_time - datetime.now().replace(tzinfo=timezone.utc)
+                ).total_seconds()
+            )
+            logging.info("Dispatch ready: %s", dispatch)
+            await self._ready_dispatch_sender.send(dispatch)
+
+    @classmethod
+    def calculate_next_run(cls, dispatch: Dispatch, _from: datetime) -> datetime | None:
+        """Calculate the next run of a dispatch.
+
+        Args:
+            dispatch: The dispatch to calculate the next run for.
+            _from: The time to calculate the next run from.
+
+        Returns:
+            The next run of the dispatch or None if the dispatch is finished.
+        """
+        next_run = dispatch.start_time
+
+        while next_run < _from:
+            match dispatch.recurrence.frequency:
+                case (
+                    Frequency.MINUTELY
+                    | Frequency.HOURLY
+                    | Frequency.DAILY
+                    | Frequency.WEEKLY
+                ):
+                    next_run = (
+                        next_run
+                        + _FREQUENCY_MAP[dispatch.recurrence.frequency]
+                        * dispatch.recurrence.interval
+                    )
+                case Frequency.MONTHLY:
+                    new_month = next_run.month + dispatch.recurrence.interval
+                    new_year = next_run.year + new_month // 12
+                    next_run = next_run.replace(year=new_year, month=new_month % 12)
+                case Frequency.UNSPECIFIED:
+                    return None
+
+        return next_run

src/frequenz/dispatch/actor/__init__.py

@@ -0,0 +1,4 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for the frequenz.dispatch package."""