/
/
/
1"""Helpers and utilities for the Home Assistant PlayerProvider."""
2
3from __future__ import annotations
4
5import logging
6import os
7from typing import TYPE_CHECKING, TypedDict, cast
8
9from music_assistant_models.enums import IdentifierType
10from music_assistant_models.errors import InvalidDataError, LoginFailed
11
12from music_assistant.providers.hass.constants import (
13 MediaPlayerEntityFeature,
14 parse_supported_features,
15)
16
17from .constants import BLOCKLISTED_HASS_INTEGRATIONS, DOMAIN
18
19if TYPE_CHECKING:
20 from collections.abc import AsyncGenerator
21
22 from hass_client.models import Entity as HassEntity
23 from hass_client.models import State as HassState
24
25 from music_assistant.mass import MusicAssistant
26 from music_assistant.providers.hass import HomeAssistantProvider
27
28
29async def get_media_player_entity_registry(
30 hass_prov: HomeAssistantProvider,
31) -> dict[str, HassEntity]:
32 """
33 Return the full registry entries of all Home Assistant media_player entities.
34
35 :param hass_prov: The Home Assistant provider to read the registry from.
36 :return: The registry entries keyed by entity ID.
37 """
38 registry = await hass_prov.get_entity_registry()
39 return await hass_prov.get_entity_registry_entries(
40 [entity_id for entity_id in registry if entity_id.startswith("media_player.")]
41 )
42
43
44async def get_hass_media_players(
45 hass_prov: HomeAssistantProvider,
46 entity_registry: dict[str, HassEntity],
47) -> AsyncGenerator[tuple[HassState, HassEntity | None]]:
48 """
49 Return all HA state objects (with registry entry) for (valid) media_player entities.
50
51 :param hass_prov: The Home Assistant provider to fetch the entity states from.
52 :param entity_registry: The registry entries of the media_player entities to consider.
53 """
54 # discover via the registry instead of a full state dump; entities without a
55 # unique_id are not registered and are therefore not discovered here
56 media_player_ids = list(entity_registry)
57 for state in await hass_prov.get_states(entity_ids=media_player_ids):
58 if "mass_player_type" in state["attributes"]:
59 # filter out mass players
60 continue
61 if "friendly_name" not in state["attributes"]:
62 # filter out invalid/unavailable players
63 continue
64 supported_features = parse_supported_features(
65 state["attributes"].get("supported_features"),
66 state["entity_id"],
67 hass_prov.logger,
68 )
69 if MediaPlayerEntityFeature.PLAY_MEDIA not in supported_features:
70 continue
71 entity_registry_entry = entity_registry.get(state["entity_id"])
72 if entity_registry_entry is not None:
73 hass_domain = entity_registry_entry["platform"]
74 if hass_domain in BLOCKLISTED_HASS_INTEGRATIONS:
75 continue
76 yield state, entity_registry_entry
77
78
79def normalized_mac(mac: str) -> str:
80 """Normalize a MAC address for comparison (lowercase, no separators)."""
81 return mac.replace(":", "").replace("-", "").lower()
82
83
84def native_player_macs(mass: MusicAssistant) -> set[str]:
85 """
86 Collect the normalized MAC addresses of all natively registered players.
87
88 Used to detect HA entities that point to a device that is already available
89 as a native Music Assistant player (e.g. an ESPHome device using Sendspin).
90 """
91 macs: set[str] = set()
92 for player in mass.players:
93 if player.provider.domain == DOMAIN:
94 continue
95 if mac := player.device_info.identifiers.get(IdentifierType.MAC_ADDRESS):
96 macs.add(normalized_mac(mac))
97 return macs
98
99
100class ESPHomeSupportedAudioFormat(TypedDict):
101 """ESPHome Supported Audio Format."""
102
103 format: str # flac, wav or mp3
104 sample_rate: int # e.g. 48000
105 num_channels: int # 1 for announcements, 2 for media
106 purpose: int # 0 for media, 1 for announcements
107 sample_bytes: int # 1 for 8 bit, 2 for 16 bit, 4 for 32 bit
108
109
110async def get_esphome_supported_audio_formats(
111 hass_prov: HomeAssistantProvider, conf_entry_id: str
112) -> list[ESPHomeSupportedAudioFormat]:
113 """Get supported audio formats for an ESPHome device."""
114 result: list[ESPHomeSupportedAudioFormat] = []
115 try:
116 # TODO: expose this in the hass client lib instead of hacking around private vars
117 ws_url = hass_prov.hass._websocket_url or "ws://supervisor/core/websocket"
118 hass_url = ws_url.replace("ws://", "http://").replace("wss://", "https://")
119 hass_url = hass_url.replace("/api/websocket", "").replace("/websocket", "")
120 api_token = hass_prov.hass._token or os.environ.get("HASSIO_TOKEN")
121 url = f"{hass_url}/api/diagnostics/config_entry/{conf_entry_id}"
122 headers = {
123 "Authorization": f"Bearer {api_token}",
124 "content-type": "application/json",
125 }
126 async with hass_prov.mass.http_session.get(url, headers=headers) as response:
127 if response.status != 200:
128 raise LoginFailed("Unable to contact Home Assistant to retrieve diagnostics")
129 data = await response.json()
130 if "data" not in data or "storage_data" not in data["data"]:
131 return result
132 if "media_player" not in data["data"]["storage_data"]:
133 raise InvalidDataError("Media player info not found in ESPHome diagnostics")
134 for media_player_obj in data["data"]["storage_data"]["media_player"]:
135 if "supported_formats" not in media_player_obj:
136 continue
137 for supported_format_obj in media_player_obj["supported_formats"]:
138 result.append(cast("ESPHomeSupportedAudioFormat", supported_format_obj))
139 except Exception as exc:
140 hass_prov.logger.warning(
141 "Failed to fetch diagnostics for ESPHome player: %s",
142 str(exc),
143 exc_info=exc if hass_prov.logger.isEnabledFor(logging.DEBUG) else None,
144 )
145 return result
146