AppDaemon APIs

The AppDaemon API comes in the form of a class called ADAPI, which provides high-level functionality for users to create their apps. This includes common functions such as listening for events/state changes, scheduling, manipulating entities, and calling services. The API is designed to be easy to use and understand, while still providing the power and flexibility needed to create complex automations.

App Creation

To use the API, create a new class that inherits from ADAPI and implement the initialize() method. This method is required for all apps and is called when the app is started.

from appdaemon.adapi import ADAPI


class MyApp(ADAPI):
    def initialize(self):
        self.log("MyApp is starting")

        # Use any of the ADAPI methods
        # handle = self.listen_state(...)
        # handle = self.listen_event(...)
        # handle = self.run_in(...)
        # handle = self.run_every(...)

Alternatively, the ADBase class can be used, which can provide some advantages, such as being able to access APIs for plugins in mulitple namespaces.

from appdaemon.adapi import ADAPI
from appdaemon.adbase import ADBase
from appdaemon.plugins.mqtt import Mqtt


class MyApp(ADBase):
    adapi: ADAPI    # This type annotation helps your IDE with autocompletion
    mqttapi: Mqtt

    def initialize(self):
        self.adapi = self.get_ad_api()
        self.adapi.log("MyApp is starting")

        # This requires having defined a plugin in the mqtt namespace in appdaemon.yaml
        self.mqttapi = self.get_plugin_api('mqtt')

        # Use any of the ADAPI methods through self.adapi
        # handle = self.adapi.listen_state(...)
        # handle = self.adapi.listen_event(...)
        # handle = self.adapi.run_in(...)
        # handle = self.adapi.run_every(...)

Entity Class

Interacting with entities is a core part of writing automation apps, so being able to easily access and manipulate them is important. AppDaemon supports this by providing entities as python objects.

The Entity class is essentially a light wrapper around ADAPI methods that pre-fills some arguments. Because of this, the entity doesn’t have to actually exist for the Entity object to be created and used. If the entity doesn’t exist, some methods will fail, but others will not. For example, get_state() will fail, but calling set_state() for an entity that doesn’t exist will create it. This is useful for creating sensor entities that are available in Home Assistant.

from appdaemon.adapi import ADAPI


class MyApp(ADAPI):
    def initialize(self):
        self.log("MyApp is starting")

        # Get light entity class
        self.kitchen_light = self.get_entity("light.kitchen_ceiling_light")

        # Assign a callback for when the state changes to on
        self.kitchen_light.listen_state(
            self.state_callback,
            attribute="brightness",
            new='on'
        )

    def state_callback(self, entity, attribute, old, new, **kwargs):
        self.log(f'{self.kitchen_light.friendly_name} turned on')

Services

AppDaemon provides some services from some built-in namespaces. These services can be called from any app, provided they use the correct namepsace. These services are listed below

Note: A service call always uses the app’s default namespace. See the section on namespaces for more information.

admin

app/create

Used to create a new app. For this service to be used, the module must be existing and provided with the module’s class. If no app name is given, the module name will be used as the app’s name by default. The service call also accepts app_file if wanting to create the app within a certain yaml file. Or app_dir, if wanting the created app’s yaml file within a certain directory. If no file or directory is given, by default the app yaml file will be generated in a directory ad_apps, using the app’s name. It should be noted that app_dir and app_file when specified, will be created within the AD’s apps directory.

data = {}
data["module"] = "web_app"
data["class"] = "WebApp"
data["namespace"] = "admin"
data["app"] = "web_app3"
data["endpoint"] = "endpoint3"
data["app_dir"] = "web_apps"
data["app_file"] = "web_apps.yaml"

self.call_service("app/create", **data)

app/edit

Used to edit an existing app. This way, an app’ args can be edited in realtime with new args

self.call_service("app/edit", app="light_app", module="light_system", namespace="admin")

app/remove

Used to remove an existing app. This way, an existing app will be deleted. If the app is the last app in the yaml file, the file will be deleted

self.call_service("app/remove", app="light_app", namespace="admin")

app/start

Starts an app that has been terminated. The app name arg is required.

self.call_service("app/start", app="light_app", namespace="admin")

app/stop

Stops a running app. The app name arg is required.

self.call_service("app/stop", app="light_app", namespace="admin")

app/restart

Restarts a running app. This service basically stops and starts the app. The app name arg is required.

self.call_service("app/restart", app="light_app", namespace="admin")

app/reload

Checks for an app update. Useful if AD is running in production mode, and app changes need to be checked and loaded.

self.call_service("app/reload", namespace="admin")

app/enable

Enables a disabled app, so it can be loaded by AD.

self.call_service("app/enable", app="living_room_app", namespace="admin")

app/disable

Disables an enabled app, so it cannot be loaded by AD. This service call is persistent, so even if AD restarts, the app will not be restarted

self.call_service("app/disable", app="living_room_app", namespace="admin")

production_mode/set

Sets the production mode AD is running on. The value of the mode arg has to be True or False.

self.call_service("production_mode/set", mode=True, namespace="admin")

All namespaces except global, and admin:

state/add_entity

Adds an existing entity to the required namespace.

self.call_service(
    "state/set",
    entity_id="sensor.test",
    state="on",
    attributes={"friendly_name" : "Sensor Test"},
    namespace="default"
)

state/set

Sets the state of an entity. This service allows any key-worded args to define what entity’s values need to be set.

self.call_service(
    "state/set",
    entity_id="sensor.test",
    state="on",
    attributes={"friendly_name" : "Sensor Test"},
    namespace="default"
)

state/remove_entity

Removes an existing entity from the required namespace.

self.call_service("state/remove_entity", entity_id="sensor.test", namespace="default")

All namespaces except admin:

event/fire

Fires an event within the specified namespace. The event arg is required.

self.call_service("event/fire", event="test_event", entity_id="appdaemon.test", namespace="hass")

rules

sequence/run

Runs a predefined sequence. The entity_id arg with the sequence full-qualified entity name is required.

self.call_service("sequence/run", entity_id ="sequence.christmas_lights", namespace="rules")

sequence/cancel

Cancels a predefined sequence. The entity_id arg with the sequence full-qualified entity name is required.

self.call_service("sequence/cancel", entity_id ="sequence.christmas_lights", namespace="rules")

Reference

Entity API

appdaemon.entity.Entity.add(self, state: str | int | float = None, attributes: dict = None) → None

Adds a non-existent entity, by creating it within a namespaces.

It should be noted that this api call, is mainly for creating AD internal entities. If wanting to create an entity within an external system, do check the system’s documentation

Parameters:

state (optional) – The state the new entity is to have
attributes (optional) – The attributes the new entity is to have

Returns:

None

Examples

>>> self.my_entity = self.get_entity("zigbee.living_room_light")

create the entity entity.

>>> self.my_entity.add(state="off", attributes={"friendly_name": "Living Room Light"})

appdaemon.entity.Entity.call_service(self, service: str, timeout: str | int | float | None = None, callback: Callable[[Any], Any] | None = None, **data: Any) → Any

Calls an entity supported service within AppDaemon.

This function can call only services that are tied to the entity, and provide any required parameters.

Parameters:

service (str) – The service name, without the domain (e.g “toggle”)
return_result (bool, option) – If return_result is provided and set to True AD will attempt to wait for the result, and return it after execution
callback – The non-async callback to be executed when complete.
**kwargs – Each service has different parameter requirements. This argument allows you to specify a comma-separated list of keyword value pairs, e.g., state = on. These parameters will be different for every service and can be discovered using the developer tools.

Returns:

Result of the call_service function if any

Examples

HASS

>>> self.my_entity = self.get_entity("light.office_1")
>>> self.my_entity.call_service("turn_on", color_name="red")

appdaemon.entity.Entity.copy(self, copy: bool = True) → dict

Gets the complete state of the entity within AD.

This is essentially a helper function, to get all data about an entity

Parameters:: copy (bool) – If set to False, it will not make a deep copy of the entity. This can help with speed of accessing the data

appdaemon.entity.Entity.exists(self) → bool: Checks the existence of the entity in AD.

appdaemon.entity.Entity.get_state(self, attribute: str | None = None, default: Any | None = None, copy: bool = True) → Any

Get the state of an entity from AppDaemon’s internals.

Home Assistant emits a state_changed event for every state change, which it sends to AppDaemon over the websocket connection made by the plugin. Appdaemon uses the data in these events to update its internal state. This method returns values from this internal state, so it does not make any external requests to Home Assistant.

Other plugins that emit state_changed events will also have their states tracked internally by AppDaemon.

It’s common for entities to have a state that’s always one of on, off, or unavailable. This applies to entities in the light, switch, binary_sensor, and input_boolean domains in Home Assistant, among others.

Parameters:

attribute (str, optional) – Optionally specify an attribute to return. If not used, the state of the entity will be returned. The value all can be used to return the entire state dict rather than a single value.
default (any, optional) – The value to return when the entity or the attribute doesn’t exist.
copy (bool, optional) – Whether to return a copy of the internal data. This is True by default in order to protect the user from accidentally modifying AppDaemon’s internal data structures, which is dangerous and can cause undefined behvaior. Only set this to False for read-only operations.

Returns:

The entire state of the entity at that given time, if if get_state() is called with no parameters. This will consist of a dictionary with a key for each entity. Under that key will be the standard entity state information.

Examples

>>> self.my_entity = self.get_entity("light.office_1")

Get the state attribute of light.office_1.

>>> state = self.my_entity.get_state()

Get the brightness attribute of light.office_1.

>>> state = self.my_entity.get_state(attribute="brightness")

Get the entire state of light.office_1.

>>> state = self.my_entity.get_state(attribute="all")

appdaemon.entity.Entity.listen_state(self, callback: StateCallback, **kwargs: Any) → str

Registers a callback to react to state changes.

This function allows the user to register a callback for a wide variety of state changes.

Parameters:

callback – Function that will be called when the callback gets triggered. It must conform to the standard state callback format documented here
new (str | Callable[[Any], bool], optional) – If given, the callback will only be invoked if the state of the selected attribute (usually state) matches this value in the new data. The data type is dependent on the specific entity and attribute. Values that look like ints or floats are often actually strings, so be careful when comparing them. The self.get_state() method is useful for checking the data type of the desired attribute. If new is a callable (lambda, function, etc), then it will be called with the new state, and the callback will only be invoked if the callable returns True.
old (str | Callable[[Any], bool], optional) – If given, the callback will only be invoked if the selected attribute (usually state) changed from this value in the new data. The data type is dependent on the specific entity and attribute. Values that look like ints or floats are often actually strings, so be careful when comparing them. The self.get_state() method is useful for checking the data type of the desired attribute. If old is a callable (lambda, function, etc), then it will be called with the old state, and the callback will only be invoked if the callable returns True.
duration (str | int | float | timedelta, optional) –
If supplied, the callback will not be invoked unless the desired state is maintained for that amount of time. This requires that a specific attribute is specified (or the default of state is used), and should be used in conjunction with either or both of the new and old parameters. When the callback is called, it is supplied with the values of entity, attr, old, and new that were current at the time the actual event occurred, since the assumption is that none of them have changed in the intervening period.

