/
/
/
1"""
2MusicAssistant Player Queues Controller.
3
4Handles all logic to PLAY Media Items, provided by Music Providers to supported players.
5
6It is loosely coupled to the MusicAssistant Music Controller and Player Controller.
7A Music Assistant Player always has a PlayerQueue associated with it
8which holds the queue items and state.
9
10The PlayerQueue is in that case the active source of the player,
11but it can also be something else, hence the loose coupling.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import random
18import time
19from typing import TYPE_CHECKING, Any, Final, cast
20
21import shortuuid
22from music_assistant_models.auth import Scope
23from music_assistant_models.enums import (
24 EventType,
25 MediaType,
26 PlaybackState,
27 PlayerType,
28 QueueOption,
29 RepeatMode,
30)
31from music_assistant_models.errors import (
32 AudioError,
33 InsufficientPermissions,
34 InvalidCommand,
35 InvalidDataError,
36 MediaNotFoundError,
37 MusicAssistantError,
38 PlayerCommandFailed,
39 PlayerUnavailableError,
40 QueueEmpty,
41)
42from music_assistant_models.media_items import (
43 Audiobook,
44 ItemMapping,
45 MediaItemType,
46 PlayableMediaItemType,
47 Playlist,
48 PodcastEpisode,
49 SoundEffect,
50 Track,
51)
52from music_assistant_models.player_queue import PlayerQueue
53
54from music_assistant.constants import (
55 ATTR_ANNOUNCEMENT_IN_PROGRESS,
56 MASS_LOGO_ONLINE,
57 PLAYLIST_MEDIA_TYPES,
58)
59from music_assistant.controllers.player_queues.autoplay import Autoplay
60from music_assistant.controllers.player_queues.config import (
61 core_config_entries,
62 queue_config_entries,
63)
64from music_assistant.controllers.player_queues.constants import (
65 CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
66 CACHE_CATEGORY_PLAYER_QUEUE_STATE,
67 CONF_AUTOPLAY_ENABLED,
68 CONF_CROSSFADE_ENABLED,
69 DEFAULT_AUTOPLAY_ENABLED,
70 DEFAULT_CROSSFADE_ENABLED,
71 PLAYBACK_START_TIMEOUT,
72 QUEUE_CACHE_SAVE_DELAY,
73)
74from music_assistant.controllers.player_queues.helpers import (
75 committed_index,
76 get_current_playback_speed,
77 handle_play_action,
78 is_dynamic_source,
79)
80from music_assistant.controllers.player_queues.managed_pool import ManagedPool
81from music_assistant.controllers.player_queues.media_resolver import MediaResolver
82from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
83from music_assistant.controllers.player_queues.queue_loader import QueueLoaderMixin
84from music_assistant.controllers.player_queues.smart_shuffle import SmartShuffle
85from music_assistant.controllers.player_queues.state import PlayerQueueData
86from music_assistant.controllers.player_queues.stream_feeder import StreamFeederMixin
87from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
88from music_assistant.helpers.api import api_command
89from music_assistant.helpers.config_entries import PLAYBACK_TARGET_TYPES
90from music_assistant.helpers.uri import parse_uri
91from music_assistant.models.music_provider import ProviderStreamLimitError
92from music_assistant.models.player import Player, PlayerMedia
93
94if TYPE_CHECKING:
95 from collections.abc import Iterator
96
97 from music_assistant_models import BackgroundTask
98 from music_assistant_models.config_entries import (
99 ConfigEntry,
100 ConfigValueOption,
101 CoreConfig,
102 )
103 from music_assistant_models.queue_item import QueueItem
104
105 from music_assistant import MusicAssistant
106 from music_assistant.constants import PlaylistPlayableItem
107 from music_assistant.controllers.music.recency import RecencyWindows
108 from music_assistant.helpers.json import SerializableType
109 from music_assistant.models.player import Player
110
111
112# the container media types worth surfacing as a queue "source" for clients to display. Individual
113# items (single tracks, radio streams, podcast episodes, live audio sources, ...) carry no grouping
114# and only clutter the "playing from" representation, so they are omitted from the wire `sources`.
115_WIRE_SOURCE_MEDIA_TYPES: Final = frozenset(
116 {
117 MediaType.ARTIST,
118 MediaType.ALBUM,
119 MediaType.PLAYLIST,
120 MediaType.PODCAST,
121 MediaType.AUDIOBOOK,
122 }
123)
124
125
126async def _is_audio_source(item: MediaItemType | ItemMapping | str) -> bool:
127 """
128 Return whether the given media names a live audio source.
129
130 :param item: One entry of a play request.
131 """
132 if not isinstance(item, str):
133 return item.media_type == MediaType.AUDIO_SOURCE
134 try:
135 media_type, _, _ = await parse_uri(item)
136 except MusicAssistantError:
137 return False
138 return media_type == MediaType.AUDIO_SOURCE
139
140
141async def _resolve_audio_source_request(
142 media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
143) -> str | None:
144 """
145 Return the uri of the live audio source a play request names, if it names one.
146
147 A live source is selected on a player rather than queued, so it cannot be lined
148 up behind or alongside other media: naming one among others is a caller error
149 rather than a request to interpret.
150
151 :param media: The media a play request was given.
152 :raises InvalidCommand: When a live source is named among other media.
153 """
154 items = media if isinstance(media, list) else [media]
155 sources = [item for item in items if await _is_audio_source(item)]
156 if not sources:
157 return None
158 if len(items) > 1:
159 raise InvalidCommand(
160 "A live audio source plays on its own: it can not be combined with other media"
161 )
162 item = sources[0]
163 return item if isinstance(item, str) else item.uri
164
165
166class PlayerQueuesController(QueueLoaderMixin, PlaybackTrackerMixin, StreamFeederMixin):
167 """
168 Controller holding all logic to enqueue music for players.
169
170 The loading, playback-tracking and stream-feeding logic lives in mixins (over the shared base);
171 this class owns the public API surface, the per-queue records and the stateful helper services.
172 """
173
174 def __init__(self, mass: MusicAssistant) -> None:
175 """Initialize core controller."""
176 super().__init__(mass)
177 # server-side per-queue records, keyed by queue_id; each bundles the wire PlayerQueue with
178 # its items, dynamic-source items and runtime-only state (see PlayerQueueData)
179 self._queue_data: dict[str, PlayerQueueData] = {}
180 # stateful helper services (own per-queue state + lifecycle), constructed with self
181 self._autoplay = Autoplay(self)
182 self._smart_shuffle = SmartShuffle(self)
183 self._managed_pool = ManagedPool(self)
184 self._media_resolver = MediaResolver(self)
185 self.manifest.name = "Player Queues controller"
186 self.manifest.description = (
187 "Music Assistant's core controller which manages the queues for all players."
188 )
189 self.manifest.icon = "playlist-music"
190
191 async def close(self) -> None:
192 """Cleanup on exit."""
193 # stop all playback
194 for queue in self.all():
195 if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
196 await self.stop(queue.queue_id)
197 # flush any pending (debounced) state writes so the latest queue survives shutdown/update
198 for queue in self.all():
199 self.mass.cancel_timer(f"save_queue_cache_{queue.queue_id}")
200 await self._save_queue_to_cache(queue.queue_id)
201
202 async def get_diagnostics(self) -> dict[str, SerializableType]:
203 """Return diagnostics info for this controller to include in diagnostics reports."""
204 queues = [queue_data.queue for queue_data in self._queue_data.values()]
205 by_state: dict[str, int] = {}
206 for queue in queues:
207 by_state[queue.state.value] = by_state.get(queue.state.value, 0) + 1
208 return {
209 "total": len(queues),
210 "active": sum(queue.active for queue in queues),
211 "by_state": by_state,
212 "flow_mode_active": sum(queue.flow_mode for queue in queues),
213 "dynamic_mode_active": sum(queue.is_dynamic for queue in queues),
214 "total_items": sum(queue.items for queue in queues),
215 }
216
217 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
218 """Return the core-module (global) config entries: the queue-controller defaults."""
219 # kept cheap (no library lookup): the config controller populates the global autoplay
220 # playlist dropdown for the UI, so this stays fast on the config value/parse path
221 return core_config_entries(self.mass)
222
223 async def update_config(self, config: CoreConfig, changed_keys: set[str]) -> None:
224 """Apply a global queue-settings change: refresh derived per-queue state and notify clients."""
225 await super().update_config(config, changed_keys)
226 if not any(key.startswith("values/") for key in changed_keys):
227 return
228 # queues that follow a changed global value may flip their effective toggles or derived
229 # indicators, so refresh and signal them (the audible effect lands on the next transition)
230 for queue_data in self._queue_data.values():
231 queue = queue_data.queue
232 self._resolve_default_toggles(queue_data)
233 queue.smart_fades_active = self.mass.streams.is_smart_fades_active(queue)
234 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
235 self.signal_update(queue.queue_id)
236
237 def get_queue_config_entries(
238 self, playlist_options: list[ConfigValueOption] | None = None
239 ) -> list[ConfigEntry]:
240 """
241 Return the per-queue config entries.
242
243 The autoplay_mode select disables the 'similar' option when no provider can supply
244 similar tracks. The crossfade_mode select's options and default depend on whether smart
245 fades are available: the smart option is disabled (shown but not selectable) and the
246 default falls back to standard crossfade when smart fades can't be used on this server.
247
248 :param playlist_options: Library playlists to offer for the 'playlist' autoplay mode.
249 Only populated when serving the entries to the UI; the parse path can omit it.
250 """
251 return queue_config_entries(self.mass, playlist_options)
252
253 def __iter__(self) -> Iterator[PlayerQueue]:
254 """Iterate over (available) players."""
255 return iter(queue_data.queue for queue_data in self._queue_data.values())
256
257 @api_command("player_queues/all", required_scope=Scope.QUEUES_READ)
258 def all(self) -> tuple[PlayerQueue, ...]:
259 """Return all registered PlayerQueues."""
260 return tuple(queue_data.queue for queue_data in self._queue_data.values())
261
262 @api_command("player_queues/get", required_scope=Scope.QUEUES_READ)
263 def get(self, queue_id: str) -> PlayerQueue | None:
264 """Return PlayerQueue by queue_id or None if not found."""
265 queue_data = self._queue_data.get(queue_id)
266 return queue_data.queue if queue_data else None
267
268 def queue_data(self, queue_id: str) -> PlayerQueueData:
269 """
270 Return the server-side record for a queue (raises if the queue is unknown).
271
272 Internal accessor for the stateful helper services so they reach per-queue state through
273 the controller rather than its private store.
274 """
275 return self._queue_data[queue_id]
276
277 def queue_data_or_none(self, queue_id: str) -> PlayerQueueData | None:
278 """Return the server-side record for a queue, or None if it is not registered."""
279 return self._queue_data.get(queue_id)
280
281 @api_command("player_queues/items", required_scope=Scope.QUEUES_READ)
282 def items(self, queue_id: str, limit: int = 500, offset: int = 0) -> list[QueueItem]:
283 """Return all QueueItems for given PlayerQueue."""
284 if (queue_data := self._queue_data.get(queue_id)) is None:
285 return []
286 return queue_data.items[offset : offset + limit]
287
288 @api_command("player_queues/get_active_queue", required_scope=Scope.QUEUES_READ)
289 def get_active_queue(self, player_id: str) -> PlayerQueue | None:
290 """Return the current active/synced queue for a player."""
291 if player := self.mass.players.get_player(player_id):
292 return self.mass.players.get_active_queue(player)
293 return None
294
295 # Queue commands
296
297 @api_command("player_queues/shuffle", required_scope=Scope.QUEUES_CONTROL)
298 async def set_shuffle(self, queue_id: str, shuffle_enabled: bool) -> None:
299 """Configure shuffle setting on the the queue."""
300 queue = self._queue_data[queue_id].queue
301 if queue.is_dynamic:
302 # a dynamic queue is an always-on, recency-orchestrated smart mix; manual shuffle
303 # (and plain linear order) have no meaning here so the toggle is locked
304 raise InvalidCommand("Cannot change shuffle while the queue is in dynamic mode")
305 if queue.shuffle_enabled == shuffle_enabled:
306 return # no change
307 await self._apply_local_shuffle(queue_id, shuffle_enabled)
308
309 def is_smart_shuffle_active(self, queue: PlayerQueue) -> bool:
310 """
311 Return whether smart shuffle is currently in effect for the queue.
312
313 A dynamic queue is always an orchestrated smart mix (the managed pool), so it always counts
314 as active; otherwise smart shuffle is active when shuffle is on and the per-queue
315 smart-shuffle setting is enabled.
316
317 :param queue: The queue to evaluate.
318 """
319 if queue.is_dynamic:
320 return True
321 return queue.shuffle_enabled and self._smart_shuffle.is_enabled(queue.queue_id)
322
323 @api_command("player_queues/autoplay", required_scope=Scope.QUEUES_CONTROL)
324 def set_autoplay(self, queue_id: str, autoplay_enabled: bool) -> None:
325 """Configure Autoplay setting on the queue."""
326 queue_data = self._queue_data[queue_id]
327 queue = queue_data.queue
328 queue_data.autoplay_override = autoplay_enabled
329 self._resolve_default_toggles(queue_data)
330 # if we're already at/near the end of the queue, kick off a refill right away
331 # (an active dynamic source manages its own refills, so leave it be)
332 if (
333 queue.autoplay_enabled
334 and not queue.is_dynamic
335 and queue.current_index is not None
336 and (queue.items - queue.current_index) < 5
337 ):
338 task_id = f"fill_autoplay_tracks_{queue_id}"
339 self.mass.call_later(5, self._fill_autoplay_tracks, queue_id, task_id=task_id)
340 self.signal_update(queue_id=queue_id)
341
342 @api_command(
343 "player_queues/dont_stop_the_music", required_scope=Scope.QUEUES_CONTROL, alias=True
344 )
345 def set_dont_stop_the_music(self, queue_id: str, dont_stop_the_music_enabled: bool) -> None:
346 """Backwards-compatible alias for the autoplay command, used by older clients."""
347 self.set_autoplay(queue_id, dont_stop_the_music_enabled)
348
349 @api_command("player_queues/repeat", required_scope=Scope.QUEUES_CONTROL)
350 async def set_repeat(self, queue_id: str, repeat_mode: RepeatMode) -> None:
351 """Configure repeat setting on the the queue."""
352 queue = self._queue_data[queue_id].queue
353 if queue.is_dynamic:
354 # a dynamic queue is an always-on flowing mix of its sources; repeat has no meaning here
355 raise InvalidCommand("Cannot change repeat while the queue is in dynamic mode")
356 if queue.repeat_mode == repeat_mode:
357 return # no change
358 queue.repeat_mode = repeat_mode
359 self.signal_update(queue_id)
360 self.update_next_item_on_player(queue_id)
361
362 @api_command("player_queues/crossfade", required_scope=Scope.QUEUES_CONTROL)
363 def set_crossfade(self, queue_id: str, crossfade_enabled: bool) -> None:
364 """Enable or disable crossfade on the queue."""
365 queue_data = self._queue_data[queue_id]
366 queue = queue_data.queue
367 effective_before = queue.crossfade_enabled
368 # the toggle always pins an explicit override, even when it matches the global default
369 queue_data.crossfade_override = crossfade_enabled
370 self._resolve_default_toggles(queue_data)
371 if queue.crossfade_enabled != effective_before:
372 # refresh the derived smart-fades indicator so the update we signal reflects the new state
373 queue.smart_fades_active = self.mass.streams.is_smart_fades_active(queue)
374 # the upcoming track is unchanged but the way it is streamed is not, so hand it over
375 # again
376 self.update_next_item_on_player(queue_id, force=True)
377 self.signal_update(queue_id)
378
379 @api_command("player_queues/overlay", required_scope=Scope.QUEUES_CONTROL)
380 async def set_overlay(
381 self,
382 queue_id: str,
383 enabled: bool | None = None,
384 source: str | None = None,
385 volume: int | None = None,
386 ) -> None:
387 """
388 Configure the audio overlay for the given queue.
389
390 The audio overlay mixes a looping sound effect (e.g. rain or white noise)
391 into the queue's audio stream. Changes take effect immediately: if the
392 queue is playing, playback is restarted from the current position.
393
394 :param queue_id: queue_id of the queue to configure.
395 :param enabled: Enable or disable the audio overlay. Omit to leave unchanged.
396 :param source: URI of the sound effect item to mix in. Omit to leave unchanged.
397 :param volume: Overlay loudness relative to the music in percent
398 (0-200, 100 = equally loud). Omit to leave unchanged.
399 """
400 queue = self._queue_data[queue_id].queue
401 changed = audible_change = False
402 if source is not None:
403 item = await self.mass.music.get_item_by_uri(source)
404 if item.media_type != MediaType.SOUND_EFFECT:
405 raise InvalidDataError("Audio overlay source must be a sound effect item")
406 mapping = ItemMapping.from_item(cast("SoundEffect", item))
407 if queue.overlay_source != mapping:
408 queue.overlay_source = mapping
409 changed = True
410 audible_change = queue.overlay_enabled
411 if volume is not None:
412 if not (0 <= volume <= 200):
413 raise InvalidDataError(f"Overlay volume must be between 0 and 200, got {volume}")
414 if queue.overlay_volume != volume:
415 queue.overlay_volume = volume
416 changed = True
417 audible_change |= queue.overlay_enabled
418 if enabled is not None and queue.overlay_enabled != enabled:
419 if enabled and queue.overlay_source is None:
420 raise InvalidCommand("Can not enable audio overlay: no overlay source selected")
421 queue.overlay_enabled = enabled
422 changed = audible_change = True
423 if not changed:
424 return
425 self.signal_update(queue_id)
426 if audible_change and queue.state == PlaybackState.PLAYING:
427 # restart playback from the current position so the change is heard
428 # immediately instead of after the player's audio buffer drains
429 await self.resume(queue_id)
430
431 # Two timebases are used in this controller when variable playback speed is in
432 # effect (atempo applied server-side):
433 # "stream-time" — seconds of audio the player has played (post-atempo).
434 # "media-time" — seconds of the original content the listener has heard.
435 # What the user expects to see on the progress bar and what
436 # we use for resume positions.
437 # Conversion: media-time = stream-time x playback_speed.
438 @api_command("player_queues/set_playback_speed", required_scope=Scope.QUEUES_CONTROL)
439 async def set_playback_speed(
440 self, queue_id: str, speed: float, queue_item_id: str | None = None
441 ) -> None:
442 """
443 Set the playback speed for the given queue item.
444
445 Variable playback speed is supported only for audiobooks and podcast episodes.
446
447 If queue_item_id is not provided,
448 the speed will be set for the current item in the queue.
449
450 :param queue_id: queue_id of the queue to configure.
451 :param speed: playback speed multiplier (0.5 to 3.0). 1.0 = normal speed.
452 """
453 if not (0.5 <= speed <= 3.0):
454 raise InvalidDataError(f"Playback speed must be between 0.5 and 3.0, got {speed}")
455 queue = self._queue_data[queue_id].queue
456 if not queue.current_item:
457 raise QueueEmpty("Cannot set playback speed: queue is empty")
458 queue_item_id = queue_item_id or queue.current_item.queue_item_id
459 queue_item = self.get_item(queue_id, queue_item_id)
460 if not queue_item:
461 raise InvalidDataError(f"Queue item {queue_item_id} not found in queue")
462 if queue_item.media_type not in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE):
463 raise InvalidCommand(
464 "Variable playback speed is only supported for audiobooks and podcast episodes"
465 )
466 if not queue_item.duration:
467 raise InvalidCommand("Cannot set playback speed for items with unknown duration")
468 current_speed = float(queue_item.extra_attributes.get("playback_speed") or 1.0)
469 if abs(current_speed - speed) < 0.001:
470 return # no change
471 # use extra_attributes of the queue item to store the playback speed
472 queue_item.extra_attributes["playback_speed"] = speed
473 # mirror onto the queue so corrected_elapsed_time advances in media-time
474 # immediately, before the next on_player_elapsed_time_corrected snapshot.
475 if queue.current_item and queue.current_item.queue_item_id == queue_item_id:
476 # close off the wallclock seconds that already ticked by at the old speed
477 # before switching, so corrected_elapsed_time doesn't multiply them by the new speed
478 if queue.state == PlaybackState.PLAYING:
479 queue.elapsed_time = queue.corrected_elapsed_time
480 queue.elapsed_time_last_updated = time.time()
481 queue.playback_speed = speed
482 self.signal_update(queue_id)
483 if queue.state == PlaybackState.PLAYING:
484 await self.resume(queue_id)
485
486 @api_command(
487 "player_queues/play_media", required_scope=Scope.QUEUES_CONTROL, allow_impersonation=True
488 )
489 async def play_media(
490 self,
491 queue_id: str,
492 media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
493 option: QueueOption | None = None,
494 radio_mode: bool = False,
495 start_item: PlayableMediaItemType | str | None = None,
496 sort_by: str | None = None,
497 start_from_beginning: bool = False,
498 shuffle: bool | None = None,
499 ) -> None:
500 """
501 Play media item(s) on the given queue.
502
503 :param queue_id: The queue_id of the queue to play media on.
504 :param media: Media that should be played (MediaItem(s) and/or uri's).
505 :param option: Which enqueue mode to use.
506 :param radio_mode: Deprecated — translated to a radio_playlist:// dynamic playlist;
507 prefer enqueuing that URI directly.
508 :param start_item: Optional item to start the playlist or album from.
509 :param sort_by: Optional sort key to order tracks before applying start_item.
510 :param start_from_beginning: Start a podcast episode at position 0, ignoring any
511 saved resume position. The stored progress itself is left untouched.
512 :param shuffle: Play the media shuffled (or explicitly in order). Only applies to the
513 options that start playing right away (play/replace), and never to a dynamic source
514 (an always-on smart mix). Omit to follow the queue's own shuffle setting, which media
515 with an order of its own (album, podcast, episode, audiobook, audio source) switches
516 off; the first item of a batch decides for the whole batch.
517 """
518 self._check_player_permission(queue_id)
519 if not self.get(queue_id):
520 raise PlayerUnavailableError(f"Queue {queue_id} is not available")
521 # A live source is not queue content: it plays on the player while the queue
522 # keeps its own items. Selecting it is the real operation, so a play request
523 # naming one is forwarded there rather than enqueued.
524 if (source_uri := await _resolve_audio_source_request(media)) is not None:
525 await self.mass.players.select_source(queue_id, source_uri)
526 return
527 # Lock is acquired by the @handle_play_action decorator on the internal handler
528 await self._handle_play_media(
529 queue_id,
530 media,
531 option,
532 radio_mode,
533 start_item,
534 sort_by,
535 start_from_beginning,
536 shuffle,
537 )
538
539 @api_command("player_queues/move_item", required_scope=Scope.QUEUES_CONTROL)
540 def move_item(self, queue_id: str, queue_item_id: str, pos_shift: int = 1) -> None:
541 """
542 Move queue item x up/down the queue.
543
544 - queue_id: id of the queue to process this request.
545 - queue_item_id: the item_id of the queueitem that needs to be moved.
546 - pos_shift: move item x positions down if positive value
547 - pos_shift: move item x positions up if negative value
548 - pos_shift: move item to the front of the upcoming items if 0
549 """
550 queue = self._queue_data[queue_id].queue
551 item_index = self.index_by_id(queue_id, queue_item_id)
552 if item_index is None:
553 raise InvalidDataError(f"Item {queue_item_id} not found in queue")
554 # the player owns everything up to this index and keeps playing it whatever the queue
555 # says, so it is out of bounds as both a source and a destination
556 boundary_index = committed_index(queue) if queue.index_in_buffer is not None else None
557 if boundary_index is not None and item_index <= boundary_index:
558 msg = f"{item_index} is already played/buffered"
559 raise IndexError(msg)
560
561 queue_items = self._queue_data[queue_id].items
562 queue_items = queue_items.copy()
563
564 if pos_shift == 0 and queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
565 new_index = (boundary_index if boundary_index is not None else 0) + 1
566 elif pos_shift == 0:
567 new_index = queue.current_index or 0
568 else:
569 new_index = item_index + pos_shift
570 lowest_index = (
571 boundary_index + 1 if boundary_index is not None else (queue.current_index or 0)
572 )
573 if new_index < lowest_index or new_index > len(queue_items):
574 return
575 # move the item in the list
576 queue_items.insert(new_index, queue_items.pop(item_index))
577 self.update_items(queue_id, queue_items)
578
579 @api_command("player_queues/move_item_end", required_scope=Scope.QUEUES_CONTROL)
580 def move_item_end(self, queue_id: str, queue_item_id: str) -> None:
581 """
582 Move queue item to the end the queue.
583
584 - queue_id: id of the queue to process this request.
585 - queue_item_id: the item_id of the queueitem that needs to be moved.
586 """
587 queue = self._queue_data[queue_id].queue
588 item_index = self.index_by_id(queue_id, queue_item_id)
589 if item_index is None:
590 raise InvalidDataError(f"Item {queue_item_id} not found in queue")
591 boundary_index = committed_index(queue) if queue.index_in_buffer is not None else None
592 if boundary_index is not None and item_index <= boundary_index:
593 msg = f"{item_index} is already played/buffered"
594 raise IndexError(msg)
595
596 queue_items = self._queue_data[queue_id].items
597 if item_index == (len(queue_items) - 1):
598 return
599 queue_items = queue_items.copy()
600
601 new_index = len(self._queue_data[queue_id].items) - 1
602
603 # move the item in the list
604 queue_items.insert(new_index, queue_items.pop(item_index))
605 self.update_items(queue_id, queue_items)
606
607 @api_command("player_queues/delete_item", required_scope=Scope.QUEUES_CONTROL)
608 def delete_item(self, queue_id: str, item_id_or_index: int | str) -> None:
609 """Delete item (by id or index) from the queue."""
610 if isinstance(item_id_or_index, str):
611 item_index = self.index_by_id(queue_id, item_id_or_index)
612 if item_index is None:
613 raise InvalidDataError(f"Item {item_id_or_index} not found in queue")
614 else:
615 item_index = item_id_or_index
616 queue = self._queue_data[queue_id].queue
617 boundary_index = committed_index(queue) if queue.index_in_buffer is not None else None
618 if boundary_index is not None and item_index <= boundary_index:
619 # ignore request if track already loaded in the buffer
620 # the frontend should guard so this is just in case
621 self.logger.warning("delete requested for item already loaded in buffer")
622 return
623 queue_items = self._queue_data[queue_id].items.copy()
624 queue_items.pop(item_index)
625 self.update_items(queue_id, queue_items)
626
627 @api_command("player_queues/clear", required_scope=Scope.QUEUES_CONTROL)
628 def clear(self, queue_id: str, skip_stop: bool = False) -> None:
629 """Clear all items in the queue, switching shuffle off with them."""
630 self._clear(queue_id, skip_stop)
631 # clearing is an explicit "start over" gesture by the user, so a shuffle that belonged to
632 # the discarded content must not carry over into whatever is played next
633 self._reset_shuffle(queue_id)
634
635 def mark_ended(self, queue_id: str) -> None:
636 """
637 Mark a queue as played to its end, keeping its items so it can be replayed.
638
639 The playback position is parked on the last item rather than cleared: a null index is
640 indistinguishable from a queue that was loaded but never started, and an index past the
641 end is silently misread by everything that does arithmetic on it. `ended` is what tells
642 clients the queue finished, and pressing play starts it over from the first item.
643
644 :param queue_id: The queue_id of the queue that reached its end.
645 """
646 queue_data = self._queue_data[queue_id]
647 queue = queue_data.queue
648 if not queue_data.items:
649 # nothing to replay, so there is nothing to advertise as finished either
650 self._clear(queue_id)
651 return
652 self.mass.streams.audio_processing.clear(queue_id)
653 queue.ended = True
654 queue.current_index = len(queue_data.items) - 1
655 queue.current_item = queue_data.items[-1]
656 queue.next_item = None
657 queue.elapsed_time = 0
658 queue.elapsed_time_last_updated = time.time()
659 queue.index_in_buffer = None
660 queue.resume_pos = 0
661 self.mass.create_task(self._cleanup_queue_audio_data(queue_id))
662 self.signal_update(queue_id)
663
664 @api_command("player_queues/save_as_playlist", required_scope=Scope.LIBRARY_WRITE)
665 async def save_as_playlist(self, queue_id: str, name: str) -> BackgroundTask:
666 """
667 Save the current queue items as a new playlist.
668
669 :param queue_id: The queue_id of the queue to save.
670 :param name: The name for the new playlist.
671 """
672 if not self.get(queue_id):
673 raise PlayerUnavailableError(f"Queue {queue_id} is not available")
674 queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
675 if not queue_items:
676 raise QueueEmpty("Cannot save an empty queue as a playlist.")
677 # collect URIs from queue items that are playlist-compatible
678 uris: list[str] = []
679 for item in queue_items:
680 if item.uri and item.media_type in PLAYLIST_MEDIA_TYPES:
681 uris.append(item.uri)
682 if not uris:
683 raise InvalidDataError("No valid items in queue to save as playlist.")
684 playlist = await self.mass.music.playlists.create_playlist(name)
685 return await self.mass.music.playlists.add_playlist_tracks(playlist.item_id, uris)
686
687 @api_command("player_queues/stop", required_scope=Scope.QUEUES_CONTROL)
688 async def stop(self, queue_id: str) -> None:
689 """
690 Handle STOP command for given queue.
691
692 - queue_id: queue_id of the playerqueue to handle the command.
693 """
694 self._check_player_permission(queue_id)
695 await self._handle_stop(queue_id)
696
697 @api_command("player_queues/play", required_scope=Scope.QUEUES_CONTROL)
698 async def play(self, queue_id: str) -> None:
699 """
700 Handle PLAY command for given queue.
701
702 :param queue_id: queue_id of the playerqueue to handle the command.
703 """
704 self._check_player_permission(queue_id)
705 if not self.get(queue_id):
706 raise PlayerUnavailableError(f"Queue {queue_id} is not available")
707 await self._handle_play(queue_id)
708
709 @api_command("player_queues/pause", required_scope=Scope.QUEUES_CONTROL)
710 async def pause(self, queue_id: str) -> None:
711 """
712 Handle PAUSE command for given queue.
713
714 - queue_id: queue_id of the playerqueue to handle the command.
715 """
716 self._check_player_permission(queue_id)
717 # cancel any pending play_index calls for this queue to prevent conflicts
718 self.mass.cancel_timer(f"queue_play_index_{queue_id}")
719 self._set_transitioning(queue_id, False)
720 if not (queue := self.get(queue_id)):
721 return
722 queue_active = queue.active
723 if queue.active and queue.state == PlaybackState.PLAYING:
724 queue.resume_pos = int(queue.corrected_elapsed_time)
725 # Use internal handler to avoid circular redirect
726 # (cmd_pause redirects to queue.pause, which calls cmd_pause again)
727 await self.mass.players._handle_cmd_pause(queue_id)
728
729 async def _watch_pause(player: Player) -> None:
730 count = 0
731 # wait for pause
732 while count < 5 and player.state.playback_state == PlaybackState.PLAYING:
733 count += 1
734 await asyncio.sleep(1)
735 # wait for unpause
736 if player.state.playback_state != PlaybackState.PAUSED:
737 return
738 count = 0
739 while count < 30 and player.state.playback_state == PlaybackState.PAUSED:
740 count += 1
741 await asyncio.sleep(1)
742 # if player is still paused when the limit is reached, send stop
743 if player.state.playback_state == PlaybackState.PAUSED:
744 await self.stop(queue_id)
745
746 # we auto stop a player from paused when its paused for 30 seconds
747 if (
748 queue_active
749 and (queue_player := self.mass.players.get_player(queue_id))
750 and not queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS)
751 ):
752 self.mass.create_task(_watch_pause(queue_player))
753
754 @api_command("player_queues/play_pause", required_scope=Scope.QUEUES_CONTROL)
755 async def play_pause(self, queue_id: str) -> None:
756 """
757 Toggle play/pause on given playerqueue.
758
759 - queue_id: queue_id of the queue to handle the command.
760 """
761 if (queue := self.get(queue_id)) and queue.state == PlaybackState.PLAYING:
762 await self.pause(queue_id)
763 return
764 await self.play(queue_id)
765
766 @api_command("player_queues/next", required_scope=Scope.QUEUES_CONTROL)
767 @handle_play_action
768 async def next(self, queue_id: str) -> None:
769 """
770 Handle NEXT TRACK command for given queue.
771
772 :param queue_id: queue_id of the queue to handle the command.
773 """
774 self._check_player_permission(queue_id)
775 if (queue := self.get(queue_id)) is None or not queue.active:
776 raise InvalidCommand(f"Queue {queue_id} is not active")
777 self._set_transitioning(queue_id, True)
778 idx = self._queue_data[queue_id].queue.current_index
779 if idx is None:
780 self.logger.warning("Queue %s has no current index", queue.display_name)
781 self._set_transitioning(queue_id, False)
782 return
783 next_index = self._get_next_index(queue_id, idx, True)
784 if next_index is None:
785 self._set_transitioning(queue_id, False)
786 return
787
788 # immediately update current item so UI shows the new track right away
789 queue.current_index = next_index
790 queue.current_item = self.get_item(queue_id, next_index)
791 queue.elapsed_time = 0
792 queue.elapsed_time_last_updated = time.time()
793 self.signal_update(queue_id)
794 if queue_player := self.mass.players.get_player(queue_id, True):
795 queue_player.update_state()
796
797 # debounce rapid next button presses using call_later
798 self.mass.call_later(
799 1,
800 self.play_index,
801 queue_id,
802 next_index,
803 task_id=f"queue_play_index_{queue_id}",
804 )
805
806 @api_command("player_queues/previous", required_scope=Scope.QUEUES_CONTROL)
807 @handle_play_action
808 async def previous(self, queue_id: str) -> None:
809 """
810 Handle PREVIOUS TRACK command for given queue.
811
812 :param queue_id: queue_id of the queue to handle the command.
813 """
814 self._check_player_permission(queue_id)
815 if (queue := self.get(queue_id)) is None or not queue.active:
816 raise InvalidCommand(f"Queue {queue_id} is not active")
817 self._set_transitioning(queue_id, True)
818 current_index = self._queue_data[queue_id].queue.current_index
819 if current_index is None:
820 self._set_transitioning(queue_id, False)
821 return
822 prev_index = int(current_index)
823 # restart current track if elapsed > 5s, otherwise go to previous
824 if self._queue_data[queue_id].queue.elapsed_time < 5:
825 prev_index = max(current_index - 1, 0)
826
827 # immediately update current item so UI shows the new track right away
828 queue.current_index = prev_index
829 queue.current_item = self.get_item(queue_id, prev_index)
830 queue.elapsed_time = 0
831 queue.elapsed_time_last_updated = time.time()
832 self.signal_update(queue_id)
833 if queue_player := self.mass.players.get_player(queue_id, True):
834 queue_player.update_state()
835
836 # debounce rapid previous button presses using call_later
837 self.mass.call_later(
838 1,
839 self.play_index,
840 queue_id,
841 prev_index,
842 task_id=f"queue_play_index_{queue_id}",
843 )
844
845 @api_command("player_queues/skip", required_scope=Scope.QUEUES_CONTROL)
846 async def skip(self, queue_id: str, seconds: int = 10) -> None:
847 """
848 Handle SKIP command for given queue.
849
850 - queue_id: queue_id of the queue to handle the command.
851 - seconds: number of seconds to skip in track. Use negative value to skip back.
852 """
853 if (queue := self.get(queue_id)) is None or not queue.active:
854 raise InvalidCommand(f"Queue {queue_id} is not active")
855 await self.seek(queue_id, int(self._queue_data[queue_id].queue.elapsed_time + seconds))
856
857 @api_command("player_queues/seek", required_scope=Scope.QUEUES_CONTROL)
858 async def seek(self, queue_id: str, position: int = 10) -> None:
859 """
860 Handle SEEK command for given queue.
861
862 - queue_id: queue_id of the queue to handle the command.
863 - position: position in seconds to seek to in the current playing item.
864 """
865 if (queue := self.get(queue_id)) is None or not queue.active:
866 raise InvalidCommand(f"Queue {queue_id} is not active")
867 queue_player = self.mass.players.get_player(queue_id, True)
868 if queue_player is None:
869 raise PlayerUnavailableError(f"Player {queue_id} is not available")
870 if not queue.current_item:
871 raise InvalidCommand(f"Queue {queue_player.state.name} has no item(s) loaded.")
872 if not queue.current_item.duration:
873 raise InvalidCommand("Can not seek items without duration.")
874 position = max(0, int(position))
875 if position > queue.current_item.duration:
876 raise InvalidCommand("Can not seek outside of duration range.")
877 if queue.current_index is None:
878 raise InvalidCommand(f"Queue {queue_player.state.name} has no current index.")
879 # Publish the seek target before rebuilding the stream to prevent progress snapback.
880 queue.elapsed_time = position
881 queue.elapsed_time_last_updated = time.time()
882 self.signal_update(queue_id)
883 await self.play_index(queue_id, queue.current_index, seek_position=position)
884
885 @api_command("player_queues/resume", required_scope=Scope.QUEUES_CONTROL)
886 @handle_play_action
887 async def resume(self, queue_id: str, fade_in: bool | None = None) -> None:
888 """
889 Handle RESUME command for given queue.
890
891 - queue_id: queue_id of the queue to handle the command.
892 """
893 self._check_player_permission(queue_id)
894 queue = self._queue_data[queue_id].queue
895 queue_items = self._queue_data[queue_id].items
896 resume_item = queue.current_item
897 if queue.state == PlaybackState.PLAYING:
898 # resume requested while already playing,
899 # use current position as resume position
900 resume_pos = queue.corrected_elapsed_time
901 fade_in = False
902 else:
903 resume_pos = queue.resume_pos or queue.elapsed_time
904
905 if queue.ended and len(queue_items) > 0:
906 # the queue played to its end and is parked on its last item,
907 # so pressing play starts it over from the beginning
908 resume_item = queue_items[0]
909 resume_pos = 0
910 elif not resume_item and queue.current_index is not None and len(queue_items) > 0:
911 resume_item = self.get_item(queue_id, queue.current_index)
912 resume_pos = 0
913 elif not resume_item and queue.current_index is None and len(queue_items) > 0:
914 # items available in queue but no previous track, start at 0
915 resume_item = self.get_item(queue_id, 0)
916 resume_pos = 0
917
918 if resume_item is not None:
919 queue_player = self.mass.players.get_player(queue_id)
920 if queue_player is None:
921 raise PlayerUnavailableError(f"Player {queue_id} is not available")
922 if (
923 fade_in is None
924 and queue_player.state.playback_state == PlaybackState.IDLE
925 and (time.time() - queue.elapsed_time_last_updated) > 60
926 ):
927 # enable fade in effect if the player is idle for a while
928 fade_in = resume_pos > 0
929 if resume_item.media_type == MediaType.RADIO:
930 # we're not able to skip in online radio so this is pointless
931 resume_pos = 0
932 await self.play_index(
933 queue_id, resume_item.queue_item_id, int(resume_pos), fade_in or False
934 )
935 else:
936 msg = f"Resume queue requested but queue {queue.display_name} is empty"
937 raise QueueEmpty(msg)
938
939 @api_command("player_queues/play_index", required_scope=Scope.QUEUES_CONTROL)
940 @handle_play_action
941 async def play_index( # noqa: PLR0915
942 self,
943 queue_id: str,
944 index: int | str,
945 seek_position: int = 0,
946 fade_in: bool = False,
947 ) -> None:
948 """Play item at index (or item_id) X in queue."""
949 self._check_player_permission(queue_id)
950 # cancel any pending play_index calls for this queue to prevent conflicts
951 self.mass.cancel_timer(f"queue_play_index_{queue_id}")
952 # we set a flag to notify the update logic that we're transitioning to a new track
953 self._set_transitioning(queue_id, True)
954 try:
955 queue_data = self._queue_data[queue_id]
956 queue = queue_data.queue
957 queue.resume_pos = 0
958 # A queue picked up from its end plays its items over from the start, so a resume point
959 # left on an audiobook/episode must not pull it back to where it was left off. The flag
960 # itself is only cleared once an item actually loaded below, so a start that never got
961 # off the ground leaves the queue finished instead of stranding it without a position.
962 restarting_ended_queue = queue.ended
963 if isinstance(index, str):
964 temp_index = self.index_by_id(queue_id, index)
965 if temp_index is None:
966 raise InvalidDataError(f"Item {index} not found in queue")
967 index = temp_index
968 # At this point index is guaranteed to be int
969 queue.index_in_buffer = index
970 queue_data.flow_mode_stream_log = []
971 queue_data.flow_buffer_completed = None
972 queue_data.flow_queue_exhausted = None
973 target_player = self.mass.players.get_player(queue_id)
974 if target_player is None:
975 raise PlayerUnavailableError(f"Player {queue_id} is not available")
976 queue_data.next_item_id_enqueued = None
977 # always update session id when we start a new playback session
978 queue_data.session_id = shortuuid.random(length=8)
979 self.mass.streams.audio_processing.start_session(
980 queue_id,
981 queue_data.session_id,
982 )
983 # handle resume point of audiobook(chapter) or podcast(episode)
984 if (
985 not seek_position
986 and not restarting_ended_queue
987 and (queue_item := self.get_item(queue_id, index))
988 and (resume_position_ms := getattr(queue_item.media_item, "resume_position_ms", 0))
989 ):
990 # the client may have fetched the item before its duration was known
991 await self._restore_probed_duration(queue_item)
992 if queue_item.duration or getattr(queue_item.media_item, "duration", 0):
993 seek_position = max(0, int((resume_position_ms - 500) / 1000))
994 else:
995 # seeking needs a duration, which is determined while streaming
996 self.logger.debug(
997 "Can not resume %s at %ss: its duration is not known (yet)",
998 queue_item.name,
999 int(resume_position_ms / 1000),
1000 )
1001
1002 # restore the persisted playback speed for a freshly queued audiobook/episode
1003 # (an in-session item already carries its speed in extra_attributes)
1004 if (
1005 (queue_item := self.get_item(queue_id, index))
1006 and queue_item.media_item is not None
1007 and queue_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1008 and "playback_speed" not in queue_item.extra_attributes
1009 ):
1010 stored_speed = await self.mass.music.get_playback_speed(
1011 cast("Audiobook | PodcastEpisode", queue_item.media_item),
1012 userid=queue_data.userid,
1013 )
1014 if stored_speed != 1.0:
1015 queue_item.extra_attributes["playback_speed"] = stored_speed
1016
1017 # try to load the item, retry with next item if it fails
1018 for attempt in range(5):
1019 try:
1020 queue_item = self.get_item(queue_id, index)
1021 if not queue_item:
1022 continue # guard
1023 await self._load_item(
1024 queue_item,
1025 is_start=True,
1026 seek_position=seek_position if attempt == 0 else 0,
1027 fade_in=fade_in if attempt == 0 else False,
1028 )
1029 # if we reach this point, loading the item succeeded, break the loop
1030 queue.current_index = index
1031 queue.current_item = queue_item
1032 # playback is under way, so the queue is no longer sitting at its end
1033 queue.ended = False
1034 # reset the elapsed clock together with the item switch (like
1035 # next/previous do), so queue updates signaled before the player
1036 # reports position don't carry the previous item's elapsed_time
1037 queue.elapsed_time = seek_position if attempt == 0 else 0
1038 queue.elapsed_time_last_updated = time.time()
1039 break
1040 except (MediaNotFoundError, AudioError) as err:
1041 item_name = queue_item.name if queue_item else "unknown"
1042 if isinstance(err, ProviderStreamLimitError):
1043 # the requested item is playable, its provider is just at capacity:
1044 # report that instead of silently advancing to another item
1045 self.logger.error("%s", err)
1046 await self.stop(queue_id)
1047 raise
1048 # Only MediaNotFoundError (item unreachable) is persistent;
1049 # keep AudioError items available so a retry can resurface
1050 # the same actionable error.
1051 if queue_item and isinstance(err, MediaNotFoundError):
1052 queue_item.available = False
1053 next_index = self._get_next_index(queue_id, index, allow_repeat=False)
1054 if next_index is None:
1055 # Surface an AudioError's own (actionable) message;
1056 # MediaNotFoundError gets the generic wording.
1057 if isinstance(err, AudioError) and str(err):
1058 msg = str(err)
1059 else:
1060 msg = f"Playback failed for {item_name} - no more tracks available"
1061 self.logger.error(msg)
1062 await self.stop(queue_id)
1063 raise MediaNotFoundError(msg) from err
1064 self.logger.warning(
1065 "Skipping unplayable item %s",
1066 item_name,
1067 )
1068 index = next_index
1069 else:
1070 # all attempts to find a playable item failed
1071 await self.stop(queue_id)
1072 raise MediaNotFoundError("No playable item found to start playback")
1073
1074 # Reset flow_mode - the streams controller will set it if flow mode is used.
1075 queue.flow_mode = False
1076 player_media = await self.player_media_from_queue_item(queue_item)
1077 # Hold the play action until the player confirms playback so the UI keeps
1078 # showing the command as in progress instead of falling back to a play button
1079 # for the time the player still needs to connect and start. The queue update
1080 # for the new item goes out first, so the item shows while it is starting.
1081 async with self.mass.players.wait_for_player_update(
1082 queue_id,
1083 attribute_name="playback_state",
1084 attribute_value=PlaybackState.PLAYING,
1085 timeout=PLAYBACK_START_TIMEOUT,
1086 ):
1087 await self.mass.players.play_media(queue_id, player_media)
1088 queue.current_index = index
1089 queue.current_item = queue_item
1090 self.signal_update(queue_id)
1091 finally:
1092 self._set_transitioning(queue_id, False)
1093
1094 @api_command("player_queues/transfer", required_scope=Scope.QUEUES_CONTROL)
1095 async def transfer_queue(
1096 self,
1097 source_queue_id: str,
1098 target_queue_id: str,
1099 auto_play: bool | None = None,
1100 ) -> None:
1101 """Transfer queue to another queue."""
1102 if not (source_queue := self.get(source_queue_id)):
1103 raise PlayerUnavailableError(f"Queue {source_queue_id} is not available")
1104 if not (target_queue := self.get(target_queue_id)):
1105 raise PlayerUnavailableError(f"Queue {target_queue_id} is not available")
1106 if auto_play is None:
1107 auto_play = source_queue.state == PlaybackState.PLAYING
1108
1109 target_player = self.mass.players.get_player(target_queue_id)
1110 if target_player is None:
1111 raise PlayerUnavailableError(f"Player {target_queue_id} is not available")
1112 # refuse targets that can never render audio (display/visualizer/lighting clients)
1113 # before anything is mutated, so a bad target does not destroy the source queue
1114 if target_player.state.type not in PLAYBACK_TARGET_TYPES:
1115 raise PlayerCommandFailed(f"Player {target_player.name} is not capable of playback")
1116 if target_player.state.active_group or target_player.state.synced_to:
1117 # edge case: the user wants to move playback from the group as a whole, to a single
1118 # player in the group or it is grouped and the command targeted at the single player.
1119 # We need to dissolve the group/sync first, and wait for the state to actually
1120 # propagate before we hand the queue over to the target player.
1121 group_id = target_player.state.active_group or target_player.state.synced_to
1122 assert group_id is not None # checked in if condition above
1123 # For an ad-hoc sync group (target is a sync member of a regular leader),
1124 # ungroup the target itself so only it is freed - ungrouping the leader would
1125 # transfer leadership to a remaining member and recurse back into this method.
1126 # For a virtual group player (active_group), release the group so its static
1127 # members are handled correctly.
1128 ungroup_target = (
1129 target_queue_id
1130 if target_player.state.synced_to and not target_player.state.active_group
1131 else group_id
1132 )
1133 async with self.mass.players.wait_for_player_update(
1134 target_queue_id,
1135 attribute_name=(
1136 "active_group" if target_player.state.active_group else "synced_to"
1137 ),
1138 attribute_value=None,
1139 timeout=5,
1140 ):
1141 await self.mass.players.cmd_ungroup(ungroup_target)
1142
1143 # capture source state before stopping (stop resets these)
1144 source_items = self._queue_data[source_queue_id].items
1145 if source_queue.state == PlaybackState.PLAYING:
1146 # use the live playback clock while actively playing
1147 source_resume_pos = int(source_queue.corrected_elapsed_time)
1148 else:
1149 # when not playing the live clock is stale, so use the stored resume position
1150 source_resume_pos = int(source_queue.resume_pos or source_queue.elapsed_time or 0)
1151 source_current_index = source_queue.current_index
1152 source_current_item = source_queue.current_item
1153
1154 # stop the source player synchronously to prevent the async stop from
1155 # clear() racing with the target's sync group formation/protocol switching
1156 if source_queue.state != PlaybackState.IDLE:
1157 await self.stop(source_queue_id)
1158
1159 target_queue.repeat_mode = source_queue.repeat_mode
1160 target_queue.shuffle_enabled = source_queue.shuffle_enabled
1161 # carry over the pinned overrides (or follow-global state) and re-resolve the target
1162 self._queue_data[target_queue_id].crossfade_override = self._queue_data[
1163 source_queue_id
1164 ].crossfade_override
1165 self._queue_data[target_queue_id].autoplay_override = self._queue_data[
1166 source_queue_id
1167 ].autoplay_override
1168 self._resolve_default_toggles(self._queue_data[target_queue_id])
1169 # refresh the derived smart-fades indicator for the target's own config/availability
1170 target_queue.smart_fades_active = self.mass.streams.is_smart_fades_active(target_queue)
1171 self._queue_data[target_queue_id].source_items = list(
1172 self._queue_data[source_queue_id].source_items
1173 )
1174 target_queue.sources = list(source_queue.sources)
1175 target_queue.is_dynamic = source_queue.is_dynamic
1176 target_queue.smart_shuffle_active = self.is_smart_shuffle_active(target_queue)
1177 self._queue_data[target_queue_id].enqueued_media_items = list(
1178 self._queue_data[source_queue_id].enqueued_media_items
1179 )
1180 self._queue_data[target_queue_id].credited_albums = set(
1181 self._queue_data[source_queue_id].credited_albums
1182 )
1183 target_queue.resume_pos = source_resume_pos
1184 target_queue.current_index = source_current_index
1185 if source_current_item:
1186 target_queue.current_item = source_current_item
1187 target_queue.current_item.queue_id = target_queue_id
1188 self._clear(source_queue_id, skip_stop=True)
1189
1190 await self.load(target_queue_id, source_items, keep_remaining=False, keep_played=False)
1191 for item in source_items:
1192 item.queue_id = target_queue_id
1193 self.update_items(target_queue_id, source_items)
1194 if auto_play:
1195 await self.resume(target_queue_id)
1196
1197 # Interaction with player
1198
1199 async def on_player_register(self, player: Player) -> None:
1200 """Register PlayerQueue for given player/queue id."""
1201 queue_id = player.player_id
1202 queue_data: PlayerQueueData | None = None
1203 # try to restore previous state
1204 try:
1205 if prev_state := await self.mass.cache.get(
1206 key=queue_id,
1207 provider=self.domain,
1208 category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1209 ):
1210 prev_items = await self.mass.cache.get(
1211 key=queue_id,
1212 provider=self.domain,
1213 category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1214 default=[],
1215 )
1216 queue_data = PlayerQueueData.from_cache(prev_state, prev_items)
1217 except Exception as err:
1218 self.logger.warning(
1219 "Failed to restore the queue(items) for %s - %s",
1220 player.state.name,
1221 str(err),
1222 )
1223 # Reset to clean state on failure
1224 queue_data = None
1225 if queue_data is None:
1226 queue_data = PlayerQueueData(
1227 queue=PlayerQueue(
1228 queue_id=queue_id,
1229 active=False,
1230 display_name=player.state.name,
1231 available=player.state.available,
1232 items=0,
1233 )
1234 )
1235 # new queues and queues restored without a pinned override follow the global defaults
1236 self._resolve_default_toggles(queue_data)
1237
1238 self._queue_data[queue_id] = queue_data
1239 # always call update to calculate state etc
1240 self.on_player_update(player, {})
1241 self.mass.signal_event(EventType.QUEUE_ADDED, object_id=queue_id, data=queue_data.queue)
1242
1243 def on_player_update(
1244 self,
1245 player: Player,
1246 changed_values: dict[str, tuple[Any, Any]],
1247 ) -> None:
1248 """
1249 Call when a PlayerQueue needs to be updated (e.g. when player updates).
1250
1251 NOTE: This is called every second if the player is playing.
1252 """
1253 if player.type == PlayerType.PROTOCOL:
1254 # protocol players do not have a queue on their own
1255 return
1256 queue_id = player.player_id
1257 if (queue := self.get(queue_id)) is None:
1258 # race condition
1259 return
1260 if player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1261 # do nothing while the announcement is in progress
1262 return
1263 # determine if this queue is currently active for this player
1264 queue.active = player.state.active_source in (queue.queue_id, None)
1265 if not queue.active and self._queue_data[queue_id].prev_state is None:
1266 queue.state = PlaybackState.IDLE
1267 # return early if the queue is not active and we have no previous state
1268 return
1269 if self._queue_data[queue_id].transitioning:
1270 # we're currently transitioning to a new track,
1271 # ignore updates from the player during this time
1272 return
1273 # queue is active and preflight checks passed, update the queue details
1274 self._update_queue_from_player(player)
1275
1276 def on_player_elapsed_time_corrected(self, player: Player) -> None:
1277 """Correct the queue's timing base if the player's real elapsed_time diverged."""
1278 if player.type == PlayerType.PROTOCOL:
1279 return
1280 queue_id = player.player_id
1281 if (queue := self.get(queue_id)) is None:
1282 return
1283 if not queue.active:
1284 return
1285 player_elapsed = player.state.corrected_elapsed_time
1286 if player_elapsed is None:
1287 return
1288 now = time.time()
1289 # queue.elapsed_time is stored in media-time so it can be displayed and
1290 # used as a resume position directly. The player reports stream-time
1291 # (post-atempo), so we scale by the current item's playback_speed.
1292 speed = get_current_playback_speed(queue)
1293 if queue.flow_mode:
1294 # _get_flow_queue_stream_index returns media-time in the current item
1295 # using each playlog entry's recorded speed.
1296 _, elapsed_time = self._get_flow_queue_stream_index(queue, player)
1297 else:
1298 elapsed_time = player_elapsed * speed
1299 if queue.current_item and queue.current_item.streamdetails:
1300 if seek_pos := queue.current_item.streamdetails.seek_position:
1301 elapsed_time += seek_pos
1302 queue.elapsed_time = elapsed_time
1303 queue.elapsed_time_last_updated = now
1304 queue.playback_speed = speed
1305 self.mass.signal_event(
1306 EventType.QUEUE_TIME_UPDATED,
1307 object_id=queue_id,
1308 data=queue.elapsed_time,
1309 )
1310
1311 def on_player_remove(self, player_id: str, permanent: bool) -> None:
1312 """Call when a player is removed from the registry."""
1313 self.mass.streams.audio_processing.clear(player_id)
1314 # cancel any pending play_index calls for this queue to prevent conflicts
1315 self.mass.cancel_timer(f"queue_play_index_{player_id}")
1316 # cancel a pending debounced cache write AND an already-started one, so neither can
1317 # recreate a deleted entry after the player is gone (the timer becomes a task once it fires)
1318 self.mass.cancel_timer(f"save_queue_cache_{player_id}")
1319 self.mass.cancel_task(f"save_queue_cache_{player_id}")
1320 self._set_transitioning(player_id, False)
1321 if permanent:
1322 self.purge_saved_queue(player_id)
1323 self._queue_data.pop(player_id, None)
1324 self._managed_pool.forget(player_id)
1325
1326 def purge_saved_queue(self, queue_id: str) -> None:
1327 """Delete the persisted state and items of the given queue."""
1328 for category in (CACHE_CATEGORY_PLAYER_QUEUE_STATE, CACHE_CATEGORY_PLAYER_QUEUE_ITEMS):
1329 # a removal runs both the player teardown and the config cleanup, so keep the
1330 # delete to one task per category instead of one per caller
1331 self.mass.create_task(
1332 self.mass.cache.delete(
1333 key=queue_id,
1334 provider=self.domain,
1335 category=category,
1336 ),
1337 task_id=f"purge_saved_queue_{queue_id}_{category}",
1338 )
1339
1340 async def load_next_queue_item(
1341 self,
1342 queue_id: str,
1343 current_item_id: str,
1344 ) -> QueueItem:
1345 """
1346 Call when a player wants the next queue item to play.
1347
1348 Raises QueueEmpty if there are no more tracks left.
1349 """
1350 queue = self.get(queue_id)
1351 if not queue:
1352 msg = f"PlayerQueue {queue_id} is not available"
1353 raise PlayerUnavailableError(msg)
1354 cur_index = self.index_by_id(queue_id, current_item_id)
1355 if cur_index is None:
1356 # this is just a guard for bad data
1357 raise QueueEmpty("Invalid item id for queue given.")
1358 next_item: QueueItem | None = None
1359 idx = 0
1360 while True:
1361 next_index = self._get_next_index(queue_id, cur_index + idx)
1362 if next_index is None:
1363 raise QueueEmpty("No more tracks left in the queue.")
1364 queue_item = self.get_item(queue_id, next_index)
1365 if queue_item is None:
1366 raise QueueEmpty("No more tracks left in the queue.")
1367 if idx >= 10:
1368 # we only allow 10 retries to prevent infinite loops
1369 raise QueueEmpty("No more (playable) tracks left in the queue.")
1370 try:
1371 await self._load_item(queue_item)
1372 # we're all set, this is our next item
1373 next_item = queue_item
1374 break
1375 except ProviderStreamLimitError:
1376 # transient source capacity, do not burn a playable item over it
1377 raise
1378 except MediaNotFoundError, AudioError:
1379 # No stream details found, skip this QueueItem
1380 self.logger.warning(
1381 "Skipping unplayable item %s (%s)", queue_item.name, queue_item.uri
1382 )
1383 queue_item.available = False
1384 idx += 1
1385 if idx != 0:
1386 # we skipped some items, signal a queue items update
1387 self.update_items(queue_id, self._queue_data[queue_id].items)
1388 if next_item is None:
1389 raise QueueEmpty("No more (playable) tracks left in the queue.")
1390
1391 # carry playback_speed forward across consecutive audiobook/podcast items
1392 current_item = self.get_item(queue_id, current_item_id)
1393 if (
1394 current_item
1395 and current_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1396 and next_item.media_type in (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
1397 ):
1398 next_item.extra_attributes["playback_speed"] = current_item.extra_attributes.get(
1399 "playback_speed", 1.0
1400 )
1401
1402 return next_item
1403
1404 def track_loaded_in_buffer(self, queue_id: str, item_id: str) -> None:
1405 """Call when a player has (started) loading a track in the buffer."""
1406 queue = self.get(queue_id)
1407 if not queue:
1408 msg = f"PlayerQueue {queue_id} is not available"
1409 raise PlayerUnavailableError(msg)
1410 # store the index of the item that is currently (being) loaded in the buffer
1411 # which helps us a bit to determine how far the player has buffered ahead
1412 current_index = self.index_by_id(queue_id, item_id)
1413 queue.index_in_buffer = current_index
1414 self.logger.debug("PlayerQueue %s loaded item %s in buffer", queue.display_name, item_id)
1415 self.signal_update(queue_id)
1416 # preload next streamdetails
1417 self._preload_next_item(queue_id, item_id)
1418 # clean up stale audio buffers for old queue items to prevent memory leaks
1419 if current_index is not None:
1420 self.mass.create_task(self._cleanup_stale_queue_buffers(queue_id, current_index))
1421
1422 def queue_buffer_completed(self, queue_id: str, queue_exhausted: bool) -> None:
1423 """
1424 Call when the flow stream has finished generating all audio data for a queue.
1425
1426 At this point all audio data for the queue has been passed to the encoding pipeline.
1427 The player will go idle once it finishes playing the remaining buffered audio.
1428
1429 We start a background task that waits for the player to go idle and checks if new
1430 items have been added to the queue in the meantime, resuming playback if so.
1431
1432 :param queue_id: The queue ID.
1433 :param queue_exhausted: Whether the flow ended because the queue ran out of items,
1434 as opposed to ending early to restart on a format change or a live item.
1435 """
1436 queue = self.get(queue_id)
1437 if not queue:
1438 return
1439 self.logger.debug("Queue flow buffer completed for %s", queue.display_name)
1440
1441 # capture session_id so we can bail out if playback restarts
1442 queue_data = self._queue_data[queue_id]
1443 original_session_id = queue_data.session_id
1444 # record so player providers can detect flow EOF without an idle report
1445 if original_session_id is not None:
1446 queue_data.flow_buffer_completed = original_session_id
1447 if queue_exhausted:
1448 queue_data.flow_queue_exhausted = original_session_id
1449
1450 async def _resume_on_idle() -> None:
1451 # wait for the player to finish playing the buffered audio and go idle
1452 idle_detected = False
1453 for _ in range(60):
1454 await asyncio.sleep(1)
1455 if not queue.active or queue_data.session_id != original_session_id:
1456 return
1457 if queue.state == PlaybackState.IDLE:
1458 idle_detected = True
1459 break
1460 if not idle_detected:
1461 return
1462 # player went idle, give it a brief moment to settle
1463 await asyncio.sleep(1)
1464 if queue.state != PlaybackState.IDLE or queue_data.session_id != original_session_id:
1465 return
1466 # check if new items were added to the queue after the flow stream ended
1467 if queue.current_index is not None and (
1468 next_item := self.get_next_item(queue_id, queue.current_index)
1469 ):
1470 next_index = self.index_by_id(queue_id, next_item.queue_item_id)
1471 if next_index is not None:
1472 self.logger.info(
1473 "Resuming playback after flow stream completed for %s",
1474 queue.display_name,
1475 )
1476 await self.play_index(queue_id, next_index)
1477
1478 task_id = f"queue_buffer_completed_{queue_id}"
1479 self.mass.create_task(_resume_on_idle(), task_id=task_id)
1480
1481 def flow_stream_finished(self, queue_id: str) -> bool:
1482 """
1483 Return whether the flow stream for the current playback session is fully generated.
1484
1485 Lets player providers detect flow EOF when the device does not report idle
1486 (e.g. a Cast group that underruns the LIVE flow stream and keeps reporting playing).
1487
1488 :param queue_id: The queue ID.
1489 """
1490 queue_data = self.queue_data_or_none(queue_id)
1491 if queue_data is None or queue_data.session_id is None:
1492 return False
1493 return queue_data.flow_buffer_completed == queue_data.session_id
1494
1495 def flow_queue_exhausted(self, queue_id: str, session_id: str) -> bool:
1496 """
1497 Return whether the given flow stream session played the queue to its end.
1498
1499 False while a session is still streaming, and for a flow stream that ended early
1500 to be restarted (a format change or a live item), where the player is expected to
1501 pick up the next stream right away.
1502
1503 :param queue_id: The queue ID.
1504 :param session_id: The stream session to check.
1505 """
1506 queue_data = self.queue_data_or_none(queue_id)
1507 if queue_data is None or queue_data.session_id != session_id:
1508 return False
1509 return queue_data.flow_queue_exhausted == session_id
1510
1511 # Main queue manipulation methods
1512
1513 async def load(
1514 self,
1515 queue_id: str,
1516 queue_items: list[QueueItem],
1517 insert_at_index: int = 0,
1518 keep_remaining: bool = True,
1519 keep_played: bool = True,
1520 shuffle: bool = False,
1521 pin_first: bool = False,
1522 ) -> None:
1523 """
1524 Load new items at index.
1525
1526 - queue_id: id of the queue to process this request.
1527 - queue_items: a list of QueueItems
1528 - insert_at_index: insert the item(s) at this index
1529 - keep_remaining: keep the remaining items after the insert
1530 - shuffle: (re)shuffle the items after insert index
1531 - pin_first: keep the first item at the insert index instead of letting the shuffle
1532 move it; only meaningful together with shuffle
1533 """
1534 prev_items = self._queue_data[queue_id].items[:insert_at_index] if keep_played else []
1535 next_items = queue_items
1536
1537 # if keep_remaining, append the old 'next' items
1538 if keep_remaining:
1539 next_items += self._queue_data[queue_id].items[insert_at_index:]
1540
1541 # we set the original insert order as attribute so we can un-shuffle
1542 for index, item in enumerate(next_items):
1543 item.sort_index += insert_at_index + index
1544 # (re)shuffle the final batch if needed: smart shuffle when enabled, else pure random
1545 if shuffle:
1546 queue = self._queue_data[queue_id].queue
1547 # a user-picked item must stay the one that plays, so hold it out of the shuffle
1548 pinned = next_items[:1] if pin_first else []
1549 shuffled = next_items[1:] if pin_first else next_items
1550 if self._smart_shuffle.is_enabled(queue_id):
1551 shuffled = await self._smart_shuffle.arrange(queue, shuffled)
1552 else:
1553 shuffled = random.sample(shuffled, len(shuffled))
1554 next_items = pinned + shuffled
1555 self.update_items(queue_id, prev_items + next_items)
1556
1557 def update_items(self, queue_id: str, queue_items: list[QueueItem]) -> None:
1558 """Update the existing queue items, mostly caused by reordering."""
1559 self._queue_data[queue_id].items = queue_items
1560 queue = self._queue_data[queue_id].queue
1561 queue.items = len(self._queue_data[queue_id].items)
1562 self.signal_update(queue_id, True)
1563 self.update_next_item_on_player(queue_id)
1564
1565 # Helper methods
1566
1567 def get_item(self, queue_id: str, item_id_or_index: int | str | None) -> QueueItem | None:
1568 """Get queue item by index or item_id."""
1569 if item_id_or_index is None:
1570 return None
1571 if (queue_data := self._queue_data.get(queue_id)) is None:
1572 return None
1573 queue_items = queue_data.items
1574 if isinstance(item_id_or_index, int) and len(queue_items) > item_id_or_index:
1575 return queue_items[item_id_or_index]
1576 if isinstance(item_id_or_index, str):
1577 return next((x for x in queue_items if x.queue_item_id == item_id_or_index), None)
1578 return None
1579
1580 def signal_update(self, queue_id: str, items_changed: bool = False) -> None:
1581 """Signal state changed of given queue."""
1582 if (queue_data := self._queue_data.get(queue_id)) is None:
1583 return
1584 queue = queue_data.queue
1585 # a mirrored shuffle write (streams controller) changes what smart shuffle
1586 # resolves to, so refresh the derived flag with every signaled update
1587 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1588 if items_changed:
1589 queue_data.items_cache_dirty = True
1590 self.mass.signal_event(EventType.QUEUE_ITEMS_UPDATED, object_id=queue_id, data=queue)
1591 self.mass.streams.audio_processing.prune(queue_id)
1592 # always send the base event
1593 self.mass.signal_event(EventType.QUEUE_UPDATED, object_id=queue_id, data=queue)
1594 # also signal update to the player itself so it can update its current_media
1595 self.mass.players.trigger_player_update(queue_id)
1596 # persist the (settings-bearing) queue state, debounced so a burst of updates or the
1597 # per-track updates during playback collapse into a single cache write
1598 self.mass.call_later(
1599 QUEUE_CACHE_SAVE_DELAY,
1600 self._save_queue_to_cache,
1601 queue_id,
1602 task_id=f"save_queue_cache_{queue_id}",
1603 )
1604
1605 def index_by_id(self, queue_id: str, queue_item_id: str) -> int | None:
1606 """Get index by queue_item_id."""
1607 if (queue_data := self._queue_data.get(queue_id)) is None:
1608 return None
1609 for index, item in enumerate(queue_data.items):
1610 if item.queue_item_id == queue_item_id:
1611 return index
1612 return None
1613
1614 async def get_tracks_for_playback(self, media_item: MediaItemType) -> list[Track]:
1615 """
1616 Return the playable tracks a media item resolves to, honoring the user's selection prefs.
1617
1618 :param media_item: The media item to resolve to playable tracks.
1619 """
1620 return await self._media_resolver.get_tracks_for_playback(media_item)
1621
1622 async def get_playlist_tracks(
1623 self, playlist: Playlist, start_item: str | None = None, sort_by: str | None = None
1624 ) -> list[PlaylistPlayableItem]:
1625 """
1626 Return the playable tracks for a playlist, honoring the user's selection prefs.
1627
1628 :param playlist: The playlist to resolve.
1629 :param start_item: Optional item URI to start the playlist from.
1630 :param sort_by: Optional sort key for the returned tracks.
1631 """
1632 return await self._media_resolver.get_playlist_tracks(playlist, start_item, sort_by)
1633
1634 async def get_dynamic_source_tracks(self, item: MediaItemType) -> list[Track]:
1635 """
1636 Return a fresh batch of tracks for a dynamic source (a dynamic playlist or radio station).
1637
1638 :param item: The dynamic playlist or radio station to fetch the next batch for.
1639 """
1640 return await self._media_resolver.get_dynamic_source_tracks(item)
1641
1642 def recency_windows(self) -> RecencyWindows:
1643 """Return the configured recency windows (a global setting; used for recency-aware gating)."""
1644 return self._smart_shuffle.windows()
1645
1646 async def player_media_from_queue_item(self, queue_item: QueueItem) -> PlayerMedia:
1647 """
1648 Parse PlayerMedia from QueueItem.
1649
1650 :param queue_item: The queue item to create media from.
1651 """
1652 queue_data = self._queue_data[queue_item.queue_id]
1653 stream_duration: int | None = None
1654 if queue_item.streamdetails:
1655 # prefer netto duration
1656 duration = queue_item.streamdetails.duration or queue_item.duration
1657 if duration and queue_item.streamdetails.seek_position:
1658 # the audio handed to the player starts at the seek position, so it is
1659 # shorter than the media item itself. seeking to (or past) the end
1660 # leaves no stream to describe, so the full length is kept instead.
1661 remaining = int(duration - queue_item.streamdetails.seek_position)
1662 stream_duration = remaining if remaining > 0 else None
1663 else:
1664 duration = queue_item.duration
1665 if queue_data.session_id is None:
1666 raise InvalidDataError("Queue session_id is None")
1667 media = PlayerMedia(
1668 uri=queue_item.uri,
1669 media_type=queue_item.media_type,
1670 title=queue_item.name,
1671 image_url=MASS_LOGO_ONLINE,
1672 duration=duration,
1673 stream_duration=stream_duration,
1674 source_id=queue_item.queue_id,
1675 queue_item_id=queue_item.queue_item_id,
1676 queue_session_id=queue_data.session_id,
1677 custom_data={
1678 "original_uri": queue_item.uri,
1679 },
1680 )
1681 if queue_item.media_item:
1682 media.title = queue_item.media_item.name
1683 media.artist = getattr(queue_item.media_item, "artist_str", "")
1684 media.album = (
1685 album.name if (album := getattr(queue_item.media_item, "album", None)) else ""
1686 )
1687 if queue_item.image:
1688 # the image format needs to be 512x512 jpeg for maximum compatibility with players
1689 # we prefer the imageproxy on the streamserver here because this request is sent
1690 # to the player itself which may not be able to reach the regular webserver
1691 media.image_url = self.mass.metadata.get_image_url(
1692 queue_item.image, size=512, image_format="jpeg", prefer_stream_server=True
1693 )
1694 return media
1695
1696 def get_next_item(self, queue_id: str, cur_index: int | str) -> QueueItem | None:
1697 """Return next QueueItem for given queue."""
1698 index: int
1699 if isinstance(cur_index, str):
1700 resolved_index = self.index_by_id(queue_id, cur_index)
1701 if resolved_index is None:
1702 return None # guard
1703 index = resolved_index
1704 else:
1705 index = cur_index
1706 # At this point index is guaranteed to be int
1707 for skip in range(5):
1708 if (next_index := self._get_next_index(queue_id, index + skip)) is None:
1709 break
1710 next_item = self.get_item(queue_id, next_index)
1711 if next_item is None:
1712 continue
1713 if not next_item.available:
1714 # ensure that we skip unavailable items (set by load_next track logic)
1715 continue
1716 return next_item
1717 return None
1718
1719 def store_sources(self, queue: PlayerQueue, items: list[MediaItemType]) -> None:
1720 """
1721 Hold the queue's full dynamic-source items server-side and project them onto `sources`.
1722
1723 :param queue: The queue whose sources are being set.
1724 :param items: The full source media items; an empty list clears the queue's sources.
1725 """
1726 self._queue_data[queue.queue_id].source_items = items
1727 # keep every occurrence server-side (a source added more than once weights it up in the
1728 # managed pool), but expose only the distinct container sources on the wire for clients to
1729 # show. Individual items (tracks, live radio streams, podcast episodes, ...) are omitted; see
1730 # `_WIRE_SOURCE_MEDIA_TYPES`. Autoplay/pool refill reads the full `source_items` above, not
1731 # this projected list, so it is unaffected.
1732 seen: set[str] = set()
1733 sources: list[ItemMapping] = []
1734 for item in items:
1735 if item.media_type not in _WIRE_SOURCE_MEDIA_TYPES and not is_dynamic_source(item):
1736 continue
1737 mapping = ItemMapping.from_item(item)
1738 if mapping.uri and mapping.uri in seen:
1739 continue
1740 if mapping.uri:
1741 seen.add(mapping.uri)
1742 sources.append(mapping)
1743 queue.sources = sources
1744 # release any materialized finite-source state whose source is no longer present
1745 self._managed_pool.retain(
1746 queue.queue_id, {item.uri for item in items if item.uri is not None}
1747 )
1748
1749 async def _save_queue_to_cache(self, queue_id: str) -> None:
1750 """Persist the queue's state (and its items when changed) to the cache."""
1751 if (queue_data := self._queue_data.get(queue_id)) is None:
1752 return
1753 try:
1754 # persistent so a cache clear/reset does not wipe the user's queues; the default
1755 # expiration still applies but is refreshed on every write. Skip the state write when its
1756 # persist-worthy content is unchanged (i.e. only playback progress advanced).
1757 state = queue_data.to_cache()
1758 significant = queue_data.cache_significant(state)
1759 if significant != queue_data.last_saved_state:
1760 await self.mass.cache.set(
1761 key=queue_id,
1762 data=state,
1763 provider=self.domain,
1764 category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1765 persistent=True,
1766 )
1767 queue_data.last_saved_state = significant
1768 if queue_data.items_cache_dirty:
1769 # only cache items with a valid media_item
1770 await self.mass.cache.set(
1771 key=queue_id,
1772 data=queue_data.items_to_cache(),
1773 provider=self.domain,
1774 category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1775 persistent=True,
1776 )
1777 queue_data.items_cache_dirty = False
1778 except Exception as err:
1779 self.logger.warning("Failed to persist the queue for %s - %s", queue_id, err)
1780
1781 def _check_player_permission(self, queue_id: str) -> None:
1782 """
1783 Check if the current user has permission to control this player/queue.
1784
1785 :param queue_id: The queue/player ID to check access for.
1786 :raises InsufficientPermissions: If the user lacks access.
1787 """
1788 current_user = get_current_user()
1789 if (
1790 current_user
1791 and current_user.player_filter
1792 and queue_id not in current_user.player_filter
1793 ):
1794 msg = f"{current_user.username} does not have access to player {queue_id}"
1795 raise InsufficientPermissions(msg)
1796
1797 @handle_play_action
1798 async def _handle_stop(self, queue_id: str) -> None:
1799 """
1800 Handle stop without checking the caller's player permissions.
1801
1802 :param queue_id: queue_id of the playerqueue to stop.
1803 """
1804 # cancel any pending play_index calls for this queue to prevent conflicts
1805 self.mass.cancel_timer(f"queue_play_index_{queue_id}")
1806 # cancel in-flight preload/enqueue-next so it can't enqueue after stop
1807 self.mass.cancel_task(f"preload_next_item_{queue_id}")
1808 self.mass.cancel_timer(f"enqueue_next_item_{queue_id}")
1809 self.mass.cancel_task(f"enqueue_next_item_{queue_id}")
1810 # a prewarm still running would attach its buffer after the teardown below has run,
1811 # leaving a stopped queue holding a provider's stream. Cancelled here rather than
1812 # alongside that teardown, where the task id can already belong to a new session.
1813 self.mass.cancel_task(f"prepare_next_audio_buffer_{queue_id}")
1814 self._set_transitioning(queue_id, False)
1815 queue_data = self._queue_data[queue_id]
1816 session_id = queue_data.session_id
1817 if (queue := self.get(queue_id)) and queue.active:
1818 if queue.state == PlaybackState.PLAYING:
1819 queue.resume_pos = int(queue.corrected_elapsed_time)
1820 try:
1821 # Use internal handler to avoid circular redirect:
1822 # public cmd_stop redirects to queue.stop when a queue is active,
1823 # which would loop back here indefinitely.
1824 await self.mass.players._handle_cmd_stop(queue_id)
1825 finally:
1826 # a device that could not be reached still gets its session torn down: an
1827 # open session keeps the item buffers producing, which holds a provider's
1828 # live session open long after the queue was told to stop
1829 if session_id is not None:
1830 # only the stopped session's audio is released. A stop that had no session
1831 # owns none of what is here, and taking it down would hit playback that
1832 # started while this stop was still waiting on the device
1833 if queue_data.session_id == session_id:
1834 queue_data.session_id = None
1835 self.mass.streams.audio_processing.clear(queue_id, session_id)
1836 self.mass.create_task(self._cleanup_queue_audio_data(queue_id, session_id))
1837
1838 @handle_play_action
1839 async def _handle_play(self, queue_id: str) -> None:
1840 """Handle play without acquiring the queue lock."""
1841 queue_player = self.mass.players.get_player(queue_id, True)
1842 if queue_player is None:
1843 raise PlayerUnavailableError(f"Player {queue_id} is not available")
1844 if (queue := self.get(queue_id)) and queue.active and queue.state == PlaybackState.PAUSED:
1845 # forward the actual play/unpause command to the player,
1846 # holding the action until the player confirms it resumed playback
1847 async with self.mass.players.wait_for_player_update(
1848 queue_id,
1849 attribute_name="playback_state",
1850 attribute_value=PlaybackState.PLAYING,
1851 timeout=PLAYBACK_START_TIMEOUT,
1852 ):
1853 await queue_player.play()
1854 return
1855 # player is not paused, perform resume instead
1856 await self.resume(queue_id)
1857
1858 def _set_transitioning(self, queue_id: str, value: bool) -> None:
1859 """Mark (or clear) whether a queue is mid-transition (no-op if it is not registered)."""
1860 if (queue_data := self._queue_data.get(queue_id)) is not None:
1861 queue_data.transitioning = value
1862
1863 def _clear(self, queue_id: str, skip_stop: bool = False) -> None:
1864 """Drop the queue's items and playback position, leaving user settings untouched."""
1865 queue = self._queue_data[queue_id].queue
1866 self.mass.streams.audio_processing.clear(queue_id)
1867 self.store_sources(queue, [])
1868 if queue.is_dynamic:
1869 # Dynamic sources impose shuffle, so clearing the source clears that shuffle too.
1870 queue.shuffle_enabled = False
1871 queue.is_dynamic = False
1872 # dropping the dynamic source changes what smart shuffle resolves to, so the derived
1873 # flag has to follow or clients keep showing a smart mix on a plain queue
1874 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1875 queue.ended = False
1876 if queue.state != PlaybackState.IDLE and not skip_stop:
1877 self.mass.create_task(self.stop(queue_id))
1878 queue.current_index = None
1879 queue.current_item = None
1880 queue.elapsed_time = 0
1881 queue.elapsed_time_last_updated = time.time()
1882 queue.index_in_buffer = None
1883 self.mass.create_task(self._cleanup_queue_audio_data(queue_id))
1884 self.update_items(queue_id, [])
1885
1886 def _reset_shuffle(self, queue_id: str) -> None:
1887 """Switch shuffle off."""
1888 queue = self._queue_data[queue_id].queue
1889 if not queue.shuffle_enabled:
1890 return
1891 queue.shuffle_enabled = False
1892 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1893 self.signal_update(queue_id)
1894
1895 async def _apply_shuffle(
1896 self, queue_id: str, option: QueueOption, shuffle: bool | None
1897 ) -> None:
1898 """
1899 Settle the queue's shuffle state for a play command before its items are resolved.
1900
1901 :param queue_id: The queue the media is played on.
1902 :param option: The enqueue option this command resolved to.
1903 :param shuffle: The state to put the queue's shuffle in; None to leave it as it is.
1904 """
1905 queue = self._queue_data[queue_id].queue
1906 if queue.is_dynamic and option in (
1907 QueueOption.PLAY,
1908 QueueOption.REPLACE,
1909 QueueOption.REPLACE_NEXT,
1910 ):
1911 # These are the options that replace the queue's sources, so the smart mix may be on
1912 # its way out - and its shuffle is never the user's own (a dynamic queue's toggle is
1913 # locked), so it must not outlive the source that imposed it. Recorded directly
1914 # because set_shuffle refuses a queue that is still a smart mix, and the items are
1915 # resolved against this flag. The state is provisional until the sources are known:
1916 # `_enter_dynamic_mode` forces shuffle back on if the queue stays dynamic.
1917 if option == QueueOption.REPLACE_NEXT:
1918 # staging leaves the shuffle the user chose alone, so it never carries a request
1919 # of its own to honour here
1920 queue.shuffle_enabled = False
1921 else:
1922 queue.shuffle_enabled = bool(shuffle)
1923 return
1924 if shuffle is None or option not in (QueueOption.PLAY, QueueOption.REPLACE):
1925 # nothing to settle: the media brings no order of its own to protect, or the option
1926 # only stages items for later and leaves the queue's shuffle state alone
1927 return
1928 if queue.shuffle_enabled == shuffle:
1929 return
1930 # routed through set_shuffle so switching shuffle off also restores the order of
1931 # the items that stay in the queue: a play keeps them, and a tail left in shuffled
1932 # order behind a queue that now reads unshuffled would contradict its own flag
1933 await self.set_shuffle(queue_id, shuffle)
1934
1935 async def _apply_local_shuffle(self, queue_id: str, shuffle_enabled: bool) -> None:
1936 """
1937 Record the queue's shuffle state and re-order the un-played tail accordingly.
1938
1939 :param queue_id: The queue to apply the shuffle state to.
1940 :param shuffle_enabled: The shuffle state to record and apply to the tail.
1941 """
1942 queue = self._queue_data[queue_id].queue
1943 queue.shuffle_enabled = shuffle_enabled
1944 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
1945 queue_items = self._queue_data[queue_id].items
1946 cur_index = committed_index(queue)
1947 if cur_index is not None:
1948 next_index = cur_index + 1
1949 next_items = queue_items[next_index:]
1950 else:
1951 next_items = []
1952 next_index = 0
1953 if not shuffle_enabled:
1954 # shuffle disabled, try to restore original sort order of the remaining items
1955 next_items.sort(key=lambda x: x.sort_index, reverse=False)
1956 await self.load(
1957 queue_id=queue_id,
1958 queue_items=next_items,
1959 insert_at_index=next_index,
1960 keep_remaining=False,
1961 shuffle=shuffle_enabled,
1962 )
1963
1964 def _resolve_default_toggles(self, queue_data: PlayerQueueData) -> None:
1965 """Set the queue's effective autoplay/crossfade from their override or the global default."""
1966 queue = queue_data.queue
1967 queue.autoplay_enabled = (
1968 queue_data.autoplay_override
1969 if queue_data.autoplay_override is not None
1970 else self.mass.config.get_raw_core_config_value(
1971 self.domain, CONF_AUTOPLAY_ENABLED, DEFAULT_AUTOPLAY_ENABLED
1972 )
1973 )
1974 queue.crossfade_enabled = (
1975 queue_data.crossfade_override
1976 if queue_data.crossfade_override is not None
1977 else self.mass.config.get_raw_core_config_value(
1978 self.domain, CONF_CROSSFADE_ENABLED, DEFAULT_CROSSFADE_ENABLED
1979 )
1980 )
1981