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