If you use duration when listening for an entire device type rather than a specific entity, or for all state changes, you may get unpredictable results, so it is recommended that this parameter is only used in conjunction with the state of specific entities.
attribute (str, optional) – Optional name of an attribute to use for the new/old checks. If not specified, the default behavior is to use the value of state. Using the value all will cause the callback to get triggered for any change in state, and the new/old values used for the callback will be the entire state dict rather than the individual value of an attribute.
timeout (str | int | float | timedelta, optional) – If given, the callback will be automatically removed after that amount of time. If activity for the listened state has occurred that would trigger a duration timer, the duration timer will still be fired even though the callback has been removed.
immediate (bool, optional) –
If given, it enables the countdown for a delay parameter to start at the time. If the duration parameter is not given, the callback runs immediately. What this means is that after the callback is registered, rather than requiring one or more state changes before it runs, it immediately checks the entity’s states based on given parameters. If the conditions are right, the callback runs immediately at the time of registering. This can be useful if, for instance, you want the callback to be triggered immediately if a light is already on, or after a duration if given.

If immediate is in use, and new and duration are both set, AppDaemon will check if the entity is already set to the new state and if so it will start the clock immediately. If new and duration are not set, immediate will trigger the callback immediately and report in its callback the new parameter as the present state of the entity. If attribute is specified, the state of the attribute will be used instead of state. In these cases, old will be ignored and when the callback is triggered, its state will be set to None.
oneshot (bool, optional) – If True, the callback will be automatically removed after the first time it gets invoked.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Note

The old and new args can be used singly or together.

Returns:: A string that uniquely identifies the callback and can be used to cancel it later if necessary. Since variables created within object methods are local to the function they are created in, it’s recommended to store the handles in the app’s instance variables, e.g. self.handle.

Examples

>>> self.my_entity = self.get_entity("light.office_1")

Listen for a state change involving light.office1 and return the state attribute.

>>> self.handle = self.my_entity.listen_state(self.my_callback)

Listen for a change involving the brightness attribute of light.office1 and return the brightness attribute.

>>> self.handle = self.my_entity.listen_state(self.my_callback, attribute="brightness")

Listen for a state change involving light.office1 turning on and return the state attribute.

>>> self.handle = self.my_entity.listen_state(self.my_callback, new="on")

Listen for a change involving light.office1 changing from brightness 100 to 200.

>>> self.handle = self.my_entity.listen_state(
    self.my_callback,
    attribute="brightness",
    old=100,
    new=200
)

Listen for a state change involving light.office1 changing to state on and remaining on for a minute.

>>> self.handle = self.my_entity.listen_state(self.my_callback, new="on", duration=60)

Listen for a state change involving light.office1 changing to state on and remaining on for a minute trigger the delay immediately if the light is already on.

>>> self.handle = self.my_entity.listen_state(self.my_callback, new="on", duration=60, immediate=True)

appdaemon.entity.Entity.is_state(self, state: Any) → bool

Checks the state of the entity against the given state

This helper function supports using both iterable and non-iterable data

Parameters:: state (any) – The state or iterable set of state data, to check against

Example

>>> light_entity_object.is_state("on")
>>> media_object.is_state(["playing", "paused"])

appdaemon.entity.Entity.set_namespace(self, namespace: str) → None

Set a new namespace for the Entity to use from that point forward.

This doesn’t change anything about the entity itself, but it does change the namespace that this instance of the entity API references. There might not be an entity with the same ID in the new namespace.

Parameters:: namespace (str) – Name of the new namespace
Returns:: None.

Examples

Get an entity

>>> self.light = self.get_entity("light.living_room")

Copy the full state from the entity

>>> state = self.my_entity.copy()

Set the new namespace

>>> self.light.set_namespace("my_namespace")

Set the state of the entity

>>> self.light.set_state(**entity_data)

Verify

>>> self.light.get_state(attribute="all")

appdaemon.entity.Entity.set_state(self, state: Any | None = None, attributes: dict | None = None, replace: bool = False, **kwargs) → dict

Update the state of the specified entity.

This causes a state_changed event to be emitted in the entity’s namespace. If that namespace is associated with a Home Assistant plugin, it will use the /api/states/<entity_id> endpoint of the REST API to update the state of the entity. This method can be useful to create entities in Home Assistant, but they won’t persist across restarts.

Parameters:

state (optional) – New state value to be set.
attributes (dict[str, Any], optional) – Optional dictionary to use for the attributes. If replace is False, then the attribute dict will use the built-in update method on this dict. If replace is True, then the attribute dict will be entirely replaced with this one.
replace (bool, optional) – Whether to replace rather than update the attributes. Defaults to False. For plugin based entities, this is not recommended, as the plugin will mostly replace the new values, when next it updates.
**kwargs (optional) – Zero or more keyword arguments. These will be applied to the attributes.

Returns:

A dictionary that represents the new state of the updated entity.

Examples

>>> self.my_entity = self.get_entity("light.living_room")

Update the state of an entity.

>>> self.my_entity.set_state(state="off")

Update the state and attribute of an entity.

>>> self.my_entity.set_state(state = "on", attributes = {"color_name": "red"})

appdaemon.entity.Entity.toggle(self, **kwargs: Any | None) → Any

Generic function, used to toggle the entity ON/OFF if supported. This function will attempt to call the toggle service if registered, either by an app or plugin within the entity’s namespace. So therefore its only functional, if the service toggle exists within the namespace the entity is operating in.

The keyword arguments accepted will vary depending on the namespace and associated plugin.

https://www.home-assistant.io/integrations/light/#action-lighttoggle

Keyword Arguments:: **kwargs (optional) – Zero or more keyword arguments. These will be applied to the attributes.

appdaemon.entity.Entity.turn_off(self, **kwargs: Any | None) → Any

Generic function, used to turn the entity OFF if supported. This function will attempt to call the turn_off service if registered, either by an app or plugin within the entity’s namespace. So therefore its only functional, if the service turn_off exists within the namespace the entity is operating in.

The keyword arguments accepted will vary depending on the namespace and associated plugin.

https://www.home-assistant.io/integrations/light/#action-lightturn_off

Keyword Arguments:: **kwargs (optional) – Zero or more keyword arguments. These will be applied to the attributes.

appdaemon.entity.Entity.turn_on(self, **kwargs) → Any

Generic helper function, used to turn the entity ON if supported. This function will attempt to call the turn_on service if registered, either by an app or plugin within the entity’s namespace. So therefore its only functional, if the service turn_on exists within the namespace the entity is operating in.

The keyword arguments accepted will vary depending on the namespace and associated plugin.

https://www.home-assistant.io/integrations/light/#action-lightturn_on

Keyword Arguments:: **kwargs (optional) – Zero or more keyword arguments. These will be applied to the attributes.

async appdaemon.entity.Entity.wait_state(self, state: Any, attribute: str | int = None, duration: int | float = 0, timeout: int | float = None) → None

Used to wait for the state of an entity’s attribute

This API call should only be used async. It should be noted that when instantiated, the api checks immediately if it’s already in the required state and will continue if it is.

Parameters:

state (Any) – The state to wait for, for the entity to be in before continuing
attribute (str, optional) – The entity’s attribute to use, if not using the entity’s state
duration (int, float) – How long the state is to hold, before continuing
timeout (int, float) – How long to wait for the state to be achieved, before timing out. When it times out, a appdaemon.exceptions.TimeOutException is raised

Returns:

None

Examples

>>> from appdaemon.exceptions import TimeOutException
>>>
>>> async def run_my_sequence(self):
>>>     sequence_object = self.get_entity("sequence.run_the_thing")
>>>     await sequence_object.call_service("run")
>>>     try:
>>>         await sequence_object.wait_state("idle", timeout=30) # wait for it to completely run
>>>     except TimeOutException:
>>>         pass # didn't complete on time

In addition to the above, there are a couple of property attributes the Entity class supports: - entity_id - namespace - domain - entity_name - state - attributes - friendly_name - last_changed - last_changed_seconds

State

Get the state of an entity from AppDaemon’s internals.

Home Assistant emits a state_changed event for every state change, which it sends to AppDaemon over the websocket connection made by the plugin. Appdaemon uses the data in these events to update its internal state. This method returns values from this internal state, so it does not make any external requests to Home Assistant.

Other plugins that emit state_changed events will also have their states tracked internally by AppDaemon.

It’s common for entities to have a state that’s always one of on, off, or unavailable. This applies to entities in the light, switch, binary_sensor, and input_boolean domains in Home Assistant, among others.

Parameters:

entity_id (str, optional) – Full entity ID or just a domain. If a full entity ID is provided, the result will be for that entity only. If a domain is provided, the result will be a dict that maps the entity IDs to their respective results.
attribute (str, optional) – Optionally specify an attribute to return. If not used, the state of the entity will be returned. The value all can be used to return the entire state dict rather than a single value.
default (any, optional) – The value to return when the entity or the attribute doesn’t exist.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. The current namespace can be changed using self.set_namespace. See the namespace documentation for more information.
copy (bool, optional) – Whether to return a copy of the internal data. This is True by default in order to protect the user from accidentally modifying AppDaemon’s internal data structures, which is dangerous and can cause undefined behvaior. Only set this to False for read-only operations.

Returns:

The state or attribute of the entity ID provided or a dict of that maps entity IDs to their respective results. If called with no parameters, this will return the entire state dict.

Examples

Get the state of the entire system.

>>> state = self.get_state()

Get the state of all switches in the system.

>>> state = self.get_state("switch")

Get the state attribute of light.office_1.

>>> state = self.get_state("light.office_1")

Get the brightness attribute of light.office_1.

>>> state = self.get_state("light.office_1", attribute="brightness")

Get the entire state of light.office_1.

>>> state = self.get_state("light.office_1", attribute="all")

appdaemon.adapi.ADAPI.set_state(self, entity_id: str, state: Any | None = None, namespace: str | None = None, attributes: dict[str, Any] | None = None, replace: bool = False, check_existence: bool = True, **kwargs: Any) → dict[str, Any]

Update the state of the specified entity.

This causes a state_changed event to be emitted in the entity’s namespace. If that namespace is associated with a Home Assistant plugin, it will use the /api/states/<entity_id> endpoint of the REST API to update the state of the entity. This method can be useful to create entities in Home Assistant, but they won’t persist across restarts.

Parameters:

