Internal Documentation

These notes are intended to assist anyone that wants to understand AppDaemon’s internals better.

Structure

The Python project follows the conventional PEP 621, using a pyproject.toml file to define its metadata. The repository is divided into various folder:

./appdaemon

source code of the Python package

./docs

source code from which this documentation is built

./tests

tests written with pytest

./conf

configuration directory, containing some sample files

AppDaemon is organized into several subsystems managed by the central AppDaemon class. Each subsystem handles a specific aspect of functionality:

Core Subsystems

• App Management (AppManagement) - Provides the mechanics to manage the lifecycle of user apps, which includes tracking the user files for changes and reloading apps as needed.

• Scheduler (Scheduler) - Provides time-based scheduling, sunrise/sunset calculations, and time travel functionality for testing.

• State (State) - Tracks and manages entity states across different namespaces, provides state persistence between AppDaemon runs, and handles callbacks based on state changes.

• Events (Events) - Provides the mechanics to fire events, hand events off to relevant plugins (such as HASS for Home Assistant), and process event callbacks

• Callbacks (Callbacks) - Container for storing and managing all registered callbacks from apps.

• Services (Services) - Provides the mechanics to register/deregister services and allows apps to call them with a <domain>/<service_name> string.

• Plugin Management (PluginManagement) - Manages external plugins that provide connectivity to external systems (Home Assistant, MQTT, etc.).

Threading & Async

• Threading (Threading) - Manages worker threads for executing app callbacks and handles thread pinning.

• Thread Async (ThreadAsync) - Bridges between threaded callback execution and the main async event loop using queues.

• Utility Loop (Utility) - Runs periodic maintenance tasks like file change detection, thread monitoring, and performance diagnostics.

Additional Subsystems

• Sequences (Sequences) - Manages configurable automation sequences with steps like service calls, waits, and conditionals.

• Futures (Futures) - Handles asynchronous operations and provides mechanisms for cancelling long-running tasks.

Startup

The top-level entrypoint is appdaemon.__main__.main(), which uses argparse to parse the launch arguments.

The ADMain class primarily provides a context manager that can be used with a with statement. This contains an ExitStack instance that gets closed when its context is exited. Several methods are wrapped with the contextmanager() decorator and are added to the stack when AppDaemon is run. This guarantees that their respective cleanups are allowed to run in the correct order, which is the reverse order that the contexts were entered.

Contexts

The various context managers that get entered as AppDaemon starts include the logic for following steps. Some of these are entered as part of the ADMain context, and some are entered in the run method. All of them are exited in reverse order as the ExitStack is closed, which happens when the ADMain context exits.

• Backstop logic to catch any exceptions and log them more prettily.

• Creates a PID file for the duration of the context, if necessary/applicapable.

• Creates a new async event loop and cleans it up afterwards.

• Attaches/removes signal handlers to catch termination signals and stop gracefully

• Startup/shutdown text in the logs about the versions and time it took to shut down

• Creates the AppDaemon and HTTP objects.

• Sets/unsets the default exception handler for the event loop to prettify any unhandled exceptions. The difference with the previous exception handler is that this one has access to the AppDaemon object

Top-level entrypoint in __main__.py

def main() -> None:
    """Top-level entrypoint for AppDaemon

    Parses the CLI arguments, configures logging, and runs the AppDaemon.
    """
    args = parse_arguments()

    CLI_LOG_CFG = PRE_LOGGING.copy()

    if args.debug is not None:
        CLI_LOG_CFG["root"]["level"] = args.debug
        logger.debug("Configured logging level from command line argument")

    logging.config.dictConfig(CLI_LOG_CFG)

    with ADMain(args) as admain:
        admain.run()

Running

ADMain runs AppDaemon by calling its start() method followed by the run_forever() method of the event loop.

ADMain.run() method

    def run(self) -> None:
        """Start AppDaemon up after initial argument parsing.

        This uses :py:meth:`~asyncio.loop.run_forever` on the event loop to run it indefinitely.
        """
        self._cleanup_stack.enter_context(self.startup_text())
        self._cleanup_stack.enter_context(self.run_context(self.loop))
        self.AD.start()
        self.logger.debug("Running async event loop forever")
        self.loop.run_forever()
        self.logger.debug("Stopped running async event loop forever")

Shutdown

Shutdown is initiated by the process running AppDaemon receiving either a SIGINT or SIGTERM signal.

