music-assistant-server

1"""
2Home Assistant Plugin for Music Assistant.
3
4The plugin is the core of all communication to/from Home Assistant and
5responsible for maintaining the WebSocket API connection to HA.
6Also, the Music Assistant integration within HA will relay its own api
7communication over the HA api for more flexibility as well as security.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import logging
14import os
15from functools import partial
16from itertools import batched
17from sys import intern
18from types import MappingProxyType
19from typing import TYPE_CHECKING, Any, NamedTuple, TypedDict, cast
20
21from hass_client import HomeAssistantClient
22from hass_client.exceptions import BaseHassClientError
23from hass_client.utils import get_websocket_url
24from music_assistant_models.auth import Scope
25from music_assistant_models.config_entries import ConfigEntry
26from music_assistant_models.enums import (
27    ConfigEntryType,
28    ContentType,
29    EventType,
30    MediaType,
31    ProviderFeature,
32    StreamType,
33)
34from music_assistant_models.errors import (
35    MusicAssistantError,
36    SetupFailedError,
37    UnsupportedFeaturedException,
38)
39from music_assistant_models.media_items.audio_format import AudioFormat
40from music_assistant_models.player_control import PlayerControl
41from music_assistant_models.streamdetails import StreamDetails
42
43from music_assistant.constants import VERBOSE_LOG_LEVEL
44from music_assistant.controllers.cache import use_cache
45from music_assistant.helpers.datetime import iso_from_utc_timestamp
46from music_assistant.helpers.json import SerializableType
47from music_assistant.helpers.util import lock, try_parse_int
48from music_assistant.models.plugin import AIEngine, PluginProvider, TTSEngine
49
50from .constants import (
51    CONF_MUTE_CONTROLS,
52    CONF_POWER_CONTROLS,
53    CONF_VOLUME_CONTROLS,
54    OFF_STATES,
55    MediaPlayerEntityFeature,
56    parse_supported_features,
57)
58from .control_entities import (
59    SEARCH_CONTROL_ENTITIES_LIMIT,
60    ControlEntitySearch,
61    HassControlEntitySearchResult,
62)
63from .helpers import ControlCapabilities, get_control_name, is_entity_id
64
65if TYPE_CHECKING:
66    from collections.abc import Callable, Collection, Mapping
67
68    from aiohttp import ClientSession
69    from hass_client.models import (
70        Area,
71        CompressedState,
72        Context,
73        Device,
74        Entity,
75        EntityStateEvent,
76        Event,
77        State,
78    )
79    from music_assistant_models.config_entries import ProviderConfig
80    from music_assistant_models.player import PlayerMedia
81    from music_assistant_models.provider import ProviderManifest
82
83    from music_assistant.mass import MusicAssistant
84    from music_assistant.models import ProviderInstanceType
85
86DOMAIN = "hass"
87CONF_URL = "url"
88CONF_AUTH_TOKEN = "token"
89CONF_VERIFY_SSL = "verify_ssl"
90FEATURE_DISCOVERY_TIMEOUT = 30
91STATE_FETCH_TIMEOUT = 30
92STATE_FETCH_BATCH_SIZE = 500
93# window to collect entity registry updates in, so an integration registering a
94# batch of entities results in a single rebuild of the engine lists
95ENGINE_REFRESH_DEBOUNCE = 2
96# window in which repeated device lookups reuse one listing, so a burst of players
97# connecting does not fetch the (unfilterable) device registry once per player
98DEVICE_REGISTRY_CACHE_TTL = 60
99# areas are renamed even less often than devices, and only ever supply a label
100AREA_REGISTRY_CACHE_TTL = 60
101
102SEARCH_CONTROL_ENTITIES_COMMAND = f"{DOMAIN}/search_control_entities"
103
104# Home Assistant entity domains that back the TTS and AI Task features.
105FEATURE_DOMAINS = ("tts", "ai_task")
106FEATURE_DOMAIN_PREFIXES = tuple(f"{domain}." for domain in FEATURE_DOMAINS)
107
108# Entity registry fields a change to which can alter the mirrored registry. Beyond the
109# mirrored fields themselves, disabled_by decides whether an entity is listed at all, and
110# config_entry_id joins them because Home Assistant can clear disabled_by while reporting
111# only the move to the other config entry.
112REGISTRY_FIELDS_AFFECTING_MIRROR = frozenset(
113    {"entity_id", "platform", "device_id", "area_id", "disabled_by", "config_entry_id"}
114)
115
116
117class DeviceMediaPlayerInfo(TypedDict):
118    """Home Assistant correlation info for a device that is natively connected elsewhere."""
119
120    # user-facing device name in HA (name_by_user or name)
121    name: str | None
122    # first enabled media_player entity of the device that supports announcements
123    announce_entity_id: str | None
124
125
126class HassRegistryEntity(NamedTuple):
127    """
128    Home Assistant entity registry entry, limited to the fields Music Assistant uses.
129
130    The entity ID is not a field: entries are always keyed by it.
131    """
132
133    platform: str
134    device_id: str | None
135    # the area the entity is assigned to directly, overriding the one of its device
136    area_id: str | None
137
138
139async def setup(
140    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
141) -> ProviderInstanceType:
142    """Initialize provider(instance) with given configuration."""
143    return HomeAssistantProvider(mass, manifest, config, set())
144
145
146def _control_config_entries() -> tuple[ConfigEntry, ...]:
147    """Return the config entries holding the entities selected as player controls."""
148    return tuple(
149        ConfigEntry(
150            key=conf_key,
151            type=ConfigEntryType.STRING,
152            multi_value=True,
153            required=True,
154            default_value=[],
155            category="player_controls",
156        )
157        for conf_key in (CONF_POWER_CONTROLS, CONF_VOLUME_CONTROLS, CONF_MUTE_CONTROLS)
158    )
159
160
161class HomeAssistantProvider(PluginProvider):
162    """Home Assistant Plugin for Music Assistant."""
163
164    hass: HomeAssistantClient
165    _listen_task: asyncio.Task[None] | None = None
166    _player_controls: dict[str, PlayerControl] | None = None
167    _unsubscribe_controls: Callable[[], None] | None = None
168    _unsubscribe_entity_registry: Callable[[], None] | None = None
169    _engine_refresh_task: asyncio.Task[None] | None = None
170    _ai_engines: list[AIEngine]
171    _tts_engines: list[TTSEngine]
172    _startup_complete: bool = False
173    _entity_registry: Mapping[str, HassRegistryEntity] | None = None
174    _entity_registry_generation: int = 0
175    _entity_registry_lock: asyncio.Lock
176    _wanted_controls: dict[str, ControlCapabilities] | None = None
177    _control_reconcile_lock: asyncio.Lock
178    _control_entity_search: ControlEntitySearch
179    _unregister_search_command: Callable[[], None] | None = None
180
181    @property
182    def url(self) -> str | None:
183        """Return the configured Home Assistant URL, or None if not configured."""
184        url = self.get_setup_value(CONF_URL)
185        if isinstance(url, str) and url:
186            return url
187        return None
188
189    @property
190    def entity_registry_generation(self) -> int:
191        """Return a counter that changes whenever the mirrored entity registry is dropped."""
192        return self._entity_registry_generation
193
194    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
195        """
196        Return the (options) config entries for the Home Assistant provider.
197
198        The connection URL and authentication token are collected by the setup flow (see
199        setup_flow.py) unless running as a Home Assistant add-on, where they are fixed; only
200        the player-control and feature options are configurable here.
201        """
202        base_entries: tuple[ConfigEntry, ...]
203        if self.mass.running_as_hass_addon:
204            # on supervisor, we use the internal url
205            # token set to None for auto retrieval
206            base_entries = (
207                ConfigEntry(
208                    key=CONF_URL,
209                    type=ConfigEntryType.STRING,
210                    label=CONF_URL,
211                    required=True,
212                    default_value="http://supervisor/core/api",
213                    value="http://supervisor/core/api",
214                    hidden=True,
215                ),
216                ConfigEntry(
217                    key=CONF_AUTH_TOKEN,
218                    type=ConfigEntryType.STRING,
219                    label=CONF_AUTH_TOKEN,
220                    required=False,
221                    default_value=None,
222                    value=None,
223                    hidden=True,
224                ),
225                ConfigEntry(
226                    key=CONF_VERIFY_SSL,
227                    type=ConfigEntryType.BOOLEAN,
228                    label=CONF_VERIFY_SSL,
229                    required=False,
230                    default_value=False,
231                    hidden=True,
232                ),
233            )
234        else:
235            # url/token/verify_ssl are collected by the setup flow instead (see setup_flow.py)
236            base_entries = ()
237
238        return (*base_entries, *_control_config_entries())
239
240    async def handle_async_init(self) -> None:
241        """Handle async initialization of the plugin."""
242        if self._listen_task and not self._listen_task.done():
243            msg = "Home Assistant listener is already running"
244            raise SetupFailedError(msg)
245        self._startup_complete = False
246        self._player_controls = {}
247        self._wanted_controls = None
248        self._control_reconcile_lock = asyncio.Lock()
249        self._ai_engines = []
250        self._tts_engines = []
251        url = get_websocket_url(cast("str", self.get_setup_value(CONF_URL)))
252        token = self.get_setup_value(CONF_AUTH_TOKEN)
253        logging.getLogger("hass_client").setLevel(self.logger.level + 10)
254        ssl = bool(self.get_setup_value(CONF_VERIFY_SSL, True))
255        http_session = self.mass.http_session if ssl else self.mass.http_session_no_ssl
256        self.hass = HomeAssistantClient(url, token, http_session)
257        self._entity_registry = None
258        self._entity_registry_lock = asyncio.Lock()
259        self._control_entity_search = ControlEntitySearch(self)
260        # registering here rather than in loaded_in_mass pairs the command with the teardown
261        # in _disconnect_hass, so a reload can never leave it registered twice
262        self._unregister_search_command = self.mass.register_api_command(
263            SEARCH_CONTROL_ENTITIES_COMMAND,
264            self.search_control_entities,
265            required_scope=Scope.CONFIG_PROVIDERS_READ,
266        )
267        try:
268            await self.hass.connect()
269        except BaseHassClientError as err:
270            await self._cleanup_failed_init()
271            err_msg = str(err) or err.__class__.__name__
272            raise SetupFailedError(err_msg) from err
273        self._listen_task = self.mass.create_task(self._hass_listener())
274        try:
275            # the registry subscription must be live before the first registry read, so no
276            # registry change can slip through unnoticed; _disconnect_hass tears the
277            # subscription down again on the failure paths below
278            await self._subscribe_entity_registry()
279            await self._resolve_startup_features()
280        except asyncio.CancelledError:
281            await self._cleanup_failed_init()
282            raise
283        except BaseHassClientError as err:
284            await self._cleanup_failed_init()
285            err_msg = str(err) or err.__class__.__name__
286            raise SetupFailedError(err_msg) from err
287        except Exception:
288            await self._cleanup_failed_init()
289            raise
290
291    async def loaded_in_mass(self) -> None:
292        """Call after the provider has been loaded."""
293        await self._register_player_controls()
294
295    async def unload(self, is_removed: bool = False) -> None:
296        """
297        Handle unload/close of the provider.
298
299        Called when provider is deregistered (e.g. MA exiting or config reloading).
300        """
301        # unregister all player controls
302        if self._player_controls:
303            for entity_id in self._player_controls:
304                self.mass.players.remove_player_control(entity_id)
305        self._startup_complete = False
306        await self._disconnect_hass()
307
308    async def update_config(self, config: ProviderConfig, changed_keys: set[str]) -> None:
309        """
310        Handle logic when the config is updated.
311
312        A change limited to the player control selection is applied in place, so adding or
313        removing a control does not drop and re-establish the Home Assistant connection.
314        Any other change reloads the provider as usual.
315
316        Raises when the in place update fails (because Home Assistant is unreachable, for
317        example); the controls are then left as they were and a later update retries.
318
319        :param config: The updated provider config.
320        :param changed_keys: The keys that changed in the given config.
321        """
322        control_keys = {
323            f"values/{conf_key}"
324            for conf_key in (CONF_POWER_CONTROLS, CONF_VOLUME_CONTROLS, CONF_MUTE_CONTROLS)
325        }
326        if not changed_keys or not changed_keys <= control_keys:
327            await super().update_config(config, changed_keys)
328            return
329        # store the new config before reconciling: the control lists are read back from it
330        self.config = config
331        await self._register_player_controls()
332
333    async def get_diagnostics(self) -> dict[str, SerializableType]:
334        """Return diagnostics info for this provider to include in diagnostics reports."""
335        return {
336            "connected": self.hass.connected,
337            "ha_version": self.hass.version,
338            "listener_active": self._listen_task is not None and not self._listen_task.done(),
339            "player_controls": len(self._player_controls) if self._player_controls else 0,
340        }
341
342    async def get_entity_registry(self) -> Mapping[str, HassRegistryEntity]:
343        """
344        Return the Home Assistant entity registry, keyed by entity ID.
345
346        Entities that are disabled in Home Assistant are absent from the result, and so are
347        entities without a unique ID: those are not part of Home Assistant's registry at all.
348
349        The result is shared between all callers and is read-only: both the mapping and
350        its entries reject writes.
351        """
352        if (registry := self._entity_registry) is not None:
353            return registry
354        async with self._entity_registry_lock:
355            if (registry := self._entity_registry) is None:
356                generation = self._entity_registry_generation
357                registry = await self._fetch_entity_registry()
358                # a registry change while the fetch was in flight leaves the listing stale
359                # on arrival, so serve it to this caller but keep it out of the cache
360                if generation == self._entity_registry_generation:
361                    self._entity_registry = registry
362            return registry
363
364    async def get_entity_registry_entries(self, entity_ids: Collection[str]) -> dict[str, Entity]:
365        """
366        Return the full Home Assistant entity registry entries of the given entities.
367
368        :param entity_ids: The entity IDs to look up.
369        :return: The registry entries keyed by entity ID; entities unknown to
370            Home Assistant are absent from the result.
371        """
372        if not entity_ids:
373            return {}
374        result = cast(
375            "dict[str, Entity | None]",
376            await self.hass.send_command(
377                "config/entity_registry/get_entries", entity_ids=list(entity_ids)
378            ),
379        )
380        return {entity_id: entry for entity_id, entry in result.items() if entry is not None}
381
382    async def get_device_registry(self) -> dict[str, Device]:
383        """
384        Return the Home Assistant device registry, keyed by device ID.
385
386        Home Assistant offers no abbreviated variant of the device registry listing, so the
387        entries carry all of their fields. The listing is reused for a short while, so a
388        device change may take up to DEVICE_REGISTRY_CACHE_TTL seconds to be reflected.
389        """
390        return await self._fetch_device_registry()
391
392    async def get_area_registry(self) -> dict[str, Area]:
393        """
394        Return the Home Assistant area registry, keyed by area ID.
395
396        The listing is reused for a short while, so an area change may take up to
397        AREA_REGISTRY_CACHE_TTL seconds to be reflected.
398        """
399        return await self._fetch_area_registry()
400
401    async def search_control_entities(
402        self,
403        search: str | None = None,
404        control_type: str | None = None,
405        limit: int = SEARCH_CONTROL_ENTITIES_LIMIT,
406    ) -> HassControlEntitySearchResult:
407        """
408        Search the Home Assistant entities that can be used as a player control.
409
410        Music Assistant's own players are never part of the result. Consecutive searches are
411        served from a short lived cache that an entity registry change drops right away, so a
412        newly added or removed entity shows up immediately, while a device or area rename can
413        lag by up to a minute.
414
415        :param search: Text to match, case insensitively, against the entity ID, the entity
416            name, its device name and its area name. Every whitespace separated word must
417            match one of those fields, though not necessarily the same one. All eligible
418            entities match when omitted.
419        :param control_type: Restrict the result to entities that can serve this control role,
420            given as one of the provider's control config keys (``power_controls``,
421            ``volume_controls`` or ``mute_controls``). All roles are returned when omitted.
422        :param limit: Maximum number of entities (not groups) to return, itself capped at
423            ``SEARCH_CONTROL_ENTITIES_MAX_LIMIT``.
424        :return: The matching entities grouped by the device and area they belong to, ordered
425            by area, device and entity name, plus a flag telling whether matches were left out
426            to honor the limit.
427        """
428        return await self._control_entity_search.search(search, control_type, limit)
429
430    async def get_media_player_device_infos(
431        self,
432        mac_addresses: Collection[str],
433        platform: str,
434    ) -> dict[str, DeviceMediaPlayerInfo]:
435        """
436        Correlate devices (by MAC address) to their HA name and media_player entity.
437
438        Used for devices that are natively connected to Music Assistant but also
439        present in Home Assistant, to pick up their HA device name and their
440        (announcement-capable) media_player entity.
441
442        :param mac_addresses: Device MAC addresses to look up (case-insensitive).
443        :param platform: The HA integration domain the media_player entities must belong to.
444        :return: Correlation info keyed by lowercased MAC address; devices unknown
445            to Home Assistant are absent from the result.
446        """
447        wanted_macs = {mac.lower() for mac in mac_addresses}
448        if not wanted_macs:
449            return {}
450        device_registry = await self.get_device_registry()
451        device_by_mac: dict[str, Device] = {
452            connection[1].lower(): device
453            for device in device_registry.values()
454            for connection in device.get("connections", [])
455            if len(connection) == 2
456            and connection[0] == "mac"
457            and connection[1].lower() in wanted_macs
458        }
459        if not device_by_mac:
460            return {}
461        media_players_by_device: dict[str, list[str]] = {}
462        for entity_id, entry in (await self.get_entity_registry()).items():
463            if (
464                entry.platform == platform
465                and entity_id.startswith("media_player.")
466                and (device_id := entry.device_id)
467            ):
468                media_players_by_device.setdefault(device_id, []).append(entity_id)
469        candidates_by_mac = {
470            mac: media_players_by_device.get(device["id"], [])
471            for mac, device in device_by_mac.items()
472        }
473        states = {
474            state["entity_id"]: state
475            for state in await self.get_states(
476                entity_ids=[
477                    entity_id
478                    for entity_ids in candidates_by_mac.values()
479                    for entity_id in entity_ids
480                ]
481            )
482        }
483
484        def _supports_announce(entity_id: str) -> bool:
485            if (state := states.get(entity_id)) is None:
486                return False
487            supported_features = parse_supported_features(
488                state["attributes"].get("supported_features"), entity_id, self.logger
489            )
490            return MediaPlayerEntityFeature.MEDIA_ANNOUNCE in supported_features
491
492        return {
493            mac: DeviceMediaPlayerInfo(
494                name=device["name_by_user"] or device["name"],
495                announce_entity_id=next(
496                    (
497                        entity_id
498                        for entity_id in candidates_by_mac[mac]
499                        if _supports_announce(entity_id)
500                    ),
501                    None,
502                ),
503            )
504            for mac, device in device_by_mac.items()
505        }
506
507    async def get_user_details(self, ha_user_id: str) -> tuple[str | None, str | None, str | None]:
508        """
509        Get user username, display name and avatar URL from Home Assistant.
510
511        Looks up the user in config/auth/list for username, and the person entity
512        for display name and picture URL.
513
514        :param ha_user_id: Home Assistant user ID.
515        :return: Tuple of (username, display_name, avatar_url) or all None if not found.
516        """
517        try:
518            username: str | None = None
519            display_name: str | None = None
520            avatar_url: str | None = None
521
522            # Get username from config/auth/list (admin endpoint, we have admin access)
523            try:
524                users = await self.hass.send_command("config/auth/list")
525                for user in users or []:
526                    if user.get("id") == ha_user_id:
527                        username = user.get("username")
528                        # Also get name as fallback display name
529                        if not display_name:
530                            display_name = user.get("name")
531                        break
532            except Exception as err:
533                self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA user list: %s", err)
534
535            # Get external URL for building avatar URL
536            ha_url: str | None = None
537            try:
538                network_urls = await self.hass.send_command("network/url")
539                if network_urls:
540                    ha_url = network_urls.get("external") or network_urls.get("internal")
541            except Exception as err:
542                self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA network URLs: %s", err)
543
544            # Find person linked to this HA user ID for display name and avatar
545            try:
546                persons = await self.hass.send_command("person/list")
547                # person/list returns {storage: [...], config: [...]}
548                all_persons = (persons.get("storage") or []) + (persons.get("config") or [])
549                for person in all_persons:
550                    if person.get("user_id") == ha_user_id:
551                        # Person name takes priority for display name
552                        if person_name := person.get("name"):
553                            display_name = person_name
554                        if (person_picture := person.get("picture")) and ha_url:
555                            avatar_url = f"{ha_url.rstrip('/')}{person_picture}"
556                        break
557            except Exception as err:
558                self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA person details: %s", err)
559
560            self.logger.log(
561                VERBOSE_LOG_LEVEL,
562                "get_user_details for %s: username=%s, display_name=%s, avatar_url=%s",
563                ha_user_id,
564                username,
565                display_name,
566                avatar_url,
567            )
568            return username, display_name, avatar_url
569        except Exception as err:
570            self.logger.warning("Failed to get HA user details: %s", err)
571            return None, None, None
572
573    async def get_states(
574        self,
575        *,
576        entity_ids: list[str] | None = None,
577        domains: Collection[str] | None = None,
578    ) -> list[State]:
579        """
580        Return the current Home Assistant state for the requested entities.
581
582        Provide explicit entity IDs and/or a set of domains; only those entities
583        are fetched.
584
585        :param entity_ids: Explicit entity IDs to fetch the current state for.
586        :param domains: Entity domains whose entities should be fetched.
587        """
588        ids: set[str] = set(entity_ids or ())
589        if domains:
590            # resolve domains to entity_ids via the registry, which is far smaller
591            # than a full state dump (it carries no attributes)
592            registry = await self.get_entity_registry()
593            ids.update(entity_id for entity_id in registry if entity_id.split(".", 1)[0] in domains)
594        if not ids:
595            return []
596        states: list[State] = []
597        async with asyncio.timeout(STATE_FETCH_TIMEOUT):
598            # exceeding hass_client's 16MB websocket message limit drops the entire
599            # connection, so bound the state dump by construction and fetch in batches
600            for batch in batched(sorted(ids), STATE_FETCH_BATCH_SIZE, strict=False):
601                states.extend(await self._fetch_states(list(batch)))
602        return states
603
604    async def resolve_image(self, path: str) -> bytes:
605        """Resolve an image from an image path."""
606        ha_url, headers, http_session = self._get_ha_http()
607        async with http_session.get(f"{ha_url}{path}", headers=headers) as response:
608            response.raise_for_status()
609            return await response.read()
610
611    async def get_ai_engines(self) -> list[AIEngine]:
612        """Return the Home Assistant AI Task entities as AI engines."""
613        return self._ai_engines
614
615    async def get_tts_engines(self) -> list[TTSEngine]:
616        """Return the Home Assistant TTS entities as TTS engines."""
617        return self._tts_engines
618
619    async def ai_query(self, query: str, engine_id: str | None = None) -> str:
620        """Handle an AI query via Home Assistant's ai_task service."""
621        entity_id = engine_id or next((engine.id for engine in self._ai_engines), None)
622        if entity_id is None:
623            raise UnsupportedFeaturedException("AI Task entity is not available")
624        result = await self.hass.send_command(
625            "call_service",
626            domain="ai_task",
627            service="generate_data",
628            service_data={
629                "task_name": "music_assistant",
630                "instructions": query,
631                "entity_id": entity_id,
632            },
633            return_response=True,
634        )
635        response = result.get("response", {}) if isinstance(result, dict) else {}
636        data = response.get("data") if isinstance(response, dict) else None
637        if not data:
638            msg = f"AI Task returned no data in response: {result}"
639            raise MusicAssistantError(msg)
640        return str(data)
641
642    async def play_announcement_on_entity(self, entity_id: str, announcement: PlayerMedia) -> None:
643        """
644        Play an announcement on a Home Assistant media_player entity.
645
646        Uses Home Assistant's announce feature, so the entity's integration ducks
647        or pauses any running playback and resumes it afterwards. Returns once the
648        announcement has finished playing (approximated by its duration).
649
650        :param entity_id: The media_player entity to play the announcement on.
651        :param announcement: The announcement to play.
652        """
653        await self.hass.call_service(
654            domain="media_player",
655            service="play_media",
656            service_data={
657                "media_content_id": announcement.uri,
658                "media_content_type": "music",
659                "announce": True,
660            },
661            target={"entity_id": entity_id},
662        )
663        # Wait until the announcement is finished playing so callers can play
664        # announcements in a sequence; HA gives no completion signal for announcements.
665        duration = await self.mass.streams.get_announcement_duration(announcement)
666        await asyncio.sleep(duration or 5)
667
668    async def get_tts_message(
669        self, message: str, language: str | None = None, engine_id: str | None = None
670    ) -> StreamDetails:
671        """Handle text-to-speech via Home Assistant's REST API."""
672        entity_id = engine_id or next((engine.id for engine in self._tts_engines), None)
673        if entity_id is None:
674            raise UnsupportedFeaturedException("TTS entity is not available")
675        ha_url, headers, http_session = self._get_ha_http()
676        # the tts_get_url payload field is called engine_id but takes a tts entity_id
677        payload: dict[str, str] = {"engine_id": entity_id, "message": message}
678        if language:
679            payload["language"] = language
680        async with http_session.post(
681            f"{ha_url}/api/tts_get_url", headers=headers, json=payload
682        ) as response:
683            response.raise_for_status()
684            data = await response.json()
685        url = str(data["url"])
686        return StreamDetails(
687            provider=self.instance_id,
688            item_id=url,
689            audio_format=AudioFormat(content_type=ContentType.MP3),
690            media_type=MediaType.SOUND_EFFECT,
691            stream_type=StreamType.HTTP,
692            path=url,
693        )
694
695    async def _hass_listener(self) -> None:
696        """Start listening on the HA websockets."""
697        try:
698            # start listening will block until the connection is lost/closed
699            await self.hass.start_listening()
700        except BaseHassClientError as err:
701            self.logger.warning("Connection to HA lost due to error: %s", err)
702        if not self._startup_complete:
703            return
704        self.logger.info("Connection to HA lost. Connection will be automatically retried later.")
705        # schedule a reload of the provider, armed under the load path's task id so any
706        # (re)load starting before it fires cancels it
707        self.available = False
708        self.mass.call_later(
709            5,
710            self.mass.load_provider,
711            self.instance_id,
712            allow_retry=True,
713            task_id=f"load_provider_{self.instance_id}",
714        )
715
716    def _on_entity_state_update(self, event: EntityStateEvent) -> None:
717        """Handle Entity State event."""
718        if entity_additions := event.get("a"):
719            for entity_id, state in entity_additions.items():
720                self._update_control_from_state_msg(entity_id, state)
721        if entity_changes := event.get("c"):
722            for entity_id, state_diff in entity_changes.items():
723                if "+" not in state_diff:
724                    continue
725                self._update_control_from_state_msg(entity_id, state_diff["+"])
726
727    async def _register_player_controls(self) -> None:
728        """Bring the registered player controls in line with the current configuration."""
729        assert self._player_controls is not None  # for type checking
730        # the wanted selection is determined inside the lock, so a reconcile that had to
731        # wait for another one cannot apply a selection that was already superseded
732        async with self._control_reconcile_lock:
733            power_controls = self._selected_control_entities(CONF_POWER_CONTROLS)
734            mute_controls = self._selected_control_entities(CONF_MUTE_CONTROLS)
735            volume_controls = self._selected_control_entities(CONF_VOLUME_CONTROLS)
736            wanted_controls: dict[str, ControlCapabilities] = {
737                entity_id: ControlCapabilities(
738                    power=entity_id in power_controls,
739                    volume=entity_id in volume_controls,
740                    mute=entity_id in mute_controls,
741                )
742                for entity_id in (*power_controls, *mute_controls, *volume_controls)
743            }
744            if wanted_controls == self._wanted_controls:
745                # the selection is unchanged, so there is no need to consult Home Assistant
746                return
747            hass_states = {
748                state["entity_id"]: state
749                for state in await self.get_states(entity_ids=list(wanted_controls))
750            }
751            for entity_id in set(self._player_controls) - set(wanted_controls):
752                del self._player_controls[entity_id]
753                self.mass.players.remove_player_control(entity_id)
754            for entity_id, capabilities in wanted_controls.items():
755                control = self._create_player_control(
756                    entity_id, hass_states.get(entity_id), capabilities
757                )
758                self._player_controls[entity_id] = control
759                await self.mass.players.register_or_update_player_control(control)
760            await self._subscribe_control_states()
761            self._wanted_controls = wanted_controls
762
763    def _selected_control_entities(self, conf_key: str) -> list[str]:
764        """
765        Return the entity IDs selected in the given player control setting.
766
767        :param conf_key: The control config key to read the selection from.
768        """
769        entity_ids: list[str] = []
770        for value in cast("list[str]", self.config.get_value(conf_key)):
771            if is_entity_id(value):
772                entity_ids.append(value)
773                continue
774            # Home Assistant rejects an entire state fetch or subscription over a single
775            # value that is not an entity ID, so a leftover selection would otherwise
776            # take down every control of this provider
777            self.logger.warning(
778                "Ignoring %r in the %s setting: it is not a Home Assistant entity ID",
779                value,
780                conf_key,
781            )
782        return entity_ids
783
784    def _create_player_control(
785        self,
786        entity_id: str,
787        hass_state: State | None,
788        capabilities: ControlCapabilities,
789    ) -> PlayerControl:
790        """
791        Return a ready to use PlayerControl for a Home Assistant entity.
792
793        :param entity_id: The entity to base the control on.
794        :param hass_state: The entity's current state, if known.
795        :param capabilities: The control roles the entity should serve.
796        """
797        entity_platform = entity_id.split(".", maxsplit=1)[0]
798        control = PlayerControl(
799            id=entity_id,
800            provider=self.instance_id,
801            name=get_control_name(entity_id, hass_state),
802        )
803        if capabilities.power:
804            control.supports_power = True
805            control.power_state = hass_state["state"] not in OFF_STATES if hass_state else False
806            control.power_on = partial(self._handle_player_control_power_on, entity_id)
807            control.power_off = partial(self._handle_player_control_power_off, entity_id)
808        if capabilities.volume:
809            control.supports_volume = True
810            if not hass_state:
811                control.volume_level = 0
812            elif entity_platform == "media_player":
813                control.volume_level = int(hass_state["attributes"].get("volume_level", 0) * 100)
814            else:
815                control.volume_level = try_parse_int(hass_state["state"]) or 0
816            control.volume_set = partial(self._handle_player_control_volume_set, entity_id)
817        if capabilities.mute:
818            control.supports_mute = True
819            if not hass_state:
820                control.volume_muted = False
821            elif entity_platform == "media_player":
822                control.volume_muted = hass_state["attributes"].get("volume_muted")
823            else:
824                control.volume_muted = hass_state["state"] not in OFF_STATES
825            control.mute_set = partial(self._handle_player_control_mute_set, entity_id)
826        return control
827
828    async def _subscribe_control_states(self) -> None:
829        """Subscribe to the Home Assistant state of all currently tracked controls."""
830        assert self._player_controls is not None  # for type checking
831        # the earlier subscription is only released once the new one is live, so a failure
832        # to subscribe leaves the controls watched by the subscription they already had
833        previous_unsubscribe = self._unsubscribe_controls
834        self._unsubscribe_controls = await self.hass.subscribe_entities(
835            self._on_entity_state_update, list(self._player_controls)
836        )
837        if previous_unsubscribe:
838            previous_unsubscribe()
839
840    async def _handle_player_control_power_on(self, entity_id: str) -> None:
841        """Handle powering on the playercontrol."""
842        await self.hass.call_service(
843            domain="homeassistant",
844            service="turn_on",
845            target={"entity_id": entity_id},
846        )
847
848    async def _handle_player_control_power_off(self, entity_id: str) -> None:
849        """Handle powering off the playercontrol."""
850        await self.hass.call_service(
851            domain="homeassistant",
852            service="turn_off",
853            target={"entity_id": entity_id},
854        )
855
856    async def _handle_player_control_mute_set(self, entity_id: str, muted: bool) -> None:
857        """Handle muting the playercontrol."""
858        if entity_id.startswith("media_player."):
859            await self.hass.call_service(
860                domain="media_player",
861                service="volume_mute",
862                service_data={"is_volume_muted": muted},
863                target={"entity_id": entity_id},
864            )
865        else:
866            await self.hass.call_service(
867                domain="homeassistant",
868                service="turn_off" if muted else "turn_on",
869                target={"entity_id": entity_id},
870            )
871
872    async def _handle_player_control_volume_set(self, entity_id: str, volume_level: int) -> None:
873        """Handle setting volume on the playercontrol."""
874        domain = entity_id.split(".", 1)[0]
875
876        if domain == "media_player":
877            await self.hass.call_service(
878                domain=domain,
879                service="volume_set",
880                service_data={"volume_level": volume_level / 100},
881                target={"entity_id": entity_id},
882            )
883            return
884
885        # At this point, `set_value` will work for both `number` or `input_number`
886        await self.hass.call_service(
887            domain=domain,
888            service="set_value",
889            target={"entity_id": entity_id},
890            service_data={"value": volume_level},
891        )
892
893    def _update_control_from_state_msg(self, entity_id: str, state: CompressedState) -> None:
894        """Update PlayerControl from state(update) message."""
895        if self._player_controls is None:
896            return
897        if not (player_control := self._player_controls.get(entity_id)):
898            return
899        entity_platform = entity_id.split(".", maxsplit=1)[0]
900        if "s" in state:
901            # state changed
902            if player_control.supports_power:
903                player_control.power_state = state["s"] not in OFF_STATES
904            if player_control.supports_mute and entity_platform != "media_player":
905                player_control.volume_muted = state["s"] not in OFF_STATES
906            if player_control.supports_volume and entity_platform != "media_player":
907                player_control.volume_level = try_parse_int(state["s"]) or 0
908        if "a" in state and (attributes := state["a"]):
909            if player_control.supports_volume and "volume_level" in attributes:
910                player_control.volume_level = int(attributes.get("volume_level", 0) * 100)
911            if player_control.supports_mute and "is_volume_muted" in attributes:
912                player_control.volume_muted = attributes.get("is_volume_muted")
913        self.mass.players.update_player_control(entity_id)
914
915    async def _fetch_states(self, entity_ids: list[str]) -> list[State]:
916        """
917        Return the current Home Assistant state of the given entities.
918
919        :param entity_ids: The entity IDs to fetch the current state for.
920        :return: The states of the requested entities; entities that currently have
921            no state are absent from the result.
922        """
923        initial_states: asyncio.Future[dict[str, CompressedState]]
924        initial_states = asyncio.get_running_loop().create_future()
925
926        def _on_initial_states(event: EntityStateEvent) -> None:
927            # only the first message of a subscription carries the full state under "a";
928            # a state change racing in ahead of it must not resolve the fetch
929            if (added := event.get("a")) is not None and not initial_states.done():
930                initial_states.set_result(added)
931
932        unsubscribe = await self.hass.subscribe_entities(_on_initial_states, entity_ids)
933        try:
934            compressed_states = await initial_states
935        finally:
936            unsubscribe()
937        return [
938            _decompress_state(entity_id, compressed_state)
939            for entity_id, compressed_state in compressed_states.items()
940        ]
941
942    def _get_ha_http(self) -> tuple[str, dict[str, str], ClientSession]:
943        """Return HA base URL (without trailing /api), auth headers, and the HTTP session."""
944        ha_url = cast("str", self.get_setup_value(CONF_URL)).rstrip("/")
945        ha_url = ha_url.removesuffix("/api")
946        token = self.get_setup_value(CONF_AUTH_TOKEN) or os.environ.get("HASSIO_TOKEN")
947        headers = {"Authorization": f"Bearer {token}"} if token else {}
948        ssl = bool(self.get_setup_value(CONF_VERIFY_SSL, True))
949        http_session = self.mass.http_session if ssl else self.mass.http_session_no_ssl
950        return ha_url, headers, http_session
951
952    async def _disconnect_hass(self) -> None:
953        """Stop listening for Home Assistant events and disconnect the client."""
954        if unregister := self._unregister_search_command:
955            self._unregister_search_command = None
956            unregister()
957        self._control_entity_search.close()
958        if unsubscribe := self._unsubscribe_controls:
959            self._unsubscribe_controls = None
960            unsubscribe()
961        if unsubscribe := self._unsubscribe_entity_registry:
962            self._unsubscribe_entity_registry = None
963            unsubscribe()
964        if refresh_task := self._engine_refresh_task:
965            self._engine_refresh_task = None
966            refresh_task.cancel()
967        if listen_task := self._listen_task:
968            self._listen_task = None
969            if not listen_task.done():
970                listen_task.cancel()
971            try:
972                await listen_task
973            except asyncio.CancelledError:
974                pass
975            except Exception as err:
976                self.logger.warning("Home Assistant listener stopped with error: %s", err)
977        await self.hass.disconnect()
978
979    async def _cleanup_failed_init(self) -> None:
980        """Clean up the Home Assistant connection after initialization fails."""
981        try:
982            await self._disconnect_hass()
983        except Exception as err:
984            self.logger.warning("Failed to disconnect from Home Assistant: %s", err)
985
986    async def _resolve_startup_features(self) -> None:
987        """Resolve Home Assistant features while the listener remains active."""
988        assert self._listen_task is not None
989        feature_task = asyncio.create_task(self._refresh_engines())
990        try:
991            try:
992                async with asyncio.timeout(FEATURE_DISCOVERY_TIMEOUT):
993                    await asyncio.wait(
994                        {feature_task, self._listen_task},
995                        return_when=asyncio.FIRST_COMPLETED,
996                    )
997            except TimeoutError as err:
998                msg = "Timed out while resolving Home Assistant feature entities"
999                raise SetupFailedError(msg) from err
1000            if not feature_task.done():
1001                msg = "Home Assistant listener stopped during startup"
1002                raise SetupFailedError(msg)
1003            if feature_task.cancelled():
1004                if self._listen_task.done():
1005                    msg = "Home Assistant listener stopped during startup"
1006                else:
1007                    msg = "Home Assistant feature resolution was cancelled"
1008                raise SetupFailedError(msg)
1009            await feature_task
1010            if self._listen_task.done():
1011                msg = "Home Assistant listener stopped during startup"
1012                raise SetupFailedError(msg)
1013            self._startup_complete = True
1014        finally:
1015            if not feature_task.done():
1016                feature_task.cancel()
1017                await asyncio.gather(feature_task, return_exceptions=True)
1018
1019    async def _refresh_engines(self) -> None:
1020        """Rebuild the TTS/AI engine lists from the Home Assistant feature entities."""
1021        tts_engines: list[TTSEngine] = []
1022        ai_engines: list[AIEngine] = []
1023        for state in await self.get_states(domains=FEATURE_DOMAINS):
1024            entity_id = state["entity_id"]
1025            entity_platform = entity_id.split(".", 1)[0]
1026            if friendly_name := state["attributes"].get("friendly_name"):
1027                name = f"{friendly_name} ({entity_id})"
1028            else:
1029                name = entity_id
1030            if entity_platform == "tts":
1031                tts_engines.append(TTSEngine(id=entity_id, name=name, provider=self))
1032            elif entity_platform == "ai_task":
1033                ai_engines.append(AIEngine(id=entity_id, name=name, provider=self))
1034        tts_engines.sort(key=lambda engine: engine.name)
1035        ai_engines.sort(key=lambda engine: engine.name)
1036        changed = (self._tts_engines, self._ai_engines) != (tts_engines, ai_engines)
1037        self._tts_engines = tts_engines
1038        self._ai_engines = ai_engines
1039        self._supported_features.discard(ProviderFeature.TTS)
1040        self._supported_features.discard(ProviderFeature.AI_QUERY)
1041        if tts_engines:
1042            self._supported_features.add(ProviderFeature.TTS)
1043        if ai_engines:
1044            self._supported_features.add(ProviderFeature.AI_QUERY)
1045        # the entities can come and go without this provider (un)loading, so tell the
1046        # consumers of our engines that their selection may need re-evaluating. They read
1047        # the lists straight from their handler, so this has to stay below the assignments.
1048        # during startup the load itself signals once we are done.
1049        if changed and self._startup_complete:
1050            self.mass.signal_event(EventType.PROVIDERS_UPDATED, data=self.mass.get_providers())
1051
1052    async def _subscribe_entity_registry(self) -> None:
1053        """Watch the Home Assistant entity registry to keep the engine lists up to date."""
1054        # register for entity registry updates, replacing any earlier subscription
1055        if unsubscribe := self._unsubscribe_entity_registry:
1056            self._unsubscribe_entity_registry = None
1057            unsubscribe()
1058        self._unsubscribe_entity_registry = await self.hass.subscribe_events(
1059            self._on_entity_registry_update, "entity_registry_updated"
1060        )
1061
1062    def _on_entity_registry_update(self, event: Event) -> None:
1063        """Handle an entity registry update event."""
1064        data = event["data"]
1065        if _affects_mirrored_registry(data):
1066            self._entity_registry = None
1067            self._entity_registry_generation += 1
1068        elif self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1069            # the kept mirror rests on which fields Home Assistant reports, so leave a
1070            # trail that tells a stale entity listing apart from a lookup that never ran
1071            self.logger.log(
1072                VERBOSE_LOG_LEVEL,
1073                "Keeping the mirrored entity registry, %s changed on %s",
1074                ", ".join(data["changes"]),
1075                data.get("entity_id", "?"),
1076            )
1077        entity_id = data.get("entity_id", "")
1078        if not entity_id.startswith(FEATURE_DOMAIN_PREFIXES):
1079            return
1080        self._schedule_engine_refresh()
1081
1082    def _schedule_engine_refresh(self) -> None:
1083        """(Re)schedule the debounced rebuild of the engine lists."""
1084        if refresh_task := self._engine_refresh_task:
1085            self._engine_refresh_task = None
1086            refresh_task.cancel()
1087        self._engine_refresh_task = self.mass.create_task(self._delayed_engine_refresh())
1088
1089    async def _delayed_engine_refresh(self) -> None:
1090        """Rebuild the engine lists once the debounce window has passed."""
1091        await asyncio.sleep(ENGINE_REFRESH_DEBOUNCE)
1092        try:
1093            await self._refresh_engines()
1094        except Exception as err:
1095            self.logger.warning("Failed to refresh Home Assistant engines: %s", err)
1096
1097    # unlike _fetch_device_registry, this listing is mirrored for the lifetime of the
1098    # connection rather than kept behind a TTL: it runs to several megabytes on a large
1099    # setup, and Home Assistant announces every change, so the mirror is both the cheaper
1100    # and the more accurate option
1101    async def _fetch_entity_registry(self) -> Mapping[str, HassRegistryEntity]:
1102        """Fetch the entity registry from Home Assistant, keyed by entity ID."""
1103        # the display variant of the registry listing carries abbreviated keys and only the
1104        # fields the Home Assistant frontend needs, making it several times smaller
1105        result = cast(
1106            "dict[str, Any]",
1107            await self.hass.send_command("config/entity_registry/list_for_display"),
1108        )
1109        # the listing repeats a handful of platform names, one device id per device and one
1110        # area id per area over all of its entities, so hold on to a single string object
1111        # per distinct value
1112        device_ids: dict[str, str] = {}
1113        area_ids: dict[str, str] = {}
1114        registry: dict[str, HassRegistryEntity] = {}
1115        for entry in result["entities"]:
1116            if (device_id := entry.get("di")) is not None:
1117                device_id = device_ids.setdefault(device_id, device_id)
1118            if (area_id := entry.get("ai")) is not None:
1119                area_id = area_ids.setdefault(area_id, area_id)
1120            registry[entry["ei"]] = HassRegistryEntity(
1121                platform=intern(entry["pl"]),
1122                device_id=device_id,
1123                area_id=area_id,
1124            )
1125        return MappingProxyType(registry)
1126
1127    # the lock sits outside the cache to keep a burst of lookups from fetching once per
1128    # caller. use_cache stores in the background, so the callers that reach a still-cold
1129    # cache can overlap: the burst costs a couple of fetches instead of one per player
1130    @lock
1131    @use_cache(expiration=DEVICE_REGISTRY_CACHE_TTL)
1132    async def _fetch_device_registry(self) -> dict[str, Any]:
1133        """Fetch the device registry from Home Assistant, keyed by device ID."""
1134        # use_cache rebuilds the cached value from this return annotation, which rules out
1135        # the Device TypedDict; get_device_registry restores the type for callers
1136        return {device["id"]: device for device in await self.hass.get_device_registry()}
1137
1138    @lock
1139    @use_cache(expiration=AREA_REGISTRY_CACHE_TTL)
1140    async def _fetch_area_registry(self) -> dict[str, Any]:
1141        """Fetch the area registry from Home Assistant, keyed by area ID."""
1142        # use_cache rebuilds the cached value from this return annotation, which rules out
1143        # the Area TypedDict; get_area_registry restores the type for callers
1144        return {area["area_id"]: area for area in await self.hass.get_area_registry()}
1145
1146
1147def _affects_mirrored_registry(data: Mapping[str, Any]) -> bool:
1148    """
1149    Return whether an entity registry update can change the mirrored entity registry.
1150
1151    :param data: The data of a Home Assistant entity_registry_updated event.
1152    """
1153    if data.get("action") != "update":
1154        # a created or removed entity always enters or leaves the listing
1155        return True
1156    # an update reports the fields it touched, so a change that only concerns fields we do
1157    # not mirror (a rename, an icon, a label) leaves our listing accurate. an update can also
1158    # report no fields at all, as a device rename re-derives a name field that Home Assistant
1159    # strips from the report, so treat that as a change of unknown reach
1160    if not (changes := data.get("changes")):
1161        return True
1162    return not REGISTRY_FIELDS_AFFECTING_MIRROR.isdisjoint(changes)
1163
1164
1165def _decompress_state(entity_id: str, compressed_state: CompressedState) -> State:
1166    """
1167    Return the full state representation of a compressed state message.
1168
1169    :param entity_id: The entity the compressed state belongs to.
1170    :param compressed_state: The compressed state as received over the websocket.
1171    """
1172    raw_context = compressed_state.get("c")
1173    context: Context = (
1174        raw_context
1175        if isinstance(raw_context, dict)
1176        else {"id": raw_context or "", "parent_id": None, "user_id": None}
1177    )
1178    last_changed = compressed_state.get("lc")
1179    # Home Assistant omits last_updated when it is identical to last_changed
1180    last_updated = compressed_state.get("lu", last_changed)
1181    return {
1182        "entity_id": entity_id,
1183        "state": compressed_state.get("s", ""),
1184        "attributes": compressed_state.get("a", {}),
1185        "last_changed": iso_from_utc_timestamp(last_changed) if last_changed else "",
1186        "last_updated": iso_from_utc_timestamp(last_updated) if last_updated else "",
1187        "context": context,
1188    }
1189