Event Handling

Advanced patterns for handling events in Home Assistant Core

Event handling is a fundamental concept in Home Assistant. The event bus is the central nervous system that allows components to communicate asynchronously. Understanding how to work with events efficiently is crucial for building responsive integrations.

Understanding the Event Bus

The event bus (EventBus) allows firing and listening for events throughout Home Assistant. Every state change, service call, and custom integration action can be represented as an event.

Key Event Types

Home Assistant Core defines several built-in event types:

• EVENT_STATE_CHANGED - Fired when an entity’s state changes
• EVENT_STATE_REPORTED - Fired when state is updated but unchanged
• EVENT_HOMEASSISTANT_START - Fired when Home Assistant starts
• EVENT_HOMEASSISTANT_STARTED - Fired after all components are loaded
• EVENT_HOMEASSISTANT_STOP - Fired when Home Assistant begins shutdown
• EVENT_SERVICE_REGISTERED - Fired when a service is registered
• EVENT_CALL_SERVICE - Fired when a service is called

Reference: homeassistant/const.py

Listening to Events

Basic Event Listener

Use async_listen to subscribe to events. This method must be run in the event loop:

from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.const import EVENT_STATE_CHANGED

@callback
def handle_event(event: Event) -> None:
    """Handle the event."""
    print(f"Event fired: {event.event_type}")
    print(f"Event data: {event.data}")

def async_setup(hass: HomeAssistant):
    """Set up the component."""
    # Register event listener
    remove_listener = hass.bus.async_listen(
        EVENT_STATE_CHANGED,
        handle_event
    )
    
    # Store the remove callback to clean up later
    return True

Reference: homeassistant/core.py:1565

Event Filters

Event filters allow you to pre-filter events before your handler is called, improving performance:

from homeassistant.core import callback

@callback
def event_filter(event_data: dict) -> bool:
    """Filter events - return True to handle, False to skip."""
    # Only handle events for specific entity
    return event_data.get("entity_id") == "light.living_room"

@callback
def handle_filtered_event(event: Event) -> None:
    """Handle only filtered events."""
    print(f"Light changed: {event.data}")

remove = hass.bus.async_listen(
    EVENT_STATE_CHANGED,
    handle_filtered_event,
    event_filter=event_filter
)

Reference: homeassistant/core.py:1569

Event filters must be decorated with @callback to ensure they run synchronously in the event loop.

Tracking State Changes

Track State Change Events

For entity-specific state tracking, use the optimized async_track_state_change_event helper:

from homeassistant.helpers.event import async_track_state_change_event
from homeassistant.core import Event, callback

@callback
def state_changed(event: Event) -> None:
    """Handle state change."""
    entity_id = event.data["entity_id"]
    old_state = event.data["old_state"]
    new_state = event.data["new_state"]
    
    if old_state and new_state:
        print(f"{entity_id}: {old_state.state} -> {new_state.state}")

# Track specific entities (more efficient than filtering manually)
remove = async_track_state_change_event(
    hass,
    ["light.living_room", "light.bedroom"],
    state_changed
)

Reference: homeassistant/helpers/event.py:309

Track by Domain

Track when entities are added to specific domains:

from homeassistant.helpers.event import async_track_state_added_domain

@callback
def handle_new_light(event: Event) -> None:
    """Handle new light entity."""
    entity_id = event.data["entity_id"]
    new_state = event.data["new_state"]
    print(f"New light added: {entity_id}")

# Track when new lights are added
remove = async_track_state_added_domain(
    hass,
    "light",
    handle_new_light
)

Reference: homeassistant/helpers/event.py:653

Advanced Tracking Patterns

Filtered State Change Tracking

Use async_track_state_change_filtered for dynamic entity tracking:

from homeassistant.helpers.event import (
    async_track_state_change_filtered,
    TrackStates,
)

@callback
def handle_change(event: Event) -> None:
    """Handle tracked state changes."""
    print(f"Tracked entity changed: {event.data['entity_id']}")

# Create tracker with initial entities and domains
track_states = TrackStates(
    all_states=False,
    entities={"light.living_room", "switch.kitchen"},
    domains={"climate"},
)

tracker = async_track_state_change_filtered(
    hass,
    track_states,
    handle_change
)