ADMain.handle_sig() method

    def handle_sig(self, signum: int):
        """Function to handle signals.

        Signals:
            SIGUSR1 will result in internal info being dumped to the DIAG log
            SIGUSR2 will reload apps with modified code/config (useful in production_mode)
            SIGHUP will force a reload of all apps
            SIGINT and SIGTEM both result in AD shutting down
        """
        match signum:
            case signal.SIGUSR1:
                self.AD.thread_async.call_async_no_wait(self.AD.sched.dump_schedule)
                self.AD.thread_async.call_async_no_wait(self.AD.callbacks.dump_callbacks)
                self.AD.thread_async.call_async_no_wait(self.AD.threading.dump_threads)
                self.AD.thread_async.call_async_no_wait(self.AD.app_management.dump_objects)
                self.AD.thread_async.call_async_no_wait(self.AD.sched.dump_sun)
            case signal.SIGUSR2:
                self.AD.thread_async.call_async_no_wait(self.AD.app_management.check_app_updates, mode=UpdateMode.NORMAL)
            case signal.SIGHUP:
                self.AD.thread_async.call_async_no_wait(self.AD.app_management.check_app_updates, mode=UpdateMode.TERMINATE)
            case (signal.SIGINT | signal.SIGTERM) as sig:
                self.logger.info(f"Received signal: {signal.Signals(sig).name}")
                self.stop()

The stop() method is called, which in turn calls the stop methods of the various subsystems. It doesn’t return until all the tasks that existed at the time of the call have finished. This usually happens almost instantly, but this has a 3s timeout just to be safe.

ADMain.stop() method

    def stop(self):
        """Stop AppDaemon and stop the event loop afterwards."""
        self.AD.stop_time = perf_counter()
        task = self.loop.create_task(self.AD.stop())
        task.add_done_callback(lambda _: self.loop.stop())

This will stop the event loop when all the tasks have finished, which breaks the run_forever() call in run() and causes the ExitStack to close.

The stop() method of the AppDaemon object is responsible for stopping all the subsystems in the correct order. It first sets a global stop event that all the subsystems check at the top of their respective loops. It then calls the stop methods of each subsystem in turn, waiting for each to finish before proceeding to the next one. Finally, it cancels any remaining tasks and waits for them to finish, with a timeout of 3 seconds.

Stop Event

AppDaemon shutdown is globally indicated by the stop Event getting set in the top-level AppDaemon object. All the subsystems check the status of this event using the is_set() method at the top of their respective loops, and exit if it is set. The general pattern is like this:

import asyncio
import contextlib

async def loop(self):
    while not self.AD.stop_event.is_set():
         ... # Do stuff
         with contextlib.suppress(asyncio.TimeoutError):
             await asyncio.wait_for(self.AD.stop_event.wait(), timeout=1)

Rather than using sleep() to wait between iterations, they use wait_for() to wait for the stop event with a timeout. The timeout is suppressed with suppress() so that it doesn’t raise an exception. Whenever the event is set, wait() returns immediately, which causes wait_for() to return immediately rather than waiting for the timeout.

Reference

class appdaemon.__main__.ADMain(args: Namespace)

Main application class for AppDaemon, which contains the parsed CLI arguments, top-level config model, and the async event loop.

When this class is instantiated, it creates a DependencyManager from the app directory. This causes

add_cleanup(cleanup_func, *args, **kwargs)

Add a cleanup function to be called on exit.

handle_sig(signum: int)

Function to handle signals.

Signals:

SIGUSR1 will result in internal info being dumped to the DIAG log SIGUSR2 will reload apps with modified code/config (useful in production_mode) SIGHUP will force a reload of all apps SIGINT and SIGTEM both result in AD shutting down

loop_context() → Generator[AbstractEventLoop]

Context manager that creates a new async event loop and cleans it up afterwards.

Includes the logic to install uvloop if it’s enabled.

model: MainConfig

Pydantic model of the top-level object for the appdaemon.yaml file.

run() → None

Start AppDaemon up after initial argument parsing.

This uses run_forever() on the event loop to run it indefinitely.

run_context(loop: AbstractEventLoop)

Context manager for the main run logic with exception handling.

setup_logging() → None

Set up logging configuration and timezone.

signal_handlers(loop: AbstractEventLoop)

Context manager for signal handler registration and cleanup.

stop()

Stop AppDaemon and stop the event loop afterwards.

class appdaemon.appdaemon.AppDaemon(logging: Logging, loop: AbstractEventLoop, ad_config_model: AppDaemonConfig, exit_stack: ExitStack | None = None)

Top-level container for the subsystem objects. This gets passed to the subsystem objects and stored in them as the self.AD attribute.

Asyncio:

ThreadPoolExecutor

Subsystems:

Attribute	Object
app_management	AppManagement
callbacks	Callbacks
events	Events
futures	Futures
http	HTTP
plugins	Plugins
scheduler	Scheduler
services	Services
sequences	Sequences
state	State
threading	Threading
utility	Utility

property app_dir: Path

Defined in the main YAML config under appdaemon.app_dir. Defaults to ./apps

property apps_enabled

Flag for whether disable_apps was set in the AppDaemon config

property config_dir

