HomeAssistant Core

The central coordinator managing the event loop, components, and system lifecycle

The HomeAssistant class is the heart of the system, coordinating all operations and managing the asyncio event loop. Every integration, automation, and service interacts with this central hub.

Core Class Structure

The HomeAssistant object is the main entry point for all Home Assistant operations:

homeassistant/core.py

class HomeAssistant:
    """Root object of the Home Assistant home automation."""
    
    def __init__(self, config_dir: str) -> None:
        """Initialize new Home Assistant object."""
        self.data = HassDict()
        self.loop = asyncio.get_running_loop()
        self.bus = EventBus(self)
        self.services = ServiceRegistry(self)
        self.states = StateMachine(self.bus, self.loop)
        self.config = Config(self, config_dir)
        self.state: CoreState = CoreState.not_running

Key Components

data

Shared storage dictionary for cross-component communication

loop

The asyncio event loop managing all async operations

bus

Event bus for publishing and subscribing to events

services

Registry of all available services across domains

states

State machine tracking all entity states

config

System configuration including paths and settings

Lifecycle Management

Startup Process

Home Assistant follows a well-defined startup sequence:

homeassistant/core.py

async def async_start(self) -> None:
    """Finalize startup from inside the event loop."""
    _LOGGER.info("Starting Home Assistant %s", __version__)
    
    self.set_state(CoreState.starting)
    self.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE)
    self.bus.async_fire_internal(EVENT_HOMEASSISTANT_START)
    
    # Wait for startup tasks with timeout
    if not self._tasks:
        pending = None
    else:
        _done, pending = await asyncio.wait(
            self._tasks, timeout=TIMEOUT_EVENT_START
        )
    
    # Allow automations to set up triggers
    await asyncio.sleep(0)
    
    self.set_state(CoreState.running)
    self.bus.async_fire_internal(EVENT_CORE_CONFIG_UPDATE)
    self.bus.async_fire_internal(EVENT_HOMEASSISTANT_STARTED)

1. Event Loop Started

   The asyncio event loop begins running in the main thread.

2. Core Integrations Load

   Essential integrations like homeassistant and persistent_notification are loaded.

3. START Event Fired

   The EVENT_HOMEASSISTANT_START event signals integrations to begin setup.

4. Tasks Complete

   All startup tasks must complete within 15 seconds (TIMEOUT_EVENT_START).

5. STARTED Event Fired

   The EVENT_HOMEASSISTANT_STARTED event indicates the system is fully operational.

Shutdown Process

Shutdown occurs in multiple stages to ensure clean termination:

homeassistant/core.py

async def async_stop(self, exit_code: int = 0, *, force: bool = False) -> None:
    """Stop Home Assistant and shuts down all threads."""
    
    # Stage 1 - Run shutdown jobs
    try:
        async with self.timeout.async_timeout(STOPPING_STAGE_SHUTDOWN_TIMEOUT):
            tasks = []
            for job in self._shutdown_jobs:
                task_or_none = self.async_run_hass_job(job.job, *job.args)
                if task_or_none:
                    tasks.append(task_or_none)
            if tasks:
                await asyncio.gather(*tasks, return_exceptions=True)
    except TimeoutError:
        _LOGGER.warning("Timed out waiting for shutdown jobs")
    
    # Stage 2 - Stop integrations
    self.set_state(CoreState.stopping)
    self.bus.async_fire_internal(EVENT_HOMEASSISTANT_STOP)
    
    # Stage 3 - Final write
    self.set_state(CoreState.final_write)
    self.bus.async_fire_internal(EVENT_HOMEASSISTANT_FINAL_WRITE)
    
    # Stage 4 - Close
    self.set_state(CoreState.not_running)
    self.bus.async_fire_internal(EVENT_HOMEASSISTANT_CLOSE)
    
    self.set_state(CoreState.stopped)

The shutdown process has timeouts at each stage to prevent a single misbehaving integration from blocking shutdown indefinitely.

Task Management

Home Assistant provides sophisticated task management capabilities:

Creating Tasks

homeassistant/core.py

