Initial structure

.github/labeler.yml

@@ -6,7 +6,6 @@
 # For more details on the configuration please see:
 # https://github.com/marketplace/actions/labeler
 
-# TODO(cookiecutter): Add different parts of the source
 # For example:
 #
 # "part:module":
@@ -65,4 +64,4 @@
 "part:actor":
   - changed-files:
     - any-glob-to-any-file:
-      - "src/frequenz/dispatch/actor/**"
+      - "src/frequenz/dispatch/actor.py"

.github/workflows/ci.yaml

@@ -180,7 +180,7 @@ jobs:
             --net=host \
             --platform linux/${{ matrix.arch }} \
             localhost/nox-cross-arch:latest \
-            bash -c "pip install -e .[dev-noxfile]; nox --install-only -e ${{ matrix.nox-session }}; pip freeze; nox -e ${{ matrix.nox-session }}"
+            bash -c "pip install -e .[dev-noxfile]; nox --install-only -e ${{ matrix.nox-session }}; pip freeze; nox -R -e ${{ matrix.nox-session }}"
         timeout-minutes: 30
 
       # This ensures that the runner has access to the pip cache.

README.md

@@ -15,22 +15,30 @@ It provides two channels for clients:
 ## Example Usage
 
 ```python
-async def run():
-    # dispatch helper sends out dispatches when they are due
-    dispatch_arrived = dispatch_helper.new_dispatches().new_receiver()
-    dispatch_ready = dispatch_helper.ready_dispatches().new_receiver()
-
-    async for selected in select(dispatch_ready, dispatch_arrived):
-        if selected_from(selected, dispatch_arrived):
-            dispatch = selected.value
-            match dispatch.type:
-                case DISPATCH_TYPE_BATTERY_CHARGE:
-                    battery_pool = microgrid.battery_pool(dispatch.battery_set, task_id)
-                    battery_pool.set_charge(dispatch.power)
-            ...
-        if selected_from(selected, dispatch_ready):
-            dispatch = selected.value
-            log.info("New dispatch arrived %s", dispatch)
+    async def run():
+        # dispatch helper sends out dispatches when they are due
+        dispatch_arrived = dispatch_helper.updated_dispatches().new_receiver()
+        dispatch_ready = dispatch_helper.ready_dispatches().new_receiver()
+
+        async for selected in select(dispatch_ready, dispatch_arrived):
+            if selected_from(selected, dispatch_ready):
+                dispatch = selected.value
+                match dispatch.type:
+                    case DISPATCH_TYPE_BATTERY_CHARGE:
+                        battery_pool = microgrid.battery_pool(dispatch.battery_set, task_id)
+                        battery_pool.set_charge(dispatch.power)
+                ...
+            if selected_from(selected, dispatch_arrived):
+                match selected.value:
+                    case Created(dispatch):
+                        log.info("New dispatch arrived %s", dispatch)
+                        ...
+                    case Updated(dispatch):
+                        log.info("Dispatch updated %s", dispatch)
+                        ...
+                    case Deleted(dispatch):
+                        log.info("Dispatch deleted %s", dispatch)
+                        ...
 ```
 
 ## Supported Platforms

mkdocs.yml

@@ -115,7 +115,7 @@ plugins:
             # See https://mkdocstrings.github.io/python/usage/#import for details
             - https://docs.python.org/3/objects.inv
             - https://frequenz-floss.github.io/frequenz-channels-python/v0.16/objects.inv
-            - https://frequenz-floss.github.io/frequenz-sdk-python/v0.25/objects.inv
+            - https://frequenz-floss.github.io/frequenz-sdk-python/v1.0-pre/objects.inv
             - https://typing-extensions.readthedocs.io/en/stable/objects.inv
             - https://frequenz-floss.github.io/frequenz-client-dispatch-python/v0.1/objects.inv
   # Note this plugin must be loaded after mkdocstrings to be able to use macros

pyproject.toml

@@ -15,7 +15,6 @@ description = "A highlevel interface for the dispatch API"
 readme = "README.md"
 license = { text = "MIT" }
 keywords = ["frequenz", "python", "actor", "frequenz-dispatch", "dispatch", "highlevel", "api"]
