duck.automation

Classes

• AutomationError: Custom exception for automation-related errors.

• AutomationThread: Threading class for executing automations.

• Automation: Base class for automating tasks, jobs, actions, etc.

Usage Example

import os
from duck.automation.dispatcher import DispatcherV1
from duck.automation.trigger import NoTrigger

class SimpleAutomation(Automation):
    # Execute a shell script
    def execute(self):
        # Or use async def execute for asynchronous execution, but parse `async_` argument to automation.
        os.system('bash some_script.sh')

# Instantiate the automation with specified parameters
automation = SimpleAutomation(
    name="Sample Automation",
    description="Sample automation",
    start_time='immediate',
    schedules=1,
    interval=0,
)  # Set automation schedules to -1 for infinite schedules

# Instantiate a trigger that executes immediately
trigger = NoTrigger()

# You can create a custom trigger by implementing the `check_trigger` method
# in your AutomationTrigger subclass and returning True or False based on
# whether the trigger condition is satisfied

# Create a dispatcher to manage and execute automations
dispatcher = DispatcherV1()  # The first argument is the main running web application (optional)

# Register the trigger and automation with the dispatcher
dispatcher.register(trigger, automation)
dispatcher.start()  # Listen for triggers and execute automations infinitely

# Alternatively, use a callback function for the automation task
def callback():
    # Perform the automation tasks here
    # Avoid using while loops; set the number of schedules to -1 for infinite schedules
    pass

automation = Automation(
    callback=callback,
    name="Sample Automation",
    description="Sample automation"
)

# Register the trigger and automation with the dispatcher
dispatcher.register(trigger, automation)
dispatcher.start()  # Listen for triggers and execute automations infinitely

Module for automations used in automating tasks, jobs, actions, and more.

This module provides classes and utilities to define and manage automated tasks. It includes a base Automation class for creating and running tasks with various scheduling options, as well as supporting threading and callback functionalities.

Submodules

• duck.automation.dispatcher
• duck.automation.trigger

Package Contents

Classes

Automation	Automation class for automating tasks, jobs, actions, etc.
AutomationThread	Threading class for Automations execution.
SampleAutomationBase	A placeholder automation class designed for testing or sampling purposes.

Data

SampleAutomation

API

class duck.automation.Automation(callback: callable = None, name: str = None, threaded: bool = True, async_: bool = False, start_time: datetime.datetime | str = 'immediate', schedules: int = 1, interval: float | int = None, description: str = None)

Automation class for automating tasks, jobs, actions, etc.

Events:

• on_pre_execute: Called before every execution cycle.

• on_post_execute: Called after every execution cycle.

• on_start: Called once when the automation starts.

• on_finish: Called once when the automation finishes.

• on_error: Called whenever there is an error during execution. Override this method to suppress or handle all errors.

Notes:

• The start method is the initial method to run an automation.

• The callback argument is optional; if not provided, the execute method must be implemented.

• Provide a name and description for more descriptive Automations.

• The join method can be used to stop the next execution cycle, meaning the automation will stop immediately before entering the next execution cycle.

• Using a while loop in the callback or execute method will cause the automation to hang, and join will not work if the callback is in a loop.

• The join method works properly before the next execution cycle or after the current execution cycle.

Automations can be asynchronously executed by providing the async_ argument.

Initialization

Initialize the Automation class.

Parameters:

• callback – A callable to be executed whenever the automation runs. If you do not provide a callback, implement the ‘execute’ method.

• name – The name of the Automation. Defaults to None.

• threaded – Indicates whether to run the automation in a new thread. Defaults to True.

• async_ – Indicates whether to run automation asynchronously. Defaults to False.

• start_time – The datetime to start the Automation. It can be “immediate” or any specific datetime.

• schedules – The number of times to run the Automation. Defaults to 1. Set to -1 for infinite schedules.

• interval – The time period in seconds between successive runs of the Automation.

• description – A brief description of the Automation. Defaults to an empty string.

