homeassistant.core

Core components of Home Assistant.

Home Assistant is a Home Automation framework for observing the state of entities and react to changes.

class homeassistant.core.Config[source]

Configuration settings for Home Assistant.

as_dict()[source]

Create a dictionary representation of this dict.

Async friendly.

distance(lat: float, lon: float) → float[source]

Calculate distance from Home Assistant.

Async friendly.

is_allowed_path(path: str) → bool[source]

Check if the path is valid for access from outside.

path(*path)[source]

Generate path to the file within the configuration directory.

Async friendly.

class homeassistant.core.Event(event_type, data=None, origin=<EventOrigin.local: 'LOCAL'>, time_fired=None)[source]

Representation of an event within the bus.

as_dict()[source]

Create a dict representation of this Event.

Async friendly.

class homeassistant.core.EventBus(hass: homeassistant.core.HomeAssistant) → None[source]

Allow the firing of and listening for events.

async_fire(event_type: str, event_data=None, origin=<EventOrigin.local: 'LOCAL'>, wait=False)[source]

Fire an event.

This method must be run in the event loop.

async_listen(event_type, listener)[source]

Listen for all events or events of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

This method must be run in the event loop.

async_listen_once(event_type, listener)[source]

Listen once for event of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

Returns registered listener that can be used with remove_listener.

This method must be run in the event loop.

async_listeners()[source]

Return dictionary with events and the number of listeners.

This method must be run in the event loop.

fire(event_type: str, event_data=None, origin=<EventOrigin.local: 'LOCAL'>)[source]

Fire an event.

listen(event_type, listener)[source]

Listen for all events or events of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

listen_once(event_type, listener)[source]

Listen once for event of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

Returns function to unsubscribe the listener.

listeners

Return dictionary with events and the number of listeners.

class homeassistant.core.HomeAssistant(loop=None)[source]

Root object of the Home Assistant home automation.

add_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Add job to the executor pool.

target: target to call. args: parameters for method to call.

async_add_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Add a job from within the eventloop.

This method must be run in the event loop.

target: target to call. args: parameters for method to call.

async_block_till_done()[source]

Block till all pending work is done.

async_run_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Run a job from within the event loop.

This method must be run in the event loop.

target: target to call. args: parameters for method to call.

async_start()[source]

Finalize startup from inside the event loop.

This method is a coroutine.

async_stop(exit_code=0) → None[source]

Stop Home Assistant and shuts down all threads.

This method is a coroutine.

async_stop_track_tasks()[source]

Stop track tasks so you can’t wait for all tasks to be done.

async_track_tasks()[source]

Track tasks so you can wait for all tasks to be done.

block_till_done() → None[source]

Block till all pending work is done.

is_running

Return if Home Assistant is running.

start() → None[source]

Start home assistant.

stop() → None[source]

Stop Home Assistant and shuts down all threads.

class homeassistant.core.State(entity_id, state, attributes=None, last_changed=None, last_updated=None)[source]

Object to represent a state within the state machine.

entity_id: the entity that is represented. state: the state of the entity attributes: extra information on entity and state last_changed: last time the state was changed, not the attributes. last_updated: last time this object was updated.

as_dict()[source]

Return a dict representation of the State.

Async friendly.

To be used for JSON serialization. Ensures: state == State.from_dict(state.as_dict())

domain

Domain of this state.

classmethod from_dict(json_dict)[source]

Initialize a state from a dict.

Async friendly.

Ensures: state == State.from_json_dict(state.to_json_dict())

name

Name of this state.

object_id

Object id of this state.

class homeassistant.core.StateMachine(bus, loop)[source]

Helper class that tracks the state of different entities.

all()[source]

Create a list of all states.

async_all()[source]

Create a list of all states.

This method must be run in the event loop.

async_entity_ids(domain_filter=None)[source]

List of entity ids that are being tracked.

This method must be run in the event loop.

async_remove(entity_id)[source]

Remove the state of an entity.