entity_id (str) – The fully qualified entity id (including the device type).
state – New state value to be set.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. See the namespace documentation for more information.
attributes (dict[str, Any], optional) – Optional dictionary to use for the attributes. If replace is False, then the attribute dict will use the built-in update method on this dict. If replace is True, then the attribute dict will be entirely replaced with this one.
replace (bool, optional) – Whether to replace rather than update the attributes. Defaults to False. For plugin based entities, this is not recommended, as the plugin will mostly replace the new values, when next it updates.
check_existence (bool, optional) – Whether to check if the entity exists before setting the state. Defaults to True, but it can be useful to set to False when using this method to create an entity.
**kwargs (optional) – Zero or more keyword arguments. Extra keyword arguments will be assigned as attributes.

Returns:

A dictionary that represents the new state of the updated entity.

Examples

Update the state of an entity.

>>> self.set_state("light.office_1", state="off")

Update the state and attribute of an entity.

>>> self.set_state(entity_id="light.office_1", state="on", attributes={"color_name": "red"})

Update the state of an entity within the specified namespace.

>>> self.set_state("light.office_1", state="off", namespace="hass")

Registers a callback to react to state changes.

The callback needs to have the following form:

>>> def my_callback(self, entity: str, attribute: str, old: Any, new: Any, **kwargs: Any) -> None: ...

Parameters:

callback – Function that will be called when the callback gets triggered. It must conform to the standard state callback format documented here
entity_id (str | Iterable[str], optional) – Entity ID or a domain. If a domain is provided, e.g., light, or binary_sensor the callback will be triggered for state changes of any entities in that domain. If a list of entities is provided, the callback will be registered for each of those entities.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. See the namespace documentation for more information. Using the value global will register the callback for all namespaces.
new (str | Callable[[Any], bool], optional) – If given, the callback will only be invoked if the state of the selected attribute (usually state) matches this value in the new data. The data type is dependent on the specific entity and attribute. Values that look like ints or floats are often actually strings, so be careful when comparing them. The self.get_state() method is useful for checking the data type of the desired attribute. If new is a callable (lambda, function, etc), then it will be called with the new state, and the callback will only be invoked if the callable returns True.
old (str | Callable[[Any], bool], optional) – If given, the callback will only be invoked if the selected attribute (usually state) changed from this value in the new data. The data type is dependent on the specific entity and attribute. Values that look like ints or floats are often actually strings, so be careful when comparing them. The self.get_state() method is useful for checking the data type of the desired attribute. If old is a callable (lambda, function, etc), then it will be called with the old state, and the callback will only be invoked if the callable returns True.
duration (str | int | float | timedelta, optional) –
If supplied, the callback will not be invoked unless the desired state is maintained for that amount of time. This requires that a specific attribute is specified (or the default of state is used), and should be used in conjunction with either or both of the new and old parameters. When the callback is called, it is supplied with the values of entity, attr, old, and new that were current at the time the actual event occurred, since the assumption is that none of them have changed in the intervening period.

If you use duration when listening for an entire device type rather than a specific entity, or for all state changes, you may get unpredictable results, so it is recommended that this parameter is only used in conjunction with the state of specific entities.
attribute (str, optional) – Optional name of an attribute to use for the new/old checks. If not specified, the default behavior is to use the value of state. Using the value all will cause the callback to get triggered for any change in state, and the new/old values used for the callback will be the entire state dict rather than the individual value of an attribute.
timeout (str | int | float | timedelta, optional) – If given, the callback will be automatically removed after that amount of time. If activity for the listened state has occurred that would trigger a duration timer, the duration timer will still be fired even though the callback has been removed.
immediate (bool, optional) –
If given, it enables the countdown for a delay parameter to start at the time. If the duration parameter is not given, the callback runs immediately. What this means is that after the callback is registered, rather than requiring one or more state changes before it runs, it immediately checks the entity’s states based on given parameters. If the conditions are right, the callback runs immediately at the time of registering. This can be useful if, for instance, you want the callback to be triggered immediately if a light is already on, or after a duration if given.

If immediate is in use, and new and duration are both set, AppDaemon will check if the entity is already set to the new state and if so it will start the clock immediately. If new and duration are not set, immediate will trigger the callback immediately and report in its callback the new parameter as the present state of the entity. If attribute is specified, the state of the attribute will be used instead of state. In these cases, old will be ignored and when the callback is triggered, its state will be set to None.
oneshot (bool, optional) – If True, the callback will be automatically removed after the first time it gets invoked.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Note

The old and new args can be used singly or together.

Returns:: A string that uniquely identifies the callback and can be used to cancel it later if necessary. Since variables created within object methods are local to the function they are created in, it’s recommended to store the handles in the app’s instance variables, e.g. self.handle.

Examples

Listen for any state change and return the state attribute.

>>> self.handle = self.listen_state(self.my_callback)

Listen for any state change involving a light and return the state attribute.

>>> self.handle = self.listen_state(self.my_callback, "light")

Listen for a state change involving light.office1 and return the state attribute.

>>> self.handle = self.listen_state(self.my_callback, entity_id="light.office_1")

Listen for a state change involving light.office1 and return the entire state as a dict.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", attribute = "all")

Listen for a change involving the brightness attribute of light.office1 and return the brightness attribute.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", attribute = "brightness")

Listen for a state change involving light.office1 turning on and return the state attribute.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", new = "on")

Listen for a state change involving light.office1 turning on when the previous state was not unknown or unavailable, and return the state attribute.

>>> self.handle = self.listen_state(
    self.my_callback,
    "light.office_1",
    new="on",
    old=lambda x: x.lower() not in {"unknown", "unavailable"}
)

Listen for a change involving light.office1 changing from brightness 100 to 200 and return the brightness attribute.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", attribute="brightness", old="100", new="200")

Listen for a state change involving light.office1 changing to state on and remaining on for a minute.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", new="on", duration=60)

Listen for a state change involving light.office1 changing to state on and remaining on for a minute trigger the delay immediately if the light is already on.

>>> self.handle = self.listen_state(self.my_callback, "light.office_1", new="on", duration=60, immediate=True)

Listen for a state change involving light.office1 and light.office2 changing to state on.

>>> self.handle = self.listen_state(self.my_callback, ["light.office_1", "light.office2"], new="on")

appdaemon.adapi.ADAPI.cancel_listen_state(self, handle: str, name: str | None = None, silent: bool = False) → bool

Cancel a listen_state() callback.

This will prevent any further calls to the callback function. Other state callbacks will not be affected.

Parameters:

handle – The handle returned when the listen_state() call was made.
name (str, optional) – The name of the app that registered the callback. Defaults to the name of the current app. This is useful if you want to get the information of a callback registered by another app.
silent (bool, optional) – If True, no warning will be issued if the handle is not found.

Returns:

Boolean.

Examples

>>> self.cancel_listen_state(self.office_light_handle)

Don’t display a warning if the handle is not found.

>>> self.cancel_listen_state(self.dummy_handle, silent=True)

appdaemon.adapi.ADAPI.info_listen_state(self, handle: str, name: str | None = None) → tuple[str, str, Any, dict[str, Any]]

Get information on state a callback from its handle.

Parameters:

handle (str) – The handle returned when the listen_state() call was made.
name (str, optional) – The name of the app that registered the callback. Defaults to the name of the current app. This is useful if you want to get the information of a callback registered by another app.

Returns:

The values supplied for namespace, entity, attribute, and kwargs when the callback was initially created.

Examples

>>> namespace, entity, attribute, kwargs = self.info_listen_state(self.handle)

Time

appdaemon.adapi.ADAPI.parse_utc_string(self, utc_string: str) → float

Convert a UTC to its string representation.

Parameters:: utc_string (str) – A string that contains a date and time to convert.
Returns:: An POSIX timestamp that is equivalent to the date and time contained in utc_string.

appdaemon.adapi.ADAPI.get_tz_offset(self) → float: Returns the timezone difference between UTC and Local Time in minutes.

appdaemon.adapi.ADAPI.convert_utc(self, utc: str) → datetime

Gets a datetime object for the specified UTC.

Home Assistant provides timestamps of several different sorts that can be used to gain additional insight into state changes. These timestamps are in UTC and are coded as ISO 8601 combined date and time strings. This function will accept one of these strings and convert it to a localised Python datetime object representing the timestamp.

Parameters:: utc – An ISO 8601 encoded date and time string in the following format: 2016-07-13T14:24:02.040658-04:00
Returns:: A localised Python datetime object representing the timestamp.

appdaemon.adapi.ADAPI.sun_up(self) → bool

Determines if the sun is currently up.

Returns:: True if the sun is up, False otherwise.
Return type:: bool

Examples

>>> if self.sun_up():
>>>    #do something

appdaemon.adapi.ADAPI.sun_down(self) → bool

Determines if the sun is currently down.

Returns:: True if the sun is down, False otherwise.
Return type:: bool

Examples

>>> if self.sun_down():
>>>    #do something

appdaemon.adapi.ADAPI.parse_time(self, time_str: str, name: str | None = None, aware: bool = False, today: bool = False, days_offset: int = 0) → time

Creates a time object from its string representation.

This functions takes a string representation of a time, or sunrise, or sunset offset and converts it to a datetime.time object.

Parameters:

time_str (str) –
A string representation of the datetime with one of the following formats:
1. HH:MM:SS[.ss] - the time in Hours Minutes, Seconds and Microseconds, 24 hour format.
b. sunrise|sunset [+|- HH:MM:SS[.ss]] - time of the next sunrise or sunset with an optional positive or negative offset in Hours Minutes, Seconds and Microseconds.

c. N deg rising|setting - time the sun will be at N degrees of elevation while either rising or setting
If the HH:MM:SS.ss format is used, the resulting datetime object will have today’s date.
name (str, optional) – Name of the calling app or module. It is used only for logging purposes.
aware (bool, optional) – If True the created datetime object will be aware of timezone.
today (bool, optional) – Instead of the default behavior which is to return the next sunrise/sunset that will occur, setting this flag to true will return today’s sunrise/sunset even if it is in the past
days_offset (int, optional) – Specify the number of days (positive or negative) for the sunset/sunrise. This can only be used in combination with the today flag

Returns:

A time object, representing the time given in the time_str argument.

Examples

>>> self.parse_time("17:30:00")
17:30:00

>>> time = self.parse_time("sunrise")
04:33:17

>>> time = self.parse_time("sunset + 00:30:00")
19:18:48

>>> time = self.parse_time("sunrise + 01:00:00")
05:33:17

appdaemon.adapi.ADAPI.parse_datetime(self, time_str: str, name: str | None = None, aware: bool = False, today: bool = False, days_offset: int = 0) → datetime

Creates a datetime object from its string representation.

This function takes a string representation of a date and time, or sunrise, or sunset offset and converts it to a datetime object.

