Internal Documentation

These notes are intended to assist anyone that wants to understand AppDaemon’s internals better. Most modules are used from within the AppDaemon object, which is a centralized depository of configuration information and references to the other objects and subsystems within AppDaemon.

appdaemon object

class appdaemon.appdaemon.AppDaemon(logging: Logging, loop: BaseEventLoop, **kwargs)

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
app_dir: str | Path

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

apps: bool

Flag for whether disable_apps was set in the AppDaemon config

config_dir: str | Path

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: BaseEventLoop

Main asyncio event loop

register_http(http: HTTP)

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

stop()

Called by the signal handler to shut AD down.

Also stops

admin

admin_loop

class appdaemon.admin_loop.AdminLoop(ad: AppDaemon)

Called by register_http(). Loop timed with admin_delay

AD: AppDaemon

Reference to the AppDaemon container object

logger: Logger

Standard python logger named AppDaemon._admin_loop

async loop()

Handles calling get_callback_update() and get_q_update()

app_management

class appdaemon.app_management.AppActions(init: ~typing.Dict[str, int] = <factory>, term: ~typing.Dict[str, int] = <factory>, total: int = 0, active: int = 0)

Stores which apps to initialize and terminate, as well as the total number of apps and the number of active apps.

init

Dictionary of apps to initialize, which ultimately happens in AppManagement._load_apps() as part of AppManagement.check_app_updates()

Type:

Dict[str, int]

term

Dictionary of apps to terminate, which ultimately happens in AppManagement._terminate_apps() as part of AppManagement.check_app_updates()

Type:

Dict[str, int]

total

Total number of apps

Type:

int

active

Number of active apps

Type:

int

class appdaemon.app_management.AppManagement(ad: AppDaemon, use_toml: bool)

Subsystem container for managing app lifecycles

AD: AppDaemon

Reference to the top-level AppDaemon container object

app_config: Dict[str, Dict[str, Dict[str, bool]]]

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().

apps_per_module(module_name: str)

Finds which apps came from a given module name

async check_app_updates(plugin: str = None, mode: UpdateMode = UpdateMode.NORMAL)

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

Called as part of utility_loop.Utility.loop()

Parameters:

  • plugin (str, optional) – Plugin to restart, if necessary. Defaults to None.

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

Check Process:

  • Refresh modified times of monitored files.

  • Checks for deleted files

  • Marks the apps for reloading or removal as necessary

  • Restarts the plugin, if specified

  • Terminates apps as necessary

  • Loads or reloads modules/pacakges as necessary

  • Loads apps from the modules/packages

async check_config(silent: bool = False, add_threads: bool = True) → AppActions | None

Wraps read_config()

Parameters:

  • silent (bool, optional) – _description_. Defaults to False.

  • add_threads (bool, optional) – _description_. Defaults to True.

Returns:

AppActions object with information about which apps to initialize and/or terminate

create_app(app=None, **kwargs)

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

edit_app(app, **kwargs)

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

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_deps_and_prios(applist: Iterable[str], mode: UpdateMode) → Dict[str, float]

Gets the dependencies and priorities for the given apps

Parameters:

  • applist (Iterable[str]) – Iterable of app names

  • mode (UpdateMode) – UpdateMode

Returns:

_description_

Return type:

_type_

get_app_file(app)

Used to get the file an app is located

get_app_from_file(file)

Finds the apps that depend on a given file

get_file_from_module(module_name: str) → Path | None

Gets the module __file__ based on the module name.

Parameters:

mod (str) – Module name

Returns:

Path of the __file__

Return type:

Optional[Path]

get_path_from_app(app_name: str) → Path

Gets the module path based on the app_name

Used in self._terminate_apps

async init_object(app_name: str)

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

Parameters:

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

init_sequence_object(name, object)

Initialize the sequence

logger: Logger

Standard python logger named AppDaemon._app_management

modules: Dict[str, ModuleType]

Dictionary of the loaded modules and their names

monitored_files: Dict[str | Path, float]

Dictionary of the Python files that are being watched for changes and their last modified times

objects: Dict[str, Dict[str, Any]]

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

read_app(reload_cfg: ModuleLoad)

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

async read_config() → Dict[str, Dict[str, Any]]

