music-assistant-server

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