"""Module to coordinate user intentions."""

from abc import abstractmethod
import asyncio
from collections.abc import Callable, Collection, Coroutine, Iterable
import dataclasses
from dataclasses import dataclass, field
from enum import Enum, StrEnum, auto
from itertools import groupby
import logging
from typing import Any, override

from propcache.api import cached_property
import voluptuous as vol

from homeassistant.components.homeassistant.exposed_entities import async_should_expose
from homeassistant.const import ATTR_ENTITY_ID, EntityStateAttribute
from homeassistant.core import Context, HomeAssistant, State, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.hass_dict import HassKey

from . import (
    area_registry as ar,
    config_validation as cv,
    device_registry as dr,
    entity_registry as er,
    floor_registry as fr,
)
from .deprecation import EnumWithDeprecatedMembers
from .typing import VolSchemaType

_LOGGER = logging.getLogger(__name__)
type _SlotsType = dict[str, Any]
type _IntentSlotsType = dict[
    str | tuple[str, str], IntentSlotInfo | VolSchemaType | Callable[[Any], Any]
]

INTENT_TURN_OFF = "HassTurnOff"
INTENT_TURN_ON = "HassTurnOn"
INTENT_TOGGLE = "HassToggle"
INTENT_GET_STATE = "HassGetState"
INTENT_NEVERMIND = "HassNevermind"
INTENT_SET_POSITION = "HassSetPosition"
INTENT_STOP_MOVING = "HassStopMoving"
INTENT_START_TIMER = "HassStartTimer"
INTENT_CANCEL_TIMER = "HassCancelTimer"
INTENT_CANCEL_ALL_TIMERS = "HassCancelAllTimers"
INTENT_INCREASE_TIMER = "HassIncreaseTimer"
INTENT_DECREASE_TIMER = "HassDecreaseTimer"
INTENT_PAUSE_TIMER = "HassPauseTimer"
INTENT_UNPAUSE_TIMER = "HassUnpauseTimer"
INTENT_TIMER_STATUS = "HassTimerStatus"
INTENT_GET_CURRENT_DATE = "HassGetCurrentDate"
INTENT_GET_CURRENT_TIME = "HassGetCurrentTime"
INTENT_RESPOND = "HassRespond"
INTENT_BROADCAST = "HassBroadcast"
INTENT_GET_TEMPERATURE = "HassClimateGetTemperature"

SLOT_SCHEMA = vol.Schema({}, extra=vol.ALLOW_EXTRA)

DATA_KEY: HassKey[dict[str, IntentHandler]] = HassKey("intent")

SPEECH_TYPE_PLAIN = "plain"
SPEECH_TYPE_SSML = "ssml"


@callback
def async_register(hass: HomeAssistant, handler: IntentHandler) -> None:
    """Register an intent with Home Assistant."""
    if (intents := hass.data.get(DATA_KEY)) is None:
        intents = {}
        hass.data[DATA_KEY] = intents

    assert getattr(handler, "intent_type", None), "intent_type should be set"

    if handler.intent_type in intents:
        _LOGGER.warning(
            "Intent %s is being overwritten by %s", handler.intent_type, handler
        )

    intents[handler.intent_type] = handler


@callback
def async_remove(hass: HomeAssistant, intent_type: str) -> None:
    """Remove an intent from Home Assistant."""
    if (intents := hass.data.get(DATA_KEY)) is None:
        return

    intents.pop(intent_type, None)


@callback
def async_get(hass: HomeAssistant) -> Iterable[IntentHandler]:
    """Return registered intents."""
    return hass.data.get(DATA_KEY, {}).values()


async def async_handle(
    hass: HomeAssistant,
    platform: str,
    intent_type: str,
    slots: _SlotsType | None = None,
    text_input: str | None = None,
    context: Context | None = None,
    language: str | None = None,
    assistant: str | None = None,
    device_id: str | None = None,
    satellite_id: str | None = None,
    conversation_agent_id: str | None = None,
) -> IntentResponse:
    """Handle an intent."""
    handler = hass.data.get(DATA_KEY, {}).get(intent_type)

    if handler is None:
        raise UnknownIntent(f"Unknown intent {intent_type}")

    if context is None:
        context = Context()

    if language is None:
        language = hass.config.language

    intent = Intent(
        hass,
        platform=platform,
        intent_type=intent_type,
        slots=slots or {},
        text_input=text_input,
        context=context,
        language=language,
        assistant=assistant,
        device_id=device_id,
        satellite_id=satellite_id,
        conversation_agent_id=conversation_agent_id,
    )

    try:
        _LOGGER.info("Triggering intent handler %s", handler)
        result = await handler.async_handle(intent)
    except vol.Invalid as err:
        _LOGGER.warning("Received invalid slot info for %s: %s", intent_type, err)
        raise InvalidSlotInfo(f"Received invalid slot info for {intent_type}") from err
    except IntentError:
        raise  # bubble up intent related errors
    except Exception as err:
        _LOGGER.exception("Error handling %s", intent_type)
        raise IntentUnexpectedError(f"Error handling {intent_type}") from err
    return result


class IntentError(HomeAssistantError):
    """Base class for intent related errors."""


class UnknownIntent(IntentError):
    """When the intent is not registered."""


