/
/
/
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 interleave_groups[ItemT](groups: list[list[ItemT]]) -> list[ItemT]:
206 """
207 Randomly interleave groups while preserving the item order within each group.
208
209 :param groups: The ordered item groups to spread across the result.
210 """
211 positioned: list[tuple[float, ItemT]] = []
212 for items in groups:
213 total = len(items)
214 for offset, item in enumerate(items):
215 positioned.append(((offset + random.random()) / total, item))
216 positioned.sort(key=lambda entry: entry[0])
217 return [item for _, item in positioned]
218
219
220# how many bounded passes to make separating directly-adjacent same-artist items
221ARTIST_REPAIR_PASSES = 4
222# how far ahead to look for a non-clashing item to swap in (keeps moves local so any
223# existing even spread is preserved)
224ARTIST_SWAP_WINDOW = 6
225
226
227def space_by_artist(artist_sets: list[set[str]], *, preceding: set[str] | None = None) -> list[int]:
228 """
229 Return an index order that best-effort keeps same-artist entries from sitting adjacent.
230
231 :param artist_sets: The lowercased artist-name set for each item, in its current order.
232 :param preceding: Artist names of the item that will sit directly before the first entry (the
233 seam with the already-queued tail); the first entry is kept clear of it too. None ignores it.
234 """
235 count = len(artist_sets)
236 order = list(range(count))
237 sets = list(artist_sets)
238 for _ in range(ARTIST_REPAIR_PASSES):
239 changed = False
240 # index -1 represents the preceding (seam) item, so the first entry is kept clear of it too
241 for index in range(-1, count - 1):
242 current = preceding if index == -1 else sets[index]
243 if not current or not current & sets[index + 1]:
244 continue
245 for target in range(index + 2, min(index + 2 + ARTIST_SWAP_WINDOW, count)):
246 if not current & sets[target]:
247 order[index + 1], order[target] = order[target], order[index + 1]
248 sets[index + 1], sets[target] = sets[target], sets[index + 1]
249 changed = True
250 break
251 if not changed:
252 break
253 return order
254