Path to the AppDaemon configuration files. Defaults to the first folder that has ./apps

• ~/.homeassistant

• /etc/appdaemon

executor: ThreadPoolExecutor

Executes functions from a pool of async threads. Configured with the threadpool_workers key. Defaults to 10.

loop: AbstractEventLoop

Main asyncio event loop

property real_time: bool

Flag for whether the AppDaemon instance is running in real time or not.

register_http(http: HTTP)

Sets the self.http attribute with a HTTP object and starts the admin loop.

start() → None

Start AppDaemon, which also starts all the component subsystems like the scheduler, etc.

• ThreadAsync

• Utility

Note: The scheduler is started by the utility loop after plugins are ready.

async stop() → None

Stop AppDaemon by calling the stop method of the subsystems.

This does not stop the event loop, but waits for all the existings tasks to finish before returning, which has a 3s timeout.

• AppManagement

• ThreadAsync

• Plugins

• Scheduler

• State

stop_event: Event

Flag to indicate that AppDaemon is stopping. Set by stop() and checked by subsystems.

stop_time: float = 0.0

Stores the value of perf_counter() when self.stop is first called.

property stopping: bool

Check if the AppDaemon instance is stopping.

class appdaemon.app_management.AppManagement(ad: AppDaemon)

Subsystem container for managing app lifecycles

AD: AppDaemon

Reference to the top-level AppDaemon container object

add_plugin_object(name: str, object: PluginBase) → None

Add the plugin object to the internal dictionary of ManagedObjects

property app_config: AllAppConfig

Keeps track of which module and class each app comes from, along with any associated global modules. Gets set at the end of check_config().

app_run_context(app: str, **kwargs)

Context manager for running an app to help during testing.

Parameters:

• app (str) – The name of the app to run. Must have an entry in the app_config root.

• **kwargs – Arbitrary keyword arguments representing configuration fields to temporarily update the app with.

async check_app_config_files(update_actions: UpdateActions)

Updates self.mtimes_config and self.app_config

async check_app_python_files(update_actions: UpdateActions)

Checks the python files in the app directory. Part of self.check_app_updates sequence

async check_app_updates(plugin_ns: str | None = None, mode: UpdateMode = UpdateMode.NORMAL, update_actions: UpdateActions | None = None) → None

Checks the states of the Python files that define the apps, reloading when necessary.

Called as part of utility_loop.Utility.loop()

NORMAL

Checks for changes and reloads apps as necessary.

INIT

Used during startup trigger processing the import paths and initializing the dependency manager.

TERMINATE

Adds all apps to the set to be terminated.

RELOAD_APPS

Adds all apps and the modules they depend on to the respective reload sets. Used by the app reload service.

PLUGIN_FAILED

Stops all the apps of a plugin that failed.

PLUGIN_RESTART

Restarts all the apps of a plugin that has started again.

TESTING

Testing mode, used during testing to load apps without starting them.

Parameters:

• plugin_ns (str, optional) – Namespace of a plugin to restart, if necessary. Defaults to None.

• mode (UpdateMode, optional) – Defaults to UpdateMode.NORMAL.

• update_actions (UpdateActions, optional) – The update actions to perform. Defaults to None.

property config_filecheck: FileCheck

Property that aliases the FileCheck instance for the app config files

async create_app(app: str = None, **kwargs)

Used to create an app, which is written to a config file

async create_app_object(app_name: str) → Any | None

Instantiates an app by name and stores it in self.objects.

This does not work on global module apps.

Parameters:

app_name (str) – Name of the app, as defined in a config file

Raises:

• PinOutofRange – Caused by passing in an invalid value for pin_thread

• MissingAppClass – When there’s a problem getting the class definition from the loaded module

• AppInstantiationError – When there’s another, unknown error creating the class from its definition

async edit_app(app: str, **kwargs)

Used to edit an app, which is already in Yaml. It is expecting the app’s name

enable_app(app: str)

Enable a disabled app by setting its disable flag to False.

error: Logger

Standard python logger named Error

filter_files: dict[str, float]

Dictionary of the modified times of the filter files and their paths.

get_app_config_files() → set[Path]

Get a set of valid app fonfig files in the app directory.

Valid files are ones that are readable, not inside an excluded directory, and not starting with a “.” character.

async get_app_config_files_async() → set[Path]

Get a set of valid app config files in the app directory.

Valid files are ones that are readable, not inside an excluded directory, and not starting with a “.” character.

get_app_file(app: str) → str

Used to get the file an app is located

get_python_files() → set[Path]

Get a set of valid Python files in the app directory.

Valid files are ones that are readable, not inside an excluded directory, and not starting with a “.” character.

async get_python_files_async() → set[Path]

Get a set of valid app config files in the app directory.

Valid files are ones that are readable, not inside an excluded directory, and not starting with a “.” character.

