music-assistant-server

1"""Snapcast Player."""
2
3from __future__ import annotations
4
5import asyncio
6from contextlib import suppress
7from typing import TYPE_CHECKING, TypedDict, cast
8
9from music_assistant_models.config_entries import ConfigEntry
10from music_assistant_models.enums import IdentifierType, PlaybackState, PlayerFeature
11from music_assistant_models.player import DeviceInfo, PlayerMedia
12from propcache import under_cached_property as cached_property
13
14from music_assistant.constants import ATTR_ANNOUNCEMENT_IN_PROGRESS, CONF_ENTRY_HTTP_PROFILE_HIDDEN
15from music_assistant.helpers.util import is_valid_mac_address
16from music_assistant.models.player import Player
17from music_assistant.providers.snapcast.constants import (
18    SNAPCLIENT_LIVENESS_POLL_INTERVAL,
19    SNAPCLIENT_STALE_THRESHOLD,
20)
21from music_assistant.providers.snapcast.ma_stream import SnapcastMAStream
22from music_assistant.providers.sync_group.constants import SGP_PREFIX
23
24if TYPE_CHECKING:
25    from music_assistant_models.config_entries import ConfigEntry
26
27    from music_assistant.providers.snapcast.provider import SnapCastProvider
28    from music_assistant.providers.snapcast.snap_cntrl_proto import SnapclientProto, SnapstreamProto
29
30
31class TrackedPlayerState(TypedDict, total=False):
32    """
33    Tracked state for the Snapcast MA player.
34
35    It is used for change detection and state synchronization, and may be
36    partially populated depending on which information is
37    currently available.
38
39    Keys prefixed with ``_attr_`` are exposed as player attributes, while the
40    remaining keys represent internal Snapcast grouping and connection state.
41    """
42
43    # Player attribute fields
44    _attr_name: str
45    _attr_volume_level: float
46    _attr_volume_muted: bool
47    _attr_available: bool
48
49    # snapclient fields
50    connected: bool
51    stream_id: str
52    stream_status: str | None
53    grp_name: str
54    grp_member_ids: list[str]
55    grp_member_avail: list[bool]
56
57
58class SnapCastPlayer(Player):
59    """SnapCastPlayer."""
60
61    def __init__(
62        self,
63        provider: SnapCastProvider,
64        player_id: str,
65        snap_client: SnapclientProto,
66    ) -> None:
67        """Init."""
68        self.snap_client = snap_client
69        super().__init__(provider, player_id)
70
71        # snapcast has fixed sample rate/bit depth
72        self._attr_supported_sample_rates = [(48000, 16)]
73
74        self._snap_ma_stream: SnapcastMAStream | None = None
75
76        self._update_worker: asyncio.Task[None] | None = None
77        self._poke_evt = asyncio.Event()
78        self._state_update_lock = asyncio.Lock()
79        self._last_tracked_state: TrackedPlayerState | None = None
80        # raw lastSeen value + monotonic time it last advanced (skew-proof liveness)
81        self._last_seen_raw: tuple[int, int] | None = None
82        self._last_seen_changed: float = 0.0
83
84    @property
85    def snap_provider(self) -> SnapCastProvider:
86        """Return the Snapcast provider instance."""
87        return cast("SnapCastProvider", self.provider)
88
89    @property
90    def requires_flow_mode(self) -> bool:
91        """Return if the player requires flow mode."""
92        return True
93
94    @property
95    def synced_to(self) -> str | None:
96        """Return the id of the player this player is synced to (sync leader)."""
97        grp_name = self.snap_group_name
98        if grp_name == self.player_id:
99            # is group leader
100            return None
101
102        grp_player_ids = self._get_player_ids_of_curr_group()
103        if len(grp_player_ids) < 2 or grp_name not in grp_player_ids:
104            return None
105
106        if leader_player := self.mass.players.get_player(grp_name):
107            return grp_name if leader_player.available else None
108
109        return None
110
111    @cached_property
112    def group_members(self) -> list[str]:
113        """Return the group members of the player."""
114        if not self._attr_available:
115            return []
116
117        grp_name = self.snap_group_name
118        if grp_name != self.player_id:
119            # only group leaders can have members
120            return []
121
122        player_ids = self._get_player_ids_of_curr_group()
123        if self.player_id not in player_ids:
124            # should not happen, unless the current
125            # state repr is invalid
126            return []
127
128        player_ids.remove(self.player_id)
129        connected = [
130            player_id
131            for player_id in player_ids
132            if (client := self.snap_provider.get_snap_client(player_id=player_id))
133            and client.connected
134        ]
135        if connected:
136            return [self.player_id, *connected]
137
138        return []
139
140    @property
141    def playback_state(self) -> PlaybackState:
142        """Return the current playback state of the player."""
143        snap_stream = self._get_active_snapstream()
144        if snap_stream is None:
145            return PlaybackState.IDLE
146
147        if snap_stream.identifier == "default" or snap_stream.status == "idle":
148            return PlaybackState.IDLE
149
150        return PlaybackState.PLAYING
151
152    @property
153    def elapsed_time(self) -> float | None:
154        """Return the elapsed time in (fractional) seconds of the current track (if any)."""
155        # using flow-mode, elapsed time will be estimated upstream from 'elapsed_time_last_updated'
156        return 0 if self.active_snap_ma_stream else None
157
158    @property
159    def elapsed_time_last_updated(self) -> float | None:
160        """
161        Return when the elapsed time was last updated.
162
163        return: The (UTC) timestamp when the elapsed time was last updated,
164        or None if it was never updated (or unknown).
165        """
166        # we only update on playback starts
167        if snap_ma_stream := self.active_snap_ma_stream:
168            return snap_ma_stream.playback_started_at
169        return None
170
171    def setup(self) -> None:
172        """Set up player."""
173        self._attr_name = self.snap_client.friendly_name
174        self._attr_available = self.snap_client.connected
175
176        host_dict = self.snap_client._client.get("host", {})
177        os, arch, ip, mac = (host_dict.get(key, "") for key in ["os", "arch", "ip", "mac"])
178        self._attr_device_info = DeviceInfo(
179            model=os,
180            manufacturer=arch,
181        )
182        if ip and (host := self.snap_client._client.get("host")):
183            self._attr_device_info.add_identifier(IdentifierType.IP_ADDRESS, host.get("ip"))
184        # Only add MAC address if it's valid (not 00:00:00:00:00:00)
185        if mac and is_valid_mac_address(mac):
186            self._attr_device_info.add_identifier(IdentifierType.MAC_ADDRESS, mac)
187        self._attr_supported_features = {
188            PlayerFeature.PLAY_MEDIA,
189            PlayerFeature.SET_MEMBERS,
190            PlayerFeature.VOLUME_SET,
191            PlayerFeature.VOLUME_MUTE,
192            PlayerFeature.PLAY_ANNOUNCEMENT,
193        }
194        self._attr_can_group_with = {self.snap_provider.instance_id}
195        # poll to detect abruptly powered-off clients (see _is_alive)
196        self._attr_needs_poll = True
197        self._attr_poll_interval = SNAPCLIENT_LIVENESS_POLL_INTERVAL
198        if not self._update_worker:
199            self._update_worker = self.mass.create_task(self._player_update_worker)
200
201    async def poll(self) -> None:
202        """Poll the snapserver so abruptly powered-off clients are detected."""
203        await self.snap_provider.refresh_server_status()
204        self.poke_player_update()
205
206    async def volume_set(self, volume_level: int) -> None:
207        """Send VOLUME_SET command to given player."""
208        # Use optimistic server state for now
209        # not guaranteed that the client respects it
210        await self.snap_client.set_volume(volume_level)
211
212    async def stop(self) -> None:
213        """Send STOP command to given player."""
214        player_group = await self.snap_provider.ensure_player_owned_group(self.player_id)
215        assert player_group is not None  # for type checking
216        await player_group.set_stream("default")
217        if ma_stream := self.active_snap_ma_stream:
218            ma_stream.request_stop_stream()
219            return
220
221        self.poke_player_update()
222
223    async def volume_mute(self, muted: bool) -> None:
224        """Send MUTE command to given player."""
225        # Use optimistic server state for now
226        # not guaranteed that the client respects it
227        # TODO: move this to the snapcast python library
228        vol = self.snap_client._client["config"]["volume"]
229        vol["muted"] = muted
230        res = await self.snap_provider._snapserver.client_volume(self.snap_client.identifier, vol)
231        if res and "muted" in res:
232            self.snap_client._client["config"]["volume"] = res
233            self.snap_client.callback()
234
235    async def set_members(
236        self,
237        player_ids_to_add: list[str] | None = None,
238        player_ids_to_remove: list[str] | None = None,
239    ) -> None:
240        """Handle SET_MEMBERS command on the player."""
241        # get the group owned by this player (identified by the group name)
242        player_group = await self.snap_provider.ensure_player_owned_group(self.player_id)
243
244        if player_group is None:
245            return
246
247        player_group.set_callback(None)
248
249        curr_ma_player_ids = [
250            ma_id
251            for cli_id in player_group.clients
252            if (ma_id := self.snap_provider._get_ma_id(cli_id))
253        ]
254
255        curr_stream_id = player_group.stream
256        sync_group_player: Player | None = None
257        if curr_ma_stream := self.snap_provider.get_snap_ma_stream(curr_stream_id):
258            media = curr_ma_stream.media
259            media_src_id = media.source_id or ""
260            if media_src_id.startswith(SGP_PREFIX):
261                sync_group_player = self.mass.players.get_player(media_src_id)
262        if sync_group_player and self.player_id in (player_ids_to_remove or []):
263            # players in sync_group_player.group_members will be rejoined
264            # remove others first
265            for id_to_remove in player_ids_to_remove or []:
266                if id_to_remove == self.player_id:
267                    continue
268                if (
269                    id_to_remove in curr_ma_player_ids
270                    and id_to_remove not in sync_group_player.group_members
271                ):
272                    await self.snap_provider.isolate_player_to_dedicated_group(
273                        id_to_remove, target_stream_id="default"
274                    )
275
276            # split remaining group into individual groups,
277            # keeps the current stream, set this group to default stream
278            await self.snap_provider.isolate_player_to_dedicated_group(
279                target_player_id=self.player_id,
280                target_stream_id="default",
281                others_stream_id=curr_stream_id,
282            )
283        else:
284            for player_id in player_ids_to_remove or []:
285                if player_id not in curr_ma_player_ids:
286                    continue
287                await self.snap_provider.isolate_player_to_dedicated_group(
288                    player_id, target_stream_id="default"
289                )
290                curr_ma_player_ids.remove(player_id)
291
292        for ma_id in player_ids_to_add or []:
293            if (
294                snap_id := self.snap_provider._get_snapclient_id(ma_id)
295            ) and ma_id not in curr_ma_player_ids:
296                await player_group.add_client(snap_id)
297
298        # some caller require instant state updates before returning
299        async with self._state_update_lock:
300            if await self._process_snapcast_client_state():
301                self.update_state()
302
303        self.snap_provider._update_group_callbacks(poke=True)
304
305    async def play_media(self, media: PlayerMedia) -> None:
306        """Handle PLAY MEDIA on given player."""
307        if self.synced_to:
308            msg = "A synced player cannot receive play commands directly"
309            raise RuntimeError(msg)
310
311        ma_stream = await self.snap_provider.get_snapcast_media_stream(
312            media, filter_settings_owner=self.player_id
313        )
314
315        if ma_stream is None or ma_stream.stream_id is None:
316            return
317
318        self._snap_ma_stream = ma_stream
319
320        # e.g. DSP settings require a restart
321        await self._snap_ma_stream.start_stream(allow_restart=True)
322
323        # if no announcement is playing we activate the stream now, otherwise it
324        # will be activated by play_announcement when the announcement is over.
325        if not self.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
326            player_group = await self.snap_provider.ensure_player_owned_group(self.player_id)
327            assert player_group is not None  # for type checking
328            await player_group.set_stream(ma_stream.stream_id)
329
330        self.poke_player_update()
331
332    async def play_announcement(
333        self, announcement: PlayerMedia, volume_level: int | None = None
334    ) -> None:
335        """Handle (provider native) playback of an announcement on given player."""
336        was_synced_to: str | None = self.synced_to
337        orig_volume_level: int | None = self.volume_level
338
339        prev_stream = self.active_snap_ma_stream
340
341        ma_stream = await self.snap_provider.get_snapcast_media_stream(
342            announcement, filter_settings_owner=self.player_id
343        )
344        player_group = await self.snap_provider.ensure_player_owned_group(self.player_id)
345
346        if ma_stream is None or ma_stream.stream_id is None or player_group is None:
347            return
348
349        await player_group.set_stream(ma_stream.stream_id)
350
351        if self.snap_provider._use_builtin_server:
352            await asyncio.sleep(self.snap_provider._snapcast_server_buffer_size / 1000.0)
353
354        if volume_level is not None:
355            await self.volume_set(volume_level)
356
357        await ma_stream.start_stream()
358        await ma_stream.wait_for_stopped()
359
360        if self.volume_level == volume_level and orig_volume_level is not None:
361            await self.volume_set(orig_volume_level)
362
363        if was_synced_to:
364            if (
365                leader_group := await self.snap_provider.ensure_player_owned_group(was_synced_to)
366            ) is None:
367                return
368            await leader_group.add_client(self.snap_client.identifier)
369        else:
370            await player_group.set_stream(
371                prev_stream.stream_id
372                if prev_stream and prev_stream.stream_id is not None
373                else "default"
374            )
375
376    async def get_config_entries(self) -> list[ConfigEntry]:
377        """Player config."""
378        return [
379            # we don't use the http server for streaming
380            CONF_ENTRY_HTTP_PROFILE_HIDDEN,
381        ]
382
383    def _handle_player_update(self, snap_client: SnapclientProto) -> None:
384        """Forward snap_client updates."""
385        self.poke_player_update()
386
387    def poke_player_update(self) -> None:
388        """Signal that a player state update should be processed."""
389        self._poke_evt.set()
390
391    async def _player_update_worker(self) -> None:
392        """Aggregate and process player state update requests."""
393        while True:
394            await self._poke_evt.wait()
395            self._poke_evt.clear()
396            while True:
397                call_update: bool = False
398                try:
399                    async with self._state_update_lock:
400                        call_update = await self._process_snapcast_client_state()
401                    if call_update:
402                        self.update_state()
403                except KeyError, AttributeError, TypeError, ValueError:
404                    # a failed update must not kill this worker (state would freeze)
405                    self.logger.exception(
406                        "Error while processing state update for player %s", self.player_id
407                    )
408                    break
409                if self._poke_evt.is_set():
410                    self._poke_evt.clear()
411                    continue
412                break
413
414    async def _process_snapcast_client_state(self) -> bool:
415        """
416        Process the latest Snapcast client state and apply changes to this player.
417
418        Returns:
419        True if changes were applied and a state update should be emitted via
420        ``update_state()``; False if no update is necessary (or if required data
421        is temporarily unavailable and the update should be retried later).
422        """
423        snap_group = self.snap_client.group
424        if snap_group is None:
425            # some data syncing error, a client is always a group member
426            # retry again later, don't call update now
427            return False
428
429        stream_id = snap_group.stream
430        snap_stream: SnapstreamProto | None = None
431        with suppress(KeyError):
432            snap_stream = self.snap_provider._snapserver.stream(stream_id)
433
434        members = list(snap_group.clients)  # snapshot
435
436        curr_state: TrackedPlayerState = {
437            "_attr_name": self.snap_client.friendly_name,
438            "_attr_volume_level": self.snap_client.volume,
439            "_attr_volume_muted": self.snap_client.muted,
440            "_attr_available": self._is_alive(),
441            "connected": self.snap_client.connected,
442            "stream_id": snap_group.stream,
443            "stream_status": snap_stream.status if snap_stream is not None else None,
444            "grp_name": snap_group.name,
445            "grp_member_ids": members,
446            "grp_member_avail": [
447                pl.available
448                for cl_id in members
449                if (pl_id := self.snap_provider._get_ma_id(cl_id))
450                and (pl := self.mass.players.get_player(pl_id))
451            ],
452        }
453
454        prev_state: TrackedPlayerState = (
455            self._last_tracked_state if self._last_tracked_state is not None else {}
456        )
457        self._last_tracked_state = curr_state
458
459        # change detection for simple attrs
460        changed_attrs = {
461            k: v for k, v in curr_state.items() if k.startswith("_attr_") and prev_state.get(k) != v
462        }
463
464        prev_connected = prev_state.get("connected", False)
465        now_connected = curr_state.get("connected", False)
466        connection_changed = prev_connected != now_connected
467
468        prev_stream_id = prev_state.get("stream_id")
469        curr_stream_id = curr_state["stream_id"]
470        prev_stream_status = prev_state.get("stream_status")
471        curr_stream_status = curr_state.get("stream_status")
472
473        stream_changed = (
474            prev_stream_id != curr_stream_id or prev_stream_status != curr_stream_status
475        )
476
477        grouping_changed = any(
478            prev_state.get(k) != curr_state.get(k)
479            for k in ("grp_name", "grp_member_ids", "grp_member_avail")
480        )
481
482        needs_processing = bool(
483            changed_attrs or grouping_changed or stream_changed or connection_changed
484        )
485        if not needs_processing:
486            return False
487
488        if connection_changed or grouping_changed:
489            self.snap_provider.poke_group_members(snap_group)
490
491        # help cleaning up unused streams
492        if curr_stream_id == "default" or (
493            (my_stream := self._snap_ma_stream)
494            and my_stream.stream_id in {prev_stream_id, curr_stream_id}
495        ):
496            self.snap_provider.update_stream_usage()
497
498        # apply changed attrs
499        for key, value in changed_attrs.items():
500            setattr(self, key, value)
501
502        # finally notify state update once
503        return True
504
505    @property
506    def active_snap_ma_stream(self) -> SnapcastMAStream | None:
507        """Return the MA stream source of the active group."""
508        grp = self.snap_client.group
509        if grp is None or grp.stream is None:
510            return None
511
512        if grp.stream == "default":
513            return None
514
515        return self.snap_provider.get_snap_ma_stream(grp.stream)
516
517    @property
518    def snap_group_name(self) -> str:
519        """Return the name of the active group."""
520        snap_group = self.snap_client.group
521        if snap_group is None:
522            return ""
523        return snap_group.name
524
525    @property
526    def current_media(self) -> PlayerMedia | None:
527        """Return the current media being played by the player."""
528        if snap_ma_stream := self.active_snap_ma_stream:
529            return snap_ma_stream.media
530        return None
531
532    @property
533    def active_source(self) -> str | None:
534        """Return the (id of) the active source of the player."""
535        grp = self.snap_client.group
536        if grp is None or grp.stream is None:
537            return None
538
539        if grp.stream == "default":
540            return None
541
542        if ma_stream := self.snap_provider.get_snap_ma_stream(grp.stream):
543            return ma_stream.source_id
544
545        # external snapcast stream
546        return grp.stream or None
547
548    def _is_alive(self) -> bool:
549        """
550        Return whether the snapclient is connected and recently seen.
551
552        The snapserver advances lastSeen on every Time message a client sends
553        (~1/s while alive), but never marks an abruptly powered-off client
554        disconnected. We treat a connected client whose lastSeen has been frozen
555        longer than the threshold as gone.
556        """
557        now = self.mass.loop.time()
558        last_seen = self.snap_client._client.get("lastSeen")
559        ls = (last_seen["sec"], last_seen["usec"]) if last_seen else None
560        if ls != self._last_seen_raw:
561            self._last_seen_raw = ls
562            self._last_seen_changed = now
563        if not self.snap_client.connected:
564            return False
565        return (now - self._last_seen_changed) < SNAPCLIENT_STALE_THRESHOLD
566
567    def _get_active_snapstream(self) -> SnapstreamProto | None:
568        """Get active stream for given player_id."""
569        if group := self.snap_client.group:
570            with suppress(KeyError):
571                return self.snap_provider._snapserver.stream(group.stream)
572        return None
573
574    def _get_player_ids_of_curr_group(self) -> list[str]:
575        snap_group = self.snap_client.group
576        if snap_group is None:
577            return []
578        return [
579            ma_id
580            for client_id in snap_group.clients
581            if (ma_id := self.snap_provider._get_ma_id(client_id))
582        ]
583
584    def _get_players_of_curr_group(self) -> list[Player]:
585        return [
586            ma_player
587            for ma_id in self._get_player_ids_of_curr_group()
588            if (ma_player := self.mass.players.get_player(ma_id))
589        ]
590