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