duck.automation.dispatcher

Module for Automators that automate specific tasks.

This module provides classes and utilities to define and manage automated tasks. It includes an AutomationDispatcher class for dispatching automations based on triggers and a custom exception class for handling dispatcher-related errors.

Classes:

• AutomationDispatcherError: Custom exception for automation dispatcher-related errors.

• AutomationDispatcher: Class for dispatching automations based on triggers.

Usage example:

Note

The following example is for handling automations manually but this can be done automatically for you if you included your automations in settings.py.

from duck.automation import Automation
from duck.automation.trigger import AutomationTrigger

# Define your automations and triggers
trigger = AutomationTrigger(...)
automation = Automation(...)

# Create a dispatcher and register automations with triggers
dispatcher = AutomationDispatcher(application)
dispatcher.register(trigger, automation)

# Start the dispatcher to listen for triggers and run automations
dispatcher.start()

Module Contents

Classes

AutomationDispatcher	Automation dispatcher class.
DispatcherV1

API

class duck.automation.dispatcher.AutomationDispatcher(application=None)

Automation dispatcher class.

Notes:

• You can create your own dispatcher class by creating a subclass from this class and implement method listen

Initialization

__executed_automations: list[duck.automation.Automation]

[]

List of automations that has been started, these may be in finished or running state.

__queue: dict[duck.automation.trigger.AutomationTrigger, list[duck.automation.Automation]]

None

Dictionary of automation triggers mapping to their Automations.

property executed_automations: list[duck.automation.Automation]

Get all automations that have been executed, whether running or finished.

abstractmethod listen()

Listen for any triggers and execute any automations associated with those triggers.

This method removes each trigger that is satisfied and executes the corresponding automations.

Implement this method to customize how automations should be run based on your specific requirements.

Notes:

• You can check if a trigger is satisfied by using the “check_trigger” method on that trigger.

• Potential RuntimeError may be raised if you try accessing self.queue.keys convert self.queue.keys to another type instead to avoid this error.

Example:

def listen(self):
    for trigger in set(self.queue.keys()):
        trigger_satisfied = trigger.check_trigger()  # whether trigger is satisfied or fulfilled
        if trigger_satisfied:
            self.run_automations(self.queue.pop(trigger))  # pop trigger and execute corresponding automations

poll: int | float

1

Poll interval to listen for triggers

property queue: dict[duck.automation.trigger.AutomationTrigger, list[duck.automation.Automation]]

Returns a dictionary mapping of triggers to their automations.

register(trigger: duck.automation.trigger.AutomationTrigger, automation: duck.automation.Automation)

This adds an Automation to ready queue.

run_automations(automations: list[duck.automation.Automation])

Run all provided automations.

start()

Start listening for registered triggers and executes automations

stop()

Stops the dispatcher alongside all its automations.

… admonition:: Notes

This only stop automations that are about to execute their next execution cycle.

unregister(trigger: duck.automation.trigger.AutomationTrigger, automation: duck.automation.Automation)

Removes an automation from the ready queue.

Returns:

The removed automation.

Return type:

Automation

exception duck.automation.dispatcher.AutomationDispatcherError

Bases: Exception

Automation Dispatcher related exceptions

Initialization

Initialize self. See help(type(self)) for accurate signature.

class duck.automation.dispatcher.DispatcherV1(application=None)

Bases: duck.automation.dispatcher.AutomationDispatcher

listen()