> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/home-assistant/core/llms.txt
> Use this file to discover all available pages before exploring further.
# Event System
> Understanding Home Assistant's event-driven architecture and event bus communication
The Event System is the backbone of Home Assistant's architecture, enabling loose coupling between components through an event-driven communication model. All significant actions in Home Assistant trigger events that other components can listen to and react upon.
## EventBus Overview
The `EventBus` manages event distribution across the entire system:
```python homeassistant/core.py theme={null}
class EventBus:
"""Allow the firing of and listening for events."""
def __init__(self, hass: HomeAssistant) -> None:
"""Initialize a new event bus."""
self._listeners: defaultdict[
EventType[Any] | str, list[_FilterableJobType[Any]]
] = defaultdict(list)
self._match_all_listeners: list[_FilterableJobType[Any]] = []
self._listeners[MATCH_ALL] = self._match_all_listeners
self._hass = hass
```
The EventBus uses a dictionary to map event types to listener lists, enabling O(1) lookup for event dispatch. The `MATCH_ALL` special key allows listeners to receive all events.
## Core Event Types
Home Assistant defines several core event types in `homeassistant/const.py`:
```python homeassistant/const.py theme={null}
# Lifecycle Events
EVENT_HOMEASSISTANT_START = EventType("homeassistant_start")
EVENT_HOMEASSISTANT_STARTED = EventType("homeassistant_started")
EVENT_HOMEASSISTANT_STOP = EventType("homeassistant_stop")
EVENT_HOMEASSISTANT_FINAL_WRITE = EventType("homeassistant_final_write")
EVENT_HOMEASSISTANT_CLOSE = EventType("homeassistant_close")
# State Events
EVENT_STATE_CHANGED: EventType[EventStateChangedData] = EventType("state_changed")
EVENT_STATE_REPORTED: EventType[EventStateReportedData] = EventType("state_reported")
# Service Events
EVENT_CALL_SERVICE = "call_service"
EVENT_SERVICE_REGISTERED = "service_registered"
EVENT_SERVICE_REMOVED = "service_removed"
# Configuration Events
EVENT_CORE_CONFIG_UPDATE = "core_config_updated"
```
### Event Type Categories
Events marking major transitions in Home Assistant's lifecycle
Events fired when entity states change or are reported
Events related to service registration and execution
Events signaling configuration changes
## Event Structure
Every event is represented by the `Event` class:
```python homeassistant/core.py theme={null}
class Event(Generic[_DataT]):
"""Representation of an event within the bus."""
def __init__(
self,
event_type: EventType[_DataT] | str,
data: _DataT | None = None,
origin: EventOrigin = EventOrigin.local,
time_fired_timestamp: float | None = None,
context: Context | None = None,
) -> None:
"""Initialize a new event."""
self.event_type = event_type
self.data: _DataT = data or {}
self.origin = origin
self.time_fired_timestamp = time_fired_timestamp or time.time()
if not context:
context = Context(id=ulid_at_time(self.time_fired_timestamp))
self.context = context
if not context.origin_event:
context.origin_event = self
```
### Event Components
* **event\_type**: String or typed identifier for the event
* **data**: Dictionary containing event-specific information
* **origin**: Whether the event originated locally or remotely
* **time\_fired\_timestamp**: Unix timestamp when the event was fired
* **context**: Context tracking the chain of events
## Firing Events
### Internal Events
For performance-critical internal events, use `async_fire_internal`:
```python homeassistant/core.py theme={null}
@callback
def async_fire_internal(
self,
event_type: EventType[_DataT] | str,
event_data: _DataT | None = None,
origin: EventOrigin = EventOrigin.local,
context: Context | None = None,
time_fired: float | None = None,
) -> None:
"""Fire an event, for internal use only."""
if self._debug:
_LOGGER.debug(
"Bus:Handling %s",
_event_repr(event_type, origin, event_data)
)
listeners = self._listeners.get(event_type, EMPTY_LIST)
if event_type not in EVENTS_EXCLUDED_FROM_MATCH_ALL:
match_all_listeners = self._match_all_listeners
else:
match_all_listeners = EMPTY_LIST
event: Event[_DataT] | None = None
for job, event_filter in listeners + match_all_listeners:
if event_filter is not None:
try:
if event_data is None or not event_filter(event_data):
continue
except Exception:
_LOGGER.exception("Error in event filter")
continue
if not event:
event = Event(
event_type, event_data, origin, time_fired, context
)
try:
self._hass.async_run_hass_job(job, event)
except Exception:
_LOGGER.exception("Error running job: %s", job)
```
`async_fire_internal` is for internal Home Assistant use only and bypasses certain checks. Integrations should use `async_fire` instead.
### Public Events
For regular event firing, use the public API:
```python theme={null}
# Async context (event loop)
await hass.bus.async_fire(
"my_custom_event",
{"key": "value"},
context=context
)
# Synchronous context (external thread)
hass.bus.fire(
"my_custom_event",
{"key": "value"},
context=context
)
```
The event type length is validated (max 64 characters).
An Event object is created with the provided data and context.
All listeners for this event type and MATCH\_ALL listeners are retrieved.
Event filters are evaluated to determine which listeners should receive the event.
Each matching listener's job is executed asynchronously.
## Listening for Events
### Basic Listener
Register a listener for a specific event type:
```python theme={null}
from homeassistant.core import callback, Event
@callback
def handle_my_event(event: Event) -> None:
"""Handle my custom event."""
data = event.data
_LOGGER.info("Received event with data: %s", data)
# Register the listener
remove_listener = hass.bus.async_listen(
"my_custom_event",
handle_my_event
)
# Later, to unregister:
remove_listener()
```
Always decorate event listeners with `@callback` to indicate they're safe to run in the event loop without creating a task. This improves performance significantly.
### Event Filters
Filters allow you to selectively receive events:
```python theme={null}
from homeassistant.core import callback
@callback
def event_filter(event_data: dict) -> bool:
"""Filter events based on data."""
return event_data.get("importance") == "high"
@callback
def handle_important_event(event: Event) -> None:
"""Handle only high-importance events."""
process_important_event(event.data)
remove_listener = hass.bus.async_listen(
"my_event",
handle_important_event,
event_filter=event_filter
)
```
Event filters must be decorated with `@callback` and should execute quickly. Slow filters will block event processing for all listeners.
### Listen Once
For one-time event handling:
```python theme={null}
@callback
def handle_startup(event: Event) -> None:
"""Handle the startup event once."""
_LOGGER.info("Home Assistant has started")
initialize_integration()
remove_listener = hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STARTED,
handle_startup
)
```
## State Change Events
State changes trigger the most common events in Home Assistant:
```python homeassistant/core.py theme={null}
class EventStateChangedData(EventStateEventData):
"""EVENT_STATE_CHANGED data.
A state changed event is fired when on state write
the state is changed.
"""
entity_id: str
new_state: State | None
old_state: State | None
```
### Listening for State Changes
```python theme={null}
from homeassistant.const import EVENT_STATE_CHANGED
from homeassistant.core import callback, Event, EventStateChangedData
@callback
def handle_state_change(event: Event[EventStateChangedData]) -> None:
"""Handle state change event."""
entity_id = event.data["entity_id"]
old_state = event.data["old_state"]
new_state = event.data["new_state"]
if old_state is None:
_LOGGER.info("New entity added: %s", entity_id)
elif new_state is None:
_LOGGER.info("Entity removed: %s", entity_id)
else:
_LOGGER.info(
"Entity %s changed from %s to %s",
entity_id,
old_state.state,
new_state.state
)
hass.bus.async_listen(EVENT_STATE_CHANGED, handle_state_change)
```
### Optimized State Tracking
For better performance when tracking specific entities, use the event helpers:
```python theme={null}
from homeassistant.helpers.event import async_track_state_change_event
@callback
def handle_light_change(event: Event[EventStateChangedData]) -> None:
"""Handle changes to specific light."""
new_state = event.data["new_state"]
if new_state and new_state.state == "on":
_LOGGER.info("Light turned on")
# This is much faster than listening to all state changes
remove_listener = async_track_state_change_event(
hass,
["light.living_room", "light.bedroom"],
handle_light_change
)
```
`async_track_state_change_event` uses an optimized index that routes events directly to listeners interested in specific entities, avoiding the need to process all state change events.
## Match All Listeners
Listen to every event (use sparingly for debugging/monitoring):
```python theme={null}
from homeassistant.const import MATCH_ALL
@callback
def log_all_events(event: Event) -> None:
"""Log every event for debugging."""
_LOGGER.debug(
"Event: %s - %s",
event.event_type,
event.data
)
remove_listener = hass.bus.async_listen(MATCH_ALL, log_all_events)
```
MATCH\_ALL listeners receive every single event, which can significantly impact performance. Use only for temporary debugging or specialized monitoring tools.
## Event Context
Every event includes a context tracking its origin:
```python homeassistant/core.py theme={null}
class Context:
"""The context that triggered something."""
def __init__(
self,
user_id: str | None = None,
parent_id: str | None = None,
id: str | None = None,
) -> None:
self.id = id or ulid_now()
self.user_id = user_id
self.parent_id = parent_id
self.origin_event: Event[Any] | None = None
```
### Using Context
```python theme={null}
# Check who triggered an event
@callback
def handle_light_change(event: Event[EventStateChangedData]) -> None:
if event.context.user_id:
user = await hass.auth.async_get_user(event.context.user_id)
_LOGGER.info("Light changed by user: %s", user.name)
else:
_LOGGER.info("Light changed by automation or integration")
# Propagate context in automation chains
context = Context(parent_id=trigger_event.context.id)
hass.bus.async_fire("my_event", {}, context=context)
```
## Event Origin
Events can originate locally or from remote systems:
```python homeassistant/core.py theme={null}
class EventOrigin(enum.Enum):
"""Represent the origin of an event."""
local = "LOCAL" # Event originated on this Home Assistant instance
remote = "REMOTE" # Event came from a remote instance
```
This is primarily used in multi-instance setups or when integrating with external systems.
## Performance Considerations
**Use @callback**: Always decorate listeners with `@callback` when possible\
**Specific Events**: Listen to specific events rather than MATCH\_ALL\
**Event Filters**: Use filters to reduce unnecessary job creation\
**Fast Execution**: Keep listener code fast; offload heavy work to tasks
## Best Practices
1. **Always unregister listeners** when your component unloads
2. **Use typed event data** with TypedDict for type safety
3. **Handle None states** in state change events (entity added/removed)
4. **Propagate context** to maintain audit trails
5. **Avoid blocking operations** in event listeners