async import_module(module_name: str) → int

Reads an app into memory by importing or reloading the module it needs

async increase_active_apps(name: str)

Marks an app as active and updates the sensors for active/inactive apps.

async increase_inactive_apps(name: str)

Marks an app as inactive and updates the sensors for active/inactive apps.

logger: Logger

Standard python logger named AppDaemon._app_management

objects: dict[str, ManagedObject]

Dictionary of dictionaries with the instantiated apps, plugins, and sequences along with some metadata. Gets populated by

• self.init_object, which instantiates the app classes

• self.init_plugin_object

• self.init_sequence_object

property python_filecheck: FileCheck

Property that aliases the FileCheck instance for the app python files

async read_config_file(file: Path) → AllAppConfig

Reads a single YAML or TOML file into a pydantic model. This also sets the config_path attribute of any AppConfigs.

This function is primarily used by the create/edit/remove app methods that write yaml files.

async remove_app(app: str, **kwargs)

Used to remove an app

reversed_graph: dict[str, set[str]] = {}

Dictionary that maps full module names to sets of those that depend on them

async start() → None

Start the app management subsystem.

This method: * Initializes admin entities * Initializes the dependency manager (INIT mode) * Loads all apps (normal mode)

async start_app(app_name: str)

Initializes a new object and runs the initialize function of the app.

This does not work on global module apps because they only exist as imported modules.

Parameters:

app_name (str) – Name of the app to start

async stop() → None

Stop the app management subsystem and all the running apps.

• Calls check_app_updates() with UpdateMode.TERMINATE

async stop_app(app_name: str, *, delete: bool = False) → bool

Stops the app

Returns:

Whether stopping was successful or not

Return type:

bool

async terminate_sequence(name: str) → bool

Terminate the sequence

update_app(app: str, **kwargs)

Update the configuration of a specified app with new keyword arguments.

Parameters:

• app (str) – The name of the app to update.

• **kwargs – Arbitrary keyword arguments representing configuration fields to update.

Notes

• Dumps the app’s configuration to a dict, merges the kwargs into it, and validates the result.

• Unlike edit_app, this method does not write to a file but updates the in-memory configuration.

• Logs warnings if the app is not found or if validation fails.

class appdaemon.callbacks.Callbacks(ad: AppDaemon)

Container for storing callbacks. Modified by Events and State

AD: AppDaemon

Reference to the AppDaemon container object

diag: Logger

Standard python logger named Diag

async dump_callbacks()

Dumps info about the callbacks to the Diag log

logger: Logger

Standard python logger named AppDaemon._callbacks

class appdaemon.dependency_manager.DependencyManager(python_files: dataclasses.InitVar[Iterable[pathlib.Path]], config_files: dataclasses.InitVar[Iterable[pathlib.Path]])

Keeps track of all the python files and the app config files (either yaml or toml)

Instantiating this class will walk the app_directory with pathlib.Path.rglob to find all the files. This happens both for app config files and app python files.

The main purpose of breaking this out from AppManagement is to make it independently testable.

dependent_apps(modules: str | Iterable[str], transitive: bool = True) → set[str]

Find the apps that are dependent on any of the modules given. This includes the transitive closure of both the module and app dependencies.

dependent_modules(modules: str | Iterable[str])

Uses find_all_dependents with the reversed dependency graph to find the transitive closure of the python module dependencies.

classmethod from_app_directory(app_dir: Path, exclude: str | Iterable[str] | None = None) → DependencyManager

Creates a new instance of the dependency manager from the given app directory

get_dependent_apps(modules: Iterable[str]) → set[str]

Finds all of the apps that depend on the given modules, even indirectly

modules_from_apps(apps: str | Iterable[str], dependents: bool = True) → set[str]

Find the importable names of all the python modules that the given apps depend on.

This includes the transitive closure by default.

Parameters:

• apps (str | Iterable[str])

• dependents (bool) – Whether to include the transitive closure

update_python_files(new_files: Iterable[Path])

Updates the dependency graph of python files.

This is used to map which modules import which other modules and requires reading the contents of each python file to find the import statements.

class appdaemon.events.EventCallback(*args, **kwargs)

class appdaemon.events.Events(ad: AppDaemon)

Subsystem container for handling all events

AD: AppDaemon

Reference to the top-level AppDaemon container object

async add_event_callback(name: str, namespace: str, cb: Callable, event: str | Iterable[str] | None = None, timeout: str | int | float | timedelta | None = None, oneshot: bool = False, pin: bool | None = None, pin_thread: int | None = None, kwargs: dict[str, Any] = None) → str | list[str] | None

Add an event callback to AppDaemon’s internal dicts.

Uses the internal callback lock to ensure that the callback is added in a thread-safe manner, and adds an entity in the admin namespace to track the callback.

