core/homeassistant/helpers/start.py

"""Helpers to help during startup."""

from __future__ import annotations

from collections.abc import Callable, Coroutine
from typing import Any

from homeassistant.const import EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STARTED
from homeassistant.core import (
    CALLBACK_TYPE,
    CoreState,
    Event,
    HassJob,
    HomeAssistant,
    callback,
)
from homeassistant.util.event_type import EventType

from .typing import NoEventData


@callback
def _async_at_core_state(
    hass: HomeAssistant,
    at_start_cb: Callable[[HomeAssistant], Coroutine[Any, Any, None] | None],
    event_type: EventType[NoEventData],
    check_state: Callable[[HomeAssistant], bool],
) -> CALLBACK_TYPE:
    """Execute a job at_start_cb when Home Assistant has the wanted state.

    The job is executed immediately if Home Assistant is in the wanted state.
    Will wait for event specified by event_type if it isn't.
    """
    at_start_job = HassJob(at_start_cb)
    if check_state(hass):
        hass.async_run_hass_job(at_start_job, hass)
        return lambda: None

    unsub: CALLBACK_TYPE | None = None

    @callback
    def _matched_event(event: Event) -> None:
        """Call the callback when Home Assistant started."""
        hass.async_run_hass_job(at_start_job, hass)
        nonlocal unsub
        unsub = None

    @callback
    def cancel() -> None:
        if unsub:
            unsub()

    unsub = hass.bus.async_listen_once(event_type, _matched_event)
    return cancel


@callback
def async_at_start(
    hass: HomeAssistant,
    at_start_cb: Callable[[HomeAssistant], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
    """Execute a job at_start_cb when Home Assistant is starting.

    The job is executed immediately if Home Assistant is already starting or started.
    Will wait for EVENT_HOMEASSISTANT_START if it isn't.
    """

    def _is_running(hass: HomeAssistant) -> bool:
        return hass.is_running

    return _async_at_core_state(
        hass, at_start_cb, EVENT_HOMEASSISTANT_START, _is_running
    )


@callback
def async_at_started(
    hass: HomeAssistant,
    at_start_cb: Callable[[HomeAssistant], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
    """Execute a job at_start_cb when Home Assistant has started.

    The job is executed immediately if Home Assistant is already started.
    Will wait for EVENT_HOMEASSISTANT_STARTED if it isn't.
    """

    def _is_started(hass: HomeAssistant) -> bool:
        return hass.state is CoreState.running

    return _async_at_core_state(
        hass, at_start_cb, EVENT_HOMEASSISTANT_STARTED, _is_started
    )