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            return
63
64        async def _do_prepare() -> None:
65            try:
66                # fetch streamdetails if not yet available
67                if not next_item.streamdetails:
68                    next_item.streamdetails = await self.mass.streams.audio.get_stream_details(
69                        queue_item=next_item
70                    )
71                self.logger.debug(
72                    "Preparing audio buffer for next track %s on queue %s",
73                    next_item.name,
74                    queue.display_name,
75                )
76                await self.mass.streams.audio.get_audio_buffer(
77                    next_item,
78                    reason="prepare_next",
79                    capacity_wait_timeout=STREAM_SLOT_WAIT_TIMEOUT,
80                    # speculative preparation gives up softly, so it must stay cheap:
81                    # leave the cross-provider search to the actual playback start
82                    allow_provider_match=False,
83                )
84            except (AudioError, MediaNotFoundError) as err:
85                self.logger.debug("Failed to prepare next audio buffer: %s", err)
86            except asyncio.CancelledError:
87                # a replacement prepare aborted this one: release the half-filled source
88                # so its slot is not pinned until the inactivity sweep
89                if (sd := next_item.streamdetails) and (buf := sd.buffer) and buf.is_buffering:
90                    await asyncio.shield(buf.clear())
91                raise
92
93        self.mass.create_task(
94            _do_prepare,
95            task_id=f"prepare_next_audio_buffer_{queue_id}",
96            abort_existing=True,
97        )
98
99    def _enqueue_next_item(self, queue_id: str, next_item: QueueItem | None) -> None:
100        """Enqueue the next item on the player."""
101        if not next_item:
102            # no next item, nothing to do...
103            return
104
105        queue_data = self._queue_data[queue_id]
106        queue = queue_data.queue
107        session_id = queue_data.session_id
108        if queue.flow_mode:
109            # ignore this for flow mode
110            return
111
112        async def _enqueue_next_item_on_player(next_item: QueueItem) -> None:
113            # Player state updates can lag behind queue loading, so wait before validating.
114            async with self.mass.players.wait_for_player_update(
115                queue_id,
116                attribute_name="playback_state",
117                attribute_value=PlaybackState.PLAYING,
118            ):
119                pass
120
121            player = self.mass.players.get_player(queue_id)
122            if (
123                player is None
124                or player.state.playback_state != PlaybackState.PLAYING
125                or player.state.active_source not in (queue.queue_id, None)
126                or queue_data.session_id != session_id
127                or queue.flow_mode
128            ):
129                # nothing re-attempts this handover, so a skip here means the player runs out
130                # of audio when the current track ends - leave a trace of why it was skipped
131                self.logger.debug(
132                    "Not enqueuing next track %s on queue %s "
133                    "(state: %s, source: %s, same session: %s, flow mode: %s)",
134                    next_item.name,
135                    queue.display_name,
136                    player.state.playback_state if player else "player unavailable",
137                    player.state.active_source if player else None,
138                    queue_data.session_id == session_id,
139                    queue.flow_mode,
140                )
141                return
142
143            current_item = queue.current_item
144            if current_item is None:
145                return
146            current_next = self.get_next_item(queue_id, current_item.queue_item_id)
147            if current_next is None or current_next.queue_item_id != next_item.queue_item_id:
148                return
149
150            await self.mass.players.enqueue_next_media(
151                player_id=queue_id,
152                media=await self.player_media_from_queue_item(next_item),
153            )
154            if queue_data.next_item_id_enqueued != next_item.queue_item_id:
155                queue_data.next_item_id_enqueued = next_item.queue_item_id
156                self.logger.debug(
157                    "Enqueued next track %s on queue %s",
158                    next_item.name,
159                    queue.display_name,
160                )
161
162        task_id = f"enqueue_next_item_{queue_id}"
163        self.mass.call_later(1, _enqueue_next_item_on_player, next_item, task_id=task_id)
164
165    def _preload_next_item(self, queue_id: str, item_id_in_buffer: str) -> None:
166        """
167        Preload the streamdetails for the next item in the queue/buffer.
168
169        This basically ensures the item is playable and fetches the stream details.
170        If an error occurs, the item will be skipped and the next item will be loaded.
171        """
172        queue = self._queue_data[queue_id].queue
173
174        async def _preload_streamdetails(item_id_in_buffer: str) -> None:
175            try:
176                # wait for the item that was loaded in the buffer is the actually playing item
177                # this prevents a race condition when we preload the next item too soon
178                # while the player is actually preloading the previously enqueued item.
179                current_item = queue.current_item
180                if current_item is None:
181                    return  # guard
182                retries = max(120, int(current_item.duration or 0) + 10)
183                for _ in range(retries):
184                    # the queue can drain to empty while we sleep (e.g. all remaining
185                    # items skipped as unplayable); stop waiting once it has no current item
186                    current_item = queue.current_item
187                    if current_item is None:
188                        return
189                    if current_item.queue_item_id == item_id_in_buffer:
190                        break
191                    await asyncio.sleep(1)
192                if next_item := await self.load_next_queue_item(queue_id, item_id_in_buffer):
193                    self.logger.debug(
194                        "Preloaded next item %s for queue %s",
195                        next_item.name,
196                        queue.display_name,
197                    )
198                    # enqueue the next item on the player
199                    self._enqueue_next_item(queue_id, next_item)
200
201            except QueueEmpty:
202                return
203
204        if not (current_item := self.get_item(queue_id, item_id_in_buffer)):
205            # this should not happen, but guard anyways
206            return
207        if current_item.media_type == MediaType.RADIO or not current_item.duration:
208            # radio items or no duration, nothing to do
209            return
210
211        task_id = f"preload_next_item_{queue_id}"
212        self.mass.create_task(
213            _preload_streamdetails,
214            item_id_in_buffer,
215            task_id=task_id,
216            abort_existing=True,
217        )
218
219    async def _cleanup_stale_queue_buffers(self, queue_id: str, current_index: int) -> None:
220        """
221        Clean up audio buffers for queue items that are no longer needed.
222
223        This clears buffers for items at index <= current_index - 2, keeping only:
224        - The previous track (current_index - 1)
225        - The current track (current_index)
226        - The next track (current_index + 1, handled by preloading)
227
228        :param queue_id: The queue ID to clean up buffers for.
229        :param current_index: The current playing index in the queue.
230        """
231        if current_index < 2:
232            return  # Nothing to clean up yet
233
234        queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
235        cleanup_threshold = current_index - 2
236        buffers_cleared = 0
237
238        for idx, item in enumerate(queue_items):
239            if idx > cleanup_threshold:
240                break  # No need to check further
241            if item.streamdetails and item.streamdetails.buffer:
242                self.logger.log(
243                    VERBOSE_LOG_LEVEL,
244                    "Clearing stale audio buffer for queue item %s (index %d) in queue %s",
245                    item.name,
246                    idx,
247                    queue_id,
248                )
249                await item.streamdetails.buffer.clear()
250                item.streamdetails.buffer = None
251                buffers_cleared += 1
252
253        if buffers_cleared > 0:
254            self.logger.debug(
255                "Cleared %d stale audio buffer(s) for queue %s (items before index %d)",
256                buffers_cleared,
257                queue_id,
258                cleanup_threshold + 1,
259            )
260
261    async def _cleanup_queue_audio_data(self, queue_id: str) -> None:
262        """
263        Clean up all audio-related data for a queue when it is stopped or cleared.
264
265        This clears:
266        - All audio buffers attached to queue item streamdetails
267        - Any pending crossfade data for the queue
268
269        :param queue_id: The queue ID to clean up.
270        """
271        self.mass.streams.audio.clear_crossfade_data(queue_id)
272
273        queue_items = queue_data.items if (queue_data := self._queue_data.get(queue_id)) else []
274        buffers_cleared = 0
275
276        for item in queue_items:
277            if item.streamdetails and item.streamdetails.buffer:
278                await item.streamdetails.buffer.clear()
279                item.streamdetails.buffer = None
280                buffers_cleared += 1
281
282        if buffers_cleared > 0:
283            self.logger.debug(
284                "Cleared %d audio buffer(s) for stopped/cleared queue %s",
285                buffers_cleared,
286                queue_id,
287            )
288