music-assistant-server

1"""
2Stream feeding for the Player Queues controller.
3
4Handles handing the next queue item to the player and preparing its audio: enqueuing the upcoming
5item on the player, preloading its stream details, warming the next track's AudioBuffer ahead of
6playback, and cleaning up stale buffers. Owns no per-queue state; it is mixed into the controller
7and reads/mutates the controller's `PlayerQueueData` records.
8"""
9
10from __future__ import annotations
11
12import asyncio
13from typing import TYPE_CHECKING
14
15from music_assistant_models.enums import (
16    MediaType,
17    PlaybackState,
18)
19from music_assistant_models.errors import (
20    AudioError,
21    MediaNotFoundError,
22    QueueEmpty,
23)
24
25from music_assistant.constants import (
26    VERBOSE_LOG_LEVEL,
27)
28from music_assistant.controllers.player_queues.base import _PlayerQueuesBase
29from music_assistant.controllers.streams.constants import STREAM_SLOT_WAIT_TIMEOUT
30
31if TYPE_CHECKING:
32    from music_assistant_models.queue_item import QueueItem
33
34
35class StreamFeederMixin(_PlayerQueuesBase):
36    """Feed the player's stream: enqueue the next item, preload/prepare its audio, clean up."""
37
38    def prepare_next_audio_buffer(self, queue_id: str) -> None:
39        """
40        Prepare the AudioBuffer for the next track in the queue.
41
42        Called ~30-60 seconds before the current track ends to ensure
43        the buffer is warm when the next track starts playing.
44        """
45        queue = self.get(queue_id)
46        if not queue or not queue.next_item:
47            return
48        next_item = queue.next_item
49        # AudioSource items are realtime/live and bypass the AudioBuffer
50        if next_item.media_type == MediaType.AUDIO_SOURCE:
51            return
52        # guard against race condition where queue.next_item still points to the
53        # currently playing track because the player state hasn't been updated yet
54        if queue.current_item and next_item.queue_item_id == queue.current_item.queue_item_id:
55            return
56        # check if buffer already exists and is valid
57        if (
58            next_item.streamdetails
59            and next_item.streamdetails.buffer
60            and next_item.streamdetails.buffer.is_valid()
61        ):
62            # reusing audio an earlier session left behind claims it for this one, so its
63            # stop releases it and the earlier session's stop no longer can
64            next_item.streamdetails.queue_session_id = self._queue_data[queue_id].session_id
65            return
66
67        async def _do_prepare() -> None:
68            try:
69                # fetch streamdetails if not yet available
70                if not next_item.streamdetails:
71                    next_item.streamdetails = await self.mass.streams.audio.get_stream_details(
72                        queue_item=next_item
73                    )
74                self.logger.debug(
75                    "Preparing audio buffer for next track %s on queue %s",
76                    next_item.name,
77                    queue.display_name,
78                )
79                await self.mass.streams.audio.get_audio_buffer(
80                    next_item,
81                    reason="prepare_next",
82                    capacity_wait_timeout=STREAM_SLOT_WAIT_TIMEOUT,
83                    # speculative preparation gives up softly, so it must stay cheap:
84                    # leave the cross-provider search to the actual playback start
85                    allow_provider_match=False,
86                )
87            except (AudioError, MediaNotFoundError) as err:
88                self.logger.debug("Failed to prepare next audio buffer: %s", err)
89            except asyncio.CancelledError:
90                # a replacement prepare aborted this one: release the half-filled source
91                # so its slot is not pinned until the inactivity sweep
92                if (sd := next_item.streamdetails) and (buf := sd.buffer) and buf.is_buffering:
93                    await asyncio.shield(buf.clear())
94                raise
95
96        self.mass.create_task(
97            _do_prepare,
98            task_id=f"prepare_next_audio_buffer_{queue_id}",
99            abort_existing=True,
100        )
101
102    def update_next_item_on_player(self, queue_id: str, force: bool = False) -> None:
103        """
104        Hand the player the track that now follows the one it is playing.
105
106        Does nothing when the player already holds that track, so a queue change that leaves the
107        upcoming track alone costs nothing.
108
109        :param queue_id: The queue whose player should be updated.
110        :param force: Hand it over even when the player already holds it, for a change that
111            alters how the same track is streamed rather than which track it is.
112        """
113        queue_data = self._queue_data[queue_id]
114        queue = queue_data.queue
115        if queue.state != PlaybackState.PLAYING or queue.current_index is None:
116            return
117        if queue.index_in_buffer is None or queue_data.transitioning:
118            # no settled position to follow: a replace clears the buffered index while it swaps
119            # the items, and a starting track moves the two indexes one after the other
120            return
121        next_item = self.get_next_item(queue_id, queue.current_index)
122        if next_item is None:
123            return
124        if not force and next_item.queue_item_id == queue_data.next_item_id_enqueued:
125            return
126        self._enqueue_next_item(queue_id, next_item)
127
128    def _enqueue_next_item(self, queue_id: str, next_item: QueueItem | None) -> None:
129        """Enqueue the next item on the player."""
130        if not next_item:
131            # no next item, nothing to do...
132            return
133
134        queue_data = self._queue_data[queue_id]
135        queue = queue_data.queue
136        session_id = queue_data.session_id
137        if queue.flow_mode:
138            # ignore this for flow mode
139            return
140
141        async def _enqueue_next_item_on_player(next_item: QueueItem) -> None:
142            # Player state updates can lag behind queue loading, so wait before validating.
143            async with self.mass.players.wait_for_player_update(
144                queue_id,
145                attribute_name="playback_state",
146                attribute_value=PlaybackState.PLAYING,
147            ):
148                pass
149
150            player = self.mass.players.get_player(queue_id)
151            if (
152                player is None
153                or player.state.playback_state != PlaybackState.PLAYING
154                or player.state.active_source not in (queue.queue_id, None)
155                or queue_data.session_id != session_id
156                or queue.flow_mode
157            ):
158                # nothing re-attempts this handover, so a skip here means the player runs out
159                # of audio when the current track ends - leave a trace of why it was skipped
160                self.logger.debug(
161                    "Not enqueuing next track %s on queue %s "
162                    "(state: %s, source: %s, same session: %s, flow mode: %s)",
163                    next_item.name,
164                    queue.display_name,
165                    player.state.playback_state if player else "player unavailable",
166                    player.state.active_source if player else None,
167                    queue_data.session_id == session_id,
168                    queue.flow_mode,
169                )
170                return
171
172            current_item = queue.current_item
173            if current_item is None:
174                return
175            current_next = self.get_next_item(queue_id, current_item.queue_item_id)
176            if current_next is None or current_next.queue_item_id != next_item.queue_item_id:
177                return
178
179            await self.mass.players.enqueue_next_media(
180                player_id=queue_id,
181                media=await self.player_media_from_queue_item(next_item),
182            )
183            if queue_data.next_item_id_enqueued != next_item.queue_item_id:
184                queue_data.next_item_id_enqueued = next_item.queue_item_id
185                self.logger.debug(
186                    "Enqueued next track %s on queue %s",
187                    next_item.name,
188                    queue.display_name,
189                )
190
191        task_id = f"enqueue_next_item_{queue_id}"
192        self.mass.call_later(1, _enqueue_next_item_on_player, next_item, task_id=task_id)
193
194    def _preload_next_item(self, queue_id: str, item_id_in_buffer: str) -> None:
195        """
196        Preload the streamdetails for the next item in the queue/buffer.
197
198        This basically ensures the item is playable and fetches the stream details.
199        If an error occurs, the item will be skipped and the next item will be loaded.
200        """
201        queue = self._queue_data[queue_id].queue
202
203        async def _preload_streamdetails(item_id_in_buffer: str) -> None:
204            try:
205                # wait for the item that was loaded in the buffer is the actually playing item
206                # this prevents a race condition when we preload the next item too soon
207                # while the player is actually preloading the previously enqueued item.
208                current_item = queue.current_item
209                if current_item is None:
210                    return  # guard
211                retries = max(120, int(current_item.duration or 0) + 10)
212                for _ in range(retries):
213                    # the queue can drain to empty while we sleep (e.g. all remaining
214                    # items skipped as unplayable); stop waiting once it has no current item
215                    current_item = queue.current_item
216                    if current_item is None:
217                        return
218                    if current_item.queue_item_id == item_id_in_buffer:
219                        break
220                    await asyncio.sleep(1)
221                if next_item := await self.load_next_queue_item(queue_id, item_id_in_buffer):
222                    self.logger.debug(
223                        "Preloaded next item %s for queue %s",
224                        next_item.name,
225                        queue.display_name,
226                    )
227                    # enqueue the next item on the player
228                    self._enqueue_next_item(queue_id, next_item)
229
230            except QueueEmpty:
231                return
232
233        if not (current_item := self.get_item(queue_id, item_id_in_buffer)):
234            # this should not happen, but guard anyways
235            return
236        if current_item.media_type == MediaType.RADIO or not current_item.duration:
237            # radio items or no duration, nothing to do
238            return
239
240        task_id = f"preload_next_item_{queue_id}"
241        self.mass.create_task(
242            _preload_streamdetails,
243            item_id_in_buffer,
244            task_id=task_id,
245            abort_existing=True,
246        )
247
248    async def _cleanup_stale_queue_buffers(self, queue_id: str, current_index: int) -> None:
249        """
250        Clean up audio buffers for queue items that are no longer needed.
251
252        This clears buffers for items at index <= current_index - 2, keeping only:
253        - The previous track (current_index - 1)
254        - The current track (current_index)
255        - The next track (current_index + 1, handled by preloading)
256
257        :param queue_id: The queue ID to clean up buffers for.
258        :param current_index: The current playing index in the queue.
259        """
260        if current_index < 2:
261            return  # Nothing to clean up yet
262
263        queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
264        cleanup_threshold = current_index - 2
265        buffers_cleared = 0
266
267        for idx, item in enumerate(queue_items):
268            if idx > cleanup_threshold:
269                break  # No need to check further
270            if (streamdetails := item.streamdetails) and (buffer := streamdetails.buffer):
271                self.logger.log(
272                    VERBOSE_LOG_LEVEL,
273                    "Clearing stale audio buffer for queue item %s (index %d) in queue %s",
274                    item.name,
275                    idx,
276                    queue_id,
277                )
278                # detached before releasing, as in _cleanup_queue_audio_data
279                streamdetails.buffer = None
280                await buffer.clear()
281                buffers_cleared += 1
282
283        if buffers_cleared > 0:
284            self.logger.debug(
285                "Cleared %d stale audio buffer(s) for queue %s (items before index %d)",
286                buffers_cleared,
287                queue_id,
288                cleanup_threshold + 1,
289            )
290
291    async def _cleanup_queue_audio_data(self, queue_id: str, session_id: str | None = None) -> None:
292        """
293        Clean up all audio-related data for a queue when it is stopped or cleared.
294
295        This clears:
296        - All audio buffers attached to queue item streamdetails
297        - Any pending crossfade data for the queue
298
299        :param queue_id: The queue ID to clean up.
300        :param session_id: The playback session being stopped. Audio the queue's currently
301            playing session claimed is left alone; everything else is released, including
302            what sessions that ended earlier left behind. None clears every buffer.
303        """
304        self.mass.streams.audio.clear_crossfade_handover(queue_id)
305
306        queue_data = self._queue_data.get(queue_id)
307        queue_items = queue_data.items if queue_data else []
308        buffers_cleared = 0
309
310        for item in queue_items:
311            if not (streamdetails := item.streamdetails) or not (buffer := streamdetails.buffer):
312                continue
313            # read the playing session per item rather than once: releasing a buffer suspends,
314            # and a session that starts during one of those waits owns what it attaches after.
315            # A session id only protects audio while that session is the one playing - sessions
316            # rotate without a stop, so a claim that is no longer current marks audio nobody
317            # will come back for.
318            playing_session = queue_data.session_id if queue_data else None
319            if (
320                session_id is not None
321                and playing_session not in (None, session_id)
322                and streamdetails.queue_session_id == playing_session
323            ):
324                # playback restarted here while this stop was still running; killing its
325                # producer would strand the session that is playing now
326                continue
327            # detach before releasing: clearing suspends on the producer's cancellation, and a
328            # session starting in that window attaches its own buffer here
329            streamdetails.buffer = None
330            await buffer.clear()
331            buffers_cleared += 1
332
333        if buffers_cleared > 0:
334            self.logger.debug(
335                "Cleared %d audio buffer(s) for stopped/cleared queue %s",
336                buffers_cleared,
337                queue_id,
338            )
339