music-assistant-server

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