Parameters:

time_str (str) –
A string representation of the datetime with one of the following formats:
a. YY-MM-DD-HH:MM:SS[.ss] - the date and time in Year, Month, Day, Hours, Minutes, Seconds and Microseconds, 24 hour format.
1. HH:MM:SS[.ss] - the time in Hours Minutes, Seconds and Microseconds, 24 hour format.
c. sunrise|sunset [+|- HH:MM:SS[.ss]] - time of the next sunrise or sunset with an optional positive or negative offset in Hours Minutes, Seconds and Microseconds.
If the HH:MM:SS.ss format is used, the resulting datetime object will have today’s date.
name (str, optional) – Name of the calling app or module. It is used only for logging purposes.
aware (bool, optional) – If True the created datetime object will be aware of timezone.
today (bool, optional) – Instead of the default behavior which is to return the next sunrise/sunset that will occur, setting this flag to true will return today’s sunrise/sunset even if it is in the past
days_offset (int, optional) – Specify the number of days (positive or negative) for the sunset/sunrise. This can only be used in combination with the today flag

Returns:

A datetime object, representing the time and date given in the time_str argument.

Examples

>>> self.parse_datetime("2018-08-09 17:30:00")
2018-08-09 17:30:00

>>> self.parse_datetime("17:30:00.01")
2019-08-15 17:30:00.010000

>>> self.parse_datetime("sunrise")
2019-08-16 05:33:17

>>> self.parse_datetime("sunset + 00:30:00")
2019-08-16 19:18:48

>>> self.parse_datetime("sunrise + 01:00:00")
2019-08-16 06:33:17

appdaemon.adapi.ADAPI.get_now(self, aware: bool = True) → datetime

Returns the current Local Date and Time.

Examples

>>> self.get_now()
2019-08-16 21:17:41.098813-04:00

appdaemon.adapi.ADAPI.get_now_ts(self, aware: bool = False) → float

Returns the current Local Timestamp.

Examples

>>> self.get_now_ts()
1565990318.728324

appdaemon.adapi.ADAPI.now_is_between(self, start_time: str | datetime, end_time: str | datetime, name: str | None = None, now: str | None = None) → bool

Determine if the current time is within the specified start and end times.

This function takes two string representations of a time ()or sunrise or sunset offset) and returns true if the current time is between those 2 times. Its implementation can correctly handle transitions across midnight.

Parameters:

start_time (str) – A string representation of the start time.
end_time (str) – A string representation of the end time.
name (str, optional) – Name of the calling app or module. It is used only for logging purposes.
now (str, optional) – If specified, now is used as the time for comparison instead of the current time. Useful for testing.

Returns:

True if the current time is within the specified start and end times, otherwise False.

Return type:

bool

Note

The string representation of the start_time and end_time should follows one of these formats:

HH:MM:SS - the time in Hours Minutes and Seconds, 24 hour format.

b. sunrise|sunset [+|- HH:MM:SS]- time of the next sunrise or sunset with an optional positive or negative offset in Hours Minutes, and Seconds.

Examples

>>> if self.now_is_between("17:30:00", "08:00:00"):
>>>     #do something

>>> if self.now_is_between("sunset - 00:45:00", "sunrise + 00:45:00"):
>>>     #do something

appdaemon.adapi.ADAPI.sunrise(self, aware: bool = False, today: bool = False, days_offset: int = 0) → datetime

Return a datetime object that represent when a sunrise will occur.

Parameters:

aware (bool, optional) – Whether the resulting datetime object will be aware of timezone.
today (bool, optional) – Defaults to False, which will return the first sunrise in the future, regardless of the day. If set to True, the function will return the sunrise for the current day, even if it is in the past.
days_offset (int, optional) – Specify the number of days (positive or negative) for the sunrise. This can only be used in combination with the today flag

Examples

>>> self.sunrise()
2023-02-02 07:11:50.150554

>>> self.sunrise(today=True)
2023-02-01 07:12:20.272403

appdaemon.adapi.ADAPI.sunset(self, aware: bool = False, today: bool = False, days_offset: int = 0) → datetime

Return a datetime object that represent when a sunset will occur.

Parameters:

aware (bool, optional) – Whether the resulting datetime object will be aware of timezone.
today (bool, optional) – Defaults to False, which will return the first sunset in the future, regardless of the day. If set to True, the function will return the sunset for the current day, even if it is in the past.
days_offset (int, optional) – Specify the number of days (positive or negative) for the sunset. This can only be used in combination with the today flag

Examples

>>> self.sunset()
2023-02-01 18:09:00.730704

>>> self.sunset(today=True, days_offset=1)
2023-02-02 18:09:46.252314

appdaemon.adapi.ADAPI.time(self) → time

Get a time object representing the current local time.

Use this instead of the standard Python methods in order to correctly account for the time when using the time travel feature, which is usually done for testing.

Examples

>>> self.time()
20:15:31.295751

appdaemon.adapi.ADAPI.datetime(self, aware: bool = False) → datetime

Get a datetime object representing the current local date and time.

Use this instead of the standard Python methods in order to correctly account for the time when using the time travel feature, which is usually done for testing.

Parameters:: aware (bool, optional) – Whether the resulting datetime object will be aware of timezone.

Examples

>>> self.datetime()
2019-08-15 20:15:55.549379

appdaemon.adapi.ADAPI.date(self) → date

Get a date object representing the current local date.

Use this instead of the standard Python methods in order to correctly account for the time when using the time travel feature, which is usually done for testing.

Examples

>>> self.date()
2019-08-15

appdaemon.adapi.ADAPI.get_timezone(self) → str: Returns the current time zone.

Scheduler

Run a function once, at the specified time of day.

Parameters:

callback – Function that will be called at the specified time. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time) – Time the callback will be triggered. It should be either a Python time object, datetime object, or a parse_time() formatted string that specifies when the callback will occur. If the time specified is in the past, the callback will occur the next day at the specified time.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run at 10:30am today, or 10:30am tomorrow if it is already after 10:30am.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, datetime.time(10, 30, 0))

Run today at 04:00pm using the parse_time() function.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "04:00:00 PM")

Run at sunset.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "sunset")

Run an hour after sunrise.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "sunrise + 01:00:00")

Run a function after a specified delay.

This method should always be used instead of time.sleep().

Parameters:

callback – Function that will be called after the specified delay. It must conform to the standard scheduler callback format documented here.
delay (str, int, float, datetime.timedelta) – Delay before the callback is executed. Numbers will be interpreted as seconds. Strings can be in the format of HH:MM, HH:MM:SS, or DD days, HH:MM:SS. If a timedelta object is given, it will be used as is.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run the specified callback after 0.5 seconds.

>>> def delayed_callback(self, **kwargs): ... # example callback
>>> self.handle = self.run_in(self.delayed_callback, 0.5)

Run the specified callback after 2.7 seconds with a custom keyword arg title.

>>> def delayed_callback(self, title: str, **kwargs): ... # example callback
>>> self.handle = self.run_in(self.delayed_callback, 2.7, title="Delayed Callback Title")

Run a function once, at the specified time of day. This is essentially an alias for run_at().

Parameters:

callback – Function that will be called at the specified time. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time) – Time the callback will be triggered. It should be either a Python time object, datetime object, or a parse_time() formatted string that specifies when the callback will occur. If the time specified is in the past, the callback will occur the next day at the specified time.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run at 10:30am today, or 10:30am tomorrow if it is already after 10:30am.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, datetime.time(10, 30, 0))

Run today at 04:00pm using the parse_time() function.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "04:00:00 PM")

Run at sunset.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "sunset")

Run an hour after sunrise.

>>> def delayed_callback(self, **kwargs): ...  # example callback
>>> handle = self.run_once(self.delayed_callback, "sunrise + 01:00:00")

Run a function at a regular time interval.

Parameters:

callback – Function that will be called at the specified time interval. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time, datetime.datetime, optional) –
Start time for the interval calculation. If this is in the future, this will be the first time the callback is triggered. If this is in the past, the intervals will be calculated forward from the start time, and the first trigger will be the first interval in the future.
- If this is a str it will be parsed with parse_time().
- If this is a datetime.time object, the current date will be assumed.
- If this is a datetime.datetime object, it will be used as is.
interval (str, int, float, datetime.timedelta) –
Time interval between callback triggers.
- If this is an int or float, it will be interpreted as seconds.
- If this is a str it will be parsed with parse_timedelta()
  - HH:MM
  - HH:MM:SS
  - DD days, HH:MM:SS
- If this is a timedelta object, the current date will be assumed.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number, which start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before an event, or positive to denote a random offset after an event.

Examples

Run every 10 minutes starting now.

  class MyApp(ADAPI):
      def initialize(self):
          self.run_every(self.timed_callback, interval=datetime.timedelta(minutes=10))

      def timed_callback(self, **kwargs): ...  # example callback

Run every 5 minutes starting in 5 seconds.

  class MyApp(ADAPI):
      def initialize(self):
          self.run_every(self.timed_callback, "now+5", 5 * 60)

      def timed_callback(self, **kwargs): ...  # example callback

Run every 17 minutes starting in 2 hours time.

  class MyApp(ADAPI):
      def initialize(self):
          start = self.get_now() + datetime.timedelta(hours=2)
          interval = datetime.timedelta(minutes=17)
          self.run_every(self.timed_callback, start, interval)

      def timed_callback(self, **kwargs): ...  # example callback

Run a function at the same time every day.

Parameters:

callback – Function that will be called every day at the specified time. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time, datetime.datetime, optional) –
Start time for the interval calculation. If this is in the future, this will be the first time the callback is triggered. If this is in the past, the intervals will be calculated forward from the start time, and the first trigger will be the first interval in the future.
- If this is a str it will be parsed with parse_time().
- If this is a datetime.time object, the current date will be assumed.
- If this is a datetime.datetime object, it will be used as is.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number, which start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run every day at 10:30am.

>>> self.run_daily(self.daily_callback, datetime.time(10, 30))

Run at 7:30pm every day using the parse_time() function.

>>> handle = self.run_daily(self.daily_callback, "07:30:00 PM")

Run every day at sunrise.

>>> handle = self.run_daily(self.daily_callback, "sunrise")

Run every day an hour after sunset.

>>> handle = self.run_daily(self.daily_callback, "sunset + 01:00:00")

Run a function at the same time every hour.

Parameters:

callback – Function that will be called every hour starting at the specified time. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time, datetime.datetime, optional) –
Start time for the interval calculation. If this is in the future, this will be the first time the callback is triggered. If this is in the past, the intervals will be calculated forward from the start time, and the first trigger will be the first interval in the future.
- If this is a str it will be parsed with parse_time().
- If this is a datetime.time object, the current date will be assumed.
- If this is a datetime.datetime object, it will be used as is.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number, which start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run every hour, on the hour.