class InvalidSlotInfo(IntentError):
    """When the slot data is invalid."""


class IntentHandleError(IntentError):
    """Error while handling intent."""

    def __init__(self, message: str = "", response_key: str | None = None) -> None:
        """Initialize error."""
        super().__init__(message)
        self.response_key = response_key


class IntentUnexpectedError(IntentError):
    """Unexpected error while handling intent."""


class MatchFailedError(IntentError):
    """Error when target matching fails."""

    def __init__(
        self,
        result: MatchTargetsResult,
        constraints: MatchTargetsConstraints,
        preferences: MatchTargetsPreferences | None = None,
    ) -> None:
        """Initialize error."""
        super().__init__()

        self.result = result
        self.constraints = constraints
        self.preferences = preferences

    @override
    def __str__(self) -> str:
        """Return string representation."""
        return (
            f"<MatchFailedError result={self.result},"
            f" constraints={self.constraints},"
            f" preferences={self.preferences}>"
        )


class NoStatesMatchedError(MatchFailedError):
    """Error when no states match the intent's constraints."""

    def __init__(
        self,
        reason: MatchFailedReason,
        name: str | None = None,
        area: str | None = None,
        floor: str | None = None,
        domains: set[str] | None = None,
        device_classes: set[str] | None = None,
    ) -> None:
        """Initialize error."""
        super().__init__(
            result=MatchTargetsResult(False, reason),
            constraints=MatchTargetsConstraints(
                name=name,
                area_name=area,
                floor_name=floor,
                domains=domains,
                device_classes=device_classes,
            ),
        )


class MatchFailedReason(Enum):
    """Possible reasons for match failure in async_match_targets."""

    NAME = auto()
    """No entities matched name constraint."""

    AREA = auto()
    """No entities matched area constraint."""

    FLOOR = auto()
    """No entities matched floor constraint."""

    DOMAIN = auto()
    """No entities matched domain constraint."""

    DEVICE_CLASS = auto()
    """No entities matched device class constraint."""

    FEATURE = auto()
    """No entities matched supported features constraint."""

    STATE = auto()
    """No entities matched required states constraint."""

    ASSISTANT = auto()
    """No entities matched exposed to assistant constraint."""

    INVALID_AREA = auto()
    """Area name from constraint does not exist."""

    INVALID_FLOOR = auto()
    """Floor name from constraint does not exist."""

    DUPLICATE_NAME = auto()
    """Two or more entities matched the same name constraint.

    Could not be disambiguated.
    """

    MULTIPLE_TARGETS = auto()
    """Two or more entities matched when a single target is required."""

    def is_no_entities_reason(self) -> bool:
        """Return True if the match failed because no entities matched."""
        return self not in (
            MatchFailedReason.INVALID_AREA,
            MatchFailedReason.INVALID_FLOOR,
            MatchFailedReason.DUPLICATE_NAME,
        )


@dataclass
class MatchTargetsResult:
    """Result from async_match_targets."""

    is_match: bool
    """True if one or more entities matched."""

    no_match_reason: MatchFailedReason | None = None
    """Reason for failed match when is_match = False."""

    states: list[State] = field(default_factory=list)
    """List of matched entity states."""

    no_match_name: str | None = None
    """Name of invalid area/floor or duplicate name.

    Used when match fails for those reasons.
    """

    areas: list[ar.AreaEntry] = field(default_factory=list)
    """Areas that were targeted."""

    floors: list[fr.FloorEntry] = field(default_factory=list)
    """Floors that were targeted."""


@dataclass
class MatchTargetsConstraints:
    """Constraints for async_match_targets."""

    name: str | None = None
    """Entity name or alias."""

    area_name: str | None = None
    """Area name, id, or alias."""

    floor_name: str | None = None
    """Floor name, id, or alias."""

    domains: Collection[str] | None = None
    """Domain names."""

    device_classes: Collection[str] | None = None
    """Device class names."""

    features: int | None = None
    """Required supported features."""

    states: Collection[str] | None = None
    """Required states for entities."""

    assistant: str | None = None
    """Name of assistant that entities should be exposed to."""

    allow_duplicate_names: bool = False
    """True if entities with duplicate names are allowed in result."""

    single_target: bool = False
    """True if result must contain a single target."""

    @property
    def has_constraints(self) -> bool:
        """Returns True if at least one constraint is set (ignores assistant)."""
        return bool(
            self.name
            or self.area_name
            or self.floor_name
            or self.domains
            or self.device_classes
            or self.features
            or self.states
            or self.single_target
        )


@dataclass
class MatchTargetsPreferences:
    """Preferences to disambiguate duplicate name matches."""

    area_id: str | None = None
    """Id of area to use when deduplicating names."""

    floor_id: str | None = None
    """Id of floor to use when deduplicating names."""


@dataclass
class MatchTargetsCandidate:
    """Candidate for async_match_targets."""

    state: State
    is_exposed: bool
    entity: er.RegistryEntry | None = None
    area: ar.AreaEntry | None = None
    device: dr.DeviceEntry | None = None
    matched_name: str | None = None