# Later, update what we're tracking
new_track_states = TrackStates(
    all_states=False,
    entities={"light.bedroom"},
    domains={"climate", "fan"},
)
tracker.async_update_listeners(new_track_states)

# Clean up when done
tracker.async_remove()

Reference: homeassistant/helpers/event.py:867

Template Tracking

Track template results and react to changes:

from homeassistant.helpers.event import (
    async_track_template_result,
    TrackTemplate,
    TrackTemplateResult,
)
from homeassistant.helpers.template import Template

@callback
def template_changed(
    event: Event | None,
    updates: list[TrackTemplateResult]
) -> None:
    """Handle template result changes."""
    for update in updates:
        print(f"Template result: {update.result}")
        print(f"Previous result: {update.last_result}")

# Create template to track
template = Template(
    "{{ states('sensor.temperature') | float > 20 }}",
    hass
)

track_template = TrackTemplate(template, variables=None)

# Start tracking
info = async_track_template_result(
    hass,
    [track_template],
    template_changed
)

# Clean up
info.async_remove()

Reference: homeassistant/helpers/event.py:1343

Time-Based Event Tracking

Track Point in Time

Schedule a callback at a specific time:

from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.util import dt as dt_util
from datetime import timedelta

@callback
def timed_callback(now: datetime) -> None:
    """Called at scheduled time."""
    print(f"Timer fired at {now}")

# Schedule for 5 minutes from now
point_in_time = dt_util.utcnow() + timedelta(minutes=5)
remove = async_track_point_in_utc_time(
    hass,
    timed_callback,
    point_in_time
)

# Cancel if needed
remove()

Reference: homeassistant/helpers/event.py:1544

Firing Events

Fire Internal Events

For internal core use, fire events directly without extra validation:

from homeassistant.core import EventOrigin

# Fire internal event (most efficient)
hass.bus.async_fire_internal(
    "my_custom_event",
    {"key": "value"},
)

Reference: homeassistant/core.py:1492

Fire Public Events

For integration events that may be consumed externally:

# Fire public event with context
hass.bus.async_fire(
    "my_integration_event",
    {"device_id": "abc123", "status": "online"},
    origin=EventOrigin.local,
    context=some_context
)

Reference: homeassistant/core.py:1470

Event Dispatching Optimization

The event system uses several optimizations:

1. Entity ID Indexing: State change events are indexed by entity_id for fast routing
2. Event Filters: Pre-filtering prevents unnecessary job creation
3. Call Soon: Events are dispatched with call_soon to ensure proper event loop iteration
4. Job Types: Callbacks are executed directly, coroutines are scheduled efficiently

Reference: homeassistant/helpers/event.py:340-367

Best Practices

Use @callback Decorator

Always decorate synchronous event handlers with @callback:

@callback
def my_handler(event: Event) -> None:
    """Handle event synchronously."""
    # Synchronous code only
    pass

Async Handlers

For async operations, use coroutine functions:

async def my_async_handler(event: Event) -> None:
    """Handle event asynchronously."""
    await some_async_operation()

Clean Up Listeners

Always remove listeners when they’re no longer needed:

class MyComponent:
    def __init__(self, hass: HomeAssistant):
        self._remove_listener = hass.bus.async_listen(
            EVENT_STATE_CHANGED,
            self.handle_event
        )
    
    async def async_will_remove_from_hass(self):
        """Clean up."""
        if self._remove_listener:
            self._remove_listener()

Avoid Long-Running Operations

Event handlers should be fast. For long operations, create a task:

@callback
def handle_event(event: Event) -> None:
    """Handle event."""
    # Create background task for slow operation
    hass.async_create_task(
        slow_operation(event.data)
    )

async def slow_operation(data: dict) -> None:
    """Perform slow operation."""
    await asyncio.sleep(10)
    # Do work

Performance Considerations

1. Use specific tracking helpers - async_track_state_change_event is more efficient than manual filtering
2. Implement event filters - Pre-filter events to avoid unnecessary processing
3. Index by entity_id - The event system automatically optimizes entity-specific tracking
4. Batch updates - When possible, batch multiple state changes together

Common Pitfalls

• Don’t block the event loop in event handlers
• Always clean up listeners to prevent memory leaks
• Don’t modify event data directly (it may be shared)
• Use @callback for synchronous handlers to avoid thread issues

Related Resources

• Async Programming
• State Management
• Service Registration