>>> runtime = datetime.time(0, 0, 0)
>>> self.run_hourly(self.run_hourly_c, runtime)

Run the callback at the same time every minute.

Parameters:

callback – Function that will be called every hour starting at the specified time. It must conform to the standard scheduler callback format documented here.
start (str, datetime.time, datetime.datetime, optional) –
Start time for the interval calculation. If this is in the future, this will be the first time the callback is triggered. If this is in the past, the intervals will be calculated forward from the start time, and the first trigger will be the first interval in the future.
- If this is a str it will be parsed with parse_time().
- If this is a datetime.time object, the current date will be assumed.
- If this is a datetime.datetime object, it will be used as is.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number, which start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer later before it’s been executed.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run every minute on the minute.

>>> time = datetime.time(0, 0, 0)
>>> self.run_minutely(self.run_minutely_c, time)

appdaemon.adapi.ADAPI.run_at_sunset(self, callback: Callable, *args, repeat: bool = False, offset: int | None = None, random_start: int | None = None, random_end: int | None = None, pin: bool | None = None, pin_thread: int | None = None, **kwargs) → str

Runs a callback every day at or around sunset.

Parameters:

callback – Function to be invoked at or around sunset. It must conform to the standard Scheduler Callback format documented here.
*args – Arbitrary positional arguments to be provided to the callback function when it is triggered.
offset (int, optional) – The time in seconds that the callback should be delayed after sunset. A negative value will result in the callback occurring before sunset. This parameter cannot be combined with random_start or random_end.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Example using timedelta.

>>> self.run_at_sunset(self.sun, offset = datetime.timedelta(minutes = -45).total_seconds())

Or you can just do the math yourself.

>>> self.run_at_sunset(self.sun, offset = 30 * 60)

Run at a random time +/- 60 minutes from sunset.

>>> self.run_at_sunset(self.sun, random_start = -60*60, random_end = 60*60)

Run at a random time between 30 and 60 minutes before sunset.

>>> self.run_at_sunset(self.sun, random_start = -60*60, random_end = 30*60)

appdaemon.adapi.ADAPI.run_at_sunrise(self, callback: Callable, *args, repeat: bool = False, offset: int | None = None, random_start: int | None = None, random_end: int | None = None, pin: bool | None = None, pin_thread: int | None = None, **kwargs) → str

Runs a callback every day at or around sunrise.

Parameters:

callback – Function to be invoked at or around sunrise. It must conform to the standard Scheduler Callback format documented here.
*args – Arbitrary positional arguments to be provided to the callback function when it is invoked.
offset (int, optional) – The time in seconds that the callback should be delayed after sunrise. A negative value will result in the callback occurring before sunrise. This parameter cannot be combined with random_start or random_end.
random_start (int, optional) – Start of range of the random time.
random_end (int, optional) – End of range of the random time.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is triggered.

Returns:

A handle that can be used to cancel the timer.

Note

The random_start value must always be numerically lower than random_end value, they can be negative to denote a random offset before and event, or positive to denote a random offset after an event.

Examples

Run 45 minutes before sunrise.

>>> self.run_at_sunrise(self.sun, offset = datetime.timedelta(minutes = -45).total_seconds())

Or you can just do the math yourself.

>>> self.run_at_sunrise(self.sun, offset = 30 * 60)

Run at a random time +/- 60 minutes from sunrise.

>>> self.run_at_sunrise(self.sun, random_start = -60*60, random_end = 60*60)

Run at a random time between 30 and 60 minutes before sunrise.

>>> self.run_at_sunrise(self.sun, random_start = -60*60, random_end = 30*60)

appdaemon.adapi.ADAPI.timer_running(self, handle: str) → bool

Check if a previously created timer is still running.

Parameters:: handle (str) – The handle returned from the original call to create the timer.
Returns:: Boolean representing whether the timer is still running.

Examples

>>> self.timer_running(handle)
True

appdaemon.adapi.ADAPI.cancel_timer(self, handle: str, silent: bool = False) → bool

Cancel a previously created timer.

Parameters:

handle (str) – The handle returned from the original call to create the timer.
silent (bool, optional) – Set to True to suppress warnings if the handle is not found. Defaults to False.

Returns:

Boolean representing whether the timer was successfully canceled.

Examples

>>> self.cancel_timer(handle)
True

>>> self.cancel_timer(handle, silent=True)

appdaemon.adapi.ADAPI.info_timer(self, handle: str) → tuple[datetime, int, dict] | None

Get information about a previously created timer.

Parameters:

handle (str) – The handle returned from the original call to create the timer.

Returns:

A tuple with the following values or None if handle is invalid or timer no longer exists.

time - datetime object representing the next time the callback will be fired
interval - repeat interval if applicable, 0 otherwise.
kwargs - the values supplied when the callback was initially created.

Examples

>>> if (info := self.info_timer(handle)) is not None:
>>>     time, interval, kwargs = info

appdaemon.adapi.ADAPI.reset_timer(self, handle: str) → bool

Reset a previously created timer.

The timer must be actively running, and not a sun-related one like sunrise/sunset for it to be reset.

Parameters:: handle (str) – The handle returned from the original call to create the timer.
Returns:: Boolean representing whether the timer reset was successful.

Examples

>>> self.reset_timer(handle)
True

Service

appdaemon.adapi.ADAPI.register_service(self, service: str, cb: Callable, namespace: str | None = None, **kwargs) → None

Register a service that can be called from other apps, the REST API, and the event stream.

This makes a function available to be called in other apps using call_service(...). The service function can accept arbitrary keyword arguments.

Registering services in namespaces that already have plugins is not recommended, as it can lead to some unpredictable behavior. Intead, it’s recommended to use a user-defined namespace or one that is not tied to plugin.

Parameters:

service – Name of the service, in the format domain/service. If the domain does not exist it will be created.
cb – The function to use for the service. This will accept both sync and async functions. Async functions are not recommended, as AppDaemon’s threading model makes them unnecessary. Async functions run in the event loop along with AppDaemon internal functions, so any blocking or delays, can cause AppDaemon itself to hang.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. See the namespace documentation for more information.
**kwargs (optional) – Zero or more keyword arguments. Extra keyword arguments will be stored alongside the service definition.

Returns:

None

Examples

>>> self.register_service("myservices/service1", self.mycallback)

>>> async def mycallback(self, namespace: str, domain: str, service: str, kwargs):
>>>     self.log("Service called")

appdaemon.adapi.ADAPI.deregister_service(self, service: str, namespace: str | None = None) → bool

Deregister a service that had been previously registered.

This will immediately remove the service from AppDaemon’s internal service registry, which will make it unavailable to other apps using the call_service() API call, as well as published as a service in the REST API

Using this function, an App can deregister a service call, it has initially registered in the service registry. This will automatically make it unavailable to other apps using the call_service() API call, as well as published as a service in the REST API and make it unavailable to the call_service command in the event stream. This function can only be used, within the app that registered it in the first place

Parameters:

service – Name of the service, in the format domain/service.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. See the namespace documentation for more information.

Returns:

True if the service was successfully deregistered, False otherwise.

Examples

>>> self.deregister_service("myservices/service1")

appdaemon.adapi.ADAPI.list_services(self, namespace: str = 'global') → list[dict[str, str]]

List all services available within AppDaemon

Parameters:: namespace (str, optional) – Optional namespace to use. The default is flobal, which will return services across all namespaces. See the namespace documentation for more information.
Returns:: List of dicts with keys namespace, domain, and service.

Examples

>>> services = self.list_services()

>>> services = self.list_services("default")

>>> services = self.list_services("mqtt")

Calls a Service within AppDaemon.

Services represent specific actions, and are generally registered by plugins or provided by AppDaemon itself. The app calls the service only by referencing the service with a string in the format <domain>/<service>, so there is no direct coupling between apps and services. This allows any app to call any service, even ones from other plugins.

Services often require additional parameters, such as entity_id, which AppDaemon will pass to the service call as appropriate, if used when calling this function. This allows arbitrary data to be passed to the service calls.

Apps can also register their own services using their self.regsiter_service method.

Parameters:

service (str) – The service name in the format <domain>/<service>. For example, light/turn_on.
namespace (str, optional) – It’s safe to ignore this parameter in most cases because the default namespace will be used. However, if a namespace is provided, the service call will be made in that namespace. If there’s a plugin associated with that namespace, it will do the service call. If no namespace is given, AppDaemon will use the app’s namespace, which can be set using the self.set_namespace method. See the section on namespaces for more information.
timeout (str | int | float, optional) – The internal AppDaemon timeout for the service call. If no value is specified, the default timeout is 60s. The default value can be changed using the appdaemon.internal_function_timeout config setting.
callback (callable) – The non-async callback to be executed when complete. It should accept a single argument, which will be the result of the service call. This is the recommended method for calling services which might take a long time to complete. This effectively bypasses the timeout argument because it only applies to this function, which will return immediately instead of waiting for the result if a callback is specified.
service_data (dict, optional) – Used as an additional dictionary to pass arguments into the service_data field of the JSON that goes to Home Assistant. This is useful if you have a dictionary that you want to pass in that has a key like target which is otherwise used for the target argument.
**data – Any other keyword arguments get passed to the service call as service_data. Each service takes different parameters, so this will vary from service to service. For example, most services require entity_id. The parameters for each service can be found in the actions tab of developer tools in the Home Assistant web interface.

Returns:

Result of the call_service function if any, see service call notes for more details.

Examples

HASS

>>> self.call_service("light/turn_on", entity_id="light.office_lamp", color_name="red")
>>> self.call_service("notify/notify", title="Hello", message="Hello World")
>>> events = self.call_service(
        "calendar/get_events",
        entity_id="calendar.home",
        start_date_time="2024-08-25 00:00:00",
        end_date_time="2024-08-27 00:00:00",
    )["result"]["response"]["calendar.home"]["events"]

MQTT

>>> self.call_service("mqtt/subscribe", topic="homeassistant/living_room/light", qos=2)
>>> self.call_service("mqtt/publish", topic="homeassistant/living_room/light", payload="on")

Utility

It’s important that the namespace arg is set to admin for these services, as they do not exist within the default namespace, and apps cannot exist in the admin namespace. If the namespace is not specified, calling the method will raise an exception.

>>> self.call_service("app/restart", app="notify_app", namespace="admin")
>>> self.call_service("app/stop", app="lights_app", namespace="admin")
>>> self.call_service("app/reload", namespace="admin")

Sequence

appdaemon.adapi.ADAPI.run_sequence(self, sequence: str | list[dict[str, dict[str, str]]], namespace: str | None = None) → Any

Run an AppDaemon Sequence.

Sequences are defined in a valid apps.yaml file or inline, and are sequences of service calls.

