/
/
/
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 if (queue_data := self._queue_data.get(queue_id)) is None:
516 # the delayed refill timer can fire after the queue was removed
517 return
518 queue = queue_data.queue
519 # restore the queue owner's user context so provider filters are respected during this
520 # background refill (dynamic-playlist generation honours the current user)
521 playback_user = (
522 await self.mass.webserver.auth.get_user(queue_data.userid)
523 if queue_data.userid
524 else None
525 )
526 set_current_user(playback_user)
527 # Top up from the queue's dynamic sources (dynamic playlists and any mixed-in finite items),
528 # weighted per source and recency-gated. fill() already sizes the batch to the pool target;
529 # the tail cap below is a defensive ceiling so the unplayed tail never grows past
530 # MANAGED_POOL_MAX.
531 pool_tracks = await self._managed_pool.fill(queue_id, is_initial=False)
532 if self._queue_data.get(queue_id) is not queue_data:
533 # the queue was removed or re-registered while tracks were fetched
534 return
535 # keep the unplayed tail within the bounded pool size (no current_index => nothing played yet)
536 played = 0 if queue.current_index is None else queue.current_index + 1
537 unplayed = max(len(queue_data.items) - played, 0)
538 headroom = max(MANAGED_POOL_MAX - unplayed, 0)
539 queue_items = [build_queue_item(queue_id, x) for x in pool_tracks[:headroom] if x.available]
540 if not queue_items:
541 return
542 await self.load(
543 queue_id,
544 queue_items,
545 insert_at_index=len(queue_data.items) + 1,
546 )
547
548 async def _fill_autoplay_tracks(self, queue_id: str) -> None:
549 """
550 Append more items to a queue that is running low, based on what is ending.
551
552 Autoplay is a single "keep going" switch; what it appends is decided by the media type
553 of the queue's last item, since that is the item the appended items follow.
554 """
555 queue = self.get(queue_id)
556 if queue is None or not queue.autoplay_enabled:
557 return
558 queue_data = self._queue_data[queue_id]
559 if not queue_data.items:
560 return
561 last_item = queue_data.items[-1]
562 if last_item.media_type in AUTOPLAY_EXCLUDED_MEDIA_TYPES:
563 return
564 # Restore the queue owner's user context so provider filters, library access and
565 # resume positions are respected during this background refill, mirroring
566 # _fill_dynamic_tracks.
567 playback_user = (
568 await self.mass.webserver.auth.get_user(queue_data.userid)
569 if queue_data.userid
570 else None
571 )
572 set_current_user(playback_user)
573 if self._queue_data.get(queue_id) is not queue_data:
574 # the queue was removed or re-registered while the user context was restored
575 return
576 if last_item.media_type in AUTOPLAY_SERIES_MEDIA_TYPES:
577 await self._fill_autoplay_next_in_series(queue_id, last_item)
578 return
579 await self._fill_autoplay_music_tracks(queue_id)
580
581 async def _fill_autoplay_next_in_series(self, queue_id: str, last_item: QueueItem) -> None:
582 """
583 Append the episode/book that follows the queue's last item, if there is one.
584
585 Nothing is appended for the last episode of a podcast or a book without a next one in
586 its collection, so the queue simply ends there.
587
588 :param queue_id: The queue to append to.
589 :param last_item: The queue's last item, an audiobook or podcast episode.
590 """
591 queue_data = self._queue_data[queue_id]
592 media_item = last_item.media_item
593 next_item: PodcastEpisode | Audiobook | None
594 try:
595 if isinstance(media_item, PodcastEpisode):
596 next_item = await self._media_resolver.get_next_podcast_episode(
597 media_item, userid=queue_data.userid
598 )
599 elif isinstance(media_item, Audiobook):
600 next_item = await self._media_resolver.get_next_audiobook(
601 media_item, userid=queue_data.userid
602 )
603 else:
604 return
605 except MusicAssistantError as err:
606 self.logger.warning(
607 "Autoplay failed to fetch the item following %s: %s", last_item.name, err
608 )
609 return
610 if next_item is None or not next_item.available:
611 self.logger.debug("Autoplay found nothing to play after %s", last_item.name)
612 return
613 if any(
614 item.media_item and item.media_item.uri == next_item.uri for item in queue_data.items
615 ):
616 # already queued (e.g. the user added it themselves), so there is nothing to do
617 return
618 if self._queue_data.get(queue_id) is not queue_data:
619 # the queue was removed or re-registered while the successor was fetched
620 return
621 await self.load(
622 queue_id,
623 [build_queue_item(queue_id, next_item)],
624 insert_at_index=len(queue_data.items) + 1,
625 )
626
627 async def _fill_autoplay_music_tracks(self, queue_id: str) -> None:
628 """Fill a Queue with additional tracks based on the configured Autoplay mode."""
629 queue = self.get(queue_id)
630 if queue is None:
631 return
632 queue_data = self._queue_data[queue_id]
633 if not queue_data.enqueued_media_items:
634 # the music refill needs what the user enqueued as its seed
635 return
636 mode = self._autoplay.resolve_mode(queue_id)
637 self.logger.debug(
638 "Filling autoplay tracks (mode: %s) for queue %s", mode.value, queue.display_name
639 )
640 existing_tracks = {
641 item.media_item
642 for item in self._queue_data[queue_id].items
643 if isinstance(item.media_item, Track)
644 }
645 try:
646 if mode == AutoplayMode.PLAYLIST:
647 tracks = await self._autoplay.get_playlist_tracks(queue, existing_tracks)
648 elif mode == AutoplayMode.LIBRARY:
649 tracks = await self._autoplay.get_library_tracks(queue, existing_tracks)
650 elif mode == AutoplayMode.SIMILAR:
651 tracks = await self._get_similar_tracks(
652 queue_id, seed_items=queue_data.enqueued_media_items
653 )
654 else:
655 # AUTO: try similar tracks first, fall back to the library mix. The similar
656 # fetch raises when no provider can supply base/similar tracks, so suppress
657 # that here to make sure the library fallback still runs.
658 tracks = []
659 with suppress(MusicAssistantError):
660 tracks = await self._get_similar_tracks(
661 queue_id, seed_items=queue_data.enqueued_media_items
662 )
663 if not tracks:
664 tracks = await self._autoplay.get_library_tracks(queue, existing_tracks)
665 except MusicAssistantError as err:
666 self.logger.warning(
667 "Autoplay failed to fetch tracks for queue %s: %s", queue.display_name, err
668 )
669 return
670 # route the autoplay batch through the recency engine so a recently-heard track isn't
671 # immediately re-added (ungated fallback keeps autoplay going if everything is recent)
672 windows = self._smart_shuffle.windows()
673 snapshot = await self.mass.music.recency.snapshot(windows, userid=queue_data.userid)
674 tracks = gate_tracks(
675 [track for track in tracks if isinstance(track, Track)], snapshot, windows
676 )
677 queue_items = [build_queue_item(queue_id, x) for x in tracks if x.available]
678 if not queue_items:
679 self.logger.info("Autoplay found no new tracks to add for queue %s", queue.display_name)
680 return
681 if self._queue_data.get(queue_id) is not queue_data:
682 # the queue was removed or re-registered while tracks were fetched
683 return
684 await self.load(
685 queue_id,
686 queue_items,
687 insert_at_index=len(queue_data.items) + 1,
688 )
689
690 @handle_play_action
691 async def _handle_play_media(
692 self,
693 queue_id: str,
694 media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
695 option: QueueOption | None = None,
696 radio_mode: bool = False,
697 start_item: PlayableMediaItemType | str | None = None,
698 sort_by: str | None = None,
699 start_from_beginning: bool = False,
700 shuffle: bool | None = None,
701 ) -> None:
702 """Handle play media without acquiring the queue lock."""
703 # cancel any pending play_index calls for this queue to prevent conflicts
704 self.mass.cancel_timer(f"queue_play_index_{queue_id}")
705 self._set_transitioning(queue_id, False)
706 # we use a contextvar to bypass the throttler for this asyncio task/context
707 # this makes sure that playback has priority over other requests that may be
708 # happening in the background
709 BYPASS_THROTTLER.set(True)
710 if not (queue := self.get(queue_id)):
711 raise PlayerUnavailableError(f"Queue {queue_id} is not available")
712 queue_data = self._queue_data[queue_id]
713 # always fetch the underlying player so we can raise early if its not available
714 queue_player = self.mass.players.get_player(queue_id, True)
715 assert queue_player is not None # for type checking
716 if queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
717 self.logger.warning("Ignore queue command: An announcement is in progress")
718 return
719
720 # save the user requesting the playback (clear it for anonymous playback)
721 playback_user = get_current_user()
722 queue_data.userid = playback_user.user_id if playback_user else None
723 if playback_user:
724 self.logger.debug(
725 "User %s requested playback.", playback_user.display_name or playback_user.username
726 )
727
728 # a single item or list of items may be provided
729 media_list = media if isinstance(media, list) else [media]
730
731 if radio_mode:
732 # radio_mode is deprecated: a "radio" is now a dynamic radio playlist. Translate each
733 # seed into the radio_playlist provider's URI and enqueue those (resolved to dynamic
734 # playlists that self-manage their refills).
735 self.logger.warning(
736 "radio_mode is deprecated; enqueue a radio_playlist:// dynamic playlist instead"
737 )
738 media_list = [
739 seed_uri
740 if (seed_uri := item if isinstance(item, str) else str(item.uri)).startswith(
741 "radio_playlist://"
742 )
743 else f"radio_playlist://playlist/{seed_uri}"
744 for item in media_list
745 ]
746 radio_mode = False
747
748 # Forget the previous queue's enqueued items when a new queue is requested. A caller that
749 # left the option to the config gets this once the first item resolved it, below: it is the
750 # option that says whether this is a new queue or an addition to the current one.
751 if option is not None and option not in (QueueOption.ADD, QueueOption.NEXT):
752 self._reset_enqueued_media_items(queue_data)
753 # An ADD/NEXT onto a queue that is already a managed pool (has a dynamic source): a finite
754 # item is kept only as a source (the bounded pool materializes it) instead of being expanded
755 # into the queue. Any other enqueue (PLAY/REPLACE, or onto a linear queue) expands finite
756 # items normally. Keys off is_dynamic since a finite-only queue records sources too.
757 # A play-next track is exempt from this (see plays_next_track below).
758 already_dynamic = queue.is_dynamic and option in (QueueOption.ADD, QueueOption.NEXT)
759
760 media_items: list[MediaItemType] = []
761 # the subset of media_items the user explicitly picked to play next
762 play_next_items: list[MediaItemType] = []
763 source_items: list[MediaItemType] = []
764 shuffle_settled = False
765 # resolve all media items
766 for item in media_list:
767 try:
768 # parse provided uri into a MA MediaItem or Basic QueueItem from URL
769 media_item: MediaItemType | ItemMapping | BrowseFolder
770 if isinstance(item, str):
771 media_item = await self.mass.music.get_item_by_uri(item)
772 elif isinstance(item, dict): # type: ignore[unreachable]
773 # TODO: Investigate why the API parser sometimes passes raw dicts instead of
774 # converting them to MediaItem objects. The parse_value function in api.py
775 # should handle dict-to-object conversion, but dicts are slipping through
776 # in some cases. This is defensive handling for that parser bug.
777 media_item = media_from_dict(item) # type: ignore[unreachable]
778 self.logger.debug("Converted to: %s", type(media_item))
779 else:
780 # item is MediaItemType | ItemMapping at this point
781 media_item = item
782
783 if isinstance(media_item, ItemMapping):
784 # Resolve any ItemMapping to its full media item, exactly as the str-uri
785 # form above already does. Everything below needs the real object: the
786 # enqueued/source bookkeeping only accepts full items (so a mapping would
787 # otherwise never count as a user-initiated play), and the dynamic check
788 # needs details such as a playlist's 'is_dynamic'.
789 if media_item.uri is None:
790 raise InvalidDataError("ItemMapping has no URI")
791 media_item = await self.mass.music.get_item_by_uri(media_item.uri)
792
793 # handle default enqueue option if needed
794 if option is None:
795 # Radio + AudioSource share a single "live_sources" enqueue default —
796 # both are live infinite streams where REPLACE is almost always the
797 # right semantic. Other media types use their per-type config key.
798 if media_item.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
799 config_key = CONF_DEFAULT_ENQUEUE_OPTION_LIVE_SOURCES
800 else:
801 config_key = f"default_enqueue_option_{media_item.media_type.value}"
802 config_value = self.get_config_value(config_key, return_type=str)
803 option = QueueOption(config_value)
804 if option not in (QueueOption.ADD, QueueOption.NEXT):
805 self._reset_enqueued_media_items(queue_data)
806 # settled from the resolved option for the same reason as the reset above
807 already_dynamic = queue.is_dynamic and option in (
808 QueueOption.ADD,
809 QueueOption.NEXT,
810 )
811
812 # Save requested media item to play on the queue so we can use it as a seed
813 # for Autoplay's music refill (the podcast/audiobook continuations resolve
814 # their successor from the queue's last item instead) and to tell which of its
815 # tracks play as part of an album the user picked.
816 # Use FIFO list to keep track of the last 10 played items
817 # Skip ItemMapping and BrowseFolder - only queue full MediaItemType objects
818 if not isinstance(media_item, BrowseFolder) and (
819 is_dynamic_source(media_item)
820 or media_item.media_type
821 in (MediaType.TRACK, MediaType.ALBUM, MediaType.PLAYLIST, MediaType.ARTIST)
822 ):
823 queue_data.enqueued_media_items.append(media_item)
824 if len(queue_data.enqueued_media_items) > 10:
825 evicted = queue_data.enqueued_media_items.pop(0)
826 # an album that dropped off the list can no longer be matched, so its
827 # credit is dead weight unless another entry still stands for it
828 if isinstance(evicted, Album) and evicted not in (
829 queue_data.enqueued_media_items
830 ):
831 queue_data.credited_albums.discard(evicted)
832 # enqueueing an album again is a new play of it, so let it be credited again
833 if isinstance(media_item, Album):
834 queue_data.credited_albums.discard(media_item)
835 if is_dynamic_source(media_item):
836 # a dynamic playlist/station is always a self-managing dynamic source
837 source_items.append(media_item)
838
839 # The shuffle state has to be settled before the items are resolved below: a
840 # shuffled queue keeps the items preceding a start_item (chosen track pinned
841 # first) instead of dropping them. The first item that resolves decides for the
842 # whole batch, because it is the only media type known this early.
843 if not shuffle_settled:
844 shuffle_settled = True
845 await self._apply_shuffle(
846 queue_id,
847 option,
848 # an explicit request always wins; only an unset one defers to the
849 # media's own order
850 False
851 if shuffle is None and media_item.media_type in ORDERED_MEDIA_TYPES
852 else shuffle,
853 )
854
855 # the user picked this exact track to play next, so it must be inserted literally
856 plays_next_track = (
857 option == QueueOption.NEXT and media_item.media_type == MediaType.TRACK
858 )
859 # collect media_items to play
860 if is_dynamic_source(media_item):
861 # a dynamic playlist/station supplies its own tracks on demand; just mark it
862 # played. The queue goes dynamic below and the bounded pool seeds its batch from
863 # all sources, so there is no need to fetch a batch here.
864 self.mass.create_task(
865 self.mass.music.mark_item_played(
866 media_item,
867 userid=queue_data.userid,
868 queue_id=queue_id,
869 user_initiated=True,
870 )
871 )
872 elif already_dynamic and not plays_next_track:
873 # feed the already-active pool: keep the finite item as a (materialized) source
874 if not isinstance(media_item, BrowseFolder):
875 source_items.append(media_item)
876 else:
877 # a play-next track never becomes a source: the pool would re-dispatch it later
878 if (
879 not plays_next_track
880 and not isinstance(media_item, BrowseFolder)
881 and media_item.media_type
882 in (
883 MediaType.TRACK,
884 MediaType.ALBUM,
885 MediaType.PLAYLIST,
886 MediaType.ARTIST,
887 )
888 ):
889 # record the finite parent as a source (kept for a later dynamic
890 # transition and for similar/autoplay seeds)
891 source_items.append(media_item)
892 # Convert start_item to string URI if needed
893 start_item_uri: str | None = None
894 if isinstance(start_item, str):
895 start_item_uri = start_item
896 elif start_item is not None:
897 start_item_uri = start_item.uri
898 resolved_items = await self._media_resolver._resolve_media_items(
899 media_item,
900 start_item_uri,
901 userid=queue_data.userid,
902 queue_id=queue_id,
903 sort_by=sort_by,
904 start_from_beginning=start_from_beginning,
905 # under shuffle "start here and play forward" has no meaning, so keep the
906 # whole playlist/album (chosen track first) instead of dropping everything
907 # before it - the chosen track is pinned in front of the shuffled rest
908 keep_preceding_items=queue.shuffle_enabled,
909 )
910 media_items += resolved_items
911 if plays_next_track:
912 play_next_items += resolved_items
913
914 except MusicAssistantError as err:
915 # invalid MA uri or item not found error
916 self.logger.warning("Skipping %s: %s", item, str(err))
917
918 if not shuffle_settled and option is not None:
919 # nothing resolved, so no media type ever decided - but the sources are replaced
920 # below all the same, and a dynamic queue's imposed shuffle must not survive that
921 await self._apply_shuffle(queue_id, option, shuffle)
922
923 # captured before the reassignment below replaces the local with the stored list
924 new_sources = bool(source_items)
925 # overwrite or append the queue's source items
926 replace_sources = option not in (QueueOption.ADD, QueueOption.NEXT)
927 if replace_sources:
928 self.store_sources(queue, source_items)
929 else:
930 self.store_sources(queue, self._queue_data[queue_id].source_items + source_items)
931 source_items = self._queue_data[queue_id].source_items
932 queue.is_dynamic = has_dynamic_source(source_items)
933 # a queue that just gained or lost its dynamic source resolves smart shuffle differently
934 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
935
936 if queue.is_dynamic:
937 if replace_sources or new_sources:
938 # the queue has (or just gained) a dynamic source: (re)build the upcoming tail into
939 # a single bounded, recency-orchestrated mix over ALL sources — existing finite
940 # content as materialized TRACKS seed(s), dynamic playlists as DYNAMIC seed(s).
941 # Only rebuilt when this enqueue changed the sources, so a play-next insert
942 # leaves the tail untouched.
943 await self._enter_dynamic_mode(queue_id, option)
944 # only explicit play-next tracks are inserted literally; container expansions are
945 # already in the pool via their source
946 media_items = play_next_items
947 if not media_items:
948 return
949 # fall through: play-next track(s) are inserted after the buffered index below
950
951 # only add valid/available items
952 queue_items: list[QueueItem] = [
953 build_queue_item(queue_id, cast("PlayableMediaItemType", x))
954 for x in media_items
955 if x and x.available
956 ]
957
958 if not queue_items:
959 raise MediaNotFoundError("No playable items found", translation_key="no_playable_items")
960
961 await self._enqueue_with_option(
962 queue_id, queue_items, option, pin_first=start_item is not None
963 )
964
965 async def _enter_dynamic_mode(self, queue_id: str, option: QueueOption | None) -> None:
966 """
967 (Re)build a queue's upcoming tail into a single bounded managed pool over all its sources.
968
969 Runs whenever an enqueue leaves the queue dynamic — both the first transition and every
970 later add. Keeps the current + already-buffered track(s), drops the rest of the upcoming
971 tail, and replaces it with a bounded, recency-orchestrated mix of all the queue's sources
972 (finite sources materialized as TRACKS seeds, dynamic playlists as DYNAMIC seeds), so the
973 queue stays a fixed-size mix instead of growing by each added source's own batch. Shuffle is
974 enabled implicitly: a dynamic queue is always a smart mix.
975
976 :param queue_id: The queue to (re)build the dynamic pool for.
977 :param option: The enqueue option that triggered the (re)build. PLAY/REPLACE start playback
978 on the rebuilt pool; ADD/NEXT/REPLACE_NEXT stage it without starting playback (behind the
979 current/buffered track, or from the front of an idle/empty queue).
980 """
981 queue_data = self._queue_data[queue_id]
982 queue = queue_data.queue
983 # a dynamic queue is an always-on smart mix; reflect that in the (now locked) shuffle state
984 queue.shuffle_enabled = True
985 queue.smart_shuffle_active = self.is_smart_shuffle_active(queue)
986 # rebuild from the buffered position so the already-prepared next track is kept and the
987 # crossfade isn't disturbed; fall back to the current index (or the front when idle/empty)
988 base_index = (
989 queue.index_in_buffer if queue.index_in_buffer is not None else queue.current_index
990 )
991 insert_at = 0 if base_index is None else base_index + 1
992 if option == QueueOption.REPLACE:
993 # A replace is a fresh queue, so the pool takes the place of the old items rather than
994 # being appended behind the one that is playing (as PLAY, which shares start_playing,
995 # deliberately does). Zeroed before the truncation below so the pool is sized against
996 # an empty queue and none of the discarded tracks are held back from it.
997 insert_at = 0
998 # as on the linear path: release the outgoing audio while its items are still on the
999 # queue, and drop the stale position
1000 await self._cleanup_queue_audio_data(queue_id)
1001 queue.index_in_buffer = None
1002 queue.ended = False
1003 # PLAY/REPLACE start playback on the rebuilt pool; ADD/NEXT/REPLACE_NEXT only stage it and
1004 # never start playback (an idle/empty queue stays idle on an add, just like the linear path)
1005 start_playing = option in (QueueOption.PLAY, QueueOption.REPLACE)
1006 # The tail is dropped before the pool is fetched, so the pool is sized and deduped against
1007 # the kept head only (the tail we are discarding must not exclude its own tracks from it).
1008 # That leaves the queue holding less than it plays - for a replace, nothing at all - across
1009 # the fetch, so hold player reconciliation off until the new items are in: it would
1010 # otherwise publish that half-built state, which is exactly the empty queue this avoids.
1011 self._set_transitioning(queue_id, True)
1012 try:
1013 queue_data.items = queue_data.items[:insert_at]
1014 queue.items = len(queue_data.items)
1015 pool_tracks = await self._managed_pool.fill(queue_id, is_initial=False)
1016 queue_items = [
1017 build_queue_item(queue_id, track) for track in pool_tracks if track.available
1018 ]
1019 if not queue_items:
1020 raise MediaNotFoundError(
1021 "No playable items found", translation_key="no_playable_items"
1022 )
1023 # the managed pool already interleaved the sources in a recency-aware order; load as-is
1024 await self.load(
1025 queue_id,
1026 queue_items,
1027 insert_at_index=insert_at,
1028 keep_remaining=False,
1029 keep_played=option != QueueOption.REPLACE,
1030 )
1031 if start_playing:
1032 await self.play_index(queue_id, insert_at)
1033 else:
1034 # give an idle/empty queue a current item without starting playback
1035 self._ensure_current_index(queue_id)
1036 finally:
1037 self._set_transitioning(queue_id, False)
1038
1039 async def _get_similar_tracks(
1040 self,
1041 queue_id: str,
1042 is_initial: bool = False,
1043 seed_items: list[MediaItemType] | None = None,
1044 ) -> list[Track]:
1045 """
1046 Fetch tracks similar to the given seeds (autoplay's similar/continuation mode).
1047
1048 :param queue_id: The queue to fetch tracks for.
1049 :param is_initial: True to interleave the base/seed tracks into the result, False to
1050 return only similar tracks.
1051 :param seed_items: Explicit seed items to base the tracks on. Defaults to the queue's
1052 sources; autoplay passes the enqueued media items instead.
1053 """
1054 queue_data = self._queue_data[queue_id]
1055 queue = queue_data.queue
1056 queue_track_items: list[Track] = [
1057 q.media_item
1058 for q in self._queue_data[queue_id].items
1059 if q.media_item and isinstance(q.media_item, Track)
1060 ]
1061 source_items = (
1062 seed_items if seed_items is not None else self._queue_data[queue_id].source_items
1063 )
1064 if not source_items:
1065 # this may happen during race conditions as this method is called delayed
1066 return []
1067 self.logger.info(
1068 "Fetching similar tracks for queue %s based on: %s",
1069 queue.display_name,
1070 ", ".join([x.name for x in source_items]),
1071 )
1072
1073 # Get user's preferred provider instances for steering provider selection
1074 preferred_provider_instances: list[str] | None = None
1075 if (
1076 queue_data.userid
1077 and (playback_user := await self.mass.webserver.auth.get_user(queue_data.userid))
1078 and playback_user.provider_filter
1079 ):
1080 preferred_provider_instances = playback_user.provider_filter
1081
1082 # Some providers have very deterministic similar-track algorithms for a single track
1083 # seed. When continuing from a single track on a refill, seed from the play history
1084 # instead so the result keeps varying.
1085 if (
1086 len(source_items) == 1
1087 and source_items[0].media_type == MediaType.TRACK
1088 and not is_initial
1089 and queue_track_items
1090 ):
1091 # Helper samples 5 internally; bound the input.
1092 seeds: list[MediaItemType] = random.sample(
1093 queue_track_items, min(len(queue_track_items), 10)
1094 )
1095 else:
1096 seeds = list(source_items)
1097
1098 radio_prov = self.mass.get_provider("radio_playlist")
1099 if radio_prov is None:
1100 return []
1101 dynamic_tracks = await cast("RadioPlaylistProvider", radio_prov).get_dynamic_tracks(
1102 seeds,
1103 include_base_tracks=is_initial,
1104 target_size=25,
1105 preferred_provider_instances=preferred_provider_instances,
1106 )
1107 # Drop anything already queued/played
1108 queued_set = set(queue_track_items)
1109 return [track for track in dynamic_tracks if track not in queued_set]
1110
1111 async def _abort_superseded_source_buffers(self, queue_item: QueueItem) -> None:
1112 """
1113 Abort the still-filling source buffers of other items in the same queue.
1114
1115 :param queue_item: The queue item that is about to start playing.
1116 """
1117 queue_data = self._queue_data.get(queue_item.queue_id)
1118 items = tuple(queue_data.items) if queue_data else ()
1119 successor: QueueItem | None = None
1120 for index, item in enumerate(items):
1121 if item.queue_item_id == queue_item.queue_item_id and index + 1 < len(items):
1122 successor = items[index + 1]
1123 break
1124 # the started item keeps its own buffer, and its direct successor keeps the prewarm
1125 # for the upcoming crossfade unless the aborts below leave the provider without a slot
1126 spared_item_ids = {queue_item.queue_item_id}
1127 if successor is not None:
1128 spared_item_ids.add(successor.queue_item_id)
1129 for item in items:
1130 if item.queue_item_id in spared_item_ids:
1131 continue
1132 await self._abort_source_buffer(item, queue_item)
1133 if successor is not None:
1134 await self._abort_source_buffer(successor, queue_item, only_when_saturated=True)
1135
1136 async def _abort_source_buffer(
1137 self,
1138 item: QueueItem,
1139 started_item: QueueItem,
1140 only_when_saturated: bool = False,
1141 ) -> None:
1142 """
1143 Cancel one item's still-filling source so its provider stream slot is handed over.
1144
1145 :param item: The queue item whose source buffer should be aborted.
1146 :param started_item: The queue item that is about to start playing.
1147 :param only_when_saturated: Only abort while the provider has no free slot left.
1148 """
1149 if item.streamdetails is None:
1150 return
1151 audio_buffer = item.streamdetails.buffer
1152 if audio_buffer is None or not audio_buffer.is_buffering:
1153 return
1154 provider = self.mass.get_provider(item.streamdetails.provider, return_unavailable=True)
1155 if not isinstance(provider, MusicProvider) or provider.max_concurrent_streams is None:
1156 return
1157 if only_when_saturated:
1158 if provider.has_available_stream_slot:
1159 # an abort above already freed a slot, so this prewarm can stay
1160 return
1161 self.logger.debug(
1162 "Aborting the prewarm of %s: %s has no free stream slot left for %s",
1163 item.name,
1164 provider.name,
1165 started_item.name,
1166 )
1167 else:
1168 self.logger.debug(
1169 "Aborting the source of %s to free a %s stream slot for %s",
1170 item.name,
1171 provider.name,
1172 started_item.name,
1173 )
1174 # the cancelled buffer stays attached: it marks the source as aborted for
1175 # the flow stream's accounting and fails is_valid() for any later reuse
1176 await audio_buffer.clear()
1177