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