Parameters:

sequence – The sequence name, referring to the correct entry in apps.yaml, or a list containing actual commands to run
namespace (str, optional) – If a namespace is provided, AppDaemon will change the state of the given entity in the given namespace. On the other hand, if no namespace is given, AppDaemon will use the last specified namespace or the default namespace. See the section on namespaces for a detailed description. In most cases, it is safe to ignore this parameter.

Returns:

A handle that can be used with cancel_sequence() to terminate the script.

Examples

Run a yaml-defined sequence called “sequence.front_room_scene”.

>>> handle = self.run_sequence("sequence.front_room_scene")

>>> handle = self.run_sequence("front_room_scene")

Run an inline sequence.

>>> handle = self.run_sequence([
        {"light/turn_on": {"entity_id": "light.office_1"}},
        {"sleep": 5},
        {"light.turn_off": {"entity_id": "light.office_1"}}
    ])

appdaemon.adapi.ADAPI.cancel_sequence(self, sequence: str | list[str] | Future) → None

Cancel an already running AppDaemon Sequence.

Parameters:: sequence – The sequence as configured to be cancelled, or the sequence entity_id or future object
Returns:: None.

Examples

>>> self.cancel_sequence("sequence.living_room_lights")

Events

Register a callback for a specific event, multiple events, or any event.

The callback needs to have the following form:

>>> def my_callback(self, event_name: str, event_data: dict[str, Any], **kwargs: Any) -> None: ...

Parameters:

callback – Function that will be called when the event is fired. It must conform to the standard event callback format documented here
event (str | list[str], optional) – Name of the event to subscribe to. Can be a standard Home Assistant event such as service_registered, an arbitrary custom event such as MODE_CHANGE or a list of events [“pressed”, “released”]. If no event is specified, listen_event() will subscribe to all events.
namespace (str, optional) – Optional namespace to use. Defaults to using the app’s current namespace. The value global will register the callback for all namespaces. See the namespace documentation for more information.
timeout (str, int, float, timedelta, optional) – If supplied, the callback will be created as normal, but the callback will be removed after the timeout.
oneshot (bool, optional) – If True, the callback will be automatically cancelled after the first state change that results in a callback. Defaults to False.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs (optional) –
One or more keyword value pairs representing app-specific parameters to supply to the callback. If the event has data that matches one of these keywords, it will be filtered by the value passed in with this function. This means that if the value in the event data does not match, the callback will not be called. If the values provided are callable (lambda, function, etc), then they’ll be invoked with the events content, and if they return True, they’ll be considered to match.

Filtering will work with any event type, but it will be necessary to figure out the data associated with the event to understand what values can be filtered on. This can be achieved by examining Home Assistant’s logfiles when the event fires.

Returns:

A handle that can be used to cancel the callback.

Examples

Listen all “MODE_CHANGE” events.

>>> self.listen_event(self.mode_event, "MODE_CHANGE")

Listen for a minimote event activating scene 3.

>>> self.listen_event(self.generic_event, "zwave.scene_activated", scene_id=3)

Listen for a minimote event activating scene 3 from a specific minimote .

>>> self.listen_event(self.generic_event, "zwave.scene_activated", entity_id="minimote_31", scene_id=3)

Listen for a minimote event activating scene 3 from certain minimote (starting with 3), matched with code.

>>> self.listen_event(
        self.generic_event,
        "zwave.scene_activated",
        entity_id=lambda x: x.starts_with("minimote_3"),
        scene_id=3
    )

Listen for some custom events of a button being pressed.

>>> self.listen_event(self.button_event, ["pressed", "released"])

appdaemon.adapi.ADAPI.cancel_listen_event(self, handle: str | Iterable[str], *, silent: bool = False) → bool | dict[str, bool]

Cancel a callback for a specific event.

Parameters:

handle (str, Iterable[str]) – Handle(s) returned from a previous call to listen_event().
silent (bool, optional) – If True, no warning will be issued if the handle is not found. Defaults to False. This is useful if you want to cancel a callback that may or may not exist.

Returns:

A single boolean if a single handle is passed, or a dict mapping the handles to boolean values. Each boolean value will be the result of canceling the corresponding handle.

Examples

Cancel a single callback. >>> self.cancel_listen_event(handle) True

Cancel multiple callbacks. >>> result = self.cancel_listen_event([handle1, handle2]) >>> all(result.values()) # Check if all handles were canceled successfully True

appdaemon.adapi.ADAPI.info_listen_event(self, handle: str) → bool

Gets information on an event callback from its handle.

Parameters:: handle – The handle returned when the listen_event() call was made.
Returns:: The values (service, kwargs) supplied when the callback was initially created.

Examples

>>> service, kwargs = self.info_listen_event(handle)

appdaemon.adapi.ADAPI.fire_event(self, event: str, namespace: str | None = None, **kwargs) → None

Fires an event on the AppDaemon bus, for apps and plugins.

Parameters:

event – Name of the event. Can be a standard Home Assistant event such as service_registered or an arbitrary custom event such as “MODE_CHANGE”.
namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases, it is safe to ignore this parameter.
**kwargs (optional) – Zero or more keyword arguments that will be supplied as part of the event.

Returns:

None.

Examples

>>> self.fire_event("MY_CUSTOM_EVENT", jam="true")

Logging

appdaemon.adapi.ADAPI.log(self, msg: str, *args, level: str | int = 'INFO', log: str | None = None, ascii_encode: bool | None = None, stack_info: bool = False, stacklevel: int = 1, extra: Mapping[str, object] | None = None, **kwargs) → None

Logs a message to AppDaemon’s main logfile.

Parameters:

msg (str) – The message to log.
level (str, optional) – String representing the standard logger levels. Defaults to INFO.
log (str, optional) – Send the message to a specific log, either system or user_defined. System logs are main_log, error_log, diag_log or access_log. Any other value in use here must have a corresponding user-defined entity in the logs section of appdaemon.yaml.
ascii_encode (bool, optional) – Switch to disable the encoding of all log messages to ascii. Set this to false if you want to log UTF-8 characters (Default is controlled by appdaemon.ascii_encode, and is True unless modified).
stack_info (bool, optional) – If True the stack info will included.
stacklevel (int, optional) – Defaults to 1.
extra (dict, optional) – Extra values to add to the log record

Returns:

None.

Examples

Log a message to the main logfile of the system.

>>> self.log("Log Test: Parameter is %s", some_variable)

Log a message to the specified logfile.

>>> self.log("Log Test: Parameter is %s", some_variable, log="test_log")

Log a message with error-level to the main logfile of the system.

>>> self.log("Log Test: Parameter is %s", some_variable, level = "ERROR")

Log a message using placeholders to the main logfile of the system.

>>> self.log("Line: __line__, module: __module__, function: __function__, Msg: Something bad happened")

Log a WARNING message (including the stack info) to the main logfile of the system.

>>> self.log("Stack is", some_value, level="WARNING", stack_info=True)

appdaemon.adapi.ADAPI.error(self, msg: str, *args, level: str | int = 'INFO', ascii_encode: bool = True, stack_info: bool = False, stacklevel: int = 1, extra: Mapping[str, object] | None = None, **kwargs) → None

Logs a message to AppDaemon’s error logfile.

Parameters:

msg (str) – The message to log.
*args – Positional arguments for populating the msg fields
level (str, optional) – String representing the standard logger levels. Defaults to INFO.
ascii_encode (bool, optional) – Switch to disable the encoding of all log messages to ascii. Set this to false if you want to log UTF-8 characters (Default: True).
stack_info (bool, optional) – If True the stack info will included.
**kwargs – Keyword arguments

Returns:

None.

Examples

Log an error message to the error logfile of the system.

>>> self.error("Some Warning string")

Log an error message with critical-level to the error logfile of the system.

>>> self.error("Some Critical string", level = "CRITICAL")

Register a callback for whenever an app logs a message.

Parameters:

callback – Function that will be called when a message is logged. It must conform to the standard event callback format documented here
level (str, optional) – Minimum level for logs to trigger the callback. Lower levels will be ignored. Default is INFO.
namespace (str, optional) – Namespace to use for the call. Defaults to admin for log callbacks. See the namespace documentation for more information.
log (str, optional) – Name of the log to listen to, default is all logs. The name should be one of the 4 built in types main_log, error_log, diag_log or access_log or a user defined log entry.
pin (bool, optional) – Optional setting to override the default thread pinning behavior. By default, this is effectively True, and pin_thread gets set when the app starts.
pin_thread (int, optional) – Specify which thread from the worker pool will run the callback. The threads each have an ID number. The ID numbers start at 0 and go through (number of threads - 1).
**kwargs (optional) – One or more keyword arguments to supply to the callback.

Returns:

A handle that can be used to cancel the callback.

Examples

Listen to all WARNING log messages of the system.

>>> self.handle = self.listen_log(self.cb, "WARNING")

Sample callback:

>>> def log_message(self, name, ts, level, type, message, kwargs):

Listen to all WARNING log messages of the main_log.

>>> self.handle = self.listen_log(self.cb, "WARNING", log="main_log")

Listen to all WARNING log messages of a user-defined logfile.

>>> self.handle = self.listen_log(self.cb, "WARNING", log="my_custom_log")

appdaemon.adapi.ADAPI.cancel_listen_log(self, handle: str) → None

Cancels the log callback for the App.

Parameters:: handle – The handle returned when the listen_log call was made.
Returns:: Boolean.

Examples

>>> self.cancel_listen_log(handle)

appdaemon.adapi.ADAPI.get_main_log(self) → Logger

Returns the underlying logger object used for the main log.

Examples

Log a critical message to the main logfile of the system.

>>> log = self.get_main_log()
>>> log.critical("Log a critical error")

appdaemon.adapi.ADAPI.get_error_log(self) → Logger

Returns the underlying logger object used for the error log.

Examples

Log an error message to the error logfile of the system.

>>> error_log = self.get_error_log()
>>> error_log.error("Log an error", stack_info=True, exc_info=True)

appdaemon.adapi.ADAPI.get_user_log(self, log: str) → Logger

Gets the specified-user logger of the App.

Parameters:: log (str) – The name of the log you want to get the underlying logger object from, as described in the logs section of appdaemon.yaml.
Returns:: The underlying logger object used for the error log.

Examples

Log an error message to a user-defined logfile.

>>> log = self.get_user_log("test_log")
>>> log.error("Log an error", stack_info=True, exc_info=True)

appdaemon.adapi.ADAPI.set_log_level(self, level: str | int) → None

Sets the log level for this App, which applies to the main log, error log, and all user logs.

Parameters:: level (str) – Log level.
Returns:: None.

Note

Supported log levels: INFO, WARNING, ERROR, CRITICAL, DEBUG, NOTSET.

