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