Returns boolean to indicate if an entity was removed.

This method must be run in the event loop.

async_set(entity_id, new_state, attributes=None, force_update=False)[source]

Set the state of an entity, add entity if it does not exist.

Attributes is an optional dict to specify attributes of this state.

If you just update the attributes and not the state, last changed will not be affected.

This method must be run in the event loop.

entity_ids(domain_filter=None)[source]

List of entity ids that are being tracked.

get(entity_id)[source]

Retrieve state of entity_id or None if not found.

Async friendly.

is_state(entity_id, state)[source]

Test if entity exists and is specified state.

Async friendly.

is_state_attr(entity_id, name, value)[source]

Test if entity exists and has a state attribute set to value.

Async friendly.

remove(entity_id)[source]

Remove the state of an entity.

Returns boolean to indicate if an entity was removed.

set(entity_id, new_state, attributes=None, force_update=False)[source]

Set the state of an entity, add entity if it does not exist.

Attributes is an optional dict to specify attributes of this state.

If you just update the attributes and not the state, last changed will not be affected.

class homeassistant.core.ServiceCall(domain, service, data=None, call_id=None)[source]

Representation of a call to a service.

class homeassistant.core.ServiceRegistry(hass)[source]

Offer the services over the eventbus.

async_call(domain, service, service_data=None, blocking=False)[source]

Call a service.

Specify blocking=True to wait till service is executed. Waits a maximum of SERVICE_CALL_LIMIT.

If blocking = True, will return boolean if service executed successfully within SERVICE_CALL_LIMIT.

This method will fire an event to call the service. This event will be picked up by this ServiceRegistry and any other ServiceRegistry that is listening on the EventBus.

Because the service is sent as an event you are not allowed to use the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.

This method is a coroutine.

async_register(domain, service, service_func, description=None, schema=None)[source]

Register a service.

Description is a dict containing key ‘description’ to describe the service and a key ‘fields’ to describe the fields.

Schema is called to coerce and validate the service data.

This method must be run in the event loop.

async_remove(domain, service)[source]

Remove a registered service from service handler.

This method must be run in the event loop.

async_services()[source]

Return dictionary with per domain a list of available services.

This method must be run in the event loop.

call(domain, service, service_data=None, blocking=False)[source]

Call a service.

Specify blocking=True to wait till service is executed. Waits a maximum of SERVICE_CALL_LIMIT.

If blocking = True, will return boolean if service executed successfully within SERVICE_CALL_LIMIT.

This method will fire an event to call the service. This event will be picked up by this ServiceRegistry and any other ServiceRegistry that is listening on the EventBus.

Because the service is sent as an event you are not allowed to use the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.

has_service(domain, service)[source]

Test if specified service exists.

Async friendly.

register(domain, service, service_func, description=None, schema=None)[source]

Register a service.

Description is a dict containing key ‘description’ to describe the service and a key ‘fields’ to describe the fields.

Schema is called to coerce and validate the service data.

remove(domain, service)[source]

Remove a registered service from service handler.

services

Return dictionary with per domain a list of available services.

Module contents

Core components of Home Assistant.

Home Assistant is a Home Automation framework for observing the state of entities and react to changes.

class homeassistant.core.Config[source]

Bases: object

Configuration settings for Home Assistant.

as_dict()[source]

Create a dictionary representation of this dict.

Async friendly.

distance(lat: float, lon: float) → float[source]

Calculate distance from Home Assistant.

Async friendly.

is_allowed_path(path: str) → bool[source]

Check if the path is valid for access from outside.

path(*path)[source]

Generate path to the file within the configuration directory.

Async friendly.

class homeassistant.core.CoreState[source]

Bases: enum.Enum

Represent the current state of Home Assistant.

not_running = 'NOT_RUNNING'
running = 'RUNNING'
starting = 'STARTING'
stopping = 'STOPPING'
class homeassistant.core.Event(event_type, data=None, origin=<EventOrigin.local: 'LOCAL'>, time_fired=None)[source]