Walks the apps directory and reads all the config files with read_config_file(), which reads individual config files and runs in the executor.

Returns:

Loaded app configuration

Return type:

Dict[str, Dict[str, Any]]

read_config_file(file) → Dict[str, Dict]

Reads a single YAML or TOML file.

remove_app(app, **kwargs)

Used to remove an app

Seems to be unreferenced?

async terminate_sequence(name: str) → bool

Terminate the sequence

use_toml: bool

Whether to use TOML files for configuration

class appdaemon.app_management.ModuleLoad(path: Path, reload: bool = False)

Dataclass containing settings for calls to AppManagement.read_app()

path

Filepath of the module or path to the __init__.py of a package.

Type:

pathlib.Path

reload

Whether to reload the app using importlib.reload

Type:

bool

name

Importable name of the module/package

Type:

str

class appdaemon.app_management.UpdateMode(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Used as an argument for AppManagement.check_app_updates() to set the mode of the check.

INIT

Triggers AppManagement._init_update_mode to run during check_app_updates

NORMAL

Normal update mode, for when AppManagement.check_app_updates() is called by utility_loop.Utility.loop()

TERMINATE

Terminate all apps

callbacks

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

dashboard

events

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, namespace, cb, event, **kwargs)

Adds a callback for an event which is called internally by apps.

Parameters:

  • name (str) – Name of the app.

  • namespace (str) – Namespace of the event.

  • cb – Callback function.

  • event (str) – Name of the event.

  • **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, handle)

Cancels an event callback.

Parameters:

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

  • handle – Previously supplied callback handle for the callback.

Returns:

None.

async fire_event(namespace: str, event: str, **kwargs)

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, data)

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.

futures

class appdaemon.futures.Futures(ad: AppDaemon)

Subsystem container for managing Future objects

AD

Reference to the AppDaemon container object

Type:

AppDaemon

http

class appdaemon.http.HTTP(ad: AppDaemon, loop, logging, appdaemon, dashboard, old_admin, admin, api, http)

Handles serving the web UI

AD: AppDaemon

Reference to the AppDaemon container object

appdaemon.http.route_secure(myfunc)

Take care of streams and service calls

appdaemon.http.secure(myfunc)

Take care of screen based security

appdaemon.http.securedata(myfunc)

Take care of streams and service calls

logging

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.

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

logging.Filter that filters duplicate messages

filter(record: LogRecord) → bool

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(config: Dict | None = None, log_level: str = 'INFO')

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, cb: Callable, level, **kwargs)

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.

  • cb – Callback function.

  • event (str) – Name of the event.

  • **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_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.

main

AppDaemon main() module.

AppDaemon module that contains main() along with argument parsing, instantiation of the AppDaemon and HTTP Objects, also creates the loop and kicks everything off

class appdaemon.__main__.ADMain

Class to encapsulate all main() functionality.

handle_sig(signum, frame)

Function to handle signals.

SIGUSR1 will result in internal info being dumped to the DIAG log SIGHUP will force a reload of all apps SIGINT and SIGTEM both result in AD shutting down

Parameters:

  • signum – Signal number being processed.

  • frame – frame - unused

Returns:

None.

init_signals()

Setup signal handling.

main()

Initial AppDaemon entry point.

Parse command line arguments, load configuration, set up logging.

run(appdaemon: AppDaemon, hadashboard, admin, aui, api, http)

Start AppDaemon up after initial argument parsing.

Parameters:

  • appdaemon – Config for AppDaemon Object.

  • hadashboard – Config for HADashboard Object.

  • admin – Config for admin Object.

  • aui – Config for aui Object.

  • api – Config for API Object

  • http – Config for HTTP Object

Returns:

None.

stop()

Called by the signal handler to shut AD down.

Returns:

None.

appdaemon.__main__.main()

Called when run from the command line.

plugins

class appdaemon.plugin_management.PluginBase(ad: AppDaemon, name, args)

Base class for plugins to set up _logging

class appdaemon.plugin_management.Plugins(ad: AppDaemon, kwargs)

Subsystem container for managing plugins

AD: AppDaemon

Reference to the top-level AppDaemon container object

error: Logger

Standard python logger named Error

logger: Logger

