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(*args, **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

property app_dir: Path

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

property apps

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: 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.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, use_dictionary_unpacking: bool = False) 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().

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)

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

Called as part of utility_loop.Utility.loop()

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

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

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

  • AppClassSignatureError – When the class has the wrong number of inputs on its __init__ method

  • 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

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_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 (str) – Name of the app to start

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

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

futures

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)

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

http

class appdaemon.http.HTTP(ad: 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.

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.

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: int)

Function to handle signals.

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

main()

Initial AppDaemon entry point.

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

run(ad_config_model: AppDaemonConfig, *args, http)

Start AppDaemon up after initial argument parsing.

Parameters:
  • ad_config_model – Config for AppDaemon Object.

  • *args – Gets used to create the HTTP object.

  • http – Main HTTP config

stop()

Called by the signal handler to shut AD down.

Returns:

None.

exception appdaemon.__main__.NoADConfig
appdaemon.__main__.main()

Called when run from the command line.

plugins

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:
stopping: bool = False

Flag that indicates whether AppDaemon is currently shutting down.

class appdaemon.plugin_management.PluginManagement(ad: AppDaemon, config: Mapping[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, PluginBase]

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

stopping: bool

Flag for if PluginManagement should be shutting down

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

scheduler

services

class appdaemon.services.ServiceCallback(*args, **kwargs)
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

Clear any services registered by the app with the given name.

deregister_service(namespace: str, domain: str, service: str, name: str) bool

Used to unregister a service

register_service(namespace: str, domain: str, service: str, callback: Callable, silent: bool = False, name: str | None = None, **kwargs: Any) None

Register a service with AppDaemon. This method should only be used by AppDaemon internals.

Services are tracked with a nested dicts structure.

Parameters:
  • namespace (str) – Namespace of the service

  • domain (str) – Domain of the service

  • service (str) – Name of the service

  • callback (Callable) – Callback function to be called when the service is invoked

  • __silent (bool, optional) – If True, do not send a registration event. Defaults to False.

  • __name (str | None, optional) – Name of the app registering the service. Defaults to None.

  • **kwargs – Additional keyword arguments to be passed to the callback function.

sequences

class appdaemon.sequences.Sequences(ad: AppDaemon)

Subsystem container for managing sequences

AD

Reference to the AppDaemon container object

Type:

AppDaemon

static normalized(sequence: str) str

Ensures the sequence name is prefixed with sequence.

async run_sequence(calling_app: str, namespace: str, sequence: str | list[dict[str, dict[str, str]]]) Task

Prepares the sequence and creates a task to run it

state

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: str | dict[str, 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: str, persist: bool, name: str | None = None) Path | bool | None

Used to Add Namespaces from Apps

async add_persistent_namespace(namespace: str, writeback: str) Path

Used to add a database file for a created namespace.

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.

async add_state_callback(name: str, namespace: str, entity: str | None, cb: StateCallback, 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)

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 (StateCallback) – Callback function to be invoked when the state changes.

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

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 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) dict[str, Any] | None

Used to Remove Namespaces from Apps

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

async remove_persistent_namespace(namespace: str) Path

Used to remove the file for a created namespace

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

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.state.StateCallback(*args, **kwargs)

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.threads.Threading(ad: AppDaemon)

Subsystem container for managing Thread objects

AD: AppDaemon

Reference to the AppDaemon container object

async calculate_pin_threads()

Assigns thread numbers to apps that are supposed to be pinned

async check_constraint(key, value, app: ADBase)

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

determine_thread(name: str, pin: bool | None, pin_thread: int | None) tuple[bool, int | None]

Determine whether the app should be pinned to a thread and which one.

Applies defaults from app management

Returns:

A tuple of (pin, pin_thread) where pin is True if the app should be pinned and pin_thread is the thread ID number

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

get_pinned_apps(thread: str)

Gets the names of apps that are pinned to a particular thread

async get_q_update()

Updates queue sizes

log_lock: allocate_lock

Threadsafe lock that helps prevent blocks of log output from different threads being mixed together

logger: Logger

Standard python logger named AppDaemon._threading

property pin_apps: bool

Whether each app should be pinned to a thread

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.

property total_threads: int

Number of threads created for apps.

By default this is automatically calculated, but can also be manually configured by the user in appdaemon.yaml.

validate_pin(name: str, pin_thread: int | None) None

Check to see if the ID for the pin thread is valid.

Raises:

PinOutofRange – if the pin_thread is not valid.

Returns:

None

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 get_uptime() timedelta

Utility function to return the uptime of AppDaemon

Returns:

The uptime of AppDaemon

Return type:

datetime.timedelta

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.PersistentDict(filename: str | Path, safe: bool, *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)
class appdaemon.utils.Subsystem(*args, **kwargs)

AppDaemon internal subsystem protocol.

AD: AppDaemon

Reference to the top-level AppDaemon object

name: str

Used for registering futures, and maybe other things?

appdaemon.utils.clean_kwargs(**kwargs)

Converts everything to strings and removes null values

appdaemon.utils.deep_compare(check: dict, data: dict) bool

Compares 2 nested dictionaries of values

