core/homeassistant/helpers/dispatcher.py

"""Helpers for Home Assistant dispatcher & internal component/platform."""

from __future__ import annotations

from collections import defaultdict
from collections.abc import Callable, Coroutine
from functools import partial
import logging
from typing import Any, overload

from homeassistant.core import (
    HassJob,
    HassJobType,
    HomeAssistant,
    callback,
    get_hassjob_callable_job_type,
)
from homeassistant.loader import bind_hass
from homeassistant.util.async_ import run_callback_threadsafe
from homeassistant.util.logging import catch_log_exception, log_exception

# Explicit reexport of 'SignalType' for backwards compatibility
from homeassistant.util.signal_type import SignalType as SignalType  # noqa: PLC0414

_LOGGER = logging.getLogger(__name__)
DATA_DISPATCHER = "dispatcher"


type _DispatcherDataType[*_Ts] = dict[
    SignalType[*_Ts] | str,
    dict[
        Callable[[*_Ts], Any] | Callable[..., Any],
        HassJob[..., None | Coroutine[Any, Any, None]] | None,
    ],
]


@overload
@bind_hass
def dispatcher_connect[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts], target: Callable[[*_Ts], None]
) -> Callable[[], None]: ...


@overload
@bind_hass
def dispatcher_connect(
    hass: HomeAssistant, signal: str, target: Callable[..., None]
) -> Callable[[], None]: ...


@bind_hass  # type: ignore[misc]  # workaround; exclude typing of 2 overload in func def
def dispatcher_connect[*_Ts](
    hass: HomeAssistant,
    signal: SignalType[*_Ts],
    target: Callable[[*_Ts], None],
) -> Callable[[], None]:
    """Connect a callable function to a signal."""
    async_unsub = run_callback_threadsafe(
        hass.loop, async_dispatcher_connect, hass, signal, target
    ).result()

    def remove_dispatcher() -> None:
        """Remove signal listener."""
        run_callback_threadsafe(hass.loop, async_unsub).result()

    return remove_dispatcher


@callback
def _async_remove_dispatcher[*_Ts](
    dispatchers: _DispatcherDataType[*_Ts],
    signal: SignalType[*_Ts] | str,
    target: Callable[[*_Ts], Any] | Callable[..., Any],
) -> None:
    """Remove signal listener."""
    try:
        signal_dispatchers = dispatchers[signal]
        del signal_dispatchers[target]
        # Cleanup the signal dict if it is now empty
        # to prevent memory leaks
        if not signal_dispatchers:
            del dispatchers[signal]
    except (KeyError, ValueError):
        # KeyError is key target listener did not exist
        # ValueError if listener did not exist within signal
        _LOGGER.warning("Unable to remove unknown dispatcher %s", target)


@overload
@callback
@bind_hass
def async_dispatcher_connect[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts], target: Callable[[*_Ts], Any]
) -> Callable[[], None]: ...


@overload
@callback
@bind_hass
def async_dispatcher_connect(
    hass: HomeAssistant, signal: str, target: Callable[..., Any]
) -> Callable[[], None]: ...


@callback
@bind_hass
def async_dispatcher_connect[*_Ts](
    hass: HomeAssistant,
    signal: SignalType[*_Ts] | str,
    target: Callable[[*_Ts], Any] | Callable[..., Any],
) -> Callable[[], None]:
    """Connect a callable function to a signal.

    This method must be run in the event loop.
    """
    if DATA_DISPATCHER not in hass.data:
        hass.data[DATA_DISPATCHER] = defaultdict(dict)
    dispatchers: _DispatcherDataType[*_Ts] = hass.data[DATA_DISPATCHER]
    dispatchers[signal][target] = None
    # Use a partial for the remove since it uses
    # less memory than a full closure since a partial copies
    # the body of the function and we don't have to store
    # many different copies of the same function
    return partial(_async_remove_dispatcher, dispatchers, signal, target)


@overload
@bind_hass
def dispatcher_send[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts], *args: *_Ts
) -> None: ...


@overload
@bind_hass
def dispatcher_send(hass: HomeAssistant, signal: str, *args: Any) -> None: ...


@bind_hass  # type: ignore[misc]  # workaround; exclude typing of 2 overload in func def
def dispatcher_send[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts], *args: *_Ts
) -> None:
    """Send signal and data."""
    hass.loop.call_soon_threadsafe(async_dispatcher_send_internal, hass, signal, *args)


def _format_err[*_Ts](
    signal: SignalType[*_Ts] | str,
    target: Callable[[*_Ts], Any] | Callable[..., Any],
    *args: Any,
) -> str:
    """Format error message."""

    return (
        # Functions wrapped in partial do not have a __name__
        f"Exception in {getattr(target, "__name__", None) or target} "
        f"when dispatching '{signal}': {args}"
    )


def _generate_job[*_Ts](
    signal: SignalType[*_Ts] | str, target: Callable[[*_Ts], Any] | Callable[..., Any]
) -> HassJob[..., Coroutine[Any, Any, None] | None]:
    """Generate a HassJob for a signal and target."""
    job_type = get_hassjob_callable_job_type(target)
    name = f"dispatcher {signal}"
    if job_type is HassJobType.Callback:
        # We will catch exceptions in the callback to avoid
        # wrapping the callback since calling wraps() is more
        # expensive than the whole dispatcher_send process
        return HassJob(target, name, job_type=job_type)
    return HassJob(
        catch_log_exception(
            target, partial(_format_err, signal, target), job_type=job_type
        ),
        name,
        job_type=job_type,
    )


@overload
@callback
@bind_hass
def async_dispatcher_send[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts], *args: *_Ts
) -> None: ...


@overload
@callback
@bind_hass
def async_dispatcher_send(hass: HomeAssistant, signal: str, *args: Any) -> None: ...


@callback
@bind_hass
def async_dispatcher_send[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts] | str, *args: *_Ts
) -> None:
    """Send signal and data.

    This method must be run in the event loop.
    """
    # We turned on asyncio debug in April 2024 in the dev containers
    # in the hope of catching some of the issues that have been
    # reported. It will take a while to get all the issues fixed in
    # custom components.
    #
    # In 2025.5 we should guard the `verify_event_loop_thread`
    # check with a check for the `hass.config.debug` flag being set as
    # long term we don't want to be checking this in production
    # environments since it is a performance hit.
    hass.verify_event_loop_thread("async_dispatcher_send")
    async_dispatcher_send_internal(hass, signal, *args)


@callback
@bind_hass
def async_dispatcher_send_internal[*_Ts](
    hass: HomeAssistant, signal: SignalType[*_Ts] | str, *args: *_Ts
) -> None:
    """Send signal and data.

    This method is intended to only be used by core internally
    and should not be considered a stable API. We will make
    breaking changes to this function in the future and it
    should not be used in integrations.

    This method must be run in the event loop.
    """
    if (maybe_dispatchers := hass.data.get(DATA_DISPATCHER)) is None:
        return
    dispatchers: _DispatcherDataType[*_Ts] = maybe_dispatchers
    if (target_list := dispatchers.get(signal)) is None:
        return

    for target, job in list(target_list.items()):
        if job is None:
            job = _generate_job(signal, target)
            target_list[target] = job
        # We do not wrap Callback jobs in catch_log_exception since
        # single use dispatchers spend more time wrapping the callback
        # than the actual callback takes to run in many cases.
        if job.job_type is HassJobType.Callback:
            try:
                job.target(*args)
            except Exception:  # noqa: BLE001
                log_exception(partial(_format_err, signal, target), *args)  # type: ignore[arg-type]
        else:
            hass.async_run_hass_job(job, *args)