__repr__()

async _async_start()

Asyncronous entry method to execute an automation.

_start()

Entry method to execute an automation.

execute()

This executes an automation. Do whatever task you want here e.g. running bash scripts or something else.

Example:

import os

class SimpleAutomation(Automation):
    def execute(self):
        os.system('bash some_script.bash')

automation = SimpleAutomation(start_time='immediate', schedules=1, interval=0)
automation.start # start the automation

property execution_cycles: int

Returns the number of times the automation has been executed (same as property ‘execution_times’).

property execution_times: int

Returns the number of times the automation has been executed.

property finished

Whether the automation has been stopped completely.

property finished_at: datetime.datetime

The datetime where this automation finished.

property first_execution: bool

Returns whether the automation has already been executed for the first time.

… admonition:: Notes

• Take a look a property execution_times to see the number of times the Automation was Executed.

get_running_app()

Returns the main running application

get_short_description()

Get the short version of description.

get_thread()

This returns the Thread for running automation.

… admonition:: Notes

• The thread is only created once.

property is_running: bool

Whether the automation is running or not.

join()

Wait for the automation to finish the current execution cycle and stop the execution

property latest_error

Returns the latest exception encountered during execution.

on_error(e)

Method called on execution automation after method “start” is called.

on_finish()

Method called on automation final finish.

on_post_execute()

Method called after automation has finished the current execution cycle.

on_pre_execute()

Method called before automation starts the current execution cycle.

on_start()

Method called once the execution has been started.

Notes:

• Method “on_start” is called before method “on_pre_execute”

prepare_stop()

Called before the main application termination.

set_running_app(app)

Set the main running application

start()

Entry method to execute an automation.

property started_at: datetime.datetime

The datetime where this automation started.

to_thread() → duck.automation.AutomationThread

This wraps an automation in new Thread so that it may be executed nicely without blocking any other automations execution.

Returns:

Returns an automation in new thread.

Return type:

AutomationThread

exception duck.automation.AutomationError

Bases: Exception

Automation based exceptions.

Initialization

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

class duck.automation.AutomationThread(group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)

Bases: threading.Thread

Threading class for Automations execution.

Initialization

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.init()) before doing anything else to the thread.

on_stop()

Method called whenever a thread finishes execution.

The callback set with method ‘set_on_stop_callback’ will be called if present.

set_on_stop_callback(callback: callable)

Set a callback that will be called whenever the thread finishes execution.

Notes:

• Make sure callback doesn’t allow any positional or keyword arguments.

start(*args, **kw)

duck.automation.SampleAutomation

‘SampleAutomationBase(…)’

class duck.automation.SampleAutomationBase(callback: callable = None, name: str = None, threaded: bool = True, async_: bool = False, start_time: datetime.datetime | str = 'immediate', schedules: int = 1, interval: float | int = None, description: str = None)

Bases: duck.automation.Automation

A placeholder automation class designed for testing or sampling purposes.

This automation is intentionally empty and does not perform any actions. It can be used as a base for creating more complex automations or for testing automation systems without side effects.

Use this class when you need a no-op (no operation) automation that will not alter the application’s state or behavior during execution.

Initialization

Initialize the Automation class.

Parameters:

• callback – A callable to be executed whenever the automation runs. If you do not provide a callback, implement the ‘execute’ method.

• name – The name of the Automation. Defaults to None.

• threaded – Indicates whether to run the automation in a new thread. Defaults to True.

• async_ – Indicates whether to run automation asynchronously. Defaults to False.

• start_time – The datetime to start the Automation. It can be “immediate” or any specific datetime.

• schedules – The number of times to run the Automation. Defaults to 1. Set to -1 for infinite schedules.

• interval – The time period in seconds between successive runs of the Automation.

• description – A brief description of the Automation. Defaults to an empty string.

execute()

No-op method.

This method is intentionally left empty and performs no actions. It is used as a placeholder or for testing purposes where no operation is required.