/
/
/
1"""
2Home Assistant PlayerProvider for Music Assistant.
3
4Allows using media_player entities in HA to be used as players in MA.
5Requires the Home Assistant Plugin.
6"""
7
8from __future__ import annotations
9
10from collections.abc import Callable
11from typing import TYPE_CHECKING, Any, cast
12
13from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption
14from music_assistant_models.enums import ConfigEntryType, ProviderFeature
15
16from music_assistant.mass import MusicAssistant
17from music_assistant.models.player_provider import PlayerProvider
18from music_assistant.providers.hass import DOMAIN as HASS_DOMAIN
19
20from .constants import (
21 CONF_PLAYERS,
22 DISABLED_REASON_NATIVE_DUPLICATE,
23 DISABLED_REASON_NATIVE_INTEGRATION,
24 NATIVE_SUPPORTED_HASS_INTEGRATIONS,
25)
26from .helpers import (
27 get_esphome_supported_audio_formats,
28 get_hass_media_players,
29 get_media_player_entity_registry,
30 native_player_macs,
31 normalized_mac,
32)
33from .player import HomeAssistantPlayer
34
35if TYPE_CHECKING:
36 from hass_client.models import CompressedState, EntityStateEvent
37 from hass_client.models import Device as HassDevice
38 from hass_client.models import Entity as HassEntity
39 from hass_client.models import State as HassState
40 from music_assistant_models.config_entries import ProviderConfig
41 from music_assistant_models.provider import ProviderManifest
42
43 from music_assistant.providers.hass import HomeAssistantProvider
44
45
46class HomeAssistantPlayerProvider(PlayerProvider):
47 """Home Assistant PlayerProvider for Music Assistant."""
48
49 hass_prov: HomeAssistantProvider
50 on_unload_callbacks: list[Callable[[], None]] | None = None
51
52 def __init__(
53 self,
54 mass: MusicAssistant,
55 manifest: ProviderManifest,
56 config: ProviderConfig,
57 hass_prov: HomeAssistantProvider,
58 ) -> None:
59 """Initialize MusicProvider."""
60 supported_features = {ProviderFeature.REMOVE_PLAYER}
61 super().__init__(mass, manifest, config, supported_features=supported_features)
62 self.hass_prov = hass_prov
63
64 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
65 """Return Config entries to configure this provider."""
66 hass_prov = cast("HomeAssistantProvider|None", self.mass.get_provider(HASS_DOMAIN))
67 player_entities: list[ConfigValueOption] = []
68 if hass_prov and hass_prov.hass.connected:
69 # entities that are already imported must stay selectable,
70 # so editing the provider config never drops them
71 selected_players = self._selected_players()
72 device_registry = await hass_prov.get_device_registry()
73 native_macs = native_player_macs(self.mass)
74 entity_registry = await get_media_player_entity_registry(hass_prov)
75 async for state, entity_registry_entry in get_hass_media_players(
76 hass_prov, entity_registry
77 ):
78 entity_id = state["entity_id"]
79 name = f"{state['attributes']['friendly_name']} ({entity_id})"
80 disabled_reason: str | None = None
81 if entity_id not in selected_players:
82 disabled_reason = self._native_support_reason(
83 entity_registry_entry, device_registry, native_macs
84 )
85 player_entities.append(
86 ConfigValueOption(
87 entity_id,
88 title=name,
89 disabled=disabled_reason is not None,
90 disabled_reason=disabled_reason,
91 )
92 )
93 return (
94 ConfigEntry(
95 key=CONF_PLAYERS,
96 type=ConfigEntryType.STRING,
97 multi_value=True,
98 required=True,
99 # the provider is added before any entity can be picked, so it starts out
100 # with an empty selection and the players are chosen in its options
101 default_value=[],
102 options=player_entities,
103 ),
104 )
105
106 async def loaded_in_mass(self) -> None:
107 """Call after the provider has been loaded."""
108 await super().loaded_in_mass()
109 player_ids = cast("list[str]", self.config.get_value(CONF_PLAYERS))
110 # prefetch the device- and entity registry
111 device_registry = await self.hass_prov.get_device_registry()
112 entity_registry = await get_media_player_entity_registry(self.hass_prov)
113 # setup players from hass entities
114 async for state, _ in get_hass_media_players(self.hass_prov, entity_registry):
115 if state["entity_id"] not in player_ids:
116 continue
117 await self._setup_player(state, entity_registry, device_registry)
118 # register for entity state updates
119 self.on_unload_callbacks = [
120 await self.hass_prov.hass.subscribe_entities(self._on_entity_state_update, player_ids)
121 ]
122 # cleanup any players that are no longer in the config
123 for player_conf in await self.mass.config.get_player_configs(
124 provider=self.instance_id, include_unavailable=True, include_disabled=True
125 ):
126 if player_conf.player_id not in player_ids:
127 await self.mass.players.remove(player_conf.player_id)
128
129 async def unload(self, is_removed: bool = False) -> None:
130 """
131 Handle unload/close of the provider.
132
133 Called when provider is deregistered (e.g. MA exiting or config reloading).
134 is_removed will be set to True when the provider is removed from the configuration.
135 """
136 if self.on_unload_callbacks:
137 for callback in self.on_unload_callbacks:
138 callback()
139
140 async def remove_player(self, player_id: str) -> None:
141 """Remove a player."""
142 player_ids = cast("list[str]", self.config.get_value(CONF_PLAYERS))
143 if player_id in player_ids:
144 player_ids.remove(player_id)
145 self.mass.config.set_raw_provider_config_value(
146 self.instance_id, CONF_PLAYERS, player_ids
147 )
148 await self.mass.players.unregister(player_id, True)
149
150 async def _setup_player(
151 self,
152 state: HassState,
153 entity_registry: dict[str, HassEntity],
154 device_registry: dict[str, HassDevice],
155 ) -> None:
156 """Handle setup of a Player from an hass entity."""
157 hass_device: HassDevice | None = None
158 hass_domain: str | None = None
159 # collect extra player data
160 extra_player_data: dict[str, Any] = {}
161 if entity_registry_entry := entity_registry.get(state["entity_id"]):
162 hass_device = device_registry.get(entity_registry_entry["device_id"])
163 hass_domain = entity_registry_entry["platform"]
164 extra_player_data["entity_registry_id"] = entity_registry_entry["id"]
165 extra_player_data["hass_domain"] = hass_domain
166 extra_player_data["hass_device_id"] = hass_device["id"] if hass_device else None
167 if hass_domain == "esphome":
168 # if the player is an ESPHome player, we need to check if it is a V2 player
169 # as the V2 player has different capabilities and needs different config entries
170 # The new media player component publishes its supported sample rates but that info
171 # is not exposed directly by HA, so we fetch it from the diagnostics.
172 esphome_supported_audio_formats = await get_esphome_supported_audio_formats(
173 self.hass_prov, entity_registry_entry["config_entry_id"]
174 )
175 extra_player_data["esphome_supported_audio_formats"] = (
176 esphome_supported_audio_formats
177 )
178 # collect device info
179 dev_info: dict[str, Any] = {}
180 if hass_device:
181 extra_player_data["hass_device_id"] = hass_device["id"]
182 if model := hass_device.get("model"):
183 dev_info["model"] = model
184 if manufacturer := hass_device.get("manufacturer"):
185 dev_info["manufacturer"] = manufacturer
186 if model_id := hass_device.get("model_id"):
187 dev_info["model_id"] = model_id
188 if sw_version := hass_device.get("sw_version"):
189 dev_info["software_version"] = sw_version
190 if connections := hass_device.get("connections"):
191 for key, value in connections:
192 if key == "mac":
193 dev_info["mac_address"] = value
194
195 # create the player
196 player = HomeAssistantPlayer(
197 provider=self,
198 hass=self.hass_prov.hass,
199 player_id=state["entity_id"],
200 hass_state=state,
201 dev_info=dev_info,
202 extra_player_data=extra_player_data,
203 entity_registry=entity_registry,
204 )
205 await self.mass.players.register(player)
206
207 def _on_entity_state_update(self, event: EntityStateEvent) -> None:
208 """Handle Entity State event."""
209
210 def update_player_from_state_msg(entity_id: str, state: CompressedState) -> None:
211 """Handle updating MA player with updated info in a HA CompressedState."""
212 player = cast("HomeAssistantPlayer | None", self.mass.players.get_player(entity_id))
213 if player is None:
214 # edge case - one of our subscribed entities was not available at startup
215 # and now came available - we should still set it up
216 player_ids = cast("list[str]", self.config.get_value(CONF_PLAYERS))
217 if entity_id not in player_ids:
218 return # should not happen, but guard just in case
219 self.mass.create_task(self._late_add_player(entity_id))
220 return
221 player.update_from_compressed_state(state)
222
223 if entity_additions := event.get("a"):
224 for entity_id, state in entity_additions.items():
225 update_player_from_state_msg(entity_id, state)
226 if entity_changes := event.get("c"):
227 for entity_id, state_diff in entity_changes.items():
228 if "+" not in state_diff:
229 continue
230 update_player_from_state_msg(entity_id, state_diff["+"])
231
232 async def _late_add_player(self, entity_id: str) -> None:
233 """Handle setup of Player from HA entity that became available after startup."""
234 # prefetch the device- and entity registry
235 device_registry = await self.hass_prov.get_device_registry()
236 entity_registry = await get_media_player_entity_registry(self.hass_prov)
237 async for state, _ in get_hass_media_players(self.hass_prov, entity_registry):
238 if state["entity_id"] != entity_id:
239 continue
240 await self._setup_player(state, entity_registry, device_registry)
241
242 def _selected_players(self) -> set[str]:
243 """Return the entity ids currently selected as players (stored config)."""
244 selected: set[str] = set()
245 if isinstance(stored_players := self.get_config_value(CONF_PLAYERS), list):
246 selected.update(cast("list[str]", stored_players))
247 return selected
248
249 @staticmethod
250 def _native_support_reason(
251 entity_registry_entry: HassEntity | None,
252 device_registry: dict[str, HassDevice],
253 native_macs: set[str],
254 ) -> str | None:
255 """Return why the entity can not be imported (natively supported), or None if it can."""
256 if entity_registry_entry is None:
257 return None
258 if entity_registry_entry["platform"] in NATIVE_SUPPORTED_HASS_INTEGRATIONS:
259 return DISABLED_REASON_NATIVE_INTEGRATION
260 device = device_registry.get(entity_registry_entry["device_id"] or "")
261 if device is not None and any(
262 normalized_mac(connection[1]) in native_macs
263 for connection in device.get("connections", [])
264 if len(connection) == 2 and connection[0] == "mac"
265 ):
266 return DISABLED_REASON_NATIVE_DUPLICATE
267 return None
268