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(hass: homeassistant.core.HomeAssistant)[source]

Configuration settings for Home Assistant.

as_dict() → dict[source]

Create a dictionary representation of the configuration.

Async friendly.

async async_load() → None[source]

Load [homeassistant] core config.

async async_store() → None[source]

Store [homeassistant] core config.

async async_update(**kwargs: Any) → None[source]

Update the configuration from a dictionary.

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

Calculate distance from Home Assistant.

Async friendly.

is_allowed_external_url(url: str) → bool[source]

Check if an external URL is allowed.

is_allowed_path(path: str) → bool[source]

Check if the path is valid for access from outside.

path(*path: str) → str[source]

Generate path to the file within the configuration directory.

Async friendly.

set_time_zone(time_zone_str: str) → None[source]

Help to set the time zone.

class homeassistant.core.Context(user_id: str = None, parent_id: str | None = None, id: str = NOTHING)[source]

The context that triggered something.

as_dict() → dict[str, str | None][source]

Return a dictionary representation of the context.

class homeassistant.core.CoreState[source]

Represent the current state of Home Assistant.

class homeassistant.core.Event(event_type: str, data: dict[str, Any] | None = None, origin: EventOrigin = <EventOrigin.local: 'LOCAL'>, time_fired: datetime.datetime | None = None, context: Context | None = None)[source]

Representation of an event within the bus.

as_dict() → dict[str, Any][source]

Create a dict representation of this Event.

Async friendly.

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

Allow the firing of and listening for events.

async_fire(event_type: str, event_data: dict[str, Any] | None = None, origin: EventOrigin = <EventOrigin.local: 'LOCAL'>, context: Context | None = None, time_fired: datetime.datetime | None = None) → None[source]

Fire an event.

This method must be run in the event loop.

async_listen(event_type: str, listener: Callable, event_filter: Callable | None = None) → CALLBACK_TYPE[source]

Listen for all events or events of a specific type.

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

An optional event_filter, which must be a callable decorated with @callback that returns a boolean value, determines if the listener callable should run.

This method must be run in the event loop.

