/
/
/
1"""
2Smart shuffle helper for the Player Queues controller.
3
4Smart shuffle reorders the upcoming queue items so recently-heard music is pushed toward the back
5and same songs/artists are spread out, while honouring intentionally-duplicated items. It reads the
6configured recency windows for a queue, takes a single play-history snapshot from the shared recency
7engine, and runs the pure ``_arrange`` algorithm. Plain (non-smart) shuffle stays a pure random
8shuffle in the controller.
9
10The algorithm works in two stages:
11- recency tiering: each item is fresh / artist-recently-played / song-recently-played, where a
12 deliberately-duplicated song uses the short duplicate repeat-gap instead of the long song window;
13- within-tier interleave: each distinct song's copies get independently randomized positions in
14 evenly-spaced strata, so duplicates stay spread without repeating the same sequence, then a
15 bounded pass separates directly-adjacent same-artist items.
16"""
17
18from __future__ import annotations
19
20import random
21from collections import Counter, defaultdict
22from typing import TYPE_CHECKING
23
24from music_assistant.constants import (
25 CONF_PLAYER_QUEUES,
26 CONF_VALUE_DISABLED,
27 CONF_VALUE_ENABLED,
28)
29from music_assistant.controllers.music.recency import RecencyWindows
30from music_assistant.controllers.player_queues.constants import (
31 CONF_SMART_SHUFFLE_ARTIST_RECENCY,
32 CONF_SMART_SHUFFLE_DUPLICATE_GAP,
33 CONF_SMART_SHUFFLE_ENABLED,
34 CONF_SMART_SHUFFLE_SONG_RECENCY,
35 SMART_SHUFFLE_ARTIST_RECENCY_DEFAULT,
36 SMART_SHUFFLE_DUPLICATE_GAP_DEFAULT,
37 SMART_SHUFFLE_SONG_RECENCY_DEFAULT,
38)
39from music_assistant.controllers.player_queues.helpers import interleave_groups, space_by_artist
40
41if TYPE_CHECKING:
42 from music_assistant_models.player_queue import PlayerQueue
43 from music_assistant_models.queue_item import QueueItem
44
45 from music_assistant.controllers.music.recency import RecencySnapshot
46 from music_assistant.controllers.player_queues.controller import PlayerQueuesController
47
48
49class SmartShuffle:
50 """Produce a recency-aware, well-spaced ordering of upcoming queue items."""
51
52 def __init__(self, queues: PlayerQueuesController) -> None:
53 """
54 Initialize the smart shuffle helper.
55
56 :param queues: The owning player queues controller.
57 """
58 self.queues = queues
59 self.mass = queues.mass
60 self.logger = queues.logger.getChild("smart_shuffle")
61
62 def is_enabled(self, queue_id: str) -> bool:
63 """
64 Return whether smart shuffle is enabled for the given queue.
65
66 Follows the global (queue controller) setting when the per-queue value is "global".
67
68 :param queue_id: The queue to read the smart-shuffle setting for.
69 """
70 return (
71 self.mass.config.get_effective_player_queue_config_value(
72 queue_id, CONF_SMART_SHUFFLE_ENABLED, CONF_VALUE_DISABLED
73 )
74 == CONF_VALUE_ENABLED
75 )
76
77 async def arrange(self, queue: PlayerQueue, items: list[QueueItem]) -> list[QueueItem]:
78 """
79 Return the items reordered with recency-aware smart shuffle.
80
81 :param queue: The queue being (re)shuffled; its owner scopes the play history.
82 :param items: The upcoming queue items to reorder.
83 """
84 windows = self.windows()
85 snapshot = await self.mass.music.recency.snapshot(
86 windows, userid=self.queues.queue_data(queue.queue_id).userid
87 )
88 return _arrange(items, snapshot, windows)
89
90 def windows(self) -> RecencyWindows:
91 """Read the configured recency windows (in seconds). These are a global-only setting."""
92 return RecencyWindows(
93 song_seconds=self._window_seconds(
94 CONF_SMART_SHUFFLE_SONG_RECENCY, SMART_SHUFFLE_SONG_RECENCY_DEFAULT
95 ),
96 artist_seconds=self._window_seconds(
97 CONF_SMART_SHUFFLE_ARTIST_RECENCY, SMART_SHUFFLE_ARTIST_RECENCY_DEFAULT
98 ),
99 duplicate_gap_seconds=self._window_seconds(
100 CONF_SMART_SHUFFLE_DUPLICATE_GAP, SMART_SHUFFLE_DUPLICATE_GAP_DEFAULT
101 ),
102 )
103
104 def _window_seconds(self, key: str, default: int) -> int:
105 """Read a window preset (seconds, 0 = off) from the global queue-controller config."""
106 raw = self.mass.config.get_raw_core_config_value(CONF_PLAYER_QUEUES, key, default)
107 try:
108 return int(raw)
109 except TypeError, ValueError:
110 return default
111
112
113def _arrange(
114 items: list[QueueItem], snapshot: RecencySnapshot, windows: RecencyWindows
115) -> list[QueueItem]:
116 """
117 Reorder items by recency tier, then spread duplicates and same-artist items within each tier.
118
119 :param items: The queue items to reorder.
120 :param snapshot: The play-history snapshot to score recency against.
121 :param windows: The configured recency windows (singleton song window vs duplicate gap).
122 """
123 if len(items) <= 2:
124 return random.sample(items, len(items))
125 counts = Counter(_song_key(item) for item in items)
126 tiers: dict[int, list[QueueItem]] = {0: [], 1: [], 2: []}
127 for item in items:
128 tiers[_tier(item, counts, snapshot, windows)].append(item)
129 result: list[QueueItem] = []
130 for tier in (0, 1, 2):
131 if bucket := tiers[tier]:
132 result.extend(_space_artists(_interleave(bucket)))
133 return result
134
135
136def _tier(
137 item: QueueItem,
138 counts: Counter[tuple[str, str]],
139 snapshot: RecencySnapshot,
140 windows: RecencyWindows,
141) -> int:
142 """Return the recency tier: 0 fresh, 1 artist recently played, 2 song recently played."""
143 media_item = item.media_item
144 if media_item is None:
145 return 0
146 # a deliberately-duplicated song uses the short repeat-gap, a singleton the long song window
147 song_window = (
148 windows.song_seconds if counts[_song_key(item)] == 1 else windows.duplicate_gap_seconds
149 )
150 if snapshot.track_recent(media_item, song_window):
151 return 2
152 if any(snapshot.artist_recent(name, windows.artist_seconds) for name in _artist_names(item)):
153 return 1
154 return 0
155
156
157def _interleave(bucket: list[QueueItem]) -> list[QueueItem]:
158 """Spread each distinct song's copies with independently randomized repeat cycles."""
159 groups: dict[tuple[str, str], list[QueueItem]] = defaultdict(list)
160 for item in bucket:
161 groups[_song_key(item)].append(item)
162 return interleave_groups(list(groups.values()))
163
164
165def _space_artists(items: list[QueueItem], *, preceding: set[str] | None = None) -> list[QueueItem]:
166 """
167 Best-effort separate directly-adjacent same-artist items.
168
169 :param items: The items to space.
170 :param preceding: Artist names of the item that will sit directly before the first item (the
171 seam with the already-queued tail); the first item is kept clear of it too. None ignores it.
172 """
173 order = space_by_artist([_artist_name_set(item) for item in items], preceding=preceding)
174 return [items[index] for index in order]
175
176
177def _song_key(item: QueueItem) -> tuple[str, str]:
178 """Return the grouping key identifying the same song (falls back to a unique id)."""
179 media_item = item.media_item
180 if media_item is None:
181 return ("", item.queue_item_id)
182 return (media_item.provider, media_item.item_id)
183
184
185def _artist_names(item: QueueItem) -> list[str]:
186 """Return the artist names for the item's media item (empty for non-track items)."""
187 return [
188 artist.name for artist in getattr(item.media_item, "artists", None) or () if artist.name
189 ]
190
191
192def _artist_name_set(item: QueueItem) -> set[str]:
193 """Return the lowercased set of artist names for the item."""
194 return {name.lower() for name in _artist_names(item)}
195