def find_areas(name: str, areas: ar.AreaRegistry) -> Iterable[ar.AreaEntry]:
    """Find all areas matching a name (including aliases)."""
    name_norm = _normalize_name(name)
    for area in areas.async_list_areas():
        # Accept name or area id
        if (area.id == name) or (_normalize_name(area.name) == name_norm):
            yield area
            continue

        if not area.aliases:
            continue

        for alias in area.aliases:
            if _normalize_name(alias) == name_norm:
                yield area
                break


def find_floors(name: str, floors: fr.FloorRegistry) -> Iterable[fr.FloorEntry]:
    """Find all floors matching a name (including aliases)."""
    name_norm = _normalize_name(name)
    for floor in floors.async_list_floors():
        # Accept name or floor id
        if (floor.floor_id == name) or (_normalize_name(floor.name) == name_norm):
            yield floor
            continue

        if not floor.aliases:
            continue

        for alias in floor.aliases:
            if _normalize_name(alias) == name_norm:
                yield floor
                break


def _normalize_name(name: str) -> str:
    """Normalize name for comparison."""
    return name.strip().casefold()


def _filter_by_name(
    hass: HomeAssistant,
    name: str,
    candidates: Iterable[MatchTargetsCandidate],
) -> Iterable[MatchTargetsCandidate]:
    """Filter candidates by name."""
    name_norm = _normalize_name(name)

    for candidate in candidates:
        # Accept entity id
        if candidate.state.entity_id == name:
            candidate.matched_name = name
            yield candidate
            continue

        for candidate_name in async_get_entity_aliases(
            hass, candidate.entity, state=candidate.state
        ):
            if _normalize_name(candidate_name) == name_norm:
                candidate.matched_name = name
                yield candidate
                break


def _filter_by_features(
    features: int,
    candidates: Iterable[MatchTargetsCandidate],
) -> Iterable[MatchTargetsCandidate]:
    """Filter candidates by supported features."""
    for candidate in candidates:
        if (candidate.entity is not None) and (
            (candidate.entity.supported_features & features) == features
        ):
            yield candidate
            continue

        supported_features = candidate.state.attributes.get(
            EntityStateAttribute.SUPPORTED_FEATURES, 0
        )
        if (supported_features & features) == features:
            yield candidate


def _filter_by_device_classes(
    device_classes: Iterable[str],
    candidates: Iterable[MatchTargetsCandidate],
) -> Iterable[MatchTargetsCandidate]:
    """Filter candidates by device classes."""
    for candidate in candidates:
        if (
            (candidate.entity is not None)
            and candidate.entity.device_class
            and (candidate.entity.device_class in device_classes)
        ):
            yield candidate
            continue

        device_class = candidate.state.attributes.get(EntityStateAttribute.DEVICE_CLASS)
        if device_class and (device_class in device_classes):
            yield candidate


def _add_areas(
    areas: ar.AreaRegistry,
    devices: dr.DeviceRegistry,
    candidates: Iterable[MatchTargetsCandidate],
) -> None:
    """Add area and device entries to match candidates."""
    for candidate in candidates:
        if candidate.entity is None:
            continue

        if candidate.entity.device_id:
            candidate.device = devices.async_get(candidate.entity.device_id)

        if candidate.entity.area_id:
            # Use entity area first
            candidate.area = areas.async_get_area(candidate.entity.area_id)
            assert candidate.area is not None
        elif (candidate.device is not None) and candidate.device.area_id:
            # Fall back to device area
            candidate.area = areas.async_get_area(candidate.device.area_id)


def _default_area_candidate_filter(
    candidate: MatchTargetsCandidate, possible_area_ids: Collection[str]
) -> bool:
    """Keep candidates in the possible areas."""
    return (candidate.area is not None) and (candidate.area.id in possible_area_ids)