Examples

>>> self.set_log_level("DEBUG")

appdaemon.adapi.ADAPI.set_error_level(self, level: str | int) → None

Sets the log level to send to the error logfile of the system.

Parameters:: level (str) – Error level.
Returns:: None.

Note

Supported log levels: INFO, WARNING, ERROR, CRITICAL, DEBUG, NOTSET.

Dashboard

appdaemon.adapi.ADAPI.dash_navigate(self, target: str, timeout: int = -1, ret: str | None = None, sticky: int = 0, deviceid: str | None = None, dashid: str | None = None) → None

Forces all connected Dashboards to navigate to a new URL.

Parameters:

target (str) – Name of the new Dashboard to navigate to (e.g., /SensorPanel). Note that this value is not a URL.
timeout (int) – Length of time to stay on the new dashboard before returning to the original. This argument is optional and if not specified, the navigation will be permanent. Note that if there is a click or touch on the new panel before the timeout expires, the timeout will be cancelled.
ret (str) – Dashboard to return to after the timeout has elapsed.
sticky (int) – Specifies whether or not to return to the original dashboard after it has been clicked on. The default behavior (sticky=0) is to remain on the new dashboard if clicked, or return to the original otherwise. By using a different value (sticky= 5), clicking the dashboard will extend the amount of time (in seconds), but it will return to the original dashboard after a period of inactivity equal to timeout.
deviceid (str) – If set, only the device which has the same deviceid will navigate.
dashid (str) – If set, all devices currently on a dashboard which the title contains the substring dashid will navigate. ex: if dashid is “kichen”, it will match devices which are on “kitchen lights”, “kitchen sensors”, “ipad - kitchen”, etc.

Returns:

None.

Examples

Switch to AlarmStatus Panel then return to current panel after 10 seconds.

>>> self.dash_navigate("/AlarmStatus", timeout=10)

Switch to Locks Panel then return to Main panel after 10 seconds.

>>> self.dash_navigate("/Locks", timeout=10, ret="/SensorPanel")

Namespace

appdaemon.adapi.ADAPI.set_namespace(self, namespace: str, writeback: str = 'safe', persist: bool = True) → None

Set the current namespace of the app

See the namespace documentation for more information.

Parameters:

namespace (str) – Name of the new namespace
writeback (str, optional) – The writeback to be used if a new namespace gets created. Will be safe by default.
persist (bool, optional) – Whether to make the namespace persistent if a new one is created. Defaults to True.

Returns:

None.

Examples

>>> self.set_namespace("hass1")

appdaemon.adapi.ADAPI.get_namespace(self) → str

Get the app’s current namespace.

See the namespace documentation for more information.

appdaemon.adapi.ADAPI.list_namespaces(self) → list[str]

Get a list of all the namespaces in AppDaemon.

Examples

>>> self.list_namespaces()

appdaemon.adapi.ADAPI.save_namespace(self, namespace: str | None = None) → None

Saves entities created in user-defined namespaces into a file.

This way, when AD restarts these entities will be reloaded into AD with its previous states within the namespace. This can be used as a basic form of non-volatile storage of entity data. Depending on the configuration of the namespace, this function can be setup to constantly be running automatically or only when AD shutdown. This function also allows for users to manually execute the command as when needed.

Parameters:: namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases it is safe to ignore this parameter.
Returns:: None.

Examples

Save all entities of the default namespace.

>>> self.save_namespace()

Threading

appdaemon.adapi.ADAPI.set_app_pin(self, pin: bool) → None

Sets an App to be pinned or unpinned.

Parameters:: pin (bool) – Sets whether the App becomes pinned or not.
Returns:: None.

Examples

The following line should be put inside the initialize() function.

>>> self.set_app_pin(True)

appdaemon.adapi.ADAPI.get_app_pin(self) → bool

Finds out if the current App is currently pinned or not.

Returns:: True if the App is pinned, False otherwise.
Return type:: bool

Examples

>>> if self.get_app_pin(True):
>>>     self.log("App pinned!")

appdaemon.adapi.ADAPI.set_pin_thread(self, thread: int) → None

Sets the thread that the App will be pinned to.

Parameters:: thread (int) – Number of the thread to pin to. Threads start at 0 and go up to the number of threads specified in appdaemon.yaml -1.
Returns:: None.

Examples

The following line should be put inside the initialize() function.

>>> self.set_pin_thread(5)

appdaemon.adapi.ADAPI.get_pin_thread(self) → int

Finds out which thread the App is pinned to.

Returns:: The thread number or -1 if the App is not pinned.
Return type:: int

Examples

>>> thread = self.get_pin_thread():
>>> self.log(f"I'm pinned to thread: {thread}")

Async

appdaemon.adapi.ADAPI.create_task(self, coro: Coroutine[Any, Any, T], callback: Callable | None = None, name: str | None = None, **kwargs) → Task[T]

Wrap the coro coroutine into a Task and schedule its execution. Return the Task object.

Uses AppDaemon’s internal event loop to run the task, so the task will be run in the same thread as the app. Running an async method like this is useful for long-running tasks because it bypasses the timeout that AppDaemon otherwise imposes on callbacks.

The callback will be run in the app’s thread, like other AppDaemon callbacks, and will have the normal timeout imposed on it.

See creating tasks for in the python documentation for more information.

Parameters:

coro – The coroutine object (not coroutine function) to be executed.
callback – The non-async callback to be executed when complete.
**kwargs (optional) – Any additional keyword arguments to send the callback.

Returns:

A Task object, which can be cancelled by calling f.cancel().

Examples

Define your callback

>>> def my_callback(self, **kwargs: Any) -> Any: ...

Create the task

>>> task = self.create_task(asyncio.sleep(3), callback=self.my_callback)

Keyword Arguments

Define your callback with a custom keyword argument my_kwarg

>>> def my_callback(self, result: Any, my_kwarg: str, **kwargs: Any) -> Any:
        self.log(f"Result: {result}, my_kwarg: {my_kwarg}")

Use the custom keyword argument when creating the task

>>> task = self.create_task(asyncio.sleep(3), callback=self.my_callback, my_kwarg="special value")

async appdaemon.adapi.ADAPI.run_in_executor(self, func: Callable[[...], T], *args, **kwargs) → T

Run a sync function from within an async function using a thread from AppDaemon’s internal thread pool.

This essentially converts a sync function into an async function, which allows async functions to use it. This is useful for even short-ish functions (even <1s execution time) because it allows the event loop to continue processing other events while waiting for the function to complete. Blocking the event loop prevents AppDaemon’s internals from running, which interferes with all other apps, and can cause issues with connection timeouts.

Parameters:

func – The function to be executed.
*args (optional) – Any additional arguments to be used by the function
**kwargs (optional) – Any additional keyword arguments to be used by the function

Returns:

None

Examples

>>> await self.run_in_executor(self.run_request)

async appdaemon.adapi.ADAPI.sleep(delay: float, result=None) → None

Pause execution for a certain time span (not available in sync apps)

Parameters:

delay (float) – Number of seconds to pause.
result (optional) – Result to return upon delay completion.

Returns:

Result or None.

Note

This function is not available in sync apps.

Examples

>>> async def myfunction(self):
>>>     await self.sleep(5)

Utility

appdaemon.adapi.ADAPI.get_app(self, name: str) → ADAPI

Gets the instantiated object of another app running within the system.

This is useful for calling functions or accessing variables that reside in different apps without requiring duplication of code.

Parameters:: name (str) – Name of the app required. This is the name specified in header section of the config file, not the module or class.
Returns:: An object reference to the class.

Examples

>>> MyApp = self.get_app("MotionLights")
>>> MyApp.turn_light_on()

appdaemon.adapi.ADAPI.get_ad_version() → str

Returns a string with the current version of AppDaemon.

Examples

>>> version = self.get_ad_version()

appdaemon.adapi.ADAPI.entity_exists(self, entity_id: str, namespace: str | None = None) → bool

Checks the existence of an entity in AD.

When working with multiple AD namespaces, it is possible to specify the namespace, so that it checks within the right namespace in in the event the app is working in a different namespace. Also when using this function, it is also possible to check if an AppDaemon entity exists.

Parameters:

entity_id (str) – The fully qualified entity id (including the device type).
namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases it is safe to ignore this parameter.

Returns:

True if the entity id exists, False otherwise.

Return type:

bool

Examples

Check if the entity light.living_room exist within the app’s namespace

>>> if self.entity_exists("light.living_room"):
>>>     #do something

Check if the entity mqtt.security_settings exist within the mqtt namespace if the app is operating in a different namespace like default

>>> if self.entity_exists("mqtt.security_settings", namespace = "mqtt"):
>>>    #do something

appdaemon.adapi.ADAPI.split_entity(self, entity_id: str, namespace: str | None = None) → list

Splits an entity into parts.

This utility function will take a fully qualified entity id of the form light.hall_light and split it into 2 values, the device and the entity, e.g. light and hall_light.

Parameters:

entity_id (str) – The fully qualified entity id (including the device type).
namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases it is safe to ignore this parameter.

Returns:

A list with 2 entries, the device and entity respectively.

Examples

Do some action if the device of the entity is scene.

>>> device, entity = self.split_entity(entity_id)
>>> if device == "scene":
>>>     #do something specific to scenes

appdaemon.adapi.ADAPI.remove_entity(self, entity_id: str, namespace: str | None = None) → None

Deletes an entity created within a namespaces.

If an entity was created, and its deemed no longer needed, by using this function, the entity can be removed from AppDaemon permanently.

Parameters:

entity_id (str) – The fully qualified entity id (including the device type).
namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases it is safe to ignore this parameter.

Returns:

None.

Examples

Delete the entity in the present namespace.

>>> self.remove_entity('sensor.living_room')

Delete the entity in the mqtt namespace.

>>> self.remove_entity('mqtt.living_room_temperature', namespace = 'mqtt')

appdaemon.adapi.ADAPI.split_device_list(devices: str) → list[str]

Converts a comma-separated list of device types to an iterable list.

This is intended to assist in use cases where the App takes a list of entities from an argument, e.g., a list of sensors to monitor. If only one entry is provided, an iterable list will still be returned to avoid the need for special processing.

Parameters:: devices (str) – A comma-separated list of devices to be split (without spaces).
Returns:: A list of split devices with 1 or more entries.

Examples

>>> for sensor in self.split_device_list(self.args["sensors"]):
>>>    #do something for each sensor, e.g., make a state subscription

appdaemon.adapi.ADAPI.get_plugin_config(self, namespace: str | None = None) → Any

Gets any useful metadata that the plugin may have available.

For instance, for the HASS plugin, this will return Home Assistant configuration data such as latitude and longitude.

