/
/
/
1"""
2Audio buffer implementation for PCM audio streaming.
3
4AudioBuffer is the primary interface for all buffered audio streaming in Music Assistant.
5It stores raw decoded PCM audio (no filters applied) and provides methods to:
6- Fill the buffer from any async generator of audio chunks
7- Get raw or processed (filtered/resampled) audio streams
8- Seek within buffered audio
9"""
10
11from __future__ import annotations
12
13import asyncio
14import logging
15import time
16from collections import deque
17from collections.abc import AsyncGenerator, Callable
18from contextlib import aclosing, suppress
19from typing import TYPE_CHECKING, Any, Final
20
21from music_assistant_models.enums import (
22 ContentType,
23 MediaType,
24 VolumeNormalizationMode,
25)
26from music_assistant_models.errors import AudioError
27from music_assistant_models.media_items import AudioFormat
28
29from music_assistant.constants import MASS_LOGGER_NAME, VERBOSE_LOG_LEVEL
30from music_assistant.controllers.streams.constants import (
31 BUFFER_SIZE_MAP,
32 CONF_BUFFER_SIZE,
33 CONF_BUFFER_SIZE_DEFAULT,
34 RADIO_BUFFER_SIZE,
35 SEEK_WAIT_THRESHOLD,
36 STREAM_SLOT_WAIT_TIMEOUT,
37 BufferMode,
38 BufferSize,
39)
40from music_assistant.helpers.audio import arriving_audio_format
41from music_assistant.helpers.ffmpeg import get_ffmpeg_stream
42from music_assistant.models.music_provider import MusicProvider
43
44if TYPE_CHECKING:
45 from music_assistant_models.streamdetails import StreamDetails
46
47 from music_assistant.mass import MusicAssistant
48
49LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.audio_buffer")
50
51# Callback signature for cancel observers: invoked when the buffer is cancelled/cleared.
52CancelCallback = Callable[[], None]
53
54# Maximum seconds to wait for the first playable audio, on top of any time the producer
55# is allowed to spend waiting for a provider source-stream slot.
56BUFFER_READY_TIMEOUT: Final[int] = 15
57
58
59class AudioBufferEOF(Exception):
60 """Exception raised when the audio buffer reaches end-of-file."""
61
62
63class AudioBufferDiscarded(Exception):
64 """
65 Raised when a passive (analysis) reader requests a chunk evicted from the retained window.
66
67 Means the reader is a full window behind playback, so its session is dropped.
68 """
69
70
71class AudioBuffer:
72 """
73 Raw PCM audio buffer with seek support and optional filter processing.
74
75 Stores audio in the original sample rate and bit depth.
76 Use get_raw_stream() for unprocessed PCM and get_stream() for
77 filtered/resampled output.
78 """
79
80 def __init__(
81 self,
82 pcm_format: AudioFormat,
83 buffer_size: BufferSize = BufferSize.BALANCED,
84 mode: BufferMode = BufferMode.SEEKABLE,
85 ready_threshold: int = 1,
86 is_realtime: bool = False,
87 ) -> None:
88 """
89 Initialize AudioBuffer.
90
91 :param pcm_format: The PCM audio format specification.
92 :param buffer_size: Buffer size preset.
93 :param mode: Buffer mode (SEEKABLE for tracks, ROLLING for radio).
94 :param ready_threshold: Seconds of audio to buffer before signaling ready.
95 :param is_realtime: Whether the source hands its audio over at playback pace.
96 """
97 self.pcm_format = pcm_format
98 self.is_realtime = is_realtime
99 self.max_size_seconds = (
100 RADIO_BUFFER_SIZE if mode == BufferMode.ROLLING else BUFFER_SIZE_MAP[buffer_size]
101 )
102 self.mode = mode
103 self._ready_threshold = ready_threshold
104 self._ready_at_chunk = ready_threshold # updated by get_buffer to account for seek
105 self._chunks: deque[bytes] = deque()
106 self._discarded_chunks = 0
107 self._lock = asyncio.Lock()
108 self._data_available = asyncio.Condition(self._lock)
109 self._space_available = asyncio.Condition(self._lock)
110 self._eof_received = False
111 self._producer_task: asyncio.Task[None] | None = None
112 self._fill_started = time.monotonic()
113 self._source_name = "unknown"
114 self._last_access_time: float = time.time()
115 self._inactivity_task: asyncio.Task[None] | None = None
116 self._cancelled = False
117 self._producer_error: Exception | None = None
118 self._background_tasks: set[asyncio.Task[None]] = set()
119 self.ready = asyncio.Event()
120 self._cancel_callbacks: list[CancelCallback] = []
121 self._ready_wait_lock = asyncio.Lock()
122
123 # -- Properties --
124
125 @property
126 def cancelled(self) -> bool:
127 """Return whether the buffer has been cancelled or cleared."""
128 if self._cancelled:
129 return True
130 return self._producer_task is not None and self._producer_task.cancelled()
131
132 @property
133 def has_error(self) -> bool:
134 """Return whether the producer encountered an error."""
135 return self._producer_error is not None
136
137 @property
138 def chunk_size_bytes(self) -> int:
139 """Return the size in bytes of one second of PCM audio."""
140 return self.pcm_format.pcm_sample_size
141
142 @property
143 def size_seconds(self) -> int:
144 """Return current size of the buffer in seconds."""
145 return len(self._chunks)
146
147 @property
148 def seconds_available(self) -> int:
149 """Return number of seconds of audio currently available."""
150 return len(self._chunks)
151
152 @property
153 def duration_available(self) -> float:
154 """Return the exact duration of resident PCM audio in seconds."""
155 return sum(len(chunk) for chunk in self._chunks) / self.pcm_format.pcm_sample_size
156
157 @property
158 def is_buffering(self) -> bool:
159 """Return whether the upstream source producer is still active."""
160 return self._producer_task is not None and not self._producer_task.done()
161
162 @property
163 def eof(self) -> bool:
164 """
165 Return whether the source stopped producing.
166
167 A source that failed after delivering audio also ends here, so pair this with
168 ``has_error`` when a clean finish is what matters.
169 """
170 return self._eof_received
171
172 @property
173 def first_buffered_chunk(self) -> int:
174 """Return the chunk number of the oldest chunk still retained in the buffer."""
175 return self._discarded_chunks
176
177 # -- Public methods --
178
179 def register_cancel_callback(self, callback: CancelCallback) -> None:
180 """
181 Register a callback to be invoked when the buffer is cancelled or cleared.
182
183 :param callback: Callable with no arguments, invoked on cancel.
184 """
185 self._cancel_callbacks.append(callback)
186
187 def is_valid(self, seek_position_ms: int = 0) -> bool:
188 """
189 Check if the buffer can serve the given seek position.
190
191 :param seek_position_ms: The position to seek to in milliseconds.
192 """
193 if self.cancelled:
194 return False
195
196 # reset inactivity timer â checking validity is activity
197 self._last_access_time = time.time()
198
199 seek_chunk = seek_position_ms // 1000
200
201 if seek_chunk < self._discarded_chunks:
202 return False
203
204 total_chunks = self._discarded_chunks + len(self._chunks)
205 if seek_chunk < total_chunks or self._eof_received:
206 return True
207
208 # The position is ahead of what the producer has made. One that runs
209 # faster than playback covers that in a fraction of the time, so waiting
210 # beats starting a new producer - but one that hands its audio over at
211 # playback pace needs exactly as long as the gap, while a fresh producer
212 # starts at the position right away (see get_buffer below).
213 chunks_ahead = seek_chunk - total_chunks
214 return chunks_ahead <= (0 if self.is_realtime else SEEK_WAIT_THRESHOLD)
215
216 async def get_raw_stream(
217 self, seek_position_ms: int = 0, exact_seek: bool = False
218 ) -> AsyncGenerator[bytes]:
219 """
220 Get raw (unprocessed) PCM audio from the buffer.
221
222 :param seek_position_ms: Starting position in milliseconds.
223 :param exact_seek: Preserve millisecond precision instead of quantizing to 100 ms.
224 """
225 if not exact_seek:
226 # align regular user seeks to 100ms steps to avoid rounding issues
227 seek_position_ms = (seek_position_ms // 100) * 100
228 chunk_number = seek_position_ms // 1000
229 # handle fractional seek: trim leading samples from the first chunk
230 fractional_ms = seek_position_ms % 1000
231 trim_bytes = 0
232 if fractional_ms > 0:
233 samples_to_trim = self.pcm_format.sample_rate * fractional_ms // 1000
234 bytes_per_sample = (self.pcm_format.bit_depth // 8) * self.pcm_format.channels
235 trim_bytes = samples_to_trim * bytes_per_sample
236
237 while True:
238 try:
239 self._last_access_time = time.time()
240 chunk = await self._get(chunk_number=chunk_number)
241 if trim_bytes > 0:
242 chunk = chunk[trim_bytes:]
243 trim_bytes = 0
244 yield chunk
245 chunk_number += 1
246 except AudioBufferEOF:
247 break
248
249 async def read_chunk_for_analysis(self, chunk_number: int) -> bytes:
250 """
251 Return one PCM chunk for a passive (analysis) reader, waiting until it is available.
252
253 A read-only accessor: it leaves the buffer untouched â no discard, no producer-space
254 signalling, no inactivity-timer reset â so an analysis reader never affects playback's
255 buffering.
256
257 :param chunk_number: Absolute chunk index to read.
258 :raises AudioBufferEOF: the stream ended before this chunk.
259 :raises AudioBufferDiscarded: the chunk has been evicted from the retained window (the
260 reader is a full window behind playback) or the buffer was torn down.
261 """
262 async with self._data_available:
263 while True:
264 if self.cancelled:
265 raise AudioBufferDiscarded
266 if chunk_number < self._discarded_chunks:
267 raise AudioBufferDiscarded
268 index = chunk_number - self._discarded_chunks
269 if index < len(self._chunks):
270 return self._chunks[index]
271 if self._producer_error:
272 raise self._producer_error
273 if self._eof_received:
274 raise AudioBufferEOF
275 await self._data_available.wait()
276
277 async def get_stream(
278 self,
279 output_format: AudioFormat,
280 seek_position_ms: int = 0,
281 filter_params: list[str] | None = None,
282 exact_seek: bool = False,
283 ) -> AsyncGenerator[bytes]:
284 """
285 Get processed audio from the buffer.
286
287 Returns audio in the requested output format with optional filters applied.
288 If no processing is needed, yields directly from the buffer.
289
290 :param output_format: The desired output PCM format.
291 :param seek_position_ms: Starting position in milliseconds.
292 :param filter_params: FFmpeg filter parameters to apply.
293 :param exact_seek: Preserve millisecond precision for the input buffer position.
294 """
295 needs_ffmpeg = bool(filter_params) or self.pcm_format != output_format
296
297 if not needs_ffmpeg:
298 async for chunk in self.get_raw_stream(
299 seek_position_ms=seek_position_ms, exact_seek=exact_seek
300 ):
301 yield chunk
302 return
303
304 async for chunk in get_ffmpeg_stream(
305 audio_input=self.get_raw_stream(
306 seek_position_ms=seek_position_ms, exact_seek=exact_seek
307 ),
308 input_format=self.pcm_format,
309 output_format=output_format,
310 filter_params=filter_params,
311 ):
312 yield chunk
313
314 def fill(
315 self,
316 audio_source: AsyncGenerator[bytes],
317 source_name: str = "unknown",
318 on_complete: Callable[[], None] | None = None,
319 ) -> None:
320 """
321 Start filling the buffer from an async generator of PCM audio chunks.
322
323 :param audio_source: Async generator yielding 1-second PCM audio chunks.
324 :param source_name: Name for logging purposes.
325 :param on_complete: Called once the source delivered everything, after its
326 generator (and the stream slot it held) has been released.
327 """
328 self._fill_started = time.monotonic()
329 self._source_name = source_name
330
331 async def _fill_task() -> None:
332 chunk_count = 0
333 status = "running"
334 try:
335 # aclosing guarantees the source generator (and any ffmpeg chain
336 # behind it) is finalized immediately when this task is cancelled,
337 # instead of lingering until garbage collection.
338 async with aclosing(audio_source):
339 async for chunk in audio_source:
340 chunk_count += 1
341 await self._put(chunk)
342 await asyncio.sleep(0)
343 await self._set_eof()
344 if on_complete is not None:
345 # isolated like the cancel callbacks in clear(): a failing
346 # callback must not retro-fail a source that delivered fully
347 try:
348 on_complete()
349 except Exception:
350 LOGGER.exception("Completion callback failed for %s", source_name)
351 except asyncio.CancelledError:
352 status = "cancelled"
353 raise
354 except Exception as err:
355 status = "aborted with error"
356 # record the error before the EOF signal below, so readers that
357 # check for a producer error never observe the abort as a clean EOF
358 self._producer_error = err
359 raise
360 finally:
361 # signal EOF even on error if we produced valid chunks,
362 # so the consumer can read all buffered data before seeing the error
363 if status == "aborted with error" and chunk_count > 0:
364 await self._set_eof()
365 LOGGER.log(
366 VERBOSE_LOG_LEVEL,
367 "fill: %s (%s chunks) for %s",
368 status,
369 chunk_count,
370 source_name,
371 )
372
373 loop = asyncio.get_running_loop()
374 task = loop.create_task(_fill_task())
375 self._attach_producer_task(task)
376
377 async def clear(self, cancel_inactivity_task: bool = True) -> None:
378 """Reset the buffer, clearing all data and cancelling active tasks."""
379 chunk_count = len(self._chunks)
380 LOGGER.log(
381 VERBOSE_LOG_LEVEL,
382 "AudioBuffer.clear: Resetting buffer (had %s chunks, producer: %s)",
383 chunk_count,
384 self._producer_task is not None,
385 )
386 if self._producer_task and not self._producer_task.done():
387 self._producer_task.cancel()
388 with suppress(asyncio.CancelledError):
389 await self._producer_task
390
391 if cancel_inactivity_task and self._inactivity_task and not self._inactivity_task.done():
392 self._inactivity_task.cancel()
393 with suppress(asyncio.CancelledError):
394 await self._inactivity_task
395
396 # signal cancel callbacks only if the stream did not complete normally
397 if not self._eof_received:
398 for callback in list(self._cancel_callbacks):
399 try:
400 callback()
401 except Exception:
402 LOGGER.exception("Cancel callback failed during clear")
403
404 async with self._lock:
405 self._chunks = deque()
406 self._discarded_chunks = 0
407 self._eof_received = False
408 self._cancelled = True
409 self._producer_error = None
410 self.ready.clear()
411 self._cancel_callbacks.clear()
412 self._data_available.notify_all()
413 self._space_available.notify_all()
414
415 @staticmethod
416 async def get_buffer(
417 mass: MusicAssistant,
418 streamdetails: StreamDetails,
419 seek_position_ms: int = 0,
420 wait_ready: bool = False,
421 reason: str = "",
422 source_wait_timeout: float | None = STREAM_SLOT_WAIT_TIMEOUT,
423 ) -> AudioBuffer:
424 """
425 Get or create an AudioBuffer for the given streamdetails.
426
427 Reuses an existing valid buffer if available.
428 Buffer size is determined from the streams controller configuration.
429
430 :param mass: The MusicAssistant instance.
431 :param streamdetails: The stream details for the media.
432 :param seek_position_ms: Position in milliseconds to start from.
433 :param wait_ready: If True, wait for the first chunk before returning.
434 :param reason: Caller context for logging (e.g. 'prepare', 'streaming').
435 :param source_wait_timeout: Maximum seconds the producer may wait for a free
436 source-stream slot on the providing music provider, or None to wait
437 without a timeout.
438 :raises AudioError: If the buffer does not become ready, wrapping the typed
439 producer error (e.g. ProviderStreamLimitError) when there is one.
440 """
441 log_prefix = f"get_buffer[{reason}]" if reason else "get_buffer"
442 # the producer may spend its source wait before the first byte arrives,
443 # so the readiness budget covers that wait on top of the audio itself
444 ready_timeout = BUFFER_READY_TIMEOUT + (source_wait_timeout or 0)
445
446 # reuse existing valid buffer
447 existing_buffer: AudioBuffer | None = streamdetails.buffer
448 if existing_buffer is not None:
449 if existing_buffer.has_error or not existing_buffer.is_valid(seek_position_ms):
450 LOGGER.debug(
451 "%s: Existing buffer invalid for %s (seek_ms: %s, discarded: %s)",
452 log_prefix,
453 streamdetails.uri,
454 seek_position_ms,
455 existing_buffer._discarded_chunks,
456 )
457 streamdetails.buffer = None
458 # a still-filling producer holds one of the provider's source-stream slots.
459 # The replacement needs a slot, so take this one back only when the provider
460 # has none free - otherwise a superseded consumer keeps draining its audio.
461 provider = mass.get_provider(streamdetails.provider, return_unavailable=True)
462 must_release_slot = (
463 existing_buffer.is_buffering
464 and isinstance(provider, MusicProvider)
465 and provider.max_concurrent_streams is not None
466 and not provider.has_available_stream_slot
467 )
468 if must_release_slot or time.time() - existing_buffer._last_access_time > 30:
469 await asyncio.shield(existing_buffer.clear())
470 # else: an active consumer is still reading via its local reference;
471 # the inactivity monitor will clean up after it finishes
472 else:
473 LOGGER.debug(
474 "%s: Reusing buffer for %s - available: %ss, seek_ms: %s, discarded: %s",
475 log_prefix,
476 streamdetails.uri,
477 existing_buffer.seconds_available,
478 seek_position_ms,
479 existing_buffer._discarded_chunks,
480 )
481 if wait_ready:
482 await existing_buffer._wait_until_ready(
483 streamdetails, ready_timeout, log_prefix
484 )
485 return existing_buffer
486
487 audio_buffer, buffer_seek_seconds = _new_buffer(
488 mass, streamdetails, seek_position_ms, log_prefix
489 )
490
491 # start filling from the media stream (seek in seconds for FFmpeg)
492 audio_source = mass.streams.audio.get_media_stream(
493 streamdetails,
494 audio_buffer.pcm_format,
495 seek_position=buffer_seek_seconds,
496 filter_params=None,
497 source_wait_timeout=source_wait_timeout,
498 )
499
500 def _source_complete() -> None:
501 # a realtime source's one stream slot frees the moment this item has
502 # fully arrived: start fetching the next item right away
503 if (
504 streamdetails.is_realtime
505 and streamdetails.media_type == MediaType.TRACK
506 and streamdetails.queue_id
507 ):
508 mass.player_queues.prepare_next_audio_buffer(streamdetails.queue_id)
509
510 audio_buffer.fill(audio_source, source_name=streamdetails.uri, on_complete=_source_complete)
511
512 if wait_ready:
513 await audio_buffer._wait_until_ready(streamdetails, ready_timeout, log_prefix)
514
515 return audio_buffer
516
517 # -- Private methods --
518
519 async def _wait_until_ready(
520 self, streamdetails: StreamDetails, ready_timeout: float, log_prefix: str
521 ) -> None:
522 """
523 Wait until this buffer can serve playback or raise its producer failure.
524
525 :param streamdetails: Stream details currently referencing this buffer.
526 :param ready_timeout: Maximum seconds to wait for enough buffered audio.
527 :param log_prefix: Caller context for logging.
528 """
529 async with self._ready_wait_lock:
530 if not self.ready.is_set():
531 try:
532 await asyncio.wait_for(self.ready.wait(), timeout=ready_timeout)
533 except TimeoutError as err:
534 # clear() does not wake this wait, and only marks the buffer cancelled
535 # once the producer is gone - so a buffer released elsewhere (to free a
536 # stream slot) lands here on an abort of our own making
537 producer = self._producer_task
538 releasing = self.cancelled or bool(producer and producer.cancelling())
539 if not releasing:
540 LOGGER.warning(
541 "%s: Gave up on %s (%s) after %.2fs, %ss buffered",
542 log_prefix,
543 streamdetails.provider,
544 streamdetails.uri,
545 time.monotonic() - self._fill_started,
546 self.seconds_available,
547 )
548 producer_error = await self._clear_failed_buffer(streamdetails)
549 if isinstance(producer_error, AudioError):
550 raise producer_error from err
551 raise AudioError("Timeout waiting for audio data") from (producer_error or err)
552 # ready was signaled but check if it was due to a producer error
553 # (ready is also set by _notify_on_producer_error)
554 if not self.has_error:
555 return
556 producer_error = await self._clear_failed_buffer(streamdetails)
557 # surface a typed producer failure (e.g. a source capacity limit) as-is,
558 # so callers can act on it instead of on a generic wrapper
559 if isinstance(producer_error, AudioError):
560 raise producer_error
561 raise AudioError("Failed to stream audio") from producer_error
562
563 async def _clear_failed_buffer(self, streamdetails: StreamDetails) -> Exception | None:
564 """
565 Detach and clear this buffer after preparation failed.
566
567 :param streamdetails: Stream details currently referencing this buffer.
568 :return: The producer error recorded before the buffer was cleared.
569 """
570 producer_error = self._producer_error
571 if streamdetails.buffer is self:
572 streamdetails.buffer = None
573 await asyncio.shield(self.clear())
574 return producer_error
575
576 async def _put(self, chunk: bytes) -> None:
577 """
578 Put a 1-second chunk of PCM audio into the buffer.
579
580 Waits for space when the buffer is full (backpressure).
581 """
582 async with self._lock:
583 if self._cancelled:
584 return
585
586 if self._eof_received:
587 LOGGER.log(
588 VERBOSE_LOG_LEVEL, "AudioBuffer._put: EOF already received, rejecting chunk"
589 )
590 return
591
592 # wait for the consumer to free space when buffer is full
593 await self._wait_for_space()
594
595 chunk_position = self._discarded_chunks + len(self._chunks)
596 self._chunks.append(chunk)
597 if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
598 LOGGER.log(
599 VERBOSE_LOG_LEVEL,
600 "AudioBuffer._put: Added chunk at position %s (size: %s bytes, buffer: %s)",
601 chunk_position,
602 len(chunk),
603 len(self._chunks),
604 )
605
606 if not self.ready.is_set() and (
607 self._discarded_chunks + len(self._chunks) >= self._ready_at_chunk
608 or len(self._chunks) >= self.max_size_seconds
609 ):
610 self._mark_ready()
611
612 self._data_available.notify_all()
613
614 async def _set_eof(self) -> None:
615 """Signal that no more data will be added to the buffer."""
616 async with self._lock:
617 LOGGER.log(
618 VERBOSE_LOG_LEVEL,
619 "AudioBuffer._set_eof: Marking EOF (buffer has %s chunks)",
620 len(self._chunks),
621 )
622 self._eof_received = True
623 if not self.ready.is_set():
624 self._mark_ready()
625 self._data_available.notify_all()
626 self._space_available.notify_all()
627
628 def _mark_ready(self) -> None:
629 """Signal that the buffer holds audio a consumer can start playing."""
630 self.ready.set()
631 # a source that failed or ended empty also lands here, without the
632 # buffer ever having become playable
633 if self._chunks and not self.has_error:
634 LOGGER.debug(
635 "AudioBuffer: %s became ready after %.2fs",
636 self._source_name,
637 time.monotonic() - self._fill_started,
638 )
639
640 async def _get(self, chunk_number: int = 0) -> bytes:
641 """
642 Get one second of audio at the given chunk position.
643
644 Waits until the chunk is available. Discards old chunks when full.
645
646 :raises AudioBufferEOF: If EOF is reached or the buffer was cleared.
647 :raises AudioError: If the chunk has been discarded or the producer failed.
648 """
649 async with self._data_available:
650 if len(self._chunks) == 0:
651 # Producer errors also set EOF after buffered data; preserve the real failure.
652 if self._producer_error:
653 raise self._producer_error
654 if self._eof_received or self.cancelled:
655 raise AudioBufferEOF
656 if self.cancelled:
657 raise AudioBufferEOF
658
659 if self.mode == BufferMode.ROLLING:
660 return await self._get_rolling()
661
662 return await self._get_seekable(chunk_number)
663
664 async def _get_rolling(self) -> bytes:
665 """
666 Pop the next chunk from the buffer (FIFO).
667
668 Must be called while holding _data_available lock.
669 """
670 while len(self._chunks) == 0:
671 if self._producer_error:
672 raise self._producer_error
673 if self.cancelled or self._eof_received:
674 raise AudioBufferEOF
675 await self._data_available.wait()
676
677 result = self._chunks.popleft()
678 self._discarded_chunks += 1
679 self._space_available.notify_all()
680 return result
681
682 async def _get_seekable(self, chunk_number: int) -> bytes:
683 """
684 Get a specific chunk by number from the buffer.
685
686 Must be called while holding _data_available lock.
687 """
688 if chunk_number < self._discarded_chunks:
689 msg = (
690 f"Chunk {chunk_number} has been discarded "
691 f"(buffer starts at {self._discarded_chunks})"
692 )
693 raise AudioError(msg)
694
695 buffer_index = chunk_number - self._discarded_chunks
696 while buffer_index >= len(self._chunks):
697 # Producer errors also set EOF after buffered data; preserve the real failure.
698 if self._producer_error:
699 raise self._producer_error
700 if self.cancelled or self._eof_received:
701 raise AudioBufferEOF
702 # if the buffer is full and we need a chunk that hasn't arrived yet,
703 # the producer is blocked waiting for space â evict to unblock it
704 if len(self._chunks) >= self.max_size_seconds:
705 self._chunks.popleft()
706 self._discarded_chunks += 1
707 buffer_index = chunk_number - self._discarded_chunks
708 self._space_available.notify_all()
709 continue
710 await self._data_available.wait()
711 buffer_index = chunk_number - self._discarded_chunks
712
713 result = self._chunks[buffer_index]
714
715 # free space for the producer when buffer is at capacity,
716 # but only if the producer is still running and needs space
717 if (
718 len(self._chunks) >= self.max_size_seconds
719 and not self._eof_received
720 and self._producer_task
721 and not self._producer_task.done()
722 ):
723 self._chunks.popleft()
724 self._discarded_chunks += 1
725 self._space_available.notify_all()
726
727 return result
728
729 async def _wait_for_space(self) -> None:
730 """Wait until buffer has space. Must be called while holding _lock."""
731 while len(self._chunks) >= self.max_size_seconds:
732 if self._cancelled:
733 return
734 await self._space_available.wait()
735
736 def _attach_producer_task(self, task: asyncio.Task[Any]) -> None:
737 """Attach a background task that fills the buffer."""
738 self._producer_task = task
739
740 def _on_producer_done(t: asyncio.Task[Any]) -> None:
741 if t.cancelled():
742 return
743 exc = t.exception()
744 if exc is not None and isinstance(exc, Exception):
745 self._producer_error = exc
746 loop = asyncio.get_running_loop()
747 task = loop.create_task(self._notify_on_producer_error())
748 self._background_tasks.add(task)
749 task.add_done_callback(self._background_tasks.discard)
750
751 task.add_done_callback(_on_producer_done)
752
753 if self._inactivity_task is None or self._inactivity_task.done():
754 self._last_access_time = time.time()
755 loop = asyncio.get_running_loop()
756 self._inactivity_task = loop.create_task(self._monitor_inactivity())
757
758 async def _monitor_inactivity(
759 self, inactivity_timeout: float = 300, check_interval: float = 30
760 ) -> None:
761 """
762 Clear the buffer once it has been inactive for inactivity_timeout seconds.
763
764 :param inactivity_timeout: Seconds without access before the buffer is released.
765 :param check_interval: Seconds between inactivity checks.
766 """
767 while True:
768 await asyncio.sleep(check_interval)
769 time_since_access = time.time() - self._last_access_time
770 # break on inactivity regardless of how many chunks remain: a rolling buffer
771 # that has drained to empty (e.g. an abandoned radio stream) must still release
772 # its resources and stop this monitor, otherwise the task loops forever
773 if time_since_access > inactivity_timeout:
774 LOGGER.log(
775 VERBOSE_LOG_LEVEL,
776 "AudioBuffer: No activity for %.1fs, clearing (%s chunks)",
777 time_since_access,
778 len(self._chunks),
779 )
780 break
781 await self.clear(cancel_inactivity_task=False)
782
783 async def _notify_on_producer_error(self) -> None:
784 """Notify waiting consumers that the producer has failed."""
785 async with self._lock:
786 if not self.ready.is_set():
787 self.ready.set()
788 self._data_available.notify_all()
789
790
791def _new_buffer(
792 mass: MusicAssistant,
793 streamdetails: StreamDetails,
794 seek_position_ms: int,
795 log_prefix: str,
796) -> tuple[AudioBuffer, int]:
797 """
798 Create the buffer for the given stream details and attach it to them.
799
800 :param mass: The MusicAssistant instance.
801 :param streamdetails: The stream details the buffer belongs to.
802 :param seek_position_ms: Position in milliseconds playback starts from.
803 :param log_prefix: Caller context for logging.
804 :return: The buffer and the position (in seconds) its producer should start at.
805 """
806 # determine buffer size from config
807 buffer_size = BufferSize(
808 mass.config.get_raw_core_config_value("streams", CONF_BUFFER_SIZE, CONF_BUFFER_SIZE_DEFAULT)
809 )
810 mode = (
811 BufferMode.ROLLING
812 if (not streamdetails.duration or not streamdetails.allow_seek)
813 else BufferMode.SEEKABLE
814 )
815
816 # convert ms to seconds for get_media_stream (FFmpeg works in seconds)
817 seek_seconds = seek_position_ms // 1000
818
819 # for large seeks without existing buffer, start at seek position.
820 # A realtime source can not produce the skipped audio any faster than playback,
821 # so it always seeks at the source instead of buffering up to the seek point.
822 buffer_seek_seconds = seek_seconds if streamdetails.is_realtime or seek_seconds > 60 else 0
823
824 pcm_format = _buffer_pcm_format(streamdetails)
825
826 # determine ready threshold: how many seconds of audio must be buffered
827 # before signaling ready for playback
828 queue = mass.player_queues.get(streamdetails.queue_id) if streamdetails.queue_id else None
829 crossfade_enabled = bool(
830 queue and queue.crossfade_enabled and streamdetails.media_type == MediaType.TRACK
831 )
832 dynamic_normalization = (
833 streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC
834 )
835 if streamdetails.is_realtime:
836 # A realtime source fills the buffer at playback pace, so every second of
837 # audio asked for here is a second of extra startup delay - on a seek or a
838 # track change as much as on a start. The queue's crossfade setting buys
839 # nothing for such a source, because its fade streams in as it arrives and
840 # is sized by the tail the outgoing track banked, not by what is resident
841 # here. Only dynamic normalization, which genuinely needs lookahead, raises
842 # this.
843 ready_threshold = 2 if dynamic_normalization else 1
844 elif crossfade_enabled:
845 ready_threshold = 8
846 elif dynamic_normalization:
847 # radio streams are continuous so the normalization will converge quickly,
848 # use a lower threshold to reduce startup latency
849 ready_threshold = 3 if streamdetails.media_type == MediaType.RADIO else 5
850 else:
851 ready_threshold = 2
852
853 # cap threshold at buffer capacity to prevent deadlock
854 max_size = RADIO_BUFFER_SIZE if mode == BufferMode.ROLLING else BUFFER_SIZE_MAP[buffer_size]
855 ready_threshold = min(ready_threshold, max_size)
856
857 LOGGER.debug(
858 "%s: Creating new buffer for %s (mode: %s, size: %s, seek_ms: %s)",
859 log_prefix,
860 streamdetails.uri,
861 mode,
862 buffer_size,
863 seek_position_ms,
864 )
865 audio_buffer = AudioBuffer(
866 pcm_format,
867 buffer_size,
868 mode,
869 ready_threshold=ready_threshold,
870 is_realtime=streamdetails.is_realtime,
871 )
872 # align chunk numbering with the actual stream start position so that
873 # get_raw_stream(seek_position_ms) requests the correct chunk number
874 audio_buffer._discarded_chunks = buffer_seek_seconds
875 # nothing has been read yet, so the source is not ahead of playback here
876 # set the chunk number at which the buffer should signal ready,
877 # accounting for seek position so we have enough data past the seek point
878 seek_chunk = seek_position_ms // 1000
879 audio_buffer._ready_at_chunk = seek_chunk + ready_threshold
880 streamdetails.buffer = audio_buffer
881
882 # attach analyze jobs for ahead-of-time processing
883 # skip AudioSource and SoundEffect â they should not feed the long-running analyzer flow
884 # (radio still runs analysis; the analyzer caps it at 10 minutes)
885 if seek_position_ms == 0 and streamdetails.media_type not in (
886 MediaType.AUDIO_SOURCE,
887 MediaType.SOUND_EFFECT,
888 ):
889 # audio analysis providers (loudness, beat tracking, key detection, etc.).
890 # Fire-and-forget: analysis setup â including a possible model (re)load â must never
891 # delay the buffer fill. The analysis worker reads the retained chunks once ready.
892 mass.create_task(mass.streams.audio_analysis.start_analysis(audio_buffer, streamdetails))
893
894 return audio_buffer, buffer_seek_seconds
895
896
897def _buffer_pcm_format(streamdetails: StreamDetails) -> AudioFormat:
898 """
899 Return the PCM format a buffer for these streamdetails holds.
900
901 The buffer stores decoded PCM, so it follows the audio that actually
902 arrives: ``audio_format`` may describe a source the provider decoded on our
903 behalf and can differ in depth or rate, in which case deriving the buffer
904 from it would resample or truncate real audio.
905
906 :param streamdetails: The stream the buffer is for.
907 """
908 arriving = arriving_audio_format(streamdetails)
909 return AudioFormat(
910 content_type=ContentType.from_bit_depth(arriving.bit_depth),
911 sample_rate=arriving.sample_rate,
912 bit_depth=arriving.bit_depth,
913 # buffer the stereo fold of a surround source, so audio analysis measures
914 # the same audio that is played back rather than the untouched surround mix
915 channels=min(arriving.channels, 2),
916 )
917