/
/
/
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
14from functools import partial
15from typing import TYPE_CHECKING, cast
16
17import shortuuid
18from hass_client import HomeAssistantClient
19from hass_client.exceptions import BaseHassClientError
20from hass_client.utils import (
21 base_url,
22 get_auth_url,
23 get_long_lived_token,
24 get_token,
25 get_websocket_url,
26)
27from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
28from music_assistant_models.enums import ConfigEntryType, ProviderFeature
29from music_assistant_models.errors import LoginFailed, SetupFailedError
30from music_assistant_models.player_control import PlayerControl
31
32from music_assistant.constants import MASS_LOGO_ONLINE, VERBOSE_LOG_LEVEL
33from music_assistant.helpers.auth import AuthenticationHelper
34from music_assistant.helpers.util import try_parse_int
35from music_assistant.models.plugin import PluginProvider
36
37from .constants import OFF_STATES, MediaPlayerEntityFeature
38
39if TYPE_CHECKING:
40 from hass_client.models import CompressedState, Device, EntityStateEvent
41 from music_assistant_models.config_entries import ProviderConfig
42 from music_assistant_models.provider import ProviderManifest
43
44 from music_assistant.mass import MusicAssistant
45 from music_assistant.models import ProviderInstanceType
46
47DOMAIN = "hass"
48CONF_URL = "url"
49CONF_AUTH_TOKEN = "token"
50CONF_ACTION_AUTH = "auth"
51CONF_VERIFY_SSL = "verify_ssl"
52CONF_POWER_CONTROLS = "power_controls"
53CONF_MUTE_CONTROLS = "mute_controls"
54CONF_VOLUME_CONTROLS = "volume_controls"
55
56SUPPORTED_FEATURES: set[ProviderFeature] = (
57 set()
58) # we don't have any special supported features (yet)
59
60
61async def setup(
62 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
63) -> ProviderInstanceType:
64 """Initialize provider(instance) with given configuration."""
65 return HomeAssistantProvider(mass, manifest, config, SUPPORTED_FEATURES)
66
67
68async def get_config_entries(
69 mass: MusicAssistant,
70 instance_id: str | None = None,
71 action: str | None = None,
72 values: dict[str, ConfigValueType] | None = None,
73) -> tuple[ConfigEntry, ...]:
74 """
75 Return Config entries to setup this provider.
76
77 instance_id: id of an existing provider instance (None if new instance setup).
78 action: [optional] action key called from config entries UI.
79 values: the (intermediate) raw values for config entries sent with the action.
80 """
81 # config flow auth action/step (authenticate button clicked)
82 if action == CONF_ACTION_AUTH and values:
83 hass_url = values[CONF_URL]
84 async with AuthenticationHelper(mass, str(values["session_id"])) as auth_helper:
85 client_id = base_url(auth_helper.callback_url)
86 auth_url = get_auth_url(
87 hass_url,
88 auth_helper.callback_url,
89 client_id=client_id,
90 state=values["session_id"],
91 )
92 result = await auth_helper.authenticate(auth_url)
93 if result["state"] != values["session_id"]:
94 msg = "session id mismatch"
95 raise LoginFailed(msg)
96 # get access token after auth was a success
97 token_details = await get_token(hass_url, result["code"], client_id=client_id)
98 # register for a long lived token
99 long_lived_token = await get_long_lived_token(
100 hass_url,
101 token_details["access_token"],
102 client_name=f"Music Assistant {shortuuid.random(6)}",
103 client_icon=MASS_LOGO_ONLINE,
104 lifespan=365 * 2,
105 )
106 # set the retrieved token on the values object to pass along
107 values[CONF_AUTH_TOKEN] = long_lived_token
108
109 base_entries: tuple[ConfigEntry, ...]
110 if mass.running_as_hass_addon:
111 # on supervisor, we use the internal url
112 # token set to None for auto retrieval
113 base_entries = (
114 ConfigEntry(
115 key=CONF_URL,
116 type=ConfigEntryType.STRING,
117 label=CONF_URL,
118 required=True,
119 default_value="http://supervisor/core/api",
120 value="http://supervisor/core/api",
121 hidden=True,
122 ),
123 ConfigEntry(
124 key=CONF_AUTH_TOKEN,
125 type=ConfigEntryType.STRING,
126 label=CONF_AUTH_TOKEN,
127 required=False,
128 default_value=None,
129 value=None,
130 hidden=True,
131 ),
132 ConfigEntry(
133 key=CONF_VERIFY_SSL,
134 type=ConfigEntryType.BOOLEAN,
135 label=CONF_VERIFY_SSL,
136 required=False,
137 default_value=False,
138 hidden=True,
139 ),
140 )
141 else:
142 # manual configuration
143 base_entries = (
144 ConfigEntry(
145 key=CONF_URL,
146 type=ConfigEntryType.STRING,
147 label="URL",
148 required=True,
149 description="URL to your Home Assistant instance (e.g. http://192.168.1.1:8123)",
150 value=cast("str", values.get(CONF_URL)) if values else None,
151 ),
152 ConfigEntry(
153 key=CONF_ACTION_AUTH,
154 type=ConfigEntryType.ACTION,
155 label="(re)Authenticate Home Assistant",
156 description="Authenticate to your home assistant "
157 "instance and generate the long lived token.",
158 action=CONF_ACTION_AUTH,
159 depends_on=CONF_URL,
160 required=False,
161 ),
162 ConfigEntry(
163 key=CONF_AUTH_TOKEN,
164 type=ConfigEntryType.SECURE_STRING,
165 label="Authentication token for HomeAssistant",
166 description="You can either paste a Long Lived Token here manually or use the "
167 "'authenticate' button to generate a token for you with logging in.",
168 depends_on=CONF_URL,
169 value=cast("str", values.get(CONF_AUTH_TOKEN)) if values else None,
170 advanced=True,
171 ),
172 ConfigEntry(
173 key=CONF_VERIFY_SSL,
174 type=ConfigEntryType.BOOLEAN,
175 label="Verify SSL",
176 required=False,
177 description="Whether or not to verify the certificate of SSL/TLS connections.",
178 advanced=True,
179 default_value=True,
180 ),
181 )
182
183 # append player controls entries (if we have an active instance)
184 if instance_id and (hass_prov := mass.get_provider(instance_id)) and hass_prov.available:
185 hass_prov = cast("HomeAssistantProvider", hass_prov)
186 return (*base_entries, *(await _get_player_control_config_entries(hass_prov.hass)))
187
188 return (
189 *base_entries,
190 ConfigEntry(
191 key=CONF_POWER_CONTROLS,
192 type=ConfigEntryType.STRING,
193 multi_value=True,
194 label=CONF_POWER_CONTROLS,
195 default_value=[],
196 ),
197 ConfigEntry(
198 key=CONF_VOLUME_CONTROLS,
199 type=ConfigEntryType.STRING,
200 multi_value=True,
201 label=CONF_VOLUME_CONTROLS,
202 default_value=[],
203 ),
204 ConfigEntry(
205 key=CONF_MUTE_CONTROLS,
206 type=ConfigEntryType.STRING,
207 multi_value=True,
208 label=CONF_MUTE_CONTROLS,
209 default_value=[],
210 ),
211 )
212
213
214async def _get_player_control_config_entries(hass: HomeAssistantClient) -> tuple[ConfigEntry, ...]:
215 """Return all HA state objects for (valid) media_player entities."""
216 all_power_entities: list[ConfigValueOption] = []
217 all_mute_entities: list[ConfigValueOption] = []
218 all_volume_entities: list[ConfigValueOption] = []
219 # collect all entities that are usable for player controls
220 if not hass.connected:
221 return ()
222 for state in await hass.get_states():
223 entity_platform = state["entity_id"].split(".")[0]
224 if "friendly_name" not in state["attributes"]:
225 name = state["entity_id"]
226 else:
227 name = f"{state['attributes']['friendly_name']} ({state['entity_id']})"
228
229 if entity_platform in ("switch", "input_boolean"):
230 # simple on/off controls are suitable as power and mute controls
231 all_power_entities.append(ConfigValueOption(name, state["entity_id"]))
232 all_mute_entities.append(ConfigValueOption(name, state["entity_id"]))
233 continue
234 if entity_platform in ("number", "input_number"):
235 # number and input_number are very similar, both are suitable for volume control
236 all_volume_entities.append(ConfigValueOption(name, state["entity_id"]))
237 continue
238
239 # media player can be used as control, depending on features
240 if entity_platform != "media_player":
241 continue
242 if "mass_player_type" in state["attributes"]:
243 # filter out mass players
244 continue
245 supported_features = MediaPlayerEntityFeature(state["attributes"]["supported_features"])
246 if MediaPlayerEntityFeature.VOLUME_MUTE in supported_features:
247 all_mute_entities.append(ConfigValueOption(name, state["entity_id"]))
248 if MediaPlayerEntityFeature.VOLUME_SET in supported_features:
249 all_volume_entities.append(ConfigValueOption(name, state["entity_id"]))
250 if (
251 MediaPlayerEntityFeature.TURN_ON in supported_features
252 and MediaPlayerEntityFeature.TURN_OFF in supported_features
253 ):
254 all_power_entities.append(ConfigValueOption(name, state["entity_id"]))
255 all_power_entities.sort(key=lambda x: x.title)
256 all_mute_entities.sort(key=lambda x: x.title)
257 all_volume_entities.sort(key=lambda x: x.title)
258 return (
259 ConfigEntry(
260 key=CONF_POWER_CONTROLS,
261 type=ConfigEntryType.STRING,
262 multi_value=True,
263 label="Player Power Control entities",
264 required=True,
265 options=all_power_entities,
266 default_value=[],
267 description="Specify which Home Assistant entities you "
268 "like to import as player Power controls in Music Assistant.",
269 category="player_controls",
270 ),
271 ConfigEntry(
272 key=CONF_VOLUME_CONTROLS,
273 type=ConfigEntryType.STRING,
274 multi_value=True,
275 label="Player Volume Control entities",
276 required=True,
277 options=all_volume_entities,
278 default_value=[],
279 description="Specify which Home Assistant entities you "
280 "like to import as player Volume controls in Music Assistant.",
281 category="player_controls",
282 ),
283 ConfigEntry(
284 key=CONF_MUTE_CONTROLS,
285 type=ConfigEntryType.STRING,
286 multi_value=True,
287 label="Player Mute Control entities",
288 required=True,
289 options=all_mute_entities,
290 default_value=[],
291 description="Specify which Home Assistant entities you "
292 "like to import as player Mute controls in Music Assistant.",
293 category="player_controls",
294 ),
295 )
296
297
298class HomeAssistantProvider(PluginProvider):
299 """Home Assistant Plugin for Music Assistant."""
300
301 hass: HomeAssistantClient
302 _listen_task: asyncio.Task[None] | None = None
303 _player_controls: dict[str, PlayerControl] | None = None
304
305 async def handle_async_init(self) -> None:
306 """Handle async initialization of the plugin."""
307 self._player_controls = {}
308 url = get_websocket_url(self.config.get_value(CONF_URL))
309 token = self.config.get_value(CONF_AUTH_TOKEN)
310 logging.getLogger("hass_client").setLevel(self.logger.level + 10)
311 ssl = bool(self.config.get_value(CONF_VERIFY_SSL))
312 http_session = self.mass.http_session if ssl else self.mass.http_session_no_ssl
313 self.hass = HomeAssistantClient(url, token, http_session)
314 try:
315 await self.hass.connect()
316 except BaseHassClientError as err:
317 err_msg = str(err) or err.__class__.__name__
318 raise SetupFailedError(err_msg) from err
319 self._listen_task = self.mass.create_task(self._hass_listener())
320
321 async def loaded_in_mass(self) -> None:
322 """Call after the provider has been loaded."""
323 await self._register_player_controls()
324
325 async def unload(self, is_removed: bool = False) -> None:
326 """
327 Handle unload/close of the provider.
328
329 Called when provider is deregistered (e.g. MA exiting or config reloading).
330 """
331 # unregister all player controls
332 if self._player_controls:
333 for entity_id in self._player_controls:
334 self.mass.players.remove_player_control(entity_id)
335 if self._listen_task and not self._listen_task.done():
336 self._listen_task.cancel()
337 await self.hass.disconnect()
338
339 async def _hass_listener(self) -> None:
340 """Start listening on the HA websockets."""
341 try:
342 # start listening will block until the connection is lost/closed
343 await self.hass.start_listening()
344 except BaseHassClientError as err:
345 self.logger.warning("Connection to HA lost due to error: %s", err)
346 self.logger.info("Connection to HA lost. Connection will be automatically retried later.")
347 # schedule a reload of the provider
348 self.available = False
349 self.mass.call_later(5, self.mass.load_provider, self.instance_id, allow_retry=True)
350
351 def _on_entity_state_update(self, event: EntityStateEvent) -> None:
352 """Handle Entity State event."""
353 if entity_additions := event.get("a"):
354 for entity_id, state in entity_additions.items():
355 self._update_control_from_state_msg(entity_id, state)
356 if entity_changes := event.get("c"):
357 for entity_id, state_diff in entity_changes.items():
358 if "+" not in state_diff:
359 continue
360 self._update_control_from_state_msg(entity_id, state_diff["+"])
361
362 async def _register_player_controls(self) -> None:
363 """Register all player controls."""
364 power_controls = cast("list[str]", self.config.get_value(CONF_POWER_CONTROLS))
365 mute_controls = cast("list[str]", self.config.get_value(CONF_MUTE_CONTROLS))
366 volume_controls = cast("list[str]", self.config.get_value(CONF_VOLUME_CONTROLS))
367 control_entity_ids: set[str] = {
368 *power_controls,
369 *mute_controls,
370 *volume_controls,
371 }
372 hass_states = {
373 state["entity_id"]: state
374 for state in await self.hass.get_states()
375 if state["entity_id"] in control_entity_ids
376 }
377 assert self._player_controls is not None # for type checking
378 for entity_id in control_entity_ids:
379 entity_platform = entity_id.split(".")[0]
380 hass_state = hass_states.get(entity_id)
381 if hass_state and (friendly_name := hass_state["attributes"].get("friendly_name")):
382 name = f"{friendly_name} ({entity_id})"
383 else:
384 name = entity_id
385 control = PlayerControl(
386 id=entity_id,
387 provider=self.instance_id,
388 name=name,
389 )
390 if entity_id in power_controls:
391 control.supports_power = True
392 control.power_state = hass_state["state"] not in OFF_STATES if hass_state else False
393 control.power_on = partial(self._handle_player_control_power_on, entity_id)
394 control.power_off = partial(self._handle_player_control_power_off, entity_id)
395 if entity_id in volume_controls:
396 control.supports_volume = True
397 if not hass_state:
398 control.volume_level = 0
399 elif entity_platform == "media_player":
400 control.volume_level = int(
401 hass_state["attributes"].get("volume_level", 0) * 100
402 )
403 else:
404 control.volume_level = try_parse_int(hass_state["state"]) or 0
405 control.volume_set = partial(self._handle_player_control_volume_set, entity_id)
406 if entity_id in mute_controls:
407 control.supports_mute = True
408 if not hass_state:
409 control.volume_muted = False
410 elif entity_platform == "media_player":
411 control.volume_muted = hass_state["attributes"].get("volume_muted")
412 elif hass_state:
413 control.volume_muted = hass_state["state"] not in OFF_STATES
414 else:
415 control.volume_muted = False
416 control.mute_set = partial(self._handle_player_control_mute_set, entity_id)
417 self._player_controls[entity_id] = control
418 await self.mass.players.register_player_control(control)
419 # register for entity state updates
420 await self.hass.subscribe_entities(self._on_entity_state_update, list(control_entity_ids))
421
422 async def _handle_player_control_power_on(self, entity_id: str) -> None:
423 """Handle powering on the playercontrol."""
424 await self.hass.call_service(
425 domain="homeassistant",
426 service="turn_on",
427 target={"entity_id": entity_id},
428 )
429
430 async def _handle_player_control_power_off(self, entity_id: str) -> None:
431 """Handle powering off the playercontrol."""
432 await self.hass.call_service(
433 domain="homeassistant",
434 service="turn_off",
435 target={"entity_id": entity_id},
436 )
437
438 async def _handle_player_control_mute_set(self, entity_id: str, muted: bool) -> None:
439 """Handle muting the playercontrol."""
440 if entity_id.startswith("media_player."):
441 await self.hass.call_service(
442 domain="media_player",
443 service="volume_mute",
444 service_data={"is_volume_muted": muted},
445 target={"entity_id": entity_id},
446 )
447 else:
448 await self.hass.call_service(
449 domain="homeassistant",
450 service="turn_off" if muted else "turn_on",
451 target={"entity_id": entity_id},
452 )
453
454 async def _handle_player_control_volume_set(self, entity_id: str, volume_level: int) -> None:
455 """Handle setting volume on the playercontrol."""
456 domain = entity_id.split(".", 1)[0]
457
458 if domain == "media_player":
459 await self.hass.call_service(
460 domain=domain,
461 service="volume_set",
462 service_data={"volume_level": volume_level / 100},
463 target={"entity_id": entity_id},
464 )
465 return
466
467 # At this point, `set_value` will work for both `number` or `input_number`
468 await self.hass.call_service(
469 domain=domain,
470 service="set_value",
471 target={"entity_id": entity_id},
472 service_data={"value": volume_level},
473 )
474
475 async def get_device_by_connection(
476 self,
477 connection_value: str,
478 connection_type: str = "mac",
479 ) -> Device | None:
480 """
481 Get device details from Home Assistant by connection type and value.
482
483 :param connection_value: The connection value (e.g. MAC address).
484 :param connection_type: The connection type (default: 'mac').
485 """
486 devices = await self.hass.get_device_registry()
487 for device in devices:
488 for connection in device.get("connections", []):
489 if (
490 len(connection) == 2
491 and connection[0] == connection_type
492 and connection[1].lower() == connection_value.lower()
493 ):
494 return device
495 return None
496
497 def _update_control_from_state_msg(self, entity_id: str, state: CompressedState) -> None:
498 """Update PlayerControl from state(update) message."""
499 if self._player_controls is None:
500 return
501 if not (player_control := self._player_controls.get(entity_id)):
502 return
503 entity_platform = entity_id.split(".")[0]
504 if "s" in state:
505 # state changed
506 if player_control.supports_power:
507 player_control.power_state = state["s"] not in OFF_STATES
508 if player_control.supports_mute and entity_platform != "media_player":
509 player_control.volume_muted = state["s"] not in OFF_STATES
510 if player_control.supports_volume and entity_platform != "media_player":
511 player_control.volume_level = try_parse_int(state["s"]) or 0
512 if "a" in state and (attributes := state["a"]):
513 if player_control.supports_volume:
514 if entity_platform == "media_player":
515 player_control.volume_level = int(attributes.get("volume_level", 0) * 100)
516 else:
517 player_control.volume_level = try_parse_int(attributes.get("value")) or 0
518 if player_control.supports_mute and entity_platform == "media_player":
519 player_control.volume_muted = attributes.get("volume_muted")
520 self.mass.players.update_player_control(entity_id)
521
522 async def get_user_details(self, ha_user_id: str) -> tuple[str | None, str | None, str | None]:
523 """
524 Get user username, display name and avatar URL from Home Assistant.
525
526 Looks up the user in config/auth/list for username, and the person entity
527 for display name and picture URL.
528
529 :param ha_user_id: Home Assistant user ID.
530 :return: Tuple of (username, display_name, avatar_url) or all None if not found.
531 """
532 try:
533 username: str | None = None
534 display_name: str | None = None
535 avatar_url: str | None = None
536
537 # Get username from config/auth/list (admin endpoint, we have admin access)
538 try:
539 users = await self.hass.send_command("config/auth/list")
540 for user in users or []:
541 if user.get("id") == ha_user_id:
542 username = user.get("username")
543 # Also get name as fallback display name
544 if not display_name:
545 display_name = user.get("name")
546 break
547 except Exception as err:
548 self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA user list: %s", err)
549
550 # Get external URL for building avatar URL
551 ha_url: str | None = None
552 try:
553 network_urls = await self.hass.send_command("network/url")
554 if network_urls:
555 ha_url = network_urls.get("external") or network_urls.get("internal")
556 except Exception as err:
557 self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA network URLs: %s", err)
558
559 # Find person linked to this HA user ID for display name and avatar
560 try:
561 persons = await self.hass.send_command("person/list")
562 # person/list returns {storage: [...], config: [...]}
563 all_persons = (persons.get("storage") or []) + (persons.get("config") or [])
564 for person in all_persons:
565 if person.get("user_id") == ha_user_id:
566 # Person name takes priority for display name
567 if person_name := person.get("name"):
568 display_name = person_name
569 if (person_picture := person.get("picture")) and ha_url:
570 avatar_url = f"{ha_url.rstrip('/')}{person_picture}"
571 break
572 except Exception as err:
573 self.logger.log(VERBOSE_LOG_LEVEL, "Failed to get HA person details: %s", err)
574
575 self.logger.log(
576 VERBOSE_LOG_LEVEL,
577 "get_user_details for %s: username=%s, display_name=%s, avatar_url=%s",
578 ha_user_id,
579 username,
580 display_name,
581 avatar_url,
582 )
583 return username, display_name, avatar_url
584 except Exception as err:
585 self.logger.warning("Failed to get HA user details: %s", err)
586 return None, None, None
587