@callback
def async_create_task[_R](
    self,
    target: Coroutine[Any, Any, _R],
    name: str | None = None,
    eager_start: bool = True,
) -> asyncio.Task[_R]:
    """Create a task from within the event loop."""
    if eager_start:
        task = create_eager_task(target, name=name, loop=self.loop)
        if task.done():
            return task
    else:
        task = self.loop.create_task(target, name=name)
    
    self._tasks.add(task)
    task.add_done_callback(self._tasks.remove)
    return task

Eager Start: When eager_start=True, the task begins executing immediately rather than waiting for the next event loop iteration. This significantly improves performance for short-running tasks.

Background Tasks

Background tasks are designed for long-running operations:

homeassistant/core.py

@callback
def async_create_background_task[_R](
    self, 
    target: Coroutine[Any, Any, _R], 
    name: str,
    eager_start: bool = True
) -> asyncio.Task[_R]:
    """Create a background task.
    
    A background task is different from a normal task:
    - Will not block startup
    - Will be automatically cancelled on shutdown  
    - Calls to async_block_till_done will not wait for completion
    """
    if eager_start:
        task = create_eager_task(target, name=name, loop=self.loop)
        if task.done():
            return task
    else:
        task = self.loop.create_task(target, name=name)
    
    self._background_tasks.add(task)
    task.add_done_callback(self._background_tasks.remove)
    return task

Use background tasks for operations that should not prevent Home Assistant from starting or stopping, such as continuous monitoring loops or periodic sync operations.

Job Execution

The HassJob system provides type-aware job execution:

homeassistant/core.py

class HassJob[**_P, _R_co]:
    """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.
    """
    
    def __init__(
        self,
        target: Callable[_P, _R_co],
        name: str | None = None,
        *,
        cancel_on_shutdown: bool | None = None,
        job_type: HassJobType | None = None,
    ) -> None:
        self.target = target
        self.name = name
        self._cancel_on_shutdown = cancel_on_shutdown
        if job_type:
            self._cache["job_type"] = job_type

Job Types

Coroutine Function

# Async functions run directly in event loop
async def my_async_handler(event):
    await some_async_operation()
    
job = HassJob(my_async_handler)
# job.job_type == HassJobType.Coroutinefunction

Thread Safety

Home Assistant enforces strict thread safety:

homeassistant/core.py

def verify_event_loop_thread(self, what: str) -> None:
    """Report and raise if not running in event loop thread."""
    if self.loop_thread_id != threading.get_ident():
        from .helpers import frame
        frame.report_non_thread_safe_operation(what)

Calling from External Threads

When calling Home Assistant methods from external threads:

from homeassistant.util.async_ import run_callback_threadsafe

# From external thread
future = run_callback_threadsafe(
    hass.loop,
    hass.states.async_set,
    "sensor.example",
    "new_value"
)
result = future.result()  # Block until complete

Always use run_callback_threadsafe when calling Home Assistant methods from background threads to avoid race conditions and data corruption.

Accessing the Instance

In contexts where passing hass is impractical, you can retrieve it:

homeassistant/core.py

@callback
def async_get_hass() -> HomeAssistant:
    """Return the HomeAssistant instance.
    
    Raises HomeAssistantError when called from the wrong thread.
    """
    if not (hass := async_get_hass_or_none()):
        raise HomeAssistantError(
            "async_get_hass called from the wrong thread"
        )
    return hass

Use async_get_hass() sparingly. It’s better to pass hass explicitly to maintain clear dependencies and improve testability.

Configuration Management

The config object provides access to system configuration:

hass.config.config_dir      # Configuration directory path
hass.config.latitude        # Home location latitude  
hass.config.longitude       # Home location longitude
hass.config.elevation       # Home elevation
hass.config.time_zone       # System timezone
hass.config.units           # Unit system (metric/imperial)
hass.config.safe_mode       # Whether running in safe mode
hass.config.recovery_mode   # Whether running in recovery mode

Data Storage

The hass.data dictionary provides shared storage:

# Store integration-specific data
hass.data["my_integration"] = MyIntegrationData()

# Use HassKey for type safety
from homeassistant.util.hass_dict import HassKey

MY_DATA_KEY: HassKey[MyIntegrationData] = HassKey("my_integration")
hass.data[MY_DATA_KEY] = MyIntegrationData()

Use the integration domain as the key to avoid conflicts. For complex data structures, consider using HassKey for type safety.