Bases: object

Representation of an event within the bus.

as_dict()[source]

Create a dict representation of this Event.

Async friendly.

data
event_type
origin
time_fired
class homeassistant.core.EventBus(hass: homeassistant.core.HomeAssistant) → None[source]

Bases: object

Allow the firing of and listening for events.

async_fire(event_type: str, event_data=None, origin=<EventOrigin.local: 'LOCAL'>, wait=False)[source]

Fire an event.

This method must be run in the event loop.

async_listen(event_type, listener)[source]

Listen for all events or events of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

This method must be run in the event loop.

async_listen_once(event_type, listener)[source]

Listen once for event of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

Returns registered listener that can be used with remove_listener.

This method must be run in the event loop.

async_listeners()[source]

Return dictionary with events and the number of listeners.

This method must be run in the event loop.

fire(event_type: str, event_data=None, origin=<EventOrigin.local: 'LOCAL'>)[source]

Fire an event.

listen(event_type, listener)[source]

Listen for all events or events of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

listen_once(event_type, listener)[source]

Listen once for event of a specific type.

To listen to all events specify the constant MATCH_ALL as event_type.

Returns function to unsubscribe the listener.

listeners

Return dictionary with events and the number of listeners.

class homeassistant.core.EventOrigin[source]

Bases: enum.Enum

Represent the origin of an event.

local = 'LOCAL'
remote = 'REMOTE'
class homeassistant.core.HomeAssistant(loop=None)[source]

Bases: object

Root object of the Home Assistant home automation.

add_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Add job to the executor pool.

target: target to call. args: parameters for method to call.

async_add_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Add a job from within the eventloop.

This method must be run in the event loop.

target: target to call. args: parameters for method to call.

async_block_till_done()[source]

Block till all pending work is done.

async_run_job(target: typing.Callable[..., NoneType], *args: typing.Any) → None[source]

Run a job from within the event loop.

This method must be run in the event loop.

target: target to call. args: parameters for method to call.

async_start()[source]

Finalize startup from inside the event loop.

This method is a coroutine.

async_stop(exit_code=0) → None[source]

Stop Home Assistant and shuts down all threads.

This method is a coroutine.

async_stop_track_tasks()[source]

Stop track tasks so you can’t wait for all tasks to be done.

async_track_tasks()[source]

Track tasks so you can wait for all tasks to be done.

block_till_done() → None[source]

Block till all pending work is done.

is_running

Return if Home Assistant is running.

start() → None[source]

Start home assistant.

stop() → None[source]

Stop Home Assistant and shuts down all threads.

class homeassistant.core.Service(func, description, fields, schema)[source]

Bases: object

Representation of a callable service.

as_dict()[source]

Return dictionary representation of this service.

description
fields
func
is_callback
is_coroutinefunction
schema
class homeassistant.core.ServiceCall(domain, service, data=None, call_id=None)[source]

Bases: object

Representation of a call to a service.

call_id
data
domain
service
class homeassistant.core.ServiceRegistry(hass)[source]

Bases: object

Offer the services over the eventbus.

async_call(domain, service, service_data=None, blocking=False)[source]

Call a service.

Specify blocking=True to wait till service is executed. Waits a maximum of SERVICE_CALL_LIMIT.

If blocking = True, will return boolean if service executed successfully within SERVICE_CALL_LIMIT.

This method will fire an event to call the service. This event will be picked up by this ServiceRegistry and any other ServiceRegistry that is listening on the EventBus.

Because the service is sent as an event you are not allowed to use the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.

This method is a coroutine.

async_register(domain, service, service_func, description=None, schema=None)[source]

Register a service.

Description is a dict containing key ‘description’ to describe the service and a key ‘fields’ to describe the fields.

Schema is called to coerce and validate the service data.

This method must be run in the event loop.

async_remove(domain, service)[source]

Remove a registered service from service handler.

This method must be run in the event loop.

async_services()[source]