Includes a feature to automatically cancel the callback after a timeout, if specified.

Parameters:

• name (str) – Name of the app registering the callback. This is important because all callbacks have to be associated with an app.

• namespace (str) – Namespace to listen for the event in. All events are fired in a namespace, and this will only listen for events in that namespace.

• cb (Callable) – Callback function.

• event (str | Iterable[str]) – Name of the event.

• timeout (int, optional)

• oneshot (bool, optional) – If True, the callback will be removed after it is executed once. Defaults to False.

• kwargs – List of values to filter on, and additional arguments to pass to the callback.

Returns:

None or the reference to the callback handle.

async cancel_event_callback(name: str, handle: str, *, silent: bool = False)

Cancels an event callback.

Parameters:

• name (str) – Name of the app that registered the callback.

• handle (str) – Handle produced by listen_event() when creating the callback.

Returns:

None.

async fire_event(namespace: str, event: str, **kwargs: Any) → dict[str, Any] | None

Fires an event.

If the namespace does not have a plugin associated with it, the event will be fired locally. If a plugin is associated, the firing of the event will be delegated to the plugin, under the understanding that when the event is fired, the plugin will notify appdaemon that it occurred, usually via the system the plugin is communicating with.

Parameters:

• namespace (str) – Namespace for the event to be fired in.

• event (str) – Name of the event.

• **kwargs – Arguments to associate with the event.

Returns:

None.

async has_log_callback(name: str)

Returns True if the app has a log callback, False otherwise.

Used to prevent callback loops. In the calling logic, if this function returns True the resulting logging event will be suppressed.

Parameters:

name (str) – Name of the app.

async info_event_callback(name: str, handle: str)

Gets the information of an event callback.

Parameters:

• name (str) – Name of the app or subsystem.

• handle (str) – Previously supplied handle for the callback.

Raises:

ValueError – an invalid name or handle was provided

Returns:

A dictionary of callback entries or rise a ValueError if an invalid handle is provided.

logger: Logger

Standard python logger named AppDaemon._events

async process_event(namespace: str, data: dict[str, Any])

Processes an event that has been received either locally or from a plugin.

Parameters:

• namespace (str) – Namespace the event was fired in.

• data – Data associated with the event.

Returns:

None.

async process_event_callbacks(namespace: str, data: dict[str, Any]) → None

Processes a pure event callback.

Locate any callbacks that may be registered for this event, check for filters and if appropriate, dispatch the event for further checking and eventual action.

Parameters:

• namespace (str) – Namespace of the event.

• data – Data associated with the event.

Returns:

None.

Custom exceptions used by AppDaemon and helper functions to format them in the logs.

appdaemon.exceptions.exception_context(logger: Logger, app_dir: Path, header: str | None = None)

Context manager to handle exceptions in a block of code

class appdaemon.futures.Futures(ad: AppDaemon)

Subsystem container for managing Future objects

AD: AppDaemon

Reference to the top-level AppDaemon container object

add_future(app_name: str, future: Future | Future)

Add a future to the registry and a callback that removes itself from the registry after it finishes.

futures: dict[str, list[Future]]

Dictionary of futures registered by app name

class appdaemon.http.HTTP(ad: AppDaemon, main_cfg: MainConfig)

Handles serving the web UI

AD: AppDaemon

Reference to the AppDaemon container object

class appdaemon.logging.AppNameFormatter(fmt=None, datefmt=None, style=None)

Logger formatter to add ‘appname’ as an interpolatable field.

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

usesTime() → bool

Override to ensure asctime is always available, as LogSubscriptionHandler depends on it being available.

class appdaemon.logging.DuplicateFilter(logger: Logger, threshold: float, delay: float, timeout: float)

logging.Filter that filters duplicate messages

filter(record: LogRecord) → bool | LogRecord

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

filtering: bool

Flag to track if the filter is active or not.

class appdaemon.logging.LogSubscriptionHandler(ad: AppDaemon, type)

Handle apps that subscribe to logs.

This Handler requires that it’s formatter is an instance of AppNameFormatter.

emit(record)

Emit a record.

If a formatter is specified, it is used to format the record. The record is then written to the stream with a trailing newline. If exception information is present, it is formatted using traceback.print_exception and appended to the stream. If the stream has an ‘encoding’ attribute, it is used to determine how to do the output to the stream.

class appdaemon.logging.Logging(*args, **kwargs)

Creates and configures the Python logging. The top-level logger is called AppDaemon. Child loggers are created with get_child().

AD: AppDaemon

Reference to the top-level AppDaemon container object

async add_log_callback(namespace: str, name: str, callback: Callable, level: str | int, pin: bool | None = None, pin_thread: int | None = None, **kwargs) → list[str] | None

Adds a callback for log which is called internally by apps.

Parameters:

• namespace (str) – Namespace of the log event.