@callback
def async_match_targets(  # noqa: C901
    hass: HomeAssistant,
    constraints: MatchTargetsConstraints,
    preferences: MatchTargetsPreferences | None = None,
    states: list[State] | None = None,
    area_candidate_filter: Callable[
        [MatchTargetsCandidate, Collection[str]], bool
    ] = _default_area_candidate_filter,
) -> MatchTargetsResult:
    """Match entities based on constraints in order to handle an intent."""
    preferences = preferences or MatchTargetsPreferences()
    filtered_by_domain = False

    if not states:
        # Get all states and filter by domain
        states = hass.states.async_all(constraints.domains)
        filtered_by_domain = True
        if not states:
            return MatchTargetsResult(False, MatchFailedReason.DOMAIN)

    candidates = [
        MatchTargetsCandidate(
            state=state,
            is_exposed=(
                async_should_expose(hass, constraints.assistant, state.entity_id)
                if constraints.assistant
                else True
            ),
        )
        for state in states
    ]

    if constraints.domains and (not filtered_by_domain):
        # Filter by domain (if we didn't already do it)
        candidates = [c for c in candidates if c.state.domain in constraints.domains]
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.DOMAIN)

    if constraints.states:
        # Filter by state
        candidates = [c for c in candidates if c.state.state in constraints.states]
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.STATE)

    # Try to exit early so we can avoid registry lookups
    if not (
        constraints.name
        or constraints.features
        or constraints.device_classes
        or constraints.area_name
        or constraints.floor_name
        or constraints.single_target
    ):
        if constraints.assistant:
            # Check exposure
            candidates = [c for c in candidates if c.is_exposed]
            if not candidates:
                return MatchTargetsResult(False, MatchFailedReason.ASSISTANT)

        return MatchTargetsResult(True, states=[c.state for c in candidates])

    # We need entity registry entries now
    ent_reg = er.async_get(hass)
    for candidate in candidates:
        candidate.entity = ent_reg.async_get(candidate.state.entity_id)

    if constraints.name:
        # Filter by entity name or alias
        candidates = list(_filter_by_name(hass, constraints.name, candidates))
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.NAME)

    if constraints.features:
        # Filter by supported features
        candidates = list(_filter_by_features(constraints.features, candidates))
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.FEATURE)

    if constraints.device_classes:
        # Filter by device class
        candidates = list(
            _filter_by_device_classes(constraints.device_classes, candidates)
        )
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.DEVICE_CLASS)

    # Check floor/area constraints
    targeted_floors: list[fr.FloorEntry] | None = None
    targeted_areas: list[ar.AreaEntry] | None = None

    # True when area information has been added to candidates
    areas_added = False

    if constraints.floor_name or constraints.area_name:
        area_reg = ar.async_get(hass)
        dev_reg = dr.async_get(hass)
        _add_areas(area_reg, dev_reg, candidates)
        areas_added = True

        if constraints.floor_name:
            # Filter by areas associated with floor
            floor_reg = fr.async_get(hass)
            targeted_floors = list(find_floors(constraints.floor_name, floor_reg))
            if not targeted_floors:
                return MatchTargetsResult(
                    False,
                    MatchFailedReason.INVALID_FLOOR,
                    no_match_name=constraints.floor_name,
                )

            possible_floor_ids = {floor.floor_id for floor in targeted_floors}
            possible_area_ids = {
                area.id
                for area in area_reg.async_list_areas()
                if area.floor_id in possible_floor_ids
            }

            candidates = [
                c for c in candidates if area_candidate_filter(c, possible_area_ids)
            ]
            if not candidates:
                return MatchTargetsResult(
                    False, MatchFailedReason.FLOOR, floors=targeted_floors
                )
        else:
            # All areas are possible
            possible_area_ids = {area.id for area in area_reg.async_list_areas()}

        if constraints.area_name:
            targeted_areas = list(find_areas(constraints.area_name, area_reg))
            if not targeted_areas:
                return MatchTargetsResult(
                    False,
                    MatchFailedReason.INVALID_AREA,
                    no_match_name=constraints.area_name,
                )

            matching_area_ids = {area.id for area in targeted_areas}

            # May be constrained by floors above
            possible_area_ids.intersection_update(matching_area_ids)
            candidates = [
                c for c in candidates if area_candidate_filter(c, possible_area_ids)
            ]
            if not candidates:
                return MatchTargetsResult(
                    False, MatchFailedReason.AREA, areas=targeted_areas
                )

    if constraints.assistant:
        # Check exposure
        candidates = [c for c in candidates if c.is_exposed]
        if not candidates:
            return MatchTargetsResult(False, MatchFailedReason.ASSISTANT)

    if constraints.name and (not constraints.allow_duplicate_names):
        # Check for duplicates
        if not areas_added:
            area_reg = ar.async_get(hass)
            dev_reg = dr.async_get(hass)
            _add_areas(area_reg, dev_reg, candidates)
            areas_added = True

        sorted_candidates = sorted(
            [c for c in candidates if c.matched_name],
            key=lambda c: c.matched_name or "",
        )
        final_candidates: list[MatchTargetsCandidate] = []
        for name, group in groupby(sorted_candidates, key=lambda c: c.matched_name):
            group_candidates = list(group)
            if len(group_candidates) < 2:
                # No duplicates for name
                final_candidates.extend(group_candidates)
                continue

            # Try to disambiguate by preferences
            if preferences.floor_id:
                group_candidates = [
                    c
                    for c in group_candidates
                    if (c.area is not None)
                    and (c.area.floor_id == preferences.floor_id)
                ]
                if len(group_candidates) < 2:
                    # Disambiguated by floor
                    final_candidates.extend(group_candidates)
                    continue

            if preferences.area_id:
                group_candidates = [
                    c
                    for c in group_candidates
                    if area_candidate_filter(c, {preferences.area_id})
                ]
                if len(group_candidates) < 2:
                    # Disambiguated by area
                    final_candidates.extend(group_candidates)
                    continue

            # Couldn't disambiguate duplicate names
            return MatchTargetsResult(
                False,
                MatchFailedReason.DUPLICATE_NAME,
                no_match_name=name,
                areas=targeted_areas or [],
                floors=targeted_floors or [],
            )

        if not final_candidates:
            return MatchTargetsResult(
                False,
                MatchFailedReason.NAME,
                areas=targeted_areas or [],
                floors=targeted_floors or [],
            )

        candidates = final_candidates

    if constraints.single_target and len(candidates) > 1:
        # Find best match using preferences
        if not (preferences.area_id or preferences.floor_id):
            # No preferences
            return MatchTargetsResult(
                False,
                MatchFailedReason.MULTIPLE_TARGETS,
                states=[c.state for c in candidates],
            )

        if not areas_added:
            area_reg = ar.async_get(hass)
            dev_reg = dr.async_get(hass)
            _add_areas(area_reg, dev_reg, candidates)
            areas_added = True

        filtered_candidates: list[MatchTargetsCandidate] = candidates
        if preferences.area_id:
            # Filter by area
            filtered_candidates = [
                c for c in candidates if area_candidate_filter(c, {preferences.area_id})
            ]

        if (len(filtered_candidates) > 1) and preferences.floor_id:
            # Filter by floor
            filtered_candidates = [
                c
                for c in candidates
                if c.area and (c.area.floor_id == preferences.floor_id)
            ]

        if len(filtered_candidates) != 1:
            # Filtering could not restrict to a single target
            return MatchTargetsResult(
                False,
                MatchFailedReason.MULTIPLE_TARGETS,
                states=[c.state for c in candidates],
            )

        # Filtering succeeded
        candidates = filtered_candidates

    return MatchTargetsResult(
        True,
        None,
        states=[c.state for c in candidates],
        areas=targeted_areas or [],
        floors=targeted_floors or [],
    )


