/
/
/
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.errors import InvalidDataError, LoginFailed
10
11from music_assistant.providers.hass.constants import MediaPlayerEntityFeature
12
13from .constants import BLOCKLISTED_HASS_INTEGRATIONS
14
15if TYPE_CHECKING:
16 from collections.abc import AsyncGenerator
17
18 from hass_client.models import State as HassState
19
20 from music_assistant.providers.hass import HomeAssistantProvider
21
22
23async def get_hass_media_players(
24 hass_prov: HomeAssistantProvider,
25) -> AsyncGenerator[HassState, None]:
26 """Return all HA state objects for (valid) media_player entities."""
27 entity_registry = {x["entity_id"]: x for x in await hass_prov.hass.get_entity_registry()}
28 for state in await hass_prov.hass.get_states():
29 if not state["entity_id"].startswith("media_player"):
30 continue
31 if "mass_player_type" in state["attributes"]:
32 # filter out mass players
33 continue
34 if "friendly_name" not in state["attributes"]:
35 # filter out invalid/unavailable players
36 continue
37 supported_features = MediaPlayerEntityFeature(state["attributes"]["supported_features"])
38 if MediaPlayerEntityFeature.PLAY_MEDIA not in supported_features:
39 continue
40 if entity_registry_entry := entity_registry.get(state["entity_id"]):
41 hass_domain = entity_registry_entry["platform"]
42 if hass_domain in BLOCKLISTED_HASS_INTEGRATIONS:
43 continue
44 yield state
45
46
47class ESPHomeSupportedAudioFormat(TypedDict):
48 """ESPHome Supported Audio Format."""
49
50 format: str # flac, wav or mp3
51 sample_rate: int # e.g. 48000
52 num_channels: int # 1 for announcements, 2 for media
53 purpose: int # 0 for media, 1 for announcements
54 sample_bytes: int # 1 for 8 bit, 2 for 16 bit, 4 for 32 bit
55
56
57async def get_esphome_supported_audio_formats(
58 hass_prov: HomeAssistantProvider, conf_entry_id: str
59) -> list[ESPHomeSupportedAudioFormat]:
60 """Get supported audio formats for an ESPHome device."""
61 result: list[ESPHomeSupportedAudioFormat] = []
62 try:
63 # TODO: expose this in the hass client lib instead of hacking around private vars
64 ws_url = hass_prov.hass._websocket_url or "ws://supervisor/core/websocket"
65 hass_url = ws_url.replace("ws://", "http://").replace("wss://", "https://")
66 hass_url = hass_url.replace("/api/websocket", "").replace("/websocket", "")
67 api_token = hass_prov.hass._token or os.environ.get("HASSIO_TOKEN")
68 url = f"{hass_url}/api/diagnostics/config_entry/{conf_entry_id}"
69 headers = {
70 "Authorization": f"Bearer {api_token}",
71 "content-type": "application/json",
72 }
73 async with hass_prov.mass.http_session.get(url, headers=headers) as response:
74 if response.status != 200:
75 raise LoginFailed("Unable to contact Home Assistant to retrieve diagnostics")
76 data = await response.json()
77 if "data" not in data or "storage_data" not in data["data"]:
78 return result
79 if "media_player" not in data["data"]["storage_data"]:
80 raise InvalidDataError("Media player info not found in ESPHome diagnostics")
81 for media_player_obj in data["data"]["storage_data"]["media_player"]:
82 if "supported_formats" not in media_player_obj:
83 continue
84 for supported_format_obj in media_player_obj["supported_formats"]:
85 result.append(cast("ESPHomeSupportedAudioFormat", supported_format_obj))
86 except Exception as exc:
87 hass_prov.logger.warning(
88 "Failed to fetch diagnostics for ESPHome player: %s",
89 str(exc),
90 exc_info=exc if hass_prov.logger.isEnabledFor(logging.DEBUG) else None,
91 )
92 return result
93