music-assistant-server

1"""Home Assistant Player implementation."""
2
3from __future__ import annotations
4
5import time
6from typing import TYPE_CHECKING, Any, cast
7
8from hass_client.exceptions import FailedCommand
9from music_assistant_models.enums import (
10    IdentifierType,
11    ImageType,
12    MediaType,
13    PlaybackState,
14    PlayerFeature,
15)
16from music_assistant_models.media_items import MediaItemImage
17
18from music_assistant.constants import (
19    CONF_ENTRY_ENABLE_ICY_METADATA_HIDDEN,
20    CONF_ENTRY_HTTP_PROFILE_FORCED_2,
21    CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3,
22    EXTERNAL_PAUSE_IDLE_TIMEOUT,
23    HIDDEN_ANNOUNCE_VOLUME_CONFIG_ENTRIES,
24    create_output_codec_config_entry,
25)
26from music_assistant.helpers.datetime import from_iso_string
27from music_assistant.models.player import DeviceInfo, Player, PlayerMedia, PlayerSource
28from music_assistant.models.player_provider import PlayerProvider
29from music_assistant.providers.hass.constants import (
30    OFF_STATES,
31    UNAVAILABLE_STATES,
32    MediaPlayerEntityFeature,
33    StateMap,
34    parse_supported_features,
35)
36
37from .constants import CONF_ENTRY_WARN_HASS_INTEGRATION, NATIVE_SUPPORTED_HASS_INTEGRATIONS
38from .helpers import ESPHomeSupportedAudioFormat, native_player_macs, normalized_mac
39
40if TYPE_CHECKING:
41    from hass_client import HomeAssistantClient
42    from hass_client.models import CompressedState
43    from hass_client.models import Entity as HassEntity
44    from hass_client.models import State as HassState
45    from music_assistant_models.config_entries import ConfigEntry
46
47    from .provider import HomeAssistantPlayerProvider
48
49
50DEFAULT_PLAYER_CONFIG_ENTRIES = (CONF_ENTRY_OUTPUT_CODEC_DEFAULT_MP3,)
51
52
53class HomeAssistantPlayer(Player):
54    """Home Assistant Player implementation."""
55
56    # the wrapped entity keeps reporting an abandoned external session as paused, and
57    # Home Assistant pushes no event when it goes stale.
58    _attr_external_pause_idle_timeout = EXTERNAL_PAUSE_IDLE_TIMEOUT
59
60    def __init__(
61        self,
62        provider: PlayerProvider,
63        hass: HomeAssistantClient,
64        player_id: str,
65        hass_state: HassState,
66        dev_info: dict[str, Any],
67        extra_player_data: dict[str, Any],
68        entity_registry: dict[str, HassEntity],
69    ) -> None:
70        """Initialize the Home Assistant Player."""
71        super().__init__(provider, player_id)
72        self.hass = hass
73        self.hass_state = hass_state
74        self._extra_data = extra_player_data
75        # Set base attributes from Home Assistant state
76        self._attr_available = hass_state["state"] not in UNAVAILABLE_STATES
77        self._attr_device_info = DeviceInfo(
78            model=dev_info.get("model", ""),
79            manufacturer=dev_info.get("manufacturer", ""),
80            software_version=dev_info.get("software_version"),
81        )
82        if mac_address := dev_info.get("mac_address"):
83            self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac_address)
84        self._attr_playback_state = StateMap.get(hass_state["state"], PlaybackState.IDLE)
85        # Work out supported features
86        self._attr_supported_features = {PlayerFeature.PLAY_MEDIA}
87        hass_supported_features = parse_supported_features(
88            hass_state["attributes"].get("supported_features"), player_id, self.logger
89        )
90        if MediaPlayerEntityFeature.VOLUME_SET in hass_supported_features:
91            self._attr_supported_features.add(PlayerFeature.VOLUME_SET)
92        if MediaPlayerEntityFeature.VOLUME_MUTE in hass_supported_features:
93            self._attr_supported_features.add(PlayerFeature.VOLUME_MUTE)
94        if MediaPlayerEntityFeature.MEDIA_ANNOUNCE in hass_supported_features:
95            self._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
96        hass_domain = extra_player_data.get("hass_domain")
97        if hass_domain and MediaPlayerEntityFeature.GROUPING in hass_supported_features:
98            self._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
99            self._attr_can_group_with = {
100                x["entity_id"]
101                for x in entity_registry.values()
102                if x["entity_id"].startswith("media_player") and x["platform"] == hass_domain
103            }
104        if (
105            MediaPlayerEntityFeature.TURN_ON in hass_supported_features
106            and MediaPlayerEntityFeature.TURN_OFF in hass_supported_features
107        ):
108            self._attr_supported_features.add(PlayerFeature.POWER)
109            self._attr_powered = hass_state["state"] not in OFF_STATES
110
111        self.extra_data["hass_supported_features"] = hass_supported_features
112        self._hass_attributes: dict[str, Any] = {}
113
114        # Add External source to support next/prev commands when playing external content
115        self._attr_source_list.append(
116            PlayerSource(
117                id="External",
118                name="External Source",
119                passive=True,
120            )
121        )
122        # Set dynamic features (PAUSE, NEXT_PREVIOUS, SEEK) via shared helper
123        self._update_hass_features(hass_supported_features)
124
125        # Derive supported sample rates from ESPHome's reported formats when available
126        esphome_formats: list[ESPHomeSupportedAudioFormat] | None = self.extra_data.get(
127            "esphome_supported_audio_formats"
128        )
129        if esphome_formats:
130            rates = sorted(
131                {(fmt["sample_rate"], (fmt["sample_bytes"] or 2) * 8) for fmt in esphome_formats}
132            )
133            self._attr_supported_sample_rates = rates or [(48000, 16)]
134
135        self._update_attributes(hass_state["attributes"])
136
137    @property
138    def requires_flow_mode(self) -> bool:
139        """Return if the player requires flow mode."""
140        # hass media players are a hot mess so play it safe and always use flow mode
141        return True
142
143    async def get_config_entries(self) -> list[ConfigEntry]:
144        """Return all (provider/player specific) Config Entries for the player."""
145        base_entries = [*DEFAULT_PLAYER_CONFIG_ENTRIES]
146        # add alert if the player (type) is also supported by a native MA provider
147        if (
148            self.extra_data.get("hass_domain") in NATIVE_SUPPORTED_HASS_INTEGRATIONS
149            or self._has_native_duplicate()
150        ):
151            base_entries = [CONF_ENTRY_WARN_HASS_INTEGRATION, *base_entries]
152        supported_formats: list[ESPHomeSupportedAudioFormat] | None = self.extra_data.get(
153            "esphome_supported_audio_formats"
154        )
155        if supported_formats:
156            # optimized config for new ESPHome mediaplayer
157            # sort on purpose field, so we prefer the media pipeline
158            # but allows fallback to announcements pipeline if no media pipeline is available
159            supported_formats.sort(key=lambda x: x["purpose"])
160            codec = supported_formats[0]["format"] if supported_formats else None
161
162            config_entries = [
163                *base_entries,
164                # New ESPHome mediaplayer (used in Voice PE) uses FLAC 48khz/16 bits
165                CONF_ENTRY_HTTP_PROFILE_FORCED_2,
166            ]
167
168            if codec is not None:
169                config_entries.append(create_output_codec_config_entry(True, codec))
170
171            config_entries.extend(
172                [
173                    CONF_ENTRY_ENABLE_ICY_METADATA_HIDDEN,
174                    # although the Voice PE supports announcements,
175                    # it does not support volume for announcements
176                    *HIDDEN_ANNOUNCE_VOLUME_CONFIG_ENTRIES,
177                ]
178            )
179
180            return config_entries
181
182        return base_entries
183
184    async def play(self) -> None:
185        """Handle PLAY command on the player."""
186        await self.hass.call_service(
187            domain="media_player",
188            service="media_play",
189            target={"entity_id": self.player_id},
190        )
191
192    async def pause(self) -> None:
193        """Handle PAUSE command on the player."""
194        await self.hass.call_service(
195            domain="media_player",
196            service="media_pause",
197            target={"entity_id": self.player_id},
198        )
199
200    async def stop(self) -> None:
201        """Send STOP command to player."""
202        try:
203            await self.hass.call_service(
204                domain="media_player",
205                service="media_stop",
206                target={"entity_id": self.player_id},
207            )
208        except FailedCommand as exc:
209            # some HA players do not support STOP
210            if "does not support" not in str(exc):
211                raise
212            if PlayerFeature.PAUSE in self.supported_features:
213                await self.pause()
214        finally:
215            self._attr_current_media = None
216            self.update_state()
217
218    async def volume_set(self, volume_level: int) -> None:
219        """Handle VOLUME_SET command on the player."""
220        await self.hass.call_service(
221            domain="media_player",
222            service="volume_set",
223            target={"entity_id": self.player_id},
224            service_data={"volume_level": volume_level / 100},
225        )
226
227    async def volume_mute(self, muted: bool) -> None:
228        """Handle VOLUME MUTE command on the player."""
229        await self.hass.call_service(
230            domain="media_player",
231            service="volume_mute",
232            target={"entity_id": self.player_id},
233            service_data={"is_volume_muted": muted},
234        )
235
236    async def power(self, powered: bool) -> None:
237        """Handle POWER command on the player."""
238        await self.hass.call_service(
239            domain="media_player",
240            service="turn_on" if powered else "turn_off",
241            target={"entity_id": self.player_id},
242        )
243
244    async def next_track(self) -> None:
245        """Handle NEXT_TRACK command on the player."""
246        await self.hass.call_service(
247            domain="media_player",
248            service="media_next_track",
249            target={"entity_id": self.player_id},
250        )
251
252    async def previous_track(self) -> None:
253        """Handle PREVIOUS_TRACK command on the player."""
254        await self.hass.call_service(
255            domain="media_player",
256            service="media_previous_track",
257            target={"entity_id": self.player_id},
258        )
259
260    async def play_media(self, media: PlayerMedia) -> None:
261        """Handle PLAY MEDIA on given player."""
262        url = await self.provider.mass.streams.resolve_stream_url(self.player_id, media)
263        extra_data: dict[str, Any] = {
264            # passing metadata to the player
265            # so far only supported by google cast, but maybe others can follow
266            "metadata": {
267                "title": media.title,
268                "artist": media.artist,
269                "metadataType": 3,
270                "album": media.album,
271                "albumName": media.album,
272                "images": [{"url": media.image_url}] if media.image_url else None,
273                "imageUrl": media.image_url,
274                "duration": media.stream_duration or media.duration,
275            },
276        }
277        if self.extra_data.get("hass_domain") == "esphome":
278            # tell esphome mediaproxy to bypass the proxy,
279            # as MA already delivers an optimized stream
280            extra_data["bypass_proxy"] = True
281
282        # stop the player if it is already playing
283        if self._attr_playback_state == PlaybackState.PLAYING:
284            await self.stop()
285
286        await self.hass.call_service(
287            domain="media_player",
288            service="play_media",
289            target={"entity_id": self.player_id},
290            service_data={
291                "media_content_id": url,
292                "media_content_type": "music",
293                "enqueue": "replace",
294                "extra": extra_data,
295            },
296        )
297
298        # Optimistically update state
299        self._attr_current_media = media
300        self._attr_elapsed_time = 0
301        self._attr_elapsed_time_last_updated = time.time()
302        self._attr_playback_state = PlaybackState.PLAYING
303        self.update_state()
304
305    async def play_announcement(
306        self, announcement: PlayerMedia, volume_level: int | None = None
307    ) -> None:
308        """Handle (provider native) playback of an announcement on given player."""
309        self.logger.info(
310            "Playing announcement %s on %s",
311            announcement.uri,
312            self.display_name,
313        )
314        if volume_level is not None:
315            self.logger.warning(
316                "Announcement volume level is not supported for player %s",
317                self.display_name,
318            )
319        hass_prov = cast("HomeAssistantPlayerProvider", self.provider).hass_prov
320        await hass_prov.play_announcement_on_entity(self.player_id, announcement)
321        self.logger.debug(
322            "Playing announcement on %s completed",
323            self.display_name,
324        )
325
326    async def set_members(
327        self,
328        player_ids_to_add: list[str] | None = None,
329        player_ids_to_remove: list[str] | None = None,
330    ) -> None:
331        """
332        Handle SET_MEMBERS command on the player.
333
334        Group or ungroup the given child player(s) to/from this player.
335        Will only be called if the PlayerFeature.SET_MEMBERS is supported.
336
337        :param player_ids_to_add: List of player_id's to add to the group.
338        :param player_ids_to_remove: List of player_id's to remove from the group.
339        """
340        for player_id_to_remove in player_ids_to_remove or []:
341            await self.hass.call_service(
342                domain="media_player",
343                service="unjoin",
344                target={"entity_id": player_id_to_remove},
345            )
346        if player_ids_to_add:
347            await self.hass.call_service(
348                domain="media_player",
349                service="join",
350                service_data={"group_members": player_ids_to_add},
351                target={"entity_id": self.player_id},
352            )
353
354    def update_from_compressed_state(self, state: CompressedState) -> None:
355        """Handle updating the player with updated info in a HA CompressedState."""
356        if "s" in state:
357            self._attr_playback_state = StateMap.get(state["s"], PlaybackState.IDLE)
358            self._attr_available = state["s"] not in UNAVAILABLE_STATES
359            if PlayerFeature.POWER in self.supported_features:
360                self._attr_powered = state["s"] not in OFF_STATES
361        if "a" in state:
362            self._update_attributes(state["a"])
363        self.update_state()
364
365    def _update_hass_features(self, hass_supported_features: MediaPlayerEntityFeature) -> None:
366        """Update player and External source features based on HA supported features."""
367        # Update player supported features for PAUSE and NEXT_PREVIOUS
368        if MediaPlayerEntityFeature.PAUSE in hass_supported_features:
369            self._attr_supported_features.add(PlayerFeature.PAUSE)
370        else:
371            self._attr_supported_features.discard(PlayerFeature.PAUSE)
372
373        has_next_prev = (
374            MediaPlayerEntityFeature.NEXT_TRACK in hass_supported_features
375            or MediaPlayerEntityFeature.PREVIOUS_TRACK in hass_supported_features
376        )
377        if has_next_prev:
378            self._attr_supported_features.add(PlayerFeature.NEXT_PREVIOUS)
379        else:
380            self._attr_supported_features.discard(PlayerFeature.NEXT_PREVIOUS)
381
382        # Update the External source capabilities
383        for source in self._attr_source_list:
384            if source.id == "External":
385                source.can_play_pause = MediaPlayerEntityFeature.PAUSE in hass_supported_features
386                source.can_next_previous = has_next_prev
387                source.can_seek = MediaPlayerEntityFeature.SEEK in hass_supported_features
388                break
389
390    def _update_attributes(self, attributes: dict[str, Any]) -> None:
391        """Update Player attributes from HA state attributes."""
392        self._hass_attributes.update(attributes)
393
394        # process optional attributes - these may not be present in all states
395        for key, value in attributes.items():
396            if key == "friendly_name":
397                self._attr_name = value
398            elif key == "media_position":
399                self._attr_elapsed_time = value
400            elif key == "media_position_updated_at":
401                self._attr_elapsed_time_last_updated = from_iso_string(value).timestamp()
402            elif key == "volume_level":
403                self._attr_volume_level = int(value * 100)
404            elif key == "is_volume_muted":
405                self._attr_volume_muted = value
406            elif key == "group_members":
407                group_members: list[str] = (
408                    [
409                        # ignore integrations that incorrectly set the group members attribute
410                        # (e.g. linkplay)
411                        x
412                        for x in value
413                        if x.startswith("media_player.")
414                    ]
415                    if value
416                    else []
417                )
418                if group_members and group_members[0] == self.player_id:
419                    # first in the list is the group leader
420                    self._attr_group_members = group_members
421                elif group_members and group_members[0] != self.player_id:
422                    # this player is not the group leader
423                    self._attr_group_members.clear()
424                else:
425                    self._attr_group_members.clear()
426            elif key == "supported_features":
427                # Update supported features dynamically via shared helper
428                hass_supported_features = parse_supported_features(
429                    value, self.player_id, self.logger
430                )
431                self.extra_data["hass_supported_features"] = hass_supported_features
432                self._update_hass_features(hass_supported_features)
433
434        # Check for external playback (not from Music Assistant).
435        # Without media_content_id we cannot reliably determine the source,
436        # so we later only react to state updates that include it.
437        media_content_id = self._hass_attributes.get("media_content_id", "")
438        is_ma_playback = media_content_id.startswith(self.mass.streams.base_url)
439        media_title = self._hass_attributes.get("media_title")
440
441        if media_content_id and is_ma_playback:
442            # MA playback - ensure active_source points to player_id for queue lookup.
443            # The actual current_media will be set by MA's queue controller.
444            self._attr_active_source = None
445        elif (
446            media_content_id
447            and media_title
448            and self.playback_state in (PlaybackState.PLAYING, PlaybackState.PAUSED)
449        ):
450            # External playback detected - set current_media from HA attributes
451            ha_content_type = self._hass_attributes.get("media_content_type", "")
452            media_type = MediaType.RADIO if ha_content_type == "radio" else MediaType.UNKNOWN
453            current_media = PlayerMedia(
454                uri=media_content_id,
455                media_type=media_type,
456                title=media_title,
457                artist=self._hass_attributes.get("media_artist"),
458                album=self._hass_attributes.get("media_album_name"),
459                image_url=self._get_image_url(self._hass_attributes),
460                duration=int(self._hass_attributes.get("media_duration", 0) or 0) or None,
461            )
462            self._attr_current_media = current_media
463            self._attr_active_source = "External"
464
465        elif self.playback_state == PlaybackState.IDLE:
466            # Clear external media if it was set
467            if self._attr_active_source and self._attr_active_source not in (
468                self.player_id,
469                None,
470            ):
471                self._attr_current_media = None
472                self._attr_active_source = None
473
474    def _get_image_url(self, attributes: dict[str, Any]) -> str | None:
475        """Get the image URL from the attributes."""
476        if entity_picture := attributes.get("entity_picture"):
477            entity_picture = str(entity_picture)
478            if entity_picture.startswith("http"):
479                return entity_picture
480
481            # Access via provider -> hass_prov
482            prov = cast("HomeAssistantPlayerProvider", self.provider)
483
484            # Use proxy for internal HA images
485            # We create a MediaItemImage with the hass provider as source
486            # This will trigger resolve_image on the hass provider when requested
487            image = MediaItemImage(
488                type=ImageType.THUMB,
489                path=entity_picture,
490                provider=prov.hass_prov.instance_id,
491                remotely_accessible=False,
492            )
493            return self.mass.metadata.get_image_url(image)
494        return None
495
496    def _has_native_duplicate(self) -> bool:
497        """Whether this device is also registered as a native Music Assistant player."""
498        if not (mac := self.device_info.identifiers.get(IdentifierType.MAC_ADDRESS)):
499            return False
500        return normalized_mac(mac) in native_player_macs(self.mass)
501