/
/
/
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.mass import MusicAssistant
14from music_assistant.models.player_provider import PlayerProvider
15
16from .constants import CONF_PLAYERS
17from .helpers import get_esphome_supported_audio_formats, get_hass_media_players
18from .player import HomeAssistantPlayer
19
20if TYPE_CHECKING:
21 from hass_client.models import CompressedState, EntityStateEvent
22 from hass_client.models import Device as HassDevice
23 from hass_client.models import Entity as HassEntity
24 from hass_client.models import State as HassState
25 from music_assistant_models.config_entries import ProviderConfig
26 from music_assistant_models.provider import ProviderManifest
27
28 from music_assistant.providers.hass import HomeAssistantProvider
29
30
31class HomeAssistantPlayerProvider(PlayerProvider):
32 """Home Assistant PlayerProvider for Music Assistant."""
33
34 hass_prov: HomeAssistantProvider
35 on_unload_callbacks: list[Callable[[], None]] | None = None
36
37 def __init__(
38 self,
39 mass: MusicAssistant,
40 manifest: ProviderManifest,
41 config: ProviderConfig,
42 hass_prov: HomeAssistantProvider,
43 ) -> None:
44 """Initialize MusicProvider."""
45 super().__init__(mass, manifest, config)
46 self.hass_prov = hass_prov
47
48 async def loaded_in_mass(self) -> None:
49 """Call after the provider has been loaded."""
50 await super().loaded_in_mass()
51 player_ids = cast("list[str]", self.config.get_value(CONF_PLAYERS))
52 # prefetch the device- and entity registry
53 device_registry = {x["id"]: x for x in await self.hass_prov.hass.get_device_registry()}
54 entity_registry = {
55 x["entity_id"]: x for x in await self.hass_prov.hass.get_entity_registry()
56 }
57 # setup players from hass entities
58 async for state in get_hass_media_players(self.hass_prov):
59 if state["entity_id"] not in player_ids:
60 continue
61 await self._setup_player(state, entity_registry, device_registry)
62 # register for entity state updates
63 self.on_unload_callbacks = [
64 await self.hass_prov.hass.subscribe_entities(self._on_entity_state_update, player_ids)
65 ]
66
67 async def unload(self, is_removed: bool = False) -> None:
68 """
69 Handle unload/close of the provider.
70
71 Called when provider is deregistered (e.g. MA exiting or config reloading).
72 is_removed will be set to True when the provider is removed from the configuration.
73 """
74 if self.on_unload_callbacks:
75 for callback in self.on_unload_callbacks:
76 callback()
77
78 async def _setup_player(
79 self,
80 state: HassState,
81 entity_registry: dict[str, HassEntity],
82 device_registry: dict[str, HassDevice],
83 ) -> None:
84 """Handle setup of a Player from an hass entity."""
85 hass_device: HassDevice | None = None
86 hass_domain: str | None = None
87 # collect extra player data
88 extra_player_data: dict[str, Any] = {}
89 if entity_registry_entry := entity_registry.get(state["entity_id"]):
90 hass_device = device_registry.get(entity_registry_entry["device_id"])
91 hass_domain = entity_registry_entry["platform"]
92 extra_player_data["entity_registry_id"] = entity_registry_entry["id"]
93 extra_player_data["hass_domain"] = hass_domain
94 extra_player_data["hass_device_id"] = hass_device["id"] if hass_device else None
95 if hass_domain == "esphome":
96 # if the player is an ESPHome player, we need to check if it is a V2 player
97 # as the V2 player has different capabilities and needs different config entries
98 # The new media player component publishes its supported sample rates but that info
99 # is not exposed directly by HA, so we fetch it from the diagnostics.
100 esphome_supported_audio_formats = await get_esphome_supported_audio_formats(
101 self.hass_prov, entity_registry_entry["config_entry_id"]
102 )
103 extra_player_data["esphome_supported_audio_formats"] = (
104 esphome_supported_audio_formats
105 )
106 # collect device info
107 dev_info: dict[str, Any] = {}
108 if hass_device:
109 extra_player_data["hass_device_id"] = hass_device["id"]
110 if model := hass_device.get("model"):
111 dev_info["model"] = model
112 if manufacturer := hass_device.get("manufacturer"):
113 dev_info["manufacturer"] = manufacturer
114 if model_id := hass_device.get("model_id"):
115 dev_info["model_id"] = model_id
116 if sw_version := hass_device.get("sw_version"):
117 dev_info["software_version"] = sw_version
118 if connections := hass_device.get("connections"):
119 for key, value in connections:
120 if key == "mac":
121 dev_info["mac_address"] = value
122
123 # create the player
124 player = HomeAssistantPlayer(
125 provider=self,
126 hass=self.hass_prov.hass,
127 player_id=state["entity_id"],
128 hass_state=state,
129 dev_info=dev_info,
130 extra_player_data=extra_player_data,
131 entity_registry=entity_registry,
132 )
133 await self.mass.players.register(player)
134
135 def _on_entity_state_update(self, event: EntityStateEvent) -> None:
136 """Handle Entity State event."""
137
138 def update_player_from_state_msg(entity_id: str, state: CompressedState) -> None:
139 """Handle updating MA player with updated info in a HA CompressedState."""
140 player = cast("HomeAssistantPlayer | None", self.mass.players.get_player(entity_id))
141 if player is None:
142 # edge case - one of our subscribed entities was not available at startup
143 # and now came available - we should still set it up
144 player_ids = cast("list[str]", self.config.get_value(CONF_PLAYERS))
145 if entity_id not in player_ids:
146 return # should not happen, but guard just in case
147 self.mass.create_task(self._late_add_player(entity_id))
148 return
149 player.update_from_compressed_state(state)
150
151 if entity_additions := event.get("a"):
152 for entity_id, state in entity_additions.items():
153 update_player_from_state_msg(entity_id, state)
154 if entity_changes := event.get("c"):
155 for entity_id, state_diff in entity_changes.items():
156 if "+" not in state_diff:
157 continue
158 update_player_from_state_msg(entity_id, state_diff["+"])
159
160 async def _late_add_player(self, entity_id: str) -> None:
161 """Handle setup of Player from HA entity that became available after startup."""
162 # prefetch the device- and entity registry
163 device_registry = {x["id"]: x for x in await self.hass_prov.hass.get_device_registry()}
164 entity_registry = {
165 x["entity_id"]: x for x in await self.hass_prov.hass.get_entity_registry()
166 }
167 async for state in get_hass_media_players(self.hass_prov):
168 if state["entity_id"] != entity_id:
169 continue
170 await self._setup_player(state, entity_registry, device_registry)
171