/
/
/
1"""Helper utilities for the player queues controller."""
2
3from __future__ import annotations
4
5import functools
6import random
7from collections.abc import Awaitable, Callable, Coroutine
8from typing import TYPE_CHECKING, Any, Concatenate, Protocol, TypedDict, TypeGuard, TypeVar
9
10from music_assistant_models.media_items import MediaItemMetadata, Playlist, Radio, Track
11from music_assistant_models.queue_item import QueueItem
12
13from music_assistant.constants import ATTR_PLAY_ACTION_IN_PROGRESS, PlaylistPlayableItem
14from music_assistant.controllers.players.constants import PlayerLockPurpose
15
16if TYPE_CHECKING:
17 from music_assistant_models.enums import ContentType, PlaybackState
18 from music_assistant_models.media_items import (
19 BrowseFolder,
20 MediaItemType,
21 PlayableMediaItemType,
22 )
23 from music_assistant_models.player_queue import PlayerQueue
24
25 from music_assistant import MusicAssistant
26 from music_assistant.controllers.player_queues.state import PlayerQueueData
27 from music_assistant.models.player import Player
28
29_SortableT = TypeVar("_SortableT", bound=PlaylistPlayableItem)
30
31
32class CompareState(TypedDict):
33 """
34 Simple object where we store the (previous) state of a queue.
35
36 Used for compare actions.
37 """
38
39 queue_id: str
40 state: PlaybackState
41 current_item_id: str | None
42 next_item_id: str | None
43 current_item: QueueItem | None
44 elapsed_time: int
45 # last_playing_elapsed_time: elapsed time from the last PLAYING state update
46 # used to determine if a track was fully played when transitioning to idle
47 last_playing_elapsed_time: int
48 stream_title: str | None
49 codec_type: ContentType | None
50 output_player_ids: list[str] | None
51
52
53class _PlayActionHost(Protocol):
54 """
55 The minimal controller surface that :func:`handle_play_action` needs.
56
57 Lets the decorator wrap actions defined either on the controller itself or on one
58 of its mixins, since both expose this surface at runtime.
59 """
60
61 mass: MusicAssistant
62 _queue_data: dict[str, PlayerQueueData]
63
64 def signal_update(self, queue_id: str, items_changed: bool = False) -> None: ...
65
66 def on_player_update(
67 self, player: Player, changed_values: dict[str, tuple[Any, Any]]
68 ) -> None: ...
69
70
71def handle_play_action[PlayActionHostT: _PlayActionHost, **P, R](
72 func: Callable[Concatenate[PlayActionHostT, P], Awaitable[R]],
73) -> Callable[Concatenate[PlayActionHostT, P], Coroutine[Any, Any, R]]:
74 """
75 Decorator for queue playback actions.
76
77 Acquires the shared playback lock for the queue's player (re-entrant)
78 and sets ATTR_PLAY_ACTION_IN_PROGRESS on the queue while the action runs.
79 Uses an internal refcount so nested actions don't clear the flag prematurely.
80
81 :param func: The function to wrap.
82 """ # noqa: D401
83
84 @functools.wraps(func)
85 async def wrapper(self: PlayActionHostT, *args: P.args, **kwargs: P.kwargs) -> R:
86 """Execute function with playback lock and play action flag set."""
87 queue_id = kwargs.get("queue_id") or args[0]
88 assert isinstance(queue_id, str) # for type checking
89 queue_data = self._queue_data.get(queue_id)
90 if queue_data is None:
91 return await func(self, *args, **kwargs)
92 queue = queue_data.queue
93 async with self.mass.players.get_player_lock(queue_id, PlayerLockPurpose.PLAYBACK):
94 prev_in_progress = queue.extra_attributes.get(ATTR_PLAY_ACTION_IN_PROGRESS, False)
95 try:
96 queue_data.play_action_refcount += 1
97 queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] = True
98 if not prev_in_progress:
99 self.signal_update(queue_id)
100 return await func(self, *args, **kwargs)
101 finally:
102 queue_data.play_action_refcount -= 1
103 if queue_data.play_action_refcount <= 0:
104 queue_data.play_action_refcount = 0
105 queue.extra_attributes[ATTR_PLAY_ACTION_IN_PROGRESS] = False
106 # the queue follows the player through a debounced update, which is also
107 # suppressed while an action is transitioning; recalculate it here so the
108 # update that clears the flag already carries the action's resulting state
109 if (player := self.mass.players.get_player(queue_id)) is not None:
110 self.on_player_update(player, {})
111 self.signal_update(queue_id)
112
113 return wrapper
114
115
116def is_dynamic_source(item: MediaItemType | BrowseFolder) -> TypeGuard[Playlist | Radio]:
117 """Return True if the item supplies its own on-demand track feed."""
118 return isinstance(item, Playlist | Radio) and item.is_dynamic
119
120
121def find_dynamic_source(queue_data: PlayerQueueData) -> MediaItemType | None:
122 """
123 Return the queue's most recently added dynamic source, if it has one.
124
125 Prefers the queue's sources and falls back to what was enqueued on it.
126
127 :param queue_data: The queue to inspect.
128 """
129 for items in (queue_data.source_items, queue_data.enqueued_media_items):
130 for item in reversed(items):
131 if is_dynamic_source(item):
132 return item
133 return None
134
135
136def has_dynamic_source(source_items: list[MediaItemType]) -> bool:
137 """Return True if any source supplies its own on-demand track feed (the queue is dynamic)."""
138 return any(is_dynamic_source(item) for item in source_items)
139
140
141def build_queue_item(queue_id: str, media_item: PlayableMediaItemType) -> QueueItem:
142 """
143 Build a QueueItem for enqueueing, keeping its media item slim.
144
145 The returned item only carries the media details needed for the queue listing and stream
146 resolution. For tracks the full metadata is dropped; it is restored from the library when
147 the item becomes the queue's current or next item, so large queues stay light on memory
148 and persisted-cache size.
149
150 :param queue_id: The id of the queue the item is created for.
151 :param media_item: The source media item to enqueue.
152 """
153 queue_item = QueueItem.from_media_item(queue_id, media_item)
154 if isinstance(queue_item.media_item, Track):
155 # the list-row artwork is already captured on QueueItem.image, so dropping the
156 # track's metadata here does not lose anything the queue listing still needs
157 queue_item.media_item.metadata = MediaItemMetadata()
158 return queue_item
159
160
161def sort_tracks(tracks: list[_SortableT], sort_by: str) -> list[_SortableT]:
162 """Sort tracks by the given sort key."""
163 key_map: dict[str, tuple[Any, bool]] = {
164 "position_desc": (lambda t: getattr(t, "position", 0) or 0, True),
165 "name": (lambda t: (t.sort_name or t.name or "").lower(), False),
166 "artist": (
167 lambda t: (
168 (t.artists[0].sort_name or t.artists[0].name).lower()
169 if hasattr(t, "artists") and t.artists
170 else ""
171 ),
172 False,
173 ),
174 "album": (
175 lambda t: (
176 (t.album.sort_name or t.album.name).lower()
177 if hasattr(t, "album") and t.album
178 else ""
179 ),
180 False,
181 ),
182 "duration": (lambda t: getattr(t, "duration", 0) or 0, False),
183 "duration_desc": (lambda t: getattr(t, "duration", 0) or 0, True),
184 "track_number": (
185 lambda t: (
186 getattr(t, "disc_number", 0) or 0,
187 getattr(t, "track_number", 0) or 0,
188 ),
189 False,
190 ),
191 }
192 if sort_by in key_map:
193 key_fn, reverse = key_map[sort_by]
194 return sorted(tracks, key=key_fn, reverse=reverse)
195 return list(tracks)
196
197
198def get_current_playback_speed(queue: PlayerQueue) -> float:
199 """Return the playback_speed of the queue's current item (1.0 if unset)."""
200 if queue.current_item is None:
201 return 1.0
202 return float(queue.current_item.extra_attributes.get("playback_speed") or 1.0)
203
204
205def committed_index(queue: PlayerQueue) -> int | None:
206 """
207 Return the highest queue index the player owns, or None if it owns none.
208
209 It holds both the playing track and the one handed to it for the transition. Nothing at or
210 below this index can be reordered: the player keeps playing what it was given whatever the
211 queue says, and the buffered position does not follow items that move around it.
212
213 :param queue: The queue to resolve the index for.
214 """
215 if queue.current_index is None:
216 return queue.index_in_buffer
217 if queue.index_in_buffer is None:
218 return queue.current_index
219 # repeat wraps the buffered index back to the front while the last track plays
220 return max(queue.current_index, queue.index_in_buffer)
221
222
223def interleave_groups[ItemT](groups: list[list[ItemT]]) -> list[ItemT]:
224 """
225 Randomly interleave groups while preserving the item order within each group.
226
227 :param groups: The ordered item groups to spread across the result.
228 """
229 positioned: list[tuple[float, ItemT]] = []
230 for items in groups:
231 total = len(items)
232 for offset, item in enumerate(items):
233 positioned.append(((offset + random.random()) / total, item))
234 positioned.sort(key=lambda entry: entry[0])
235 return [item for _, item in positioned]
236
237
238# how many bounded passes to make separating directly-adjacent same-artist items
239ARTIST_REPAIR_PASSES = 4
240# how far ahead to look for a non-clashing item to swap in (keeps moves local so any
241# existing even spread is preserved)
242ARTIST_SWAP_WINDOW = 6
243
244
245def space_by_artist(artist_sets: list[set[str]], *, preceding: set[str] | None = None) -> list[int]:
246 """
247 Return an index order that best-effort keeps same-artist entries from sitting adjacent.
248
249 :param artist_sets: The lowercased artist-name set for each item, in its current order.
250 :param preceding: Artist names of the item that will sit directly before the first entry (the
251 seam with the already-queued tail); the first entry is kept clear of it too. None ignores it.
252 """
253 count = len(artist_sets)
254 order = list(range(count))
255 sets = list(artist_sets)
256 for _ in range(ARTIST_REPAIR_PASSES):
257 changed = False
258 # index -1 represents the preceding (seam) item, so the first entry is kept clear of it too
259 for index in range(-1, count - 1):
260 current = preceding if index == -1 else sets[index]
261 if not current or not current & sets[index + 1]:
262 continue
263 for target in range(index + 2, min(index + 2 + ARTIST_SWAP_WINDOW, count)):
264 if not current & sets[target]:
265 order[index + 1], order[target] = order[target], order[index + 1]
266 sets[index + 1], sets[target] = sets[target], sets[index + 1]
267 changed = True
268 break
269 if not changed:
270 break
271 return order
272