Add Dispatcher.manage to manage dispatchable actors

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

Author: Marenz

RELEASE_NOTES.md

@@ -19,6 +19,5 @@ This release introduces a more flexible and powerful mechanism for managing disp
 ## New Features
 
 * A new feature "merger strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge & unify consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
-
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.
-
+* A new function `Dispatcher.manage` was added to simplify dispatchable actor management initialization.

src/frequenz/dispatch/_dispatcher.py

@@ -4,12 +4,20 @@
 """A highlevel interface for the dispatch API."""
 
 
+import logging
+from typing import Callable
+
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
+from frequenz.sdk.actor import Actor
 
+from ._actor_dispatcher import ActorDispatcher, DispatchInfo
 from ._bg_service import DispatchScheduler, MergeStrategy
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
+from ._merge_strategies import MergeByIdentity
+
+_logger = logging.getLogger(__name__)
 
 
 class Dispatcher:
@@ -178,11 +186,53 @@ def __init__(
             microgrid_id,
             self._client,
         )
+        self._actor_dispatchers: dict[str, ActorDispatcher] = {}
 
-    async def start(self) -> None:
+    def start(self) -> None:
         """Start the local dispatch service."""
         self._bg_service.start()
 
+    async def manage(
+        self,
+        dispatch_type: str,
+        *,
+        actor_factory: Callable[[DispatchInfo, Receiver[DispatchInfo]], Actor],
+        merge_strategy: MergeByIdentity | None = None,
+    ) -> None:
+        """Manage actors for a given dispatch type.
+
+        Creates and manages an ActorDispatcher for the given type that will
+        start, stop and reconfigure actors based on received dispatches.
+
+        Args:
+            dispatch_type: The type of the dispatch to manage.
+            actor_factory: The factory to create actors.
+            merge_strategy: The strategy to merge running intervals.
+        """
+        dispatcher = self._actor_dispatchers.get(dispatch_type)
+
+        if dispatcher is not None:
+            _logger.debug(
+                "Ignoring duplicate actor dispatcher request for %r", dispatch_type
+            )
+            return
+
+        def id_identity(dispatch: Dispatch) -> int:
+            return dispatch.id
+
+        dispatcher = ActorDispatcher(
+            actor_factory=actor_factory,
+            running_status_receiver=await self.new_running_state_event_receiver(
+                dispatch_type, merge_strategy=merge_strategy
+            ),
+            map_dispatch=(
+                id_identity if merge_strategy is None else merge_strategy.identity
+            ),
+        )
+
+        self._actor_dispatchers[dispatch_type] = dispatcher
+        dispatcher.start()
+
     @property
     def client(self) -> Client:
         """Return the client."""

tests/test_mananging_actor.py

@@ -8,20 +8,39 @@
 import logging
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
-from typing import AsyncIterator, Iterator, cast
+from typing import AsyncIterator, Callable, Iterator, cast
+from unittest.mock import patch
 
 import async_solipsism
+import pytest
 import time_machine
 from frequenz.channels import Broadcast, Receiver, Sender
+from frequenz.client.dispatch import recurrence
 from frequenz.client.dispatch.recurrence import Frequency
+from frequenz.client.dispatch.test.client import FakeClient
 from frequenz.client.dispatch.test.generator import DispatchGenerator
 from frequenz.sdk.actor import Actor
 from pytest import fixture
 
-from frequenz.dispatch import ActorDispatcher, Dispatch, DispatchInfo
+from frequenz.dispatch import (
+    ActorDispatcher,
+    Dispatch,
+    Dispatcher,
+    DispatchInfo,
+    MergeByIdentity,
+    MergeByType,
+    MergeByTypeTarget,
+    MergeStrategy,
+)
 from frequenz.dispatch._bg_service import DispatchScheduler
 
 
+@fixture
+def generator() -> DispatchGenerator:
+    """Return a dispatch generator."""
+    return DispatchGenerator()
+
+
 @fixture
 def event_loop_policy() -> async_solipsism.EventLoopPolicy:
     """Set the event loop policy to use async_solipsism."""
@@ -232,3 +251,80 @@ async def test_dry_run(test_env: TestEnv, fake_time: time_machine.Coordinates) -
 
     # Give await actor.stop a chance to run
     await asyncio.sleep(1)
+
+
+@pytest.mark.parametrize("strategy", [MergeByTypeTarget(), MergeByType(), None])
+async def test_manage_abstraction(
+    fake_time: time_machine.Coordinates,
+    generator: DispatchGenerator,
+    strategy: MergeByIdentity | None,
+) -> None:
+    """Test Dispatcher.manage sets up correctly."""
+    identity: Callable[[Dispatch], int] = (
+        strategy.identity if strategy else lambda dispatch: dispatch.id
+    )
+
+    class MyFakeClient(FakeClient):
+        """Fake client for testing."""
+
+        def __init__(self, *, server_url: str, key: str):
+            assert server_url
+            assert key
+            super().__init__()
+
+    mid = 1
+
+    # Patch `Client` class in Dispatcher with MyFakeClient
+    with patch("frequenz.dispatch._dispatcher.Client", MyFakeClient):
+        dispatcher = Dispatcher(
+            microgrid_id=mid, server_url="grpc://test-url", key="test-key"
+        )
+        dispatcher.start()
+
+        channel = Broadcast[Dispatch](name="dispatch ready test channel")
+        sender = channel.new_sender()
+
+        async def new_mock_receiver(
+            _: Dispatcher, dispatch_type: str, *, merge_strategy: MergeStrategy | None
+        ) -> Receiver[Dispatch]:
+            assert dispatch_type == "MANAGE_TEST"
+            assert merge_strategy is strategy
+            return channel.new_receiver()
+
+        with patch(
+            "frequenz.dispatch._dispatcher.Dispatcher.new_running_state_event_receiver",
+            new_mock_receiver,
+        ):
+            await dispatcher.manage(
+                dispatch_type="MANAGE_TEST",
+                actor_factory=MockActor,
+                merge_strategy=strategy,
+            )
+
+        # pylint: disable=protected-access
+        assert "MANAGE_TEST" in dispatcher._actor_dispatchers
+        actor_manager = dispatcher._actor_dispatchers["MANAGE_TEST"]
+        assert actor_manager._actor_factory == MockActor
+
+        dispatch = Dispatch(
+            replace(
+                generator.generate_dispatch(),
+                start_time=_now(),
+                duration=timedelta(minutes=10),
+                recurrence=recurrence.RecurrenceRule(),
+                active=True,
+                type="MANAGE_TEST",
+            )
+        )
+
+        fake_time.move_to(dispatch.start_time + timedelta(seconds=1))
+        assert dispatch.started
+
+        # Send a dispatch to start an actor instance
+        await sender.send(dispatch)
+
+        # Give the actor a chance to start
+        await asyncio.sleep(1)
+
+        # Check if actor instance is created
+        assert identity(dispatch) in actor_manager._actors