music-assistant-server

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