-# TODO(cookiecutter): Remove and add more classifiers if appropriate
 classifiers = [
   "Development Status :: 3 - Alpha",
   "Intended Audience :: Developers",
@@ -26,21 +25,27 @@ classifiers = [
   "Typing :: Typed",
 ]
 requires-python = ">= 3.11, < 4"
-# TODO(cookiecutter): Remove and add more dependencies if appropriate
 dependencies = [
+  "python-dateutil >= 2.8.2, < 3.0",
   "typing-extensions == 4.10.0",
   # Make sure to update the version for cross-referencing also in the
   # mkdocs.yml file when changing the version here (look for the config key
   # plugins.mkdocstrings.handlers.python.import)
-  "frequenz-sdk == 0.25.2",
+  "frequenz-sdk == v1.0.0-rc5",
+  "frequenz-channels == 1.0.0b2",
+  "frequenz-dispatch-client @ git+https://github.com/frequenz-floss/frequenz-client-dispatch-python.git@00dcb3c",
+  "frequenz-api-dispatch @ git+https://github.com/frequenz-floss/frequenz-api-dispatch.git@223315c",
+#  "frequenz-api-dispatch >= 0.13.0, < 0.14",
+  "frequenz-client-base >= 0.2.1, < 0.3.0",
+  # Directly use unreleased commit until the first release
+  "frequenz-client-common @ git+https://github.com/frequenz-floss/frequenz-client-common-python.git@5e6e7b7",
 ]
 dynamic = ["version"]
 
 [[project.authors]]
 name = "Frequenz Energy-as-a-Service GmbH"
 email = "[email protected]"
 
-# TODO(cookiecutter): Remove and add more optional dependencies if appropriate
 [project.optional-dependencies]
 dev-flake8 = [
   "flake8 == 7.0.0",
@@ -64,11 +69,13 @@ dev-mkdocs = [
 dev-mypy = [
   "mypy == 1.8.0",
   "types-Markdown == 3.5.0.20240129",
+  "types-python-dateutil==2.8.19.20240311",
   # For checking the noxfile, docs/ script, and tests
   "frequenz-dispatch[dev-mkdocs,dev-noxfile,dev-pytest]",
 ]
 dev-noxfile = [
-  "nox == 2023.4.22",
+  "uv == 0.1.14",
+  "nox == 2024.3.02",
   "frequenz-repo-config[lib] == 0.9.1",
 ]
 dev-pylint = [
@@ -82,6 +89,7 @@ dev-pytest = [
   "pytest-mock == 3.12.0",
   "pytest-asyncio == 0.23.5",
   "async-solipsism == 0.5",
+  "time-machine == 2.14.0",
 ]
 dev = [
   "frequenz-dispatch[dev-mkdocs,dev-flake8,dev-formatting,dev-mkdocs,dev-mypy,dev-noxfile,dev-pylint,dev-pytest]",
@@ -159,11 +167,15 @@ namespace_packages = true
 # used but getting the original ignored error when removing the type: ignore.
 # See for example: https://github.com/python/mypy/issues/2960
 #no_incremental = true
-packages = ["frequenz.actor.dispatch"]
+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

@@ -0,0 +1,78 @@
+# License: MIT
+# 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, DispatchEvent
+
+__all__ = ["Dispatcher"]
+
+
+class Dispatcher:
+    """A highlevel interface for the dispatch API.
+
+    This class provides a highlevel interface to the dispatch API.
+    It provides two channels:
+
+    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.
+
+    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._updated_channel = Broadcast[DispatchEvent]("new_dispatches")
+        self._actor = DispatchActor(
+            microgrid_id,
+            grpc_channel,
+            svc_addr,
+            self._updated_channel.new_sender(),
+            self._ready_channel.new_sender(),
+        )
+
+    async def start(self) -> None:
+        """Start the actor."""
+        self._actor.start()
+
+    def updated_dispatches(self) -> Receiver[DispatchEvent]:
+        """Return new, updated or deleted dispatches receiver.
+
+        Returns:
+            A new receiver for new dispatches.
+        """
+        return self._updated_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()