@callback
def async_match_states(
    hass: HomeAssistant,
    name: str | None = None,
    area_name: str | None = None,
    floor_name: str | None = None,
    domains: Collection[str] | None = None,
    device_classes: Collection[str] | None = None,
    states: list[State] | None = None,
    assistant: str | None = None,
) -> Iterable[State]:
    """Return states matching the constraints.

    Simplified interface to async_match_targets.
    """
    result = async_match_targets(
        hass,
        constraints=MatchTargetsConstraints(
            name=name,
            area_name=area_name,
            floor_name=floor_name,
            domains=domains,
            device_classes=device_classes,
            assistant=assistant,
        ),
        states=states,
    )
    return result.states


@callback
def async_test_feature(state: State, feature: int, feature_name: str) -> None:
    """Test if state supports a feature."""
    if state.attributes.get(EntityStateAttribute.SUPPORTED_FEATURES, 0) & feature == 0:
        raise IntentHandleError(f"Entity {state.name} does not support {feature_name}")


class IntentHandler:
    """Intent handler registration."""

    intent_type: str
    platforms: set[str] | None = None
    description: str | None = None

    @property
    def slot_schema(self) -> dict | None:
        """Return a slot schema."""
        return None

    @callback
    def async_can_handle(self, intent_obj: Intent) -> bool:
        """Test if an intent can be handled."""
        return self.platforms is None or intent_obj.platform in self.platforms

    @callback
    def async_validate_slots(self, slots: _SlotsType) -> _SlotsType:
        """Validate slot information."""
        if self.slot_schema is None:
            return slots

        return self._slot_schema(slots)  # type: ignore[no-any-return]

    @cached_property
    def _slot_schema(self) -> vol.Schema:
        """Create validation schema for slots."""
        assert self.slot_schema is not None
        return vol.Schema(
            {
                key: SLOT_SCHEMA.extend({"value": validator})
                for key, validator in self.slot_schema.items()
            },
            extra=vol.ALLOW_EXTRA,
        )

    async def async_handle(self, intent_obj: Intent) -> IntentResponse:
        """Handle the intent."""
        raise NotImplementedError

    @override
    def __repr__(self) -> str:
        """Represent a string of an intent handler."""
        return f"<{self.__class__.__name__} - {self.intent_type}>"


def non_empty_string(value: Any) -> str:
    """Coerce value to string and fail if string is empty or whitespace."""
    value_str = cv.string(value)
    if not value_str.strip():
        raise vol.Invalid("string value is empty")

    return value_str


@dataclass(kw_only=True)
class IntentSlotInfo:
    """Details about how intent slots are processed and validated."""

    service_data_name: str | None = None
    """Optional name of the service data input to map to this slot."""

    description: str | None = None
    """Human readable description of the slot."""

    value_schema: VolSchemaType | Callable[[Any], Any] = vol.Any
    """Validator for the slot."""


def _convert_slot_info(
    key: str | tuple[str, str],
    value: IntentSlotInfo | VolSchemaType | Callable[[Any], Any],
) -> tuple[str, IntentSlotInfo]:
    """Create an IntentSlotInfo from the various supported input arguments."""
    if isinstance(value, IntentSlotInfo):
        if not isinstance(key, str):
            raise TypeError("Tuple key and IntentSlotDescription value not supported")
        return key, value
    if isinstance(key, tuple):
        return key[0], IntentSlotInfo(service_data_name=key[1], value_schema=value)
    return key, IntentSlotInfo(value_schema=value)


