music-assistant-server

1"""
2Queue loading for the Player Queues controller.
3
4Applies the enqueue option (play/replace/next/add) to a batch of resolved items, loads a single
5media item into the queue, resumes from the play-log when the queue is empty, computes the next
6index, and refills the queue (dynamic managed-pool fill and autoplay fill). Owns no per-queue state;
7it is mixed into the controller and reads/mutates the controller's `PlayerQueueData` records.
8"""
9# ruff: noqa: PLR0915
10
11from __future__ import annotations
12
13import random
14from contextlib import suppress
15from typing import TYPE_CHECKING, cast
16
17from music_assistant_models.enums import (
18    MediaType,
19    PlaybackState,
20    QueueOption,
21    RepeatMode,
22)
23from music_assistant_models.errors import (
24    InvalidDataError,
25    MediaNotFoundError,
26    MusicAssistantError,
27    PlayerUnavailableError,
28)
29from music_assistant_models.media_items import (
30    Album,
31    Audiobook,
32    BrowseFolder,
33    ItemMapping,
34    MediaItemType,
35    PlayableMediaItemType,
36    PodcastEpisode,
37    Track,
38    UniqueList,
39    media_from_dict,
40)
41
42from music_assistant.constants import ATTR_ANNOUNCEMENT_IN_PROGRESS
43from music_assistant.controllers.player_queues.autoplay import (
44    AUTOPLAY_EXCLUDED_MEDIA_TYPES,
45    AUTOPLAY_SERIES_MEDIA_TYPES,
46    AutoplayMode,
47)
48from music_assistant.controllers.player_queues.base import _PlayerQueuesBase
49from music_assistant.controllers.player_queues.constants import (
50    CONF_DEFAULT_ENQUEUE_OPTION_LIVE_SOURCES,
51    MANAGED_POOL_MAX,
52    PROBED_DURATION_MEDIA_TYPES,
53)
54from music_assistant.controllers.player_queues.helpers import (
55    build_queue_item,
56    handle_play_action,
57    has_dynamic_source,
58    is_dynamic_source,
59)
60from music_assistant.controllers.player_queues.managed_pool import gate_tracks
61from music_assistant.controllers.streams.audio_buffer import AudioBuffer
62from music_assistant.controllers.webserver.helpers.auth_middleware import (
63    get_current_user,
64    set_current_user,
65)
66from music_assistant.helpers.audio import get_probed_duration, store_probed_duration
67from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
68
69if TYPE_CHECKING:
70    from music_assistant_models.media_items.metadata import MediaItemImage
71    from music_assistant_models.queue_item import QueueItem
72
73    from music_assistant.providers.radio_playlist import RadioPlaylistProvider
74
75
76class QueueLoaderMixin(_PlayerQueuesBase):
77    """Load items into a queue: apply the enqueue option, resolve single items, refill the pool."""
78
79    async def _enqueue_with_option(
80        self,
81        queue_id: str,
82        queue_items: list[QueueItem],
83        option: QueueOption | None,
84        pin_first: bool = False,
85    ) -> None:
86        """
87        Load queue items into the queue according to the given enqueue option.
88
89        :param queue_id: The queue to load the items into.
90        :param queue_items: The items to load.
91        :param option: The enqueue option to apply.
92        :param pin_first: The first item was explicitly picked by the user (a start_item), so it
93            must keep its position when the batch is shuffled instead of being moved at random.
94        """
95        queue = self._queue_data[queue_id].queue
96        # A queue that played to its end is finished, so anything enqueued onto it starts a fresh
97        # queue rather than stacking onto the items that already played. Only an explicit ADD keeps
98        # them: there the added items continue the queue from where it ended, and the index is moved
99        # onto the first of them below so pressing play starts there instead of replaying the last
100        # item. ADD never starts playback by itself.
101        continues_ended_queue = queue.ended and option == QueueOption.ADD
102        items_before_add = len(self._queue_data[queue_id].items)
103        if queue.ended and not continues_ended_queue:
104            # mechanical clear: the shuffle state for this batch was already settled by the caller
105            self._clear(queue_id, skip_stop=True)
106        if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
107            cur_index = (
108                queue.index_in_buffer
109                if queue.index_in_buffer is not None
110                else (queue.current_index if queue.current_index is not None else 0)
111            )
112        else:
113            cur_index = queue.current_index or 0
114        insert_at_index = cur_index + 1
115        shuffle = queue.shuffle_enabled and len(queue_items) > 1
116        # a user-picked start item must be the one that actually starts playing, so keep it in
117        # front of the shuffled rest instead of letting the shuffle move it to a random slot
118        pin_first = pin_first and shuffle
119
120        # handle replace: clear all items and replace with the new items
121        if option == QueueOption.REPLACE:
122            if pin_first:
123                await self._load_pinned_first(
124                    queue_id,
125                    queue_items,
126                    insert_at_index=0,
127                    keep_remaining=False,
128                    keep_played=False,
129                )
130            else:
131                await self.load(
132                    queue_id,
133                    queue_items=queue_items,
134                    keep_remaining=False,
135                    keep_played=False,
136                    shuffle=shuffle,
137                )
138            await self.play_index(queue_id, 0)
139            return
140        # handle next: add item(s) in the index next to the playing/loaded/buffered index
141        if option == QueueOption.NEXT:
142            if shuffle:
143                # honour "play next" under shuffle: the first new item goes right after the
144                # buffered index so it plays next, the rest of the batch is shuffled into the tail
145                # behind it. insert_at_index is the first un-buffered slot, so the track the player
146                # already prepared for crossfade is left untouched.
147                await self._load_pinned_first(queue_id, queue_items, insert_at_index)
148            else:
149                await self.load(
150                    queue_id,
151                    queue_items=queue_items,
152                    insert_at_index=insert_at_index,
153                    shuffle=shuffle,
154                )
155            self._ensure_current_index(queue_id)
156            return
157        if option == QueueOption.REPLACE_NEXT:
158            if pin_first:
159                await self._load_pinned_first(
160                    queue_id, queue_items, insert_at_index, keep_remaining=False
161                )
162            else:
163                await self.load(
164                    queue_id,
165                    queue_items=queue_items,
166                    insert_at_index=insert_at_index,
167                    keep_remaining=False,
168                    shuffle=shuffle,
169                )
170            self._ensure_current_index(queue_id)
171            return
172        # handle play: replace current loaded/playing index with new item(s)
173        if option == QueueOption.PLAY:
174            # an idle/empty queue has no current item to insert after, so insert at and
175            # start from the very first index instead of skipping past it
176            play_at_index = 0 if queue.current_index is None else insert_at_index
177            if pin_first:
178                await self._load_pinned_first(queue_id, queue_items, play_at_index)
179            else:
180                await self.load(
181                    queue_id,
182                    queue_items=queue_items,
183                    insert_at_index=play_at_index,
184                    shuffle=shuffle,
185                )
186            next_index = min(play_at_index, len(self._queue_data[queue_id].items) - 1)
187            await self.play_index(queue_id, next_index)
188            return
189        # handle add: add/append item(s) to the remaining queue items
190        if option == QueueOption.ADD:
191            # When shuffling, mix the new items into the not-yet-played tail. While playing,
192            # keep the item right after the buffered one in place: it has already been enqueued
193            # to the player (and prepared for crossfade), so reshuffling it would swap the
194            # upcoming track underneath the player and cause an abrupt, non-crossfaded switch.
195            if not queue.shuffle_enabled:
196                add_at_index = len(self._queue_data[queue_id].items) + 1
197            elif queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
198                add_at_index = insert_at_index + 1
199            else:
200                add_at_index = insert_at_index
201            await self.load(
202                queue_id=queue_id,
203                queue_items=queue_items,
204                insert_at_index=add_at_index,
205                shuffle=queue.shuffle_enabled,
206            )
207            if continues_ended_queue:
208                self._continue_ended_queue(queue_id, items_before_add)
209                return
210            self._ensure_current_index(queue_id)
211
212    async def _load_pinned_first(
213        self,
214        queue_id: str,
215        queue_items: list[QueueItem],
216        insert_at_index: int,
217        keep_remaining: bool = True,
218        keep_played: bool = True,
219    ) -> None:
220        """
221        Insert the first item at the given index and shuffle the rest of the batch behind it.
222
223        :param queue_id: The queue to load the items into.
224        :param queue_items: The items to load; the first one keeps the given index.
225        :param insert_at_index: The index to place the first item at.
226        :param keep_remaining: Keep the queue's existing items from the insert index onwards.
227        :param keep_played: Keep the queue's existing items before the insert index.
228        """
229        await self.load(
230            queue_id,
231            queue_items=queue_items[:1],
232            insert_at_index=insert_at_index,
233            keep_remaining=keep_remaining,
234            keep_played=keep_played,
235        )
236        await self.load(
237            queue_id,
238            queue_items=queue_items[1:],
239            insert_at_index=insert_at_index + 1,
240            shuffle=True,
241        )
242
243    def _ensure_current_index(self, queue_id: str) -> None:
244        """
245        Point the current index at the first item when the queue does not have one yet.
246
247        NEXT/ADD/REPLACE_NEXT stage items without starting playback; on an empty queue there is no
248        current index, so set it to the first item to give the queue a current item. A queue that
249        already has content keeps its current index untouched (its items are inserted after it).
250
251        :param queue_id: The queue to update.
252        """
253        queue = self._queue_data[queue_id].queue
254        if queue.current_index is not None:
255            return
256        queue.current_index = 0
257        queue.current_item = self.get_item(queue_id, 0)
258        self.signal_update(queue_id)
259
260    def _continue_ended_queue(self, queue_id: str, first_added_index: int) -> None:
261        """
262        Point a finished queue at the first item just added to it, without starting playback.
263
264        The items that already played are kept, so the queue is no longer finished but its position
265        still sits on its old last item. Moving it onto the added items is what makes a play press
266        start there rather than replay the item the queue ended on.
267
268        :param queue_id: The queue that was added to.
269        :param first_added_index: Index of the first of the added items.
270        """
271        queue = self._queue_data[queue_id].queue
272        queue.ended = False
273        if (current_item := self.get_item(queue_id, first_added_index)) is None:
274            return
275        queue.current_index = first_added_index
276        queue.current_item = current_item
277        # ending the queue cleared the next item; refresh it so a batch of added items reports
278        # what follows instead of looking like there is nothing after the first one
279        queue.next_item = self.get_next_item(queue_id, first_added_index)
280        self.signal_update(queue_id)
281
282    async def _load_item(
283        self,
284        queue_item: QueueItem,
285        next_index: int | None,
286        is_start: bool = False,
287        seek_position: int = 0,
288        fade_in: bool = False,
289    ) -> None:
290        """Try to load the stream details for the given queue item."""
291        queue_id = queue_item.queue_id
292        queue = self._queue_data[queue_id].queue
293
294        # we use a contextvar to bypass the throttler for this asyncio task/context
295        # this makes sure that playback has priority over other requests that may be
296        # happening in the background
297        BYPASS_THROTTLER.set(True)
298
299        self.logger.debug(
300            "(pre)loading (next) item for queue %s...",
301            queue.display_name,
302        )
303
304        if not queue_item.available:
305            raise MediaNotFoundError(f"Item {queue_item.uri} is not available")
306
307        # work out if we are playing an album and if we should prefer album
308        # loudness
309        next_track_from_same_album = (
310            next_index is not None
311            and (next_item := self.get_item(queue_id, next_index))
312            and (
313                queue_item.media_item
314                and hasattr(queue_item.media_item, "album")
315                and queue_item.media_item.album
316                and next_item.media_item
317                and hasattr(next_item.media_item, "album")
318                and next_item.media_item.album
319                and queue_item.media_item.album.item_id == next_item.media_item.album.item_id
320            )
321        )
322        current_index = self.index_by_id(queue_id, queue_item.queue_item_id)
323        if current_index is None:
324            previous_track_from_same_album = False
325        else:
326            previous_index = max(current_index - 1, 0)
327            previous_track_from_same_album = (
328                previous_index > 0
329                and (previous_item := self.get_item(queue_id, previous_index)) is not None
330                and previous_item.media_item is not None
331                and hasattr(previous_item.media_item, "album")
332                and previous_item.media_item.album is not None
333                and queue_item.media_item is not None
334                and hasattr(queue_item.media_item, "album")
335                and queue_item.media_item.album is not None
336                and queue_item.media_item.album.item_id == previous_item.media_item.album.item_id
337            )
338        playing_album_tracks = next_track_from_same_album or previous_track_from_same_album
339        if queue_item.media_item and isinstance(queue_item.media_item, Track):
340            album = queue_item.media_item.album
341            # prefer the full library media item so we have all metadata and provider(quality) info
342            # always request the full library item as there might be other qualities available
343            if library_item := await self.mass.music.get_library_item_by_prov_id(
344                queue_item.media_item.media_type,
345                queue_item.media_item.item_id,
346                queue_item.media_item.provider,
347            ):
348                queue_item.media_item = cast("Track", library_item)
349            elif not queue_item.media_item.image or queue_item.media_item.provider.startswith(
350                "ytmusic"
351            ):
352                # Youtube Music has poor thumbs by default, so we always fetch the full item
353                # this also catches the case where they have an unavailable item in a listing
354                fetched_item = await self.mass.music.get_item_by_uri(queue_item.uri)
355                queue_item.media_item = cast("Track", fetched_item)
356
357            # ensure we got the full (original) album set
358            if album and (
359                library_album := await self.mass.music.get_library_item_by_prov_id(
360                    album.media_type,
361                    album.item_id,
362                    album.provider,
363                )
364            ):
365                queue_item.media_item.album = cast("Album", library_album)
366            elif album:
367                # Restore original album if we have no better alternative from the library
368                queue_item.media_item.album = album
369            # prefer album image over track image
370            if queue_item.media_item.album and queue_item.media_item.album.image:
371                org_images: list[MediaItemImage] = queue_item.media_item.metadata.images or []
372                queue_item.media_item.metadata.images = UniqueList(
373                    [
374                        queue_item.media_item.album.image,
375                        *org_images,
376                    ]
377                )
378        # Fetch streamdetails (reuses existing if buffer is still valid for the seek).
379        queue_item.streamdetails = await self.mass.streams.audio.get_stream_details(
380            queue_item=queue_item,
381            seek_position=seek_position,
382            fade_in=fade_in,
383            prefer_album_loudness=bool(playing_album_tracks),
384        )
385        # update queue_item.duration from streamdetails if we got a better value
386        self._apply_probed_duration(queue_item)
387
388        # pre-initialize the AudioBuffer so audio is ready
389        # when the player requests it. For the current/first track this ensures
390        # immediate playback start. For preloaded next tracks we skip this and
391        # initialize the buffer ~30s before the current track ends instead.
392        # AudioSource items are realtime/live and bypass the AudioBuffer.
393        if is_start and queue_item.streamdetails.media_type != MediaType.AUDIO_SOURCE:
394            await AudioBuffer.get_buffer(
395                self.mass,
396                queue_item.streamdetails,
397                seek_position_ms=int(seek_position * 1000),
398                wait_ready=True,
399                reason="prepare",
400            )
401            # the first chunk is in, so the source has been probed and a duration the
402            # provider did not report is known before playback starts
403            self._apply_probed_duration(queue_item)
404
405    def _apply_probed_duration(self, queue_item: QueueItem) -> None:
406        """
407        Apply a duration determined while streaming to the queue item and its media item.
408
409        :param queue_item: The queue item whose streamdetails to take the duration from.
410        """
411        streamdetails = queue_item.streamdetails
412        if streamdetails is None or not streamdetails.duration:
413            return
414        duration = int(streamdetails.duration)
415        if not self._set_missing_duration(queue_item, duration):
416            return
417        if uri := getattr(queue_item.media_item, "uri", None):
418            # store it so listings and later playbacks have it up front
419            self.mass.create_task(store_probed_duration(self.mass, uri, duration))
420
421    async def _restore_probed_duration(self, queue_item: QueueItem) -> None:
422        """
423        Apply the duration determined during an earlier playback to an item that lacks one.
424
425        :param queue_item: The queue item to fill the duration of.
426        """
427        if queue_item.media_type not in PROBED_DURATION_MEDIA_TYPES:
428            return
429        if not (uri := getattr(queue_item.media_item, "uri", None)):
430            return
431        if queue_item.duration and getattr(queue_item.media_item, "duration", None):
432            return
433        if duration := await get_probed_duration(self.mass, uri):
434            self._set_missing_duration(queue_item, duration)
435
436    def _set_missing_duration(self, queue_item: QueueItem, duration: int) -> bool:
437        """
438        Fill in the duration of a queue item and its media item, leaving known ones alone.
439
440        :param queue_item: The queue item to fill the duration of.
441        :param duration: The duration in seconds.
442        :return: True if the item (or its media item) did not have a duration yet.
443        """
444        if queue_item.media_type not in PROBED_DURATION_MEDIA_TYPES:
445            return False
446        media_item = queue_item.media_item
447        # an ItemMapping or any other reference without a duration is left untouched
448        media_item_duration = getattr(media_item, "duration", None)
449        if queue_item.duration and media_item_duration != 0:
450            return False
451        if not queue_item.duration:
452            queue_item.duration = duration
453        if media_item_duration == 0:
454            media_item.duration = duration  # type: ignore[union-attr]
455        self.signal_update(queue_item.queue_id, items_changed=True)
456        return True
457
458    def _get_next_index(
459        self,
460        queue_id: str,
461        cur_index: int | None,
462        is_skip: bool = False,
463        allow_repeat: bool = True,
464    ) -> int | None:
465        """
466        Return the next index for the queue, accounting for repeat settings.
467
468        Will return None if there are no (more) items in the queue.
469        """
470        queue = self._queue_data[queue_id].queue
471        queue_items = self._queue_data[queue_id].items
472        if not queue_items or cur_index is None:
473            # queue is empty
474            return None
475        # handle repeat single track
476        if queue.repeat_mode == RepeatMode.ONE and not is_skip:
477            return cur_index if allow_repeat else None
478        # handle cur_index is last index of the queue
479        if cur_index >= (len(queue_items) - 1):
480            if allow_repeat and queue.repeat_mode == RepeatMode.ALL:
481                # if repeat all is enabled, we simply start again from the beginning
482                return 0
483            return None
484        # all other: just the next index
485        return cur_index + 1
486
487    async def _fill_dynamic_tracks(self, queue_id: str) -> None:
488        """Fill a Queue with (additional) tracks from its dynamic sources."""
489        self.logger.debug(
490            "Filling dynamic tracks for queue %s",
491            queue_id,
492        )
493        queue_data = self._queue_data[queue_id]
494        queue = queue_data.queue
495        # restore the queue owner's user context so provider filters are respected during this
496        # background refill (dynamic-playlist generation honours the current user)
497        playback_user = (
498            await self.mass.webserver.auth.get_user(queue_data.userid)
499            if queue_data.userid
500            else None
501        )
502        set_current_user(playback_user)
503        # Top up from the queue's dynamic sources (dynamic playlists and any mixed-in finite items),
504        # weighted per source and recency-gated. fill() already sizes the batch to the pool target;
505        # the tail cap below is a defensive ceiling so the unplayed tail never grows past
506        # MANAGED_POOL_MAX.
507        pool_tracks = await self._managed_pool.fill(queue_id, is_initial=False)
508        # keep the unplayed tail within the bounded pool size (no current_index => nothing played yet)
509        played = 0 if queue.current_index is None else queue.current_index + 1
510        unplayed = max(len(self._queue_data[queue_id].items) - played, 0)
511        headroom = max(MANAGED_POOL_MAX - unplayed, 0)
512        queue_items = [build_queue_item(queue_id, x) for x in pool_tracks[:headroom] if x.available]
513        if not queue_items:
514            return
515        await self.load(
516            queue_id,
517            queue_items,
518            insert_at_index=len(self._queue_data[queue_id].items) + 1,
519        )
520
521    async def _fill_autoplay_tracks(self, queue_id: str) -> None:
522        """
523        Append more items to a queue that is running low, based on what is ending.
524
525        Autoplay is a single "keep going" switch; what it appends is decided by the media type
526        of the queue's last item, since that is the item the appended items follow.
527        """
528        queue = self.get(queue_id)
529        if queue is None or not queue.autoplay_enabled:
530            return
531        queue_data = self._queue_data[queue_id]
532        if not queue_data.items:
533            return
534        last_item = queue_data.items[-1]
535        if last_item.media_type in AUTOPLAY_EXCLUDED_MEDIA_TYPES:
536            return
537        # Restore the queue owner's user context so provider filters, library access and
538        # resume positions are respected during this background refill, mirroring
539        # _fill_dynamic_tracks.
540        playback_user = (
541            await self.mass.webserver.auth.get_user(queue_data.userid)
542            if queue_data.userid
543            else None
544        )
545        set_current_user(playback_user)
546        if last_item.media_type in AUTOPLAY_SERIES_MEDIA_TYPES:
547            await self._fill_autoplay_next_in_series(queue_id, last_item)
548            return
549        await self._fill_autoplay_music_tracks(queue_id)
550
551    async def _fill_autoplay_next_in_series(self, queue_id: str, last_item: QueueItem) -> None:
552        """
553        Append the episode/book that follows the queue's last item, if there is one.
554
555        Nothing is appended for the last episode of a podcast or a book without a next one in
556        its collection, so the queue simply ends there.
557
558        :param queue_id: The queue to append to.
559        :param last_item: The queue's last item, an audiobook or podcast episode.
560        """
561        queue_data = self._queue_data[queue_id]
562        media_item = last_item.media_item
563        next_item: PodcastEpisode | Audiobook | None
564        try:
565            if isinstance(media_item, PodcastEpisode):
566                next_item = await self._media_resolver.get_next_podcast_episode(
567                    media_item, userid=queue_data.userid
568                )
569            elif isinstance(media_item, Audiobook):
570                next_item = await self._media_resolver.get_next_audiobook(
571                    media_item, userid=queue_data.userid
572                )
573            else:
574                return
575        except MusicAssistantError as err:
576            self.logger.warning(
577                "Autoplay failed to fetch the item following %s: %s", last_item.name, err
578            )
579            return
580        if next_item is None or not next_item.available:
581            self.logger.debug("Autoplay found nothing to play after %s", last_item.name)
582            return
583        if any(
584            item.media_item and item.media_item.uri == next_item.uri for item in queue_data.items
585        ):
586            # already queued (e.g. the user added it themselves), so there is nothing to do
587            return
588        await self.load(
589            queue_id,
590            [build_queue_item(queue_id, next_item)],
591            insert_at_index=len(queue_data.items) + 1,
592        )
593
594    async def _fill_autoplay_music_tracks(self, queue_id: str) -> None:
595        """Fill a Queue with additional tracks based on the configured Autoplay mode."""
596        queue = self.get(queue_id)
597        if queue is None:
598            return
599        queue_data = self._queue_data[queue_id]
600        if not queue_data.enqueued_media_items:
601            # the music refill needs what the user enqueued as its seed
602            return
603        mode = self._autoplay.resolve_mode(queue_id)
604        self.logger.debug(
605            "Filling autoplay tracks (mode: %s) for queue %s", mode.value, queue.display_name
606        )
607        existing_tracks = {
608            item.media_item
609            for item in self._queue_data[queue_id].items
610            if isinstance(item.media_item, Track)
611        }
612        try:
613            if mode == AutoplayMode.PLAYLIST:
614                tracks = await self._autoplay.get_playlist_tracks(queue, existing_tracks)
615            elif mode == AutoplayMode.LIBRARY:
616                tracks = await self._autoplay.get_library_tracks(queue, existing_tracks)
617            elif mode == AutoplayMode.SIMILAR:
618                tracks = await self._get_similar_tracks(
619                    queue_id, seed_items=queue_data.enqueued_media_items
620                )
621            else:
622                # AUTO: try similar tracks first, fall back to the library mix. The similar
623                # fetch raises when no provider can supply base/similar tracks, so suppress
624                # that here to make sure the library fallback still runs.
625                tracks = []
626                with suppress(MusicAssistantError):
627                    tracks = await self._get_similar_tracks(
628                        queue_id, seed_items=queue_data.enqueued_media_items
629                    )
630                if not tracks:
631                    tracks = await self._autoplay.get_library_tracks(queue, existing_tracks)
632        except MusicAssistantError as err:
633            self.logger.warning(
634                "Autoplay failed to fetch tracks for queue %s: %s", queue.display_name, err
635            )
636            return
637        # route the autoplay batch through the recency engine so a recently-heard track isn't
638        # immediately re-added (ungated fallback keeps autoplay going if everything is recent)
639        windows = self._smart_shuffle.windows()
640        snapshot = await self.mass.music.recency.snapshot(windows, userid=queue_data.userid)
641        tracks = gate_tracks(
642            [track for track in tracks if isinstance(track, Track)], snapshot, windows
643        )
644        queue_items = [build_queue_item(queue_id, x) for x in tracks if x.available]
645        if not queue_items:
646            self.logger.info("Autoplay found no new tracks to add for queue %s", queue.display_name)
647            return
648        await self.load(
649            queue_id,
650            queue_items,
651            insert_at_index=len(self._queue_data[queue_id].items) + 1,
652        )
653
654    @handle_play_action
655    async def _handle_play_media(
656        self,
657        queue_id: str,
658        media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
659        option: QueueOption | None = None,
660        radio_mode: bool = False,
661        start_item: PlayableMediaItemType | str | None = None,
662        sort_by: str | None = None,
663        start_from_beginning: bool = False,
664        shuffle: bool | None = None,
665    ) -> None:
666        """Handle play media without acquiring the queue lock."""
667        # cancel any pending play_index calls for this queue to prevent conflicts
668        self.mass.cancel_timer(f"queue_play_index_{queue_id}")
669        self._set_transitioning(queue_id, False)
670        # we use a contextvar to bypass the throttler for this asyncio task/context
671        # this makes sure that playback has priority over other requests that may be
672        # happening in the background
673        BYPASS_THROTTLER.set(True)
674        if not (queue := self.get(queue_id)):
675            raise PlayerUnavailableError(f"Queue {queue_id} is not available")
676        queue_data = self._queue_data[queue_id]
677        # always fetch the underlying player so we can raise early if its not available
678        queue_player = self.mass.players.get_player(queue_id, True)
679        assert queue_player is not None  # for type checking
680        if queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
681            self.logger.warning("Ignore queue command: An announcement is in progress")
682            return
683
684        # save the user requesting the playback (clear it for anonymous playback)
685        playback_user = get_current_user()
686        queue_data.userid = playback_user.user_id if playback_user else None
687        if playback_user:
688            self.logger.debug(
689                "User %s requested playback.", playback_user.display_name or playback_user.username
690            )
691
692        # a single item or list of items may be provided
693        media_list = media if isinstance(media, list) else [media]
694
695        if radio_mode:
696            # radio_mode is deprecated: a "radio" is now a dynamic radio playlist. Translate each
697            # seed into the radio_playlist provider's URI and enqueue those (resolved to dynamic
698            # playlists that self-manage their refills).
699            self.logger.warning(
700                "radio_mode is deprecated; enqueue a radio_playlist:// dynamic playlist instead"
701            )
702            media_list = [
703                seed_uri
704                if (seed_uri := item if isinstance(item, str) else str(item.uri)).startswith(
705                    "radio_playlist://"
706                )
707                else f"radio_playlist://playlist/{seed_uri}"
708                for item in media_list
709            ]
710            radio_mode = False
711
712        # clear queue if needed
713        if option == QueueOption.REPLACE:
714            self._clear(queue_id, skip_stop=True)
715        # Clear the 'enqueued media item' list when a new queue is requested
716        if option not in (QueueOption.ADD, QueueOption.NEXT):
717            queue_data.enqueued_media_items.clear()
718        # The shuffle state has to be settled before the items are resolved below: a shuffled queue
719        # keeps the items preceding a start_item (chosen track pinned first) instead of dropping
720        # them. When the option still has to be derived, this runs as soon as it is known.
721        if option is not None:
722            await self._apply_shuffle_intent(queue_id, option, shuffle)
723
724        # An ADD/NEXT onto a queue that is already a managed pool (has a dynamic source): a finite
725        # item is kept only as a source (the bounded pool materializes it) instead of being expanded
726        # into the queue. Any other enqueue (PLAY/REPLACE, or onto a linear queue) expands finite
727        # items normally. Keys off is_dynamic since a finite-only queue records sources too.
728        already_dynamic = queue.is_dynamic and option in (QueueOption.ADD, QueueOption.NEXT)
729
730        media_items: list[MediaItemType] = []
731        source_items: list[MediaItemType] = []
732        # resolve all media items
733        for item in media_list:
734            try:
735                # parse provided uri into a MA MediaItem or Basic QueueItem from URL
736                media_item: MediaItemType | ItemMapping | BrowseFolder
737                if isinstance(item, str):
738                    media_item = await self.mass.music.get_item_by_uri(item)
739                elif isinstance(item, dict):  # type: ignore[unreachable]
740                    # TODO: Investigate why the API parser sometimes passes raw dicts instead of
741                    # converting them to MediaItem objects. The parse_value function in api.py
742                    # should handle dict-to-object conversion, but dicts are slipping through
743                    # in some cases. This is defensive handling for that parser bug.
744                    media_item = media_from_dict(item)  # type: ignore[unreachable]
745                    self.logger.debug("Converted to: %s", type(media_item))
746                else:
747                    # item is MediaItemType | ItemMapping at this point
748                    media_item = item
749
750                if isinstance(media_item, ItemMapping):
751                    # Resolve any ItemMapping to its full media item, exactly as the str-uri
752                    # form above already does. Everything below needs the real object: the
753                    # enqueued/source bookkeeping only accepts full items (so a mapping would
754                    # otherwise never count as a user-initiated play), and the dynamic check
755                    # needs details such as a playlist's 'is_dynamic'.
756                    if media_item.uri is None:
757                        raise InvalidDataError("ItemMapping has no URI")
758                    media_item = await self.mass.music.get_item_by_uri(media_item.uri)
759
760                # Save requested media item to play on the queue so we can use it as a seed
761                # for Autoplay's music refill (the podcast/audiobook continuations resolve
762                # their successor from the queue's last item instead).
763                # Use FIFO list to keep track of the last 10 played items
764                # Skip ItemMapping and BrowseFolder - only queue full MediaItemType objects
765                if not isinstance(media_item, BrowseFolder) and (
766                    is_dynamic_source(media_item)
767                    or media_item.media_type
768                    in (MediaType.TRACK, MediaType.ALBUM, MediaType.PLAYLIST, MediaType.ARTIST)
769                ):
770                    queue_data.enqueued_media_items.append(media_item)
771                    if len(queue_data.enqueued_media_items) > 10:
772                        queue_data.enqueued_media_items.pop(0)
773                    if is_dynamic_source(media_item):
774                        # a dynamic playlist/station is always a self-managing dynamic source
775                        source_items.append(media_item)
776
777                # handle default enqueue option if needed
778                if option is None:
779                    # Radio + AudioSource share a single "live_sources" enqueue default —
780                    # both are live infinite streams where REPLACE is almost always the
781                    # right semantic. Other media types use their per-type config key.
782                    if media_item.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
783                        config_key = CONF_DEFAULT_ENQUEUE_OPTION_LIVE_SOURCES
784                    else:
785                        config_key = f"default_enqueue_option_{media_item.media_type.value}"
786                    config_value = self.get_config_value(config_key, return_type=str)
787                    option = QueueOption(config_value)
788                    if option == QueueOption.REPLACE:
789                        self._clear(queue_id, skip_stop=True)
790                    await self._apply_shuffle_intent(queue_id, option, shuffle)
791
792                # collect media_items to play
793                if is_dynamic_source(media_item):
794                    # a dynamic playlist/station supplies its own tracks on demand; just mark it
795                    # played. The queue goes dynamic below and the bounded pool seeds its batch from
796                    # all sources, so there is no need to fetch a batch here.
797                    self.mass.create_task(
798                        self.mass.music.mark_item_played(
799                            media_item,
800                            userid=queue_data.userid,
801                            queue_id=queue_id,
802                            user_initiated=True,
803                        )
804                    )
805                elif already_dynamic:
806                    # feed the already-active pool: keep the finite item as a (materialized) source
807                    if not isinstance(media_item, BrowseFolder):
808                        source_items.append(media_item)
809                else:
810                    # not (yet) a managed pool: record the finite parent as a source (kept for a
811                    # later dynamic transition and for similar/autoplay seeds) and expand it into
812                    # the linear queue
813                    if not isinstance(media_item, BrowseFolder) and media_item.media_type in (
814                        MediaType.TRACK,
815                        MediaType.ALBUM,
816                        MediaType.PLAYLIST,
817                        MediaType.ARTIST,
818                    ):
819                        source_items.append(media_item)
820                    # Convert start_item to string URI if needed
821                    start_item_uri: str | None = None
822                    if isinstance(start_item, str):
823                        start_item_uri = start_item
824                    elif start_item is not None:
825                        start_item_uri = start_item.uri
826                    media_items += await self._media_resolver._resolve_media_items(
827                        media_item,
828                        start_item_uri,
829                        userid=queue_data.userid,
830                        queue_id=queue_id,
831                        sort_by=sort_by,
832                        start_from_beginning=start_from_beginning,
833                        # under shuffle "start here and play forward" has no meaning, so keep the
834                        # whole playlist/album (chosen track first) instead of dropping everything
835                        # before it - the chosen track is pinned in front of the shuffled rest
836                        keep_preceding_items=queue.shuffle_enabled,
837                    )
838
839            except MusicAssistantError as err:
840                # invalid MA uri or item not found error
841                self.logger.warning("Skipping %s: %s", item, str(err))
842
843        # overwrite or append the queue's source items
844        replace_sources = option not in (QueueOption.ADD, QueueOption.NEXT)
845        if replace_sources:
846            self.store_sources(queue, source_items)
847        else:
848            self.store_sources(queue, self._queue_data[queue_id].source_items + source_items)
849        source_items = self._queue_data[queue_id].source_items
850        queue.is_dynamic = has_dynamic_source(source_items)
851        # a queue that just gained or lost its dynamic source resolves smart shuffle differently
852        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
853
854        if queue.is_dynamic:
855            # the queue has (or just gained) a dynamic source: (re)build the upcoming tail into a
856            # single bounded, recency-orchestrated mix over ALL sources — existing finite content as
857            # materialized TRACKS seed(s), dynamic playlists as DYNAMIC seed(s). Every add rebuilds
858            # from the buffer position, so the queue stays a fixed-size mix instead of growing by
859            # each added source's own batch.
860            await self._enter_dynamic_mode(queue_id, option)
861            return
862
863        # only add valid/available items
864        queue_items: list[QueueItem] = [
865            build_queue_item(queue_id, cast("PlayableMediaItemType", x))
866            for x in media_items
867            if x and x.available
868        ]
869
870        if not queue_items:
871            raise MediaNotFoundError("No playable items found", translation_key="no_playable_items")
872
873        await self._enqueue_with_option(
874            queue_id, queue_items, option, pin_first=start_item is not None
875        )
876
877    async def _enter_dynamic_mode(self, queue_id: str, option: QueueOption | None) -> None:
878        """
879        (Re)build a queue's upcoming tail into a single bounded managed pool over all its sources.
880
881        Runs whenever an enqueue leaves the queue dynamic — both the first transition and every
882        later add. Keeps the current + already-buffered track(s), drops the rest of the upcoming
883        tail, and replaces it with a bounded, recency-orchestrated mix of all the queue's sources
884        (finite sources materialized as TRACKS seeds, dynamic playlists as DYNAMIC seeds), so the
885        queue stays a fixed-size mix instead of growing by each added source's own batch. Shuffle is
886        enabled implicitly: a dynamic queue is always a smart mix.
887
888        :param queue_id: The queue to (re)build the dynamic pool for.
889        :param option: The enqueue option that triggered the (re)build. PLAY/REPLACE start playback
890            on the rebuilt pool; ADD/NEXT/REPLACE_NEXT stage it without starting playback (behind the
891            current/buffered track, or from the front of an idle/empty queue).
892        """
893        queue_data = self._queue_data[queue_id]
894        queue = queue_data.queue
895        # a dynamic queue is an always-on smart mix; reflect that in the (now locked) shuffle state
896        queue.shuffle_enabled = True
897        queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
898        # rebuild from the buffered position so the already-prepared next track is kept and the
899        # crossfade isn't disturbed; fall back to the current index (or the front when idle/empty)
900        base_index = (
901            queue.index_in_buffer if queue.index_in_buffer is not None else queue.current_index
902        )
903        insert_at = 0 if base_index is None else base_index + 1
904        # PLAY/REPLACE start playback on the rebuilt pool; ADD/NEXT/REPLACE_NEXT only stage it and
905        # never start playback (an idle/empty queue stays idle on an add, just like the linear path)
906        start_playing = option in (QueueOption.PLAY, QueueOption.REPLACE)
907        # drop the finite upcoming tail up front so the pool is sized and deduped against the kept
908        # head only (the tail we are discarding must not exclude its own tracks from the new pool)
909        queue_data.items = queue_data.items[:insert_at]
910        queue.items = len(queue_data.items)
911        pool_tracks = await self._managed_pool.fill(queue_id, is_initial=False)
912        queue_items = [
913            build_queue_item(queue_id, track) for track in pool_tracks if track.available
914        ]
915        if not queue_items:
916            raise MediaNotFoundError("No playable items found", translation_key="no_playable_items")
917        # the managed pool already interleaved the sources in a recency-aware order; load as-is
918        await self.load(queue_id, queue_items, insert_at_index=insert_at, keep_remaining=False)
919        if start_playing:
920            await self.play_index(queue_id, insert_at)
921        else:
922            # give an idle/empty queue a current item without starting playback
923            self._ensure_current_index(queue_id)
924
925    async def _get_similar_tracks(
926        self,
927        queue_id: str,
928        is_initial: bool = False,
929        seed_items: list[MediaItemType] | None = None,
930    ) -> list[Track]:
931        """
932        Fetch tracks similar to the given seeds (autoplay's similar/continuation mode).
933
934        :param queue_id: The queue to fetch tracks for.
935        :param is_initial: True to interleave the base/seed tracks into the result, False to
936            return only similar tracks.
937        :param seed_items: Explicit seed items to base the tracks on. Defaults to the queue's
938            sources; autoplay passes the enqueued media items instead.
939        """
940        queue_data = self._queue_data[queue_id]
941        queue = queue_data.queue
942        queue_track_items: list[Track] = [
943            q.media_item
944            for q in self._queue_data[queue_id].items
945            if q.media_item and isinstance(q.media_item, Track)
946        ]
947        source_items = (
948            seed_items if seed_items is not None else self._queue_data[queue_id].source_items
949        )
950        if not source_items:
951            # this may happen during race conditions as this method is called delayed
952            return []
953        self.logger.info(
954            "Fetching similar tracks for queue %s based on: %s",
955            queue.display_name,
956            ", ".join([x.name for x in source_items]),
957        )
958
959        # Get user's preferred provider instances for steering provider selection
960        preferred_provider_instances: list[str] | None = None
961        if (
962            queue_data.userid
963            and (playback_user := await self.mass.webserver.auth.get_user(queue_data.userid))
964            and playback_user.provider_filter
965        ):
966            preferred_provider_instances = playback_user.provider_filter
967
968        # Some providers have very deterministic similar-track algorithms for a single track
969        # seed. When continuing from a single track on a refill, seed from the play history
970        # instead so the result keeps varying.
971        if (
972            len(source_items) == 1
973            and source_items[0].media_type == MediaType.TRACK
974            and not is_initial
975            and queue_track_items
976        ):
977            # Helper samples 5 internally; bound the input.
978            seeds: list[MediaItemType] = random.sample(
979                queue_track_items, min(len(queue_track_items), 10)
980            )
981        else:
982            seeds = list(source_items)
983
984        radio_prov = self.mass.get_provider("radio_playlist")
985        if radio_prov is None:
986            return []
987        dynamic_tracks = await cast("RadioPlaylistProvider", radio_prov).get_dynamic_tracks(
988            seeds,
989            include_base_tracks=is_initial,
990            target_size=25,
991            preferred_provider_instances=preferred_provider_instances,
992        )
993        # Drop anything already queued/played
994        queued_set = set(queue_track_items)
995        return [track for track in dynamic_tracks if track not in queued_set]
996