music-assistant-server

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