class DynamicServiceIntentHandler(IntentHandler):
    """Service Intent handler registration (dynamic).

    Service specific intent handler that calls a service by name/entity_id.
    """

    # We use a small timeout in service calls to (hopefully) pass validation
    # checks, but not try to wait for the call to fully complete.
    service_timeout: float = 0.2

    def __init__(
        self,
        intent_type: str,
        *,
        required_slots: _IntentSlotsType | None = None,
        optional_slots: _IntentSlotsType | None = None,
        required_domains: set[str] | None = None,
        required_features: int | None = None,
        required_states: set[str] | None = None,
        description: str | None = None,
        platforms: set[str] | None = None,
        device_classes: set[type[StrEnum]] | None = None,
    ) -> None:
        """Create Service Intent Handler."""
        self.intent_type = intent_type
        self.required_domains = required_domains
        self.required_features = required_features
        self.required_states = required_states
        self.description = description
        self.platforms = platforms
        self.device_classes = device_classes

        self.required_slots: dict[str, IntentSlotInfo] = dict(
            _convert_slot_info(key, value)
            for key, value in (required_slots or {}).items()
        )
        self.optional_slots: dict[str, IntentSlotInfo] = dict(
            _convert_slot_info(key, value)
            for key, value in (optional_slots or {}).items()
        )

    @cached_property
    @override
    def slot_schema(self) -> dict:
        """Return a slot schema."""
        domain_validator = (
            vol.In(list(self.required_domains)) if self.required_domains else cv.string
        )
        slot_schema = {
            vol.Any("name", "area", "floor"): non_empty_string,
            vol.Optional("domain"): vol.All(cv.ensure_list, [domain_validator]),
        }
        if self.device_classes:
            # The typical way to match enums is with vol.Coerce, but we build a
            # flat list to make the API simpler to describe programmatically
            flattened_device_classes = vol.In(
                [
                    device_class.value
                    for device_class_enum in self.device_classes
                    for device_class in device_class_enum
                ]
            )
            slot_schema.update(
                {
                    vol.Optional("device_class"): vol.All(
                        cv.ensure_list,
                        [flattened_device_classes],
                    )
                }
            )

        slot_schema.update(
            {
                vol.Optional("preferred_area_id"): cv.string,
                vol.Optional("preferred_floor_id"): cv.string,
            }
        )

        if self.required_slots:
            slot_schema.update(
                {
                    vol.Required(
                        key, description=slot_info.description
                    ): slot_info.value_schema
                    for key, slot_info in self.required_slots.items()
                }
            )

        if self.optional_slots:
            slot_schema.update(
                {
                    vol.Optional(
                        key, description=slot_info.description
                    ): slot_info.value_schema
                    for key, slot_info in self.optional_slots.items()
                }
            )

        return slot_schema

    @abstractmethod
    def get_domain_and_service(
        self, intent_obj: Intent, state: State
    ) -> tuple[str, str]:
        """Get the domain and service name to call."""
        raise NotImplementedError

    @override
    async def async_handle(self, intent_obj: Intent) -> IntentResponse:
        """Handle the hass intent."""
        hass = intent_obj.hass
        slots = self.async_validate_slots(intent_obj.slots)

        name_slot = slots.get("name", {})
        entity_name: str | None = name_slot.get("value")
        entity_text: str | None = name_slot.get("text")
        if entity_name == "all":
            # Don't match on name if targeting all entities
            entity_name = None

        # Get area/floor info
        area_slot = slots.get("area", {})
        area_id = area_slot.get("value")

        floor_slot = slots.get("floor", {})
        floor_id = floor_slot.get("value")

        # Optional domain/device class filters.
        # Convert to sets for speed.
        domains: set[str] | None = self.required_domains
        device_classes: set[str] | None = None

        if "domain" in slots:
            domains = set(slots["domain"]["value"])

        if "device_class" in slots:
            device_classes = set(slots["device_class"]["value"])

        match_constraints = MatchTargetsConstraints(
            name=entity_name,
            area_name=area_id,
            floor_name=floor_id,
            domains=domains,
            device_classes=device_classes,
            assistant=intent_obj.assistant,
            features=self.required_features,
            states=self.required_states,
        )
        if not match_constraints.has_constraints:
            # Fail if attempting to target all devices in the house
            raise IntentHandleError("Service handler cannot target all devices")

        match_preferences = MatchTargetsPreferences(
            area_id=slots.get("preferred_area_id", {}).get("value"),
            floor_id=slots.get("preferred_floor_id", {}).get("value"),
        )

        match_result = async_match_targets(hass, match_constraints, match_preferences)
        if not match_result.is_match:
            raise MatchFailedError(
                result=match_result,
                constraints=match_constraints,
                preferences=match_preferences,
            )

        # Ensure name is text
        if ("name" in slots) and entity_text:
            slots["name"]["value"] = entity_text

        # Replace area/floor values with the resolved ids for use in templates
        if ("area" in slots) and match_result.areas:
            slots["area"]["value"] = match_result.areas[0].id

        if ("floor" in slots) and match_result.floors:
            slots["floor"]["value"] = match_result.floors[0].floor_id

        # Update intent slots to include any transformations done by the schemas
        intent_obj.slots = slots

        return await self.async_handle_states(
            intent_obj, match_result, match_constraints, match_preferences
        )

    async def async_handle_states(
        self,
        intent_obj: Intent,
        match_result: MatchTargetsResult,
        match_constraints: MatchTargetsConstraints,
        match_preferences: MatchTargetsPreferences | None = None,
    ) -> IntentResponse:
        """Complete action on matched entity states."""
        states = match_result.states
        response = intent_obj.create_response()

        hass = intent_obj.hass
        success_results: list[IntentResponseTarget] = []

        if match_result.floors:
            success_results.extend(
                IntentResponseTarget(
                    type=IntentResponseTargetType.FLOOR,
                    name=floor.name,
                    id=floor.floor_id,
                )
                for floor in match_result.floors
            )
        elif match_result.areas:
            success_results.extend(
                IntentResponseTarget(
                    type=IntentResponseTargetType.AREA, name=area.name, id=area.id
                )
                for area in match_result.areas
            )

        service_coros: list[Coroutine[Any, Any, None]] = []
        for state in states:
            domain, service = self.get_domain_and_service(intent_obj, state)
            service_coros.append(
                self.async_call_service(domain, service, intent_obj, state)
            )

        # Handle service calls in parallel, noting failures as they occur.
        failed_results: list[IntentResponseTarget] = []
        for state, service_coro in zip(
            states, asyncio.as_completed(service_coros), strict=False
        ):
            target = IntentResponseTarget(
                type=IntentResponseTargetType.ENTITY,
                name=state.name,
                id=state.entity_id,
            )

            try:
                await service_coro
                success_results.append(target)
            except Exception:
                failed_results.append(target)
                _LOGGER.exception("Service call failed for %s", state.entity_id)

        if not success_results:
            # If no entities succeeded, raise an error.
            failed_entity_ids = [target.id for target in failed_results]
            raise IntentHandleError(
                f"Failed to call {service} for: {failed_entity_ids}"
            )

        response.async_set_results(
            success_results=success_results, failed_results=failed_results
        )

        # Update all states
        states = [hass.states.get(state.entity_id) or state for state in states]
        response.async_set_states(states)

        return response

    async def async_call_service(
        self, domain: str, service: str, intent_obj: Intent, state: State
    ) -> None:
        """Call service on entity."""
        hass = intent_obj.hass

        service_data: dict[str, Any] = {ATTR_ENTITY_ID: state.entity_id}
        if self.required_slots:
            for key, slot_info in self.required_slots.items():
                service_data[slot_info.service_data_name or key] = intent_obj.slots[
                    key
                ]["value"]

        if self.optional_slots:
            for key, slot_info in self.optional_slots.items():
                if value := intent_obj.slots.get(key):
                    service_data[slot_info.service_data_name or key] = value["value"]

        await self._run_then_background(
            hass.async_create_task_internal(
                hass.services.async_call(
                    domain,
                    service,
                    service_data,
                    context=intent_obj.context,
                    blocking=True,
                ),
                f"intent_call_service_{domain}_{service}",
            )
        )

    async def _run_then_background(self, task: asyncio.Task[Any]) -> None:
        """Run task with timeout to (hopefully) catch validation errors.

        After the timeout the task will continue to run in the background.
        """
        done, _ = await asyncio.wait({task}, timeout=self.service_timeout)
        if done:
            # Task finished within the timeout. Re-raise any exception
            # (e.g. validation errors) so the caller can handle it.
            task.result()


