/
/
/
1"""
2Autoplay helper for the Player Queues controller.
3
4Autoplay is the single "keep going" switch of a queue; what it appends depends on the media
5type of the item that is ending. Music continues with a fresh batch of tracks, a podcast
6episode or audiobook with its own successor, and a live source is left alone. This helper
7resolves the configured Autoplay mode for a queue and produces the next batch of tracks for
8the library- and playlist-based modes. The similar-tracks (radio) modes reuse the controller's
9existing dynamic-radio machinery.
10"""
11
12from __future__ import annotations
13
14import random
15from contextlib import suppress
16from enum import StrEnum
17from typing import TYPE_CHECKING
18
19from music_assistant_models.enums import MediaType
20from music_assistant_models.errors import MusicAssistantError
21from music_assistant_models.media_items import Playlist, Track
22
23from music_assistant.constants import CONF_PLAYER_QUEUES, CONF_VALUE_GLOBAL
24from music_assistant.controllers.player_queues.constants import (
25 CONF_AUTOPLAY_MODE,
26 CONF_AUTOPLAY_PLAYLIST,
27)
28
29if TYPE_CHECKING:
30 from music_assistant_models.config_entries import ConfigValueType
31 from music_assistant_models.player_queue import PlayerQueue
32
33 from music_assistant.controllers.player_queues.controller import PlayerQueuesController
34
35# number of library tracks to append on each refill
36AUTOPLAY_BATCH_SIZE = 25
37# how many of the most recently enqueued items to derive a genre bias from
38GENRE_SEED_ITEM_COUNT = 3
39# media types that carry a useful genre signal for the library mix
40GENRE_SEED_MEDIA_TYPES = (MediaType.TRACK, MediaType.ALBUM, MediaType.ARTIST)
41# media types that end on their own but have a natural successor of their own: Autoplay
42# continues these with the next episode/book instead of appending music
43AUTOPLAY_SERIES_MEDIA_TYPES = (MediaType.AUDIOBOOK, MediaType.PODCAST_EPISODE)
44# media types Autoplay does not apply to: a live source has no natural end (stopping it means
45# the source stopped, not that the queue ran out) and a sound effect is a one-off
46AUTOPLAY_EXCLUDED_MEDIA_TYPES = (
47 MediaType.RADIO,
48 MediaType.AUDIO_SOURCE,
49 MediaType.SOUND_EFFECT,
50)
51
52
53class AutoplayMode(StrEnum):
54 """Enum with the available Autoplay (queue refill) strategies."""
55
56 AUTO = "auto"
57 SIMILAR = "similar"
58 LIBRARY = "library"
59 PLAYLIST = "playlist"
60
61
62AUTOPLAY_MODE_DEFAULT_VALUE = AutoplayMode.AUTO.value
63
64
65class Autoplay:
66 """Resolve the Autoplay mode and produce the next batch of tracks for a queue."""
67
68 def __init__(self, queues: PlayerQueuesController) -> None:
69 """
70 Initialize the Autoplay helper.
71
72 :param queues: The owning player queues controller.
73 """
74 self.queues = queues
75 self.mass = queues.mass
76 self.logger = queues.logger.getChild("autoplay")
77
78 def resolve_mode(self, queue_id: str) -> AutoplayMode:
79 """
80 Return the configured Autoplay mode for the given queue.
81
82 Follows the global (queue controller) mode when the per-queue value is "global".
83
84 :param queue_id: The queue to read the configured Autoplay mode for.
85 """
86 raw = self.mass.config.get_effective_player_queue_config_value(
87 queue_id, CONF_AUTOPLAY_MODE, AUTOPLAY_MODE_DEFAULT_VALUE
88 )
89 try:
90 return AutoplayMode(str(raw))
91 except ValueError:
92 return AutoplayMode.AUTO
93
94 async def get_library_tracks(self, queue: PlayerQueue, exclude: set[Track]) -> list[Track]:
95 """
96 Return a fresh batch of library tracks for the 'infinite library mix' mode.
97
98 Tracks are ordered by random/least-played and biased towards the genre(s) of what
99 was recently played, falling back to a whole-library random mix when no genre match
100 is available.
101
102 :param queue: The queue being refilled.
103 :param exclude: Tracks already present in the queue, to avoid immediate repeats.
104 """
105 candidates: list[Track] = []
106 if genre_ids := await self._collect_genre_ids(queue):
107 with suppress(MusicAssistantError):
108 candidates += await self.mass.music.tracks.library_items(
109 genre=genre_ids,
110 limit=AUTOPLAY_BATCH_SIZE * 3,
111 order_by="random_play_count",
112 summary=False,
113 )
114 # top up with a whole-library random mix when the genre selection yields too few usable
115 # tracks (gauge on the deduped result so unavailable/excluded matches don't mask a shortfall)
116 result = self._dedupe(candidates, exclude)
117 if len(result) < AUTOPLAY_BATCH_SIZE:
118 with suppress(MusicAssistantError):
119 candidates += await self.mass.music.tracks.library_items(
120 limit=AUTOPLAY_BATCH_SIZE * 3,
121 order_by="random_play_count",
122 summary=False,
123 )
124 result = self._dedupe(candidates, exclude)
125 return result
126
127 async def get_playlist_tracks(self, queue: PlayerQueue, exclude: set[Track]) -> list[Track]:
128 """
129 Return a random batch of tracks from the configured Autoplay playlist.
130
131 :param queue: The queue being refilled.
132 :param exclude: Tracks already present in the queue, to avoid immediate repeats.
133 """
134 uri = self._autoplay_playlist_uri(queue.queue_id)
135 if not uri:
136 self.logger.warning(
137 "Autoplay playlist mode is selected for %s but no playlist is configured",
138 queue.display_name,
139 )
140 return []
141 try:
142 playlist = await self.mass.music.get_item_by_uri(str(uri))
143 except MusicAssistantError as err:
144 self.logger.warning("Autoplay playlist %s is not available: %s", uri, err)
145 return []
146 if not isinstance(playlist, Playlist):
147 return []
148 tracks = [
149 track
150 for track in await self.queues.get_playlist_tracks(playlist, start_item=None)
151 if isinstance(track, Track)
152 ]
153 random.shuffle(tracks)
154 return self._dedupe(tracks, exclude)
155
156 def _autoplay_playlist_uri(self, queue_id: str) -> ConfigValueType:
157 """
158 Return the configured Autoplay playlist, taken from the level the mode resolves from.
159
160 A queue that follows the global Autoplay mode also follows the global playlist, so a
161 leftover per-queue playlist can't override the global one.
162
163 :param queue_id: The queue to read the Autoplay playlist for.
164 """
165 raw_mode = self.mass.config.get_raw_player_queue_config_value(
166 queue_id, CONF_AUTOPLAY_MODE, CONF_VALUE_GLOBAL
167 )
168 if raw_mode in (CONF_VALUE_GLOBAL, None):
169 return self.mass.config.get_raw_core_config_value(
170 CONF_PLAYER_QUEUES, CONF_AUTOPLAY_PLAYLIST
171 )
172 return self.mass.config.get_raw_player_queue_config_value(queue_id, CONF_AUTOPLAY_PLAYLIST)
173
174 async def _collect_genre_ids(self, queue: PlayerQueue) -> list[int]:
175 """Collect library genre ids from the queue's most recently enqueued items."""
176 genre_ids: set[int] = set()
177 seeds = [
178 item
179 for item in reversed(self.queues.queue_data(queue.queue_id).enqueued_media_items)
180 if item.media_type in GENRE_SEED_MEDIA_TYPES
181 ][:GENRE_SEED_ITEM_COUNT]
182 for seed in seeds:
183 library_id = await self._resolve_library_id(
184 seed.media_type, seed.item_id, seed.provider
185 )
186 if library_id is None:
187 continue
188 with suppress(MusicAssistantError):
189 for genre in await self.mass.music.genres.get_genres_for_media_item(
190 seed.media_type, library_id
191 ):
192 with suppress(ValueError, TypeError):
193 genre_ids.add(int(genre.item_id))
194 return list(genre_ids)
195
196 async def _resolve_library_id(
197 self, media_type: MediaType, item_id: str, provider: str
198 ) -> str | int | None:
199 """Resolve a media item to its library (database) id, or None when not in the library."""
200 if provider == "library":
201 return item_id
202 controller = self.mass.music.get_controller(media_type)
203 with suppress(MusicAssistantError):
204 if library_item := await controller.get_library_item_by_prov_id(item_id, provider):
205 return library_item.item_id
206 return None
207
208 def _dedupe(self, tracks: list[Track], exclude: set[Track]) -> list[Track]:
209 """Drop unavailable/duplicate/excluded tracks, capped to the batch size."""
210 result: list[Track] = []
211 seen: set[Track] = set()
212 for track in tracks:
213 if not track.available or track in exclude or track in seen:
214 continue
215 seen.add(track)
216 result.append(track)
217 if len(result) >= AUTOPLAY_BATCH_SIZE:
218 break
219 return result
220