Parameters:: namespace (str) – Select the namespace of the plugin for which data is desired.
Returns:: A dictionary containing all the configuration information available from the Home Assistant /api/config endpoint.

Examples

>>> config = self.get_plugin_config()
>>> self.log(f'My current position is {config["latitude"]}(Lat), {config["longitude"]}(Long)')
My current position is 50.8333(Lat), 4.3333(Long)

appdaemon.adapi.ADAPI.friendly_name(self, entity_id: str, namespace: str | None = None) → str

Gets the Friendly Name of an entity.

Parameters:

entity_id (str) – The fully qualified entity id (including the device type).
namespace (str, optional) – Namespace to use for the call. See the section on namespaces for a detailed description. In most cases it is safe to ignore this parameter.

Returns:

The friendly name of the entity if it exists or the entity id if not.

Return type:

str

Examples

>>> tracker = "device_tracker.andrew"
>>> friendly_name = self.friendly_name(tracker)
>>> tracker_state = self.get_tracker_state(tracker)
>>> self.log(f"{tracker}  ({friendly_name}) is {tracker_state}.")
device_tracker.andrew (Andrew Tracker) is on.

appdaemon.adapi.ADAPI.set_production_mode(self, mode: bool = True) → bool | None

Deactivates or activates the production mode in AppDaemon.

When called without declaring passing any arguments, mode defaults to True.

Parameters:: mode (bool) – If it is True the production mode is activated, or deactivated otherwise.
Returns:: The specified mode or None if a wrong parameter is passed.

appdaemon.adapi.ADAPI.start_app(self, app: str) → None

Starts an App which can either be running or not.

This API call cannot start an app which has already been disabled in the App Config. It essentially only runs the initialize() function in the app, and changes to attributes like class name or app config are not taken into account.

Parameters:: app (str) – Name of the app.
Returns:: None.

Examples

>>> self.start_app("lights_app")

appdaemon.adapi.ADAPI.stop_app(self, app: str) → None

Stops an App which is running.

Parameters:: app (str) – Name of the app.
Returns:: None.

Examples

>>> self.stop_app("lights_app")

appdaemon.adapi.ADAPI.restart_app(self, app: str) → None

Restarts an App which can either be running or not.

Parameters:: app (str) – Name of the app.
Returns:: None.

Examples

>>> self.restart_app("lights_app")

appdaemon.adapi.ADAPI.reload_apps(self) → None

Reloads the apps, and loads up those that have changes made to their .yaml or .py files.

This utility function can be used if AppDaemon is running in production mode, and it is needed to reload apps that changes have been made to.

Returns:: None.

Examples

>>> self.reload_apps()

Dialogflow

appdaemon.adapi.ADAPI.get_dialogflow_intent(self, data: dict) → Any | None

Gets the intent’s action from the Google Home response.

Parameters:: data – Response received from Google Home.
Returns:: A string representing the Intent from the interaction model that was requested, or None, if no action was received.

Examples

>>> intent = ADAPI.get_dialogflow_intent(data)

appdaemon.adapi.ADAPI.get_dialogflow_slot_value(data, slot=None) → Any | None

Gets slots’ values from the interaction model.

Parameters:

data – Response received from Google Home.
slot (str) – Name of the slot. If a name is not specified, all slots will be returned as a dictionary. If a name is specified but is not found, None will be returned.

Returns:

A string representing the value of the slot from the interaction model, or a hash of slots.

Examples

>>> beer_type = ADAPI.get_dialogflow_intent(data, "beer_type")
>>> all_slots = ADAPI.get_dialogflow_intent(data)

appdaemon.adapi.ADAPI.format_dialogflow_response(self, speech=None) → Any | None

Formats a response to be returned to Google Home, including speech.

Parameters:: speech (str) – The text for Google Home to say.
Returns:: None.

Examples

>>> ADAPI.format_dialogflow_response(speech = "Hello World")

Alexa

appdaemon.adapi.ADAPI.get_alexa_intent(data: dict) → str | None

Gets the Intent’s name from the Alexa response.

Parameters:: data – Response received from Alexa.
Returns:: A string representing the Intent’s name from the interaction model that was requested, or None, if no Intent was received.

Examples

>>> intent = ADAPI.get_alexa_intent(data)

appdaemon.adapi.ADAPI.get_alexa_slot_value(data, slot=None) → str | None

Gets values for slots from the interaction model.

Parameters:

data – The request data received from Alexa.
slot – Name of the slot. If a name is not specified, all slots will be returned as a dictionary. If a name is specified but is not found, None will be returned.

Returns:

A string representing the value of the slot from the interaction model, or a hash of slots.

Examples

>>> beer_type = ADAPI.get_alexa_intent(data, "beer_type")
>>> all_slots = ADAPI.get_alexa_intent(data)

appdaemon.adapi.ADAPI.format_alexa_response(speech: str | None = None, card: str | None = None, title: str | None = None) → dict

Formats a response to be returned to Alexa including speech and a card.

Parameters:

speech (str) – The text for Alexa to say.
card (str) – Text for the card.
title (str) – Title for the card.

Returns:

None.

Examples

>>> ADAPI.format_alexa_response(speech = "Hello World", card = "Greetings to the world", title = "Hello")

appdaemon.adapi.ADAPI.get_alexa_error(data: dict) → str | None

Gets the error message from the Alexa API response.

Parameters:: data – Response received from the Alexa API.
Returns:: A string representing the value of message, or None if no error message was received.

API

appdaemon.adapi.ADAPI.register_endpoint(self, callback: Callable[[Any, dict], Any], endpoint: str | None = None, **kwargs) → str | None

Registers an endpoint for API calls into the current App.

Parameters:

callback – The function to be called when a request is made to the named endpoint.
endpoint (str, optional) – The name of the endpoint to be used for the call (Default: None).
endpoints (This must be unique across all)
given (and when not)
endpoint. (the name of the app is used as the)
instance. (It is possible to register multiple endpoints to a single app)

Keyword Arguments:

**kwargs (optional) – Zero or more keyword arguments.

Returns:

A handle that can be used to remove the registration.

Examples

It should be noted that the register function, should return a string (can be empty), and an HTTP OK status response (e.g., 200. If this is not added as a returned response, the function will generate an error each time it is processed. If the POST request contains JSON data, the decoded data will be passed as the argument to the callback. Otherwise the callback argument will contain the query string. A request kwarg contains the http request object.

>>> self.register_endpoint(self.my_callback)
>>> self.register_endpoint(self.alexa_cb, "alexa")

>>> async def alexa_cb(self, json_obj, kwargs):
>>>     self.log(json_obj)
>>>     response = {"message": "Hello World"}
>>>     return response, 200

appdaemon.adapi.ADAPI.deregister_endpoint(self, handle: str) → None

Removes a previously registered endpoint.

Parameters:: handle – A handle returned by a previous call to register_endpoint
Returns:: None.

Examples

>>> self.deregister_endpoint(handle)

WebRoute ~~~

appdaemon.adapi.ADAPI.register_route(self, callback: Callable[[Any, dict], Any], route: str | None = None, **kwargs: dict[str, Any]) → str | None

Registers a route for Web requests into the current App.: By registering an app web route, this allows to make use of AD’s internal web server to serve web clients. All routes registered using this api call, can be accessed using http://AD_IP:Port/app/route.

Parameters:

callback – The function to be called when a request is made to the named route. This must be an async function
route (str, optional) – The name of the route to be used for the request (Default: the app’s name).

Keyword Arguments:

**kwargs (optional) – Zero or more keyword arguments.

Returns:

A handle that can be used to remove the registration.

Examples

It should be noted that the register function, should return a aiohttp Response.

>>> from aiohttp import web

>>> def initialize(self):
>>>   self.register_route(my_callback)
>>>   self.register_route(stream_cb, "camera")
>>>
>>> async def camera(self, request, kwargs):
>>>   return web.Response(text="test", content_type="text/html")

appdaemon.adapi.ADAPI.deregister_route(self, handle: str) → None

Removes a previously registered app route.

Parameters:: handle – A handle returned by a previous call to register_app_route
Returns:: None.

Examples

>>> self.deregister_route(handle)

Other

appdaemon.adapi.ADAPI.run_in_thread(self, callback: Callable, thread: int, **kwargs) → None

Schedules a callback to be run in a different thread from the current one.

Parameters:

callback – Function to be run on the new thread.
thread (int) – Thread number (0 - number of threads).
**kwargs – Arbitrary keyword parameters to be provided to the callback function when it is invoked.

Returns:

None.

Examples

>>> self.run_in_thread(my_callback, 8)

appdaemon.adapi.ADAPI.submit_to_executor(self, func: Callable[[...], T], *args, callback: Callable | None = None, **kwargs) → Future[T]

Submit a sync function from within another sync function to be executed using a thread from AppDaemon’s internal thread pool.

This function does not wait for the result of the submitted function and immediately returns a Future object. This is useful for executing long-running functions without blocking the thread for other callbacks. The result can be retrieved later using the Future object, but it’s recommended to use a callback to handle the result instead.

Parameters:

func – The function to be executed.
*args (optional) – Any additional arguments to be used by the function
callback (optional) – A callback function to be executed when the function has completed.
**kwargs (optional) – Any additional keyword arguments to be used by the function.

Returns:

A Future object representing the result of the function.

Examples

Submit a long-running function to be executed in the background

>>> def initialize(self):
        self.long_future = self.submit_to_executor(self.long_request, url, callback=self.result_callback)

Long running function:

>>> def long_request(self, url: str):
        import requests
        res = requests.get(url)
        return res.json()

Callback to handle the result:

>>> def result_callback(self, result: dict, **kwargs):
        # Set the attributes of a sensor with the result
        self.set_state("sensor.url_result", state="ready", attributes=result, replace=True)

appdaemon.adapi.ADAPI.get_thread_info(self) → Any

Gets information on AppDaemon worker threads.

Returns:: A dictionary containing all the information for AppDaemon worker threads.

Examples

>>> thread_info = self.get_thread_info()

appdaemon.adapi.ADAPI.get_scheduler_entries(self)

Gets information on AppDaemon scheduler entries.

Returns:: A dictionary containing all the information for entries in the AppDaemon scheduler.

Examples

>>> schedule = self.get_scheduler_entries()

appdaemon.adapi.ADAPI.get_callback_entries(self) → list

Gets information on AppDaemon callback entries.

Returns:: A dictionary containing all the information for entries in the AppDaemon state, and event callback table.

Examples

>>> callbacks = self.get_callback_entries()

appdaemon.adapi.ADAPI.depends_on_module(self, *modules: list[str]) → None

Registers a global_modules dependency for an app.

Parameters:: *modules – Modules to register a dependency on.
Returns:: None.

Examples

>>> import some_module
>>> import another_module
>>> # later
>>> self.depends_on_module('some_module')