/
/
/
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(self, audio_source: AsyncGenerator[bytes], source_name: str = "unknown") -> None:
315 """
316 Start filling the buffer from an async generator of PCM audio chunks.
317
318 :param audio_source: Async generator yielding 1-second PCM audio chunks.
319 :param source_name: Name for logging purposes.
320 """
321 self._fill_started = time.monotonic()
322 self._source_name = source_name
323
324 async def _fill_task() -> None:
325 chunk_count = 0
326 status = "running"
327 try:
328 # aclosing guarantees the source generator (and any ffmpeg chain
329 # behind it) is finalized immediately when this task is cancelled,
330 # instead of lingering until garbage collection.
331 async with aclosing(audio_source):
332 async for chunk in audio_source:
333 chunk_count += 1
334 await self._put(chunk)
335 await asyncio.sleep(0)
336 await self._set_eof()
337 except asyncio.CancelledError:
338 status = "cancelled"
339 raise
340 except Exception as err:
341 status = "aborted with error"
342 # record the error before the EOF signal below, so readers that
343 # check for a producer error never observe the abort as a clean EOF
344 self._producer_error = err
345 raise
346 finally:
347 # signal EOF even on error if we produced valid chunks,
348 # so the consumer can read all buffered data before seeing the error
349 if status == "aborted with error" and chunk_count > 0:
350 await self._set_eof()
351 LOGGER.log(
352 VERBOSE_LOG_LEVEL,
353 "fill: %s (%s chunks) for %s",
354 status,
355 chunk_count,
356 source_name,
357 )
358
359 loop = asyncio.get_running_loop()
360 task = loop.create_task(_fill_task())
361 self._attach_producer_task(task)
362
363 async def clear(self, cancel_inactivity_task: bool = True) -> None:
364 """Reset the buffer, clearing all data and cancelling active tasks."""
365 chunk_count = len(self._chunks)
366 LOGGER.log(
367 VERBOSE_LOG_LEVEL,
368 "AudioBuffer.clear: Resetting buffer (had %s chunks, producer: %s)",
369 chunk_count,
370 self._producer_task is not None,
371 )
372 if self._producer_task and not self._producer_task.done():
373 self._producer_task.cancel()
374 with suppress(asyncio.CancelledError):
375 await self._producer_task
376
377 if cancel_inactivity_task and self._inactivity_task and not self._inactivity_task.done():
378 self._inactivity_task.cancel()
379 with suppress(asyncio.CancelledError):
380 await self._inactivity_task
381
382 # signal cancel callbacks only if the stream did not complete normally
383 if not self._eof_received:
384 for callback in list(self._cancel_callbacks):
385 try:
386 callback()
387 except Exception:
388 LOGGER.exception("Cancel callback failed during clear")
389
390 async with self._lock:
391 self._chunks = deque()
392 self._discarded_chunks = 0
393 self._eof_received = False
394 self._cancelled = True
395 self._producer_error = None
396 self.ready.clear()
397 self._cancel_callbacks.clear()
398 self._data_available.notify_all()
399 self._space_available.notify_all()
400
401 @staticmethod
402 async def get_buffer(
403 mass: MusicAssistant,
404 streamdetails: StreamDetails,
405 seek_position_ms: int = 0,
406 wait_ready: bool = False,
407 reason: str = "",
408 source_wait_timeout: float | None = STREAM_SLOT_WAIT_TIMEOUT,
409 ) -> AudioBuffer:
410 """
411 Get or create an AudioBuffer for the given streamdetails.
412
413 Reuses an existing valid buffer if available.
414 Buffer size is determined from the streams controller configuration.
415
416 :param mass: The MusicAssistant instance.
417 :param streamdetails: The stream details for the media.
418 :param seek_position_ms: Position in milliseconds to start from.
419 :param wait_ready: If True, wait for the first chunk before returning.
420 :param reason: Caller context for logging (e.g. 'prepare', 'streaming').
421 :param source_wait_timeout: Maximum seconds the producer may wait for a free
422 source-stream slot on the providing music provider, or None to wait
423 without a timeout.
424 :raises AudioError: If the buffer does not become ready, wrapping the typed
425 producer error (e.g. ProviderStreamLimitError) when there is one.
426 """
427 log_prefix = f"get_buffer[{reason}]" if reason else "get_buffer"
428 # the producer may spend its source wait before the first byte arrives,
429 # so the readiness budget covers that wait on top of the audio itself
430 ready_timeout = BUFFER_READY_TIMEOUT + (source_wait_timeout or 0)
431 # determine buffer size from config
432 buffer_size = BufferSize(
433 mass.config.get_raw_core_config_value(
434 "streams", CONF_BUFFER_SIZE, CONF_BUFFER_SIZE_DEFAULT
435 )
436 )
437 mode = (
438 BufferMode.ROLLING
439 if (not streamdetails.duration or not streamdetails.allow_seek)
440 else BufferMode.SEEKABLE
441 )
442
443 # reuse existing valid buffer
444 existing_buffer: AudioBuffer | None = streamdetails.buffer
445 if existing_buffer is not None:
446 if existing_buffer.has_error or not existing_buffer.is_valid(seek_position_ms):
447 LOGGER.debug(
448 "%s: Existing buffer invalid for %s (seek_ms: %s, discarded: %s)",
449 log_prefix,
450 streamdetails.uri,
451 seek_position_ms,
452 existing_buffer._discarded_chunks,
453 )
454 streamdetails.buffer = None
455 # a still-filling producer holds one of the provider's source-stream slots.
456 # The replacement needs a slot, so take this one back only when the provider
457 # has none free - otherwise a superseded consumer keeps draining its audio.
458 provider = mass.get_provider(streamdetails.provider, return_unavailable=True)
459 must_release_slot = (
460 existing_buffer.is_buffering
461 and isinstance(provider, MusicProvider)
462 and provider.max_concurrent_streams is not None
463 and not provider.has_available_stream_slot
464 )
465 if must_release_slot or time.time() - existing_buffer._last_access_time > 30:
466 await asyncio.shield(existing_buffer.clear())
467 # else: an active consumer is still reading via its local reference;
468 # the inactivity monitor will clean up after it finishes
469 else:
470 LOGGER.debug(
471 "%s: Reusing buffer for %s - available: %ss, seek_ms: %s, discarded: %s",
472 log_prefix,
473 streamdetails.uri,
474 existing_buffer.seconds_available,
475 seek_position_ms,
476 existing_buffer._discarded_chunks,
477 )
478 if wait_ready:
479 await existing_buffer._wait_until_ready(
480 streamdetails, ready_timeout, log_prefix
481 )
482 return existing_buffer
483
484 # convert ms to seconds for get_media_stream (FFmpeg works in seconds)
485 seek_seconds = seek_position_ms // 1000
486
487 # for large seeks without existing buffer, start at seek position.
488 # A realtime source can not produce the skipped audio any faster than playback,
489 # so it always seeks at the source instead of buffering up to the seek point.
490 buffer_seek_seconds = seek_seconds if streamdetails.is_realtime or seek_seconds > 60 else 0
491
492 pcm_format = _buffer_pcm_format(streamdetails)
493
494 # determine ready threshold: how many seconds of audio must be buffered
495 # before signaling ready for playback
496 queue = mass.player_queues.get(streamdetails.queue_id) if streamdetails.queue_id else None
497 crossfade_enabled = bool(
498 queue and queue.crossfade_enabled and streamdetails.media_type == MediaType.TRACK
499 )
500 dynamic_normalization = (
501 streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC
502 )
503 if streamdetails.is_realtime:
504 # A realtime source fills the buffer at playback pace, so every second of
505 # audio asked for here is a second of extra startup delay - on a seek or a
506 # track change as much as on a start. The queue's crossfade setting buys
507 # nothing for such a source, because its fade streams in as it arrives and
508 # is sized by the tail the outgoing track banked, not by what is resident
509 # here. Only dynamic normalization, which genuinely needs lookahead, raises
510 # this.
511 ready_threshold = 2 if dynamic_normalization else 1
512 elif crossfade_enabled:
513 ready_threshold = 8
514 elif dynamic_normalization:
515 # radio streams are continuous so the normalization will converge quickly,
516 # use a lower threshold to reduce startup latency
517 ready_threshold = 3 if streamdetails.media_type == MediaType.RADIO else 5
518 else:
519 ready_threshold = 2
520
521 # cap threshold at buffer capacity to prevent deadlock
522 max_size = RADIO_BUFFER_SIZE if mode == BufferMode.ROLLING else BUFFER_SIZE_MAP[buffer_size]
523 ready_threshold = min(ready_threshold, max_size)
524
525 LOGGER.debug(
526 "%s: Creating new buffer for %s (mode: %s, size: %s, seek_ms: %s)",
527 log_prefix,
528 streamdetails.uri,
529 mode,
530 buffer_size,
531 seek_position_ms,
532 )
533 audio_buffer = AudioBuffer(
534 pcm_format,
535 buffer_size,
536 mode,
537 ready_threshold=ready_threshold,
538 is_realtime=streamdetails.is_realtime,
539 )
540 # align chunk numbering with the actual stream start position so that
541 # get_raw_stream(seek_position_ms) requests the correct chunk number
542 audio_buffer._discarded_chunks = buffer_seek_seconds
543 # set the chunk number at which the buffer should signal ready,
544 # accounting for seek position so we have enough data past the seek point
545 seek_chunk = seek_position_ms // 1000
546 audio_buffer._ready_at_chunk = seek_chunk + ready_threshold
547 streamdetails.buffer = audio_buffer
548
549 # attach analyze jobs for ahead-of-time processing
550 # skip AudioSource and SoundEffect â they should not feed the long-running analyzer flow
551 # (radio still runs analysis; the analyzer caps it at 10 minutes)
552 if seek_position_ms == 0 and streamdetails.media_type not in (
553 MediaType.AUDIO_SOURCE,
554 MediaType.SOUND_EFFECT,
555 ):
556 # audio analysis providers (loudness, beat tracking, key detection, etc.).
557 # Fire-and-forget: analysis setup â including a possible model (re)load â must never
558 # delay the buffer fill. The analysis worker reads the retained chunks once ready.
559 mass.create_task(
560 mass.streams.audio_analysis.start_analysis(audio_buffer, streamdetails)
561 )
562
563 # start filling from the media stream (seek in seconds for FFmpeg)
564 audio_source = mass.streams.audio.get_media_stream(
565 streamdetails,
566 pcm_format,
567 seek_position=buffer_seek_seconds,
568 filter_params=None,
569 source_wait_timeout=source_wait_timeout,
570 )
571 audio_buffer.fill(audio_source, source_name=streamdetails.uri)
572
573 if wait_ready:
574 await audio_buffer._wait_until_ready(streamdetails, ready_timeout, log_prefix)
575
576 return audio_buffer
577
578 # -- Private methods --
579
580 async def _wait_until_ready(
581 self, streamdetails: StreamDetails, ready_timeout: float, log_prefix: str
582 ) -> None:
583 """
584 Wait until this buffer can serve playback or raise its producer failure.
585
586 :param streamdetails: Stream details currently referencing this buffer.
587 :param ready_timeout: Maximum seconds to wait for enough buffered audio.
588 :param log_prefix: Caller context for logging.
589 """
590 async with self._ready_wait_lock:
591 if not self.ready.is_set():
592 try:
593 await asyncio.wait_for(self.ready.wait(), timeout=ready_timeout)
594 except TimeoutError as err:
595 # clear() does not wake this wait, and only marks the buffer cancelled
596 # once the producer is gone - so a buffer released elsewhere (to free a
597 # stream slot) lands here on an abort of our own making
598 producer = self._producer_task
599 releasing = self.cancelled or bool(producer and producer.cancelling())
600 if not releasing:
601 LOGGER.warning(
602 "%s: Gave up on %s (%s) after %.2fs, %ss buffered",
603 log_prefix,
604 streamdetails.provider,
605 streamdetails.uri,
606 time.monotonic() - self._fill_started,
607 self.seconds_available,
608 )
609 producer_error = await self._clear_failed_buffer(streamdetails)
610 if isinstance(producer_error, AudioError):
611 raise producer_error from err
612 raise AudioError("Timeout waiting for audio data") from (producer_error or err)
613 # ready was signaled but check if it was due to a producer error
614 # (ready is also set by _notify_on_producer_error)
615 if not self.has_error:
616 return
617 producer_error = await self._clear_failed_buffer(streamdetails)
618 # surface a typed producer failure (e.g. a source capacity limit) as-is,
619 # so callers can act on it instead of on a generic wrapper
620 if isinstance(producer_error, AudioError):
621 raise producer_error
622 raise AudioError("Failed to stream audio") from producer_error
623
624 async def _clear_failed_buffer(self, streamdetails: StreamDetails) -> Exception | None:
625 """
626 Detach and clear this buffer after preparation failed.
627
628 :param streamdetails: Stream details currently referencing this buffer.
629 :return: The producer error recorded before the buffer was cleared.
630 """
631 producer_error = self._producer_error
632 if streamdetails.buffer is self:
633 streamdetails.buffer = None
634 await asyncio.shield(self.clear())
635 return producer_error
636
637 async def _put(self, chunk: bytes) -> None:
638 """
639 Put a 1-second chunk of PCM audio into the buffer.
640
641 Waits for space when the buffer is full (backpressure).
642 """
643 async with self._lock:
644 if self._cancelled:
645 return
646
647 if self._eof_received:
648 LOGGER.log(
649 VERBOSE_LOG_LEVEL, "AudioBuffer._put: EOF already received, rejecting chunk"
650 )
651 return
652
653 # wait for the consumer to free space when buffer is full
654 await self._wait_for_space()
655
656 chunk_position = self._discarded_chunks + len(self._chunks)
657 self._chunks.append(chunk)
658 if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
659 LOGGER.log(
660 VERBOSE_LOG_LEVEL,
661 "AudioBuffer._put: Added chunk at position %s (size: %s bytes, buffer: %s)",
662 chunk_position,
663 len(chunk),
664 len(self._chunks),
665 )
666
667 if not self.ready.is_set() and (
668 self._discarded_chunks + len(self._chunks) >= self._ready_at_chunk
669 or len(self._chunks) >= self.max_size_seconds
670 ):
671 self._mark_ready()
672
673 self._data_available.notify_all()
674
675 async def _set_eof(self) -> None:
676 """Signal that no more data will be added to the buffer."""
677 async with self._lock:
678 LOGGER.log(
679 VERBOSE_LOG_LEVEL,
680 "AudioBuffer._set_eof: Marking EOF (buffer has %s chunks)",
681 len(self._chunks),
682 )
683 self._eof_received = True
684 if not self.ready.is_set():
685 self._mark_ready()
686 self._data_available.notify_all()
687 self._space_available.notify_all()
688
689 def _mark_ready(self) -> None:
690 """Signal that the buffer holds audio a consumer can start playing."""
691 self.ready.set()
692 # a source that failed or ended empty also lands here, without the
693 # buffer ever having become playable
694 if self._chunks and not self.has_error:
695 LOGGER.debug(
696 "AudioBuffer: %s became ready after %.2fs",
697 self._source_name,
698 time.monotonic() - self._fill_started,
699 )
700
701 async def _get(self, chunk_number: int = 0) -> bytes:
702 """
703 Get one second of audio at the given chunk position.
704
705 Waits until the chunk is available. Discards old chunks when full.
706
707 :raises AudioBufferEOF: If EOF is reached or the buffer was cleared.
708 :raises AudioError: If the chunk has been discarded or the producer failed.
709 """
710 async with self._data_available:
711 if len(self._chunks) == 0:
712 # Producer errors also set EOF after buffered data; preserve the real failure.
713 if self._producer_error:
714 raise self._producer_error
715 if self._eof_received or self.cancelled:
716 raise AudioBufferEOF
717 if self.cancelled:
718 raise AudioBufferEOF
719
720 if self.mode == BufferMode.ROLLING:
721 return await self._get_rolling()
722
723 return await self._get_seekable(chunk_number)
724
725 async def _get_rolling(self) -> bytes:
726 """
727 Pop the next chunk from the buffer (FIFO).
728
729 Must be called while holding _data_available lock.
730 """
731 while len(self._chunks) == 0:
732 if self._producer_error:
733 raise self._producer_error
734 if self.cancelled or self._eof_received:
735 raise AudioBufferEOF
736 await self._data_available.wait()
737
738 result = self._chunks.popleft()
739 self._discarded_chunks += 1
740 self._space_available.notify_all()
741 return result
742
743 async def _get_seekable(self, chunk_number: int) -> bytes:
744 """
745 Get a specific chunk by number from the buffer.
746
747 Must be called while holding _data_available lock.
748 """
749 if chunk_number < self._discarded_chunks:
750 msg = (
751 f"Chunk {chunk_number} has been discarded "
752 f"(buffer starts at {self._discarded_chunks})"
753 )
754 raise AudioError(msg)
755
756 buffer_index = chunk_number - self._discarded_chunks
757 while buffer_index >= len(self._chunks):
758 # Producer errors also set EOF after buffered data; preserve the real failure.
759 if self._producer_error:
760 raise self._producer_error
761 if self.cancelled or self._eof_received:
762 raise AudioBufferEOF
763 # if the buffer is full and we need a chunk that hasn't arrived yet,
764 # the producer is blocked waiting for space â evict to unblock it
765 if len(self._chunks) >= self.max_size_seconds:
766 self._chunks.popleft()
767 self._discarded_chunks += 1
768 buffer_index = chunk_number - self._discarded_chunks
769 self._space_available.notify_all()
770 continue
771 await self._data_available.wait()
772 buffer_index = chunk_number - self._discarded_chunks
773
774 result = self._chunks[buffer_index]
775
776 # free space for the producer when buffer is at capacity,
777 # but only if the producer is still running and needs space
778 if (
779 len(self._chunks) >= self.max_size_seconds
780 and not self._eof_received
781 and self._producer_task
782 and not self._producer_task.done()
783 ):
784 self._chunks.popleft()
785 self._discarded_chunks += 1
786 self._space_available.notify_all()
787
788 return result
789
790 async def _wait_for_space(self) -> None:
791 """Wait until buffer has space. Must be called while holding _lock."""
792 while len(self._chunks) >= self.max_size_seconds:
793 if self._cancelled:
794 return
795 await self._space_available.wait()
796
797 def _attach_producer_task(self, task: asyncio.Task[Any]) -> None:
798 """Attach a background task that fills the buffer."""
799 self._producer_task = task
800
801 def _on_producer_done(t: asyncio.Task[Any]) -> None:
802 if t.cancelled():
803 return
804 exc = t.exception()
805 if exc is not None and isinstance(exc, Exception):
806 self._producer_error = exc
807 loop = asyncio.get_running_loop()
808 task = loop.create_task(self._notify_on_producer_error())
809 self._background_tasks.add(task)
810 task.add_done_callback(self._background_tasks.discard)
811
812 task.add_done_callback(_on_producer_done)
813
814 if self._inactivity_task is None or self._inactivity_task.done():
815 self._last_access_time = time.time()
816 loop = asyncio.get_running_loop()
817 self._inactivity_task = loop.create_task(self._monitor_inactivity())
818
819 async def _monitor_inactivity(
820 self, inactivity_timeout: float = 300, check_interval: float = 30
821 ) -> None:
822 """
823 Clear the buffer once it has been inactive for inactivity_timeout seconds.
824
825 :param inactivity_timeout: Seconds without access before the buffer is released.
826 :param check_interval: Seconds between inactivity checks.
827 """
828 while True:
829 await asyncio.sleep(check_interval)
830 time_since_access = time.time() - self._last_access_time
831 # break on inactivity regardless of how many chunks remain: a rolling buffer
832 # that has drained to empty (e.g. an abandoned radio stream) must still release
833 # its resources and stop this monitor, otherwise the task loops forever
834 if time_since_access > inactivity_timeout:
835 LOGGER.log(
836 VERBOSE_LOG_LEVEL,
837 "AudioBuffer: No activity for %.1fs, clearing (%s chunks)",
838 time_since_access,
839 len(self._chunks),
840 )
841 break
842 await self.clear(cancel_inactivity_task=False)
843
844 async def _notify_on_producer_error(self) -> None:
845 """Notify waiting consumers that the producer has failed."""
846 async with self._lock:
847 if not self.ready.is_set():
848 self.ready.set()
849 self._data_available.notify_all()
850
851
852def _buffer_pcm_format(streamdetails: StreamDetails) -> AudioFormat:
853 """
854 Return the PCM format a buffer for these streamdetails holds.
855
856 The buffer stores decoded PCM, so it follows the audio that actually
857 arrives: ``audio_format`` may describe a source the provider decoded on our
858 behalf and can differ in depth or rate, in which case deriving the buffer
859 from it would resample or truncate real audio.
860
861 :param streamdetails: The stream the buffer is for.
862 """
863 arriving = arriving_audio_format(streamdetails)
864 return AudioFormat(
865 content_type=ContentType.from_bit_depth(arriving.bit_depth),
866 sample_rate=arriving.sample_rate,
867 bit_depth=arriving.bit_depth,
868 # buffer the stereo fold of a surround source, so audio analysis measures
869 # the same audio that is played back rather than the untouched surround mix
870 channels=min(arriving.channels, 2),
871 )
872