class ServiceIntentHandler(DynamicServiceIntentHandler):
    """Service Intent handler registration.

    Service specific intent handler that calls a service by name/entity_id.
    """

    def __init__(
        self,
        intent_type: str,
        domain: str,
        service: str,
        *,
        required_slots: _IntentSlotsType | None = None,
        optional_slots: _IntentSlotsType | None = None,
        required_domains: set[str] | None = None,
        required_features: int | None = None,
        required_states: set[str] | None = None,
        description: str | None = None,
        platforms: set[str] | None = None,
        device_classes: set[type[StrEnum]] | None = None,
    ) -> None:
        """Create service handler."""
        super().__init__(
            intent_type,
            required_slots=required_slots,
            optional_slots=optional_slots,
            required_domains=required_domains,
            required_features=required_features,
            required_states=required_states,
            description=description,
            platforms=platforms,
            device_classes=device_classes,
        )
        self.domain = domain
        self.service = service

    @override
    def get_domain_and_service(
        self, intent_obj: Intent, state: State
    ) -> tuple[str, str]:
        """Get the domain and service name to call."""
        return (self.domain, self.service)


class Intent:
    """Hold the intent."""

    __slots__ = [
        "assistant",
        "context",
        "conversation_agent_id",
        "device_id",
        "hass",
        "intent_type",
        "language",
        "platform",
        "satellite_id",
        "slots",
        "text_input",
    ]

    def __init__(
        self,
        hass: HomeAssistant,
        platform: str,
        intent_type: str,
        slots: _SlotsType,
        text_input: str | None,
        context: Context,
        language: str,
        assistant: str | None = None,
        device_id: str | None = None,
        satellite_id: str | None = None,
        conversation_agent_id: str | None = None,
    ) -> None:
        """Initialize an intent."""
        self.hass = hass
        self.platform = platform
        self.intent_type = intent_type
        self.slots = slots
        self.text_input = text_input
        self.context = context
        self.language = language
        self.assistant = assistant
        self.device_id = device_id
        self.satellite_id = satellite_id
        self.conversation_agent_id = conversation_agent_id

    @callback
    def create_response(self) -> IntentResponse:
        """Create a response."""
        return IntentResponse(language=self.language, intent=self)


class IntentResponseType(
    Enum,
    metaclass=EnumWithDeprecatedMembers,
    deprecated={
        "PARTIAL_ACTION_DONE": (
            "IntentResponseType.ACTION_DONE or IntentResponseType.ERROR",
            "2026.3.0",
        ),
    },
):
    """Type of the intent response."""

    ACTION_DONE = "action_done"
    """Intent caused an action to occur"""

    PARTIAL_ACTION_DONE = "partial_action_done"
    """Deprecated. Intent caused an action, but it could only be partially done"""

    QUERY_ANSWER = "query_answer"
    """Response is an answer to a query"""

    ERROR = "error"
    """Response is an error"""


