A highlevel interface for the dispatch API
Project description
Dispatch Highlevel Interface
Introduction
A highlevel interface for the dispatch API.
See the documentation for more information.
Usage
The Dispatcher
class, the main entry point for the API, provides two channels:
- Lifecycle events: A channel that sends a message whenever a [Dispatch][frequenz.dispatch.Dispatch] is created, updated or deleted.
- Running status change: Sends a dispatch message whenever a dispatch is ready to be executed according to the schedule or the running status of the dispatch changed in a way that could potentially require the actor to start, stop or reconfigure itself.
Example using the running status change channel
import os
from frequenz.dispatch import Dispatcher, RunningState
from unittest.mock import MagicMock
async def run():
url = os.getenv("DISPATCH_API_URL", "grpc://fz-0004.frequenz.io:50051")
key = os.getenv("DISPATCH_API_KEY", "some-key")
microgrid_id = 1
dispatcher = Dispatcher(
microgrid_id=microgrid_id,
server_url=url,
key=key
)
await dispatcher.start()
actor = MagicMock() # replace with your actor
changed_running_status_rx = dispatcher.running_status_change.new_receiver()
async for dispatch in changed_running_status_rx:
match dispatch.running("DEMO_TYPE"):
case RunningState.RUNNING:
print(f"Executing dispatch {dispatch.id}, due on {dispatch.start_time}")
if actor.is_running:
actor.reconfigure(
components=dispatch.selector,
run_parameters=dispatch.payload, # custom actor parameters
dry_run=dispatch.dry_run,
until=dispatch.until,
) # this will reconfigure the actor
else:
# this will start a new actor with the given components
# and run it for the duration of the dispatch
actor.start(
components=dispatch.selector,
run_parameters=dispatch.payload, # custom actor parameters
dry_run=dispatch.dry_run,
until=dispatch.until,
)
case RunningState.STOPPED:
actor.stop() # this will stop the actor
case RunningState.DIFFERENT_TYPE:
pass # dispatch not for this type
Supported Platforms
The following platforms are officially supported (tested):
- Python: 3.11
- Operating System: Ubuntu Linux 20.04
- Architectures: amd64, arm64
Contributing
If you want to know how to build this project and contribute to it, please check out the Contributing Guide.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
frequenz-dispatch-0.3.3.tar.gz
(16.8 kB
view hashes)
Built Distribution
Close
Hashes for frequenz_dispatch-0.3.3-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 8b77b165dcafa189d212bc04f8b6b552540a52fe2416ed80a4baa43f834e194c |
|
MD5 | 9dfa1e6296048ab91ec8a750ec78d1b6 |
|
BLAKE2b-256 | 652801ac83d0f615750d8a7e0c81e4e0c86785fe450d3130a28fe0af4f4434f9 |