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