HomeAssistantVS/custom_components/spook/services.py

"""Spook - Your homie."""

from __future__ import annotations

from abc import ABC, abstractmethod
import asyncio
from dataclasses import dataclass, field
import importlib
from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, TypeVar, final

import voluptuous as vol

from homeassistant.const import EVENT_COMPONENT_LOADED, EVENT_CORE_CONFIG_UPDATE
from homeassistant.core import (
    Event,
    HomeAssistant,
    Service,
    ServiceCall,
    ServiceResponse,
    SupportsResponse,
    callback,
)
from homeassistant.exceptions import Unauthorized, UnknownUser
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import DATA_INSTANCES, EntityComponent
from homeassistant.helpers.entity_platform import DATA_ENTITY_PLATFORM
from homeassistant.helpers.service import (
    SERVICE_DESCRIPTION_CACHE,
    async_register_admin_service,
    async_set_service_schema,
)
from homeassistant.helpers.translation import (
    _async_get_translations_cache,
    async_get_cached_translations,
    async_get_translations,
)
from homeassistant.loader import async_get_integration
from homeassistant.setup import ATTR_COMPONENT

from .const import DOMAIN, LOGGER
from .core_compat import load_service_descriptions
from .service_icons import async_inject_service_icons, async_remove_service_icons

if TYPE_CHECKING:
    from collections.abc import Callable, Iterable
    from types import ModuleType


_EntityT = TypeVar("_EntityT", bound=Entity, default=Entity)
GHOST = "👻"
SERVICE_TRANSLATION_CATEGORY = "services"
SELECTOR_TRANSLATION_CATEGORY = "selector"