• name (str) – Name of the app.

• callback (Callable) – Callback function.

• level (str | int) – Log level

• **kwargs – List of values to filter on, and additional arguments to pass to the callback.

Returns:

None or a list of the callback handles, 1 for each logging level above the one given

async cancel_log_callback(name: str, handles: str | List[str])

Cancels log callback(s).

Parameters:

• name (str) – Name of the app or module.

• handles (Union[str, List[str]]) – Callback handle or list of them

get_access() → Logger

Returns:

Python logger named Access

Return type:

Logger

get_child(name: str) → Logger

Creates a logger with the name AppDaemon.<name>. Automatically adds a DuplicateFilter with the config options from main_log:

• filter_threshold

• filter_repeat_delay

• filter_timeout

Parameters:

name (str) – Child name for the logger.

Returns:

Child logger

Return type:

Logger

get_diag() → Logger

Returns:

Python logger named Diag

Return type:

Logger

get_error() → Logger

Gets the top-level error log

Returns:

Python logger named Error

Return type:

Logger

get_logger() → Logger

Gets the top-level log

Returns:

Python logger named AppDaemon

Return type:

Logger

async process_log_callbacks(namespace, log_data)

Process Log callbacks

register_ad(ad: AppDaemon)

Adds a reference to the top-level AppDaemon object. This is necessary because the Logging object gets created first.

class appdaemon.plugin_management.PluginBase(ad: AppDaemon, name: str, config: PluginConfig)

Base class for plugins to set up _logging

property all_namespaces: list[str]

A list of namespaces that includes the main namespace as well as any extra ones.

first_time: bool = True

Flag for this being the first time the plugin has made a connection.

The first connection a plugin makes is handled a little differently because it’ll be at startup and it’ll be before any apps have been loaded.

property namespace: str

Main namespace of the plugin

property namespaces: list[str]

Extra namespaces for the plugin

async notify_plugin_started(meta: dict, state: dict)

Notifys the AD internals that the plugin has started

• sets the namespace state in self.AD.state

• adds the plugin entity in self.AD.state

• sets the plugin object to active

• fires a plugin_started event

Parameters:

• meta (dict)

• state (dict)

class appdaemon.plugin_management.PluginManagement(ad: AppDaemon, config: dict[str, PluginConfig])

Subsystem container for managing plugins

AD: AppDaemon

Reference to the top-level AppDaemon container object

config: dict[str, PluginConfig]

Config as defined in the appdaemon.plugins section of appdaemon.yaml

error: Logger

Standard python logger named Error

get_plugin_from_namespace(namespace: str) → str

Gets the name of the plugin that’s associated with the given namespace.

This function is needed because plugins can have multiple namespaces associated with them.

logger: Logger

Standard python logger named AppDaemon._plugin_management

property plugin_dir: Path

Built-in plugin base directory

plugin_meta: dict[str, dict[str, Any]]

Dictionary storing the metadata for the loaded plugins. {<namespace>: <metadata dict>}

property plugin_names: set[str]

Names of the built-in plugins

plugin_objs: dict[str, dict[str, PluginBase | bool | str]]

Dictionary storing the instantiated plugin objects. {<namespace>: { "object": <PluginBase>, "active": <bool>, "name": <str> }}

property plugin_paths: list[Path]

Paths to the built-in plugins

process_meta(meta: dict, name: str)

Looks for certain keys in the metadata dict to override ones in the original AD config. For example, latitude and longitude from a Hass plugin

async refresh_update_time(plugin_name: str)

Updates the internal time for when the plugin’s state was last updated

async stop()

Stops all the plugins and clears the callbacks for them.

This needs to be async in order to wait for the apps to all be cleared before stopping the plugins.

async wait_for_plugins(timeout: float | None = None)

Waits for the user-configured plugin startup conditions.

Specifically, this waits for each of their ready events

class appdaemon.scheduler.Scheduler(ad: AppDaemon)

AppDaemon subsystem to manage internal scheduling, calculate the times of sun-based events, and parse datetime strings.

property active: bool

Whether the core scheduler loop is running.

async get_next_period(interval: str | int | float | timedelta, start: time | datetime | str | None = None) → datetime

Calculate the next execution time for a periodic interval.

If start is “immediate”, returns the current time. Otherwise, calculates a start time (defaulting to “now”) and advances by the interval until a future time is reached.

async get_now(tz: str | BaseTzInfo | None = None) → datetime

Get the current time for the scheduler.

The time represented will be the same regardless of the timestamp. The timezone only influences how the time is displayed.

Parameters:

tz – The timezone to use for the current time. If None, uses the timezone configured in appdaemon.yaml.

async loop()

Core scheduler loop, which processes scheduled callbacks and sleeping between them.

make_naive(dt: datetime) → datetime

Convert a timezone-aware datetime to a naive datetime in local timezone.