Return dictionary with per domain a list of available services.

This method must be run in the event loop.

call(domain, service, service_data=None, blocking=False)[source]

Call a service.

Specify blocking=True to wait till service is executed. Waits a maximum of SERVICE_CALL_LIMIT.

If blocking = True, will return boolean if service executed successfully within SERVICE_CALL_LIMIT.

This method will fire an event to call the service. This event will be picked up by this ServiceRegistry and any other ServiceRegistry that is listening on the EventBus.

Because the service is sent as an event you are not allowed to use the keys ATTR_DOMAIN and ATTR_SERVICE in your service_data.

has_service(domain, service)[source]

Test if specified service exists.

Async friendly.

register(domain, service, service_func, description=None, schema=None)[source]

Register a service.

Description is a dict containing key ‘description’ to describe the service and a key ‘fields’ to describe the fields.

Schema is called to coerce and validate the service data.

remove(domain, service)[source]

Remove a registered service from service handler.

services

Return dictionary with per domain a list of available services.

class homeassistant.core.State(entity_id, state, attributes=None, last_changed=None, last_updated=None)[source]

Bases: object

Object to represent a state within the state machine.

entity_id: the entity that is represented. state: the state of the entity attributes: extra information on entity and state last_changed: last time the state was changed, not the attributes. last_updated: last time this object was updated.

as_dict()[source]

Return a dict representation of the State.

Async friendly.

To be used for JSON serialization. Ensures: state == State.from_dict(state.as_dict())

attributes
domain

Domain of this state.

entity_id
classmethod from_dict(json_dict)[source]

Initialize a state from a dict.

Async friendly.

Ensures: state == State.from_json_dict(state.to_json_dict())

last_changed
last_updated
name

Name of this state.

object_id

Object id of this state.

state
class homeassistant.core.StateMachine(bus, loop)[source]

Bases: object

Helper class that tracks the state of different entities.

all()[source]

Create a list of all states.

async_all()[source]

Create a list of all states.

This method must be run in the event loop.

async_entity_ids(domain_filter=None)[source]

List of entity ids that are being tracked.

This method must be run in the event loop.

async_remove(entity_id)[source]

Remove the state of an entity.

Returns boolean to indicate if an entity was removed.

This method must be run in the event loop.

async_set(entity_id, new_state, attributes=None, force_update=False)[source]

Set the state of an entity, add entity if it does not exist.

Attributes is an optional dict to specify attributes of this state.

If you just update the attributes and not the state, last changed will not be affected.

This method must be run in the event loop.

entity_ids(domain_filter=None)[source]

List of entity ids that are being tracked.

get(entity_id)[source]

Retrieve state of entity_id or None if not found.

Async friendly.

is_state(entity_id, state)[source]

Test if entity exists and is specified state.

Async friendly.

is_state_attr(entity_id, name, value)[source]

Test if entity exists and has a state attribute set to value.

Async friendly.

remove(entity_id)[source]

Remove the state of an entity.

Returns boolean to indicate if an entity was removed.

set(entity_id, new_state, attributes=None, force_update=False)[source]

Set the state of an entity, add entity if it does not exist.

Attributes is an optional dict to specify attributes of this state.

If you just update the attributes and not the state, last changed will not be affected.

homeassistant.core.async_loop_exception_handler(loop, context)[source]

Handle all exception inside the core loop.

homeassistant.core.callback(func: typing.Callable[..., NoneType]) → typing.Callable[..., NoneType][source]

Annotation to mark method as safe to call from within the event loop.

homeassistant.core.is_callback(func: typing.Callable[..., typing.Any]) → bool[source]

Check if function is safe to be called in the event loop.

homeassistant.core.split_entity_id(entity_id: str) → typing.List[str][source]

Split a state entity_id into domain, object_id.

homeassistant.core.valid_entity_id(entity_id: str) → bool[source]

Test if an entity ID is a valid format.

homeassistant.core.valid_state(state: str) → bool[source]

Test if an state is valid.