class AbstractSpookServiceBase(ABC):
    """Abstract base class to hold a Spook service."""

    hass: HomeAssistant
    domain: str
    service: str
    schema: dict[str | vol.Marker, Any] | None = None

    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize the service."""
        self.hass = hass

    @abstractmethod
    @callback
    def async_register(self) -> bool:
        """Register the service with Home Assistant.

        Returns True when the service was actually registered.
        """
        raise NotImplementedError

    @final
    @callback
    def async_unregister(self) -> None:
        """Unregister the service from Home Assistant."""
        LOGGER.debug(
            "Unregistering Spook service: %s.%s",
            self.domain,
            self.service,
        )

        self.hass.services.async_remove(self.domain, self.service)


class ReplaceExistingService(AbstractSpookServiceBase):
    """Service replaces/may replace an existing service."""

    overriden_service: Service | None = None


class AbstractSpookService(AbstractSpookServiceBase):
    """Abstract class to hold a Spook service."""

    supports_response: SupportsResponse = SupportsResponse.NONE

    @final
    @callback
    def async_register(self) -> bool:
        """Register the service with Home Assistant."""
        # Only register the service if the domain is the spook integration
        # or if the target integration is loaded.
        if self.domain != DOMAIN and self.domain not in self.hass.config.components:
            LOGGER.debug(
                "Not registering Spook %s.%s service, %s is not loaded",
                self.domain,
                self.service,
                self.domain,
            )
            return False

        LOGGER.debug(
            "Registering Spook service: %s.%s",
            self.domain,
            self.service,
        )

        self.hass.services.async_register(
            domain=self.domain,
            service=self.service,
            service_func=self.async_handle_service,
            schema=vol.Schema(self.schema) if self.schema else None,
            supports_response=self.supports_response,
        )

        return True

    @abstractmethod
    async def async_handle_service(self, call: ServiceCall) -> ServiceResponse:
        """Handle the service call."""
        raise NotImplementedError


class AbstractSpookAdminService(AbstractSpookServiceBase):
    """Abstract class to hold a Spook admin service."""

    supports_response: SupportsResponse = SupportsResponse.NONE

    @final
    @callback
    def async_register(self) -> bool:
        """Register the service with Home Assistant."""
        if self.domain != DOMAIN and self.domain not in self.hass.config.components:
            LOGGER.debug(
                "Not registering Spook %s.%s admin service, %s is not loaded",
                self.domain,
                self.service,
                self.domain,
            )
            return False

        LOGGER.debug(
            "Registering Spook admin service: %s.%s",
            self.domain,
            self.service,
        )
        async_register_admin_service(
            hass=self.hass,
            domain=self.domain,
            service=self.service,
            service_func=self.async_handle_service,
            schema=vol.Schema(self.schema) if self.schema else None,
            supports_response=self.supports_response,
        )

        return True

    @abstractmethod
    async def async_handle_service(self, call: ServiceCall) -> None:
        """Handle the service call."""
        raise NotImplementedError


class AbstractSpookEntityService(AbstractSpookServiceBase, Generic[_EntityT]):
    """Abstract class to hold a Spook entity service."""

    platform: str
    required_features: list[int] | None = None
    supports_response: SupportsResponse = SupportsResponse.NONE

    @final
    @callback
    def async_register(self) -> bool:
        """Register the service with Home Assistant."""
        LOGGER.debug(
            "Registering Spook entity service: %s.%s for platform %s",
            self.domain,
            self.service,
            self.platform,
        )

        if not (
            platform := next(
                (
                    platform
                    for platform in self.hass.data.get(DATA_ENTITY_PLATFORM, {}).get(
                        self.domain, []
                    )
                    if platform.domain == self.platform
                ),
                None,
            )
        ):
            msg = (
                f"Could not find platform {self.platform} for domain "
                f"{self.domain} to register service: "
                f"{self.domain}.{self.service}"
            )
            raise RuntimeError(msg)

        platform.async_register_entity_service(
            name=self.service,
            func=self.async_handle_service,
            schema=self.schema,
            required_features=self.required_features,
            supports_response=self.supports_response,
        )

        return True

    @abstractmethod
    async def async_handle_service(
        self,
        entity: _EntityT,
        call: ServiceCall,
    ) -> ServiceResponse:
        """Handle the service call."""
        raise NotImplementedError


class AbstractSpookEntityComponentService(AbstractSpookServiceBase, Generic[_EntityT]):
    """Abstract class to hold a Spook entity component service."""

    required_features: list[int] | None = None
    supports_response: SupportsResponse = SupportsResponse.NONE
    #: For an action that changes how something is set up rather than what it
    #: is doing, which Home Assistant keeps to admins. Automations still pass.
    admin_only: bool = False

    @final
    @callback
    def async_register(self) -> bool:
        """Register the service with Home Assistant."""
        LOGGER.debug(
            "Registering Spook entity component service: %s.%s",
            self.domain,
            self.service,
        )

        # Not every component is there when Spook is. Home Assistant loads
        # calendar and todo only once an integration brings a calendar or a
        # to-do list along, which can be well after Spook, or never. The
        # manager waits for it and registers the action then.
        if self.domain not in self.hass.data.get(DATA_INSTANCES, {}):
            LOGGER.debug(
                "Not registering Spook %s.%s service yet, %s is not loaded",
                self.domain,
                self.service,
                self.domain,
            )
            return False

        component: EntityComponent[Entity] = self.hass.data[DATA_INSTANCES][self.domain]

        component.async_register_entity_service(
            name=self.service,
            func=(
                self._async_handle_service_as_admin
                if self.admin_only
                else self.async_handle_service
            ),
            schema=self.schema,
            required_features=self.required_features,
            supports_response=self.supports_response,
        )

        return True

    async def _async_handle_service_as_admin(
        self,
        entity: _EntityT,
        call: ServiceCall,
    ) -> ServiceResponse:
        """Handle the call, if whoever made it may change how things are set up.

        The same check Home Assistant does for an admin-only entity action. It
        is done here rather than asked of Home Assistant, because the option
        to ask arrived in a later version than the oldest one Spook runs on,
        and passing it there fails registering every one of these actions.
        """
        if call.context.user_id:
            user = await self.hass.auth.async_get_user(call.context.user_id)
            if user is None:
                raise UnknownUser(context=call.context)
            if not user.is_admin:
                raise Unauthorized(context=call.context)

        return await self.async_handle_service(entity, call)

    @abstractmethod
    async def async_handle_service(
        self,
        entity: _EntityT,
        call: ServiceCall,
    ) -> ServiceResponse:
        """Handle the service call."""
        raise NotImplementedError


# Most of these track something it put into Home Assistant, to undo on unload.
@dataclass
class SpookServiceManager:  # pylint: disable=too-many-instance-attributes
    """Class to manage Spook services."""

    hass: HomeAssistant

    _services: set[AbstractSpookService] = field(default_factory=set)
    _service_schemas: dict[str, Any] = field(default_factory=dict)
    _service_translation_overrides: dict[tuple[str, str, str], str | None] = field(
        default_factory=dict
    )
    # The same, for the option labels of the selectors those actions use.
    _selector_translation_overrides: dict[tuple[str, str, str], str | None] = field(
        default_factory=dict
    )
    # The icons Spook gave its actions on other domains, as (domain, action).
    _injected_service_icons: set[tuple[str, str]] = field(default_factory=set)
    # Services for a domain that was not loaded yet, by that domain.
    _waiting_for_domain: dict[str, list[AbstractSpookService]] = field(
        default_factory=dict
    )
    # Everything to undo on unload: the listeners, and any translation
    # injection still on its way.
    _on_unload: list[Callable[[], None]] = field(default_factory=list)
    # The languages Spook's strings went into. Not just the server's: every
    # person picks their own in their profile, and the frontend asks for that
    # one. #1820.
    _languages: set[str] = field(default_factory=set)
    # A language whose injection is on its way, so a second request for it
    # waits for that one instead of reading the cache halfway.
    _language_tasks: dict[str, asyncio.Task[None]] = field(default_factory=dict)
    _starting_languages: set[str] = field(default_factory=set)
    # Whether Home Assistant's translation loading is still being followed.
    # Turned off the moment unloading starts, for a load already under way.
    follows_translation_loads: bool = False

    def __post_init__(self) -> None:
        """Post initialization."""
        LOGGER.debug("Spook service manager initialized")

    async def async_setup(self) -> None:
        """Set up the Spook services."""
        LOGGER.debug("Setting up Spook services")

        # Load service schemas
        integration = await async_get_integration(self.hass, DOMAIN)
        self._service_schemas = await self.hass.async_add_executor_job(
            load_service_descriptions,
            integration,
        )

        modules: list[ModuleType] = []

        def _load_all_service_modules() -> None:
            """Load all service modules."""
            for module_file in Path(__file__).parent.rglob(
                "ectoplasms/*/services/*.py"
            ):
                if module_file.name == "__init__.py":
                    continue
                module_path = str(module_file.relative_to(Path(__file__).parent))[
                    :-3
                ].replace("/", ".")
                modules.append(importlib.import_module(f".{module_path}", __package__))

        await self.hass.async_add_import_executor_job(_load_all_service_modules)

        # Listening starts before anything is parked, and nothing between here
        # and the end of the loop below waits on anything. A domain that loads
        # in the meantime is either there for the loop to register straight
        # away, or loads afterwards with the listener already in place.
        self._on_unload = [
            self.hass.bus.async_listen(
                EVENT_COMPONENT_LOADED,
                self._async_component_loaded,
            ),
            self.hass.bus.async_listen(
                EVENT_CORE_CONFIG_UPDATE,
                self._async_core_config_updated,
            ),
        ]

        for module in modules:
            self._async_setup_service_module(module)

        self._async_follow_translation_loads()
        await self.async_inject_service_translations()

    @callback
    def _async_follow_translation_loads(self) -> None:
        """Inject into every language Home Assistant loads, as it loads it.

        Home Assistant loads a language the moment somebody asks for it, and
        says nothing when it does. Spook's strings for a domain that is not
        its own have to be written into that language by hand, so it follows
        the loading on this one cache: once a language is in for the first
        time, Spook's strings go in too, before the caller reads them.

        The cache keeps its attributes in slots, so its method cannot be
        swapped on the instance. Its class can, for a subclass that adds no
        slots of its own, and only this Home Assistant's cache is touched.
        """
        translations_cache = _async_get_translations_cache(self.hass)
        original_class = type(translations_cache)
        if not callable(getattr(original_class, "async_load", None)):
            LOGGER.warning(
                "Unable to follow Home Assistant's translation loading, "
                "Spook's actions are only translated in the server's language"
            )
            return

        manager = self

        class _FollowedTranslationCache(original_class):  # type: ignore[misc,valid-type]
            """Home Assistant's translation cache, with Spook along."""

            __slots__ = ()

            async def async_load(self, language: str, components: set[str]) -> None:
                """Load as ever, then add Spook's strings to a new language."""
                await super().async_load(language, components)
                # Unloading can have happened while that waited on the lock.
                if manager.follows_translation_loads:
                    await manager.async_follow_language(language)

        try:
            translations_cache.__class__ = _FollowedTranslationCache
        except TypeError:
            LOGGER.warning(
                "Unable to follow Home Assistant's translation loading, "
                "Spook's actions are only translated in the server's language"
            )
            return

        self.follows_translation_loads = True

        @callback
        def _stop_following() -> None:
            self.follows_translation_loads = False
            # Only when nothing swapped it again since.
            if translations_cache.__class__ is _FollowedTranslationCache:
                translations_cache.__class__ = original_class

        self._on_unload.append(_stop_following)

    async def async_follow_language(self, language: str) -> None:
        """Translate Spook's actions into a language loaded for the first time.

        One injection per language, which every request for that language
        waits for. Only one that went through marks the language as done; a
        failed or cancelled one is tried again by the next request.
        """
        if language in self._languages or language in self._starting_languages:
            return

        task = self._language_tasks.get(language)
        if task is None or task.done():
            # A finished one is still listed until its callback has run; one
            # that failed is not a reason to stop trying.
            task = self._async_start_language_task(language)
        elif task is asyncio.current_task():
            # The injection itself loading this language again.
            return

        try:
            # Shielded: one caller being cancelled must not cancel the
            # injection the others are waiting for.
            await asyncio.shield(task)
        except asyncio.CancelledError:
            # Only swallowed when unloading cancelled Spook's part; the caller
            # being cancelled itself goes on as it should.
            current = asyncio.current_task()
            if current is not None and current.cancelling():
                raise
        # pylint: disable-next=broad-exception-caught
        except Exception:  # noqa: BLE001, S110
            # Logged once, where the task finishes. The caller only asked for
            # translations, and Spook's part going wrong must not take theirs
            # with it.
            pass

    @callback
    def _async_start_language_task(self, language: str) -> asyncio.Task[None]:
        """Start injecting into a language, in a task unloading cancels."""
        # Home Assistant starts a task eagerly: the injection runs, and loads
        # this language again, before the task is handed back to be listed.
        self._starting_languages.add(language)
        try:
            task = self.hass.async_create_task(
                self.async_inject_service_translations([language]),
                f"Inject Spook service translations in {language}",
            )
        finally:
            self._starting_languages.discard(language)
        self._language_tasks[language] = task
        self._on_unload.append(task.cancel)

        @callback
        def _finished(_task: asyncio.Task[None]) -> None:
            self._language_tasks.pop(language, None)
            if task.cancel in self._on_unload:
                self._on_unload.remove(task.cancel)
            if task.cancelled():
                return
            if (err := task.exception()) is not None:
                LOGGER.error(
                    "Spook could not translate its actions into %s: %s",
                    language,
                    err,
                )
                return
            self._languages.add(language)

        task.add_done_callback(_finished)
        return task

    @callback
    def _async_setup_service_module(self, module: ModuleType) -> None:
        """Set up a single service module, isolating failures.

        A service that fails to set up must not prevent the rest of Spook
        from loading.
        """
        try:
            service = module.SpookService(self.hass)
        # pylint: disable-next=broad-exception-caught
        except Exception:  # noqa: BLE001
            LOGGER.exception(
                "Spook service %s failed to set up and has been skipped; "
                "please report this issue at "
                "https://github.com/frenck/spook/issues",
                module.__name__,
            )
            return

        self._async_setup_service(service, module.__name__)

    @callback
    def _async_setup_service(self, service: AbstractSpookService, name: str) -> None:
        """Register one service, isolating failures.

        One for a domain that is not loaded yet waits for it, and is set up
        through here again once it is.
        """
        try:
            if isinstance(
                service,
                ReplaceExistingService,
            ) and self.hass.services.has_service(service.domain, service.service):
                LOGGER.debug(
                    "Unregistering service that will be overriden service: %s.%s",
                    service.domain,
                    service.service,
                )
                # pylint: disable=protected-access
                service.overriden_service = (
                    self.hass.services._services[service.domain]  # noqa: SLF001
                ).pop(service.service)

            if not self.async_register_service(service):
                self._waiting_for_domain.setdefault(service.domain, []).append(service)
        # pylint: disable-next=broad-exception-caught
        except Exception:  # noqa: BLE001
            # If the service this one overrides was already unregistered,
            # restore it; a failing setup must not silently remove a core
            # service until the next restart.
            if (
                isinstance(service, ReplaceExistingService)
                and service.overriden_service is not None
            ):
                # pylint: disable-next=protected-access
                self.hass.services._services.setdefault(  # noqa: SLF001
                    service.domain,
                    {},
                )[service.service] = service.overriden_service
            LOGGER.exception(
                "Spook service %s failed to set up and has been skipped; "
                "please report this issue at "
                "https://github.com/frenck/spook/issues",
                name,
            )

    @callback
    def _async_component_loaded(self, event: Event) -> None:
        """Register the services that were waiting for this domain."""
        if not (
            waiting := self._waiting_for_domain.pop(event.data[ATTR_COMPONENT], [])
        ):
            return

        for service in waiting:
            self._async_setup_service(service, type(service).__module__)

        # The descriptions went in with the registration, the translations and
        # icons did not: those are injected for every registered service in one
        # go.
        self._async_reinject_service_translations()

    @callback
    def _async_reinject_service_translations(
        self, languages: Iterable[str] | None = None
    ) -> None:
        """Inject the translations again, in a task unloading cancels.

        Injecting waits on loading translations before it writes anything.
        Left running through an unload, it would write Spook's strings back
        for actions that were just taken away, after unload put the originals
        back.
        """
        task = self.hass.async_create_task(
            self.async_inject_service_translations(languages),
            "Inject Spook service translations",
        )
        self._on_unload.append(task.cancel)

        @callback
        def _finished(_task: asyncio.Task[None]) -> None:
            # Already gone when unloading cleared the list and cancelled it.
            if task.cancel in self._on_unload:
                self._on_unload.remove(task.cancel)

        task.add_done_callback(_finished)

    @callback
    def async_register_service(self, service: AbstractSpookService) -> bool:
        """Register a Spook service.

        Returns False when the domain it belongs to is not loaded (yet).
        """
        # A service aimed at an integration that is not set up never lands in
        # Home Assistant. Injecting a description for it would then describe
        # an action that does not exist, which core refuses with a KeyError.
        if not service.async_register():
            return False

        self._services.add(service)

        # Override service description with Spook's if the service is not
        # for the Spook integration.
        if service.domain != DOMAIN and (
            service_schema := self._service_schemas.get(
                f"{service.domain}_{service.service}",
            )
        ):
            LOGGER.debug(
                "Injecting Spook service schema for: %s.%s",
                service.domain,
                service.service,
            )
            async_set_service_schema(
                self.hass,
                domain=service.domain,
                service=service.service,
                schema=service_schema,
            )

        return True

    @callback
    def _service_schema_key(self, service: AbstractSpookService) -> str:
        """Return the services.yaml key for a Spook service."""
        if service.domain == DOMAIN:
            return service.service
        return f"{service.domain}_{service.service}"

    @callback
    def _service_translation_strings(
        self,
        service: AbstractSpookService,
        cached_spook_translations: dict[str, str],
    ) -> dict[str, str]:
        """Return service translation strings mapped to the target domain."""
        schema_key = self._service_schema_key(service)
        spook_prefix = f"component.{DOMAIN}.services.{schema_key}."
        target_prefix = f"component.{service.domain}.services.{service.service}."

        return {
            f"{target_prefix}{key.removeprefix(spook_prefix)}": (
                f"{value} {GHOST}"
                if key == f"{spook_prefix}name" and GHOST not in value
                else value
            )
            for key, value in cached_spook_translations.items()
            if key.startswith(spook_prefix)
        }

    @callback
    def _translation_component_cache(
        self,
        language: str,
        domain: str,
        *,
        create: bool = False,
        category: str = SERVICE_TRANSLATION_CATEGORY,
    ) -> dict[str, str] | None:
        """Return the Home Assistant translation cache for a component."""
        translations_cache = _async_get_translations_cache(self.hass)

        try:
            cache = translations_cache.cache_data.cache
        except AttributeError:
            LOGGER.warning(
                "Unable to access Home Assistant's translation cache, "
                "skipping Spook service translation update"
            )
            return None

        if not isinstance(cache, dict):
            LOGGER.warning(
                "Home Assistant's translation cache has an unexpected structure, "
                "skipping Spook service translation update"
            )
            return None

        if create:
            return (
                cache.setdefault(language, {})
                .setdefault(category, {})
                .setdefault(domain, {})
            )

        return cache.get(language, {}).get(category, {}).get(domain)

    @callback
    def _inject_service_translation_strings(
        self,
        service: AbstractSpookService,
        cached_spook_translations: dict[str, str],
        language: str,
    ) -> None:
        """Inject service translation strings into Home Assistant's cache."""
        component_cache = self._translation_component_cache(
            language,
            service.domain,
            create=True,
        )
        if component_cache is None:
            return

        cached_translations = async_get_cached_translations(
            self.hass,
            language,
            SERVICE_TRANSLATION_CATEGORY,
            service.domain,
        )

        for key, value in self._service_translation_strings(
            service,
            cached_spook_translations,
        ).items():
            self._service_translation_overrides.setdefault(
                (language, service.domain, key), cached_translations.get(key)
            )
            component_cache[key] = value

    @callback
    def _selector_translation_keys(self, service: AbstractSpookService) -> set[str]:
        """Return the selector translation keys a Spook service's fields use."""
        schema = self._service_schemas.get(self._service_schema_key(service)) or {}
        keys: set[str] = set()
        for field_schema in (schema.get("fields") or {}).values():
            for selector_config in (
                (field_schema or {}).get("selector") or {}
            ).values():
                if isinstance(selector_config, dict) and (
                    key := selector_config.get("translation_key")
                ):
                    keys.add(key)
        return keys

    @callback
    def _inject_selector_translation_strings(
        self,
        service: AbstractSpookService,
        cached_spook_translations: dict[str, str],
        language: str,
    ) -> None:
        """Inject the option labels of a Spook service's selectors.

        Home Assistant looks a selector's labels up under the domain the action
        belongs to, `component.todo.selector...` for a `todo` action, and Spook
        keeps them under its own. Without this, the options show as their raw
        values.
        """
        if not (keys := self._selector_translation_keys(service)):
            return

        component_cache = self._translation_component_cache(
            language,
            service.domain,
            create=True,
            category=SELECTOR_TRANSLATION_CATEGORY,
        )
        if component_cache is None:
            return

        cached_translations = async_get_cached_translations(
            self.hass,
            language,
            SELECTOR_TRANSLATION_CATEGORY,
            service.domain,
        )

        for key in keys:
            spook_prefix = f"component.{DOMAIN}.selector.{key}."
            target_prefix = f"component.{service.domain}.selector.{key}."
            for spook_key, value in cached_spook_translations.items():
                if not spook_key.startswith(spook_prefix):
                    continue
                target_key = f"{target_prefix}{spook_key.removeprefix(spook_prefix)}"
                self._selector_translation_overrides.setdefault(
                    (language, service.domain, target_key),
                    cached_translations.get(target_key),
                )
                component_cache[target_key] = value

    async def async_inject_service_translations(
        self, languages: Iterable[str] | None = None
    ) -> None:
        """Inject Spook service strings and icons into Home Assistant.

        Into the languages given, or every language Spook has written into.
        """
        services = [
            service
            for service in self._services
            if self._service_schema_key(service) in self._service_schemas
        ]

        if not services:
            return

        if languages is None:
            # The server's own is always one of them. So is a language still
            # on its first injection: that one took its list of actions when it
            # started, and an action registered since would be left out.
            self._languages.add(self.hass.config.language)
            languages = self._languages | set(self._language_tasks)

        domains = {DOMAIN, *(service.domain for service in services)}
        for language in list(languages):
            await self._async_inject_translations_for(language, services, domains)

        await async_inject_service_icons(
            self.hass,
            (
                (service.domain, service.service, self._service_schema_key(service))
                for service in services
            ),
            self._injected_service_icons,
        )

    async def _async_inject_translations_for(
        self,
        language: str,
        services: list[AbstractSpookService],
        domains: set[str],
    ) -> None:
        """Inject the strings and selector labels of these services in a language."""
        await async_get_translations(
            self.hass, language, SERVICE_TRANSLATION_CATEGORY, domains
        )
        cached_spook_translations = async_get_cached_translations(
            self.hass, language, SERVICE_TRANSLATION_CATEGORY, DOMAIN
        )
        for service in services:
            self._inject_service_translation_strings(
                service, cached_spook_translations, language
            )

        await async_get_translations(
            self.hass, language, SELECTOR_TRANSLATION_CATEGORY, domains
        )
        cached_spook_selector_translations = async_get_cached_translations(
            self.hass, language, SELECTOR_TRANSLATION_CATEGORY, DOMAIN
        )
        for service in services:
            self._inject_selector_translation_strings(
                service, cached_spook_selector_translations, language
            )

    @callback
    def _async_core_config_updated(self, event: Event) -> None:
        """Re-inject service translations when the language changes."""
        if "language" not in event.data:
            return

        # The language it was is kept: somebody can still have it as theirs.
        self._languages.add(self.hass.config.language)
        self._async_reinject_service_translations([self.hass.config.language])

    @callback
    def async_clear_service_translation_overrides(self) -> None:
        """Restore translation strings that were overridden by Spook."""
        self._restore(self._service_translation_overrides, SERVICE_TRANSLATION_CATEGORY)
        self._restore(
            self._selector_translation_overrides, SELECTOR_TRANSLATION_CATEGORY
        )

    @callback
    def _restore(
        self,
        overrides: dict[tuple[str, str, str], str | None],
        category: str,
    ) -> None:
        """Put back what Spook overrode, and take away what it only added."""
        for (language, domain, key), original_value in overrides.items():
            component_cache = self._translation_component_cache(
                language, domain, category=category
            )
            if component_cache is None:
                continue

            if original_value is None:
                component_cache.pop(key, None)
            else:
                component_cache[key] = original_value

        overrides.clear()

    @callback
    def async_on_unload(self) -> None:
        """Tear down the Spook services."""
        LOGGER.debug("Tearing down Spook services")
        # A copy, as cancelling a finished injection takes it off the list.
        for undo in list(self._on_unload):
            undo()
        self._on_unload.clear()
        self._waiting_for_domain.clear()

        for service in self._services:
            LOGGER.debug(
                "Unregistering service: %s.%s",
                service.domain,
                service.service,
            )
            service.async_unregister()

            if (
                isinstance(service, ReplaceExistingService)
                and service.overriden_service
            ):
                LOGGER.debug(
                    "Restoring service that was overriden previously: %s.%s",
                    service.domain,
                    service.service,
                )

                # pylint: disable-next=protected-access
                self.hass.services._services.setdefault(  # noqa: SLF001
                    service.domain,
                    {},
                )[service.service] = service.overriden_service

                # Flush service description schema cache
                self.hass.data.pop(SERVICE_DESCRIPTION_CACHE, None)

        self.async_clear_service_translation_overrides()
        async_remove_service_icons(self.hass, self._injected_service_icons)