appdaemon.utils.dt_to_str(dt: <module 'datetime' from '/home/docs/.asdf/installs/python/3.11.12/lib/python3.11/datetime.py'>, tz: ~datetime.tzinfo | None = None, *, round: bool = False) str | Literal['never']

Convert a datetime object to a string.

This function provides a single place for standardizing the conversion of datetimes to strings.

Parameters:
  • dt (datetime) – The datetime object to convert.

  • tz (tzinfo, optional) – Optional timezone to apply. Defaults to None.

  • round (bool, optional) – Whether to round the datetime to the nearest second. Defaults to False.

appdaemon.utils.executor_decorator(func: Callable[[...], R]) Callable[[...], Coroutine[Any, Any, R]]

Decorate a sync function to turn it into an async function that runs in a separate thread.

appdaemon.utils.format_timedelta(td: str | int | float | timedelta | None) str

Format a timedelta object into a human-readable string.

There are different brackets for lengths of time that will format the strings differently.

Uses parse_timedelta to convert the input into a timedelta object before formatting the string.

Examples

>>> format_timedelta(0.025374)
'25.374ms'
>>> format_timedelta(0.687)
'687ms'
>>> format_timedelta(2.5)
'2.5s'
>>> format_timedelta(25)
'25s'
>>> format_timedelta(None)
'never'
>>> format_timedelta(0)
'No time'
appdaemon.utils.get_object_size(obj, seen=None)

Recursively finds size of objects

appdaemon.utils.has_expanded_kwargs(func)

Determines whether or not to use keyword argument expansion on this function by finding if there’s a **kwargs expansion somewhere.

Handles unwrapping (removing decorators) if necessary.

appdaemon.utils.make_endpoint(base: str, endpoint: str) str

Formats a URL appropriately with slashes

appdaemon.utils.parse_timedelta(s: str | int | float | timedelta | None) timedelta

Convert disparate types into a timedelta object.

Parameters:

s (str | int | float | timedelta | None) – The value to convert. Can be a string, int, float, or timedelta. Numbers get interpreted as seconds. Strings can in different formats either HH:MM:SS, MM:SS, or SS.

Returns:

Timedelta object.

Examples

>>> parse_timedelta(0.025374)
datetime.timedelta(microseconds=25374)
>>> parse_timedelta(0.687)
datetime.timedelta(microseconds=687000)
>>> parse_timedelta(2.5)
datetime.timedelta(seconds=2, microseconds=500000)
>>> parse_timedelta("25")
datetime.timedelta(seconds=25)
>>> parse_timedelta("02:30")
datetime.timedelta(seconds=150)
>>> parse_timedelta("00:00:00")
datetime.timedelta(0)
appdaemon.utils.read_config_file(file: Path, app_config: bool = False) dict[str, dict | list]

Reads a single YAML or TOML file.

This includes all the mechanics for including secrets and environment variables.

Parameters:

app_config – Flag for whether to add the config_path key to the loaded dictionaries

appdaemon.utils.recursive_get_files(base: Path, suffix: str, exclude: set[str] | None = None) Generator[Path, None, None]

Recursively generate file paths.

Parameters:
  • base (Path) – The base directory to start searching from.

  • suffix (str) – The file extension to filter by.

  • exclude (set[str]) – A set of directory names to exclude from the search.

Yields:

Path objects to files that have the matching extension and are readable.

appdaemon.utils.run_coroutine_threadsafe(self: ADBase, coro: Coroutine[Any, Any, R], timeout: str | int | float | timedelta | None = None) R

Run an instantiated coroutine (async) from sync code.

This wraps the native python function asyncio.run_coroutine_threadsafe with logic to add a timeout. See scheduling from other threads for more details.

Parameters:
  • self (ADBase) – Needs to have a self.AD attribute with a reference to the AppDaemon object.

  • coro (Coroutine) – An instantiated coroutine that hasn’t been awaited.

  • timeout (float | None, optional) – Optional timeout to use. If no value is provided then the value set in appdaemon.internal_function_timeout in the appdaemon.yaml file will be used.

Returns:

Result from the coroutine

async appdaemon.utils.run_in_executor(self: Subsystem, fn: Callable[[...], R], *args, **kwargs) R

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

appdaemon.utils.sync_decorator(coro_func: Callable[[P], Awaitable[R]]) Callable[[P], R]

Wrap a coroutine function to ensure it gets run in the main thread.

This allows users to run async ADAPI methods as if they were regular sync methods. It works by checking to see if the function is being run in the main thread, which has the async event loop in it. If it is the main loop, then it creates a task and returns it. If it isn’t, then it runs the coroutine in the main thread using run_coroutine_threadsafe.

See scheduling from other threads for more details.

appdaemon.utils.warning_decorator(start_text: str | None = None, success_text: str | None = None, error_text: str | None = None, finally_text: str | None = None, reraise: bool = False) Callable[[Callable[[...], Coroutine[Any, Any, R]]], Callable[[...], Coroutine[Any, Any, R]]]

Decorate an async function to log messages at various stages around it running.

By default this does not reraise any exceptions that occur during the execution of the wrapped function.

Only works on methods of AppDaemon subsystems because it uses the attributes:
  • self.logger

  • self.AD

Raises:

By default, only ever re-raises an AppDaemonException

appdaemon.utils.write_config_file(file: Path, **kwargs)

Writes a single YAML or TOML file.