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