music-assistant-server

1"""
2Base class for players that delegate playback to linked protocol players.
3
4A protocol-backed player owns no native audio path: playback, transport state,
5current media and volume are provided by one or more linked protocol players
6(AirPlay, DLNA, Chromecast, Sendspin, ...). Subclasses only supply the set of
7backing protocol player ids and may add their own native capabilities on top
8(for example native multiroom grouping).
9"""
10
11from __future__ import annotations
12
13from typing import TYPE_CHECKING
14
15from music_assistant_models.enums import PlaybackState, PlayerFeature
16
17from music_assistant.constants import EXTERNAL_SOURCES
18from music_assistant.models.player import Player
19
20if TYPE_CHECKING:
21    from music_assistant_models.player import PlayerMedia, PlayerSource
22
23# Protocol domains where an external source (e.g. Spotify Connect) can play
24# independently of Music Assistant, so we surface that protocol player's state.
25EXTERNAL_SOURCE_PROTOCOLS = {"chromecast", "dlna"}
26
27# Features forwarded from a protocol player that is playing an external source.
28# Volume and mute are excluded: the base Player resolves those to the protocol player.
29FORWARDED_FEATURES = {
30    PlayerFeature.PAUSE,
31    PlayerFeature.SEEK,
32    PlayerFeature.NEXT_PREVIOUS,
33}
34
35
36class ProtocolBackedPlayer(Player):
37    """
38    Base for players whose playback is delegated to linked protocol players.
39
40    The player has no PLAY_MEDIA capability of its own; the player controller routes
41    playback to one of the linked protocol players. This base surfaces the delegated
42    playback/state/media and any external source active on a linked protocol player.
43    """
44
45    @property
46    def available(self) -> bool:
47        """Return if the player is currently available."""
48        return any(
49            (p := self.mass.players.get_player(pid)) and p.available_for_playback
50            for pid in self._backing_protocol_player_ids()
51        )
52
53    @property
54    def needs_setup(self) -> bool:
55        """Return if the player needs setup (a protocol is connected but not set up)."""
56        if self.available:
57            return False
58        return self._get_protocol_player_needing_setup() is not None
59
60    @property
61    def setup_reason(self) -> str | None:
62        """Return why the player needs setup, or None when it is ready to use."""
63        if self.available:
64            return None
65        if protocol_player := self._get_protocol_player_needing_setup():
66            return protocol_player.setup_reason
67        return None
68
69    @property
70    def supported_features(self) -> set[PlayerFeature]:
71        """Return the supported features of the player."""
72        if ext_player := self._get_protocol_player_with_external_source():
73            # Keep this player's own native capabilities (a subclass may add some, e.g.
74            # grouping) and add the forwardable transport controls of the external source.
75            return self._attr_supported_features | (
76                ext_player.supported_features & FORWARDED_FEATURES
77            )
78        return self._attr_supported_features
79
80    @property
81    def active_source(self) -> str | None:
82        """Return the active source of the player."""
83        if ext_player := self._get_protocol_player_with_external_source():
84            return ext_player.active_source
85        return None
86
87    @property
88    def playback_state(self) -> PlaybackState:
89        """Return the current playback state of the player."""
90        if ext_player := self._get_protocol_player_with_external_source():
91            return ext_player.playback_state
92        return self._attr_playback_state
93
94    @property
95    def elapsed_time(self) -> float | None:
96        """Return the elapsed time in (fractional) seconds of the current track."""
97        if ext_player := self._get_protocol_player_with_external_source():
98            return ext_player.elapsed_time
99        return None
100
101    @property
102    def elapsed_time_last_updated(self) -> float | None:
103        """Return when the elapsed time was last updated."""
104        if ext_player := self._get_protocol_player_with_external_source():
105            return ext_player.elapsed_time_last_updated
106        return None
107
108    @property
109    def current_media(self) -> PlayerMedia | None:
110        """Return the current media being played by the player."""
111        if ext_player := self._get_protocol_player_with_external_source():
112            return ext_player.current_media
113        if protocol_player := self._get_active_output_protocol_player():
114            # while playing through an output protocol player, surface its raw current_media
115            # so consumers of this player's raw value (e.g. a sync group mirroring its leader)
116            # can resolve the active queue item. Reading the protocol player's .state here would
117            # route back through this player's __final_current_media and lose the queue item id.
118            return protocol_player.current_media
119        return None
120
121    @property
122    def source_list(self) -> list[PlayerSource]:
123        """Return list of available sources for this player."""
124        if ext_player := self._get_protocol_player_with_external_source():
125            # if an external source is active, show sources from that protocol player
126            return ext_player.source_list
127        return super().source_list
128
129    async def stop(self) -> None:
130        """Handle STOP command on the player."""
131        if ext_player := self._get_protocol_player_with_external_source():
132            await ext_player.stop()
133
134    async def play(self) -> None:
135        """Handle PLAY command on the player."""
136        if ext_player := self._get_protocol_player_with_external_source():
137            await ext_player.play()
138
139    async def pause(self) -> None:
140        """Handle PAUSE command on the player."""
141        if ext_player := self._get_protocol_player_with_external_source():
142            await ext_player.pause()
143
144    async def next_track(self) -> None:
145        """Handle NEXT_TRACK command on the player."""
146        if ext_player := self._get_protocol_player_with_external_source():
147            await ext_player.next_track()
148
149    async def previous_track(self) -> None:
150        """Handle PREVIOUS_TRACK command on the player."""
151        if ext_player := self._get_protocol_player_with_external_source():
152            await ext_player.previous_track()
153
154    async def seek(self, position: int) -> None:
155        """Handle SEEK command on the player."""
156        if ext_player := self._get_protocol_player_with_external_source():
157            await ext_player.seek(position)
158            self.mass.players.trigger_player_update(ext_player.player_id, debounce_delay=2)
159
160    def _backing_protocol_player_ids(self) -> list[str]:
161        """Return the ids of the protocol players backing this player."""
162        raise NotImplementedError
163
164    def _get_protocol_player_needing_setup(self) -> Player | None:
165        """Return the first connected protocol player that still needs setup, if any."""
166        for pid in self._backing_protocol_player_ids():
167            protocol_player = self.mass.players.get_player(pid)
168            if protocol_player and protocol_player.available and protocol_player.needs_setup:
169                return protocol_player
170        return None
171
172    def _get_active_output_protocol_player(self) -> Player | None:
173        """Return the protocol player currently selected as this player's output, if any."""
174        if self.active_output_protocol and self.active_output_protocol != "native":
175            return self.mass.players.get_player(self.active_output_protocol)
176        return None
177
178    def _get_protocol_player_with_external_source(self) -> Player | None:
179        """
180        Return a chromecast or dlna protocol player that has an external source active.
181
182        Prefers chromecast over dlna because dlna metadata tends to be less reliable.
183        """
184        if self.active_output_protocol:
185            # if an output protocol is active, don't consider external sources
186            return None
187        result: Player | None = None
188        for pid in self._backing_protocol_player_ids():
189            protocol_player = self.mass.players.get_player(pid)
190            if not protocol_player or not protocol_player.available:
191                continue
192            if protocol_player.provider.domain not in EXTERNAL_SOURCE_PROTOCOLS:
193                continue
194            if (
195                protocol_player.active_source
196                and protocol_player.active_source.lower() in EXTERNAL_SOURCES
197                and protocol_player.playback_state != PlaybackState.IDLE
198            ):
199                # chromecast is preferred, return immediately
200                if protocol_player.provider.domain == "chromecast":
201                    return protocol_player
202                result = result or protocol_player
203        return result
204