Add initial structure

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

Author: Marenz

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,181 @@
+# 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),
+}
+
+
+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."""
+        while True:
+            await self._fetch()
+            await asyncio.sleep(timedelta(minutes=5).total_seconds())
+
+    async def _fetch(self) -> None:
+        """Fetch all relevant dispatches."""
+        old_dispatches = self._dispatches
+        self._dispatches = {}
+        # breakpoint()
+        try:
+            logging.info("Fetching dispatches")
+            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
+
+            await asyncio.sleep(
+                (
+                    next_time - datetime.now().replace(tzinfo=timezone.utc)
+                ).total_seconds()
+            )
+
+            await self._ready_dispatch_sender.send(dispatch)
+            logging.info("Dispatch ready: %s", dispatch)
+
+    def _calculate_next_run(
+        self, 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

@@ -2,17 +2,81 @@
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
 """Tests for the frequenz.dispatch.actor package."""
-import pytest
 
-from frequenz.dispatch.actor import delete_me
 
+from dataclasses import dataclass
+from typing import Iterator
 
-def test_frequenz_dispatch_succeeds() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function succeeds."""
-    assert delete_me() is True
+import async_solipsism
+import grpc.aio
+from frequenz.channels import Broadcast, Receiver
+from frequenz.client.dispatch.test.client import FakeClient, to_create_params
+from frequenz.client.dispatch.test.generator import DispatchGenerator
+from frequenz.client.dispatch.types import Dispatch
+from pytest import fixture
 
+from frequenz.dispatch.actor import DispatchActor
 
-def test_frequenz_dispatch_fails() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function fails."""
-    with pytest.raises(RuntimeError, match="This function should be removed!"):
-        delete_me(blow_up=True)
+
+# This method replaces the event loop for all tests in the file.
+@fixture()
+def event_loop() -> Iterator[async_solipsism.EventLoop]:
+    """Replace the loop with one that doesn't interact with the outside world."""
+    loop = async_solipsism.EventLoop()
+    yield loop
+    loop.close()
+
+
+@dataclass
+class ActorTestEnv:
+    """Test environment for the actor."""
+
+    actor: DispatchActor
+    updated_dispatches: Receiver[Dispatch]
+    ready_dispatches: Receiver[Dispatch]
+    client: FakeClient
+
+
+@fixture
+def actor_env() -> ActorTestEnv:
+    """Return an actor test environment."""
+    updated_dispatches = Broadcast[Dispatch]("updated_dispatches")
+    ready_dispatches = Broadcast[Dispatch]("ready_dispatches")
+
+    actor = DispatchActor(
+        microgrid_id=1,
+        grpc_channel=grpc.aio.insecure_channel("mock"),
+        svc_addr="localhost",
+        updated_dispatch_sender=updated_dispatches.new_sender(),
+        ready_dispatch_sender=ready_dispatches.new_sender(),
+    )
+
+    client = FakeClient()
+    actor._client = client  # pylint: disable=protected-access
+
+    actor.start()
+
+    return ActorTestEnv(
+        actor,
+        updated_dispatches.new_receiver(),
+        ready_dispatches.new_receiver(),
+        client,
+    )
+
+
+@fixture
+def generator() -> DispatchGenerator:
+    """Return a dispatch generator."""
+    return DispatchGenerator()
+
+
+async def test_new_dispatch_created(
+    actor_env: ActorTestEnv, generator: DispatchGenerator
+) -> None:
+    """Test that a new dispatch is created."""
+    breakpoint()
+    sample = generator.generate_dispatch()
+    await actor_env.client.create(**to_create_params(sample))
+
+    dispatch = await actor_env.updated_dispatches.receive()
+    assert dispatch == sample