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 _enqueue_next_item(self, queue_id: str, next_item: QueueItem | None) -> None:
103        """Enqueue the next item on the player."""
104        if not next_item:
105            # no next item, nothing to do...
106            return
107
108        queue_data = self._queue_data[queue_id]
109        queue = queue_data.queue
110        session_id = queue_data.session_id
111        if queue.flow_mode:
112            # ignore this for flow mode
113            return
114
115        async def _enqueue_next_item_on_player(next_item: QueueItem) -> None:
116            # Player state updates can lag behind queue loading, so wait before validating.
117            async with self.mass.players.wait_for_player_update(
118                queue_id,
119                attribute_name="playback_state",
120                attribute_value=PlaybackState.PLAYING,
121            ):
122                pass
123
124            player = self.mass.players.get_player(queue_id)
125            if (
126                player is None
127                or player.state.playback_state != PlaybackState.PLAYING
128                or player.state.active_source not in (queue.queue_id, None)
129                or queue_data.session_id != session_id
130                or queue.flow_mode
131            ):
132                # nothing re-attempts this handover, so a skip here means the player runs out
133                # of audio when the current track ends - leave a trace of why it was skipped
134                self.logger.debug(
135                    "Not enqueuing next track %s on queue %s "
136                    "(state: %s, source: %s, same session: %s, flow mode: %s)",
137                    next_item.name,
138                    queue.display_name,
139                    player.state.playback_state if player else "player unavailable",
140                    player.state.active_source if player else None,
141                    queue_data.session_id == session_id,
142                    queue.flow_mode,
143                )
144                return
145
146            current_item = queue.current_item
147            if current_item is None:
148                return
149            current_next = self.get_next_item(queue_id, current_item.queue_item_id)
150            if current_next is None or current_next.queue_item_id != next_item.queue_item_id:
151                return
152
153            await self.mass.players.enqueue_next_media(
154                player_id=queue_id,
155                media=await self.player_media_from_queue_item(next_item),
156            )
157            if queue_data.next_item_id_enqueued != next_item.queue_item_id:
158                queue_data.next_item_id_enqueued = next_item.queue_item_id
159                self.logger.debug(
160                    "Enqueued next track %s on queue %s",
161                    next_item.name,
162                    queue.display_name,
163                )
164
165        task_id = f"enqueue_next_item_{queue_id}"
166        self.mass.call_later(1, _enqueue_next_item_on_player, next_item, task_id=task_id)
167
168    def _preload_next_item(self, queue_id: str, item_id_in_buffer: str) -> None:
169        """
170        Preload the streamdetails for the next item in the queue/buffer.
171
172        This basically ensures the item is playable and fetches the stream details.
173        If an error occurs, the item will be skipped and the next item will be loaded.
174        """
175        queue = self._queue_data[queue_id].queue
176
177        async def _preload_streamdetails(item_id_in_buffer: str) -> None:
178            try:
179                # wait for the item that was loaded in the buffer is the actually playing item
180                # this prevents a race condition when we preload the next item too soon
181                # while the player is actually preloading the previously enqueued item.
182                current_item = queue.current_item
183                if current_item is None:
184                    return  # guard
185                retries = max(120, int(current_item.duration or 0) + 10)
186                for _ in range(retries):
187                    # the queue can drain to empty while we sleep (e.g. all remaining
188                    # items skipped as unplayable); stop waiting once it has no current item
189                    current_item = queue.current_item
190                    if current_item is None:
191                        return
192                    if current_item.queue_item_id == item_id_in_buffer:
193                        break
194                    await asyncio.sleep(1)
195                if next_item := await self.load_next_queue_item(queue_id, item_id_in_buffer):
196                    self.logger.debug(
197                        "Preloaded next item %s for queue %s",
198                        next_item.name,
199                        queue.display_name,
200                    )
201                    # enqueue the next item on the player
202                    self._enqueue_next_item(queue_id, next_item)
203
204            except QueueEmpty:
205                return
206
207        if not (current_item := self.get_item(queue_id, item_id_in_buffer)):
208            # this should not happen, but guard anyways
209            return
210        if current_item.media_type == MediaType.RADIO or not current_item.duration:
211            # radio items or no duration, nothing to do
212            return
213
214        task_id = f"preload_next_item_{queue_id}"
215        self.mass.create_task(
216            _preload_streamdetails,
217            item_id_in_buffer,
218            task_id=task_id,
219            abort_existing=True,
220        )
221
222    async def _cleanup_stale_queue_buffers(self, queue_id: str, current_index: int) -> None:
223        """
224        Clean up audio buffers for queue items that are no longer needed.
225
226        This clears buffers for items at index <= current_index - 2, keeping only:
227        - The previous track (current_index - 1)
228        - The current track (current_index)
229        - The next track (current_index + 1, handled by preloading)
230
231        :param queue_id: The queue ID to clean up buffers for.
232        :param current_index: The current playing index in the queue.
233        """
234        if current_index < 2:
235            return  # Nothing to clean up yet
236
237        queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
238        cleanup_threshold = current_index - 2
239        buffers_cleared = 0
240
241        for idx, item in enumerate(queue_items):
242            if idx > cleanup_threshold:
243                break  # No need to check further
244            if (streamdetails := item.streamdetails) and (buffer := streamdetails.buffer):
245                self.logger.log(
246                    VERBOSE_LOG_LEVEL,
247                    "Clearing stale audio buffer for queue item %s (index %d) in queue %s",
248                    item.name,
249                    idx,
250                    queue_id,
251                )
252                # detached before releasing, as in _cleanup_queue_audio_data
253                streamdetails.buffer = None
254                await buffer.clear()
255                buffers_cleared += 1
256
257        if buffers_cleared > 0:
258            self.logger.debug(
259                "Cleared %d stale audio buffer(s) for queue %s (items before index %d)",
260                buffers_cleared,
261                queue_id,
262                cleanup_threshold + 1,
263            )
264
265    async def _cleanup_queue_audio_data(self, queue_id: str, session_id: str | None = None) -> None:
266        """
267        Clean up all audio-related data for a queue when it is stopped or cleared.
268
269        This clears:
270        - All audio buffers attached to queue item streamdetails
271        - Any pending crossfade data for the queue
272
273        :param queue_id: The queue ID to clean up.
274        :param session_id: The playback session being stopped. Audio the queue's currently
275            playing session claimed is left alone; everything else is released, including
276            what sessions that ended earlier left behind. None clears every buffer.
277        """
278        self.mass.streams.audio.clear_crossfade_data(queue_id)
279
280        queue_data = self._queue_data.get(queue_id)
281        queue_items = queue_data.items if queue_data else []
282        buffers_cleared = 0
283
284        for item in queue_items:
285            if not (streamdetails := item.streamdetails) or not (buffer := streamdetails.buffer):
286                continue
287            # read the playing session per item rather than once: releasing a buffer suspends,
288            # and a session that starts during one of those waits owns what it attaches after.
289            # A session id only protects audio while that session is the one playing - sessions
290            # rotate without a stop, so a claim that is no longer current marks audio nobody
291            # will come back for.
292            playing_session = queue_data.session_id if queue_data else None
293            if (
294                session_id is not None
295                and playing_session not in (None, session_id)
296                and streamdetails.queue_session_id == playing_session
297            ):
298                # playback restarted here while this stop was still running; killing its
299                # producer would strand the session that is playing now
300                continue
301            # detach before releasing: clearing suspends on the producer's cancellation, and a
302            # session starting in that window attaches its own buffer here
303            streamdetails.buffer = None
304            await buffer.clear()
305            buffers_cleared += 1
306
307        if buffers_cleared > 0:
308            self.logger.debug(
309                "Cleared %d audio buffer(s) for stopped/cleared queue %s",
310                buffers_cleared,
311                queue_id,
312            )
313