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
+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,73 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""The dispatch actor."""
+
+import asyncio
+import logging
+
+import grpc
+from frequenz.channels import Sender
+from frequenz.client.dispatch import Client
+from frequenz.client.dispatch.types import Dispatch
+from frequenz.sdk.actor import Actor
+
+
+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,
+        new_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.
+            new_dispatch_sender: A sender for new 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._microgrid_id = microgrid_id
+        self._new_dispatch_sender = new_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(5)
+
+    async def _fetch(self) -> None:
+        """Fetch all relevant dispatches."""
+        old_dispatches = self._dispatches
+        self._dispatches = {}
+
+        try:
+            logging.info("Fetching dispatches")
+            async for dispatch in self._client.list(microgrid_id=1):
+                self._dispatches[dispatch.id] = dispatch
+
+                if dispatch.id not in old_dispatches:
+                    logging.info("New dispatch: %s", dispatch)
+                    await self._new_dispatch_sender.send(dispatch)
+
+        except grpc.aio.AioRpcError as error:
+            logging.error("Error fetching dispatches: %s", error)
+            self._dispatches = old_dispatches

src/frequenz/dispatch/actor/__init__.py

@@ -2,17 +2,3 @@
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
 """Tests for the frequenz.dispatch.actor package."""
-import pytest
-
-from frequenz.dispatch.actor import delete_me
-
-
-def test_frequenz_dispatch_succeeds() -> None:  # TODO(cookiecutter): Remove
-    """Test that the delete_me function succeeds."""
-    assert delete_me() is True
-
-
-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)