This is used for display purposes only. The scheduler internally works with timezone-aware UTC datetimes.

async parse_datetime(input_: str | time | datetime, aware: bool = False, today: bool | None = None, days_offset: int = 0, *, now: datetime | None = None) → datetime

Parse a variety of inputs into a datetime object.

Parameters:

• input (str | time | datetime) – The input to parse. Can be a string, time, or datetime object.

• aware (bool, optional) – If False, the resulting datetime will be naive (without timezone). Defaults to True.

• today (bool, optional) – If True, forces the result to have the same date as the now datetime. False is effectively equivalent to next. The default value is None, which doesn’t try to coerce the output at all. This results in slightly different date results for different input types. For example, a time string will be given the same date as the one in the now datetime, but a sun event string will be the datetime of the next one.

• days_offset (int, optional) – Number of days to offset from the current date for sunrise/sunset parsing. If this is negative, this will unset the today argument, which allows the result to be in the past.

• now (datetime, optional) – The current time to use as reference. If not provided, will call get_now().

property realtime: bool

Whether the scheduler is running in real time (timewarp == 1).

async reset_timer(name: str, handle: str) → bool

Only used by the ADAPI to reset an internal timer.

async restart_timer(uuid_: str, args: dict[str, Any]) → dict

Used to restart a timer. This directly modifies the internal schedule dict.

static sanitize_timer_kwargs(app: ADBase, kwargs: dict) → dict

Removes keywords from the keywords

start() → None

Starts the scheduler, which creates the the async task for loop() and adds some cleanup callbacks using the Python-native add_done_callback().

stop() → None

Stops the scheduler by cancelling the task for loop()

timer_running(name: str, handle: str) → bool

Check if the handler is still running by checking for the existence of the handle in the schedule.

class appdaemon.services.ServiceCallback(*args, **kwargs)

Simple Protocol for callbacks for service results.

__call__(result: Any) → None

Required form of the callback function for a service result.

class appdaemon.services.Services(ad: AppDaemon)

Subsystem container for handling services

AD

Reference to the AppDaemon container object

Type:

AppDaemon

services

AppDaemon’s internal service registry, which is a set of nested dicts, organized like this:

• Namespace

• Domain

• Service name

• Service Info

Type:

dict[str, dict[str, dict[str, dict[str, Any]]]]

services_lock

Re-entrant lock for preventing the service dict from being read and modified at the same time.

Type:

threading.RLock

class appdaemon.state.State(ad: AppDaemon)

Subsystem container for tracking states

AD

Reference to the AppDaemon container object

Type:

AppDaemon

async add_entity(namespace: str, entity: str, state: Any, attributes: dict | None = None) → None

Adds an entity to the internal state registry and fires the __AD_ENTITY_ADDED event

async add_namespace(namespace: str, writeback: ADWritebackType, persist: bool, name: str | None = None) → Path | Literal[False] | None

Add a state namespace.

Fires a __AD_NAMESPACE_ADDED event in the admin namespace if it’s actually added.

Parameters:

• namespace (str) – Name of the namespace to add.

• writeback (Literal["safe", "hybrid"]) – Writeback strategy for the namespace.

• persist (bool) – If True, the namespace will be persistent by saving it to a file.

• name (str, optional) – Name of the app adding the namespace.

Returns:

The path to the namespace database file if added successfully, False if it

already exists, or None if the namespace isn’t persistent.

Return type:

Path | Literal[False] | None

async add_persistent_namespace(namespace: str, writeback: ADWritebackType = ADWritebackType.safe) → Path | None

Add a namespace that’s stored in a persistent file.

This needs to be an async method to make sure it gets run from the event loop in the main thread. Otherwise, the DbfilenameShelf can get messed up because it’s not thread-safe. In some systems, it’ll complain about being accessed from multiple threads, depending on what database driver is used in the background.

async add_state_callback(name: str, namespace: str, entity: str | None, cb: StateCallback | AsyncStateCallback, timeout: str | int | float | timedelta | None = None, oneshot: bool = False, immediate: bool = False, pin: bool | None = None, pin_thread: int | None = None, kwargs: dict[str, Any] | None = None)

Add a state callback to AppDaemon’s internal dicts.

Uses the internal callback lock to ensure that the callback is added in a thread-safe manner.

Parameters:

• name – Name of the app registering the callback. This is important because all callbacks have to be associated with an app.

• namespace – Namespace of the entity to listen to.

• entity (str, optional) – Entity ID for listening to state changes. If None, the callback will be invoked for all state changes in the namespace.

• cb (StateCallbackType) – Callback function to be invoked when the state changes. Can be sync or async.

• oneshot (bool, optional) – If True, the callback will be removed after it is executed once. Defaults to False.

• immediate (bool, optional) – If True, the callback will be executed immediately if the entity is already in the new state. Defaults to False.