class IntentResponseErrorCode(StrEnum):
    """Reason for an intent response error."""

    NO_INTENT_MATCH = "no_intent_match"
    """Text could not be matched to an intent"""

    NO_VALID_TARGETS = "no_valid_targets"
    """Intent was matched, but no valid areas/devices/entities were targeted"""

    FAILED_TO_HANDLE = "failed_to_handle"
    """Unexpected error occurred while handling intent"""

    UNKNOWN = "unknown"
    """Error outside the scope of intent processing"""


class IntentResponseTargetType(StrEnum):
    """Type of target for an intent response."""

    AREA = "area"
    FLOOR = "floor"
    DEVICE = "device"
    ENTITY = "entity"
    DOMAIN = "domain"
    DEVICE_CLASS = "device_class"
    CUSTOM = "custom"


@dataclass(slots=True)
class IntentResponseTarget:
    """Target of the intent response."""

    name: str
    type: IntentResponseTargetType
    id: str | None = None


class IntentResponse:
    """Response to an intent."""

    def __init__(
        self,
        language: str,
        intent: Intent | None = None,
    ) -> None:
        """Initialize an IntentResponse."""
        self.language = language
        self.intent = intent
        self.speech: dict[str, dict[str, Any]] = {}
        self.reprompt: dict[str, dict[str, Any]] = {}
        self.card: dict[str, dict[str, str]] = {}
        self.error_code: IntentResponseErrorCode | None = None
        self.success_results: list[IntentResponseTarget] = []
        self.failed_results: list[IntentResponseTarget] = []
        self.matched_states: list[State] = []
        self.unmatched_states: list[State] = []
        self.speech_slots: dict[str, Any] = {}
        self.response_type = IntentResponseType.ACTION_DONE

    @callback
    def async_set_speech(
        self,
        speech: str,
        speech_type: str = "plain",
        extra_data: Any | None = None,
    ) -> None:
        """Set speech response."""
        self.speech[speech_type] = {
            "speech": speech,
            "extra_data": extra_data,
        }

    @callback
    def async_set_reprompt(
        self,
        speech: str,
        speech_type: str = "plain",
        extra_data: Any | None = None,
    ) -> None:
        """Set reprompt response."""
        self.reprompt[speech_type] = {
            "reprompt": speech,
            "extra_data": extra_data,
        }

    @callback
    def async_set_card(
        self, title: str, content: str, card_type: str = "simple"
    ) -> None:
        """Set card response."""
        self.card[card_type] = {"title": title, "content": content}

    @callback
    def async_set_error(self, code: IntentResponseErrorCode, message: str) -> None:
        """Set response error."""
        self.response_type = IntentResponseType.ERROR
        self.error_code = code

        # Speak error message
        self.async_set_speech(message)

    @callback
    def async_set_results(
        self,
        success_results: list[IntentResponseTarget],
        failed_results: list[IntentResponseTarget] | None = None,
    ) -> None:
        """Set response results."""
        self.success_results = success_results
        self.failed_results = failed_results if failed_results is not None else []

    @callback
    def async_set_states(
        self, matched_states: list[State], unmatched_states: list[State] | None = None
    ) -> None:
        """Set matched/unmatched entity states during intent handling."""
        self.matched_states = matched_states
        self.unmatched_states = unmatched_states or []

    @callback
    def async_set_speech_slots(self, speech_slots: dict[str, Any]) -> None:
        """Set slots that will be used in the response template of the default agent."""
        self.speech_slots = speech_slots

    @callback
    def as_dict(self) -> dict[str, Any]:
        """Return a dictionary representation of an intent response."""
        response_dict: dict[str, Any] = {
            "speech": {k: dict(v) for k, v in self.speech.items()},
            "card": {k: dict(v) for k, v in self.card.items()},
            "language": self.language,
            "response_type": self.response_type.value,
        }

        if self.reprompt:
            response_dict["reprompt"] = {k: dict(v) for k, v in self.reprompt.items()}
        if self.speech_slots:
            response_dict["speech_slots"] = self.speech_slots.copy()

        response_data: dict[str, Any] = {}

        if self.response_type is IntentResponseType.ERROR:
            assert self.error_code is not None, "error code is required"
            response_data["code"] = self.error_code.value
        else:
            # action done or query answer
            response_data["success"] = [
                dataclasses.asdict(target) for target in self.success_results
            ]

            response_data["failed"] = [
                dataclasses.asdict(target) for target in self.failed_results
            ]

        response_dict["data"] = response_data

        return response_dict


@callback
def async_get_entity_aliases(
    hass: HomeAssistant,
    entity_entry: er.RegistryEntry | None,
    *,
    state: State,
    allow_empty: bool = True,
) -> list[str]:
    """Get all names/aliases for an entity.

    If no entity registry entry is provided, returns a list with just the
    state name. Otherwise, delegates to the entity registry to resolve aliases,
    where COMPUTED_NAME aliases are replaced with the computed full entity name.

    The returned list preserves the order set by the user.
    """
    if entity_entry is None:
        return [state.name.strip()]

    return er.async_get_entity_aliases(hass, entity_entry, allow_empty=allow_empty)
