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