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