async_listen_once(event_type: str, listener: Callable) → Callable[[], None][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() → dict[str, int][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: dict | None = None, origin: EventOrigin = <EventOrigin.local: 'LOCAL'>, context: Context | None = None) → None[source]

Fire an event.

listen(event_type: str, listener: Callable) → Callable[[], None][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: str, listener: Callable[[homeassistant.core.Event], None]) → Callable[[], None][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.

property listeners

Return dictionary with events and the number of listeners.

class homeassistant.core.EventOrigin[source]

Represent the origin of an event.

class homeassistant.core.HassJob(target: Callable)[source]

Represent a job to be run later.

We check the callable type in advance so we can avoid checking it every time we run the job.

class homeassistant.core.HassJobType[source]

Represent a job type.

class homeassistant.core.HomeAssistant[source]

Root object of the Home Assistant home automation.

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

Add job to the executor pool.

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

async_add_executor_job(target: Callable[..., T], *args: Any) → Awaitable[T][source]

Add an executor job from within the event loop.

async_add_hass_job(hassjob: HassJob, *args: Any) → asyncio.Future | None[source]

Add a HassJob from within the event loop.

This method must be run in the event loop. hassjob: HassJob to call. args: parameters for method to call.

async_add_job(target: Callable[..., Any], *args: Any) → asyncio.Future | None[source]

Add 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 async_block_till_done() → None[source]

Block until all pending work is done.

async_create_task(target: collections.abc.Awaitable) → _asyncio.Task[source]

Create a task from within the eventloop.

This method must be run in the event loop.

target: target to call.

async async_run(*, attach_signals: bool = True) → int[source]

Home Assistant main entry point.

Start Home Assistant and block until stopped.

This method is a coroutine.

async_run_hass_job(hassjob: HassJob, *args: Any) → asyncio.Future | None[source]

Run a HassJob from within the event loop.

This method must be run in the event loop.

hassjob: HassJob args: parameters for method to call.

async_run_job(target: Callable[..., None | Awaitable], *args: Any) → asyncio.Future | 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 async_start() → None[source]

Finalize startup from inside the event loop.

This method is a coroutine.

async async_stop(exit_code: int = 0, *, force: bool = False) → None[source]

Stop Home Assistant and shuts down all threads.

The “force” flag commands async_stop to proceed regardless of Home Assistan’t current state. You should not set this flag unless you’re testing.

This method is a coroutine.

async_stop_track_tasks() → None[source]

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

async_track_tasks() → None[source]

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

block_till_done() → None[source]

Block until all pending work is done.

create_task(target: collections.abc.Awaitable) → None[source]

Add task to the executor pool.

target: target to call.

property is_running

Return if Home Assistant is running.

property is_stopping

Return if Home Assistant is stopping.

start() → int[source]

Start Home Assistant.

Note: This function is only used for testing. For regular use, use “await hass.run()”.

stop() → None[source]

Stop Home Assistant and shuts down all threads.

class homeassistant.core.Service(func: Callable, schema: vol.Schema | None, context: Context | None = None)[source]

Representation of a callable service.

class homeassistant.core.ServiceCall(domain: str, service: str, data: dict | None = None, context: Context | None = None)[source]

Representation of a call to a service.

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

Offer the services over the eventbus.

async async_call(domain: str, service: str, service_data: dict | None = None, blocking: bool = False, context: Context | None = None, limit: float | None = 10, target: dict | None = None) → bool | None[source]

Call a service.

Specify blocking=True to wait until service is executed. Waits a maximum of limit, which may be None for no timeout.

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

This method will fire an event to indicate the service has been called.

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: str, service: str, service_func: Callable, schema: vol.Schema | None = None) → None[source]

Register a service.

Schema is called to coerce and validate the service data.

This method must be run in the event loop.

async_remove(domain: str, service: str) → None[source]

Remove a registered service from service handler.

This method must be run in the event loop.

async_services() → dict[str, dict[str, Service]][source]

Return dictionary with per domain a list of available services.

This method must be run in the event loop.

call(domain: str, service: str, service_data: dict | None = None, blocking: bool = False, context: Context | None = None, limit: float | None = 10, target: dict | None = None) → bool | None[source]

Call a service.

See description of async_call for details.

has_service(domain: str, service: str) → bool[source]

Test if specified service exists.

Async friendly.

register(domain: str, service: str, service_func: Callable, schema: vol.Schema | None = None) → None[source]

Register a service.

Schema is called to coerce and validate the service data.

remove(domain: str, service: str) → None[source]

Remove a registered service from service handler.

property services

Return dictionary with per domain a list of available services.

class homeassistant.core.State(entity_id: str, state: str, attributes: Mapping[str, Any] | None = None, last_changed: datetime.datetime | None = None, last_updated: datetime.datetime | None = None, context: Context | None = None, validate_entity_id: bool | None = True)[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. context: Context in which it was created domain: Domain of this state. object_id: Object id of this state.

as_dict() → 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())

classmethod from_dict(json_dict: dict) → Any[source]

Initialize a state from a dict.

Async friendly.

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

property name

Name of this state.

class homeassistant.core.StateMachine(bus: homeassistant.core.EventBus, loop: asyncio.events.AbstractEventLoop)[source]

Helper class that tracks the state of different entities.

all(domain_filter: str | Iterable | None = None) → list[State][source]

Create a list of all states.

async_all(domain_filter: str | Iterable | None = None) → list[State][source]

Create a list of all states matching the filter.

This method must be run in the event loop.

async_available(entity_id: str) → bool[source]

Check to see if an entity_id is available to be used.

async_entity_ids(domain_filter: str | Iterable | None = None) → list[str][source]

List of entity ids that are being tracked.

This method must be run in the event loop.

async_entity_ids_count(domain_filter: str | Iterable | None = None) → int[source]

Count the entity ids that are being tracked.

This method must be run in the event loop.

async_remove(entity_id: str, context: Context | None = None) → bool[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_reserve(entity_id: str) → None[source]

Reserve a state in the state machine for an entity being added.

This must not fire an event when the state is reserved.

This avoids a race condition where multiple entities with the same entity_id are added.

async_set(entity_id: str, new_state: str, attributes: Mapping[str, Any] | None = None, force_update: bool = False, context: Context | None = None) → None[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: str | None = None) → list[str][source]

List of entity ids that are being tracked.

get(entity_id: str) → State | None[source]

Retrieve state of entity_id or None if not found.

Async friendly.

is_state(entity_id: str, state: str) → bool[source]

Test if entity exists and is in specified state.

Async friendly.

remove(entity_id: str) → bool[source]

Remove the state of an entity.

Returns boolean to indicate if an entity was removed.

set(entity_id: str, new_state: str, attributes: Mapping[str, Any] | None = None, force_update: bool = False, context: Context | None = None) → None[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.callback(func: CALLABLE_T) → CALLABLE_T[source]

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

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

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

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

Split a state entity ID into domain and object ID.

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

Test if an entity ID is a valid format.

Format: <domain>.<entity> where both are slugs.

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

Test if a state is valid.