Standard python logger named AppDaemon._plugin_management

plugin_meta: Dict[str, dict]

Dictionary storing the metadata for the loaded plugins

plugin_objs: Dict[str, Any]

Dictionary storing the instantiated plugin objects

scheduler

services

class appdaemon.services.Services(ad: AppDaemon)

Subsystem container for handling services

AD

Reference to the AppDaemon container object

Type:

AppDaemon

clear_services(name: str) → None

Used to clear services

deregister_service(namespace: str, domain: str, service: str, **kwargs: dict) → bool

Used to unregister a service

async run_service(coro: Awaitable) → Any

Used to process a service call

sequences

class appdaemon.sequences.Sequences(ad: AppDaemon)

Subsystem container for managing sequences

AD

Reference to the AppDaemon container object

Type:

AppDaemon

async loop_step(namespace: str, command: str, parameters: dict, loop_step: dict) → None

Used to loop a step command

state

class appdaemon.state.State(ad: AppDaemon)

Subsystem container for tracking states

AD

Reference to the AppDaemon container object

Type:

AppDaemon

async add_namespace(namespace, writeback, persist, name=None)

Used to Add Namespaces from Apps

add_persistent_namespace(namespace, writeback)

Used to add a database file for a created namespace

async remove_entity(namespace, entity)

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

async remove_namespace(namespace)

Used to Remove Namespaces from Apps

remove_persistent_namespace(namespace)

Used to remove the file for a created namespace

stream

thread_async

class appdaemon.thread_async.ThreadAsync(ad: AppDaemon)

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

threading

class appdaemon.threading.Threading(ad: AppDaemon, kwargs)

Subsystem container for managing Thread objects

AD: AppDaemon

Reference to the AppDaemon container object

async check_constraint(key, value, app)

Used to check Constraint

async check_days_constraint(args, name)

Used to check days Constraint

async check_state_constraint(args, new_state, name)

Used to check state Constraint

async check_time_constraint(args, name)

Used to check time Constraint

diag: Logger

Standard python logger named Diag

async get_callback_update()

Updates the sensors with information about how many callbacks have been fired. Called by the AdminLoop

  • sensor.callbacks_average_fired

  • sensor.callbacks_average_executed

async get_q_update()

Updates queue sizes

logger: Logger

Standard python logger named AppDaemon._threading

threads: Dict[str, Dict[str, Thread | Queue]]

Dictionary with keys of the thread ID (string beginning with thread-) and values of another dictionary with thread and queue keys that have values of Thread and Queue objects respectively.

utility_loop

Module to handle utility functions within AppDaemon.

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 loop()

The main utility loop.

Loops until stop() is called, checks for file changes, overdue threads, thread starvation, and schedules regular state refreshes.

stop()

Called by the AppDaemon object to terminate the loop cleanly

Returns:

None

utils

class appdaemon.utils.AttrDict(*args, **kwargs)

Dictionary subclass whose entries can be accessed by attributes (as well as normally).

static from_nested_dict(data)

Construct nested AttrDicts from nested dictionaries.

class appdaemon.utils.EntityStateAttrs(dict)
class appdaemon.utils.PersistentDict(filename, safe, *args, **kwargs)

Dict-like object that uses a Shelf to persist its contents.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

class appdaemon.utils.StateAttrs(dict)
appdaemon.utils.get_object_size(obj, seen=None)

Recursively finds size of objects

appdaemon.utils.read_config_file(path) → Dict[str, Dict]

Reads a single YAML or TOML file.

appdaemon.utils.recursive_reload(module: ModuleType, reloaded: set = None)

Recursive reload function that does a depth-first search through all sub-modules and reloads them in the correct order of dependence.

Adapted from https://gist.github.com/KristianHolsheimer/f139646259056c1dffbf79169f84c5de

async appdaemon.utils.run_in_executor(self, fn, *args, **kwargs) → Any

Runs the function with the given arguments in the instance of ThreadPoolExecutor in the top-level AppDaemon object.

Parameters:

  • self – Needs to have an AD attribute with the AppDaemon object

  • fn (function) – Function to run in the executor

  • *args – Any positional arguments to use with the function

  • **kwargs – Any keyword arguments to use with the function

Returns:

Whatever the function returns