music-assistant-server
114.5 KB•PY
player_queues.py
114.5 KB • 2,620 lines • python
1"""
2MusicAssistant Player Queues Controller.
3
4Handles all logic to PLAY Media Items, provided by Music Providers to supported players.
5
6It is loosely coupled to the MusicAssistant Music Controller and Player Controller.
7A Music Assistant Player always has a PlayerQueue associated with it
8which holds the queue items and state.
9
10The PlayerQueue is in that case the active source of the player,
11but it can also be something else, hence the loose coupling.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import random
18import time
19from contextlib import suppress
20from types import NoneType
21from typing import TYPE_CHECKING, Any, TypedDict, cast
22
23import shortuuid
24from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
25from music_assistant_models.enums import (
26 ConfigEntryType,
27 ContentType,
28 EventType,
29 MediaType,
30 PlaybackState,
31 PlayerFeature,
32 ProviderFeature,
33 QueueOption,
34 RepeatMode,
35)
36from music_assistant_models.errors import (
37 AudioError,
38 InvalidCommand,
39 InvalidDataError,
40 MediaNotFoundError,
41 MusicAssistantError,
42 PlayerUnavailableError,
43 QueueEmpty,
44 UnsupportedFeaturedException,
45)
46from music_assistant_models.media_items import (
47 Album,
48 Artist,
49 Audiobook,
50 BrowseFolder,
51 ItemMapping,
52 MediaItemType,
53 PlayableMediaItemType,
54 Playlist,
55 Podcast,
56 PodcastEpisode,
57 Track,
58 UniqueList,
59 media_from_dict,
60)
61from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
62from music_assistant_models.player_queue import PlayerQueue
63from music_assistant_models.queue_item import QueueItem
64
65from music_assistant.constants import (
66 ATTR_ANNOUNCEMENT_IN_PROGRESS,
67 CONF_FLOW_MODE,
68 MASS_LOGO_ONLINE,
69 VERBOSE_LOG_LEVEL,
70)
71from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
72from music_assistant.helpers.api import api_command
73from music_assistant.helpers.audio import get_stream_details, get_stream_dsp_details
74from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
75from music_assistant.helpers.util import get_changed_keys, percentage
76from music_assistant.models.core_controller import CoreController
77from music_assistant.models.player import Player, PlayerMedia
78
79if TYPE_CHECKING:
80 from collections.abc import Iterator
81
82 from music_assistant_models.auth import User
83 from music_assistant_models.media_items.metadata import MediaItemImage
84
85 from music_assistant import MusicAssistant
86 from music_assistant.models.player import Player
87
88
89CONF_DEFAULT_ENQUEUE_SELECT_ARTIST = "default_enqueue_select_artist"
90CONF_DEFAULT_ENQUEUE_SELECT_ALBUM = "default_enqueue_select_album"
91
92ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE = "all_tracks"
93ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE = "all_tracks"
94
95CONF_DEFAULT_ENQUEUE_OPTION_ARTIST = "default_enqueue_option_artist"
96CONF_DEFAULT_ENQUEUE_OPTION_ALBUM = "default_enqueue_option_album"
97CONF_DEFAULT_ENQUEUE_OPTION_TRACK = "default_enqueue_option_track"
98CONF_DEFAULT_ENQUEUE_OPTION_RADIO = "default_enqueue_option_radio"
99CONF_DEFAULT_ENQUEUE_OPTION_PLAYLIST = "default_enqueue_option_playlist"
100CONF_DEFAULT_ENQUEUE_OPTION_AUDIOBOOK = "default_enqueue_option_audiobook"
101CONF_DEFAULT_ENQUEUE_OPTION_PODCAST = "default_enqueue_option_podcast"
102CONF_DEFAULT_ENQUEUE_OPTION_PODCAST_EPISODE = "default_enqueue_option_podcast_episode"
103CONF_DEFAULT_ENQUEUE_OPTION_FOLDER = "default_enqueue_option_folder"
104CONF_DEFAULT_ENQUEUE_OPTION_UNKNOWN = "default_enqueue_option_unknown"
105RADIO_TRACK_MAX_DURATION_SECS = 20 * 60 # 20 minutes
106CACHE_CATEGORY_PLAYER_QUEUE_STATE = 0
107CACHE_CATEGORY_PLAYER_QUEUE_ITEMS = 1
108
109
110class CompareState(TypedDict):
111 """Simple object where we store the (previous) state of a queue.
112
113 Used for compare actions.
114 """
115
116 queue_id: str
117 state: PlaybackState
118 current_item_id: str | None
119 next_item_id: str | None
120 current_item: QueueItem | None
121 elapsed_time: int
122 # last_playing_elapsed_time: elapsed time from the last PLAYING state update
123 # used to determine if a track was fully played when transitioning to idle
124 last_playing_elapsed_time: int
125 stream_title: str | None
126 codec_type: ContentType | None
127 output_formats: list[str] | None
128
129
130class PlayerQueuesController(CoreController):
131 """Controller holding all logic to enqueue music for players."""
132
133 domain: str = "player_queues"
134
135 def __init__(self, mass: MusicAssistant) -> None:
136 """Initialize core controller."""
137 super().__init__(mass)
138 self._queues: dict[str, PlayerQueue] = {}
139 self._queue_items: dict[str, list[QueueItem]] = {}
140 self._prev_states: dict[str, CompareState] = {}
141 self._transitioning_players: set[str] = set()
142 self.manifest.name = "Player Queues controller"
143 self.manifest.description = (
144 "Music Assistant's core controller which manages the queues for all players."
145 )
146 self.manifest.icon = "playlist-music"
147
148 async def close(self) -> None:
149 """Cleanup on exit."""
150 # stop all playback
151 for queue in self.all():
152 if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
153 await self.stop(queue.queue_id)
154
155 async def get_config_entries(
156 self,
157 action: str | None = None,
158 values: dict[str, ConfigValueType] | None = None,
159 ) -> tuple[ConfigEntry, ...]:
160 """Return all Config Entries for this core module (if any)."""
161 enqueue_options = [ConfigValueOption(x.name, x.value) for x in QueueOption]
162 return (
163 ConfigEntry(
164 key=CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
165 type=ConfigEntryType.STRING,
166 default_value=ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE,
167 label="Items to select when you play a (in-library) artist.",
168 options=[
169 ConfigValueOption(
170 title="Only in-library tracks",
171 value="library_tracks",
172 ),
173 ConfigValueOption(
174 title="All tracks from all albums in the library",
175 value="library_album_tracks",
176 ),
177 ConfigValueOption(
178 title="All (top) tracks from (all) streaming provider(s)",
179 value="all_tracks",
180 ),
181 ConfigValueOption(
182 title="All tracks from all albums from (all) streaming provider(s)",
183 value="all_album_tracks",
184 ),
185 ],
186 ),
187 ConfigEntry(
188 key=CONF_DEFAULT_ENQUEUE_SELECT_ALBUM,
189 type=ConfigEntryType.STRING,
190 default_value=ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE,
191 label="Items to select when you play a (in-library) album.",
192 options=[
193 ConfigValueOption(
194 title="Only in-library tracks",
195 value="library_tracks",
196 ),
197 ConfigValueOption(
198 title="All tracks for album on (streaming) provider",
199 value="all_tracks",
200 ),
201 ],
202 ),
203 ConfigEntry(
204 key=CONF_DEFAULT_ENQUEUE_OPTION_ARTIST,
205 type=ConfigEntryType.STRING,
206 default_value=QueueOption.REPLACE.value,
207 label="Default enqueue option for Artist item(s).",
208 options=enqueue_options,
209 description="Define the default enqueue action for this mediatype.",
210 ),
211 ConfigEntry(
212 key=CONF_DEFAULT_ENQUEUE_OPTION_ALBUM,
213 type=ConfigEntryType.STRING,
214 default_value=QueueOption.REPLACE.value,
215 label="Default enqueue option for Album item(s).",
216 options=enqueue_options,
217 description="Define the default enqueue action for this mediatype.",
218 ),
219 ConfigEntry(
220 key=CONF_DEFAULT_ENQUEUE_OPTION_TRACK,
221 type=ConfigEntryType.STRING,
222 default_value=QueueOption.PLAY.value,
223 label="Default enqueue option for Track item(s).",
224 options=enqueue_options,
225 description="Define the default enqueue action for this mediatype.",
226 ),
227 ConfigEntry(
228 key=CONF_DEFAULT_ENQUEUE_OPTION_RADIO,
229 type=ConfigEntryType.STRING,
230 default_value=QueueOption.REPLACE.value,
231 label="Default enqueue option for Radio item(s).",
232 options=enqueue_options,
233 description="Define the default enqueue action for this mediatype.",
234 ),
235 ConfigEntry(
236 key=CONF_DEFAULT_ENQUEUE_OPTION_PLAYLIST,
237 type=ConfigEntryType.STRING,
238 default_value=QueueOption.REPLACE.value,
239 label="Default enqueue option for Playlist item(s).",
240 options=enqueue_options,
241 description="Define the default enqueue action for this mediatype.",
242 ),
243 ConfigEntry(
244 key=CONF_DEFAULT_ENQUEUE_OPTION_AUDIOBOOK,
245 type=ConfigEntryType.STRING,
246 default_value=QueueOption.REPLACE.value,
247 label="Default enqueue option for Audiobook item(s).",
248 options=enqueue_options,
249 hidden=True,
250 ),
251 ConfigEntry(
252 key=CONF_DEFAULT_ENQUEUE_OPTION_PODCAST,
253 type=ConfigEntryType.STRING,
254 default_value=QueueOption.REPLACE.value,
255 label="Default enqueue option for Podcast item(s).",
256 options=enqueue_options,
257 hidden=True,
258 ),
259 ConfigEntry(
260 key=CONF_DEFAULT_ENQUEUE_OPTION_PODCAST_EPISODE,
261 type=ConfigEntryType.STRING,
262 default_value=QueueOption.REPLACE.value,
263 label="Default enqueue option for Podcast-episode item(s).",
264 options=enqueue_options,
265 hidden=True,
266 ),
267 ConfigEntry(
268 key=CONF_DEFAULT_ENQUEUE_OPTION_FOLDER,
269 type=ConfigEntryType.STRING,
270 default_value=QueueOption.REPLACE.value,
271 label="Default enqueue option for Folder item(s).",
272 options=enqueue_options,
273 hidden=True,
274 ),
275 )
276
277 def __iter__(self) -> Iterator[PlayerQueue]:
278 """Iterate over (available) players."""
279 return iter(self._queues.values())
280
281 @api_command("player_queues/all")
282 def all(self) -> tuple[PlayerQueue, ...]:
283 """Return all registered PlayerQueues."""
284 return tuple(self._queues.values())
285
286 @api_command("player_queues/get")
287 def get(self, queue_id: str) -> PlayerQueue | None:
288 """Return PlayerQueue by queue_id or None if not found."""
289 return self._queues.get(queue_id)
290
291 @api_command("player_queues/items")
292 def items(self, queue_id: str, limit: int = 500, offset: int = 0) -> list[QueueItem]:
293 """Return all QueueItems for given PlayerQueue."""
294 if queue_id not in self._queue_items:
295 return []
296
297 return self._queue_items[queue_id][offset : offset + limit]
298
299 @api_command("player_queues/get_active_queue")
300 def get_active_queue(self, player_id: str) -> PlayerQueue | None:
301 """Return the current active/synced queue for a player."""
302 if player := self.mass.players.get(player_id):
303 return self.mass.players.get_active_queue(player)
304 return None
305
306 # Queue commands
307
308 @api_command("player_queues/shuffle")
309 async def set_shuffle(self, queue_id: str, shuffle_enabled: bool) -> None:
310 """Configure shuffle setting on the the queue."""
311 queue = self._queues[queue_id]
312 if queue.shuffle_enabled == shuffle_enabled:
313 return # no change
314 queue.shuffle_enabled = shuffle_enabled
315 queue_items = self._queue_items[queue_id]
316 cur_index = queue.index_in_buffer or queue.current_index
317 if cur_index is not None:
318 next_index = cur_index + 1
319 next_items = queue_items[next_index:]
320 else:
321 next_items = []
322 next_index = 0
323 if not shuffle_enabled:
324 # shuffle disabled, try to restore original sort order of the remaining items
325 next_items.sort(key=lambda x: x.sort_index, reverse=False)
326 await self.load(
327 queue_id=queue_id,
328 queue_items=next_items,
329 insert_at_index=next_index,
330 keep_remaining=False,
331 shuffle=shuffle_enabled,
332 )
333
334 @api_command("player_queues/dont_stop_the_music")
335 def set_dont_stop_the_music(self, queue_id: str, dont_stop_the_music_enabled: bool) -> None:
336 """Configure Don't stop the music setting on the queue."""
337 providers_available_with_similar_tracks = any(
338 ProviderFeature.SIMILAR_TRACKS in provider.supported_features
339 for provider in self.mass.music.providers
340 )
341 if dont_stop_the_music_enabled and not providers_available_with_similar_tracks:
342 raise UnsupportedFeaturedException(
343 "Don't stop the music is not supported by any of the available music providers"
344 )
345 queue = self._queues[queue_id]
346 queue.dont_stop_the_music_enabled = dont_stop_the_music_enabled
347 self.signal_update(queue_id=queue_id)
348 # if this happens to be the last track in the queue, fill the radio source
349 if (
350 queue.dont_stop_the_music_enabled
351 and queue.enqueued_media_items
352 and queue.current_index is not None
353 and (queue.items - queue.current_index) <= 1
354 ):
355 queue.radio_source = queue.enqueued_media_items
356 task_id = f"fill_radio_tracks_{queue_id}"
357 self.mass.call_later(5, self._fill_radio_tracks, queue_id, task_id=task_id)
358
359 @api_command("player_queues/repeat")
360 def set_repeat(self, queue_id: str, repeat_mode: RepeatMode) -> None:
361 """Configure repeat setting on the the queue."""
362 queue = self._queues[queue_id]
363 if queue.repeat_mode == repeat_mode:
364 return # no change
365 queue.repeat_mode = repeat_mode
366 self.signal_update(queue_id)
367 if (
368 queue.state == PlaybackState.PLAYING
369 and queue.index_in_buffer is not None
370 and queue.index_in_buffer == queue.current_index
371 ):
372 # if the queue is playing,
373 # ensure to (re)queue the next track because it might have changed
374 # note that we only do this if the player has loaded the current track
375 # if not, we wait until it has loaded to prevent conflicts
376 if next_item := self.get_next_item(queue_id, queue.index_in_buffer):
377 self._enqueue_next_item(queue_id, next_item)
378
379 @api_command("player_queues/play_media")
380 async def play_media(
381 self,
382 queue_id: str,
383 media: MediaItemType | ItemMapping | str | list[MediaItemType | ItemMapping | str],
384 option: QueueOption | None = None,
385 radio_mode: bool = False,
386 start_item: PlayableMediaItemType | str | None = None,
387 username: str | None = None,
388 ) -> None:
389 """Play media item(s) on the given queue.
390
391 :param queue_id: The queue_id of the queue to play media on.
392 :param media: Media that should be played (MediaItem(s) and/or uri's).
393 :param option: Which enqueue mode to use.
394 :param radio_mode: Enable radio mode for the given item(s).
395 :param start_item: Optional item to start the playlist or album from.
396 :param username: The username of the user requesting the playback.
397 Setting the username allows for overriding the logged-in user
398 to account for playback history per user when the play_media is
399 called from a shared context (like a web hook or automation).
400 """
401 # ruff: noqa: PLR0915
402 # we use a contextvar to bypass the throttler for this asyncio task/context
403 # this makes sure that playback has priority over other requests that may be
404 # happening in the background
405 BYPASS_THROTTLER.set(True)
406 if not (queue := self.get(queue_id)):
407 raise PlayerUnavailableError(f"Queue {queue_id} is not available")
408 # always fetch the underlying player so we can raise early if its not available
409 queue_player = self.mass.players.get(queue_id, True)
410 assert queue_player is not None # for type checking
411 if queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
412 self.logger.warning("Ignore queue command: An announcement is in progress")
413 return
414
415 # save the user requesting the playback
416 playback_user: User | None
417 if username and (user := await self.mass.webserver.auth.get_user_by_username(username)):
418 playback_user = user
419 else:
420 playback_user = get_current_user()
421 queue.userid = playback_user.user_id if playback_user else None
422
423 # a single item or list of items may be provided
424 media_list = media if isinstance(media, list) else [media]
425
426 # clear queue if needed
427 if option == QueueOption.REPLACE:
428 self.clear(queue_id)
429 # Clear the 'enqueued media item' list when a new queue is requested
430 if option not in (QueueOption.ADD, QueueOption.NEXT):
431 queue.enqueued_media_items.clear()
432
433 media_items: list[MediaItemType] = []
434 radio_source: list[MediaItemType] = []
435 # resolve all media items
436 for item in media_list:
437 try:
438 # parse provided uri into a MA MediaItem or Basic QueueItem from URL
439 media_item: MediaItemType | ItemMapping | BrowseFolder
440 if isinstance(item, str):
441 media_item = await self.mass.music.get_item_by_uri(item)
442 elif isinstance(item, dict): # type: ignore[unreachable]
443 # TODO: Investigate why the API parser sometimes passes raw dicts instead of
444 # converting them to MediaItem objects. The parse_value function in api.py
445 # should handle dict-to-object conversion, but dicts are slipping through
446 # in some cases. This is defensive handling for that parser bug.
447 media_item = media_from_dict(item) # type: ignore[unreachable]
448 self.logger.debug("Converted to: %s", type(media_item))
449 else:
450 # item is MediaItemType | ItemMapping at this point
451 media_item = item
452
453 # Save requested media item to play on the queue so we can use it as a source
454 # for Don't stop the music. Use FIFO list to keep track of the last 10 played items
455 # Skip ItemMapping and BrowseFolder - only queue full MediaItemType objects
456 if not isinstance(
457 media_item, (ItemMapping, BrowseFolder)
458 ) and media_item.media_type in (
459 MediaType.TRACK,
460 MediaType.ALBUM,
461 MediaType.PLAYLIST,
462 MediaType.ARTIST,
463 ):
464 queue.enqueued_media_items.append(media_item)
465 if len(queue.enqueued_media_items) > 10:
466 queue.enqueued_media_items.pop(0)
467
468 # handle default enqueue option if needed
469 if option is None:
470 config_value = await self.mass.config.get_core_config_value(
471 self.domain,
472 f"default_enqueue_option_{media_item.media_type.value}",
473 return_type=str,
474 )
475 option = QueueOption(config_value)
476 if option == QueueOption.REPLACE:
477 self.clear(queue_id, skip_stop=True)
478
479 # collect media_items to play
480 if radio_mode:
481 # Type guard for mypy - only add full MediaItemType to radio_source
482 if not isinstance(media_item, (ItemMapping, BrowseFolder)):
483 radio_source.append(media_item)
484 else:
485 # Convert start_item to string URI if needed
486 start_item_uri: str | None = None
487 if isinstance(start_item, str):
488 start_item_uri = start_item
489 elif start_item is not None:
490 start_item_uri = start_item.uri
491 media_items += await self._resolve_media_items(
492 media_item, start_item_uri, queue_id=queue_id
493 )
494
495 except MusicAssistantError as err:
496 # invalid MA uri or item not found error
497 self.logger.warning("Skipping %s: %s", item, str(err))
498
499 # overwrite or append radio source items
500 if option not in (QueueOption.ADD, QueueOption.NEXT):
501 queue.radio_source = radio_source
502 else:
503 queue.radio_source += radio_source
504 # Use collected media items to calculate the radio if radio mode is on
505 if radio_mode:
506 radio_tracks = await self._get_radio_tracks(
507 queue_id=queue_id, is_initial_radio_mode=True
508 )
509 media_items = list(radio_tracks)
510
511 # only add valid/available items
512 queue_items: list[QueueItem] = []
513 for x in media_items:
514 if not x or not x.available:
515 continue
516 queue_items.append(
517 QueueItem.from_media_item(queue_id, cast("PlayableMediaItemType", x))
518 )
519
520 if not queue_items:
521 raise MediaNotFoundError("No playable items found")
522
523 # load the items into the queue
524 if queue.state in (PlaybackState.PLAYING, PlaybackState.PAUSED):
525 cur_index = queue.index_in_buffer or queue.current_index or 0
526 else:
527 cur_index = queue.current_index or 0
528 insert_at_index = cur_index + 1
529 # Radio modes are already shuffled in a pattern we would like to keep.
530 shuffle = queue.shuffle_enabled and len(queue_items) > 1 and not radio_mode
531
532 # handle replace: clear all items and replace with the new items
533 if option == QueueOption.REPLACE:
534 await self.load(
535 queue_id,
536 queue_items=queue_items,
537 keep_remaining=False,
538 keep_played=False,
539 shuffle=shuffle,
540 )
541 await self.play_index(queue_id, 0)
542 return
543 # handle next: add item(s) in the index next to the playing/loaded/buffered index
544 if option == QueueOption.NEXT:
545 await self.load(
546 queue_id,
547 queue_items=queue_items,
548 insert_at_index=insert_at_index,
549 shuffle=shuffle,
550 )
551 return
552 if option == QueueOption.REPLACE_NEXT:
553 await self.load(
554 queue_id,
555 queue_items=queue_items,
556 insert_at_index=insert_at_index,
557 keep_remaining=False,
558 shuffle=shuffle,
559 )
560 return
561 # handle play: replace current loaded/playing index with new item(s)
562 if option == QueueOption.PLAY:
563 await self.load(
564 queue_id,
565 queue_items=queue_items,
566 insert_at_index=insert_at_index,
567 shuffle=shuffle,
568 )
569 next_index = min(insert_at_index, len(self._queue_items[queue_id]) - 1)
570 await self.play_index(queue_id, next_index)
571 return
572 # handle add: add/append item(s) to the remaining queue items
573 if option == QueueOption.ADD:
574 await self.load(
575 queue_id=queue_id,
576 queue_items=queue_items,
577 insert_at_index=insert_at_index
578 if queue.shuffle_enabled
579 else len(self._queue_items[queue_id]) + 1,
580 shuffle=queue.shuffle_enabled,
581 )
582 # handle edgecase, queue is empty and items are only added (not played)
583 # mark first item as new index
584 if queue.current_index is None:
585 queue.current_index = 0
586 queue.current_item = self.get_item(queue_id, 0)
587 queue.items = len(queue_items)
588 self.signal_update(queue_id)
589
590 @api_command("player_queues/move_item")
591 def move_item(self, queue_id: str, queue_item_id: str, pos_shift: int = 1) -> None:
592 """
593 Move queue item x up/down the queue.
594
595 - queue_id: id of the queue to process this request.
596 - queue_item_id: the item_id of the queueitem that needs to be moved.
597 - pos_shift: move item x positions down if positive value
598 - pos_shift: move item x positions up if negative value
599 - pos_shift: move item to top of queue as next item if 0.
600 """
601 queue = self._queues[queue_id]
602 item_index = self.index_by_id(queue_id, queue_item_id)
603 if item_index is None:
604 raise InvalidDataError(f"Item {queue_item_id} not found in queue")
605 if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
606 msg = f"{item_index} is already played/buffered"
607 raise IndexError(msg)
608
609 queue_items = self._queue_items[queue_id]
610 queue_items = queue_items.copy()
611
612 if pos_shift == 0 and queue.state == PlaybackState.PLAYING:
613 new_index = (queue.current_index or 0) + 1
614 elif pos_shift == 0:
615 new_index = queue.current_index or 0
616 else:
617 new_index = item_index + pos_shift
618 if (new_index < (queue.current_index or 0)) or (new_index > len(queue_items)):
619 return
620 # move the item in the list
621 queue_items.insert(new_index, queue_items.pop(item_index))
622 self.update_items(queue_id, queue_items)
623
624 @api_command("player_queues/move_item_end")
625 def move_item_end(self, queue_id: str, queue_item_id: str) -> None:
626 """
627 Move queue item to the end the queue.
628
629 - queue_id: id of the queue to process this request.
630 - queue_item_id: the item_id of the queueitem that needs to be moved.
631 """
632 queue = self._queues[queue_id]
633 item_index = self.index_by_id(queue_id, queue_item_id)
634 if item_index is None:
635 raise InvalidDataError(f"Item {queue_item_id} not found in queue")
636 if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
637 msg = f"{item_index} is already played/buffered"
638 raise IndexError(msg)
639
640 queue_items = self._queue_items[queue_id]
641 if item_index == (len(queue_items) - 1):
642 return
643 queue_items = queue_items.copy()
644
645 new_index = len(self._queue_items[queue_id]) - 1
646
647 # move the item in the list
648 queue_items.insert(new_index, queue_items.pop(item_index))
649 self.update_items(queue_id, queue_items)
650
651 @api_command("player_queues/delete_item")
652 def delete_item(self, queue_id: str, item_id_or_index: int | str) -> None:
653 """Delete item (by id or index) from the queue."""
654 if isinstance(item_id_or_index, str):
655 item_index = self.index_by_id(queue_id, item_id_or_index)
656 if item_index is None:
657 raise InvalidDataError(f"Item {item_id_or_index} not found in queue")
658 else:
659 item_index = item_id_or_index
660 queue = self._queues[queue_id]
661 if queue.index_in_buffer is not None and item_index <= queue.index_in_buffer:
662 # ignore request if track already loaded in the buffer
663 # the frontend should guard so this is just in case
664 self.logger.warning("delete requested for item already loaded in buffer")
665 return
666 queue_items = self._queue_items[queue_id]
667 queue_items.pop(item_index)
668 self.update_items(queue_id, queue_items)
669
670 @api_command("player_queues/clear")
671 def clear(self, queue_id: str, skip_stop: bool = False) -> None:
672 """Clear all items in the queue."""
673 queue = self._queues[queue_id]
674 queue.radio_source = []
675 if queue.state != PlaybackState.IDLE and not skip_stop:
676 self.mass.create_task(self.stop(queue_id))
677 queue.current_index = None
678 queue.current_item = None
679 queue.elapsed_time = 0
680 queue.elapsed_time_last_updated = time.time()
681 queue.index_in_buffer = None
682 self.update_items(queue_id, [])
683
684 @api_command("player_queues/stop")
685 async def stop(self, queue_id: str) -> None:
686 """
687 Handle STOP command for given queue.
688
689 - queue_id: queue_id of the playerqueue to handle the command.
690 """
691 queue_player = self.mass.players.get(queue_id, True)
692 if queue_player is None:
693 raise PlayerUnavailableError(f"Player {queue_id} is not available")
694 if (queue := self.get(queue_id)) and queue.active:
695 if queue.state == PlaybackState.PLAYING:
696 queue.resume_pos = int(queue.corrected_elapsed_time)
697 # forward the actual command to the player
698 if temp_player := self.mass.players.get(queue_id):
699 await temp_player.stop()
700
701 @api_command("player_queues/play")
702 async def play(self, queue_id: str) -> None:
703 """
704 Handle PLAY command for given queue.
705
706 - queue_id: queue_id of the playerqueue to handle the command.
707 """
708 queue_player = self.mass.players.get(queue_id, True)
709 if queue_player is None:
710 raise PlayerUnavailableError(f"Player {queue_id} is not available")
711 if (
712 (queue := self._queues.get(queue_id))
713 and queue.active
714 and queue.state == PlaybackState.PAUSED
715 ):
716 # forward the actual play/unpause command to the player
717 await queue_player.play()
718 return
719 # player is not paused, perform resume instead
720 await self.resume(queue_id)
721
722 @api_command("player_queues/pause")
723 async def pause(self, queue_id: str) -> None:
724 """Handle PAUSE command for given queue.
725
726 - queue_id: queue_id of the playerqueue to handle the command.
727 """
728 if queue := self._queues.get(queue_id):
729 if queue.state == PlaybackState.PLAYING:
730 queue.resume_pos = int(queue.corrected_elapsed_time)
731 # forward the actual command to the player controller
732 queue_player = self.mass.players.get(queue_id)
733 assert queue_player is not None # for type checking
734 if not (self.mass.players.get_player_provider(queue_id)):
735 return # guard
736
737 if PlayerFeature.PAUSE not in queue_player.supported_features:
738 # if player does not support pause, we need to send stop
739 await queue_player.stop()
740 return
741 await queue_player.pause()
742
743 async def _watch_pause() -> None:
744 count = 0
745 # wait for pause
746 while count < 5 and queue_player.playback_state == PlaybackState.PLAYING:
747 count += 1
748 await asyncio.sleep(1)
749 # wait for unpause
750 if queue_player.playback_state != PlaybackState.PAUSED:
751 return
752 count = 0
753 while count < 30 and queue_player.playback_state == PlaybackState.PAUSED:
754 count += 1
755 await asyncio.sleep(1)
756 # if player is still paused when the limit is reached, send stop
757 if queue_player.playback_state == PlaybackState.PAUSED:
758 await queue_player.stop()
759
760 # we auto stop a player from paused when its paused for 30 seconds
761 if not queue_player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
762 self.mass.create_task(_watch_pause())
763
764 @api_command("player_queues/play_pause")
765 async def play_pause(self, queue_id: str) -> None:
766 """Toggle play/pause on given playerqueue.
767
768 - queue_id: queue_id of the queue to handle the command.
769 """
770 if (queue := self._queues.get(queue_id)) and queue.state == PlaybackState.PLAYING:
771 await self.pause(queue_id)
772 return
773 await self.play(queue_id)
774
775 @api_command("player_queues/next")
776 async def next(self, queue_id: str) -> None:
777 """Handle NEXT TRACK command for given queue.
778
779 - queue_id: queue_id of the queue to handle the command.
780 """
781 if (queue := self.get(queue_id)) is None or not queue.active:
782 # TODO: forward to underlying player if not active
783 return
784 idx = self._queues[queue_id].current_index
785 if idx is None:
786 self.logger.warning("Queue %s has no current index", queue.display_name)
787 return
788 attempts = 5
789 while attempts:
790 try:
791 if (next_index := self._get_next_index(queue_id, idx, True)) is not None:
792 await self.play_index(queue_id, next_index, debounce=True)
793 break
794 except MediaNotFoundError:
795 self.logger.warning(
796 "Failed to fetch next track for queue %s - trying next item",
797 queue.display_name,
798 )
799 idx += 1
800 attempts -= 1
801
802 @api_command("player_queues/previous")
803 async def previous(self, queue_id: str) -> None:
804 """Handle PREVIOUS TRACK command for given queue.
805
806 - queue_id: queue_id of the queue to handle the command.
807 """
808 if (queue := self.get(queue_id)) is None or not queue.active:
809 # TODO: forward to underlying player if not active
810 return
811 current_index = self._queues[queue_id].current_index
812 if current_index is None:
813 return
814 next_index = int(current_index)
815 # restart current track if current track has played longer than 4
816 # otherwise skip to previous track
817 if self._queues[queue_id].elapsed_time < 5:
818 next_index = max(current_index - 1, 0)
819 await self.play_index(queue_id, next_index, debounce=True)
820
821 @api_command("player_queues/skip")
822 async def skip(self, queue_id: str, seconds: int = 10) -> None:
823 """Handle SKIP command for given queue.
824
825 - queue_id: queue_id of the queue to handle the command.
826 - seconds: number of seconds to skip in track. Use negative value to skip back.
827 """
828 if (queue := self.get(queue_id)) is None or not queue.active:
829 # TODO: forward to underlying player if not active
830 return
831 await self.seek(queue_id, int(self._queues[queue_id].elapsed_time + seconds))
832
833 @api_command("player_queues/seek")
834 async def seek(self, queue_id: str, position: int = 10) -> None:
835 """Handle SEEK command for given queue.
836
837 - queue_id: queue_id of the queue to handle the command.
838 - position: position in seconds to seek to in the current playing item.
839 """
840 if not (queue := self.get(queue_id)):
841 return
842 queue_player = self.mass.players.get(queue_id, True)
843 if queue_player is None:
844 raise PlayerUnavailableError(f"Player {queue_id} is not available")
845 if not queue.current_item:
846 raise InvalidCommand(f"Queue {queue_player.display_name} has no item(s) loaded.")
847 if not queue.current_item.duration:
848 raise InvalidCommand("Can not seek items without duration.")
849 position = max(0, int(position))
850 if position > queue.current_item.duration:
851 raise InvalidCommand("Can not seek outside of duration range.")
852 if queue.current_index is None:
853 raise InvalidCommand(f"Queue {queue_player.display_name} has no current index.")
854 await self.play_index(queue_id, queue.current_index, seek_position=position)
855
856 @api_command("player_queues/resume")
857 async def resume(self, queue_id: str, fade_in: bool | None = None) -> None:
858 """Handle RESUME command for given queue.
859
860 - queue_id: queue_id of the queue to handle the command.
861 """
862 queue = self._queues[queue_id]
863 queue_items = self._queue_items[queue_id]
864 resume_item = queue.current_item
865 if queue.state == PlaybackState.PLAYING:
866 # resume requested while already playing,
867 # use current position as resume position
868 resume_pos = queue.corrected_elapsed_time
869 fade_in = False
870 else:
871 resume_pos = queue.resume_pos or queue.elapsed_time
872
873 if not resume_item and queue.current_index is not None and len(queue_items) > 0:
874 resume_item = self.get_item(queue_id, queue.current_index)
875 resume_pos = 0
876 elif not resume_item and queue.current_index is None and len(queue_items) > 0:
877 # items available in queue but no previous track, start at 0
878 resume_item = self.get_item(queue_id, 0)
879 resume_pos = 0
880
881 if resume_item is not None:
882 queue_player = self.mass.players.get(queue_id)
883 if queue_player is None:
884 raise PlayerUnavailableError(f"Player {queue_id} is not available")
885 if (
886 fade_in is None
887 and queue_player.playback_state == PlaybackState.IDLE
888 and (time.time() - queue.elapsed_time_last_updated) > 60
889 ):
890 # enable fade in effect if the player is idle for a while
891 fade_in = resume_pos > 0
892 if resume_item.media_type == MediaType.RADIO:
893 # we're not able to skip in online radio so this is pointless
894 resume_pos = 0
895 await self.play_index(
896 queue_id, resume_item.queue_item_id, int(resume_pos), fade_in or False
897 )
898 else:
899 # Queue is empty, try to resume from playlog
900 if await self._try_resume_from_playlog(queue):
901 return
902 msg = f"Resume queue requested but queue {queue.display_name} is empty"
903 raise QueueEmpty(msg)
904
905 @api_command("player_queues/play_index")
906 async def play_index(
907 self,
908 queue_id: str,
909 index: int | str,
910 seek_position: int = 0,
911 fade_in: bool = False,
912 debounce: bool = False,
913 ) -> None:
914 """Play item at index (or item_id) X in queue."""
915 queue = self._queues[queue_id]
916 queue.resume_pos = 0
917 if isinstance(index, str):
918 temp_index = self.index_by_id(queue_id, index)
919 if temp_index is None:
920 raise InvalidDataError(f"Item {index} not found in queue")
921 index = temp_index
922 # At this point index is guaranteed to be int
923 queue.current_index = index
924 # update current item and elapsed time and signal update
925 # this way the UI knows immediately that a new item is loading
926 queue.current_item = self.get_item(queue_id, index)
927 queue.elapsed_time = seek_position
928 queue.elapsed_time_last_updated = time.time()
929 self.signal_update(queue_id)
930 queue.index_in_buffer = index
931 queue.flow_mode_stream_log = []
932 target_player = self.mass.players.get(queue_id)
933 if target_player is None:
934 raise PlayerUnavailableError(f"Player {queue_id} is not available")
935 enqueue_supported = PlayerFeature.ENQUEUE in target_player.supported_features
936 queue.next_item_id_enqueued = None
937 # always update session id when we start a new playback session
938 queue.session_id = shortuuid.random(length=8)
939 # handle resume point of audiobook(chapter) or podcast(episode)
940 if (
941 not seek_position
942 and (queue_item := self.get_item(queue_id, index))
943 and (resume_position_ms := getattr(queue_item.media_item, "resume_position_ms", 0))
944 ):
945 seek_position = max(0, int((resume_position_ms - 500) / 1000))
946
947 # send play_media request to player
948 # NOTE that we debounce this a bit to account for someone hitting the next button
949 # like a madman. This will prevent the player from being overloaded with requests.
950 async def _play_index(index: int, debounce: bool) -> None:
951 for attempt in range(5):
952 try:
953 queue_item = self.get_item(queue_id, index)
954 if not queue_item:
955 continue # guard
956 await self._load_item(
957 queue_item,
958 self._get_next_index(queue_id, index),
959 is_start=True,
960 seek_position=seek_position if attempt == 0 else 0,
961 fade_in=fade_in if attempt == 0 else False,
962 )
963 # if we reach this point, loading the item succeeded, break the loop
964 queue.current_index = index
965 queue.current_item = queue_item
966 break
967 except (MediaNotFoundError, AudioError):
968 # the requested index can not be played.
969 if queue_item:
970 self.logger.warning(
971 "Skipping unplayable item %s (%s)",
972 queue_item.name,
973 queue_item.uri,
974 )
975 queue_item.available = False
976 next_index = self._get_next_index(queue_id, index, allow_repeat=False)
977 if next_index is None:
978 raise MediaNotFoundError("No next item available")
979 index = next_index
980 else:
981 # all attempts to find a playable item failed
982 raise MediaNotFoundError("No playable item found to start playback")
983
984 # work out if we need to use flow mode
985 prefer_flow_mode = await self.mass.config.get_player_config_value(
986 queue_id, CONF_FLOW_MODE, default=False
987 )
988 flow_mode = (
989 prefer_flow_mode or not enqueue_supported
990 ) and queue_item.media_type not in (
991 # don't use flow mode for duration-less streams
992 MediaType.RADIO,
993 MediaType.PLUGIN_SOURCE,
994 )
995 await asyncio.sleep(0.5 if debounce else 0.1)
996 queue.flow_mode = flow_mode
997 await self.mass.players.play_media(
998 player_id=queue_id,
999 media=await self.player_media_from_queue_item(queue_item, flow_mode),
1000 )
1001 queue.current_index = index
1002 queue.current_item = queue_item
1003 await asyncio.sleep(2)
1004 self._transitioning_players.discard(queue_id)
1005
1006 # we set a flag to notify the update logic that we're transitioning to a new track
1007 self._transitioning_players.add(queue_id)
1008
1009 # we debounce the play_index command to handle the case where someone
1010 # is spamming next/previous on the player
1011 task_id = f"play_index_{queue_id}"
1012 if existing_task := self.mass.get_task(task_id):
1013 existing_task.cancel()
1014 with suppress(asyncio.CancelledError):
1015 await existing_task
1016 task = self.mass.create_task(
1017 _play_index,
1018 index,
1019 debounce,
1020 task_id=task_id,
1021 )
1022 await task
1023 self.signal_update(queue_id)
1024
1025 @api_command("player_queues/transfer")
1026 async def transfer_queue(
1027 self,
1028 source_queue_id: str,
1029 target_queue_id: str,
1030 auto_play: bool | None = None,
1031 ) -> None:
1032 """Transfer queue to another queue."""
1033 if not (source_queue := self.get(source_queue_id)):
1034 raise PlayerUnavailableError(f"Queue {source_queue_id} is not available")
1035 if not (target_queue := self.get(target_queue_id)):
1036 raise PlayerUnavailableError(f"Queue {target_queue_id} is not available")
1037 if auto_play is None:
1038 auto_play = source_queue.state == PlaybackState.PLAYING
1039
1040 target_player = self.mass.players.get(target_queue_id)
1041 if target_player is None:
1042 raise PlayerUnavailableError(f"Player {target_queue_id} is not available")
1043 if target_player.active_group or target_player.synced_to:
1044 # edge case: the user wants to move playback from the group as a whole, to a single
1045 # player in the group or it is grouped and the command targeted at the single player.
1046 # We need to dissolve the group first.
1047 group_id = target_player.active_group or target_player.synced_to
1048 assert group_id is not None # checked in if condition above
1049 await self.mass.players.cmd_ungroup(group_id)
1050 await asyncio.sleep(3)
1051
1052 source_items = self._queue_items[source_queue_id]
1053 target_queue.repeat_mode = source_queue.repeat_mode
1054 target_queue.shuffle_enabled = source_queue.shuffle_enabled
1055 target_queue.dont_stop_the_music_enabled = source_queue.dont_stop_the_music_enabled
1056 target_queue.radio_source = source_queue.radio_source
1057 target_queue.enqueued_media_items = source_queue.enqueued_media_items
1058 target_queue.resume_pos = int(source_queue.elapsed_time)
1059 target_queue.current_index = source_queue.current_index
1060 if source_queue.current_item:
1061 target_queue.current_item = source_queue.current_item
1062 target_queue.current_item.queue_id = target_queue_id
1063 self.clear(source_queue_id)
1064
1065 await self.load(target_queue_id, source_items, keep_remaining=False, keep_played=False)
1066 for item in source_items:
1067 item.queue_id = target_queue_id
1068 self.update_items(target_queue_id, source_items)
1069 if auto_play:
1070 await self.resume(target_queue_id)
1071
1072 # Interaction with player
1073
1074 async def on_player_register(self, player: Player) -> None:
1075 """Register PlayerQueue for given player/queue id."""
1076 queue_id = player.player_id
1077 queue: PlayerQueue | None = None
1078 queue_items: list[QueueItem] = []
1079 # try to restore previous state
1080 if prev_state := await self.mass.cache.get(
1081 key=queue_id,
1082 provider=self.domain,
1083 category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1084 ):
1085 try:
1086 queue = PlayerQueue.from_dict(prev_state)
1087 prev_items = await self.mass.cache.get(
1088 key=queue_id,
1089 provider=self.domain,
1090 category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1091 default=[],
1092 )
1093 queue_items = []
1094 for idx, item_data in enumerate(prev_items):
1095 qi = QueueItem.from_cache(item_data)
1096 if not qi.media_item:
1097 # Skip items with missing media_item - this can happen if
1098 # MA was killed during shutdown while cache was being written
1099 self.logger.debug(
1100 "Skipping queue item %s (index %d) restored from cache "
1101 "without media_item",
1102 qi.name,
1103 idx,
1104 )
1105 continue
1106 queue_items.append(qi)
1107 if queue.enqueued_media_items:
1108 # we need to restore the MediaItem objects for the enqueued media items
1109 # Items from cache may be dicts that need deserialization
1110 restored_enqueued_items: list[MediaItemType] = []
1111 cached_items: list[dict[str, Any] | MediaItemType] = cast(
1112 "list[dict[str, Any] | MediaItemType]", queue.enqueued_media_items
1113 )
1114 for item in cached_items:
1115 if isinstance(item, dict):
1116 restored_item = media_from_dict(item)
1117 restored_enqueued_items.append(cast("MediaItemType", restored_item))
1118 else:
1119 restored_enqueued_items.append(item)
1120 queue.enqueued_media_items = restored_enqueued_items
1121 except Exception as err:
1122 self.logger.warning(
1123 "Failed to restore the queue(items) for %s - %s",
1124 player.display_name,
1125 str(err),
1126 )
1127 # Reset to clean state on failure
1128 queue = None
1129 queue_items = []
1130 if queue is None:
1131 queue = PlayerQueue(
1132 queue_id=queue_id,
1133 active=False,
1134 display_name=player.display_name,
1135 available=player.available,
1136 dont_stop_the_music_enabled=False,
1137 items=0,
1138 )
1139
1140 self._queues[queue_id] = queue
1141 self._queue_items[queue_id] = queue_items
1142 # always call update to calculate state etc
1143 self.on_player_update(player, {})
1144 self.mass.signal_event(EventType.QUEUE_ADDED, object_id=queue_id, data=queue)
1145
1146 def on_player_update(
1147 self,
1148 player: Player,
1149 changed_values: dict[str, tuple[Any, Any]],
1150 ) -> None:
1151 """
1152 Call when a PlayerQueue needs to be updated (e.g. when player updates).
1153
1154 NOTE: This is called every second if the player is playing.
1155 """
1156 queue_id = player.player_id
1157 if (queue := self._queues.get(queue_id)) is None:
1158 # race condition
1159 return
1160 if player.extra_data.get(ATTR_ANNOUNCEMENT_IN_PROGRESS):
1161 # do nothing while the announcement is in progress
1162 return
1163 # determine if this queue is currently active for this player
1164 queue.active = player.active_source in (queue.queue_id, None)
1165 if not queue.active and queue_id not in self._prev_states:
1166 queue.state = PlaybackState.IDLE
1167 # return early if the queue is not active and we have no previous state
1168 return
1169 if queue.queue_id in self._transitioning_players:
1170 # we're currently transitioning to a new track,
1171 # ignore updates from the player during this time
1172 return
1173
1174 # queue is active and preflight checks passed, update the queue details
1175 self._update_queue_from_player(player)
1176
1177 def on_player_remove(self, player_id: str, permanent: bool) -> None:
1178 """Call when a player is removed from the registry."""
1179 if permanent:
1180 # if the player is permanently removed, we also remove the cached queue data
1181 self.mass.create_task(
1182 self.mass.cache.delete(
1183 key=player_id,
1184 provider=self.domain,
1185 category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1186 )
1187 )
1188 self.mass.create_task(
1189 self.mass.cache.delete(
1190 key=player_id,
1191 provider=self.domain,
1192 category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1193 )
1194 )
1195 self._queues.pop(player_id, None)
1196 self._queue_items.pop(player_id, None)
1197
1198 async def load_next_queue_item(
1199 self,
1200 queue_id: str,
1201 current_item_id: str,
1202 ) -> QueueItem:
1203 """
1204 Call when a player wants the next queue item to play.
1205
1206 Raises QueueEmpty if there are no more tracks left.
1207 """
1208 queue = self.get(queue_id)
1209 if not queue:
1210 msg = f"PlayerQueue {queue_id} is not available"
1211 raise PlayerUnavailableError(msg)
1212 cur_index = self.index_by_id(queue_id, current_item_id)
1213 if cur_index is None:
1214 # this is just a guard for bad data
1215 raise QueueEmpty("Invalid item id for queue given.")
1216 next_item: QueueItem | None = None
1217 idx = 0
1218 while True:
1219 next_index = self._get_next_index(queue_id, cur_index + idx)
1220 if next_index is None:
1221 raise QueueEmpty("No more tracks left in the queue.")
1222 queue_item = self.get_item(queue_id, next_index)
1223 if queue_item is None:
1224 raise QueueEmpty("No more tracks left in the queue.")
1225 if idx >= 10:
1226 # we only allow 10 retries to prevent infinite loops
1227 raise QueueEmpty("No more (playable) tracks left in the queue.")
1228 try:
1229 await self._load_item(queue_item, next_index)
1230 # we're all set, this is our next item
1231 next_item = queue_item
1232 break
1233 except (MediaNotFoundError, AudioError):
1234 # No stream details found, skip this QueueItem
1235 self.logger.warning(
1236 "Skipping unplayable item %s (%s)", queue_item.name, queue_item.uri
1237 )
1238 queue_item.available = False
1239 idx += 1
1240 if idx != 0:
1241 # we skipped some items, signal a queue items update
1242 self.update_items(queue_id, self._queue_items[queue_id])
1243 if next_item is None:
1244 raise QueueEmpty("No more (playable) tracks left in the queue.")
1245
1246 return next_item
1247
1248 async def _load_item(
1249 self,
1250 queue_item: QueueItem,
1251 next_index: int | None,
1252 is_start: bool = False,
1253 seek_position: int = 0,
1254 fade_in: bool = False,
1255 ) -> None:
1256 """Try to load the stream details for the given queue item."""
1257 queue_id = queue_item.queue_id
1258 queue = self._queues[queue_id]
1259
1260 # we use a contextvar to bypass the throttler for this asyncio task/context
1261 # this makes sure that playback has priority over other requests that may be
1262 # happening in the background
1263 BYPASS_THROTTLER.set(True)
1264
1265 self.logger.debug(
1266 "(pre)loading (next) item for queue %s...",
1267 queue.display_name,
1268 )
1269
1270 if not queue_item.available:
1271 raise MediaNotFoundError(f"Item {queue_item.uri} is not available")
1272
1273 # work out if we are playing an album and if we should prefer album
1274 # loudness
1275 next_track_from_same_album = (
1276 next_index is not None
1277 and (next_item := self.get_item(queue_id, next_index))
1278 and (
1279 queue_item.media_item
1280 and hasattr(queue_item.media_item, "album")
1281 and queue_item.media_item.album
1282 and next_item.media_item
1283 and hasattr(next_item.media_item, "album")
1284 and next_item.media_item.album
1285 and queue_item.media_item.album.item_id == next_item.media_item.album.item_id
1286 )
1287 )
1288 current_index = self.index_by_id(queue_id, queue_item.queue_item_id)
1289 if current_index is None:
1290 previous_track_from_same_album = False
1291 else:
1292 previous_index = max(current_index - 1, 0)
1293 previous_track_from_same_album = (
1294 previous_index > 0
1295 and (previous_item := self.get_item(queue_id, previous_index)) is not None
1296 and previous_item.media_item is not None
1297 and hasattr(previous_item.media_item, "album")
1298 and previous_item.media_item.album is not None
1299 and queue_item.media_item is not None
1300 and hasattr(queue_item.media_item, "album")
1301 and queue_item.media_item.album is not None
1302 and queue_item.media_item.album.item_id == previous_item.media_item.album.item_id
1303 )
1304 playing_album_tracks = next_track_from_same_album or previous_track_from_same_album
1305 if queue_item.media_item and isinstance(queue_item.media_item, Track):
1306 album = queue_item.media_item.album
1307 # prefer the full library media item so we have all metadata and provider(quality) info
1308 # always request the full library item as there might be other qualities available
1309 if library_item := await self.mass.music.get_library_item_by_prov_id(
1310 queue_item.media_item.media_type,
1311 queue_item.media_item.item_id,
1312 queue_item.media_item.provider,
1313 ):
1314 queue_item.media_item = cast("Track", library_item)
1315 elif not queue_item.media_item.image or queue_item.media_item.provider.startswith(
1316 "ytmusic"
1317 ):
1318 # Youtube Music has poor thumbs by default, so we always fetch the full item
1319 # this also catches the case where they have an unavailable item in a listing
1320 fetched_item = await self.mass.music.get_item_by_uri(queue_item.uri)
1321 queue_item.media_item = cast("Track", fetched_item)
1322
1323 # ensure we got the full (original) album set
1324 if album and (
1325 library_album := await self.mass.music.get_library_item_by_prov_id(
1326 album.media_type,
1327 album.item_id,
1328 album.provider,
1329 )
1330 ):
1331 queue_item.media_item.album = cast("Album", library_album)
1332 elif album:
1333 # Restore original album if we have no better alternative from the library
1334 queue_item.media_item.album = album
1335 # prefer album image over track image
1336 if queue_item.media_item.album and queue_item.media_item.album.image:
1337 org_images: list[MediaItemImage] = queue_item.media_item.metadata.images or []
1338 queue_item.media_item.metadata.images = UniqueList(
1339 [
1340 queue_item.media_item.album.image,
1341 *org_images,
1342 ]
1343 )
1344 # Fetch the streamdetails, which could raise in case of an unplayable item.
1345 # For example, YT Music returns Radio Items that are not playable.
1346 queue_item.streamdetails = await get_stream_details(
1347 mass=self.mass,
1348 queue_item=queue_item,
1349 seek_position=seek_position,
1350 fade_in=fade_in,
1351 prefer_album_loudness=bool(playing_album_tracks),
1352 )
1353
1354 def track_loaded_in_buffer(self, queue_id: str, item_id: str) -> None:
1355 """Call when a player has (started) loading a track in the buffer."""
1356 queue = self.get(queue_id)
1357 if not queue:
1358 msg = f"PlayerQueue {queue_id} is not available"
1359 raise PlayerUnavailableError(msg)
1360 # store the index of the item that is currently (being) loaded in the buffer
1361 # which helps us a bit to determine how far the player has buffered ahead
1362 queue.index_in_buffer = self.index_by_id(queue_id, item_id)
1363 self.logger.debug("PlayerQueue %s loaded item %s in buffer", queue.display_name, item_id)
1364 self.signal_update(queue_id)
1365 # preload next streamdetails
1366 self._preload_next_item(queue_id, item_id)
1367
1368 # Main queue manipulation methods
1369
1370 async def load(
1371 self,
1372 queue_id: str,
1373 queue_items: list[QueueItem],
1374 insert_at_index: int = 0,
1375 keep_remaining: bool = True,
1376 keep_played: bool = True,
1377 shuffle: bool = False,
1378 ) -> None:
1379 """Load new items at index.
1380
1381 - queue_id: id of the queue to process this request.
1382 - queue_items: a list of QueueItems
1383 - insert_at_index: insert the item(s) at this index
1384 - keep_remaining: keep the remaining items after the insert
1385 - shuffle: (re)shuffle the items after insert index
1386 """
1387 prev_items = self._queue_items[queue_id][:insert_at_index] if keep_played else []
1388 next_items = queue_items
1389
1390 # if keep_remaining, append the old 'next' items
1391 if keep_remaining:
1392 next_items += self._queue_items[queue_id][insert_at_index:]
1393
1394 # we set the original insert order as attribute so we can un-shuffle
1395 for index, item in enumerate(next_items):
1396 item.sort_index += insert_at_index + index
1397 # (re)shuffle the final batch if needed
1398 if shuffle:
1399 next_items = await _smart_shuffle(next_items)
1400 self.update_items(queue_id, prev_items + next_items)
1401
1402 def update_items(self, queue_id: str, queue_items: list[QueueItem]) -> None:
1403 """Update the existing queue items, mostly caused by reordering."""
1404 self._queue_items[queue_id] = queue_items
1405 queue = self._queues[queue_id]
1406 queue.items = len(self._queue_items[queue_id])
1407 # to track if the queue items changed we set a timestamp
1408 # this is a simple way to detect changes in the list of items
1409 # without having to compare the entire list
1410 queue.items_last_updated = time.time()
1411 self.signal_update(queue_id, True)
1412 if (
1413 queue.state == PlaybackState.PLAYING
1414 and queue.index_in_buffer is not None
1415 and queue.index_in_buffer == queue.current_index
1416 ):
1417 # if the queue is playing,
1418 # ensure to (re)queue the next track because it might have changed
1419 # note that we only do this if the player has loaded the current track
1420 # if not, we wait until it has loaded to prevent conflicts
1421 if next_item := self.get_next_item(queue_id, queue.index_in_buffer):
1422 self._enqueue_next_item(queue_id, next_item)
1423
1424 # Helper methods
1425
1426 def get_item(self, queue_id: str, item_id_or_index: int | str | None) -> QueueItem | None:
1427 """Get queue item by index or item_id."""
1428 if item_id_or_index is None:
1429 return None
1430 if (queue_items := self._queue_items.get(queue_id)) is None:
1431 return None
1432 if isinstance(item_id_or_index, int) and len(queue_items) > item_id_or_index:
1433 return queue_items[item_id_or_index]
1434 if isinstance(item_id_or_index, str):
1435 return next((x for x in queue_items if x.queue_item_id == item_id_or_index), None)
1436 return None
1437
1438 def signal_update(self, queue_id: str, items_changed: bool = False) -> None:
1439 """Signal state changed of given queue."""
1440 queue = self._queues[queue_id]
1441 if items_changed:
1442 self.mass.signal_event(EventType.QUEUE_ITEMS_UPDATED, object_id=queue_id, data=queue)
1443 # save items in cache - only cache items with valid media_item
1444 cache_data = [
1445 x.to_cache() for x in self._queue_items[queue_id] if x.media_item is not None
1446 ]
1447 self.mass.create_task(
1448 self.mass.cache.set(
1449 key=queue_id,
1450 data=cache_data,
1451 provider=self.domain,
1452 category=CACHE_CATEGORY_PLAYER_QUEUE_ITEMS,
1453 )
1454 )
1455 # always send the base event
1456 self.mass.signal_event(EventType.QUEUE_UPDATED, object_id=queue_id, data=queue)
1457 # also signal update to the player itself so it can update its current_media
1458 self.mass.players.trigger_player_update(queue_id)
1459 # save state
1460 self.mass.create_task(
1461 self.mass.cache.set(
1462 key=queue_id,
1463 data=queue.to_cache(),
1464 provider=self.domain,
1465 category=CACHE_CATEGORY_PLAYER_QUEUE_STATE,
1466 )
1467 )
1468
1469 def index_by_id(self, queue_id: str, queue_item_id: str) -> int | None:
1470 """Get index by queue_item_id."""
1471 queue_items = self._queue_items[queue_id]
1472 for index, item in enumerate(queue_items):
1473 if item.queue_item_id == queue_item_id:
1474 return index
1475 return None
1476
1477 async def player_media_from_queue_item(
1478 self, queue_item: QueueItem, flow_mode: bool
1479 ) -> PlayerMedia:
1480 """Parse PlayerMedia from QueueItem."""
1481 queue = self._queues[queue_item.queue_id]
1482 if flow_mode:
1483 duration = None
1484 elif queue_item.streamdetails:
1485 # prefer netto duration
1486 # when seeking, the player only receives the remaining duration
1487 duration = queue_item.streamdetails.duration or queue_item.duration
1488 if duration and queue_item.streamdetails.seek_position:
1489 duration = duration - queue_item.streamdetails.seek_position
1490 else:
1491 duration = queue_item.duration
1492 if queue.session_id is None:
1493 # handle error or return early
1494 raise InvalidDataError("Queue session_id is None")
1495 media = PlayerMedia(
1496 uri=await self.mass.streams.resolve_stream_url(
1497 queue.session_id, queue_item, flow_mode=flow_mode
1498 ),
1499 media_type=MediaType.FLOW_STREAM if flow_mode else queue_item.media_type,
1500 title="Music Assistant" if flow_mode else queue_item.name,
1501 image_url=MASS_LOGO_ONLINE,
1502 duration=duration,
1503 source_id=queue_item.queue_id,
1504 queue_item_id=queue_item.queue_item_id,
1505 )
1506 if not flow_mode and queue_item.media_item:
1507 media.title = queue_item.media_item.name
1508 media.artist = getattr(queue_item.media_item, "artist_str", "")
1509 media.album = (
1510 album.name if (album := getattr(queue_item.media_item, "album", None)) else ""
1511 )
1512 if queue_item.image:
1513 # the image format needs to be 500x500 jpeg for maximum compatibility with players
1514 # we prefer the imageproxy on the streamserver here because this request is sent
1515 # to the player itself which may not be able to reach the regular webserver
1516 media.image_url = self.mass.metadata.get_image_url(
1517 queue_item.image, size=500, prefer_stream_server=True
1518 )
1519 return media
1520
1521 async def get_artist_tracks(self, artist: Artist) -> list[Track]:
1522 """Return tracks for given artist, based on user preference."""
1523 artist_items_conf = self.mass.config.get_raw_core_config_value(
1524 self.domain,
1525 CONF_DEFAULT_ENQUEUE_SELECT_ARTIST,
1526 ENQUEUE_SELECT_ARTIST_DEFAULT_VALUE,
1527 )
1528 self.logger.info(
1529 "Fetching tracks to play for artist %s",
1530 artist.name,
1531 )
1532 if artist_items_conf in ("library_tracks", "all_tracks"):
1533 all_items = await self.mass.music.artists.tracks(
1534 artist.item_id,
1535 artist.provider,
1536 in_library_only=artist_items_conf == "library_tracks",
1537 )
1538 random.shuffle(all_items)
1539 return all_items
1540 if artist_items_conf in ("library_album_tracks", "all_album_tracks"):
1541 all_tracks: list[Track] = []
1542 for library_album in await self.mass.music.artists.albums(
1543 artist.item_id,
1544 artist.provider,
1545 in_library_only=artist_items_conf == "library_album_tracks",
1546 ):
1547 for album_track in await self.mass.music.albums.tracks(
1548 library_album.item_id, library_album.provider
1549 ):
1550 if album_track not in all_tracks:
1551 all_tracks.append(album_track)
1552 random.shuffle(all_tracks)
1553 return all_tracks
1554 return []
1555
1556 async def get_album_tracks(self, album: Album, start_item: str | None) -> list[Track]:
1557 """Return tracks for given album, based on user preference."""
1558 album_items_conf = self.mass.config.get_raw_core_config_value(
1559 self.domain,
1560 CONF_DEFAULT_ENQUEUE_SELECT_ALBUM,
1561 ENQUEUE_SELECT_ALBUM_DEFAULT_VALUE,
1562 )
1563 result: list[Track] = []
1564 start_item_found = False
1565 self.logger.info(
1566 "Fetching tracks to play for album %s",
1567 album.name,
1568 )
1569 for album_track in await self.mass.music.albums.tracks(
1570 item_id=album.item_id,
1571 provider_instance_id_or_domain=album.provider,
1572 in_library_only=album_items_conf == "library_tracks",
1573 ):
1574 if not album_track.available:
1575 continue
1576 if start_item in (album_track.item_id, album_track.uri):
1577 start_item_found = True
1578 if start_item is not None and not start_item_found:
1579 continue
1580 result.append(album_track)
1581 return result
1582
1583 async def get_playlist_tracks(self, playlist: Playlist, start_item: str | None) -> list[Track]:
1584 """Return tracks for given playlist, based on user preference."""
1585 result: list[Track] = []
1586 start_item_found = False
1587 self.logger.info(
1588 "Fetching tracks to play for playlist %s",
1589 playlist.name,
1590 )
1591 # TODO: Handle other sort options etc.
1592 async for playlist_track in self.mass.music.playlists.tracks(
1593 playlist.item_id, playlist.provider
1594 ):
1595 if not playlist_track.available:
1596 continue
1597 if start_item in (playlist_track.item_id, playlist_track.uri):
1598 start_item_found = True
1599 if start_item is not None and not start_item_found:
1600 continue
1601 result.append(playlist_track)
1602 return result
1603
1604 async def get_audiobook_resume_point(
1605 self, audio_book: Audiobook, chapter: str | int | None = None, userid: str | None = None
1606 ) -> int:
1607 """Return resume point (in milliseconds) for given audio book."""
1608 self.logger.debug(
1609 "Fetching resume point to play for audio book %s",
1610 audio_book.name,
1611 )
1612 if chapter is not None:
1613 # user explicitly selected a chapter to play
1614 start_chapter = int(chapter) if isinstance(chapter, str) else chapter
1615 if chapters := audio_book.metadata.chapters:
1616 if _chapter := next((x for x in chapters if x.position == start_chapter), None):
1617 return int(_chapter.start * 1000)
1618 raise InvalidDataError(
1619 f"Unable to resolve chapter to play for Audiobook {audio_book.name}"
1620 )
1621 full_played, resume_position_ms = await self.mass.music.get_resume_position(
1622 audio_book, userid=userid
1623 )
1624 return 0 if full_played else resume_position_ms
1625
1626 async def get_next_podcast_episodes(
1627 self,
1628 podcast: Podcast | None,
1629 episode: PodcastEpisode | str | None,
1630 userid: str | None = None,
1631 ) -> UniqueList[PodcastEpisode]:
1632 """Return (next) episode(s) and resume point for given podcast."""
1633 if podcast is None and isinstance(episode, str | NoneType):
1634 raise InvalidDataError("Either podcast or episode must be provided")
1635 if podcast is None:
1636 # single podcast episode requested
1637 assert isinstance(episode, PodcastEpisode) # checked above
1638 self.logger.debug(
1639 "Fetching resume point to play for Podcast episode %s",
1640 episode.name,
1641 )
1642 (
1643 fully_played,
1644 resume_position_ms,
1645 ) = await self.mass.music.get_resume_position(episode, userid=userid)
1646 episode.fully_played = fully_played
1647 episode.resume_position_ms = 0 if fully_played else resume_position_ms
1648 return UniqueList([episode])
1649 # podcast with optional start episode requested
1650 self.logger.debug(
1651 "Fetching episode(s) and resume point to play for Podcast %s",
1652 podcast.name,
1653 )
1654 all_episodes = [
1655 x async for x in self.mass.music.podcasts.episodes(podcast.item_id, podcast.provider)
1656 ]
1657 all_episodes.sort(key=lambda x: x.position)
1658 # if a episode was provided, a user explicitly selected a episode to play
1659 # so we need to find the index of the episode in the list
1660 resolved_episode: PodcastEpisode | None = None
1661 if isinstance(episode, PodcastEpisode):
1662 resolved_episode = next((x for x in all_episodes if x.uri == episode.uri), None)
1663 if resolved_episode:
1664 # ensure we have accurate resume info
1665 (
1666 fully_played,
1667 resume_position_ms,
1668 ) = await self.mass.music.get_resume_position(resolved_episode, userid=userid)
1669 resolved_episode.resume_position_ms = 0 if fully_played else resume_position_ms
1670 elif isinstance(episode, str):
1671 resolved_episode = next(
1672 (x for x in all_episodes if episode in (x.uri, x.item_id)), None
1673 )
1674 if resolved_episode:
1675 # ensure we have accurate resume info
1676 (
1677 fully_played,
1678 resume_position_ms,
1679 ) = await self.mass.music.get_resume_position(resolved_episode, userid=userid)
1680 resolved_episode.resume_position_ms = 0 if fully_played else resume_position_ms
1681 else:
1682 # get first episode that is not fully played
1683 for ep in all_episodes:
1684 if ep.fully_played:
1685 continue
1686 # ensure we have accurate resume info
1687 (
1688 fully_played,
1689 resume_position_ms,
1690 ) = await self.mass.music.get_resume_position(ep, userid=userid)
1691 if fully_played:
1692 continue
1693 ep.resume_position_ms = resume_position_ms
1694 resolved_episode = ep
1695 break
1696 else:
1697 # no episodes found that are not fully played, so we start at the beginning
1698 resolved_episode = next((x for x in all_episodes), None)
1699 if resolved_episode is None:
1700 raise InvalidDataError(f"Unable to resolve episode to play for Podcast {podcast.name}")
1701 # get the index of the episode
1702 episode_index = all_episodes.index(resolved_episode)
1703 # return the (remaining) episode(s) to play
1704 return UniqueList(all_episodes[episode_index:])
1705
1706 def _get_next_index(
1707 self,
1708 queue_id: str,
1709 cur_index: int | None,
1710 is_skip: bool = False,
1711 allow_repeat: bool = True,
1712 ) -> int | None:
1713 """
1714 Return the next index for the queue, accounting for repeat settings.
1715
1716 Will return None if there are no (more) items in the queue.
1717 """
1718 queue = self._queues[queue_id]
1719 queue_items = self._queue_items[queue_id]
1720 if not queue_items or cur_index is None:
1721 # queue is empty
1722 return None
1723 # handle repeat single track
1724 if queue.repeat_mode == RepeatMode.ONE and not is_skip:
1725 return cur_index if allow_repeat else None
1726 # handle cur_index is last index of the queue
1727 if cur_index >= (len(queue_items) - 1):
1728 if allow_repeat and queue.repeat_mode == RepeatMode.ALL:
1729 # if repeat all is enabled, we simply start again from the beginning
1730 return 0
1731 return None
1732 # all other: just the next index
1733 return cur_index + 1
1734
1735 def get_next_item(self, queue_id: str, cur_index: int | str) -> QueueItem | None:
1736 """Return next QueueItem for given queue."""
1737 index: int
1738 if isinstance(cur_index, str):
1739 resolved_index = self.index_by_id(queue_id, cur_index)
1740 if resolved_index is None:
1741 return None # guard
1742 index = resolved_index
1743 else:
1744 index = cur_index
1745 # At this point index is guaranteed to be int
1746 for skip in range(5):
1747 if (next_index := self._get_next_index(queue_id, index + skip)) is None:
1748 break
1749 next_item = self.get_item(queue_id, next_index)
1750 if next_item is None:
1751 continue
1752 if not next_item.available:
1753 # ensure that we skip unavailable items (set by load_next track logic)
1754 continue
1755 return next_item
1756 return None
1757
1758 async def _fill_radio_tracks(self, queue_id: str) -> None:
1759 """Fill a Queue with (additional) Radio tracks."""
1760 self.logger.debug(
1761 "Filling radio tracks for queue %s",
1762 queue_id,
1763 )
1764 tracks = await self._get_radio_tracks(queue_id=queue_id, is_initial_radio_mode=False)
1765 # fill queue - filter out unavailable items
1766 queue_items = [QueueItem.from_media_item(queue_id, x) for x in tracks if x.available]
1767 await self.load(
1768 queue_id,
1769 queue_items,
1770 insert_at_index=len(self._queue_items[queue_id]) + 1,
1771 )
1772
1773 def _enqueue_next_item(self, queue_id: str, next_item: QueueItem | None) -> None:
1774 """Enqueue the next item on the player."""
1775 if not next_item:
1776 # no next item, nothing to do...
1777 return
1778
1779 queue = self._queues[queue_id]
1780 if queue.flow_mode:
1781 # ignore this for flow mode
1782 return
1783
1784 async def _enqueue_next_item_on_player(next_item: QueueItem) -> None:
1785 await self.mass.players.enqueue_next_media(
1786 player_id=queue_id,
1787 media=await self.player_media_from_queue_item(next_item, False),
1788 )
1789 if queue.next_item_id_enqueued != next_item.queue_item_id:
1790 queue.next_item_id_enqueued = next_item.queue_item_id
1791 self.logger.debug(
1792 "Enqueued next track %s on queue %s",
1793 next_item.name,
1794 self._queues[queue_id].display_name,
1795 )
1796
1797 task_id = f"enqueue_next_item_{queue_id}"
1798 self.mass.call_later(0.5, _enqueue_next_item_on_player, next_item, task_id=task_id)
1799
1800 def _preload_next_item(self, queue_id: str, item_id_in_buffer: str) -> None:
1801 """
1802 Preload the streamdetails for the next item in the queue/buffer.
1803
1804 This basically ensures the item is playable and fetches the stream details.
1805 If an error occurs, the item will be skipped and the next item will be loaded.
1806 """
1807 queue = self._queues[queue_id]
1808
1809 async def _preload_streamdetails(item_id_in_buffer: str) -> None:
1810 try:
1811 # wait for the item that was loaded in the buffer is the actually playing item
1812 # this prevents a race condition when we preload the next item too soon
1813 # while the player is actually preloading the previously enqueued item.
1814 retries = 120
1815 while retries > 0:
1816 if not queue.current_item:
1817 return # guard
1818 if queue.current_item.queue_item_id == item_id_in_buffer:
1819 break
1820 retries -= 1
1821 await asyncio.sleep(1)
1822 if next_item := await self.load_next_queue_item(queue_id, item_id_in_buffer):
1823 self.logger.debug(
1824 "Preloaded next item %s for queue %s",
1825 next_item.name,
1826 queue.display_name,
1827 )
1828 # enqueue the next item on the player
1829 self._enqueue_next_item(queue_id, next_item)
1830
1831 except QueueEmpty:
1832 return
1833
1834 if not (current_item := self.get_item(queue_id, item_id_in_buffer)):
1835 # this should not happen, but guard anyways
1836 return
1837 if current_item.media_type == MediaType.RADIO or not current_item.duration:
1838 # radio items or no duration, nothing to do
1839 return
1840
1841 task_id = f"preload_next_item_{queue_id}"
1842 self.mass.create_task(
1843 _preload_streamdetails,
1844 item_id_in_buffer,
1845 task_id=task_id,
1846 abort_existing=True,
1847 )
1848
1849 async def _resolve_media_items(
1850 self,
1851 media_item: MediaItemType | ItemMapping | BrowseFolder,
1852 start_item: str | None = None,
1853 userid: str | None = None,
1854 queue_id: str | None = None,
1855 ) -> list[MediaItemType]:
1856 """Resolve/unwrap media items to enqueue."""
1857 # resolve Itemmapping to full media item
1858 if isinstance(media_item, ItemMapping):
1859 if media_item.uri is None:
1860 raise InvalidDataError("ItemMapping has no URI")
1861 media_item = await self.mass.music.get_item_by_uri(media_item.uri)
1862 if media_item.media_type == MediaType.PLAYLIST:
1863 media_item = cast("Playlist", media_item)
1864 self.mass.create_task(
1865 self.mass.music.mark_item_played(
1866 media_item, userid=userid, queue_id=queue_id, user_initiated=True
1867 )
1868 )
1869 return list(await self.get_playlist_tracks(media_item, start_item))
1870 if media_item.media_type == MediaType.ARTIST:
1871 media_item = cast("Artist", media_item)
1872 self.mass.create_task(
1873 self.mass.music.mark_item_played(media_item, queue_id=queue_id, user_initiated=True)
1874 )
1875 return list(await self.get_artist_tracks(media_item))
1876 if media_item.media_type == MediaType.ALBUM:
1877 media_item = cast("Album", media_item)
1878 self.mass.create_task(
1879 self.mass.music.mark_item_played(
1880 media_item, userid=userid, queue_id=queue_id, user_initiated=True
1881 )
1882 )
1883 return list(await self.get_album_tracks(media_item, start_item))
1884 if media_item.media_type == MediaType.AUDIOBOOK:
1885 media_item = cast("Audiobook", media_item)
1886 # ensure we grab the correct/latest resume point info
1887 media_item.resume_position_ms = await self.get_audiobook_resume_point(
1888 media_item, start_item, userid=userid
1889 )
1890 return [media_item]
1891 if media_item.media_type == MediaType.PODCAST:
1892 media_item = cast("Podcast", media_item)
1893 self.mass.create_task(
1894 self.mass.music.mark_item_played(
1895 media_item, userid=userid, queue_id=queue_id, user_initiated=True
1896 )
1897 )
1898 return list(await self.get_next_podcast_episodes(media_item, start_item, userid=userid))
1899 if media_item.media_type == MediaType.PODCAST_EPISODE:
1900 media_item = cast("PodcastEpisode", media_item)
1901 return list(await self.get_next_podcast_episodes(None, media_item, userid=userid))
1902 if media_item.media_type == MediaType.FOLDER:
1903 media_item = cast("BrowseFolder", media_item)
1904 return list(await self._get_folder_tracks(media_item))
1905 # all other: single track or radio item
1906 return [cast("MediaItemType", media_item)]
1907
1908 async def _try_resume_from_playlog(self, queue: PlayerQueue) -> bool:
1909 """Try to resume playback from playlog when queue is empty.
1910
1911 Attempts to find user-initiated recently played items in the following order:
1912 1. By userid AND queue_id
1913 2. By queue_id only
1914 3. By userid only (if available)
1915 4. Any recently played item
1916
1917 :param queue: The queue to resume playback on.
1918 :return: True if playback was started, False otherwise.
1919 """
1920 # Try different filter combinations in order of specificity
1921 filter_attempts: list[tuple[str | None, str | None, str]] = []
1922 if queue.userid:
1923 filter_attempts.append((queue.userid, queue.queue_id, "userid + queue_id match"))
1924 filter_attempts.append((None, queue.queue_id, "queue_id match"))
1925 if queue.userid:
1926 filter_attempts.append((queue.userid, None, "userid match"))
1927 filter_attempts.append((None, None, "any recent item"))
1928
1929 for userid, queue_id, match_type in filter_attempts:
1930 items = await self.mass.music.recently_played(
1931 limit=5,
1932 fully_played_only=False,
1933 user_initiated_only=True,
1934 userid=userid,
1935 queue_id=queue_id,
1936 )
1937 for item in items:
1938 if not item.uri:
1939 continue
1940 try:
1941 await self.play_media(queue.queue_id, item)
1942 self.logger.info(
1943 "Resumed queue %s from playlog (%s)", queue.display_name, match_type
1944 )
1945 return True
1946 except MusicAssistantError as err:
1947 self.logger.debug("Failed to resume with item %s: %s", item.name, err)
1948 continue
1949
1950 return False
1951
1952 async def _get_radio_tracks(
1953 self, queue_id: str, is_initial_radio_mode: bool = False
1954 ) -> list[Track]:
1955 """Call the registered music providers for dynamic tracks."""
1956 queue = self._queues[queue_id]
1957 queue_track_items: list[Track] = [
1958 q.media_item
1959 for q in self._queue_items[queue_id]
1960 if q.media_item and isinstance(q.media_item, Track)
1961 ]
1962 if not queue.radio_source:
1963 # this may happen during race conditions as this method is called delayed
1964 return []
1965 self.logger.info(
1966 "Fetching radio tracks for queue %s based on: %s",
1967 queue.display_name,
1968 ", ".join([x.name for x in queue.radio_source]),
1969 )
1970
1971 # Get user's preferred provider instances for steering provider selection
1972 preferred_provider_instances: list[str] | None = None
1973 if (
1974 queue.userid
1975 and (playback_user := await self.mass.webserver.auth.get_user(queue.userid))
1976 and playback_user.provider_filter
1977 ):
1978 preferred_provider_instances = playback_user.provider_filter
1979
1980 available_base_tracks: list[Track] = []
1981 base_track_sample_size = 5
1982 # Some providers have very deterministic similar track algorithms when providing
1983 # a single track item. When we have a radio mode based on 1 track and we have to
1984 # refill the queue (ie not initial radio mode), we use the play history as base tracks
1985 if (
1986 len(queue.radio_source) == 1
1987 and queue.radio_source[0].media_type == MediaType.TRACK
1988 and not is_initial_radio_mode
1989 ):
1990 available_base_tracks = queue_track_items
1991 else:
1992 # Grab all the available base tracks based on the selected source items.
1993 # shuffle the source items, just in case
1994 for radio_item in random.sample(queue.radio_source, len(queue.radio_source)):
1995 ctrl = self.mass.music.get_controller(radio_item.media_type)
1996 try:
1997 available_base_tracks += [
1998 track
1999 for track in await ctrl.radio_mode_base_tracks(
2000 radio_item, # type: ignore[arg-type]
2001 preferred_provider_instances,
2002 )
2003 # Avoid duplicate base tracks
2004 if track not in available_base_tracks
2005 ]
2006 except UnsupportedFeaturedException as err:
2007 self.logger.debug(
2008 "Skip loading radio items for %s: %s ",
2009 radio_item.uri,
2010 str(err),
2011 )
2012 if not available_base_tracks:
2013 raise UnsupportedFeaturedException("Radio mode not available for source items")
2014
2015 # Sample tracks from the base tracks, which will be used to calculate the dynamic ones
2016 base_tracks = random.sample(
2017 available_base_tracks,
2018 min(base_track_sample_size, len(available_base_tracks)),
2019 )
2020 # Use a set to avoid duplicate dynamic tracks
2021 dynamic_tracks: set[Track] = set()
2022 # Use base tracks + Trackcontroller to obtain similar tracks for every base Track
2023 for allow_lookup in (False, True):
2024 if dynamic_tracks:
2025 break
2026 for base_track in base_tracks:
2027 try:
2028 _similar_tracks = await self.mass.music.tracks.similar_tracks(
2029 base_track.item_id,
2030 base_track.provider,
2031 allow_lookup=allow_lookup,
2032 preferred_provider_instances=preferred_provider_instances,
2033 )
2034 except MediaNotFoundError:
2035 # Some providers don't have similar tracks for all items. For example,
2036 # Tidal can sometimes return a 404 when the 'similar_tracks' endpoint is called.
2037 # in that case, just skip the track.
2038 self.logger.debug("Similar tracks not found for track %s", base_track.name)
2039 continue
2040 for track in _similar_tracks:
2041 if (
2042 track not in base_tracks
2043 # Exclude tracks we have already played / queued
2044 and track not in queue_track_items
2045 # Ignore tracks that are too long for radio mode, e.g. mixes
2046 and track.duration <= RADIO_TRACK_MAX_DURATION_SECS
2047 ):
2048 dynamic_tracks.add(track)
2049 if len(dynamic_tracks) >= 50:
2050 break
2051 queue_tracks: list[Track] = []
2052 dynamic_tracks_list = list(dynamic_tracks)
2053 # Only include the sampled base tracks when the radio mode is first initialized
2054 if is_initial_radio_mode:
2055 queue_tracks += [base_tracks[0]]
2056 # Exhaust base tracks with the pattern of BDDBDDBDD (1 base track + 2 dynamic tracks)
2057 if len(base_tracks) > 1:
2058 for base_track in base_tracks[1:]:
2059 queue_tracks += [base_track]
2060 if len(dynamic_tracks_list) > 2:
2061 queue_tracks += random.sample(dynamic_tracks_list, 2)
2062 else:
2063 queue_tracks += dynamic_tracks_list
2064 # Add dynamic tracks to the queue, make sure to exclude already picked tracks
2065 remaining_dynamic_tracks = [t for t in dynamic_tracks_list if t not in queue_tracks]
2066 if remaining_dynamic_tracks:
2067 queue_tracks += random.sample(
2068 remaining_dynamic_tracks, min(len(remaining_dynamic_tracks), 25)
2069 )
2070 return queue_tracks
2071
2072 async def _get_folder_tracks(self, folder: BrowseFolder) -> list[Track]:
2073 """Fetch (playable) tracks for given browse folder."""
2074 self.logger.info(
2075 "Fetching tracks to play for folder %s",
2076 folder.name,
2077 )
2078 tracks: list[Track] = []
2079 for item in await self.mass.music.browse(folder.path):
2080 if not item.is_playable:
2081 continue
2082 # recursively fetch tracks from all media types
2083 resolved = await self._resolve_media_items(item)
2084 tracks += [x for x in resolved if isinstance(x, Track)]
2085
2086 return tracks
2087
2088 def _update_queue_from_player(
2089 self,
2090 player: Player,
2091 ) -> None:
2092 """Update the Queue when the player state changed."""
2093 queue_id = player.player_id
2094 queue = self._queues[queue_id]
2095
2096 # basic properties
2097 queue.display_name = player.display_name
2098 queue.available = player.available
2099 queue.items = len(self._queue_items[queue_id])
2100
2101 queue.state = (
2102 player.playback_state or PlaybackState.IDLE if queue.active else PlaybackState.IDLE
2103 )
2104 # update current item/index from player report
2105 if queue.active and queue.state in (
2106 PlaybackState.PLAYING,
2107 PlaybackState.PAUSED,
2108 ):
2109 # NOTE: If the queue is not playing (yet) we will not update the current index
2110 # to ensure we keep the previously known current index
2111 if queue.flow_mode:
2112 # flow mode active, the player is playing one long stream
2113 # so we need to calculate the current index and elapsed time
2114 current_index, elapsed_time = self._get_flow_queue_stream_index(queue, player)
2115 elif item_id := self._parse_player_current_item_id(queue_id, player):
2116 # normal mode, the player itself will report the current item
2117 elapsed_time = int(player.corrected_elapsed_time or 0)
2118 current_index = self.index_by_id(queue_id, item_id)
2119 else:
2120 # this may happen if the player is still transitioning between tracks
2121 # we ignore this for now and keep the current index as is
2122 return
2123
2124 # get current/next item based on current index
2125 queue.current_index = current_index
2126 queue.current_item = current_item = self.get_item(queue_id, current_index)
2127 queue.next_item = (
2128 self.get_next_item(queue_id, current_index)
2129 if current_item and current_index is not None
2130 else None
2131 )
2132
2133 # correct elapsed time when seeking
2134 if (
2135 not queue.flow_mode
2136 and current_item
2137 and current_item.streamdetails
2138 and current_item.streamdetails.seek_position
2139 ):
2140 elapsed_time += current_item.streamdetails.seek_position
2141 queue.elapsed_time = elapsed_time
2142 queue.elapsed_time_last_updated = time.time()
2143
2144 elif not queue.current_item and queue.current_index is not None:
2145 current_index = queue.current_index
2146 queue.current_item = current_item = self.get_item(queue_id, current_index)
2147 queue.next_item = (
2148 self.get_next_item(queue_id, current_index)
2149 if current_item and current_index is not None
2150 else None
2151 )
2152
2153 # This is enough to detect any changes in the DSPDetails
2154 # (so child count changed, or any output format changed)
2155 output_formats = []
2156 if output_format := player.extra_data.get("output_format"):
2157 output_formats.append(str(output_format))
2158 for child_id in player.group_members:
2159 if (child := self.mass.players.get(child_id)) and (
2160 output_format := child.extra_data.get("output_format")
2161 ):
2162 output_formats.append(str(output_format))
2163 else:
2164 output_formats.append("unknown")
2165
2166 # basic throttle: do not send state changed events if queue did not actually change
2167 prev_state: CompareState = self._prev_states.get(
2168 queue_id,
2169 CompareState(
2170 queue_id=queue_id,
2171 state=PlaybackState.IDLE,
2172 current_item_id=None,
2173 next_item_id=None,
2174 current_item=None,
2175 elapsed_time=0,
2176 last_playing_elapsed_time=0,
2177 stream_title=None,
2178 codec_type=None,
2179 output_formats=None,
2180 ),
2181 )
2182 # update last_playing_elapsed_time only when the player is actively playing
2183 # use corrected_elapsed_time which accounts for time since last update
2184 # this preserves the last known elapsed time when transitioning to idle/paused
2185 prev_playing_elapsed = prev_state["last_playing_elapsed_time"]
2186 prev_item_id = prev_state["current_item_id"]
2187 current_item_id = queue.current_item.queue_item_id if queue.current_item else None
2188 if queue.state == PlaybackState.PLAYING:
2189 current_elapsed = int(queue.corrected_elapsed_time)
2190 if current_item_id != prev_item_id:
2191 # new track started, reset the elapsed time tracker
2192 last_playing_elapsed_time = current_elapsed
2193 else:
2194 # same track, use the max of current and previous to handle timing issues
2195 last_playing_elapsed_time = max(current_elapsed, prev_playing_elapsed)
2196 else:
2197 last_playing_elapsed_time = prev_playing_elapsed
2198 new_state = CompareState(
2199 queue_id=queue_id,
2200 state=queue.state,
2201 current_item_id=queue.current_item.queue_item_id if queue.current_item else None,
2202 next_item_id=queue.next_item.queue_item_id if queue.next_item else None,
2203 current_item=queue.current_item,
2204 elapsed_time=int(queue.elapsed_time),
2205 last_playing_elapsed_time=last_playing_elapsed_time,
2206 stream_title=(
2207 queue.current_item.streamdetails.stream_title
2208 if queue.current_item and queue.current_item.streamdetails
2209 else None
2210 ),
2211 codec_type=(
2212 queue.current_item.streamdetails.audio_format.codec_type
2213 if queue.current_item and queue.current_item.streamdetails
2214 else None
2215 ),
2216 output_formats=output_formats,
2217 )
2218 changed_keys = get_changed_keys(dict(prev_state), dict(new_state))
2219 with suppress(KeyError):
2220 changed_keys.remove("next_item_id")
2221 with suppress(KeyError):
2222 changed_keys.remove("last_playing_elapsed_time")
2223
2224 # store the new state
2225 if queue.active:
2226 self._prev_states[queue_id] = new_state
2227 else:
2228 self._prev_states.pop(queue_id, None)
2229
2230 # return early if nothing changed
2231 if len(changed_keys) == 0:
2232 return
2233
2234 # signal update and store state
2235 send_update = True
2236 if changed_keys == {"elapsed_time"}:
2237 # only elapsed time changed, do not send full queue update
2238 send_update = False
2239 prev_time = prev_state.get("elapsed_time") or 0
2240 cur_time = new_state.get("elapsed_time") or 0
2241 if abs(cur_time - prev_time) > 2:
2242 # send dedicated event for time updates when seeking
2243 self.mass.signal_event(
2244 EventType.QUEUE_TIME_UPDATED,
2245 object_id=queue_id,
2246 data=queue.elapsed_time,
2247 )
2248 # also signal update to the player itself so it can update its current_media
2249 self.mass.players.trigger_player_update(queue_id)
2250
2251 if send_update:
2252 self.signal_update(queue_id)
2253
2254 if "output_formats" in changed_keys:
2255 # refresh DSP details since they may have changed
2256 dsp = get_stream_dsp_details(self.mass, queue_id)
2257 if queue.current_item and queue.current_item.streamdetails:
2258 queue.current_item.streamdetails.dsp = dsp
2259 if queue.next_item and queue.next_item.streamdetails:
2260 queue.next_item.streamdetails.dsp = dsp
2261
2262 # handle updating stream_metadata if needed
2263 if (
2264 queue.current_item
2265 and (streamdetails := queue.current_item.streamdetails)
2266 and streamdetails.stream_metadata_update_callback
2267 and (
2268 streamdetails.stream_metadata_last_updated is None
2269 or (
2270 time.time() - streamdetails.stream_metadata_last_updated
2271 >= streamdetails.stream_metadata_update_interval
2272 )
2273 )
2274 ):
2275 streamdetails.stream_metadata_last_updated = time.time()
2276 self.mass.create_task(
2277 streamdetails.stream_metadata_update_callback(
2278 streamdetails, int(queue.corrected_elapsed_time)
2279 )
2280 )
2281
2282 # handle sending a playback progress report
2283 # we do this every 30 seconds or when the state changes
2284 if (
2285 changed_keys.intersection({"state", "current_item_id"})
2286 or int(queue.elapsed_time) % 30 == 0
2287 ):
2288 self._handle_playback_progress_report(queue, prev_state, new_state)
2289
2290 # check if we need to clear the queue if we reached the end
2291 if "state" in changed_keys and queue.state == PlaybackState.IDLE:
2292 self._handle_end_of_queue(queue, prev_state, new_state)
2293
2294 # watch dynamic radio items refill if needed
2295 if "current_item_id" in changed_keys:
2296 # auto enable radio mode if dont stop the music is enabled
2297 if (
2298 queue.dont_stop_the_music_enabled
2299 and queue.enqueued_media_items
2300 and queue.current_index is not None
2301 and (queue.items - queue.current_index) <= 1
2302 ):
2303 # We have received the last item in the queue and Don't stop the music is enabled
2304 # set the played media item(s) as radio items (which will refill the queue)
2305 # note that this will fail if there are no media items for which we have
2306 # a dynamic radio source.
2307 self.logger.debug(
2308 "End of queue detected and Don't stop the music is enabled for %s"
2309 " - setting enqueued media items as radio source: %s",
2310 queue.display_name,
2311 ", ".join([x.uri for x in queue.enqueued_media_items]), # type: ignore[misc] # uri set in __post_init__
2312 )
2313 queue.radio_source = queue.enqueued_media_items
2314 # auto fill radio tracks if less than 5 tracks left in the queue
2315 if (
2316 queue.radio_source
2317 and queue.current_index is not None
2318 and (queue.items - queue.current_index) < 5
2319 ):
2320 task_id = f"fill_radio_tracks_{queue_id}"
2321 self.mass.call_later(5, self._fill_radio_tracks, queue_id, task_id=task_id)
2322
2323 def _get_flow_queue_stream_index(
2324 self, queue: PlayerQueue, player: Player
2325 ) -> tuple[int | None, int]:
2326 """Calculate current queue index and current track elapsed time when flow mode is active."""
2327 elapsed_time_queue_total = player.corrected_elapsed_time or 0
2328 if queue.current_index is None and not queue.flow_mode_stream_log:
2329 return queue.current_index, int(queue.elapsed_time)
2330
2331 # For each track that has been streamed/buffered to the player,
2332 # a playlog entry will be created with the queue item id
2333 # and the amount of seconds streamed. We traverse the playlog to figure
2334 # out where we are in the queue, accounting for actual streamed
2335 # seconds (and not duration) and skipped seconds. If a track has been repeated,
2336 # it will simply be in the playlog multiple times.
2337 played_time = 0.0
2338 queue_index: int | None = queue.current_index or 0
2339 track_time = 0.0
2340 for play_log_entry in queue.flow_mode_stream_log:
2341 queue_item_duration = (
2342 # NOTE: 'seconds_streamed' can actually be 0 if there was a stream error!
2343 play_log_entry.seconds_streamed
2344 if play_log_entry.seconds_streamed is not None
2345 else play_log_entry.duration or 3600 * 24 * 7
2346 )
2347 if elapsed_time_queue_total > (queue_item_duration + played_time):
2348 # total elapsed time is more than (streamed) track duration
2349 # this track has been fully played, move on.
2350 played_time += queue_item_duration
2351 else:
2352 # no more seconds left to divide, this is our track
2353 # account for any seeking by adding the skipped/seeked seconds
2354 queue_index = self.index_by_id(queue.queue_id, play_log_entry.queue_item_id)
2355 queue_item = self.get_item(queue.queue_id, queue_index)
2356 if queue_item and queue_item.streamdetails:
2357 track_sec_skipped = queue_item.streamdetails.seek_position
2358 else:
2359 track_sec_skipped = 0
2360 track_time = elapsed_time_queue_total + track_sec_skipped - played_time
2361 break
2362 if player.playback_state != PlaybackState.PLAYING:
2363 # if the player is not playing, we can't be sure that the elapsed time is correct
2364 # so we just return the queue index and the elapsed time
2365 return queue.current_index, int(queue.elapsed_time)
2366 return queue_index, int(track_time)
2367
2368 def _parse_player_current_item_id(self, queue_id: str, player: Player) -> str | None:
2369 """Parse QueueItem ID from Player's current url."""
2370 if not player._current_media:
2371 # YES, we use player._current_media on purpose here because we need the raw metadata
2372 return None
2373 # prefer queue_id and queue_item_id within the current media
2374 if player._current_media.source_id == queue_id and player._current_media.queue_item_id:
2375 return player._current_media.queue_item_id
2376 # special case for sonos players
2377 if player._current_media.uri and player._current_media.uri.startswith(f"mass:{queue_id}"):
2378 if player._current_media.queue_item_id:
2379 return player._current_media.queue_item_id
2380 return player._current_media.uri.split(":")[-1]
2381 # try to extract the item id from a mass stream url
2382 if (
2383 player._current_media.uri
2384 and queue_id in player._current_media.uri
2385 and self.mass.streams.base_url in player._current_media.uri
2386 ):
2387 current_item_id = player._current_media.uri.rsplit("/")[-1].split(".")[0]
2388 if self.get_item(queue_id, current_item_id):
2389 return current_item_id
2390 # try to extract the item id from a queue_id/item_id combi
2391 if (
2392 player._current_media.uri
2393 and queue_id in player._current_media.uri
2394 and "/" in player._current_media.uri
2395 ):
2396 current_item_id = player._current_media.uri.split("/")[1]
2397 if self.get_item(queue_id, current_item_id):
2398 return current_item_id
2399
2400 return None
2401
2402 def _handle_end_of_queue(
2403 self, queue: PlayerQueue, prev_state: CompareState, new_state: CompareState
2404 ) -> None:
2405 """Check if the queue should be cleared after the current item."""
2406 # check if queue state changed to stopped (from playing/paused to idle)
2407 if not (
2408 prev_state["state"] in (PlaybackState.PLAYING, PlaybackState.PAUSED)
2409 and new_state["state"] == PlaybackState.IDLE
2410 ):
2411 return
2412 # check if no more items in the queue (next_item should be None at end of queue)
2413 if queue.next_item is not None:
2414 return
2415 # check if we had a previous item playing
2416 if prev_state["current_item_id"] is None:
2417 return
2418
2419 async def _clear_queue_delayed() -> None:
2420 for _ in range(5):
2421 await asyncio.sleep(1)
2422 if queue.state != PlaybackState.IDLE:
2423 return
2424 if queue.next_item is not None:
2425 return
2426 self.logger.info("End of queue reached, clearing items")
2427 self.clear(queue.queue_id)
2428
2429 # all checks passed, we stopped playback at the last (or single) track of the queue
2430 # now determine if the item was fully played before clearing
2431
2432 # For flow mode, check if the last track was fully streamed using the stream log
2433 # This is more reliable than elapsed_time which can be reset/incorrect
2434 if queue.flow_mode and queue.flow_mode_stream_log:
2435 last_log_entry = queue.flow_mode_stream_log[-1]
2436 if last_log_entry.seconds_streamed is not None:
2437 # The last track finished streaming, safe to clear queue
2438 self.mass.create_task(_clear_queue_delayed())
2439 return
2440
2441 # For non-flow mode, use prev_state values since queue state may have been updated/reset
2442 prev_item = prev_state["current_item"]
2443 if prev_item and (streamdetails := prev_item.streamdetails):
2444 duration = streamdetails.duration or prev_item.duration or 24 * 3600
2445 elif prev_item:
2446 duration = prev_item.duration or 24 * 3600
2447 else:
2448 # No current item means player has already cleared it, safe to clear queue
2449 self.mass.create_task(_clear_queue_delayed())
2450 return
2451
2452 # use last_playing_elapsed_time which preserves the elapsed time from when the player
2453 # was still playing (before transitioning to idle where elapsed_time may be reset to 0)
2454 seconds_played = int(prev_state["last_playing_elapsed_time"])
2455 # debounce this a bit to make sure we're not clearing the queue by accident
2456 # only clear if the last track was played to near completion (within 5 seconds of end)
2457 if seconds_played >= (duration or 3600) - 5:
2458 self.mass.create_task(_clear_queue_delayed())
2459
2460 def _handle_playback_progress_report(
2461 self, queue: PlayerQueue, prev_state: CompareState, new_state: CompareState
2462 ) -> None:
2463 """Handle playback progress report."""
2464 # detect change in current index to report that a item has been played
2465 prev_item_id = prev_state["current_item_id"]
2466 cur_item_id = new_state["current_item_id"]
2467 if prev_item_id is None and cur_item_id is None:
2468 return
2469
2470 if prev_item_id is not None and prev_item_id != cur_item_id:
2471 # we have a new item, so we need report the previous one
2472 is_current_item = False
2473 item_to_report = prev_state["current_item"]
2474 seconds_played = int(prev_state["elapsed_time"])
2475 else:
2476 # report on current item
2477 is_current_item = True
2478 item_to_report = self.get_item(queue.queue_id, cur_item_id) or new_state["current_item"]
2479 seconds_played = int(new_state["elapsed_time"])
2480
2481 if not item_to_report:
2482 return # guard against invalid items
2483
2484 if not (media_item := item_to_report.media_item):
2485 # only report on media items
2486 return
2487 assert media_item.uri is not None # uri is set in __post_init__
2488
2489 if item_to_report.streamdetails and item_to_report.streamdetails.stream_error:
2490 # Ignore items that had a stream error
2491 return
2492
2493 if item_to_report.streamdetails and item_to_report.streamdetails.duration:
2494 duration = int(item_to_report.streamdetails.duration)
2495 else:
2496 duration = int(item_to_report.duration or 3 * 3600)
2497
2498 if seconds_played < 5:
2499 # ignore items that have been played less than 5 seconds
2500 # this also filters out a bounce effect where the previous item
2501 # gets reported with 0 elapsed seconds after a new item starts playing
2502 return
2503
2504 # determine if item is fully played
2505 # for podcasts and audiobooks we account for the last 60 seconds
2506 percentage_played = percentage(seconds_played, duration)
2507 if not is_current_item and item_to_report.media_type in (
2508 MediaType.AUDIOBOOK,
2509 MediaType.PODCAST_EPISODE,
2510 ):
2511 fully_played = seconds_played >= duration - 60
2512 elif not is_current_item:
2513 # 90% of the track must be played to be considered fully played
2514 fully_played = percentage_played >= 90
2515 else:
2516 fully_played = seconds_played >= duration - 10
2517
2518 is_playing = is_current_item and queue.state == PlaybackState.PLAYING
2519 if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
2520 self.logger.debug(
2521 "%s %s '%s' (%s) - Fully played: %s - Progress: %s (%s/%ss)",
2522 queue.display_name,
2523 "is playing" if is_playing else "played",
2524 item_to_report.name,
2525 item_to_report.uri,
2526 fully_played,
2527 f"{percentage_played}%",
2528 seconds_played,
2529 duration,
2530 )
2531 # add entry to playlog - this also handles resume of podcasts/audiobooks
2532 self.mass.create_task(
2533 self.mass.music.mark_item_played(
2534 media_item,
2535 fully_played=fully_played,
2536 seconds_played=seconds_played,
2537 is_playing=is_playing,
2538 userid=queue.userid,
2539 queue_id=queue.queue_id,
2540 user_initiated=False,
2541 )
2542 )
2543
2544 album: Album | ItemMapping | None = getattr(media_item, "album", None)
2545 # signal 'media item played' event,
2546 # which is useful for plugins that want to do scrobbling
2547 artists: list[Artist | ItemMapping] = getattr(media_item, "artists", [])
2548 artists_names = [a.name for a in artists]
2549 self.mass.signal_event(
2550 EventType.MEDIA_ITEM_PLAYED,
2551 object_id=media_item.uri,
2552 data=MediaItemPlaybackProgressReport(
2553 uri=media_item.uri,
2554 media_type=media_item.media_type,
2555 name=media_item.name,
2556 version=getattr(media_item, "version", None),
2557 artist=(
2558 getattr(media_item, "artist_str", None) or artists_names[0]
2559 if artists_names
2560 else None
2561 ),
2562 artists=artists_names,
2563 artist_mbids=[a.mbid for a in artists if a.mbid] if artists else None,
2564 album=album.name if album else None,
2565 album_mbid=album.mbid if album else None,
2566 album_artist=(album.artist_str if isinstance(album, Album) else None),
2567 album_artist_mbids=(
2568 [a.mbid for a in album.artists if a.mbid] if isinstance(album, Album) else None
2569 ),
2570 image_url=(
2571 self.mass.metadata.get_image_url(
2572 item_to_report.media_item.image, prefer_proxy=False
2573 )
2574 if item_to_report.media_item.image
2575 else None
2576 ),
2577 duration=duration,
2578 mbid=(getattr(media_item, "mbid", None)),
2579 seconds_played=seconds_played,
2580 fully_played=fully_played,
2581 is_playing=is_playing,
2582 userid=queue.userid,
2583 ),
2584 )
2585
2586
2587async def _smart_shuffle(items: list[QueueItem]) -> list[QueueItem]:
2588 """Shuffle queue items, avoiding identical tracks next to each other.
2589
2590 Best-effort approach to prevent the same track from appearing adjacent.
2591 Does a random shuffle first, then makes a limited number of passes to
2592 swap adjacent duplicates with a random item further in the list.
2593
2594 :param items: List of queue items to shuffle.
2595 """
2596 if len(items) <= 2:
2597 return random.sample(items, len(items)) if len(items) == 2 else items
2598
2599 # Start with a random shuffle
2600 shuffled = random.sample(items, len(items))
2601
2602 # Make a few passes to fix adjacent duplicates
2603 max_passes = 3
2604 for _ in range(max_passes):
2605 swapped = False
2606 for i in range(len(shuffled) - 1):
2607 if shuffled[i].name == shuffled[i + 1].name:
2608 # Found adjacent duplicate - swap with random position at least 2 away
2609 swap_candidates = [j for j in range(len(shuffled)) if abs(j - i - 1) >= 2]
2610 if swap_candidates:
2611 swap_pos = random.choice(swap_candidates)
2612 shuffled[i + 1], shuffled[swap_pos] = shuffled[swap_pos], shuffled[i + 1]
2613 swapped = True
2614 if not swapped:
2615 break
2616 # Yield to event loop between passes
2617 await asyncio.sleep(0)
2618
2619 return shuffled
2620