music-assistant-server

1"""
2Helpers for hosting a shared listening experience.
3
4Provides the SharedPlaybackSession abstraction that plugin providers (e.g. the
5party plugin) build on to let a group of guests listen to the same queue.
6Two modes are supported:
7
8- VENUE: an existing real player owns the queue and plays out loud;
9  guests may optionally listen in on their own device when the venue
10  player supports grouping with it.
11- REMOTE: a hidden Sendspin virtual player owns the queue and leads the
12  group; every guest's web player can be attached, so all playback
13  happens on the guests' own devices (silent-disco style).
14
15NOTE: the virtual player backing a REMOTE session lives in memory of the
16Sendspin provider. When that provider reloads, the session is gone and the
17owning plugin is responsible for re-creating it; passing the same session_id
18to :meth:`SharedPlaybackSession.create_remote` yields the same player_id.
19"""
20
21from __future__ import annotations
22
23import asyncio
24import logging
25from enum import StrEnum
26from functools import partial
27from typing import TYPE_CHECKING, cast
28
29from music_assistant_models.enums import PlayerFeature
30from music_assistant_models.errors import (
31    MusicAssistantError,
32    SetupFailedError,
33    UnsupportedFeaturedException,
34)
35
36from music_assistant.helpers.util import join_task
37
38if TYPE_CHECKING:
39    from music_assistant.mass import MusicAssistant
40    from music_assistant.models.player import Player
41    from music_assistant.providers.sendspin.provider import SendspinProvider
42
43SENDSPIN_DOMAIN = "sendspin"
44REMOTE_CREATION_CLEANUP_TIMEOUT = 15.0
45REMOTE_REMOVAL_CLEANUP_DELAYS = (0.0, 1.0, 5.0)
46REMOTE_REMOVAL_CLEANUP_TIMEOUT = 5.0
47
48LOGGER = logging.getLogger(__name__)
49
50
51class SharedPlaybackMode(StrEnum):
52    """Mode of a shared playback session."""
53
54    VENUE = "venue"
55    REMOTE = "remote"
56
57
58def is_remote_session_host(mass: MusicAssistant, player_id: str) -> bool:
59    """
60    Return whether the given player is the virtual host of a REMOTE session.
61
62    :param mass: MusicAssistant instance.
63    :param player_id: Player to inspect.
64    """
65    sendspin = cast("SendspinProvider | None", mass.get_provider(SENDSPIN_DOMAIN))
66    return sendspin is not None and sendspin.is_virtual_player(player_id)
67
68
69class SharedPlaybackSession:
70    """
71    A player/queue that hosts a shared listening experience.
72
73    Use the :meth:`create_venue` or :meth:`create_remote` factory to create a
74    session; the owning plugin drives playback on :attr:`queue_id` and calls
75    :meth:`close` when the session ends.
76    """
77
78    def __init__(self, mass: MusicAssistant, mode: SharedPlaybackMode, player_id: str) -> None:
79        """Initialize the session. Use the create_venue/create_remote factories instead."""
80        self.mass = mass
81        self._mode = mode
82        self._player_id = player_id
83        self._guest_listeners: set[str] = set()
84
85    @classmethod
86    async def create_venue(
87        cls, mass: MusicAssistant, venue_player_id: str
88    ) -> SharedPlaybackSession:
89        """
90        Create a session hosted by an existing (real) player.
91
92        :param mass: MusicAssistant instance.
93        :param venue_player_id: The player_id of the player that owns the queue
94            and plays out loud.
95        :raises SetupFailedError: If the venue player is unknown.
96        :return: The created session.
97        """
98        if mass.players.get_player(venue_player_id) is None:
99            raise SetupFailedError(f"Venue player {venue_player_id} is not available")
100        return cls(mass, SharedPlaybackMode.VENUE, venue_player_id)
101
102    @classmethod
103    async def create_remote(
104        cls,
105        mass: MusicAssistant,
106        owner_instance_id: str,
107        display_name: str,
108        session_id: str | None = None,
109    ) -> SharedPlaybackSession:
110        """
111        Create a session hosted by a hidden Sendspin virtual player.
112
113        :param mass: MusicAssistant instance.
114        :param owner_instance_id: Instance id of the plugin provider that owns
115            the session (the virtual player is removed when it unloads).
116        :param display_name: Human readable name for the virtual player.
117        :param session_id: Optional stable id for the virtual player so the
118            owner can re-create the session with the same player_id.
119        :raises SetupFailedError: If the Sendspin provider is not loaded.
120        :return: The created session.
121        """
122        sendspin = cast("SendspinProvider | None", mass.get_provider(SENDSPIN_DOMAIN))
123        if sendspin is None:
124            raise SetupFailedError("The Sendspin provider is required for a remote session")
125        creation = sendspin.create_virtual_player(
126            owner_instance_id=owner_instance_id,
127            display_name=display_name,
128            player_id=session_id,
129        )
130        try:
131            creation_task = mass.create_task(creation, eager_start=False)
132        except Exception:
133            creation.close()
134            raise
135        cleanup_required = asyncio.get_running_loop().create_future()
136        cleanup = cls._cleanup_cancelled_remote_creation(
137            mass,
138            sendspin,
139            creation_task,
140            cleanup_required,
141        )
142        try:
143            mass.create_task(cleanup, eager_start=False)
144        except Exception:
145            cleanup.close()
146            cls._cancel_and_observe_creation(mass, sendspin, creation_task)
147            raise
148        try:
149            # join: cancelling the caller must not abort the creation halfway, or the
150            # cleanup task can no longer remove the player it left behind
151            player_id = await join_task(creation_task)
152        except asyncio.CancelledError:
153            cleanup_required.set_result(True)
154            raise
155        except Exception:
156            cleanup_required.set_result(False)
157            raise
158        cleanup_required.set_result(False)
159        return cls(mass, SharedPlaybackMode.REMOTE, player_id)
160
161    @property
162    def mode(self) -> SharedPlaybackMode:
163        """Return the mode of this session."""
164        return self._mode
165
166    @property
167    def player_id(self) -> str:
168        """Return the player_id of the player that hosts this session."""
169        return self._player_id
170
171    @property
172    def queue_id(self) -> str:
173        """Return the queue_id of the queue that hosts this session."""
174        # a player-owned queue always has the same id as the player
175        return self._player_id
176
177    def can_listen_in(self, web_player_id: str) -> bool:
178        """
179        Return whether the given guest web player can listen in on this session.
180
181        :param web_player_id: The player_id of the guest's web player.
182        """
183        if (host_player := self._get_host_player()) is None:
184            return False
185        if PlayerFeature.SET_MEMBERS not in host_player.state.supported_features:
186            return False
187        # state.can_group_with handles all protocol expansion and translation,
188        # for both a real venue player and a (virtual) Sendspin host player
189        return (
190            web_player_id in host_player.state.can_group_with
191            or web_player_id in host_player.state.group_members
192        )
193
194    async def add_guest_listener(self, web_player_id: str) -> None:
195        """
196        Attach a guest's web player to this session so it plays the same audio.
197
198        :param web_player_id: The player_id of the guest's web player.
199        :raises UnsupportedFeaturedException: If the session host does not
200            support grouping with the given player.
201        """
202        if not self.can_listen_in(web_player_id):
203            raise UnsupportedFeaturedException(
204                f"Player {web_player_id} can not listen in on this session"
205            )
206        await self.mass.players.cmd_set_members(self._player_id, player_ids_to_add=[web_player_id])
207        self._guest_listeners.add(web_player_id)
208
209    async def restore_guest_listeners(self) -> None:
210        """
211        Restore tracked guest listeners missing from the host player's group.
212
213        Missing or temporarily incompatible guest players remain tracked so a
214        later playback transition can restore them after they reconnect.
215        """
216        host_player = self._get_host_player()
217        if host_player is None or not self._guest_listeners:
218            return
219        group_members = set(host_player.state.group_members)
220        for web_player_id in sorted(self._guest_listeners):
221            if web_player_id in group_members:
222                continue
223            guest_player = self.mass.players.get_player(web_player_id)
224            if (
225                guest_player is None
226                or not guest_player.state.available
227                or not self.can_listen_in(web_player_id)
228            ):
229                continue
230            try:
231                await self.mass.players.cmd_set_members(
232                    self._player_id,
233                    player_ids_to_add=[web_player_id],
234                )
235            except MusicAssistantError as err:
236                LOGGER.warning(
237                    "Could not restore guest listener %s to shared playback session %s: %s",
238                    web_player_id,
239                    self._player_id,
240                    err,
241                )
242                continue
243            group_members.add(web_player_id)
244
245    async def remove_guest_listener(self, web_player_id: str) -> None:
246        """
247        Detach a guest's web player from this session.
248
249        :param web_player_id: The player_id of the guest's web player.
250        """
251        self._guest_listeners.discard(web_player_id)
252        if self._get_host_player() is None:
253            return
254        await self.mass.players.cmd_set_members(
255            self._player_id, player_ids_to_remove=[web_player_id]
256        )
257
258    async def close(self) -> None:
259        """
260        Tear down the session.
261
262        In REMOTE mode the virtual player (and its queue) is removed entirely.
263        In VENUE mode only the guest listeners added through this session are
264        detached; the venue player itself is left untouched.
265        """
266        if self._mode == SharedPlaybackMode.REMOTE:
267            sendspin = cast("SendspinProvider | None", self.mass.get_provider(SENDSPIN_DOMAIN))
268            if sendspin is not None and sendspin.is_virtual_player(self._player_id):
269                await sendspin.remove_virtual_player(self._player_id)
270            self._guest_listeners.clear()
271            return
272        if self._guest_listeners and self._get_host_player() is not None:
273            await self.mass.players.cmd_set_members(
274                self._player_id, player_ids_to_remove=list(self._guest_listeners)
275            )
276        self._guest_listeners.clear()
277
278    @classmethod
279    async def _cleanup_cancelled_remote_creation(
280        cls,
281        mass: MusicAssistant,
282        sendspin: SendspinProvider,
283        creation_task: asyncio.Task[str],
284        cleanup_required: asyncio.Future[bool],
285    ) -> None:
286        """
287        Clean up a remote virtual player when its session creation is cancelled.
288
289        :param mass: MusicAssistant instance.
290        :param sendspin: Sendspin provider that owns the virtual player.
291        :param creation_task: In-flight virtual-player creation task.
292        :param cleanup_required: Signal indicating whether cleanup is needed.
293        """
294        if not await asyncio.shield(cleanup_required):
295            return
296        try:
297            done, _ = await asyncio.wait(
298                (creation_task,),
299                timeout=REMOTE_CREATION_CLEANUP_TIMEOUT,
300            )
301            if not done:
302                LOGGER.warning("Timed out waiting for cancelled remote session creation")
303                cls._cancel_and_observe_creation(mass, sendspin, creation_task)
304                return
305        except asyncio.CancelledError:
306            cls._cancel_and_observe_creation(mass, sendspin, creation_task)
307            raise
308
309        try:
310            player_id = creation_task.result()
311        except asyncio.CancelledError:
312            return
313        except Exception as err:
314            LOGGER.debug("Cancelled remote session creation failed: %s", err)
315            return
316
317        await cls._cleanup_cancelled_remote_player(sendspin, player_id)
318
319    @staticmethod
320    async def _cleanup_cancelled_remote_player(
321        sendspin: SendspinProvider,
322        player_id: str,
323    ) -> None:
324        """
325        Remove a virtual player left by cancelled remote session creation.
326
327        :param sendspin: Sendspin provider that owns the virtual player.
328        :param player_id: Virtual player to remove.
329        """
330        last_error: Exception | None = None
331        for delay in REMOTE_REMOVAL_CLEANUP_DELAYS:
332            if delay:
333                await asyncio.sleep(delay)
334            try:
335                if not sendspin.is_virtual_player(player_id):
336                    return
337                async with asyncio.timeout(REMOTE_REMOVAL_CLEANUP_TIMEOUT):
338                    await sendspin.remove_virtual_player(player_id)
339                return
340            except Exception as err:
341                last_error = err
342        LOGGER.warning(
343            "Could not clean up cancelled remote session %s: %s",
344            player_id,
345            last_error,
346        )
347
348    @classmethod
349    def _cancel_and_observe_creation(
350        cls,
351        mass: MusicAssistant,
352        sendspin: SendspinProvider,
353        task: asyncio.Task[str],
354    ) -> None:
355        """Cancel virtual-player creation and observe its eventual result."""
356        task.cancel()
357        cls._observe_late_remote_creation(mass, sendspin, task)
358
359    @classmethod
360    def _observe_late_remote_creation(
361        cls,
362        mass: MusicAssistant,
363        sendspin: SendspinProvider,
364        task: asyncio.Task[str],
365    ) -> None:
366        """Observe creation after bounded cleanup stops waiting for it."""
367        task.add_done_callback(partial(cls._handle_late_remote_creation, mass, sendspin))
368
369    @classmethod
370    def _handle_late_remote_creation(
371        cls,
372        mass: MusicAssistant,
373        sendspin: SendspinProvider,
374        task: asyncio.Task[str],
375    ) -> None:
376        """Schedule cleanup when cancelled creation eventually returns a player."""
377        if task.cancelled():
378            return
379        try:
380            player_id = task.result()
381        except Exception as err:
382            LOGGER.debug("Cancelled remote session creation failed: %s", err)
383            return
384        cleanup = cls._cleanup_cancelled_remote_player(sendspin, player_id)
385        try:
386            mass.create_task(cleanup, eager_start=False)
387        except Exception as err:
388            cleanup.close()
389            LOGGER.warning(
390                "Could not schedule cancelled remote session cleanup for %s: %s",
391                player_id,
392                err,
393            )
394
395    def _get_host_player(self) -> Player | None:
396        """Return the (available) player hosting this session, if any."""
397        player = self.mass.players.get_player(self._player_id)
398        if player is None or not player.state.available:
399            return None
400        return player
401