• kwargs (dict, optional) – Additional parameters arguments to be passed to the callback function.

Returns:

A string made from uuid4().hex that is used to identify the callback. This can be used to cancel the callback later.

close_namespaces() → None

Close all the persistent namespaces, which includes saving them.

async info_state_callback(handle: str, name: str) → tuple[str, str, Any, dict[str, Any]]

Get information about a state callback

Needs to be async to use the callback lock.

Parameters:

• handle (str) – Handle from when the callback was registered.

• name (str) – Name of the app that registered the callback. Every callback is registered under an app, so this is required to find the callback information.

Returns:

A tuple with the namespace, entity, attribute, and kwargs of the callback

async periodic_save(interval: str | int | float | timedelta) → None

Periodically save all namespaces that are persistent with writeback_type ‘hybrid’

register_state_services(namespace: str) → None

Register the set of state services for the given namespace.

async remove_entity(namespace: str, entity: str) → None

Removes an entity.

If the namespace does not have a plugin associated with it, the entity will be removed locally only. If a plugin is associated, the entity will be removed via the plugin and locally.

Parameters:

• namespace (str) – Namespace for the event to be fired in.

• entity (str) – Name of the entity.

Returns:

None.

async remove_entity_simple(namespace: str, entity_id: str) → None

Used to remove an internal AD entity

Fires the __AD_ENTITY_REMOVED event in a new task

async remove_namespace(namespace: str) → PersistentDict | dict[str, Any] | None

Remove a state namespace. Must not be configured by the appdaemon.yaml file, and must have been added by an app.

Fires an __AD_NAMESPACE_REMOVED event in the admin namespace if it’s actually removed.

async remove_persistent_namespace(namespace: str, state: PersistentDict) → Path | None

Used to remove the file for a created namespace

save_hybrid_namespaces() → None

Save all the persistent namespaces with the hybrid writeback type

async set_state(name: str, namespace: str, entity: str, _silent: bool = False, *, state: Any | None = None, attributes: dict | None = None, replace: bool = False, **kwargs: Any) → dict[str, Any]

async set_state(name: str, namespace: str, entity: str, _silent: bool = False, **kwargs: Any) → dict[str, Any]

Sets the internal state of an entity.

Fires the state_changed event under the namespace, and uses relevant plugin objects based on namespace.

Parameters:

• name – Only used for a log message

• namespace

• entity

• __silent

• state

• attributes

• replace

set_state_simple(namespace: str, entity_id: str, state: Any)

Set state without any checks or triggering amy events, and only if the entity exists

update_namespace_state(namespace: str | list[str], state: dict)

Uses the update method of dict

If the namespace argument is a list, then the state is expected to be a dictionary with each

class appdaemon.thread_async.ThreadAsync(ad: AppDaemon)

Module to translate from the thread world to the async world via queues

start() → None

Start the thread_async loop

stop()

Stops the thread/async loop by putting a sentinel value in the queue.

class appdaemon.utility_loop.Utility(ad: AppDaemon)

Subsystem container for managing the utility loop

Checks for file changes, overdue threads, thread starvation, and schedules regular state refreshes.

AD: AppDaemon

Reference to the AppDaemon container object

async _init_loop()

Initialize the utility loop components.

• Sets up stats

• Starts the web server if configured

• Waits for all plugins to initialize

• Registers services

• Starts the scheduler

• Initializes apps if apps are enabled

_loop_iteration_context() → AsyncGenerator[LoopTiming]

Async context manager for running the utility loop().

• Contains logic for warnings

  • Exceptions are logged with tracebacks, but not raised

  • Warnings are logged if the utility loop takes too long

• Handles the timing of the utility loop

Yields:

LoopTiming – Timing object for recording operation timestamps.

app_update_event: Event

Event that gets set and cleared _loop_iteration_context() method, which wraps each iteration of the while loop in loop()

async get_uptime() → timedelta

Get the uptime of AppDaemon as a timedelta.

async loop()

Run the utility loop, which handles the following:

• Checking for file changes to update/reload apps if necessary with check_app_updates()

• Checking for thread starvation

• Checking for overdue threads

• Save hybrid namespaces

• Gives the plugins a chance to run their own utility functions

• Updates performance entities

loop_task: Task

The task for the loop() method

async sleep(delay: float, *, timeout_ok: bool)

Sleep for a specified number of seconds.

The purpose of this method is to make sleeping easily and quickly interruptible. This is done by using asyncio.wait_for() to wait for the stop event to be set, and (usually) ignoring the timeout.

Parameters:

• seconds (float) – Number of seconds to sleep.

• timeout_ok (bool) – If True, does not raise TimeoutError if the sleep is interrupted by stop_event.

start() → None

Starts the utility loop by creating the async task for the loop() method.