music-assistant-server

1"""
2Audio streaming helpers that interact with core controllers and providers.
3
4This module contains all audio stream acquisition and processing functions
5that need access to the MusicAssistant instance. Generic audio utilities
6that do not need controller interaction live in helpers/audio.py.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import logging
13import os
14import re
15import time
16from collections import deque
17from collections.abc import AsyncGenerator, Callable, Iterable
18from contextlib import aclosing, asynccontextmanager, nullcontext, suppress
19from dataclasses import dataclass
20from functools import partial
21from typing import TYPE_CHECKING, Any, cast
22from urllib.parse import urlparse
23from weakref import WeakValueDictionary
24
25import aiofiles
26import aiofiles.os
27import aiohttp
28import shortuuid
29from aiohttp import ClientConnectorSSLError, ClientResponseError, ClientTimeout
30from music_assistant_models.audio_processing import (
31    AudioDSPDetails,
32    AudioOutputDetails,
33    AudioQueueProcessing,
34)
35from music_assistant_models.dsp import (
36    AudioChannel,
37    ConvolutionFilter,
38    DSPConfig,
39    DSPFilter,
40    DSPState,
41)
42from music_assistant_models.enums import (
43    ContentType,
44    CrossfadeMode,
45    MediaType,
46    PlayerFeature,
47    ProviderFeature,
48    ProviderType,
49    StreamType,
50    VolumeNormalizationMode,
51)
52from music_assistant_models.errors import (
53    AudioError,
54    InvalidDataError,
55    MediaNotFoundError,
56    MusicAssistantError,
57    ProviderPermissionDenied,
58    ProviderUnavailableError,
59    QueueEmpty,
60    RetriesExhausted,
61)
62from music_assistant_models.media_items import Album, AudioFormat, Track
63from music_assistant_models.player_queue import PlayLogEntry
64from music_assistant_models.streamdetails import MultiPartPath, StreamMetadata
65
66from music_assistant.constants import (
67    CONF_CROSSFADE_DURATION,
68    CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES,
69    CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
70    CONF_FLOW_MODE_SAMPLE_RATE,
71    CONF_OUTPUT_CHANNELS,
72    CONF_PLAYER_QUEUES,
73    CONF_VALUE_DISABLED,
74    CONF_VALUE_ENABLED,
75    CONF_VOLUME_NORMALIZATION,
76    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
77    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
78    CONF_VOLUME_NORMALIZATION_RADIO,
79    CONF_VOLUME_NORMALIZATION_TARGET,
80    CONF_VOLUME_NORMALIZATION_TRACKS,
81    DSP_IRS_DIRNAME,
82    FLOW_MODE_SAMPLE_RATE_48000,
83    FLOW_MODE_SAMPLE_RATE_96000,
84    FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
85    FLOW_MODE_SAMPLE_RATE_HIGHEST,
86    FLOW_MODE_SAMPLE_RATE_SMART,
87    INTERNAL_PCM_FORMAT,
88    MASS_LOGGER_NAME,
89    STREAM_STALL_TIMEOUT,
90    STREAM_START_TIMEOUT,
91    VERBOSE_LOG_LEVEL,
92)
93from music_assistant.controllers.streams.audio_analysis import (
94    LOUDNESS_ANALYSIS_DOMAIN,
95)
96from music_assistant.controllers.streams.audio_buffer import AudioBuffer
97from music_assistant.controllers.streams.audio_processing import (
98    AudioOutputPlan,
99    get_normalization_details,
100)
101from music_assistant.controllers.streams.constants import (
102    CACHE_CATEGORY_RESOLVED_RADIO_URL,
103    CACHE_PROVIDER,
104    CONF_ALLOW_CROSSFADE_SAME_ALBUM,
105    DEFAULT_VOLUME_NORMALIZATION_MODE,
106    OUTCOME_ONLY_NORMALIZATION_MODES,
107    STREAM_SLOT_MATCH_TIMEOUT,
108    STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT,
109    STREAM_SLOT_WAIT_TIMEOUT,
110    STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY,
111    STREAMDETAILS_INBAND_TITLE_KEY,
112)
113from music_assistant.controllers.streams.ogg_handler import get_chained_ogg_stream
114from music_assistant.controllers.streams.smart_fades import SmartFadesMixer
115from music_assistant.controllers.streams.smart_fades.fades import SmartFade, StandardCrossFade
116from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
117from music_assistant.helpers import ssl as ssl_util
118from music_assistant.helpers.aiohttp_client import encoded_request_url
119from music_assistant.helpers.audio import (
120    HTTP_HEADERS,
121    HTTP_HEADERS_ICY,
122    arriving_audio_format,
123    audio_source_silence_keepalive,
124    build_concat_filelist,
125    calculate_content_length,
126    get_bit_rate,
127    get_normalization_mode,
128    get_parts_from_position,
129    is_grouping_preventing_dsp,
130    iter_pcm_slices,
131    parse_extinf_metadata,
132    realtime_pcm_pacer,
133    resample_pcm_audio,
134    resolve_output_player_ids,
135)
136from music_assistant.helpers.compare import compare_item_ids
137from music_assistant.helpers.dsp import ComplexFilter, filter_to_ffmpeg_params
138from music_assistant.helpers.ffmpeg import (
139    FFMpeg,
140    get_ffmpeg_overlay_stream,
141    get_ffmpeg_stream,
142)
143from music_assistant.helpers.named_pipe import read_named_pipe
144from music_assistant.helpers.playlists import (
145    HLS_CONTENT_TYPES,
146    PLAYLIST_CONTENT_TYPES,
147    PLAYLIST_READ_TIMEOUT,
148    IsHLSPlaylist,
149    PlaylistItem,
150    parse_m3u,
151    parse_playlist_data,
152    read_playlist_body,
153)
154from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
155from music_assistant.helpers.util import (
156    clean_stream_title,
157    detect_charset,
158    parse_quoted_stream_title,
159    parse_title_and_version,
160    remove_file,
161)
162from music_assistant.models.music_provider import MusicProvider, ProviderStreamLimitError
163
164if TYPE_CHECKING:
165    from music_assistant_models.media_items import ProviderMapping
166    from music_assistant_models.player_queue import PlayerQueue
167    from music_assistant_models.queue_item import QueueItem
168    from music_assistant_models.streamdetails import StreamDetails
169
170    from music_assistant.mass import MusicAssistant
171    from music_assistant.models.player import Player
172    from music_assistant.models.plugin import PluginProvider
173    from music_assistant.models.provider import Provider
174
175# ruff: noqa: PLR0915
176
177# Seconds of PCM at the start of a track that are yielded straight to the player,
178# never held back for a crossfade.
179WARMUP_DURATION = 8
180# Minimum overlap worth blending; below this the tail plays out and the boundary
181# is a hard cut. The configured mode picks the fade, this only decides whether a
182# boundary can carry one at all.
183MIN_CROSSFADE_DURATION = 3
184
185# Bounded wait for the fade-in prefetcher to release a stream at the handover. In
186# normal operation it returns on its next chunk; only a stalled source takes longer,
187# and then the flow stream is better off opening the track itself.
188PREFETCH_HANDOVER_TIMEOUT = 5.0
189
190# Bounded wait for a fade the outgoing stream is still mixing, when the incoming
191# item's own request arrives first. A speaker asks for the next url several seconds
192# before the current track ends, so without this wait a fade that is nearly ready is
193# thrown away. Kept short: the speaker is waiting on its first byte for this long.
194CROSSFADE_HANDOFF_WAIT = 5.0
195
196# Bounded wait at a boundary for a realtime incoming track to start delivering.
197# Its buffer only exists once its session produces audio, which happens around the
198# moment the outgoing track's audio ends; the wait trades a little of the player's
199# lead for the fade, and a source that never shows up loses only the fade.
200REALTIME_FADE_SOURCE_WAIT = 5.0
201
202# Chunk size for the realtime AudioSource path; small enough to keep ffmpeg→consumer
203# latency below ~50 ms while still amortising per-chunk overhead.
204AUDIO_SOURCE_CHUNK_SECONDS = 0.02
205
206# Terminal errors get_icy_radio_stream raises once a single mirror is exhausted; the
207# multi-mirror reader treats these as the signal to fail over to the next URL.
208RADIO_MIRROR_FAILOVER_ERRORS = (
209    MediaNotFoundError,
210    ProviderPermissionDenied,
211    ProviderUnavailableError,
212    RetriesExhausted,
213    InvalidDataError,
214)
215
216
217@dataclass
218class CrossfadeData:
219    """Data class to hold crossfade data."""
220
221    data: bytes
222    fade_in_media_duration: float
223    pcm_format: AudioFormat  # Format of the 'data' bytes (current/previous track's format)
224    queue_item_id: str
225    # Mode of the fade the 'data' bytes were blended with
226    crossfade_mode: CrossfadeMode = CrossfadeMode.DISABLED
227    # Offset for the fade_in track's elapsed time calculation, to account for crossfade duration and trim
228    elapsed_time_offset: float = 0.0
229    # Normalization mode the intro PCM was baked with, used to pin the next track's body to the same mode
230    normalization_mode: VolumeNormalizationMode | None = None
231
232
233def _snap_supported_rate_up(target: int, supported_sample_rates: list[int]) -> int:
234    """Snap target up, falling back to its highest supported divisor or the maximum."""
235    if target in supported_sample_rates:
236        return target
237    higher = [r for r in supported_sample_rates if r > target]
238    if higher:
239        return min(higher)
240    same_family = [r for r in supported_sample_rates if target % r == 0]
241    return max(same_family) if same_family else max(supported_sample_rates)
242
243
244def _snap_supported_rate_down(target: int, supported_sample_rates: list[int]) -> int:
245    """Snap target down to the highest supported rate <= target, falling back to min."""
246    if target in supported_sample_rates:
247        return target
248    lower = [r for r in supported_sample_rates if r < target]
249    return max(lower) if lower else min(supported_sample_rates)
250
251
252def overlay_active(queue: PlayerQueue) -> bool:
253    """Return True if the given queue has an audio overlay enabled and a source selected."""
254    return queue.overlay_enabled and queue.overlay_source is not None
255
256
257class _TailHold:
258    """
259    Grow a fade-out holdback out of what a source delivered ahead of playback.
260
261    Withholding a fixed window starves a source that delivers near playback pace
262    (a realtime session, a slow provider, a seek close to the end of a track). The
263    only audio that may be withheld is what the player is provably ahead by, past a
264    safety reserve, and only half of that, so its lead keeps growing too. Once the
265    source is done, the rest is resident and the full window is available.
266
267    A source read faster than playback banks that lead a little at a time, so it is
268    carried across a crossfaded boundary rather than remeasured per track: a queue
269    played from a source barely above playback pace only earns a full window several
270    tracks in, and starting from zero every track would never earn one at all. What
271    is carried is measured from audio emitted downstream, never from what a source
272    delivered - a fade consumes an overlap from both tracks and emits it once, so
273    the two are not the same number and only the first one the player ever hears.
274    """
275
276    # the player's supply must stay at least this far ahead of the wall clock
277    _LEAD_RESERVE_S = 3.0
278
279    def __init__(
280        self,
281        pcm_format: AudioFormat,
282        queue_item: QueueItem,
283        carried_lead: float = 0.0,
284        carried_at: float | None = None,
285    ) -> None:
286        """
287        Initialize the tracker for one track's stream.
288
289        :param pcm_format: PCM format of the stream's chunks.
290        :param queue_item: The item being streamed; its source buffer is resolved at
291            hold time, because opening the stream is what creates it - and a capacity
292            reselection can hand the item different details altogether.
293        :param carried_lead: Seconds of audio the player still held unplayed when this
294            stream took over, from the boundary this item was faded in across.
295        :param carried_at: When that lead was measured. Nothing is handed over between
296            then and this stream's first bytes while the player keeps playing, so the
297            carry is aged by that gap; defaults to now.
298        """
299        self._pcm_format = pcm_format
300        self._queue_item = queue_item
301        self._carried_lead = max(0.0, carried_lead)
302        self._carried_at = carried_at if carried_at is not None else asyncio.get_event_loop().time()
303        self._started: float | None = None
304        self._last_noted = 0.0
305        self._received_bytes = 0
306
307    # an arrival gap this long is a suspension (pause, sink hold), not elapsed
308    # listening; counting it would wrongly erase the banked surplus for good
309    _SUSPEND_FORGIVE_S = 5.0
310
311    def note_bytes(self, count: int) -> None:
312        """
313        Record stream bytes as they arrive (anchors the clock on the first ones).
314
315        :param count: Number of PCM bytes received.
316        """
317        now = asyncio.get_event_loop().time()
318        if self._started is None:
319            self._started = now
320            # the player drained while the boundary was being worked out and this
321            # stream was opening; what it still holds is that much less
322            self._carried_lead = max(0.0, self._carried_lead - (now - self._carried_at))
323        elif now - self._last_noted > self._SUSPEND_FORGIVE_S:
324            self._started += now - self._last_noted
325        self._last_noted = now
326        self._received_bytes += count
327
328    def hold_target(self, max_bytes: int, frame_size: int) -> int:
329        """
330        Return how many bytes of tail may currently be held back.
331
332        :param max_bytes: The full fade-out window (the cap).
333        :param frame_size: PCM frame size the target is aligned down to.
334        """
335        if self._started is None:
336            return 0
337        streamdetails = self._queue_item.streamdetails
338        audio_buffer = cast("AudioBuffer | None", streamdetails.buffer) if streamdetails else None
339        if audio_buffer is not None:
340            if audio_buffer.has_error:
341                # a failed source is skipped without a fade, so its remaining audio
342                # is better off played out than held back for one
343                return 0
344            if audio_buffer.eof:
345                # the source is done: everything left is resident, hold the full window
346                return max_bytes
347        elapsed = asyncio.get_event_loop().time() - self._started
348        received_seconds = self._received_bytes / self._pcm_format.pcm_sample_size
349        spare_seconds = self._carried_lead + received_seconds - elapsed - self._LEAD_RESERVE_S
350        surplus_bytes = int(max(0.0, spare_seconds) * self._pcm_format.pcm_sample_size) // 2
351        return min(max_bytes, surplus_bytes // frame_size * frame_size)
352
353    def banked_lead(self, emitted_bytes: int) -> float:
354        """
355        Return the lead to hand the next item, out of what was actually emitted.
356
357        Measured against audio yielded downstream rather than audio the source
358        delivered: a crossfade consumes an overlap from both tracks and emits it
359        once, and the planner trims what it does not use, so what a source handed
360        over is not what the player received. Reads as 0 until this stream
361        produced its first bytes.
362
363        :param emitted_bytes: PCM bytes this stream yielded downstream.
364        """
365        if self._started is None:
366            return 0.0
367        now = asyncio.get_event_loop().time()
368        emitted_seconds = emitted_bytes / self._pcm_format.pcm_sample_size
369        # note_bytes forgives a long arrival gap to keep the in-track holdback alive
370        # across a suspension. A stalled source looks the same as a paused player from
371        # here, and it is only safe to forgive the second, so the banked value pays for
372        # every second since audio last arrived.
373        idle = max(0.0, now - self._last_noted)
374        return max(0.0, self._carried_lead + emitted_seconds - (now - self._started) - idle)
375
376
377async def _incoming_overlap_stream(
378    collected: bytes,
379    stream: AsyncGenerator[bytes],
380    target_size: int,
381    overshoot: bytearray,
382    on_pulled: Callable[[int], None],
383) -> AsyncGenerator[bytes]:
384    """
385    Yield exactly the incoming track's overlap: what is in hand, then the live stream.
386
387    :param collected: Overlap bytes already collected when the mix starts.
388    :param stream: The incoming track's stream, read further as needed; bytes read
389        beyond the overlap are not lost (see ``overshoot``) and the stream itself
390        stays open for the track's body.
391    :param target_size: Exact number of overlap bytes to yield.
392    :param overshoot: Receives bytes read beyond the overlap (they open the body).
393    :param on_pulled: Called with the size of every chunk taken off the stream here,
394        as it is taken - these bypass the caller's own read loop.
395    """
396    taken = 0
397    if collected:
398        part = collected[:target_size]
399        overshoot.extend(collected[target_size:])
400        taken = len(part)
401        yield part
402    while taken < target_size:
403        try:
404            next_chunk = await anext(stream)
405        except StopAsyncIteration:
406            return
407        on_pulled(len(next_chunk))
408        remaining = target_size - taken
409        part = next_chunk[:remaining]
410        overshoot.extend(next_chunk[remaining:])
411        taken += len(part)
412        yield part
413
414
415class _IncomingFadePrefetcher:
416    """
417    Collect the incoming track's fade-in while the outgoing track's tail is held back.
418
419    A flow stream emits nothing while it gathers the audio a transition blends in, so the
420    player hears that wait as lost lead. Gathering it alongside the held-back tail instead
421    of after it keeps audio flowing right up to the transition. The collected audio and the
422    still-open stream are handed over together, so the track is decoded exactly once and the
423    seam is a plain continuation.
424    """
425
426    def __init__(
427        self, audio: StreamsAudio, pcm_format: AudioFormat, session_id: str | None
428    ) -> None:
429        """
430        Initialize the prefetcher for one flow stream.
431
432        :param audio: Audio sub-controller used to open the incoming track's stream.
433        :param pcm_format: Shared PCM format of the flow stream.
434        :param session_id: Queue session that owns the flow stream.
435        """
436        self._audio = audio
437        self._pcm_format = pcm_format
438        self._session_id = session_id
439        self._queue_item_id: str | None = None
440        self._streamdetails: StreamDetails | None = None
441        self._seek_position = 0
442        self._stream: AsyncGenerator[bytes] | None = None
443        self._chunks: deque[bytes] = deque()
444        self._target = 0
445        self._failed = False
446        self._collected_at_handover = 0
447        self._task: asyncio.Task[None] | None = None
448
449    def ensure_started(
450        self,
451        queue: PlayerQueue,
452        queue_item: QueueItem,
453        crossfade_mode: CrossfadeMode,
454        standard_crossfade_duration: int,
455    ) -> None:
456        """
457        Start collecting the next track's fade-in when it can be served from its buffer.
458
459        Does nothing when a prefetch is already running or the next track is not prepared
460        yet, so this is safe (and cheap) to call for every chunk of the outgoing track.
461
462        :param queue: Queue being streamed.
463        :param queue_item: Queue item whose tail is currently held back.
464        :param crossfade_mode: Crossfade mode selected for this queue item.
465        :param standard_crossfade_duration: Configured standard overlap in seconds.
466        """
467        if self._task is not None or crossfade_mode == CrossfadeMode.DISABLED:
468            return
469        next_item = self._audio.mass.player_queues.get_next_item(
470            queue.queue_id, queue_item.queue_item_id
471        )
472        if (
473            next_item is None
474            or next_item.queue_item_id == queue_item.queue_item_id
475            or next_item.media_type != MediaType.TRACK
476            or (streamdetails := next_item.streamdetails) is None
477            # without a duration the read below cannot be kept clear of the track's end
478            or not streamdetails.duration
479            or (audio_buffer := cast("AudioBuffer | None", streamdetails.buffer)) is None
480            or audio_buffer.has_error
481            or not audio_buffer.is_valid()
482        ):
483            return
484        overlap: float = (
485            SMART_CROSSFADE_DURATION
486            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
487            else standard_crossfade_duration
488        )
489        # never read a track to its end in the background: that would report it to its
490        # provider as streamed before a single second of it has reached the player.
491        # A track always plays at its own pace, so what is left of it after the seek is
492        # also what is left of the stream.
493        seek_position = int(streamdetails.seek_position)
494        overlap = min(overlap, (streamdetails.duration - seek_position) / 2)
495        if overlap <= 0:
496            return
497        self._target = int(self._pcm_format.pcm_sample_size * overlap)
498        self._queue_item_id = next_item.queue_item_id
499        self._streamdetails = streamdetails
500        self._seek_position = seek_position
501        self._chunks = deque()
502        self._stream = self._audio.get_queue_item_stream(
503            next_item,
504            pcm_format=self._pcm_format,
505            seek_position=seek_position,
506            playback_speed=cast("float", next_item.extra_attributes.get("playback_speed", 1.0)),
507            raise_on_error=False,
508            session_id=self._session_id,
509            prepared_buffer=audio_buffer,
510        )
511        self._task = asyncio.create_task(self._collect(self._stream, self._chunks))
512        self._audio.logger.debug(
513            "Prefetching %.0f seconds of %s while the tail of %s is held back",
514            overlap,
515            next_item.name,
516            queue_item.name,
517        )
518
519    async def take(self, queue_item: QueueItem, seek_position: int) -> AsyncGenerator[bytes] | None:
520        """
521        Hand over the prefetched stream for the given queue item.
522
523        Returns None unless the prefetch is for exactly the track and position the flow
524        stream is about to play and is still usable; the caller then opens the stream
525        itself, which also gives a broken source its chance to be re-resolved.
526
527        :param queue_item: Queue item the flow stream is about to play.
528        :param seek_position: Position in seconds the item is to be streamed from.
529        """
530        if self._task is None:
531            return None
532        if (
533            self._queue_item_id != queue_item.queue_item_id
534            or self._streamdetails is not queue_item.streamdetails
535            or self._seek_position != seek_position
536        ):
537            await self.close()
538            return None
539        # stop collecting: from here the flow stream reads the same generator itself
540        self._target = 0
541        try:
542            # the collector only sees the new target once its next chunk arrives, so a
543            # source that stalled would hold the handover; give up on it instead
544            await asyncio.wait_for(self._task, timeout=PREFETCH_HANDOVER_TIMEOUT)
545        except TimeoutError:
546            await self.close()
547            return None
548        if self._failed or (self._streamdetails is not None and self._streamdetails.stream_error):
549            await self.close()
550            return None
551        chunks, stream = self._chunks, self._stream
552        assert stream is not None
553        self._collected_at_handover = sum(len(chunk) for chunk in chunks)
554        self._reset()
555        return self._replay(chunks, stream)
556
557    @property
558    def collected_at_handover(self) -> int:
559        """Return how many bytes the last handover already had in hand."""
560        return self._collected_at_handover
561
562    async def close(self) -> None:
563        """Abandon a pending prefetch and release the incoming track's stream."""
564        task, stream = self._task, self._stream
565        # stop the collector before dropping the handles, so a task still running
566        # cannot write into the state a next prefetch starts from
567        self._target = 0
568        if task is not None:
569            task.cancel()
570            # gather consumes the collector's own cancellation but still lets a
571            # cancellation of this task through, so a stopped flow really stops
572            await asyncio.gather(task, return_exceptions=True)
573        if stream is not None:
574            await stream.aclose()
575        self._reset()
576
577    # --- Private methods ---
578
579    def _reset(self) -> None:
580        """Drop the handles of the current prefetch so a next one can start."""
581        self._task = None
582        self._stream = None
583        self._queue_item_id = None
584        self._streamdetails = None
585        self._seek_position = 0
586        self._chunks = deque()
587        self._target = 0
588        self._failed = False
589
590    async def _collect(self, stream: AsyncGenerator[bytes], chunks: deque[bytes]) -> None:
591        """Read the incoming track until the fade-in target is reached."""
592        collected = 0
593        try:
594            async for chunk in stream:
595                chunks.append(chunk)
596                collected += len(chunk)
597                # re-read the target every chunk: it drops to zero on handover
598                if collected >= self._target:
599                    return
600            # the target is kept clear of the track's end, so running out here means the
601            # source gave up early and the flow stream is better off opening it again
602            self._failed = True
603        except Exception as err:
604            # the flow stream opens the track itself rather than inheriting a dead stream
605            self._failed = True
606            self._audio.logger.warning("Failed to prefetch the incoming fade-in: %s", err)
607
608    async def _replay(
609        self, chunks: deque[bytes], stream: AsyncGenerator[bytes]
610    ) -> AsyncGenerator[bytes]:
611        """Yield the collected audio, then continue from the same stream."""
612        async with aclosing(stream):
613            while chunks:
614                yield chunks.popleft()
615            async for chunk in stream:
616                yield chunk
617
618
619class StreamsAudio:
620    """Audio stream acquisition and processing for the streams controller."""
621
622    def __init__(self, mass: MusicAssistant) -> None:
623        """
624        Initialize StreamsAudio.
625
626        :param mass: The MusicAssistant instance.
627        """
628        self.mass = mass
629        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.streams.audio")
630        self._crossfade_data: dict[str, CrossfadeData] = {}
631        # per queue: seconds of audio the player still held unplayed at the last
632        # crossfaded boundary, and when that was measured. The next item's holdback
633        # continues from the lead already earned, aged by the gap in between.
634        self._playback_lead: dict[str, tuple[float, float]] = {}
635        # queue_id -> (item the fade is being built for, set once its data is stored).
636        # The speaker asks for the next item's url before the outgoing stream is done,
637        # so without this the handoff is a race the fade loses.
638        self._crossfade_pending: dict[str, tuple[str, asyncio.Event]] = {}
639        self._smart_fades_mixer: SmartFadesMixer | None = None
640        # serializes buffer preparation per queue item, so concurrent callers share
641        # the single source (and the single capacity reselection) instead of racing
642        self._audio_buffer_locks: WeakValueDictionary[tuple[str, str], asyncio.Lock] = (
643            WeakValueDictionary()
644        )
645
646    def setup(self) -> None:
647        """Set up the audio sub-controller (called after all core controllers are created)."""
648        self._smart_fades_mixer = SmartFadesMixer(self.mass.streams)
649
650    @property
651    def smart_fades_mixer(self) -> SmartFadesMixer:
652        """Return the smart fades mixer."""
653        assert self._smart_fades_mixer is not None, "StreamsAudio.setup() not called"
654        return self._smart_fades_mixer
655
656    # --- Public methods ---
657
658    async def get_stream_details(
659        self,
660        queue_item: QueueItem,
661        seek_position: int = 0,
662        fade_in: bool = False,
663        prefer_album_loudness: bool = False,
664        excluded_provider_instances: set[str] | None = None,
665    ) -> StreamDetails:
666        """
667        Get streamdetails for the given QueueItem.
668
669        This is called just-in-time when a PlayerQueue wants a MediaItem to be played.
670        Do not try to request streamdetails too much in advance as this is expiring data.
671
672        :param queue_item: Queue item to resolve.
673        :param seek_position: Requested playback position in seconds.
674        :param fade_in: Whether playback should fade in.
675        :param prefer_album_loudness: Whether album loudness should be preferred.
676        :param excluded_provider_instances: Provider instances to skip during this selection.
677        """
678        mass = self.mass
679        streamdetails: StreamDetails | None = None
680        excluded_provider_instances = excluded_provider_instances or set()
681        time_start = time.time()
682        self.logger.debug("Getting streamdetails for %s", queue_item.uri)
683
684        if not queue_item.media_item and not queue_item.streamdetails:
685            # in case of a non-media item queue item, the streamdetails should already be provided
686            # this should not happen, but guard it just in case
687            raise MediaNotFoundError(
688                f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
689            )
690
691        if (
692            queue_item.streamdetails
693            # cached details of an excluded instance are exactly what we select away from
694            and queue_item.streamdetails.provider not in excluded_provider_instances
695            and (
696                # reuse if the buffer can serve this seek position (fast seek path)
697                (
698                    queue_item.streamdetails.buffer
699                    and queue_item.streamdetails.buffer.is_valid(int(seek_position * 1000))
700                )
701                # or reuse if streamdetails hasn't expired yet (new buffer will be created)
702                or (queue_item.streamdetails.created_at + queue_item.streamdetails.expiration)
703                > time.time()
704            )
705        ):
706            streamdetails = queue_item.streamdetails
707        else:
708            # need to (re)create streamdetails
709            # retrieve streamdetails from provider
710
711            media_item = queue_item.media_item
712            assert media_item is not None  # for type checking
713            preferred_providers: list[str] = []
714            if (
715                (pq_data := mass.player_queues.queue_data_or_none(queue_item.queue_id))
716                and pq_data.userid
717                and (playback_user := await mass.webserver.auth.get_user(pq_data.userid))
718                and playback_user.provider_filter
719            ):
720                # handle steering into user preferred providerinstance
721                preferred_providers = playback_user.provider_filter
722            candidates = self._get_streamdetail_candidates(
723                media_item.provider_mappings,
724                preferred_providers,
725                excluded_provider_instances,
726            )
727            streamdetails = await self._request_streamdetails(candidates, media_item.media_type)
728
729            if not streamdetails:
730                msg = f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
731                raise MediaNotFoundError(msg)
732
733            # work out how to handle radio stream
734            if (
735                streamdetails.stream_type in (StreamType.ICY, StreamType.HLS, StreamType.HTTP)
736                and streamdetails.media_type == MediaType.RADIO
737                and isinstance(streamdetails.path, str)
738            ):
739                resolved_url, stream_type = await self.resolve_radio_stream(streamdetails.path)
740                streamdetails.path = resolved_url
741                streamdetails.stream_type = stream_type
742                # Set up metadata monitoring callback for HLS radio streams, if not already set
743                if (
744                    stream_type == StreamType.HLS
745                    and not streamdetails.stream_metadata_update_callback
746                ):
747                    streamdetails.stream_metadata_update_callback = partial(
748                        self._update_hls_radio_metadata
749                    )
750                    streamdetails.stream_metadata_update_interval = 5
751
752        # providers report an unknown duration as either None or 0
753        if not streamdetails.duration:
754            if queue_item.media_item and queue_item.media_item.duration:
755                streamdetails.duration = queue_item.media_item.duration
756            elif queue_item.duration:
757                streamdetails.duration = queue_item.duration
758        if seek_position and not streamdetails.allow_seek:
759            self.logger.warning("seeking is not possible on this stream!")
760            seek_position = 0
761        elif seek_position and not streamdetails.duration:
762            self.logger.warning("seeking is not possible on duration-less streams!")
763            seek_position = 0
764
765        if streamdetails.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
766            # radio stations and live audio sources hand over their audio at playback pace
767            streamdetails.is_realtime = True
768
769        # set queue_id on the streamdetails so we know what is being streamed
770        streamdetails.queue_id = queue_item.queue_id
771        # handle skip/fade_in details
772        streamdetails.seek_position = seek_position
773        streamdetails.fade_in = fade_in
774
775        streamdetails.prefer_album_loudness = prefer_album_loudness
776        conf_volume_normalization_target = float(
777            mass.streams.get_config_value(CONF_VOLUME_NORMALIZATION_TARGET, return_type=int)
778        )
779        # guard against invalid volume normalization values
780        # range and default_value are guaranteed to be set for this constant
781        volume_range = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.range
782        assert volume_range is not None
783        if (
784            conf_volume_normalization_target < volume_range[0]
785            or conf_volume_normalization_target >= volume_range[1]
786        ):
787            default_val = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value
788            assert isinstance(default_val, (int, float))
789            conf_volume_normalization_target = float(default_val)
790            self.logger.warning(
791                "Invalid volume normalization target configured, resetting to default of %s LUFS",
792                CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value,
793            )
794        streamdetails.target_loudness = conf_volume_normalization_target
795        volume_normalization_enabled = (
796            mass.config.get_effective_player_queue_config_value(
797                streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
798            )
799            != CONF_VALUE_DISABLED
800        )
801        streamdetails.volume_normalization_mode = get_normalization_mode(
802            self._get_volume_normalization_preference(streamdetails),
803            volume_normalization_enabled,
804            streamdetails,
805            self.mass.streams.source_normalizes_audio(streamdetails),
806        )
807
808        self.logger.debug(
809            "Retrieved streamdetails for %s in %s milliseconds",
810            queue_item.uri,
811            int((time.time() - time_start) * 1000),
812        )
813        return streamdetails
814
815    async def get_audio_buffer(
816        self,
817        queue_item: QueueItem,
818        seek_position_ms: int = 0,
819        reason: str = "",
820        capacity_wait_timeout: float = STREAM_SLOT_PLAYBACK_WAIT_TIMEOUT,
821        allow_provider_match: bool = True,
822    ) -> AudioBuffer:
823        """
824        Return a ready AudioBuffer for the given queue item.
825
826        Compatible provider mappings are reselected while the owning provider has no free
827        source-stream slot. Other AudioErrors propagate as on a direct buffer request.
828
829        :param queue_item: Queue item whose source should be buffered.
830        :param seek_position_ms: Position in milliseconds to start from.
831        :param reason: Caller context for logging (e.g. 'prepare_next', 'streaming').
832        :param capacity_wait_timeout: Total seconds to spend waiting for source capacity.
833        :param allow_provider_match: Whether an on-demand cross-provider match may widen
834            the candidates when all are saturated.
835        :raises ProviderStreamLimitError: If no source slot becomes available within the budget.
836        """
837        lock_key = (queue_item.queue_id, queue_item.queue_item_id)
838        if (buffer_lock := self._audio_buffer_locks.get(lock_key)) is None:
839            buffer_lock = asyncio.Lock()
840            self._audio_buffer_locks[lock_key] = buffer_lock
841        async with buffer_lock:
842            return await self._get_audio_buffer(
843                queue_item, seek_position_ms, reason, capacity_wait_timeout, allow_provider_match
844            )
845
846    async def get_media_stream(
847        self,
848        streamdetails: StreamDetails,
849        pcm_format: AudioFormat,
850        seek_position: int = 0,
851        filter_params: list[str] | None = None,
852        chunk_seconds: float = 1.0,
853        source_wait_timeout: float | None = STREAM_SLOT_WAIT_TIMEOUT,
854    ) -> AsyncGenerator[bytes]:
855        """
856        Get audio stream for given media details as raw PCM.
857
858        :param streamdetails: Details of the stream to fetch.
859        :param pcm_format: Target PCM format the consumer expects.
860        :param seek_position: Seek offset in seconds (only honoured when the
861            source allows seeking; ignored for live AudioSources).
862        :param filter_params: Optional ffmpeg filter expressions.
863        :param chunk_seconds: Size of each yielded chunk in seconds of audio.
864            Defaults to 1 s for track-like sources; callers streaming live
865            AudioSources should pass a much smaller value (e.g. 0.02) to keep
866            end-to-end latency low.
867        :param source_wait_timeout: Maximum seconds to wait for a free source-stream slot
868            on the providing music provider, or None to wait without a timeout.
869        :raises ProviderStreamLimitError: If the provider has no free slot within the timeout.
870        """
871        media_stream = self._get_media_stream(
872            streamdetails,
873            pcm_format,
874            seek_position,
875            filter_params,
876            chunk_seconds,
877        )
878        # resolve the exact owning instance (even when flagged unavailable) so the
879        # slot is charged to the account that issued the streamdetails
880        provider = self.mass.get_provider(streamdetails.provider, return_unavailable=True)
881        stream_slot = (
882            provider.acquire_stream_slot(source_wait_timeout)
883            if isinstance(provider, MusicProvider)
884            else nullcontext()
885        )
886        async with stream_slot, aclosing(media_stream):
887            async for chunk in media_stream:
888                yield chunk
889
890    async def resolve_radio_stream(self, url: str) -> tuple[str, StreamType]:
891        """
892        Resolve a streaming radio URL.
893
894        Unwraps playlists and determines stream type (ICY, HLS, SHOUTCAST, IN_BAND, HTTP).
895
896        :param url: Radio stream URL to resolve
897        """
898        mass = self.mass
899        if cache := await mass.cache.get(
900            key=url, provider=CACHE_PROVIDER, category=CACHE_CATEGORY_RESOLVED_RADIO_URL
901        ):
902            if TYPE_CHECKING:
903                cache = cast("tuple[str, str]", cache)
904            return (cache[0], StreamType(cache[1]))
905
906        stream_type = StreamType.HTTP
907        timeout = ClientTimeout(total=None, connect=10, sock_read=5)
908        playlist_data: bytes | None = None
909        playlist_charset: str | None = None
910
911        try:
912            async with self._connect_radio_stream(
913                url, headers=HTTP_HEADERS_ICY, allow_redirects=True, timeout=timeout
914            ) as resp:
915                headers = resp.headers
916                resp.raise_for_status()
917                if not resp.headers:
918                    raise InvalidDataError("no headers found")
919                # media types are case insensitive, the comparisons below are all lower case
920                content_type = headers.get("content-type", "").lower()
921                # a server declaring HLS settles it: a media playlist is free to carry none
922                # of the tags the parser recognises an HLS playlist by
923                is_hls = any(hls_type in content_type for hls_type in HLS_CONTENT_TYPES)
924                if not is_hls and (
925                    url.endswith((".m3u", ".m3u8", ".pls"))
926                    or ".m3u?" in url
927                    or ".m3u8?" in url
928                    or ".pls?" in url
929                    or any(
930                        playlist_type in content_type for playlist_type in PLAYLIST_CONTENT_TYPES
931                    )
932                ):
933                    # take the playlist from this very response: a separate request would
934                    # go out with another user agent and stricter TLS than the rest of the
935                    # radio paths, so a host could answer it differently
936                    try:
937                        # the probe has no total timeout, so bound the body on its own:
938                        # a server trickling bytes would otherwise stall resolving for hours
939                        async with asyncio.timeout(PLAYLIST_READ_TIMEOUT):
940                            playlist_data = await read_playlist_body(resp.content)
941                    except aiohttp.ClientError as err:
942                        # the endpoint answered as a playlist, so a truncated body is a bad
943                        # playlist - not a reason to fall back to streaming the URL directly
944                        raise InvalidDataError(f"Error while fetching playlist {url}") from err
945                    playlist_charset = resp.charset
946
947            if headers.get("icy-metaint") is not None:
948                stream_type = StreamType.ICY
949            elif is_hls:
950                stream_type = StreamType.HLS
951            elif content_type in ("application/ogg", "audio/ogg"):
952                # Ogg streams (Opus/Vorbis) have in-band metadata via Vorbis comments
953                stream_type = StreamType.IN_BAND
954
955            if playlist_data is not None:
956                try:
957                    substreams = await parse_playlist_data(url, playlist_data, playlist_charset)
958                    if not any(x for x in substreams if x.length):
959                        for line in substreams:
960                            if not line.is_url:
961                                continue
962                            return await self.resolve_radio_stream(line.path)
963                        raise InvalidDataError("No content found in playlist")
964                except IsHLSPlaylist:
965                    stream_type = StreamType.HLS
966
967        except TimeoutError as err:
968            self.logger.warning("Timeout while parsing radio URL %s", url)
969            raise InvalidDataError(f"Timeout connecting to {url}") from err
970
971        except aiohttp.ClientResponseError as err:
972            if err.status == 404:
973                raise MediaNotFoundError(f"Radio stream not found: {url}") from err
974            if err.status == 403:
975                raise InvalidDataError(f"Access denied to radio stream: {url}") from err
976            if err.status >= 500:
977                raise InvalidDataError(
978                    f"Radio stream server error (HTTP {err.status}): {url}"
979                ) from err
980            if err.status == 400:
981                # 400 errors might be from legacy Shoutcast servers
982                return await self._handle_client_error_for_radio_stream(url, err, stream_type)
983            raise InvalidDataError(f"HTTP error {err.status} from {url}") from err
984
985        except aiohttp.ClientError as err:
986            return await self._handle_client_error_for_radio_stream(url, err, stream_type)
987
988        return await self._cache_radio_result(url, stream_type)
989
990    async def get_icy_radio_stream(
991        self, url: str, streamdetails: StreamDetails
992    ) -> AsyncGenerator[bytes]:
993        """
994        Stream radio audio with ICY metadata support, reconnecting on disconnect.
995
996        Requires icy-metaint header support. Stream type should be validated
997        by resolve_radio_stream() before calling this function.
998
999        :param url: Radio stream URL
1000        :param streamdetails: StreamDetails to update with metadata
1001        """
1002        self.logger.debug("Start streaming radio with ICY metadata from url %s", url)
1003        timeout = ClientTimeout(total=0, connect=30, sock_read=5 * 60)
1004        # Budget for *consecutive* reconnects that delivered no audio. A connection
1005        # that actually streamed data resets it, so a healthy long-running stream can
1006        # reconnect indefinitely while a dead/looping one bails out instead of spinning.
1007        failed_reconnects = 0
1008        max_failed_reconnects = 25
1009
1010        while True:
1011            streamed_data = False
1012            try:
1013                async with self._connect_radio_stream(
1014                    url, allow_redirects=True, headers=HTTP_HEADERS_ICY, timeout=timeout
1015                ) as resp:
1016                    # surface a non-200 (e.g. on reconnect) as a ClientResponseError so the
1017                    # terminal/HTTP handling below applies instead of failing on the header
1018                    resp.raise_for_status()
1019                    meta_int_str = resp.headers.get("icy-metaint")
1020                    if not meta_int_str:
1021                        raise InvalidDataError(f"No icy-metaint header for radio stream: {url}")
1022                    try:
1023                        meta_int = int(meta_int_str)
1024                    except ValueError as err:
1025                        raise InvalidDataError(
1026                            f"Invalid icy-metaint value for radio stream: {url}"
1027                        ) from err
1028                    if meta_int <= 0:
1029                        raise InvalidDataError(f"Invalid icy-metaint value for radio stream: {url}")
1030                    # readexactly raises IncompleteReadError when the server closes the
1031                    # connection mid-frame; that (and the network errors below) drops us
1032                    # out to the reconnect handler so a live stream survives the blip.
1033                    while True:
1034                        chunk = await resp.content.readexactly(meta_int)
1035                        streamed_data = True
1036                        yield chunk
1037                        meta_byte = await resp.content.readexactly(1)
1038                        if meta_byte == b"\x00":
1039                            continue
1040                        meta_length = ord(meta_byte) * 16
1041                        meta_data = await resp.content.readexactly(meta_length)
1042                        self._parse_icy_metadata(meta_data, streamdetails)
1043            except asyncio.CancelledError:
1044                self.logger.debug("ICY radio stream cancelled for %s", url)
1045                raise
1046            except aiohttp.ClientResponseError as err:
1047                if err.status == 404:
1048                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
1049                if err.status == 403:
1050                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
1051                raise ProviderUnavailableError(
1052                    f"Radio stream returned HTTP {err.status}: {err}"
1053                ) from err
1054            except (
1055                asyncio.IncompleteReadError,
1056                aiohttp.ClientConnectionError,
1057                aiohttp.ClientPayloadError,
1058                aiohttp.ServerDisconnectedError,
1059            ) as err:
1060                if streamed_data:
1061                    # a healthy session that dropped - reconnect without spending budget
1062                    failed_reconnects = 0
1063                    self.logger.debug("ICY radio stream dropped, reconnecting: %s", err)
1064                else:
1065                    failed_reconnects += 1
1066                    if failed_reconnects > max_failed_reconnects:
1067                        raise RetriesExhausted(
1068                            f"ICY radio stream failed after {max_failed_reconnects} "
1069                            f"reconnects without data: {err}"
1070                        ) from err
1071                    self.logger.warning(
1072                        "ICY radio stream reconnect produced no data (%d/%d): %s",
1073                        failed_reconnects,
1074                        max_failed_reconnects,
1075                        err,
1076                    )
1077                await asyncio.sleep(0.5)
1078
1079    async def get_reconnecting_icy_radio_stream(
1080        self, url: str | list[MultiPartPath], streamdetails: StreamDetails
1081    ) -> AsyncGenerator[bytes]:
1082        """
1083        Yield ICY radio audio with metadata, failing over across mirror URLs.
1084
1085        A single URL is delegated to :meth:`get_icy_radio_stream`, which already reconnects
1086        on disconnect. Multiple URLs are treated as interchangeable mirrors and tried in turn;
1087        a mirror that delivers audio resets the failover budget, so a healthy mirror keeps
1088        streaming while a set of unreachable mirrors raises the last error instead of spinning.
1089
1090        :param url: One stream URL, or a list of mirror URLs to fail over between.
1091        :param streamdetails: StreamDetails to update with metadata.
1092        """
1093        urls = self._normalize_reconnecting_urls(url)
1094        if len(urls) == 1:
1095            async for chunk in self.get_icy_radio_stream(urls[0], streamdetails):
1096                yield chunk
1097            return
1098
1099        url_index = 0
1100        failed_rotations = 0
1101        max_failed_rotations = len(urls) * 2
1102        last_err: MusicAssistantError | None = None
1103        while failed_rotations <= max_failed_rotations:
1104            current_url = urls[url_index % len(urls)]
1105            url_index += 1
1106            delivered_audio = False
1107            try:
1108                async for chunk in self.get_icy_radio_stream(current_url, streamdetails):
1109                    delivered_audio = True
1110                    failed_rotations = 0
1111                    # release the previous failure while healthy: it pins the full
1112                    # exception traceback (with frames) for the lifetime of the stream
1113                    last_err = None
1114                    yield chunk
1115                return
1116            except RADIO_MIRROR_FAILOVER_ERRORS as err:
1117                last_err = err
1118                if not delivered_audio:
1119                    failed_rotations += 1
1120                self.logger.warning(
1121                    "ICY radio mirror %s failed, trying next url (%d/%d): %s",
1122                    current_url,
1123                    failed_rotations,
1124                    max_failed_rotations,
1125                    err,
1126                )
1127        if last_err is not None:
1128            raise last_err
1129
1130    async def get_reconnecting_radio_stream(self, url: str) -> AsyncGenerator[bytes]:
1131        """
1132        Yield continuous radio stream data, automatically reconnecting on disconnect.
1133
1134        :param url: URL of the radio stream.
1135        """
1136        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
1137        reconnect_count = 0
1138        max_reconnects = 1000  # Allow many reconnects for long-running radio
1139
1140        while reconnect_count <= max_reconnects:
1141            try:
1142                async with self._connect_radio_stream(
1143                    url, allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
1144                ) as resp:
1145                    chunk_count = 0
1146                    async for chunk in resp.content.iter_any():
1147                        chunk_count += 1
1148                        yield chunk
1149
1150                    # Connection closed normally - reconnect
1151                    self.logger.debug(
1152                        "Radio stream connection closed after %d chunks, reconnecting... "
1153                        "(reconnect #%d)",
1154                        chunk_count,
1155                        reconnect_count,
1156                    )
1157                    reconnect_count += 1
1158                    await asyncio.sleep(0.1)  # Brief delay before reconnect
1159
1160            except asyncio.CancelledError:
1161                self.logger.debug("Radio stream cancelled for %s", url)
1162                raise
1163            except (
1164                aiohttp.ClientConnectionError,
1165                aiohttp.ClientPayloadError,
1166                aiohttp.ServerDisconnectedError,
1167            ) as err:
1168                # Transient network errors - retry
1169                self.logger.warning("Radio stream error (reconnect #%d): %s", reconnect_count, err)
1170                reconnect_count += 1
1171                if reconnect_count > max_reconnects:
1172                    raise RetriesExhausted(
1173                        f"Radio stream failed after {max_reconnects} reconnects: {err}"
1174                    ) from err
1175                await asyncio.sleep(0.5)
1176            except aiohttp.ClientResponseError as err:
1177                if err.status == 404:
1178                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
1179                if err.status == 403:
1180                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
1181                # Other HTTP errors (5xx etc) - could be temporary
1182                raise ProviderUnavailableError(
1183                    f"Radio stream returned HTTP {err.status}: {err}"
1184                ) from err
1185
1186        self.logger.warning("Radio stream reached max reconnects (%d) for %s", max_reconnects, url)
1187
1188    async def get_hls_substream(self, url: str) -> PlaylistItem:
1189        """Select the (highest quality) HLS substream for given HLS playlist/URL."""
1190        mass = self.mass
1191        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
1192        # fetch master playlist and select (best) child playlist
1193        # https://datatracker.ietf.org/doc/html/draft-pantos-http-live-streaming-19#section-10
1194        async with mass.http_session_no_ssl.get(
1195            encoded_request_url(url), allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
1196        ) as resp:
1197            resp.raise_for_status()
1198            raw_data = await resp.read()
1199            encoding = await detect_charset(raw_data, preferred=resp.charset)
1200            master_m3u_data = raw_data.decode(encoding, errors="replace")
1201        substreams = parse_m3u(master_m3u_data)
1202        # There is a chance that we did not get a master playlist with subplaylists
1203        # but just a single master/sub playlist with the actual audio stream(s)
1204        # so we need to detect if the playlist child's contain audio streams or
1205        # sub-playlists.
1206        if any(
1207            x
1208            for x in substreams
1209            if (x.length or x.path.endswith((".mp4", ".aac")))
1210            and not x.path.endswith((".m3u", ".m3u8"))
1211        ):
1212            return PlaylistItem(path=url, key=substreams[0].key)
1213        # sort substreams on best quality (highest bandwidth) when available
1214        if any(x for x in substreams if x.stream_info):
1215            substreams.sort(
1216                key=lambda x: int(
1217                    x.stream_info.get("BANDWIDTH", "0") if x.stream_info is not None else 0
1218                ),
1219                reverse=True,
1220            )
1221        substream = substreams[0]
1222        if not substream.path.startswith("http"):
1223            # path is relative, stitch it together
1224            base_path = url.rsplit("/", 1)[0]
1225            substream.path = base_path + "/" + substream.path
1226        return substream
1227
1228    async def get_multi_file_stream(
1229        self,
1230        streamdetails: StreamDetails,
1231        seek_position: int = 0,
1232    ) -> AsyncGenerator[bytes]:
1233        """
1234        Return audio stream for a concatenation of multiple files.
1235
1236        Arguments:
1237        seek_position: The position to seek to in seconds
1238        """
1239        if not isinstance(streamdetails.path, list):
1240            raise InvalidDataError("Multi-file streamdetails requires a list of MultiPartPath")
1241        parts, seek_position = get_parts_from_position(streamdetails.path, seek_position)
1242        files_list = [part.path for part in parts]
1243
1244        # concat input files
1245        temp_file = f"/tmp/{shortuuid.random(20)}.txt"  # noqa: S108
1246        async with aiofiles.open(temp_file, "w") as f:
1247            await f.write(build_concat_filelist(files_list))
1248
1249        try:
1250            async for chunk in get_ffmpeg_stream(
1251                audio_input=temp_file,
1252                input_format=streamdetails.audio_format,
1253                output_format=AudioFormat(
1254                    content_type=ContentType.NUT,
1255                    sample_rate=streamdetails.audio_format.sample_rate,
1256                    bit_depth=streamdetails.audio_format.bit_depth,
1257                    channels=streamdetails.audio_format.channels,
1258                ),
1259                extra_input_args=[
1260                    "-safe",
1261                    "0",
1262                    "-f",
1263                    "concat",
1264                    "-i",
1265                    temp_file,
1266                    "-ss",
1267                    str(seek_position),
1268                ],
1269            ):
1270                yield chunk
1271        finally:
1272            await remove_file(temp_file)
1273
1274    def get_player_output_plan(
1275        self,
1276        player_id: str,
1277        input_format: AudioFormat,
1278        output_format: AudioFormat,
1279        *,
1280        shared_player_ids: Iterable[str] | None = None,
1281        handoff_format: AudioFormat | None = None,
1282        queue_id: str | None = None,
1283        session_id: str | None = None,
1284        queue_item_id: str | None = None,
1285    ) -> AudioOutputPlan:
1286        """
1287        Return executable filters and matching output details for a player.
1288
1289        :param player_id: Destination player identifier.
1290        :param input_format: PCM format entering player-specific processing.
1291        :param output_format: Furthest downstream output format known to the server.
1292        :param shared_player_ids: Additional players receiving this identical output path.
1293            An empty iterable marks a path that can gain shared destinations later.
1294        :param handoff_format: Earlier provider handoff format when it differs.
1295        :param queue_id: Explicit queue identifier for the processing snapshot.
1296        :param session_id: Explicit queue session identifier for the processing snapshot.
1297        :param queue_item_id: Queue item for a single-item output path.
1298        """
1299        filter_params: list[str | ComplexFilter] = []
1300        player = self.mass.players.get_player(player_id)
1301        destination_player_id = (
1302            player.protocol_parent_id if player and player.protocol_parent_id else player_id
1303        )
1304        resolved_shared_player_ids = (
1305            resolve_output_player_ids(self.mass, shared_player_ids) - {destination_player_id}
1306            if shared_player_ids is not None
1307            else None
1308        )
1309        destination_player_ids = {destination_player_id, *(resolved_shared_player_ids or ())}
1310        if player:
1311            dsp_config_id = self._resolve_player_dsp_config_id(player)
1312            dsp = self._resolve_player_dsp_config(player)
1313            configured_dsp = self.mass.config.get_player_dsp_config(dsp_config_id)
1314            if configured_dsp.enabled and not dsp.enabled and is_grouping_preventing_dsp(player):
1315                dsp_state = DSPState.DISABLED_BY_UNSUPPORTED_GROUP
1316            else:
1317                dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1318        else:
1319            dsp_config_id = player_id
1320            dsp = self.mass.config.get_player_dsp_config(player_id)
1321            dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1322
1323        enabled_filters = [dsp_filter for dsp_filter in dsp.filters if dsp_filter.enabled]
1324        # a neutral filter (0 dB gain, centered balance) emits no params; exclude
1325        # it so it is not reported as an active, non-bit-perfect stage
1326        effective_filters: list[DSPFilter] = []
1327        if dsp.enabled:
1328            if dsp.input_gain != 0:
1329                filter_params.append(f"volume={dsp.input_gain}dB")
1330            ir_dir = os.path.join(self.mass.storage_path, DSP_IRS_DIRNAME)
1331            known_ir_ids = {record["ir_id"] for record in self.mass.config.get_dsp_irs()}
1332            for dsp_filter in enabled_filters:
1333                if isinstance(dsp_filter, ConvolutionFilter) and dsp_filter.ir_id:
1334                    # ffmpeg fails to open the graph if the impulse response file is gone,
1335                    # which costs the player all audio, so drop the filter instead
1336                    if dsp_filter.ir_id not in known_ir_ids:
1337                        self.logger.warning(
1338                            "Skipping the convolution filter of player %s: "
1339                            "impulse response %s is not stored",
1340                            player_id,
1341                            dsp_filter.ir_id,
1342                        )
1343                        continue
1344                params = filter_to_ffmpeg_params(dsp_filter, input_format, ir_dir=ir_dir)
1345                if not params:
1346                    continue
1347                filter_params.extend(params)
1348                effective_filters.append(dsp_filter)
1349            if dsp.output_gain != 0:
1350                filter_params.append(f"volume={dsp.output_gain}dB")
1351
1352        channel_value = self._get_output_channels(player, player_id)
1353        source_channel = None
1354        channel_mix = ""
1355        # a single channel source is already the downmix and holds no FL/FR to select
1356        # from, where a pan would silently resolve every gain to zero
1357        if input_format.channels > 1:
1358            if channel_value == "left":
1359                source_channel = AudioChannel.FL
1360                channel_mix = "FL"
1361            elif channel_value == "right":
1362                source_channel = AudioChannel.FR
1363                channel_mix = "FR"
1364            elif channel_value == "mono":
1365                # both source channels feed the downmix, so report ALL to keep the
1366                # output from ever being presented as bit perfect
1367                source_channel = AudioChannel.ALL
1368                channel_mix = "0.5*FL+0.5*FR"
1369        if channel_mix:
1370            # the pan runs in the command that emits the handoff format, and it feeds
1371            # every channel of it explicitly: leaving ffmpeg to upmix from a single
1372            # channel costs 3 dB through its rematrix
1373            if (handoff_format or output_format).channels == 1:
1374                filter_params.append(f"pan=mono|c0={channel_mix}")
1375            else:
1376                filter_params.append(f"pan=stereo|c0={channel_mix}|c1={channel_mix}")
1377
1378        output_details = AudioOutputDetails(
1379            player_ids=sorted(destination_player_ids),
1380            dsp=AudioDSPDetails(
1381                state=dsp_state,
1382                input_gain=dsp.input_gain if dsp.enabled else 0.0,
1383                filters=effective_filters,
1384                output_gain=dsp.output_gain if dsp.enabled else 0.0,
1385                preset_id=dsp.preset_id,
1386            ),
1387            source_channel=source_channel,
1388            output_format=output_format,
1389        )
1390        output_plan = AudioOutputPlan(
1391            filter_params=filter_params,
1392            output_details=output_details,
1393            input_format=input_format,
1394            handoff_format=handoff_format,
1395            dsp_config_id=dsp_config_id,
1396        )
1397        if queue_id is not None and session_id is not None:
1398            self.mass.streams.audio_processing.update_output(
1399                destination_player_id,
1400                output_plan,
1401                shared_player_ids=resolved_shared_player_ids,
1402                queue_id=queue_id,
1403                session_id=session_id,
1404                queue_item_id=queue_item_id,
1405            )
1406        self.logger.log(
1407            VERBOSE_LOG_LEVEL,
1408            "Generated ffmpeg params for player %s: %s",
1409            player_id,
1410            filter_params,
1411        )
1412        return output_plan
1413
1414    async def get_output_format(
1415        self,
1416        output_format_str: str,
1417        player: Player,
1418        content_sample_rate: int,
1419        content_bit_depth: int,
1420        media_type: MediaType = MediaType.UNKNOWN,
1421    ) -> AudioFormat:
1422        """Parse (player specific) output format details for given format string."""
1423        content_type: ContentType = ContentType.try_parse(output_format_str)
1424        player_supported_rates = player.get_supported_sample_rates()
1425        supported_sample_rates = [sr for sr, _ in player_supported_rates]
1426        if content_sample_rate in supported_sample_rates:
1427            output_sample_rate = content_sample_rate
1428        else:
1429            output_sample_rate = max(supported_sample_rates)
1430        # only consider bit depths that are actually paired with the chosen sample rate
1431        bit_depths_for_rate = [
1432            bd for (sr, bd) in player_supported_rates if sr == output_sample_rate
1433        ]
1434        output_bit_depth = min(content_bit_depth, max(bit_depths_for_rate, default=16))
1435
1436        if not content_type.is_lossless():
1437            # no point in having a higher bit depth for lossy formats
1438            output_bit_depth = 16
1439            output_sample_rate = min(48000, output_sample_rate)
1440        if media_type not in (MediaType.TRACK, MediaType.AUDIO_SOURCE, MediaType.FLOW_STREAM):
1441            # no point in having a higher bit depth for non-track media types (e.g. TTS, radio)
1442            output_bit_depth = min(output_bit_depth, 16)
1443        if output_format_str == "pcm":
1444            content_type = ContentType.from_bit_depth(output_bit_depth)
1445
1446        output_channels_str = self._get_output_channels(player, player.player_id)
1447        fmt = AudioFormat(
1448            content_type=content_type,
1449            sample_rate=output_sample_rate,
1450            bit_depth=output_bit_depth,
1451            channels=1 if output_channels_str != "stereo" else 2,
1452        )
1453        fmt.bit_rate = get_bit_rate(fmt)
1454        return fmt
1455
1456    async def select_pcm_format(
1457        self,
1458        player: Player,
1459        streamdetails: StreamDetails,
1460        crossfade_enabled: bool,
1461        overlay_active: bool = False,
1462    ) -> AudioFormat:
1463        """
1464        Select the internal PCM format for streaming a single queue item.
1465
1466        Used by the per-item (non-flow) stream path. The sample rate is the highest
1467        rate the player supports that is <= the source rate, so the source is never
1468        upsampled. The bit depth follows the source unless audio processing
1469        (crossfade, volume normalization, DSP) is active — those need F32 headroom
1470        to avoid clipping/precision loss. Surround sources are folded down to stereo.
1471        Realtime AudioSource items skip all processing and get a pure passthrough
1472        format (source rate/bit depth when the player supports them).
1473
1474        :param player: The player requesting the stream.
1475        :param streamdetails: Stream details for the current item.
1476        :param crossfade_enabled: Whether crossfade is enabled for this stream.
1477        :param overlay_active: Whether an audio overlay will be mixed into this stream.
1478        """
1479        if streamdetails.media_type == MediaType.AUDIO_SOURCE:
1480            return self._select_audio_source_pcm_format(player, streamdetails)
1481        supported_sample_rates = [sr for sr, _ in player.get_supported_sample_rates()]
1482        # snap-down: pick the highest supported rate <= source. when the source rate
1483        # is below every supported rate (e.g. 22 kHz content on a 44.1k-only player),
1484        # fall back to the lowest supported rate instead of a hardcoded 48 kHz that
1485        # the player may not actually support.
1486        output_sample_rate = max(
1487            (r for r in supported_sample_rates if r <= streamdetails.audio_format.sample_rate),
1488            default=min(supported_sample_rates),
1489        )
1490        content_type, bit_depth = self._pick_pcm_bit_depth(
1491            (player,),
1492            streamdetails,
1493            crossfade_enabled,
1494            overlay_active,
1495        )
1496        pcm_format = AudioFormat(
1497            sample_rate=output_sample_rate,
1498            content_type=content_type,
1499            bit_depth=bit_depth,
1500            # fold surround sources down to stereo right at the decode step: no
1501            # output format carries more than two channels, so a wider PCM format
1502            # only makes every bytes-to-seconds sum on the stream come out short
1503            channels=min(streamdetails.audio_format.channels, 2),
1504        )
1505        if crossfade_enabled or overlay_active:
1506            pcm_format.channels = 2
1507        return pcm_format
1508
1509    async def select_flow_pcm_format(
1510        self,
1511        player: Player,
1512        start_streamdetails: StreamDetails | None = None,
1513        crossfade_enabled: bool = False,
1514        overlay_active: bool = False,
1515        fallback_sample_rate: int | None = None,
1516        output_players: Iterable[Player] | None = None,
1517    ) -> AudioFormat:
1518        """
1519        Select the internal PCM format for a Queue Flow Mode stream.
1520
1521        Used by the gapless flow path that stitches multiple queue items into one
1522        continuous PCM stream. The sample rate is driven by the player's
1523        ``CONF_FLOW_MODE_SAMPLE_RATE`` setting (smart/bit_perfect/48k/96k/highest)
1524        — for the anchored modes it follows the first track's rate, for the fixed
1525        modes it snaps to the configured rate. The bit depth follows the first
1526        track's source unless audio processing is active (then F32 for headroom),
1527        avoiding an unnecessary up-convert to 32-bit when none of the consumers
1528        will benefit from it. When the first item is a realtime AudioSource, the
1529        flow mode config is ignored and a pure passthrough format is used so the
1530        source audio is delivered with minimum overhead and latency.
1531
1532        :param player: The player the flow stream is being prepared for.
1533        :param start_streamdetails: Stream details of the first track in the flow.
1534            Required for the anchored modes ('smart' / 'bit_perfect') and for the
1535            bit-depth optimization. May be omitted for the fixed-rate modes — when
1536            omitted the bit depth defaults to F32.
1537        :param crossfade_enabled: Whether the queue will use crossfade transitions.
1538        :param overlay_active: Whether an audio overlay will be mixed into the stream.
1539        :param fallback_sample_rate: Preferred rate when the first item format is unknown.
1540        :param output_players: All players consuming the shared PCM stream. Their common
1541            sample rates and processing requirements determine the session format.
1542        """
1543        players = tuple(output_players) if output_players is not None else (player,)
1544        if not players:
1545            raise AudioError("At least one output player is required")
1546        supported_sample_rates = sorted(
1547            set.intersection(
1548                *(
1549                    {sample_rate for sample_rate, _ in item.get_supported_sample_rates()}
1550                    for item in players
1551                )
1552            )
1553        )
1554        if not supported_sample_rates:
1555            raise AudioError("Output players do not share a supported sample rate")
1556        if start_streamdetails is not None and (
1557            start_streamdetails.media_type == MediaType.AUDIO_SOURCE
1558        ):
1559            return self._select_audio_source_pcm_format(
1560                player,
1561                start_streamdetails,
1562                supported_sample_rates=supported_sample_rates,
1563            )
1564        flow_mode_conf = cast(
1565            "str",
1566            player.config.get_value(CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART),
1567        )
1568
1569        if flow_mode_conf == FLOW_MODE_SAMPLE_RATE_HIGHEST:
1570            output_sample_rate = max(supported_sample_rates)
1571        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_48000:
1572            # for the fixed-rate modes, the user picked a specific bandwidth/quality
1573            # ceiling; prefer the highest supported rate <= target
1574            output_sample_rate = _snap_supported_rate_down(48000, supported_sample_rates)
1575        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_96000:
1576            output_sample_rate = _snap_supported_rate_down(96000, supported_sample_rates)
1577        else:
1578            # smart or bit_perfect (default): anchor the flow at the starting track's
1579            # sample rate; if the player doesn't natively support it, upsample to the
1580            # closest higher supported rate
1581            target_rate = (
1582                start_streamdetails.audio_format.sample_rate
1583                if start_streamdetails
1584                else (
1585                    fallback_sample_rate
1586                    if fallback_sample_rate is not None
1587                    else max(supported_sample_rates)
1588                )
1589            )
1590            output_sample_rate = _snap_supported_rate_up(target_rate, supported_sample_rates)
1591
1592        content_type, bit_depth = self._pick_pcm_bit_depth(
1593            players, start_streamdetails, crossfade_enabled, overlay_active
1594        )
1595        return AudioFormat(
1596            content_type=content_type,
1597            sample_rate=output_sample_rate,
1598            bit_depth=bit_depth,
1599            channels=2,
1600        )
1601
1602    async def get_audio_source_stream(
1603        self,
1604        streamdetails: StreamDetails,
1605        pcm_format: AudioFormat,
1606        raise_on_error: bool = True,
1607        display_name: str | None = None,
1608        on_no_audio: Callable[[], None] | None = None,
1609    ) -> AsyncGenerator[bytes]:
1610        """
1611        Get the realtime PCM stream for a live AudioSource.
1612
1613        AudioSources are live/realtime: bytes flow at the producer's pace, with
1614        no pre-buffering, no loudness hydration, no volume normalization, no
1615        crossfade/fade-in, no playback-speed shift, no next-track preload. The
1616        path stays as small as possible to keep end-to-end latency low.
1617
1618        Fast path: when the source PCM format already matches the consumer's
1619        ``pcm_format``, the provider's bytes are paced in Python and forwarded
1620        directly — no ffmpeg in the data path.
1621
1622        Slow path: when formats differ, ffmpeg resamples/recodes the stream
1623        (with ``-readrate`` pacing) via ``get_media_stream``.
1624
1625        :param streamdetails: The stream details of the source to stream.
1626        :param pcm_format: Output PCM format the consumer wants.
1627        :param raise_on_error: Re-raise stream errors instead of swallowing them.
1628        :param display_name: Name to identify the source by in the logs.
1629        :param on_no_audio: Called when the stream failed without ever producing
1630            audio, so the caller can mark its own copy of the source unplayable.
1631        """
1632        logger = self.logger.getChild("audio_source_stream")
1633        name = display_name or streamdetails.uri
1634        bytes_received = 0
1635        try:
1636            async for chunk in self._iter_audio_source_pcm(streamdetails, pcm_format):
1637                bytes_received += len(chunk)
1638                yield chunk
1639        except AudioError as err:
1640            streamdetails.stream_error = True
1641            if bytes_received == 0 and not isinstance(err, ProviderStreamLimitError):
1642                if on_no_audio is not None:
1643                    on_no_audio()
1644            if raise_on_error:
1645                raise
1646            logger.error(
1647                "AudioError while streaming AudioSource %s (%s): %s",
1648                name,
1649                streamdetails.uri,
1650                err,
1651            )
1652        except asyncio.CancelledError:
1653            raise
1654        except Exception:
1655            streamdetails.stream_error = True
1656            if raise_on_error:
1657                raise
1658            logger.exception(
1659                "Unexpected error while streaming AudioSource %s (%s)",
1660                name,
1661                streamdetails.uri,
1662            )
1663        finally:
1664            streamdetails.seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1665
1666    async def get_queue_item_stream(
1667        self,
1668        queue_item: QueueItem,
1669        pcm_format: AudioFormat,
1670        seek_position: float = 0,
1671        playback_speed: float = 1.0,
1672        raise_on_error: bool = True,
1673        normalization_override: VolumeNormalizationMode | None = None,
1674        session_id: str | None = None,
1675        prepared_buffer: AudioBuffer | None = None,
1676        exact_seek: bool = False,
1677    ) -> AsyncGenerator[bytes]:
1678        """
1679        Get the (PCM) audio stream for a single queue item.
1680
1681        Audio is always served from the AudioBuffer which stores raw decoded PCM.
1682        Volume normalization and other filters are applied on-the-fly when reading
1683        from the buffer.
1684
1685        AudioSource items dispatch to ``get_audio_source_stream`` instead: they
1686        are realtime and bypass the buffering/normalization/filter machinery.
1687
1688        :param normalization_override: Force this volume normalization mode instead of
1689            re-evaluating it from the (possibly just-updated) loudness measurement. Used by
1690            the crossfade path to keep a track's replayed intro and its body on the same mode.
1691        :param session_id: Queue session that owns processing-detail updates.
1692        :param prepared_buffer: Existing buffer that must be used without opening a new source.
1693        :param exact_seek: Preserve millisecond precision instead of user-seek quantization.
1694        """
1695        streamdetails = queue_item.streamdetails
1696        assert streamdetails
1697
1698        # streamdetails are cached and reused for retries; reset this before any
1699        # media-type-specific dispatch so AudioSource failures do not stick.
1700        streamdetails.stream_error = False
1701
1702        if queue_item.media_type == MediaType.AUDIO_SOURCE:
1703
1704            def _mark_item_unavailable() -> None:
1705                queue_item.available = False
1706
1707            async for chunk in self.get_audio_source_stream(
1708                streamdetails=streamdetails,
1709                pcm_format=pcm_format,
1710                raise_on_error=raise_on_error,
1711                display_name=queue_item.name,
1712                on_no_audio=_mark_item_unavailable,
1713            ):
1714                yield chunk
1715            return
1716        filter_params: list[str] = []
1717
1718        logger = self.logger.getChild("queue_item_stream")
1719
1720        if normalization_override is not None:
1721            # crossfade path pins the body to the intro's mode; skip hydration/re-eval that could flip it
1722            streamdetails.volume_normalization_mode = normalization_override
1723        else:
1724            # hydrate loudness from audio analysis (just-in-time, so that a measurement
1725            # completed during a previous play is picked up here). A live analyzer run
1726            # may have already populated streamdetails.loudness in memory — don't clobber
1727            # that, and don't clobber a value set upstream by the music provider.
1728            if streamdetails.loudness is None:
1729                if analysis := await self.mass.streams.audio_analysis.get_audio_analysis(
1730                    streamdetails.item_id,
1731                    streamdetails.provider,
1732                    media_type=streamdetails.media_type,
1733                    # use the authoritative EBU R128 value, not another provider's loudness proxy
1734                    priority=(LOUDNESS_ANALYSIS_DOMAIN,),
1735                ):
1736                    if analysis.loudness_integrated is not None:
1737                        streamdetails.loudness = round(analysis.loudness_integrated, 2)
1738                    if analysis.loudness_album is not None and streamdetails.loudness_album is None:
1739                        streamdetails.loudness_album = round(analysis.loudness_album, 2)
1740
1741            # re-evaluate normalization mode: the background loudness analyzer may have
1742            # updated streamdetails.loudness since get_stream_details was called
1743            if streamdetails.queue_id:
1744                volume_normalization_enabled = (
1745                    self.mass.config.get_effective_player_queue_config_value(
1746                        streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
1747                    )
1748                    != CONF_VALUE_DISABLED
1749                )
1750                streamdetails.volume_normalization_mode = get_normalization_mode(
1751                    self._get_volume_normalization_preference(streamdetails),
1752                    volume_normalization_enabled,
1753                    streamdetails,
1754                    self.mass.streams.source_normalizes_audio(streamdetails),
1755                )
1756
1757        # get or create the AudioBuffer (stores raw decoded PCM). This runs before the
1758        # filters are built because a source-capacity reselection can hand back another
1759        # provider's streamdetails, which everything below must then work with.
1760        seek_position_ms = int(seek_position * 1000)
1761        try:
1762            if prepared_buffer is not None:
1763                if streamdetails.buffer is not prepared_buffer or not prepared_buffer.is_valid(
1764                    seek_position_ms
1765                ):
1766                    raise AudioError("Prepared crossfade buffer is no longer available")
1767                audio_buffer = prepared_buffer
1768            else:
1769                audio_buffer = await self.get_audio_buffer(
1770                    queue_item, seek_position_ms=seek_position_ms, reason="streaming"
1771                )
1772        except AudioError as err:
1773            streamdetails.stream_error = True
1774            if raise_on_error:
1775                raise
1776            logger.error(
1777                "AudioError while preparing queue item %s (%s): %s",
1778                queue_item.name,
1779                streamdetails.uri,
1780                err,
1781            )
1782            return
1783        streamdetails = queue_item.streamdetails
1784        assert streamdetails  # for type checking
1785        if normalization_override is not None:
1786            # a capacity reselection hands back freshly resolved details, so the
1787            # crossfade's intro/body normalization pin must be re-applied to them
1788            streamdetails.volume_normalization_mode = normalization_override
1789
1790        # handle volume normalization
1791        gain_correct: float | None = None
1792        if streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC:
1793            filter_rule = (
1794                f"loudnorm=I={streamdetails.target_loudness}"
1795                ":TP=-2.0:LRA=10.0:offset=0.0:print_format=json"
1796            )
1797            filter_params.append(filter_rule)
1798        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.FIXED_GAIN:
1799            config_key = (
1800                CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS
1801                if streamdetails.media_type == MediaType.TRACK
1802                else CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO
1803            )
1804            gain_value = self.mass.streams.get_config_value(config_key, return_type=float)
1805            gain_correct = round(gain_value, 2)
1806            filter_params.append(f"volume={gain_correct}dB")
1807        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.MEASUREMENT_ONLY:
1808            target_loudness = (
1809                float(streamdetails.target_loudness)
1810                if streamdetails.target_loudness is not None
1811                else 0.0
1812            )
1813            if streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
1814                gain_correct = target_loudness - float(streamdetails.loudness_album)
1815            elif streamdetails.loudness is not None:
1816                gain_correct = target_loudness - float(streamdetails.loudness)
1817            else:
1818                gain_correct = 0.0
1819            gain_correct = round(gain_correct, 2)
1820            filter_params.append(f"volume={gain_correct}dB")
1821        streamdetails.volume_normalization_gain_correct = gain_correct
1822
1823        # handle playback speed
1824        if playback_speed != 1.0:
1825            filter_params.append(f"atempo={playback_speed}")
1826
1827        # handle optional fade-in
1828        if streamdetails.fade_in:
1829            filter_params.insert(0, "afade=type=in:start_time=0:duration=3")
1830
1831        logger.log(
1832            VERBOSE_LOG_LEVEL,
1833            "Starting queue item stream for %s (%s)"
1834            " - using fade-in: %s"
1835            " - using volume normalization: %s"
1836            " - using playback speed: %s",
1837            queue_item.name,
1838            streamdetails.uri,
1839            streamdetails.fade_in,
1840            streamdetails.volume_normalization_mode,
1841            playback_speed,
1842        )
1843
1844        if (
1845            streamdetails.queue_id
1846            and (queue_data := self.mass.player_queues.queue_data_or_none(streamdetails.queue_id))
1847            and (processing_session_id := session_id or queue_data.session_id)
1848        ):
1849            self.mass.streams.audio_processing.update_item_runtime(
1850                queue_id=streamdetails.queue_id,
1851                session_id=processing_session_id,
1852                queue_item_id=queue_item.queue_item_id,
1853                input_format=audio_buffer.pcm_format,
1854                pcm_format=pcm_format,
1855                normalization=get_normalization_details(streamdetails, gain_correct),
1856                playback_speed=playback_speed,
1857                alters_audio=streamdetails.fade_in,
1858            )
1859        # read from buffer with filters applied (volume normalization, speed, fade-in, etc.)
1860        # if no processing needed, this yields directly from the buffer
1861        media_stream_gen = audio_buffer.get_stream(
1862            output_format=pcm_format,
1863            seek_position_ms=seek_position_ms,
1864            filter_params=filter_params or None,
1865            exact_seek=exact_seek,
1866        )
1867
1868        first_chunk_received = False
1869        bytes_received = 0
1870        finished = False
1871        next_buffer_triggered = False
1872        stream_started_at = asyncio.get_event_loop().time()
1873        try:
1874            async for chunk in media_stream_gen:
1875                bytes_received += len(chunk)
1876                if not first_chunk_received:
1877                    first_chunk_received = True
1878                    logger.log(
1879                        VERBOSE_LOG_LEVEL,
1880                        "First audio chunk received for %s (%s) after %.2f seconds",
1881                        queue_item.name,
1882                        streamdetails.uri,
1883                        asyncio.get_event_loop().time() - stream_started_at,
1884                    )
1885                # trigger pre-buffering of the next item well before end
1886                # to ensure the raw PCM is ready when the next item needs to be streamed.
1887                # tracks and sound effects are finite files that fill and close immediately;
1888                # live sources (radio, audio_source) open an upstream connection that would
1889                # sit idle and likely time out before the player actually consumes it.
1890                # a realtime source is excluded for the same reason from the other side:
1891                # the next item's audio does not exist yet at any point during this one,
1892                # so only the source itself can say when it does - it triggers the
1893                # pre-buffer through prepare_next_audio_buffer() when it gets there.
1894                if (
1895                    not next_buffer_triggered
1896                    and streamdetails.duration
1897                    and not streamdetails.is_realtime
1898                    and (queue := self.mass.player_queues.get_active_queue(queue_item.queue_id))
1899                    and queue.next_item
1900                    and queue.next_item.queue_item_id != queue_item.queue_item_id
1901                    and queue.next_item.media_type in (MediaType.TRACK, MediaType.SOUND_EFFECT)
1902                    and (bytes_received / pcm_format.pcm_sample_size + seek_position)
1903                    >= streamdetails.duration - 60
1904                ):
1905                    next_buffer_triggered = True
1906                    self.mass.player_queues.prepare_next_audio_buffer(queue_item.queue_id)
1907                yield chunk
1908                del chunk
1909            finished = True
1910        except AudioError as err:
1911            streamdetails.stream_error = True
1912            # revoke availability when the stream never produced any audio
1913            if bytes_received == 0 and not isinstance(err, ProviderStreamLimitError):
1914                queue_item.available = False
1915            if raise_on_error:
1916                raise
1917            logger.error(
1918                "AudioError while streaming queue item %s (%s): %s",
1919                queue_item.name,
1920                streamdetails.uri,
1921                err,
1922            )
1923        except asyncio.CancelledError:
1924            raise
1925        except Exception:
1926            streamdetails.stream_error = True
1927            if raise_on_error:
1928                raise
1929            logger.exception(
1930                "Unexpected error while streaming queue item %s (%s)",
1931                queue_item.name,
1932                streamdetails.uri,
1933            )
1934        finally:
1935            seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1936            streamdetails.seconds_streamed = seconds_streamed
1937            logger.log(
1938                VERBOSE_LOG_LEVEL,
1939                "stream %s for %s in %.2f seconds - seconds streamed/buffered: %.2f",
1940                "aborted" if not finished else "finished",
1941                streamdetails.uri,
1942                asyncio.get_event_loop().time() - stream_started_at,
1943                seconds_streamed,
1944            )
1945            self._notify_provider_streamed(streamdetails, finished, seconds_streamed)
1946
1947    async def get_queue_item_stream_with_smartfade(
1948        self,
1949        player: Player,
1950        queue_item: QueueItem,
1951        pcm_format: AudioFormat,
1952        crossfade_mode: CrossfadeMode = CrossfadeMode.SMART_CROSSFADE,
1953        standard_crossfade_duration: int = 10,
1954        session_id: str | None = None,
1955    ) -> AsyncGenerator[bytes]:
1956        """
1957        Return one queue item with a crossfade into the next item.
1958
1959        :param player: Player consuming the stream.
1960        :param queue_item: Queue item to stream.
1961        :param pcm_format: Shared PCM format.
1962        :param crossfade_mode: Effective crossfade mode.
1963        :param standard_crossfade_duration: Configured standard crossfade duration.
1964        :param session_id: Queue session that owns processing-detail updates.
1965        """
1966        queue = self.mass.player_queues.get(queue_item.queue_id)
1967        if not queue:
1968            raise RuntimeError(f"Queue {queue_item.queue_id} not found")
1969
1970        streamdetails = queue_item.streamdetails
1971        assert streamdetails
1972        crossfade_data = self._crossfade_data.get(queue.queue_id)
1973        if crossfade_data is None and not streamdetails.seek_position:
1974            # the outgoing stream may still be building this item's fade. A seek
1975            # discards the fade below either way, and waiting for one would only
1976            # hold the response open on audio that is about to be thrown away.
1977            crossfade_data = await self._await_pending_crossfade(queue, queue_item)
1978
1979        if crossfade_data and streamdetails.seek_position > 0:
1980            # don't do crossfade when seeking into track
1981            self.logger.debug(
1982                "Discarding crossfade data for queue %s - seeking into track (pos=%s)",
1983                queue.display_name,
1984                streamdetails.seek_position,
1985            )
1986            crossfade_data = None
1987        if crossfade_data and (crossfade_data.queue_item_id != queue_item.queue_item_id):
1988            # edge case alert: the next item changed just while we were preloading/crossfading
1989            self.logger.warning(
1990                "Skipping crossfade data for queue %s - next item changed!"
1991                " (expected queue_item_id=%s, got=%s)",
1992                queue.display_name,
1993                crossfade_data.queue_item_id,
1994                queue_item.queue_item_id,
1995            )
1996            crossfade_data = None
1997            self._crossfade_data.pop(queue.queue_id, None)
1998        elif not crossfade_data:
1999            self.logger.debug(
2000                "No crossfade data available for queue %s (queue_item_id=%s)",
2001                queue.display_name,
2002                queue_item.queue_item_id,
2003            )
2004
2005        # only a fade actually handed over proves the player still holds the lead it
2006        # earned; a fresh start, a seek or a lost handoff leaves its buffer unknown,
2007        # so the holdback starts from nothing again
2008        earned_lead, earned_at = self._playback_lead.pop(queue.queue_id, (0.0, 0.0))
2009        carried_lead = earned_lead if crossfade_data else 0.0
2010        carried_at = earned_at if crossfade_data else None
2011
2012        self.logger.debug(
2013            "Start Streaming queue track: %s (%s) for queue %s on player %s"
2014            "- crossfade mode: %s "
2015            "- crossfading from previous track: %s "
2016            "- lead carried into this item: %.1fs ",
2017            queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
2018            queue_item.name,
2019            queue.display_name,
2020            player.name,
2021            crossfade_mode,
2022            "true" if crossfade_data else "false",
2023            carried_lead,
2024        )
2025        # report the fade this item was actually faded into; the fade leaving it is
2026        # only known once the next item's overlap has been selected further down
2027        self._report_crossfade_mode(
2028            queue.queue_id,
2029            queue_item,
2030            pcm_format,
2031            crossfade_data.crossfade_mode if crossfade_data else CrossfadeMode.DISABLED,
2032            session_id,
2033            # only radio carries an overlay outside flow mode, and this path is tracks-only
2034            overlay_enabled=False,
2035        )
2036
2037        buffer = bytearray()
2038        bytes_written = 0
2039        # calculate crossfade buffer size; a realtime source's holdback only ever
2040        # withholds its banked surplus, so the smart window is a ceiling there
2041        crossfade_buffer_duration = (
2042            SMART_CROSSFADE_DURATION
2043            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
2044            else standard_crossfade_duration
2045        )
2046        crossfade_buffer_duration = min(
2047            crossfade_buffer_duration,
2048            int(streamdetails.duration / 2)
2049            if streamdetails.duration
2050            else crossfade_buffer_duration,
2051        )
2052        # skip crossfade if buffer would be too small to be meaningful
2053        if crossfade_buffer_duration < MIN_CROSSFADE_DURATION:
2054            crossfade_buffer_duration = 0
2055        # Ensure crossfade buffer size is aligned to frame boundaries
2056        # Frame size = bytes_per_sample * channels
2057        bytes_per_sample = pcm_format.bit_depth // 8
2058        frame_size = bytes_per_sample * pcm_format.channels
2059        crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
2060        # Round down to nearest frame boundary
2061        crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
2062        fade_out_data: bytes | None = None
2063        uncredited_tail_bytes = 0
2064
2065        # pin the body to DYNAMIC when the intro was baked DYNAMIC,
2066        # else a late measurement flips it and causes a volume jump
2067        norm_override: VolumeNormalizationMode | None = None
2068        if crossfade_data and crossfade_data.normalization_mode == VolumeNormalizationMode.DYNAMIC:
2069            norm_override = VolumeNormalizationMode.DYNAMIC
2070
2071        exact_buffer_seek = crossfade_data is not None
2072        if crossfade_data:
2073            # reported media-time (TRIM + CF) is decoupled from the raw buffer seek below (X)
2074            streamdetails.seek_position = crossfade_data.elapsed_time_offset
2075            # yield the POST portion (resample if previous track's format differs)
2076            if crossfade_data.pcm_format != pcm_format:
2077                async for _chunk in resample_pcm_audio(
2078                    crossfade_data.data, crossfade_data.pcm_format, pcm_format
2079                ):
2080                    yield _chunk
2081                    bytes_written += len(_chunk)
2082            else:
2083                for pcm_slice in iter_pcm_slices(crossfade_data.data, pcm_format, 1000):
2084                    yield pcm_slice
2085                    await asyncio.sleep(0)
2086                bytes_written += len(crossfade_data.data)
2087            # skip past the source media already consumed by the crossfade
2088            discard_position = crossfade_data.fade_in_media_duration
2089            crossfade_data = None
2090            self._crossfade_data.pop(queue.queue_id, None)
2091        else:
2092            discard_position = float(streamdetails.seek_position)
2093
2094        # Yield the first WARMUP_DURATION worth of audio immediately so playback starts
2095        # right away. After that, start accumulating the crossfade holdback buffer.
2096        warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
2097        warmup_bytes = 0
2098        total_chunks_received = 0
2099        playback_speed = cast("float", queue_item.extra_attributes.get("playback_speed", 1.0))
2100        # the holdback is grown out of the audio banked ahead of playback instead of
2101        # armed as one fixed window, so a source delivering near playback pace keeps
2102        # feeding the player
2103        tail_hold = (
2104            _TailHold(pcm_format, queue_item, carried_lead=carried_lead, carried_at=carried_at)
2105            if crossfade_buffer_size > 0
2106            else None
2107        )
2108        async for chunk in self.get_queue_item_stream(
2109            queue_item,
2110            pcm_format,
2111            seek_position=discard_position,
2112            playback_speed=playback_speed,
2113            normalization_override=norm_override,
2114            session_id=session_id,
2115            exact_seek=exact_buffer_seek,
2116        ):
2117            total_chunks_received += 1
2118            if tail_hold is not None:
2119                tail_hold.note_bytes(len(chunk))
2120
2121            if warmup_bytes < warmup_size:
2122                # warmup: yield directly, don't buffer
2123                yield chunk
2124                warmup_bytes += len(chunk)
2125                bytes_written += len(chunk)
2126                del chunk
2127                continue
2128
2129            buffer.extend(chunk)
2130            del chunk
2131            hold_target = (
2132                tail_hold.hold_target(crossfade_buffer_size, frame_size)
2133                if tail_hold is not None
2134                else 0
2135            )
2136            if len(buffer) <= hold_target:
2137                await asyncio.sleep(0)
2138                continue
2139            # yield everything above the current holdback window; the slice can
2140            # run short of a whole second when the window is small, so credit
2141            # what is actually yielded - a nominal full-second credit inflates
2142            # the play log and the reported duration
2143            while len(buffer) > hold_target:
2144                pcm_slice = bytes(buffer[: pcm_format.pcm_sample_size])
2145                yield pcm_slice
2146                bytes_written += len(pcm_slice)
2147                del buffer[: len(pcm_slice)]
2148                await asyncio.sleep(0)
2149
2150        #### HANDLE END OF TRACK
2151
2152        # get next track for crossfade
2153        crossfade_start_time = asyncio.get_event_loop().time()
2154        next_queue_item: QueueItem | None
2155        try:
2156            self.logger.debug(
2157                "Preloading NEXT track for crossfade for queue %s", queue.display_name
2158            )
2159            next_queue_item = await self.mass.player_queues.load_next_queue_item(
2160                queue.queue_id, queue_item.queue_item_id
2161            )
2162            # set index_in_buffer to prevent our next track is overwritten while preloading
2163            if next_queue_item.streamdetails is None:
2164                raise InvalidDataError(
2165                    f"No streamdetails for next queue item {next_queue_item.queue_item_id}"
2166                )
2167            queue.index_in_buffer = self.mass.player_queues.index_by_id(
2168                queue.queue_id, next_queue_item.queue_item_id
2169            )
2170        except QueueEmpty:
2171            # end of queue reached, no next item
2172            next_queue_item = None
2173
2174        crossfade_allowed = False
2175        transition_mode = CrossfadeMode.DISABLED
2176        fade_in_buffer_duration = 0.0
2177        fade_in_playback_speed = 1.0
2178        # a fade needs enough of the outgoing track to overlap with; a holdback that
2179        # armed late (or not at all) leaves less than that
2180        min_fade_out_size = int(pcm_format.pcm_sample_size * MIN_CROSSFADE_DURATION)
2181        # Claim the handoff the moment the next item is known, before the awaits that
2182        # size the fade: the speaker can ask for that item's url during them, and a
2183        # marker registered afterwards would arrive too late to be waited for.
2184        handoff: asyncio.Event | None = None
2185        if next_queue_item is not None:
2186            handoff = asyncio.Event()
2187            self._crossfade_pending[queue.queue_id] = (
2188                next_queue_item.queue_item_id,
2189                handoff,
2190            )
2191        try:
2192            if (
2193                len(buffer) >= min_fade_out_size
2194                and next_queue_item
2195                and next_queue_item.streamdetails
2196            ):
2197                fade_in_playback_speed = cast(
2198                    "float", next_queue_item.extra_attributes.get("playback_speed", 1.0)
2199                )
2200                next_pcm = await self.select_pcm_format(
2201                    player=player,
2202                    streamdetails=next_queue_item.streamdetails,
2203                    crossfade_enabled=True,
2204                )
2205                crossfade_allowed = self.crossfade_allowed(
2206                    queue_item,
2207                    crossfade_mode=crossfade_mode,
2208                    player_id=player.player_id,
2209                    flow_mode=False,
2210                    next_queue_item=next_queue_item,
2211                    sample_rate=pcm_format.sample_rate,
2212                    next_sample_rate=next_pcm.sample_rate,
2213                )
2214                if crossfade_allowed:
2215                    # a realtime incoming track has audio to read only once its session
2216                    # produces; give it a bounded chance to show up
2217                    await self._await_realtime_fade_source(next_queue_item.streamdetails)
2218                    transition_mode, fade_in_buffer_duration = self._select_buffered_crossfade(
2219                        next_queue_item.streamdetails,
2220                        crossfade_mode,
2221                        standard_crossfade_duration,
2222                        fade_out_seconds=len(buffer) / pcm_format.pcm_sample_size,
2223                        playback_speed=fade_in_playback_speed,
2224                    )
2225                    crossfade_allowed = transition_mode != CrossfadeMode.DISABLED
2226            if not crossfade_allowed:
2227                # no crossfade enabled/allowed, just yield the buffer last part
2228                bytes_written += len(buffer)
2229                for pcm_slice in iter_pcm_slices(bytes(buffer), pcm_format, 1000):
2230                    yield pcm_slice
2231                    await asyncio.sleep(0)
2232            else:
2233                assert next_queue_item is not None
2234                assert next_queue_item.streamdetails is not None
2235                assert next_queue_item.streamdetails.buffer is not None
2236                fade_in_audio_buffer = cast("AudioBuffer", next_queue_item.streamdetails.buffer)
2237                # the remaining buffer is the fade-out tail of the current track
2238                fade_out_data = bytes(buffer)
2239                buffer = bytearray()
2240                fade_in_buffer_size = int(pcm_format.pcm_sample_size * fade_in_buffer_duration)
2241                fade_in_buffer_size = (fade_in_buffer_size // frame_size) * frame_size
2242                # initialized before the try block — the except handler reads these
2243                first_part_written = 0
2244                second_part_buf = bytearray()
2245                try:
2246                    # wrap the next track's stream in a counting generator that caps
2247                    # at the resident fade-in size and tracks how many bytes were consumed
2248                    fade_in_bytes_consumed = 0
2249
2250                    _next_item = next_queue_item
2251
2252                    async def _limited_fade_in() -> AsyncGenerator[bytes]:
2253                        nonlocal fade_in_bytes_consumed
2254                        fade_in_stream = self.get_queue_item_stream(
2255                            _next_item,
2256                            pcm_format,
2257                            playback_speed=fade_in_playback_speed,
2258                            session_id=session_id,
2259                            prepared_buffer=fade_in_audio_buffer,
2260                        )
2261                        async with aclosing(fade_in_stream):
2262                            async for chunk in fade_in_stream:
2263                                remaining = fade_in_buffer_size - fade_in_bytes_consumed
2264                                if remaining <= 0:
2265                                    break
2266                                if len(chunk) >= remaining:
2267                                    fade_in_bytes_consumed += remaining
2268                                    yield chunk[:remaining]
2269                                    break
2270                                fade_in_bytes_consumed += len(chunk)
2271                                yield chunk
2272
2273                    smart_fade = await self.smart_fades_mixer.build(
2274                        fade_in_streamdetails=next_queue_item.streamdetails,
2275                        fade_out_streamdetails=streamdetails,
2276                        pcm_format=pcm_format,
2277                        standard_crossfade_duration=standard_crossfade_duration,
2278                        mode=transition_mode,
2279                        fade_out_data=fade_out_data,
2280                        fade_in_bytes_len=fade_in_buffer_size,
2281                    )
2282                    # the mixer degrades to a standard fade when the smart one cannot be planned
2283                    applied_mode = (
2284                        CrossfadeMode.STANDARD_CROSSFADE
2285                        if isinstance(smart_fade, StandardCrossFade)
2286                        else transition_mode
2287                    )
2288                    crossfade_timing = smart_fade.timing_info
2289                    # Split mix output at end-of-overlap: PRE+CF to A, POST to B's intro.
2290                    fadeout_share_bytes = int(
2291                        (
2292                            crossfade_timing.pre_crossfade_duration
2293                            + crossfade_timing.crossfade_duration
2294                        )
2295                        * pcm_format.pcm_sample_size
2296                    )
2297                    fadeout_share_bytes = (fadeout_share_bytes // frame_size) * frame_size
2298                    mix_stream = self.smart_fades_mixer.mix(
2299                        smart_fade,
2300                        fade_in_part=_limited_fade_in(),
2301                        fade_out_part=fade_out_data,
2302                        pcm_format=pcm_format,
2303                    )
2304                    # aclosing so an aborted stream tears the mix (and its feeder,
2305                    # which holds a read on the fade-in stream) down first
2306                    async with aclosing(mix_stream):
2307                        async for mix_chunk in mix_stream:
2308                            if first_part_written < fadeout_share_bytes:
2309                                # split this chunk so A gets exactly fadeout_share_bytes
2310                                remaining = fadeout_share_bytes - first_part_written
2311                                if len(mix_chunk) > remaining:
2312                                    yield mix_chunk[:remaining]
2313                                    first_part_written += remaining
2314                                    bytes_written += remaining
2315                                    second_part_buf.extend(mix_chunk[remaining:])
2316                                else:
2317                                    yield mix_chunk
2318                                    first_part_written += len(mix_chunk)
2319                                    bytes_written += len(mix_chunk)
2320                            else:
2321                                second_part_buf.extend(mix_chunk)
2322                    # tail consumed by the mix but not credited to bytes_written
2323                    uncredited_tail_bytes = len(fade_out_data) - first_part_written
2324                    self._report_crossfade_mode(
2325                        queue.queue_id,
2326                        queue_item,
2327                        pcm_format,
2328                        applied_mode,
2329                        session_id,
2330                        overlay_enabled=False,
2331                    )
2332                    self._crossfade_data[queue_item.queue_id] = CrossfadeData(
2333                        data=bytes(second_part_buf),
2334                        fade_in_media_duration=(fade_in_bytes_consumed / pcm_format.pcm_sample_size)
2335                        * fade_in_playback_speed,
2336                        pcm_format=pcm_format,
2337                        queue_item_id=next_queue_item.queue_item_id,
2338                        crossfade_mode=applied_mode,
2339                        elapsed_time_offset=(
2340                            crossfade_timing.fadein_trimmed_duration
2341                            + crossfade_timing.crossfade_duration
2342                        )
2343                        * fade_in_playback_speed,
2344                        normalization_mode=next_queue_item.streamdetails.volume_normalization_mode,
2345                    )
2346                    crossfade_elapsed = asyncio.get_event_loop().time() - crossfade_start_time
2347                    self.logger.debug(
2348                        "Stored crossfade data for queue %s"
2349                        " - next queue_item_id: %s (preparation took %.1fs)",
2350                        queue.display_name,
2351                        next_queue_item.queue_item_id,
2352                        crossfade_elapsed,
2353                    )
2354                except Exception as err:
2355                    if first_part_written or second_part_buf:
2356                        # partial mix already played — concat'd fade_out_data would duplicate audio
2357                        raise
2358                    # crossfade failed, fall back to just yielding the fade_out_data
2359                    self.logger.warning(
2360                        "Crossfade failed for queue %s: %s",
2361                        queue.display_name,
2362                        err,
2363                    )
2364                    next_queue_item = None
2365                    for pcm_slice in iter_pcm_slices(fade_out_data, pcm_format, 1000):
2366                        yield pcm_slice
2367                        await asyncio.sleep(0)
2368                    bytes_written += len(fade_out_data)
2369                    del fade_out_data
2370        finally:
2371            # a waiter must be released whichever way this went: the fade landed, it
2372            # was never allowed, the mixer fell back, or the stream was torn down
2373            if handoff is not None:
2374                handoff.set()
2375                if self._crossfade_pending.get(queue.queue_id, (None, None))[1] is handoff:
2376                    del self._crossfade_pending[queue.queue_id]
2377        # make sure the buffer gets cleaned up
2378        del buffer
2379        # a capacity reselection inside the stream replaces the queue item's details,
2380        # so rebind before the writebacks land on an orphaned object
2381        streamdetails = queue_item.streamdetails or streamdetails
2382        # update duration details based on the actual pcm data we sent
2383        # this also accounts for crossfade and silence stripping
2384        seconds_streamed = bytes_written / pcm_format.pcm_sample_size
2385        streamdetails.seconds_streamed = seconds_streamed
2386        # an externally aborted source ends in a clean EOF mid-track, so the
2387        # streamed length must not be written back as the item's duration
2388        source_buffer = streamdetails.buffer
2389        if source_buffer is None or not source_buffer.cancelled:
2390            uncredited_tail_seconds = uncredited_tail_bytes / pcm_format.pcm_sample_size
2391            # streamdetails.duration is in media-time; seconds_streamed is stream-time
2392            # (post-atempo), so we scale by playback_speed to recover media-time.
2393            streamdetails.duration = int(
2394                streamdetails.seek_position
2395                + (seconds_streamed + uncredited_tail_seconds) * playback_speed
2396            )
2397            # propagate accurate duration to queue_item so UI displays it
2398            queue_item.duration = streamdetails.duration
2399        # bank what the player still holds unplayed, measured now that this item's audio
2400        # and its fade are both handed over. Capped at the largest window a fade can ask
2401        # for: more lead buys nothing, and over-reading it is the direction that starves.
2402        if tail_hold is not None:
2403            self._playback_lead[queue.queue_id] = (
2404                min(tail_hold.banked_lead(bytes_written), float(SMART_CROSSFADE_DURATION)),
2405                asyncio.get_event_loop().time(),
2406            )
2407        self.logger.debug(
2408            "Finished Streaming queue track: %s (%s) on queue %s "
2409            "- crossfade data prepared for next track: %s"
2410            " - lead banked for the next item: %.1fs",
2411            streamdetails.uri,
2412            queue_item.name,
2413            queue.display_name,
2414            (
2415                next_queue_item.name
2416                if next_queue_item and queue_item.queue_id in self._crossfade_data
2417                else "N/A"
2418            ),
2419            self._playback_lead.get(queue.queue_id, (0.0, 0.0))[0],
2420        )
2421
2422    async def get_queue_flow_stream(
2423        self,
2424        queue: PlayerQueue,
2425        start_queue_item: QueueItem,
2426        pcm_format: AudioFormat,
2427        session_id: str | None = None,
2428        protocol_player: Player | None = None,
2429    ) -> AsyncGenerator[bytes]:
2430        """
2431        Get a flow stream of all tracks in the queue as raw PCM audio.
2432
2433        yields chunks of exactly 1 second of audio in the given pcm_format.
2434
2435        :param queue: Queue being streamed.
2436        :param start_queue_item: First queue item in the flow stream.
2437        :param pcm_format: Shared PCM format for the complete flow stream.
2438        :param session_id: Queue session that owns processing-detail updates.
2439        :param protocol_player: The protocol player actually consuming the flow stream.
2440            Must be the same player that was used to select ``pcm_format`` so
2441            restart decisions are made against the correct supported sample rates
2442            and flow mode configuration. Falls back to the queue's player when omitted.
2443        """
2444        # ruff: noqa: PLR0915
2445        assert pcm_format.content_type.is_pcm()
2446        queue_track = None
2447        # seconds of audio the player still holds unplayed, earned by every track this
2448        # stream already fed it, and when that was last measured; a flow stream never
2449        # restarts, so neither does its lead. Boundary work hands over nothing while the
2450        # player keeps playing, so the timestamp is what keeps the carry honest - which
2451        # also covers the paths that leave a track without banking anything.
2452        flow_lead = 0.0
2453        flow_lead_at = 0.0
2454        last_fadeout_part: bytes = b""
2455        last_streamdetails: StreamDetails | None = None
2456        last_queue_track: QueueItem | None = None
2457        last_play_log_entry: PlayLogEntry | None = None
2458        # Snapshot the queue's current session_id. PlayerQueues rotates this on
2459        # every new stream session, so if a newer producer takes over the queue
2460        # (rapid track switch, sync-group reform, dynamic leader handoff) the
2461        # snapshot will no longer match and we exit cleanly on the next yield or
2462        # playlog append — preventing two producers from writing to the same
2463        # pq_data.flow_mode_stream_log.
2464        pq_data = self.mass.player_queues.queue_data(queue.queue_id)
2465        flow_session_id = session_id or pq_data.session_id
2466        if flow_session_id is None or pq_data.session_id != flow_session_id:
2467            self.logger.debug(
2468                "Ignoring stale flow stream for queue %s (session %s, active %s)",
2469                queue.display_name,
2470                flow_session_id,
2471                pq_data.session_id,
2472            )
2473            return
2474        queue.flow_mode = True
2475        # A session can also be handed a second producer, which the session check does not
2476        # catch: players such as DLNA renderers sometimes open the same flow url twice to
2477        # probe the audio. Append to the list published here rather than to whatever the
2478        # queue currently holds, so the entries of a producer that has since been replaced
2479        # end up in a list nobody reads instead of interleaving with the live one's.
2480        flow_log: list[PlayLogEntry] = []
2481        pq_data.flow_mode_stream_log = flow_log
2482        if not start_queue_item:
2483            # this can happen in some (edge case) race conditions
2484            return
2485        pcm_sample_size = pcm_format.pcm_sample_size
2486        if start_queue_item.media_type != MediaType.TRACK:
2487            # no crossfade on non-tracks
2488            crossfade_mode = CrossfadeMode.DISABLED
2489            standard_crossfade_duration = 0
2490        else:
2491            crossfade_mode = self.mass.streams.get_crossfade_mode(queue)
2492            # crossfade duration is a global (queue controller) setting; fallback matches
2493            # CONF_ENTRY_CROSSFADE_DURATION's default
2494            standard_crossfade_duration = self.mass.config.get_raw_core_config_value(
2495                CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8
2496            )
2497        flow_mode_sample_rate_conf, flow_supported_sample_rates = self._flow_restart_context(
2498            queue.queue_id, protocol_player
2499        )
2500        # note: get_crossfade_mode() already falls back to standard when smart fades aren't
2501        # available (no analysis provider / minimal buffer), so crossfade_mode is safe to use.
2502        self.logger.info(
2503            "Start Queue Flow stream for Queue %s - crossfade: %s %s",
2504            queue.display_name,
2505            crossfade_mode,
2506            f"({standard_crossfade_duration}s)"
2507            if crossfade_mode == CrossfadeMode.STANDARD_CROSSFADE
2508            else "",
2509        )
2510        total_chunks_received = 0
2511
2512        def _superseded() -> bool:
2513            """Return True if a newer stream session has taken over this queue."""
2514            return pq_data.session_id != flow_session_id
2515
2516        queue_exhausted = False
2517        incoming_prefetcher = _IncomingFadePrefetcher(self, pcm_format, flow_session_id)
2518        try:
2519            while True:
2520                # every outgoing tail this iteration hands over, wherever it is flushed.
2521                # Those bytes are the previous item's media time, so they are kept out
2522                # of bytes_written, but the player receives them here and the lead is
2523                # measured from what the player received.
2524                outgoing_emitted = 0
2525                # bail out early if a newer producer has taken over this queue,
2526                # so we don't append another entry to a stream log we no longer own
2527                if _superseded():
2528                    self.logger.debug(
2529                        "Flow stream for queue %s superseded (session %s -> %s) "
2530                        "- exiting before next track",
2531                        queue.display_name,
2532                        flow_session_id,
2533                        pq_data.session_id,
2534                    )
2535                    return
2536                # get (next) queue item to stream
2537                if queue_track is None:
2538                    queue_track = start_queue_item
2539                else:
2540                    try:
2541                        queue_track = await self.mass.player_queues.load_next_queue_item(
2542                            queue.queue_id, queue_track.queue_item_id
2543                        )
2544                    except QueueEmpty:
2545                        queue_exhausted = True
2546                        break
2547
2548                if self._flow_stream_needs_restart(
2549                    queue_track,
2550                    pcm_format,
2551                    flow_supported_sample_rates,
2552                    flow_mode_sample_rate_conf,
2553                    is_first_track=queue_track is start_queue_item,
2554                ):
2555                    break
2556
2557                if queue_track.streamdetails is None:
2558                    self.logger.error(
2559                        "No StreamDetails for queue item %s (%s) on queue %s - skipping track",
2560                        queue_track.queue_item_id,
2561                        queue_track.name,
2562                        queue.display_name,
2563                    )
2564                    continue
2565                # a realtime source gets a fade decided from what its boundary can
2566                # actually deliver (see _select_buffered_crossfade)
2567                item_crossfade_mode = crossfade_mode
2568                self.logger.debug(
2569                    "Start Streaming queue track: %s (%s) for queue %s",
2570                    queue_track.streamdetails.uri,
2571                    queue_track.name,
2572                    queue.display_name,
2573                )
2574                # last chance to bail before mutating the stream log: a newer producer
2575                # may have taken over while we were awaiting load_next_queue_item
2576                if _superseded():
2577                    self.logger.debug(
2578                        "Flow stream for queue %s superseded - exiting before playlog append",
2579                        queue.display_name,
2580                    )
2581                    return
2582                track_playback_speed = cast(
2583                    "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2584                )
2585                # calculate crossfade buffer size; a realtime source's holdback only
2586                # ever withholds its banked surplus, so the smart window is a
2587                # ceiling there
2588                crossfade_buffer_duration = (
2589                    SMART_CROSSFADE_DURATION
2590                    if item_crossfade_mode == CrossfadeMode.SMART_CROSSFADE
2591                    else standard_crossfade_duration
2592                )
2593                crossfade_buffer_duration = min(
2594                    crossfade_buffer_duration,
2595                    int(queue_track.streamdetails.duration / 2)
2596                    if queue_track.streamdetails.duration
2597                    else crossfade_buffer_duration,
2598                )
2599                # skip crossfade if buffer would be too small to be meaningful
2600                if crossfade_buffer_duration < MIN_CROSSFADE_DURATION:
2601                    crossfade_buffer_duration = 0
2602                # Ensure crossfade buffer size is aligned to frame boundaries
2603                # Frame size = bytes_per_sample * channels
2604                bytes_per_sample = pcm_format.bit_depth // 8
2605                frame_size = bytes_per_sample * pcm_format.channels
2606                crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
2607                # Round down to nearest frame boundary
2608                crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
2609                warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
2610
2611                # raw_seek_position feeds the PCM buffer; streamdetails.seek_position
2612                # (overwritten below) only drives reported elapsed time.
2613                raw_seek_position = queue_track.streamdetails.seek_position
2614                # Build eagerly so seek_position is set before PlayLogEntry is appended —
2615                # consumer-paced mix() would otherwise let the queue briefly report 0.
2616                crossfade_smart_fade: SmartFade | None = None
2617                collect_resident = 0.0
2618                incoming_crossfade_size = crossfade_buffer_size
2619                incoming_audio_buffer: AudioBuffer | None = None
2620                build_seconds = 0.0
2621                transition_mode = CrossfadeMode.DISABLED
2622                applied_mode = CrossfadeMode.DISABLED
2623                outgoing_queue_track = last_queue_track
2624                if last_fadeout_part and last_streamdetails:
2625                    incoming_duration = 0.0
2626                    if crossfade_buffer_size > 0 and item_crossfade_mode != CrossfadeMode.DISABLED:
2627                        # a realtime incoming track has audio to read only once its
2628                        # session produces; give it a bounded chance to show up
2629                        await self._await_realtime_fade_source(queue_track.streamdetails)
2630                        transition_mode, incoming_duration = self._select_buffered_crossfade(
2631                            queue_track.streamdetails,
2632                            item_crossfade_mode,
2633                            standard_crossfade_duration,
2634                            fade_out_seconds=len(last_fadeout_part) / pcm_sample_size,
2635                            playback_speed=track_playback_speed,
2636                        )
2637                    if transition_mode == CrossfadeMode.DISABLED:
2638                        # nothing to fade into: flush the held-back tail of the previous track
2639                        for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2640                            yield pcm_slice
2641                            await asyncio.sleep(0)
2642                        outgoing_emitted += len(last_fadeout_part)
2643                        last_fadeout_part = b""
2644                        last_streamdetails = None
2645                        last_play_log_entry = None
2646                        last_queue_track = None
2647                    else:
2648                        assert queue_track.streamdetails.buffer is not None
2649                        incoming_audio_buffer = cast(
2650                            "AudioBuffer", queue_track.streamdetails.buffer
2651                        )
2652                        incoming_crossfade_size = int(
2653                            pcm_format.pcm_sample_size * incoming_duration
2654                        )
2655                        incoming_crossfade_size = (
2656                            incoming_crossfade_size // frame_size
2657                        ) * frame_size
2658                        collect_resident = incoming_audio_buffer.duration_available
2659                        applied_mode = transition_mode
2660                        build_started = asyncio.get_event_loop().time()
2661                        crossfade_smart_fade = await self.smart_fades_mixer.build(
2662                            fade_in_streamdetails=queue_track.streamdetails,
2663                            fade_out_streamdetails=last_streamdetails,
2664                            pcm_format=pcm_format,
2665                            standard_crossfade_duration=standard_crossfade_duration,
2666                            mode=transition_mode,
2667                            fade_out_data=last_fadeout_part,
2668                            fade_in_bytes_len=incoming_crossfade_size,
2669                        )
2670                        build_seconds = asyncio.get_event_loop().time() - build_started
2671                        timing_info = crossfade_smart_fade.timing_info
2672                        if isinstance(crossfade_smart_fade, StandardCrossFade):
2673                            # the mixer degrades to a standard fade when the smart one
2674                            # cannot be planned, so that is what will really be applied
2675                            applied_mode = CrossfadeMode.STANDARD_CROSSFADE
2676                            # A standard fade blends its overlap and passes everything after it
2677                            # through untouched, so only the overlap has to be in hand before
2678                            # the transition can start. Holding back the rest buys nothing and
2679                            # keeps the player waiting - a smart fade does need its full window,
2680                            # which is only chosen when the analysis it needs is already there.
2681                            blended_seconds = (
2682                                timing_info.fadein_trimmed_duration + timing_info.crossfade_duration
2683                            )
2684                            blended_size = int(pcm_format.pcm_sample_size * blended_seconds)
2685                            incoming_crossfade_size = min(
2686                                incoming_crossfade_size,
2687                                (blended_size // frame_size) * frame_size,
2688                            )
2689                        queue_track.streamdetails.seek_position = (
2690                            raw_seek_position
2691                            + (timing_info.fadein_trimmed_duration + timing_info.crossfade_duration)
2692                            * track_playback_speed
2693                        )
2694                # no fade is credited to this track until one is really rendered below
2695                self._report_crossfade_mode(
2696                    queue.queue_id,
2697                    queue_track,
2698                    pcm_format,
2699                    CrossfadeMode.DISABLED,
2700                    flow_session_id,
2701                    overlay_enabled=overlay_active(queue),
2702                )
2703                # append to play log so the queue controller can work out which track is playing
2704                play_log_entry = PlayLogEntry(queue_track.queue_item_id)
2705                flow_log.append(play_log_entry)
2706
2707                bytes_written = 0
2708                crossfade_buffer = bytearray()
2709                warmup_bytes = 0
2710                first_chunk_received = False
2711                # the holdback is grown out of the audio banked ahead of playback instead
2712                # of armed as one fixed window, so a source delivering near playback pace
2713                # keeps feeding the player
2714                tail_hold = (
2715                    _TailHold(
2716                        pcm_format, queue_track, carried_lead=flow_lead, carried_at=flow_lead_at
2717                    )
2718                    if item_crossfade_mode != CrossfadeMode.DISABLED
2719                    else None
2720                )
2721
2722                item_stream = await incoming_prefetcher.take(queue_track, int(raw_seek_position))
2723                prefetched_size = incoming_prefetcher.collected_at_handover if item_stream else 0
2724                if item_stream is None:
2725                    item_stream = self.get_queue_item_stream(
2726                        queue_track,
2727                        pcm_format=pcm_format,
2728                        seek_position=int(raw_seek_position),
2729                        playback_speed=cast(
2730                            "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2731                        ),
2732                        raise_on_error=False,
2733                        session_id=flow_session_id,
2734                        prepared_buffer=incoming_audio_buffer,
2735                    )
2736
2737                # closing here releases the decoders on an early exit,
2738                # instead of leaving them to the garbage collector
2739                async with aclosing(item_stream):
2740                    async for chunk in item_stream:
2741                        # if a newer producer has taken over this queue, stop sending
2742                        # audio and exit cleanly before the outer-loop end-of-track
2743                        # bookkeeping mutates seconds_streamed / duration on the log
2744                        if _superseded():
2745                            self.logger.debug(
2746                                "Flow stream for queue %s superseded - stopping chunk yield",
2747                                queue.display_name,
2748                            )
2749                            return
2750                        total_chunks_received += 1
2751                        if tail_hold is not None:
2752                            tail_hold.note_bytes(len(chunk))
2753                        if not first_chunk_received:
2754                            first_chunk_received = True
2755                            # inform the queue that the track is now loaded in the buffer
2756                            # so the next track can be preloaded
2757                            self.mass.player_queues.track_loaded_in_buffer(
2758                                queue.queue_id, queue_track.queue_item_id
2759                            )
2760
2761                        if item_crossfade_mode == CrossfadeMode.DISABLED:
2762                            # no cross/smart fade: yield chunks directly without intermediate buffer
2763                            yield chunk
2764                            bytes_written += len(chunk)
2765                            del chunk
2766                            continue
2767
2768                        # Warmup: yield chunks directly until we have streamed WARMUP_DURATION
2769                        # worth of audio, so playback starts immediately. Skip warmup when
2770                        # crossfade data from the previous track is pending — we need a full
2771                        # buffer for the mix.
2772                        if warmup_bytes < warmup_size and not last_fadeout_part:
2773                            yield chunk
2774                            warmup_bytes += len(chunk)
2775                            bytes_written += len(chunk)
2776                            del chunk
2777                            continue
2778
2779                        if not last_fadeout_part:
2780                            # the tail is being held back, so the audio the next transition
2781                            # blends in can be gathered alongside it instead of after it
2782                            incoming_prefetcher.ensure_started(
2783                                queue,
2784                                queue_track,
2785                                item_crossfade_mode,
2786                                standard_crossfade_duration,
2787                            )
2788
2789                        # accumulate chunks in the crossfade buffer: the outgoing tail
2790                        # window, or (at a boundary) whatever of the incoming overlap
2791                        # arrived before the mix starts. The window is whatever the
2792                        # source has banked ahead of playback right now.
2793                        crossfade_buffer.extend(chunk)
2794                        del chunk
2795                        hold_target = (
2796                            tail_hold.hold_target(crossfade_buffer_size, frame_size)
2797                            if tail_hold is not None
2798                            else 0
2799                        )
2800                        if not last_fadeout_part and len(crossfade_buffer) <= hold_target:
2801                            await asyncio.sleep(0)
2802                            continue
2803                        # handle crossfade of previous track and new track
2804                        if (
2805                            last_fadeout_part
2806                            and last_streamdetails
2807                            and crossfade_smart_fade is not None
2808                            and last_play_log_entry is not None
2809                        ):
2810                            self.logger.debug(
2811                                "Starting the transition into %s with %.1fs of its overlap"
2812                                " in hand (%.1fs prefetched, %.1fs build,"
2813                                " %.1fs was resident at the boundary)",
2814                                queue_track.name,
2815                                len(crossfade_buffer) / pcm_sample_size,
2816                                prefetched_size / pcm_sample_size,
2817                                build_seconds,
2818                                collect_resident,
2819                            )
2820                            # The mixer consumes the incoming overlap as it arrives and
2821                            # emits the blend at that same pace, so the transition
2822                            # streams instead of first collecting the whole overlap.
2823                            overlap_overshoot = bytearray()
2824                            mix_start_collected = len(crossfade_buffer)
2825                            overlap_pulled = 0
2826
2827                            def _note_overlap_bytes(
2828                                count: int, hold: _TailHold | None = tail_hold
2829                            ) -> None:
2830                                # the mixer reads the stream itself for the length of the
2831                                # overlap; noting those bytes as they arrive keeps the
2832                                # holdback's clock running, where one update afterwards
2833                                # would read as a suspension and bank a false surplus
2834                                nonlocal overlap_pulled
2835                                overlap_pulled += count
2836                                if hold is not None:
2837                                    hold.note_bytes(count)
2838
2839                            overlap_stream = _incoming_overlap_stream(
2840                                bytes(crossfade_buffer),
2841                                item_stream,
2842                                incoming_crossfade_size,
2843                                overlap_overshoot,
2844                                _note_overlap_bytes,
2845                            )
2846                            crossfade_buffer = bytearray()
2847                            # The mix output is split live as it flows: the first
2848                            # fadeout_share bytes are the outgoing track's (its held
2849                            # tail, processed), the rest belong to this one. Credited
2850                            # per chunk, because a single correction afterwards would
2851                            # leave the queue's position mapping on the wrong track
2852                            # for the whole (source-paced) duration of the blend. The
2853                            # pre-counted tail makes way for that live credit.
2854                            fadeout_share_seconds = (
2855                                timing_info.pre_crossfade_duration + timing_info.crossfade_duration
2856                            )
2857                            fadeout_share = int(fadeout_share_seconds * pcm_sample_size)
2858                            fadeout_share = (fadeout_share // frame_size) * frame_size
2859                            assert last_play_log_entry.seconds_streamed is not None
2860                            last_play_log_entry.seconds_streamed -= (
2861                                len(last_fadeout_part) / pcm_sample_size
2862                            )
2863                            mix_stream = self.smart_fades_mixer.mix(
2864                                crossfade_smart_fade,
2865                                fade_in_part=overlap_stream,
2866                                fade_out_part=last_fadeout_part,
2867                                pcm_format=pcm_format,
2868                            )
2869                            try:
2870                                crossfade_bytes_written = 0
2871                                # closed before item_stream on an aborted flow: its
2872                                # teardown stops the feeder that still holds a read
2873                                # on item_stream, which must not be closed mid-read
2874                                async with aclosing(mix_stream):
2875                                    async for mix_chunk in mix_stream:
2876                                        yield mix_chunk
2877                                        outgoing_part = min(
2878                                            len(mix_chunk),
2879                                            max(0, fadeout_share - crossfade_bytes_written),
2880                                        )
2881                                        last_play_log_entry.seconds_streamed += (
2882                                            outgoing_part / pcm_sample_size
2883                                        )
2884                                        bytes_written += len(mix_chunk) - outgoing_part
2885                                        outgoing_emitted += outgoing_part
2886                                        crossfade_bytes_written += len(mix_chunk)
2887                                remaining_bytes = bytes(overlap_overshoot)
2888                            except Exception as mix_err:
2889                                if crossfade_bytes_written:
2890                                    # partial mix already played — concat'd tail would duplicate audio
2891                                    raise
2892                                self.logger.warning(
2893                                    "Crossfade mixer failed for %s, falling back to simple concat: %s",
2894                                    queue_track.name,
2895                                    mix_err,
2896                                )
2897                                # the tail was un-counted for the live credit above;
2898                                # it now plays as ordinary outgoing audio
2899                                last_play_log_entry.seconds_streamed += (
2900                                    len(last_fadeout_part) / pcm_sample_size
2901                                )
2902                                for pcm_slice in iter_pcm_slices(
2903                                    last_fadeout_part, pcm_format, 1000
2904                                ):
2905                                    yield pcm_slice
2906                                    await asyncio.sleep(0)
2907                                outgoing_emitted += len(last_fadeout_part)
2908                                crossfade_bytes_written = 0
2909                                remaining_bytes = b""
2910                                # mix failed — undo the eager seek_position
2911                                queue_track.streamdetails.seek_position = raw_seek_position
2912                                # The mixer teardown cancels its feeder, which was
2913                                # likely parked reading item_stream - that ends the
2914                                # stream itself. Play the track from a fresh stream
2915                                # (its buffer still holds what the mixer consumed)
2916                                # rather than silently losing its body.
2917                                await item_stream.aclose()
2918                                fallback_stream = self.get_queue_item_stream(
2919                                    queue_track,
2920                                    pcm_format=pcm_format,
2921                                    seek_position=int(raw_seek_position),
2922                                    playback_speed=track_playback_speed,
2923                                    raise_on_error=False,
2924                                    session_id=flow_session_id,
2925                                )
2926                                async with aclosing(fallback_stream):
2927                                    async for fallback_chunk in fallback_stream:
2928                                        if _superseded():
2929                                            return
2930                                        yield fallback_chunk
2931                                        bytes_written += len(fallback_chunk)
2932                            if crossfade_bytes_written:
2933                                # the blend really played, so credit both of its sides with it
2934                                for faded_item in (queue_track, outgoing_queue_track):
2935                                    if faded_item is None:
2936                                        continue
2937                                    self._report_crossfade_mode(
2938                                        queue.queue_id,
2939                                        faded_item,
2940                                        pcm_format,
2941                                        applied_mode,
2942                                        flow_session_id,
2943                                        overlay_enabled=overlay_active(queue),
2944                                    )
2945                            if remaining_bytes:
2946                                for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2947                                    yield pcm_slice
2948                                    await asyncio.sleep(0)
2949                                bytes_written += len(remaining_bytes)
2950                                del remaining_bytes
2951                            # the position was reported for the planned overlap; an
2952                            # incoming stream that ended short blended less than that,
2953                            # and the track must not be reported past its own audio
2954                            blended = max(
2955                                0, mix_start_collected + overlap_pulled - len(overlap_overshoot)
2956                            )
2957                            queue_track.streamdetails.seek_position = min(
2958                                queue_track.streamdetails.seek_position,
2959                                raw_seek_position
2960                                + blended / pcm_sample_size * track_playback_speed,
2961                            )
2962                            last_fadeout_part = b""
2963                            last_streamdetails = None
2964                            last_queue_track = None
2965                            crossfade_buffer = bytearray()
2966                            warmup_bytes = 0
2967
2968                        # yield everything above the current holdback window; the
2969                        # slice can run short of a whole second when the window is
2970                        # small, so credit what is actually yielded - a nominal
2971                        # full-second credit inflates the play log and pins the
2972                        # queue's position mapping to the wrong track
2973                        while len(crossfade_buffer) > hold_target:
2974                            pcm_slice = bytes(crossfade_buffer[:pcm_sample_size])
2975                            yield pcm_slice
2976                            bytes_written += len(pcm_slice)
2977                            del crossfade_buffer[: len(pcm_slice)]
2978                            await asyncio.sleep(0)
2979
2980                # A source error after partial audio must not look like a completed item.
2981                # Progress reporting skips items with stream_error, so the item is not
2982                # marked played; move on to the next queue item like the zero-audio path.
2983                if first_chunk_received and queue_track.streamdetails.stream_error:
2984                    if _superseded():
2985                        return
2986                    self.logger.warning(
2987                        "Track %s (%s) on queue %s aborted by a stream error - skipping",
2988                        queue_track.name,
2989                        queue_track.streamdetails.uri,
2990                        queue.display_name,
2991                    )
2992                    # the audio sent so far will still play out; keep the play log entry
2993                    # honest about how much of this item was actually streamed
2994                    play_log_entry.seconds_streamed = bytes_written / pcm_sample_size
2995                    if last_fadeout_part:
2996                        # crossfade into this item never happened — undo the eager seek_position
2997                        queue_track.streamdetails.seek_position = raw_seek_position
2998                    continue
2999
3000                #### HANDLE END OF TRACK
3001                if not first_chunk_received:
3002                    self.logger.warning(
3003                        "Track %s (%s) on queue %s produced no audio data - skipping",
3004                        queue_track.name,
3005                        queue_track.streamdetails.uri if queue_track.streamdetails else "unknown",
3006                        queue.display_name,
3007                    )
3008                    queue_track.streamdetails.stream_error = True
3009                    play_log_entry.seconds_streamed = 0
3010                    if last_fadeout_part:
3011                        queue_track.streamdetails.seek_position = raw_seek_position
3012                    continue
3013                if last_fadeout_part:
3014                    # edge case: we did not get enough data to make the crossfade
3015                    # attribute these bytes to the previous track (they are its tail)
3016                    for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
3017                        yield pcm_slice
3018                        await asyncio.sleep(0)
3019                    outgoing_emitted += len(last_fadeout_part)
3020                    # no crossfade happened — undo the eager seek_position
3021                    queue_track.streamdetails.seek_position = raw_seek_position
3022                    # full tail was pre-counted and is now yielded as-is
3023                    last_fadeout_part = b""
3024                # a fade needs enough of the outgoing track to overlap with; a holdback that
3025                # armed late (or not at all) leaves less than that
3026                min_fade_out_size = int(pcm_sample_size * MIN_CROSSFADE_DURATION)
3027                if len(crossfade_buffer) >= min_fade_out_size and self.crossfade_allowed(
3028                    queue_track,
3029                    crossfade_mode=item_crossfade_mode,
3030                    player_id=queue.queue_id,
3031                    flow_mode=True,
3032                ):
3033                    last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:])
3034                    last_streamdetails = queue_track.streamdetails
3035                    last_queue_track = queue_track
3036                    last_play_log_entry = play_log_entry
3037                    remaining_bytes = bytes(crossfade_buffer[:-crossfade_buffer_size])
3038                    if remaining_bytes:
3039                        for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
3040                            yield pcm_slice
3041                            await asyncio.sleep(0)
3042                        bytes_written += len(remaining_bytes)
3043                    del remaining_bytes
3044                elif item_crossfade_mode != CrossfadeMode.DISABLED and crossfade_buffer:
3045                    bytes_written += len(crossfade_buffer)
3046                    for pcm_slice in iter_pcm_slices(bytes(crossfade_buffer), pcm_format, 1000):
3047                        yield pcm_slice
3048                        await asyncio.sleep(0)
3049                crossfade_buffer = bytearray()
3050                # one flow stream feeds the whole queue, so the lead it earned belongs to
3051                # the next track in it too; remeasuring per track throws it away
3052                if tail_hold is not None:
3053                    flow_lead = min(
3054                        tail_hold.banked_lead(bytes_written + outgoing_emitted),
3055                        float(SMART_CROSSFADE_DURATION),
3056                    )
3057                    flow_lead_at = asyncio.get_event_loop().time()
3058                    self.logger.debug(
3059                        "Flow stream for queue %s banked a lead of %.1fs after %s",
3060                        queue.display_name,
3061                        flow_lead,
3062                        queue_track.name,
3063                    )
3064
3065                # update duration details based on the actual pcm data we sent
3066                # this also accounts for crossfade and silence stripping
3067                seconds_streamed = bytes_written / pcm_sample_size
3068                queue_track.streamdetails.seconds_streamed = seconds_streamed
3069                play_log_entry.seconds_streamed = seconds_streamed
3070                # an externally aborted source ends in a clean EOF mid-track, so the
3071                # streamed length must not be written back as the item's duration
3072                source_buffer = queue_track.streamdetails.buffer
3073                source_aborted = source_buffer is not None and source_buffer.cancelled
3074                if not source_aborted:
3075                    # the held-back crossfade tail still counts as this track's media-time
3076                    tail_seconds = len(last_fadeout_part) / pcm_sample_size
3077                    # streamdetails.duration is in media-time; seconds_streamed is stream-time
3078                    # (post-atempo), so we scale by the track's playback_speed to recover media-time.
3079                    queue_track.streamdetails.duration = int(
3080                        queue_track.streamdetails.seek_position
3081                        + (seconds_streamed + tail_seconds) * track_playback_speed
3082                    )
3083                    # propagate accurate duration to queue_item so UI displays it
3084                    queue_track.duration = queue_track.streamdetails.duration
3085                    play_log_entry.duration = queue_track.streamdetails.duration
3086                if last_play_log_entry is play_log_entry and last_fadeout_part:
3087                    # Pre-count the full crossfade tail so the queue index calculation
3088                    # doesn't undercount while waiting for the next track's crossfade mix.
3089                    # This will be corrected to crossfade_total/2 once the mix completes.
3090                    assert play_log_entry.seconds_streamed is not None
3091                    play_log_entry.seconds_streamed += len(last_fadeout_part) / pcm_sample_size
3092                self.logger.debug(
3093                    "Finished Streaming queue track: %s (%s) on queue %s",
3094                    queue_track.streamdetails.uri,
3095                    queue_track.name,
3096                    queue.display_name,
3097                )
3098        finally:
3099            await incoming_prefetcher.close()
3100        #### HANDLE END OF QUEUE FLOW STREAM
3101        # skip end-of-queue bookkeeping if a newer producer has superseded us;
3102        # the new producer owns queue_buffer_completed and the play log now
3103        if _superseded():
3104            self.logger.debug(
3105                "Flow stream for queue %s superseded - skipping end-of-queue handling",
3106                queue.display_name,
3107            )
3108            return
3109        # end of queue flow: make sure we yield the last_fadeout_part
3110        if last_fadeout_part:
3111            for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
3112                yield pcm_slice
3113                await asyncio.sleep(0)
3114            # correct seconds streamed - the duration already includes the tail
3115            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
3116            streamdetails = queue_track.streamdetails
3117            assert streamdetails is not None
3118            streamdetails.seconds_streamed = (
3119                streamdetails.seconds_streamed or 0
3120            ) + last_part_seconds
3121            # also update the play log entry so elapsed time tracking stays in sync
3122            if last_play_log_entry:
3123                assert last_play_log_entry.seconds_streamed is not None
3124                # full tail was pre-counted and is now yielded as-is
3125                last_play_log_entry.duration = streamdetails.duration
3126            last_fadeout_part = b""
3127        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
3128        # only signal completion if we are still the active producer — a later
3129        # producer would (incorrectly) see this as its own completion otherwise
3130        if not _superseded():
3131            # inform the queue controller that all audio data has been generated
3132            # so it can handle the case where new items were added after the flow stream ended
3133            self.mass.player_queues.queue_buffer_completed(queue.queue_id, queue_exhausted)
3134
3135    async def get_overlay_mixed_stream(
3136        self,
3137        queue: PlayerQueue,
3138        audio_input: AsyncGenerator[bytes],
3139        pcm_format: AudioFormat,
3140    ) -> AsyncGenerator[bytes]:
3141        """
3142        Mix the queue's audio overlay (looping sound effect) into the given PCM stream.
3143
3144        The mixed output has the exact same PCM format, duration and chunking as the
3145        input stream. If the overlay source can not be resolved, the original stream
3146        is passed through unchanged so playback is never interrupted.
3147
3148        :param queue: The PlayerQueue holding the overlay source and volume.
3149        :param audio_input: The audio stream (raw PCM in ``pcm_format``) to mix into.
3150        :param pcm_format: PCM format of both the input and the mixed output.
3151        """
3152        overlay_input = await self._resolve_overlay_input(queue)
3153        if overlay_input is None:
3154            # overlay source unavailable: degrade gracefully to music-only
3155            async for chunk in audio_input:
3156                yield chunk
3157            return
3158        async for chunk in get_ffmpeg_overlay_stream(
3159            audio_input=audio_input,
3160            overlay_input=overlay_input,
3161            pcm_format=pcm_format,
3162            overlay_volume=queue.overlay_volume,
3163            chunk_size=pcm_format.pcm_sample_size,
3164        ):
3165            yield chunk
3166
3167    def crossfade_allowed(
3168        self,
3169        queue_item: QueueItem,
3170        crossfade_mode: CrossfadeMode,
3171        player_id: str,
3172        flow_mode: bool = False,
3173        next_queue_item: QueueItem | None = None,
3174        sample_rate: int | None = None,
3175        next_sample_rate: int | None = None,
3176    ) -> bool:
3177        """Get the crossfade config for a queue item."""
3178        if crossfade_mode == CrossfadeMode.DISABLED:
3179            return False
3180        if not (self.mass.player_queues.get(queue_item.queue_id)):
3181            return False  # just a guard
3182        if not (self.mass.players.get_player(player_id)):
3183            return False  # just a guard
3184        if queue_item.media_type != MediaType.TRACK:
3185            self.logger.debug("Skipping crossfade: current item is not a track")
3186            return False
3187        # check if the next item is part of the same album
3188        next_item = next_queue_item or self.mass.player_queues.get_next_item(
3189            queue_item.queue_id, queue_item.queue_item_id
3190        )
3191        if not next_item:
3192            # there is no next item!
3193            return False
3194        # check if next item is a track
3195        if next_item.media_type != MediaType.TRACK:
3196            self.logger.debug("Skipping crossfade: next item is not a track")
3197            return False
3198        # an item picks up its library album only once it is loaded, so a queue fed straight
3199        # from a provider can hold the provider album on the side that is not loaded yet.
3200        # Matching on the provider mappings recognises both shapes as the same album; the
3201        # uri-based equality of the album objects does not.
3202        if (
3203            isinstance(queue_item.media_item, Track)
3204            and isinstance(next_item.media_item, Track)
3205            and queue_item.media_item.album
3206            and next_item.media_item.album
3207            and compare_item_ids(queue_item.media_item.album, next_item.media_item.album)
3208            and not self.mass.config.get_raw_core_config_value(
3209                "streams", CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
3210            )
3211        ):
3212            # in general, crossfade is not desired for tracks of the same (gapless) album
3213            # because we have no accurate way to determine if the album is gapless or not,
3214            # for now we just never crossfade between tracks of the same album
3215            self.logger.debug("Skipping crossfade: next item is part of the same album")
3216            return False
3217
3218        # check if we're allowed to crossfade on different sample rates
3219        if (
3220            not flow_mode
3221            and sample_rate
3222            and next_sample_rate
3223            and sample_rate != next_sample_rate
3224            and not self.mass.config.get_raw_player_config_value(
3225                player_id,
3226                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key,
3227                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.default_value,
3228            )
3229        ):
3230            self.logger.debug(
3231                "Skipping crossfade: player(protocol) does not support gapless playback "
3232                "with different sample rates (%s vs %s)",
3233                sample_rate,
3234                next_sample_rate,
3235            )
3236            return False
3237
3238        return True
3239
3240    def clear_crossfade_data(self, queue_id: str) -> None:
3241        """
3242        Clear any pending crossfade data for a queue.
3243
3244        :param queue_id: The queue ID to clear crossfade data for.
3245        """
3246        if queue_id in self._crossfade_data:
3247            self.logger.debug("Clearing crossfade data for queue %s", queue_id)
3248            del self._crossfade_data[queue_id]
3249        # the player's buffer is no longer accounted for, so the next item must
3250        # re-earn its holdback rather than trust a lead measured before the break
3251        self._playback_lead.pop(queue_id, None)
3252        # and release anything waiting on a fade this queue will never finish mixing
3253        if pending := self._crossfade_pending.pop(queue_id, None):
3254            pending[1].set()
3255
3256    async def get_shoutcast_stream(
3257        self, url: str, streamdetails: StreamDetails
3258    ) -> AsyncGenerator[bytes]:
3259        """
3260        Yield audio from a legacy Shoutcast server, with ICY metadata parsed inline.
3261
3262        :param url: Shoutcast stream URL.
3263        :param streamdetails: StreamDetails to update with ICY metadata as it arrives.
3264        """
3265        self.logger.debug("Start streaming from legacy Shoutcast server: %s", url)
3266
3267        parsed = urlparse(url)
3268        host = parsed.hostname
3269        port = parsed.port or 80
3270        path = parsed.path or "/"
3271        if parsed.query:
3272            path = f"{path}?{parsed.query}"
3273
3274        try:
3275            # Open raw socket connection
3276            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=30)
3277        except TimeoutError as err:
3278            raise AudioError(f"Timeout connecting to Shoutcast stream {url}") from err
3279        except (OSError, ConnectionError) as err:
3280            raise AudioError(f"Failed to connect to Shoutcast stream {url}") from err
3281
3282        try:
3283            # Send HTTP request with ICY metadata header
3284            request = (
3285                f"GET {path} HTTP/1.1\r\n"
3286                f"Host: {host}\r\n"
3287                f"User-Agent: {HTTP_HEADERS['User-Agent']}\r\n"
3288                f"Icy-MetaData: 1\r\n\r\n"
3289            )
3290            writer.write(request.encode())
3291            await writer.drain()
3292
3293            # Read and parse response line
3294            try:
3295                response_line = await asyncio.wait_for(reader.readline(), timeout=10)
3296            except TimeoutError as err:
3297                raise AudioError("Timeout reading Shoutcast response") from err
3298
3299            if not response_line.startswith(b"ICY"):
3300                raise InvalidDataError("Invalid Shoutcast response")
3301
3302            # Read headers until empty line
3303            headers: dict[str, str] = {}
3304            while True:
3305                try:
3306                    line = await asyncio.wait_for(reader.readline(), timeout=5)
3307                except TimeoutError as err:
3308                    raise AudioError("Timeout reading Shoutcast headers") from err
3309
3310                if line in (b"\r\n", b"\n", b""):
3311                    break
3312
3313                if b":" in line:
3314                    try:
3315                        key, value = line.decode("latin-1", errors="ignore").split(":", 1)
3316                        headers[key.strip().lower()] = value.strip()
3317                    except UnicodeDecodeError, ValueError:
3318                        continue
3319
3320            # Get metadata interval
3321            meta_int_str = headers.get("icy-metaint")
3322            if not meta_int_str:
3323                raise InvalidDataError("No icy-metaint header in Shoutcast response")
3324
3325            try:
3326                meta_int = int(meta_int_str)
3327            except ValueError as err:
3328                raise InvalidDataError("Invalid icy-metaint value") from err
3329
3330            self.logger.debug("Connected to Shoutcast stream %s (icy-metaint: %s)", url, meta_int)
3331
3332            # Stream audio data with metadata parsing
3333            while True:
3334                try:
3335                    # Read audio chunk
3336                    audio_chunk = await reader.readexactly(meta_int)
3337                    yield audio_chunk
3338
3339                    # Read metadata length
3340                    meta_byte = await reader.readexactly(1)
3341                    if meta_byte == b"\x00":
3342                        continue
3343
3344                    meta_length = ord(meta_byte) * 16
3345                    meta_data = await reader.readexactly(meta_length)
3346                    self._parse_icy_metadata(meta_data, streamdetails)
3347
3348                except asyncio.exceptions.IncompleteReadError:
3349                    # End of stream
3350                    break
3351
3352        finally:
3353            writer.close()
3354            await writer.wait_closed()
3355
3356    # --- Private methods ---
3357
3358    def _notify_provider_streamed(
3359        self, streamdetails: StreamDetails, finished: bool, seconds_streamed: float
3360    ) -> None:
3361        """Report a (mostly) streamed item back to the provider that owns it."""
3362        if not finished and seconds_streamed < 90:
3363            return
3364        provider = self.mass.get_provider(streamdetails.provider)
3365        # plugin providers serve playable items too, but on_streamed is MusicProvider-only
3366        if provider is None or provider.type != ProviderType.MUSIC:
3367            return
3368        music_prov = cast("MusicProvider", provider)
3369        self.mass.create_task(music_prov.on_streamed(streamdetails))
3370
3371    def _get_volume_normalization_preference(
3372        self, streamdetails: StreamDetails
3373    ) -> VolumeNormalizationMode:
3374        """Return the configured normalization preference for the stream's media type."""
3375        conf_key = (
3376            CONF_VOLUME_NORMALIZATION_RADIO
3377            if streamdetails.media_type == MediaType.RADIO
3378            else CONF_VOLUME_NORMALIZATION_TRACKS
3379        )
3380        preference = VolumeNormalizationMode(
3381            self.mass.streams.get_config_value(conf_key, return_type=str)
3382        )
3383        # a stored value the options never offered is not a preference: nothing
3384        # validates a saved config value against them
3385        if preference in OUTCOME_ONLY_NORMALIZATION_MODES:
3386            return DEFAULT_VOLUME_NORMALIZATION_MODE
3387        return preference
3388
3389    def _update_radio_stream_metadata(
3390        self,
3391        streamdetails: StreamDetails,
3392        artist: str | None,
3393        title: str,
3394        image_url: str | None = None,
3395        album: str | None = None,
3396    ) -> None:
3397        """
3398        Update radio stream metadata and trigger artwork lookup.
3399
3400        :param streamdetails: The stream details to update.
3401        :param artist: Artist name (will be normalized).
3402        :param title: Track title (will be cleaned for display).
3403        :param image_url: Optional image URL from stream metadata.
3404        :param album: Optional album name.
3405        """
3406        station_image_url = image_url or self.mass.metadata.get_radio_stream_station_image(
3407            streamdetails
3408        )
3409        artist_normalized = (
3410            self.mass.metadata.normalize_radio_artist_name(artist) if artist else None
3411        )
3412        display_title, _ = parse_title_and_version(title, strip_for_display=True)
3413
3414        streamdetails.stream_metadata = StreamMetadata(
3415            title=display_title,
3416            artist=artist_normalized,
3417            album=album,
3418            image_url=station_image_url,
3419        )
3420        streamdetails.stream_metadata_last_updated = time.time()
3421        if streamdetails.queue_id:
3422            self.mass.player_queues.signal_update(streamdetails.queue_id)
3423
3424        # Fetch artwork in background (track, album then artist)
3425        if artist and title and not image_url:
3426            self.mass.call_later(
3427                0.2,
3428                self.mass.metadata.update_radio_stream_artwork,
3429                streamdetails,
3430                task_id=f"update_radio_artwork_{streamdetails.queue_id}",
3431            )
3432
3433    async def _cache_radio_result(
3434        self,
3435        url: str,
3436        stream_type: StreamType,
3437        resolved_url: str | None = None,
3438    ) -> tuple[str, StreamType]:
3439        """Cache and return a radio stream resolution result."""
3440        result = (resolved_url or url, stream_type)
3441        await self.mass.cache.set(
3442            url,
3443            result,
3444            expiration=3600 * 3,
3445            provider=CACHE_PROVIDER,
3446            category=CACHE_CATEGORY_RESOLVED_RADIO_URL,
3447        )
3448        return result
3449
3450    async def _handle_client_error_for_radio_stream(
3451        self, url: str, err: aiohttp.ClientError, fallback_stream_type: StreamType
3452    ) -> tuple[str, StreamType]:
3453        """Handle aiohttp client errors during radio stream resolution."""
3454        # Prefer the final post-redirect URL: aiohttp follows redirects before raising,
3455        # but the original url may just point at a redirector rather than the ICY endpoint.
3456        request_info = getattr(err, "request_info", None)
3457        validate_url = str(request_info.url) if request_info is not None else url
3458
3459        # Check if this is a Shoutcast/ICY response that aiohttp can't parse
3460        if isinstance(err, aiohttp.ClientResponseError) and "ICY" in str(err).upper():
3461            self.logger.debug(
3462                "ICY response detected for %s, validating Shoutcast stream", validate_url
3463            )
3464            if await self._validate_shoutcast_stream(validate_url):
3465                return await self._cache_radio_result(
3466                    url, StreamType.SHOUTCAST, resolved_url=validate_url
3467                )
3468            self.logger.warning(
3469                "ICY response detected but Shoutcast validation failed for %s", validate_url
3470            )
3471            return await self._cache_radio_result(
3472                url, fallback_stream_type, resolved_url=validate_url
3473            )
3474
3475        # Other aiohttp errors - might still be Shoutcast, check it
3476        self.logger.debug("aiohttp error for %s, checking if legacy Shoutcast stream", validate_url)
3477        if await self._validate_shoutcast_stream(validate_url):
3478            return await self._cache_radio_result(
3479                url, StreamType.SHOUTCAST, resolved_url=validate_url
3480            )
3481
3482        # Unknown error - still try to stream
3483        self.logger.warning(
3484            "Failed to parse radio URL %s: %s - attempting direct stream", validate_url, str(err)
3485        )
3486        return await self._cache_radio_result(url, fallback_stream_type, resolved_url=validate_url)
3487
3488    async def _get_audio_buffer(
3489        self,
3490        queue_item: QueueItem,
3491        seek_position_ms: int,
3492        reason: str,
3493        capacity_wait_timeout: float,
3494        allow_provider_match: bool,
3495    ) -> AudioBuffer:
3496        """
3497        Create or reuse a ready AudioBuffer within one queue-item preparation lock.
3498
3499        :param queue_item: Queue item whose source should be buffered.
3500        :param seek_position_ms: Position in milliseconds to start from.
3501        :param reason: Caller context for logging.
3502        :param capacity_wait_timeout: Total seconds to spend waiting for source capacity.
3503        :param allow_provider_match: Whether an on-demand cross-provider match may widen
3504            the candidates when all are saturated.
3505        """
3506        loop = asyncio.get_running_loop()
3507        # the playback intent lives on the details we start from; keep it across a reselection
3508        initial_streamdetails = queue_item.streamdetails
3509        seek_position = (
3510            int(initial_streamdetails.seek_position)
3511            if initial_streamdetails
3512            else seek_position_ms // 1000
3513        )
3514        fade_in = bool(initial_streamdetails and initial_streamdetails.fade_in)
3515        prefer_album_loudness = bool(
3516            initial_streamdetails and initial_streamdetails.prefer_album_loudness
3517        )
3518        all_candidate_instances = {
3519            provider.instance_id
3520            for mapping in (
3521                queue_item.media_item.provider_mappings if queue_item.media_item else ()
3522            )
3523            if mapping.available
3524            for provider in self._get_mapping_providers(mapping)
3525        }
3526        if initial_streamdetails is not None:
3527            all_candidate_instances.add(initial_streamdetails.provider)
3528        # a track may also exist on streaming providers it has no mapping for yet; such a
3529        # match is only searched once, and only when every known candidate is saturated
3530        match_pending = (
3531            allow_provider_match
3532            and isinstance(queue_item.media_item, Track)
3533            and self._has_alternative_match_providers(queue_item.media_item)
3534        )
3535
3536        deadline = loop.time() + capacity_wait_timeout
3537        busy_instances: set[str] = set()
3538        final_pass = False
3539        last_capacity_error: ProviderStreamLimitError | None = None
3540        last_failed_streamdetails: StreamDetails | None = None
3541        while True:
3542            if queue_item.streamdetails is None or (
3543                queue_item.streamdetails.provider in busy_instances and not final_pass
3544            ):
3545                try:
3546                    queue_item.streamdetails = await self.get_stream_details(
3547                        queue_item,
3548                        seek_position=seek_position,
3549                        fade_in=fade_in,
3550                        prefer_album_loudness=prefer_album_loudness,
3551                        excluded_provider_instances=busy_instances,
3552                    )
3553                except (AudioError, MediaNotFoundError) as err:
3554                    if last_capacity_error is None:
3555                        raise
3556                    if final_pass:
3557                        # capacity was the root cause, surface the typed (actionable) error
3558                        raise last_capacity_error from err
3559                    # no usable alternative mapping: restore the capacity-blocked details
3560                    # and spend the remaining budget blocking on that provider's slot
3561                    final_pass = True
3562                    continue
3563                finally:
3564                    if queue_item.streamdetails is None:
3565                        # never leave the queue item without streamdetails on any exit,
3566                        # including a cancellation or a non-audio provider failure
3567                        queue_item.streamdetails = last_failed_streamdetails
3568            streamdetails = queue_item.streamdetails
3569            assert streamdetails is not None  # for type checking
3570            remaining = max(deadline - loop.time(), 0)
3571            alternatives_left = bool(
3572                all_candidate_instances - busy_instances - {streamdetails.provider}
3573            )
3574            # probe (0s) whenever a reselection can still follow: a free slot is still
3575            # acquired instantly, while a busy one fails fast instead of spending the
3576            # whole budget on this candidate. Block only on the last resort.
3577            source_wait = (
3578                0.0
3579                if (not final_pass and (alternatives_left or busy_instances or match_pending))
3580                else remaining
3581            )
3582            # record whose audio this is before it exists: a queue stop releases only the
3583            # buffers of the session it is tearing down, and details resolved by an earlier
3584            # session are reused as they are, so the claim has to be made where the buffer
3585            # is attached rather than where the details came from. The queue's own session
3586            # is the owner rather than the one a caller asks for: a superseded request that
3587            # reuses a live buffer must not take it from the session still playing it
3588            streamdetails.queue_session_id = (
3589                queue_data.session_id
3590                if (queue_data := self.mass.player_queues.queue_data_or_none(queue_item.queue_id))
3591                else None
3592            )
3593            try:
3594                return await AudioBuffer.get_buffer(
3595                    mass=self.mass,
3596                    streamdetails=streamdetails,
3597                    seek_position_ms=seek_position_ms,
3598                    wait_ready=True,
3599                    reason=reason,
3600                    source_wait_timeout=source_wait,
3601                )
3602            except ProviderStreamLimitError as err:
3603                last_capacity_error = err
3604                last_failed_streamdetails = streamdetails
3605                busy_instances.add(err.provider_instance)
3606                if final_pass or loop.time() >= deadline:
3607                    raise
3608                if all_candidate_instances.issubset(busy_instances):
3609                    discovered: set[str] = set()
3610                    if match_pending:
3611                        match_pending = False
3612                        try:
3613                            discovered = await self._discover_alternative_provider_mappings(
3614                                queue_item, busy_instances, max(deadline - loop.time(), 0)
3615                            )
3616                        except Exception as err:
3617                            # discovery is best-effort: any failure falls back to the
3618                            # final blocking wait instead of replacing the typed error
3619                            self.logger.warning(
3620                                "Alternative provider search for %s failed: %s",
3621                                queue_item.name,
3622                                err,
3623                            )
3624                    if discovered:
3625                        all_candidate_instances.update(discovered)
3626                    else:
3627                        # every candidate is saturated: one last blocking wait on the best one
3628                        busy_instances.clear()
3629                        final_pass = True
3630                queue_item.streamdetails = None
3631            except AudioError:
3632                if last_capacity_error is None or final_pass:
3633                    raise
3634                # a broken alternate must not turn a transient capacity miss into a hard
3635                # failure: restore the blocked details and spend the rest of the budget there
3636                queue_item.streamdetails = last_failed_streamdetails
3637                final_pass = True
3638
3639    def _get_streamdetail_candidates(
3640        self,
3641        provider_mappings: Iterable[ProviderMapping],
3642        preferred_providers: list[str],
3643        excluded_provider_instances: set[str],
3644    ) -> list[tuple[ProviderMapping, Provider]]:
3645        """
3646        Return mapping candidates in steering, quality, and instance-fallback order.
3647
3648        :param provider_mappings: Mappings attached to the media item.
3649        :param preferred_providers: Provider instances tried before widening to the rest.
3650        :param excluded_provider_instances: Provider instances unavailable to this attempt.
3651        :return: Ordered provider mapping candidates.
3652        """
3653        ordered_mappings = sorted(
3654            provider_mappings, key=lambda mapping: mapping.quality or 0, reverse=True
3655        )
3656        preferred_candidates: list[tuple[ProviderMapping, Provider]] = []
3657        fallback_candidates: list[tuple[ProviderMapping, Provider]] = []
3658        seen_candidates: set[tuple[str, str]] = set()
3659        for mapping in ordered_mappings:
3660            if not mapping.available:
3661                self.logger.debug("Skipping unavailable %s", mapping)
3662                continue
3663            for provider in self._get_mapping_providers(mapping):
3664                candidate_id = (provider.instance_id, mapping.item_id)
3665                if (
3666                    candidate_id in seen_candidates
3667                    or provider.instance_id in excluded_provider_instances
3668                ):
3669                    continue
3670                seen_candidates.add(candidate_id)
3671                candidate = (mapping, provider)
3672                if provider.instance_id in preferred_providers:
3673                    preferred_candidates.append(candidate)
3674                else:
3675                    fallback_candidates.append(candidate)
3676        return [*preferred_candidates, *fallback_candidates]
3677
3678    def _get_mapping_providers(self, mapping: ProviderMapping) -> list[Provider]:
3679        """
3680        Return the mapped provider followed by compatible instances of its streaming catalog.
3681
3682        :param mapping: Provider mapping whose item ID will be requested.
3683        :return: Loaded provider instances that can resolve the mapping.
3684        """
3685        providers: list[Provider] = []
3686        if (
3687            primary_provider := self.mass.get_provider(
3688                mapping.provider_instance, return_unavailable=True
3689            )
3690        ) and primary_provider.available:
3691            providers.append(primary_provider)
3692        # another account of the same streaming catalog serves the same item ID,
3693        # so it can stand in when the mapped instance can not
3694        for provider in self.mass.providers:
3695            if (
3696                not isinstance(provider, MusicProvider)
3697                or not provider.available
3698                or not provider.is_streaming_provider
3699                or provider.domain != mapping.provider_domain
3700                or provider in providers
3701            ):
3702                continue
3703            providers.append(provider)
3704        if not providers:
3705            self.logger.debug("Skipping %s - provider not available", mapping)
3706        return providers
3707
3708    def _is_match_candidate_provider(
3709        self, provider: MusicProvider, known_domains: set[str]
3710    ) -> bool:
3711        """
3712        Return whether a provider is eligible to search a track match on.
3713
3714        :param provider: Music provider to check.
3715        :param known_domains: Provider domains the track already has mappings for.
3716        """
3717        return (
3718            provider.available
3719            and provider.is_streaming_provider
3720            and ProviderFeature.SEARCH in provider.supported_features
3721            and provider.domain not in known_domains
3722            and MediaType.TRACK in provider.supported_media_types
3723        )
3724
3725    def _has_alternative_match_providers(self, media_item: Track) -> bool:
3726        """
3727        Return whether any configured streaming provider could carry an unmapped match.
3728
3729        :param media_item: Track whose existing mappings define the known provider domains.
3730        """
3731        known_domains = {mapping.provider_domain for mapping in media_item.provider_mappings}
3732        return any(
3733            self._is_match_candidate_provider(provider, known_domains)
3734            for provider in self.mass.music.providers
3735        )
3736
3737    async def _discover_alternative_provider_mappings(
3738        self, queue_item: QueueItem, busy_instances: set[str], remaining: float
3739    ) -> set[str]:
3740        """
3741        Search other streaming providers for the queue item's track and widen its mappings.
3742
3743        A found mapping is added to the media item (and persisted for library items) so the
3744        capacity reselection can continue on the discovered provider.
3745
3746        :param queue_item: Queue item whose track should be matched on another provider.
3747        :param busy_instances: Provider instances already known to be saturated.
3748        :param remaining: Seconds left of the caller's capacity budget.
3749        :return: Provider instances able to serve the discovered mappings.
3750        """
3751        media_item = queue_item.media_item
3752        if not isinstance(media_item, Track):
3753            return set()
3754        known_domains = {mapping.provider_domain for mapping in media_item.provider_mappings}
3755        eligible = [
3756            provider
3757            for provider in self.mass.music.providers
3758            if self._is_match_candidate_provider(provider, known_domains)
3759            and provider.instance_id not in busy_instances
3760            and provider.has_available_stream_slot
3761        ]
3762        if not eligible:
3763            return set()
3764        # mirror the playback user's provider steering for the search order
3765        if (
3766            (pq_data := self.mass.player_queues.queue_data_or_none(queue_item.queue_id))
3767            and pq_data.userid
3768            and (playback_user := await self.mass.webserver.auth.get_user(pq_data.userid))
3769            and playback_user.provider_filter
3770        ):
3771            preferred = set(playback_user.provider_filter)
3772            eligible.sort(key=lambda provider: provider.instance_id not in preferred)
3773        # one instance per domain: a found mapping widens to sibling instances anyway
3774        candidates: list[MusicProvider] = []
3775        for provider in eligible:
3776            if provider.domain in known_domains:
3777                continue
3778            known_domains.add(provider.domain)
3779            candidates.append(provider)
3780        # the track's own album is free, sufficient evidence for the strict compare and
3781        # avoids match_provider's multi-provider album lookup on every call
3782        ref_albums = [media_item.album] if isinstance(media_item.album, Album) else []
3783        matches: list[ProviderMapping] = []
3784        try:
3785            async with asyncio.timeout(min(STREAM_SLOT_MATCH_TIMEOUT, remaining)):
3786                for provider in candidates:
3787                    # one failing provider must not end the search on the others
3788                    try:
3789                        matches = await self.mass.music.tracks.match_provider(
3790                            media_item, provider, strict=True, ref_albums=ref_albums
3791                        )
3792                    except Exception as err:
3793                        self.logger.debug("Searching a match on %s failed: %s", provider.name, err)
3794                        continue
3795                    if matches:
3796                        break
3797        except TimeoutError:
3798            self.logger.debug("Searching an alternative provider for %s timed out", media_item.name)
3799        if not matches:
3800            return set()
3801        media_item.provider_mappings.update(matches)
3802        if media_item.provider == "library":
3803            # persist in the background so future plays have the mapping ahead of time;
3804            # cancellation of this playback must never interrupt the library write
3805            self.mass.create_task(
3806                self.mass.music.tracks.add_provider_mappings(media_item.item_id, matches)
3807            )
3808        self.logger.info(
3809            "All known sources for %s are at their stream limit, "
3810            "using a matching track found on %s",
3811            media_item.name,
3812            matches[0].provider_domain,
3813        )
3814        return {
3815            provider.instance_id
3816            for mapping in matches
3817            for provider in self._get_mapping_providers(mapping)
3818        }
3819
3820    async def _request_streamdetails(
3821        self,
3822        candidates: Iterable[tuple[ProviderMapping, Provider]],
3823        media_type: MediaType,
3824    ) -> StreamDetails | None:
3825        """
3826        Request stream details from ordered provider mapping candidates.
3827
3828        :param candidates: Candidates in mapping and compatible-instance order.
3829        :param media_type: Media type requested from each provider.
3830        :return: The first resolved stream details, or None when every candidate failed.
3831        :raises AudioError: The last (actionable) audio error when no candidate resolved.
3832        """
3833        last_audio_error: AudioError | None = None
3834        for mapping, provider in candidates:
3835            # music and plugin providers share this signature, so either type can own the item
3836            token = BYPASS_THROTTLER.set(True)
3837            try:
3838                stream_prov = cast("MusicProvider | PluginProvider", provider)
3839                return await stream_prov.get_stream_details(mapping.item_id, media_type)
3840            except AudioError as err:
3841                # remember the last one so its (actionable) message can be re-raised
3842                last_audio_error = err
3843                self.logger.warning("%s", err)
3844            except MusicAssistantError as err:
3845                self.logger.warning("%s", err)
3846            finally:
3847                BYPASS_THROTTLER.reset(token)
3848        if last_audio_error is not None:
3849            raise last_audio_error
3850        return None
3851
3852    async def _get_media_stream(
3853        self,
3854        streamdetails: StreamDetails,
3855        pcm_format: AudioFormat,
3856        seek_position: int,
3857        filter_params: list[str] | None,
3858        chunk_seconds: float,
3859    ) -> AsyncGenerator[bytes]:
3860        """
3861        Stream one provider source as raw PCM.
3862
3863        :param streamdetails: Details of the stream to fetch.
3864        :param pcm_format: Target PCM format the consumer expects.
3865        :param seek_position: Requested seek offset in seconds.
3866        :param filter_params: Optional ffmpeg filter expressions.
3867        :param chunk_seconds: Size of each yielded chunk in seconds of audio.
3868        """
3869        mass = self.mass
3870        logger = self.logger.getChild("media_stream")
3871        logger.log(VERBOSE_LOG_LEVEL, "Starting media stream for %s", streamdetails.uri)
3872        # copy: the args below are appended per call, while the StreamDetails is cached on
3873        # the queue item and reused across calls (retry, seek, background analysis)
3874        extra_input_args = list(streamdetails.extra_input_args or [])
3875        # the resolver below zeroes out seek_position where the seek is delegated to the
3876        # source itself, so keep the requested position for the duration writeback
3877        requested_seek_position = seek_position
3878
3879        # work out audio source for these streamdetails
3880        audio_source, seek_position, extra_input_args = await self._resolve_media_stream_source(
3881            streamdetails, seek_position, extra_input_args
3882        )
3883
3884        # pace ffmpeg at native rate for live sources; the producer (e.g.
3885        # librespot's pipe backend) may otherwise write faster than realtime.
3886        # The initial burst grants a small bounded read-ahead so downstream
3887        # jitter does not immediately underrun the player. Providers that need
3888        # different pacing can pass their own -re/-readrate args to override.
3889        if (
3890            streamdetails.media_type == MediaType.AUDIO_SOURCE
3891            and "-re" not in extra_input_args
3892            and "-readrate" not in extra_input_args
3893        ):
3894            extra_input_args += ["-readrate", "1", "-readrate_initial_burst", "0.5"]
3895
3896        # handle seek support
3897        if seek_position and streamdetails.duration and streamdetails.allow_seek:
3898            extra_input_args += ["-ss", str(int(seek_position))]
3899
3900        bytes_sent = 0
3901        finished = False
3902        cancelled = False
3903        first_chunk_received = False
3904        ffmpeg_loglevel = "debug" if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) else "info"
3905        ffmpeg_input_format = arriving_audio_format(streamdetails)
3906        ffmpeg_proc = FFMpeg(
3907            audio_input=audio_source,
3908            input_format=ffmpeg_input_format,
3909            output_format=pcm_format,
3910            filter_params=filter_params,
3911            extra_input_args=extra_input_args,
3912            collect_log_history=True,
3913            loglevel=ffmpeg_loglevel,
3914        )
3915
3916        try:
3917            await ffmpeg_proc.start()
3918            assert ffmpeg_proc.proc is not None  # for type checking
3919            if logger.isEnabledFor(VERBOSE_LOG_LEVEL):
3920                logger.log(
3921                    VERBOSE_LOG_LEVEL,
3922                    "Started media stream for %s - using streamtype: %s "
3923                    "- pcm format: %s - ffmpeg PID: %s",
3924                    streamdetails.uri,
3925                    streamdetails.stream_type,
3926                    pcm_format.content_type.value,
3927                    ffmpeg_proc.proc.pid,
3928                )
3929            else:
3930                logger.debug(
3931                    "Started media stream for %s - using streamtype: %s",
3932                    streamdetails.uri,
3933                    streamdetails.stream_type,
3934                )
3935            stream_start = mass.loop.time()
3936            chunk_size = calculate_content_length(pcm_format, chunk_seconds)
3937            chunk_iter = ffmpeg_proc.iter_chunked(chunk_size)
3938            while True:
3939                # Time the read, not the yield: catches a stalled source, ignores backpressure.
3940                read_timeout = (
3941                    STREAM_START_TIMEOUT if not first_chunk_received else STREAM_STALL_TIMEOUT
3942                )
3943                try:
3944                    async with asyncio.timeout(read_timeout):
3945                        chunk = await anext(chunk_iter)
3946                except StopAsyncIteration:
3947                    break
3948                except TimeoutError as err:
3949                    raise AudioError(f"Source stalled: no audio for {read_timeout}s") from err
3950                if not first_chunk_received:
3951                    # At this point ffmpeg has started and should now know the codec used
3952                    # for encoding the audio.
3953                    # Note: ffmpeg_proc.input_format is the same object as
3954                    # ffmpeg_input_format, so sample_rate / bit_depth / bit_rate
3955                    # parsed from the ffmpeg log already live on streamdetails too.
3956                    first_chunk_received = True
3957                    # Skip the codec_type writeback when the provider declared a
3958                    # decoded format: audio_format already holds the authoritative
3959                    # source codec and the probed value would just be the
3960                    # post-decode wire format (e.g. PCM for Spotify Connect).
3961                    if streamdetails.decoded_audio_format is None:
3962                        streamdetails.audio_format.codec_type = ffmpeg_proc.input_format.codec_type
3963                    # Some providers omit (or report 0 for) the item duration; ffmpeg can
3964                    # usually probe it from the source. Only apply when missing so we
3965                    # don't clobber an accurate provider value with a rounded one.
3966                    if ffmpeg_proc.parsed_duration is not None and not streamdetails.duration:
3967                        streamdetails.duration = ffmpeg_proc.parsed_duration
3968                    logger.debug(
3969                        "First chunk received after %.2f seconds (codec detected: %s)",
3970                        mass.loop.time() - stream_start,
3971                        ffmpeg_proc.input_format.codec_type,
3972                    )
3973                yield chunk
3974                bytes_sent += len(chunk)
3975
3976            # end of audio/track reached
3977            logger.debug("End of media stream reached for %s", streamdetails.uri)
3978            # wait until stderr also completed reading
3979            await ffmpeg_proc.wait_with_timeout(5)
3980            logger.log(
3981                VERBOSE_LOG_LEVEL,
3982                "FFmpeg process ended with return code %s for %s",
3983                ffmpeg_proc.returncode,
3984                streamdetails.uri,
3985            )
3986            # a nested source raises through the stdin feeder, where ffmpeg's own exit
3987            # would otherwise flatten it into a generic AudioError
3988            if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
3989                raise feeder_exception
3990            if ffmpeg_proc.returncode not in (0, None):
3991                log_trail = "\n".join(list(ffmpeg_proc.log_history)[-5:])
3992                raise AudioError(f"FFMpeg exited with code {ffmpeg_proc.returncode}: {log_trail}")
3993            if bytes_sent == 0:
3994                # edge case: no audio data was received at all
3995                raise AudioError("No audio was received")
3996            finished = True
3997        except (Exception, GeneratorExit, asyncio.CancelledError) as err:
3998            if isinstance(err, asyncio.CancelledError | GeneratorExit):
3999                # we were cancelled, just raise
4000                cancelled = True
4001                raise
4002            if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
4003                if isinstance(feeder_exception, ProviderStreamLimitError):
4004                    raise ffmpeg_proc.stdin_feeder_exception
4005                err = feeder_exception
4006            if isinstance(err, ProviderStreamLimitError):
4007                raise
4008            # dump the last 10 lines of the log in case of an unclean exit
4009            logger.warning("\n".join(list(ffmpeg_proc.log_history)[-10:]))
4010            raise AudioError(f"Error while streaming: {err}") from err
4011        finally:
4012            # An ffmpeg wedged on an input that will never deliver again pays close()'s
4013            # full drain - some 12 seconds in practice - under a held player lock,
4014            # before the SIGKILL that was always coming. Once the process has exited
4015            # close() is free, and it is what cleans up the stdin feeder behind it.
4016            if ffmpeg_proc.returncode is not None:
4017                await ffmpeg_proc.close()
4018            else:
4019                await ffmpeg_proc.kill()
4020            # determine how many seconds we've received
4021            # for pcm output we can calculate this easily
4022            seconds_received = bytes_sent / pcm_format.pcm_sample_size if bytes_sent else 0
4023            # store accurate duration, but only for a playthrough from the very start:
4024            # a seeked stream yields the remaining audio, not the item's full length
4025            if finished and not requested_seek_position and seconds_received:
4026                streamdetails.duration = int(seconds_received)
4027
4028            logger.log(
4029                VERBOSE_LOG_LEVEL,
4030                "stream %s (with code %s) for %s",
4031                "cancelled" if cancelled else "finished" if finished else "aborted",
4032                ffmpeg_proc.returncode,
4033                streamdetails.uri,
4034            )
4035
4036    def _report_crossfade_mode(
4037        self,
4038        queue_id: str,
4039        queue_item: QueueItem,
4040        pcm_format: AudioFormat,
4041        crossfade_mode: CrossfadeMode,
4042        session_id: str | None,
4043        *,
4044        overlay_enabled: bool,
4045    ) -> None:
4046        """
4047        Publish the crossfade that is actually applied to a queue item's audio.
4048
4049        :param queue_id: Queue the item is streamed from.
4050        :param queue_item: Queue item the fade touches.
4051        :param pcm_format: Shared PCM format leaving queue processing.
4052        :param crossfade_mode: Mode of the applied fade, SOURCE when the item's own
4053            source applies it, or DISABLED when none is applied.
4054        :param session_id: Queue session that owns processing-detail updates.
4055        :param overlay_enabled: Whether an overlay is mixed into this stream.
4056        """
4057        if session_id is None or queue_item.streamdetails is None:
4058            return
4059        self.mass.streams.audio_processing.update_item_context(
4060            queue_id=queue_id,
4061            session_id=session_id,
4062            queue_item_id=queue_item.queue_item_id,
4063            queue_processing=AudioQueueProcessing(
4064                pcm_format=pcm_format,
4065                playback_speed=cast(
4066                    "float", queue_item.extra_attributes.get("playback_speed", 1.0)
4067                ),
4068                crossfade_mode=crossfade_mode,
4069                overlay_active=overlay_enabled,
4070            ),
4071            alters_audio=queue_item.streamdetails.fade_in,
4072        )
4073
4074    async def _await_pending_crossfade(
4075        self, queue: PlayerQueue, queue_item: QueueItem
4076    ) -> CrossfadeData | None:
4077        """
4078        Wait, briefly, for a fade into this item that the outgoing stream is still mixing.
4079
4080        :param queue: The queue this request belongs to.
4081        :param queue_item: The item whose stream is starting.
4082        :return: The fade data if it landed in time, else None.
4083        """
4084        pending = self._crossfade_pending.get(queue.queue_id)
4085        if pending is None or pending[0] != queue_item.queue_item_id:
4086            return None
4087        handoff = pending[1]
4088        self.logger.debug(
4089            "Waiting up to %.1fs for the fade into %s being mixed for queue %s",
4090            CROSSFADE_HANDOFF_WAIT,
4091            queue_item.name,
4092            queue.display_name,
4093        )
4094        waited_from = asyncio.get_event_loop().time()
4095        with suppress(TimeoutError):
4096            await asyncio.wait_for(handoff.wait(), CROSSFADE_HANDOFF_WAIT)
4097        crossfade_data = self._crossfade_data.get(queue.queue_id)
4098        self.logger.debug(
4099            "Waited %.1fs for the fade into %s on queue %s - %s",
4100            asyncio.get_event_loop().time() - waited_from,
4101            queue_item.name,
4102            queue.display_name,
4103            "landed" if crossfade_data else "gave up",
4104        )
4105        return crossfade_data
4106
4107    async def _await_realtime_fade_source(self, streamdetails: StreamDetails) -> None:
4108        """
4109        Give a realtime incoming track a bounded chance to start delivering.
4110
4111        :param streamdetails: Stream details of the incoming (fade-in) track.
4112        """
4113        if not streamdetails.is_realtime:
4114            return
4115        loop = asyncio.get_event_loop()
4116        deadline = loop.time() + REALTIME_FADE_SOURCE_WAIT
4117        while True:
4118            audio_buffer = cast("AudioBuffer | None", streamdetails.buffer)
4119            if audio_buffer is not None:
4120                if audio_buffer.has_error:
4121                    return
4122                with suppress(TimeoutError):
4123                    await asyncio.wait_for(audio_buffer.ready.wait(), deadline - loop.time())
4124                return
4125            if loop.time() >= deadline:
4126                return
4127            # the buffer appears when the source's session starts producing
4128            await asyncio.sleep(0.1)
4129
4130    def _select_buffered_crossfade(
4131        self,
4132        streamdetails: StreamDetails,
4133        crossfade_mode: CrossfadeMode,
4134        standard_crossfade_duration: int,
4135        fade_out_seconds: float,
4136        playback_speed: float = 1.0,
4137    ) -> tuple[CrossfadeMode, float]:
4138        """
4139        Select the crossfade this boundary can carry.
4140
4141        The configured mode picks the fade; the held-back outgoing tail sizes its
4142        window, up to that mode's ceiling and to what the incoming track can supply.
4143        Too short a tail to blend at all means no fade rather than a different one.
4144
4145        :param streamdetails: Incoming track stream details.
4146        :param crossfade_mode: Requested crossfade mode.
4147        :param standard_crossfade_duration: Configured standard overlap in seconds.
4148        :param fade_out_seconds: Held-back outgoing tail in seconds.
4149        :param playback_speed: Incoming track playback-speed multiplier.
4150        :return: Effective mode and fade-in duration in seconds.
4151        """
4152        audio_buffer = streamdetails.buffer
4153        if (
4154            crossfade_mode == CrossfadeMode.DISABLED
4155            or playback_speed <= 0
4156            or audio_buffer is None
4157            or audio_buffer.has_error
4158            or not audio_buffer.is_valid()
4159            or not audio_buffer.ready.is_set()
4160        ):
4161            return CrossfadeMode.DISABLED, 0
4162
4163        # The blend streams, so the incoming window does not have to be resident:
4164        # it arrives while the blend plays. The tail we held back is what bounds it.
4165        window = min(
4166            SMART_CROSSFADE_DURATION
4167            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
4168            else standard_crossfade_duration,
4169            fade_out_seconds,
4170        )
4171        if audio_buffer.eof:
4172            # the source is done, so what is resident is all there will ever be
4173            window = min(window, audio_buffer.duration_available / playback_speed)
4174        if streamdetails.duration:
4175            # a short incoming track cannot supply a long overlap, and blending into
4176            # more than half of it would leave the listener no clean part of it. The
4177            # window is stream time, the track's remaining audio is media time.
4178            remaining_media = max(0.0, streamdetails.duration - streamdetails.seek_position)
4179            window = min(window, remaining_media / playback_speed / 2)
4180        if window < MIN_CROSSFADE_DURATION:
4181            return CrossfadeMode.DISABLED, 0
4182        self.logger.debug(
4183            "Using a %.1f second %s for %s",
4184            window,
4185            crossfade_mode.value,
4186            streamdetails.uri,
4187        )
4188        return crossfade_mode, window
4189
4190    async def _resolve_media_stream_source(
4191        self,
4192        streamdetails: StreamDetails,
4193        seek_position: int,
4194        extra_input_args: list[str],
4195    ) -> tuple[str | AsyncGenerator[bytes], int, list[str]]:
4196        """
4197        Resolve the input consumed by ffmpeg for the given stream details.
4198
4199        :param streamdetails: Details of the stream to fetch.
4200        :param seek_position: Requested seek offset in seconds.
4201        :param extra_input_args: Provider-supplied ffmpeg input arguments.
4202        :return: The ffmpeg input, the remaining seek offset and the ffmpeg input arguments.
4203        """
4204        stream_type = streamdetails.stream_type
4205        if stream_type == StreamType.CUSTOM:
4206            if streamdetails.media_type == MediaType.AUDIO_SOURCE:
4207                audio_source = self._open_audio_source_generator(
4208                    streamdetails,
4209                    seek_position=seek_position if streamdetails.can_seek else 0,
4210                )
4211            else:
4212                # MusicProvider and PluginProvider both expose get_audio_stream with the same
4213                # shape. Pin the exact instance: a domain fallback would stream from a sibling
4214                # account while the source-stream slot is charged to the issuing instance.
4215                provider = self.mass.get_provider(streamdetails.provider, return_unavailable=True)
4216                if provider is None or not provider.available:
4217                    raise ProviderUnavailableError(
4218                        f"Provider {streamdetails.provider} for stream is no longer available"
4219                    )
4220                provider = cast("MusicProvider | PluginProvider", provider)
4221                audio_source = provider.get_audio_stream(
4222                    streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
4223                )
4224            return audio_source, 0 if streamdetails.can_seek else seek_position, extra_input_args
4225        if stream_type == StreamType.ICY:
4226            assert streamdetails.path is not None
4227            assert isinstance(streamdetails.path, (str, list))
4228            audio_source = self.get_reconnecting_icy_radio_stream(streamdetails.path, streamdetails)
4229            return audio_source, 0, extra_input_args
4230        if stream_type == StreamType.SHOUTCAST:
4231            assert isinstance(streamdetails.path, str)
4232            return self.get_shoutcast_stream(streamdetails.path, streamdetails), 0, extra_input_args
4233        if stream_type == StreamType.IN_BAND:
4234            assert isinstance(streamdetails.path, str)  # for type checking
4235
4236            # For IN_BAND (OGG/Opus) radio streams, use chained OGG handler.
4237            # This handles the chained OGG format by stitching logical bitstreams together
4238            # so FFmpeg sees a single continuous stream. Metadata is extracted in-band.
4239            audio_source = get_chained_ogg_stream(
4240                self.mass,
4241                streamdetails.path,
4242                metadata_callback=partial(self._handle_inband_metadata, streamdetails),
4243            )
4244            # seeking not possible on radio streams
4245            return audio_source, 0, extra_input_args
4246        if stream_type == StreamType.HLS:
4247            assert isinstance(streamdetails.path, str)  # for type checking
4248            substream = await self.get_hls_substream(streamdetails.path)
4249            if streamdetails.media_type == MediaType.RADIO:
4250                # HLS streams (especially the BBC) struggle when they're played directly
4251                # with ffmpeg, where they just stop after some minutes,
4252                # so we tell ffmpeg to loop around in this case.
4253                extra_input_args += ["-stream_loop", "-1", "-re"]
4254            return substream.path, seek_position, extra_input_args
4255
4256        # all other stream types (HTTP, FILE, etc)
4257        if stream_type == StreamType.ENCRYPTED_HTTP:
4258            assert streamdetails.decryption_key is not None  # for type checking
4259            extra_input_args += ["-decryption_key", streamdetails.decryption_key]
4260        if isinstance(streamdetails.path, list):
4261            # multi part stream, which handles the seek itself
4262            return self.get_multi_file_stream(streamdetails, seek_position), 0, extra_input_args
4263        # regular single file/url stream
4264        assert isinstance(streamdetails.path, str)  # for type checking
4265        return streamdetails.path, seek_position, extra_input_args
4266
4267    async def _iter_audio_source_pcm(
4268        self,
4269        streamdetails: StreamDetails,
4270        pcm_format: AudioFormat,
4271    ) -> AsyncGenerator[bytes]:
4272        """Yield PCM for an AudioSource, bypassing ffmpeg when formats match."""
4273        # deliberately the advertised format: an AudioSource provider states the
4274        # PCM it delivers here, and providers that advertise a codec instead rely
4275        # on the ffmpeg path below to notice their source ending
4276        if streamdetails.audio_format == pcm_format:
4277            source_gen = self._open_audio_source_generator(streamdetails)
4278            async for chunk in realtime_pcm_pacer(source_gen, pcm_format):
4279                yield chunk
4280            return
4281        # format mismatch → fall back to ffmpeg for resampling (still small chunks)
4282        async for chunk in self.get_media_stream(
4283            streamdetails=streamdetails,
4284            pcm_format=pcm_format,
4285            filter_params=None,
4286            chunk_seconds=AUDIO_SOURCE_CHUNK_SECONDS,
4287        ):
4288            yield chunk
4289
4290    def _open_audio_source_generator(
4291        self,
4292        streamdetails: StreamDetails,
4293        seek_position: int = 0,
4294    ) -> AsyncGenerator[bytes]:
4295        """
4296        Open the raw PCM generator for an AudioSource.
4297
4298        :param streamdetails: Details of the AudioSource to stream.
4299        :param seek_position: Requested seek offset in seconds.
4300        """
4301        if streamdetails.stream_type == StreamType.CUSTOM:
4302            # pin the exact instance, see _resolve_media_stream_source
4303            provider = self.mass.get_provider(streamdetails.provider, return_unavailable=True)
4304            if provider is None or not provider.available:
4305                raise ProviderUnavailableError(
4306                    f"Provider {streamdetails.provider} for stream is no longer available"
4307                )
4308            provider = cast("MusicProvider | PluginProvider", provider)
4309            audio_source = provider.get_audio_stream(
4310                streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
4311            )
4312            return audio_source_silence_keepalive(
4313                audio_source, arriving_audio_format(streamdetails)
4314            )
4315        if streamdetails.stream_type == StreamType.NAMED_PIPE:
4316            assert isinstance(streamdetails.path, str)  # for type checking
4317            return read_named_pipe(streamdetails.path)
4318        raise AudioError(f"Unsupported stream_type {streamdetails.stream_type} for AudioSource")
4319
4320    def _handle_inband_metadata(
4321        self, streamdetails: StreamDetails, metadata: dict[str, str]
4322    ) -> None:
4323        """Handle metadata extracted from a chained Ogg stream."""
4324        title = metadata.get("title", "")
4325        artist = metadata.get("artist", "")
4326        album = metadata.get("album", "")
4327        if not artist and " - " in title:
4328            artist, title = title.split(" - ", 1)
4329        if not (title or artist):
4330            return
4331
4332        stream_title = f"{artist} - {title}" if artist and title else title or artist
4333        cleaned_title = clean_stream_title(stream_title)
4334        if not cleaned_title:
4335            return
4336        if self._record_inband_stream_title(streamdetails, cleaned_title):
4337            return
4338        if cleaned_title != streamdetails.stream_title:
4339            self.logger.log(VERBOSE_LOG_LEVEL, "In-band metadata: %s", cleaned_title)
4340            streamdetails.stream_title = cleaned_title
4341            self._update_radio_stream_metadata(
4342                streamdetails,
4343                artist=artist or None,
4344                title=title or cleaned_title,
4345                album=album or None,
4346            )
4347
4348    def _record_inband_stream_title(self, streamdetails: StreamDetails, cleaned_title: str) -> bool:
4349        """
4350        Record an in-band stream title for provider-owned metadata, if applicable.
4351
4352        When a provider opts into owning stream_metadata (and stream_title is only
4353        a derived view of it), writing either from the stream reader would fight the
4354        provider. The cleaned in-band title is recorded on StreamDetails.data instead,
4355        as the identity signal for the provider callback.
4356
4357        :param streamdetails: StreamDetails carrying the stream.
4358        :param cleaned_title: Cleaned in-band stream title.
4359        :returns: True when recorded (the caller must not write stream metadata);
4360            False when no provider callback exists and normal handling applies.
4361        """
4362        if (
4363            streamdetails.stream_metadata_update_callback is None
4364            or streamdetails.data is None
4365            or not streamdetails.data.get(STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY)
4366        ):
4367            return False
4368        if streamdetails.data.get(STREAMDETAILS_INBAND_TITLE_KEY) != cleaned_title:
4369            # occupancy approximates how far this detection leads audible playback
4370            buffer = streamdetails.buffer
4371            self.logger.debug(
4372                "In-band stream title: %s (buffer occupancy: %ss)",
4373                cleaned_title,
4374                buffer.size_seconds if buffer is not None else "unknown",
4375            )
4376            streamdetails.data[STREAMDETAILS_INBAND_TITLE_KEY] = cleaned_title
4377        return True
4378
4379    def _parse_icy_metadata(self, meta_data: bytes, streamdetails: StreamDetails) -> None:
4380        """
4381        Parse ICY metadata and update streamdetails.
4382
4383        Sets the cleaned stream title and, when the title parses as "Artist - Track",
4384        triggers a radio-artwork metadata update.
4385
4386        :param meta_data: Raw metadata bytes from an ICY stream chunk.
4387        :param streamdetails: StreamDetails to update with parsed title and metadata.
4388        """
4389        if not meta_data:
4390            return
4391
4392        meta_data = meta_data.rstrip(b"\0")
4393        # Match StreamTitle, handling apostrophes in titles
4394        stream_title_re = re.search(rb"StreamTitle='(.*?)';", meta_data)
4395
4396        if not stream_title_re:
4397            self.logger.log(
4398                VERBOSE_LOG_LEVEL,
4399                "ICY metadata does not contain StreamTitle field. Raw: %s",
4400                meta_data.decode("utf-8", errors="replace")[:200],
4401            )
4402            return
4403
4404        try:
4405            # in 99% of the cases the stream title is utf-8 encoded
4406            stream_title = stream_title_re.group(1).decode("utf-8")
4407        except UnicodeDecodeError:
4408            # fallback to iso-8859-1
4409            stream_title = stream_title_re.group(1).decode("iso-8859-1", errors="replace")
4410
4411        cleaned_stream_title = clean_stream_title(stream_title)
4412
4413        if not cleaned_stream_title:
4414            return
4415
4416        if self._record_inband_stream_title(streamdetails, cleaned_stream_title):
4417            return
4418
4419        if cleaned_stream_title == streamdetails.stream_title:
4420            return
4421
4422        self.logger.log(VERBOSE_LOG_LEVEL, "ICY Radio streamtitle original: %s", stream_title)
4423        self.logger.log(
4424            VERBOSE_LOG_LEVEL, "ICY Radio streamtitle cleaned: %s", cleaned_stream_title
4425        )
4426        streamdetails.stream_title = cleaned_stream_title
4427
4428        # Prefer station-provided cover art from the ICY 'StreamUrl' field (when it is
4429        # an image) over the MusicBrainz artwork lookup in _update_radio_stream_metadata.
4430        image_url = self._parse_icy_image_url(meta_data)
4431
4432        # Parse the original title for structured fields first so stations that announce
4433        # an album can refine the artwork lookup; fall back to the "Artist - Track" split.
4434        album: str | None = None
4435        if parsed := parse_quoted_stream_title(stream_title):
4436            track_name, artist_name_raw, album = parsed
4437        elif " - " in cleaned_stream_title:
4438            artist_name_raw, track_name = (
4439                part.strip() for part in cleaned_stream_title.split(" - ", 1)
4440            )
4441        else:
4442            return
4443
4444        if artist_name_raw and track_name:
4445            self.logger.debug(
4446                "ICY metadata: artist='%s', track='%s', album='%s'",
4447                artist_name_raw,
4448                track_name,
4449                album,
4450            )
4451            self._update_radio_stream_metadata(
4452                streamdetails,
4453                artist=artist_name_raw,
4454                title=track_name,
4455                album=album,
4456                image_url=image_url,
4457            )
4458
4459    def _parse_icy_image_url(self, meta_data: bytes) -> str | None:
4460        """
4461        Return a PNG or JPEG cover-art URL from the ICY 'StreamUrl' field, if present.
4462
4463        :param meta_data: Raw metadata bytes from an ICY stream chunk.
4464        """
4465        # The trailing semicolon is optional to match sources that omit it.
4466        stream_url_re = re.search(rb"StreamUrl='([^']*)'", meta_data)
4467        if not stream_url_re:
4468            return None
4469        try:
4470            image_url = stream_url_re.group(1).decode("utf-8").strip()
4471        except UnicodeDecodeError:
4472            return None
4473        if not image_url:
4474            return None
4475        # StreamUrl is not a standardized artwork field (reference clients such as VLC
4476        # ignore it and it conventionally holds a station website link), so only accept
4477        # values that point at a PNG or JPEG image.
4478        parsed = urlparse(image_url)
4479        if parsed.scheme not in ("http", "https"):
4480            return None
4481        if not parsed.path.lower().endswith((".png", ".jpg", ".jpeg")):
4482            return None
4483        self.logger.debug("ICY metadata: StreamUrl image='%s'", image_url)
4484        return image_url
4485
4486    async def _validate_shoutcast_stream(self, url: str) -> bool:
4487        """
4488        Return True if the URL responds with a legacy Shoutcast "ICY 200 OK" line.
4489
4490        :param url: The URL to validate.
4491        """
4492        try:
4493            parsed = urlparse(url)
4494            host = parsed.hostname
4495            port = parsed.port or 80
4496            path = parsed.path or "/"
4497            if parsed.query:
4498                path = f"{path}?{parsed.query}"
4499
4500            # Open raw socket connection with timeout
4501            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=10)
4502            try:
4503                # Send minimal HTTP request with ICY metadata header
4504                request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nIcy-MetaData: 1\r\n\r\n"
4505                writer.write(request.encode())
4506                await writer.drain()
4507
4508                # Read just the response line
4509                response_line = await asyncio.wait_for(reader.readline(), timeout=5)
4510            finally:
4511                writer.close()
4512                await writer.wait_closed()
4513
4514            # Check if response starts with "ICY"
4515            decoded_line = response_line.decode("latin-1", errors="ignore").strip()
4516            return decoded_line.startswith("ICY")
4517
4518        except TimeoutError:
4519            self.logger.debug("Timeout during Shoutcast validation for %s", url)
4520            return False
4521        except OSError, ConnectionError:
4522            self.logger.debug("Connection failed during Shoutcast validation for %s", url)
4523            return False
4524        except UnicodeDecodeError:
4525            self.logger.debug("Invalid response encoding during Shoutcast validation for %s", url)
4526            return False
4527
4528    def _resolve_player_dsp_config(self, player: Player) -> DSPConfig:
4529        """
4530        Resolve the effective DSP config for a player.
4531
4532        Single source of truth shared by every code path that needs to know
4533        whether DSP will run for this player. Protocol wrappers defer to their
4534        parent player; single-leg ``player_group`` instances that don't expose
4535        ``MULTI_DEVICE_DSP`` defer to their first member; players whose grouping
4536        context prevents DSP get a disabled config back regardless.
4537
4538        :param player: The player to resolve DSP config for.
4539        """
4540        dsp_player_id = self._resolve_player_dsp_config_id(player)
4541        dsp = self.mass.config.get_player_dsp_config(dsp_player_id)
4542        if is_grouping_preventing_dsp(player):
4543            dsp.enabled = False
4544        elif player.provider.domain == "player_group" and (
4545            PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
4546        ):
4547            if not player.state.group_members:
4548                dsp.enabled = False
4549        return dsp
4550
4551    def _resolve_player_dsp_config_id(self, player: Player) -> str:
4552        """
4553        Return the player identifier that supplies the effective DSP config.
4554
4555        :param player: Player whose DSP config source should be resolved.
4556        """
4557        dsp_player_id = player.protocol_parent_id or player.player_id
4558        if (
4559            not is_grouping_preventing_dsp(player)
4560            and player.provider.domain == "player_group"
4561            and PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
4562            and player.state.group_members
4563        ):
4564            child_player = self.mass.players.get_player(player.state.group_members[0])
4565            assert child_player is not None
4566            dsp_player_id = child_player.player_id
4567        return dsp_player_id
4568
4569    def _get_output_channels(self, player: Player | None, player_id: str) -> str:
4570        """
4571        Return the configured output channels for the rendering player.
4572
4573        The value may be stored on the rendering player(protocol) itself (the
4574        protocol section of the config UI) or on its visible parent player (the
4575        native section); the rendering player's own stored value wins.
4576        """
4577        parent_id = player.protocol_parent_id if player and player.protocol_parent_id else player_id
4578        parent_value = self.mass.config.get_raw_player_config_value(
4579            parent_id, CONF_OUTPUT_CHANNELS, "stereo"
4580        )
4581        return self.mass.config.get_raw_player_config_value(
4582            player.player_id if player else player_id, CONF_OUTPUT_CHANNELS, parent_value
4583        )
4584
4585    def _pick_pcm_bit_depth(
4586        self,
4587        players: Iterable[Player],
4588        streamdetails: StreamDetails | None,
4589        crossfade_enabled: bool,
4590        overlay_active: bool = False,
4591    ) -> tuple[ContentType, int]:
4592        """
4593        Return ``(content_type, bit_depth)`` for an internal PCM stream.
4594
4595        F32 is chosen when audio processing (crossfade, audio overlay, volume
4596        normalization, DSP) will run on the stream — those need the extra
4597        headroom to avoid clipping and precision loss. Otherwise the source's
4598        native bit depth is reused so we don't waste memory upcasting a 16-bit
4599        stream to 32-bit just to pass it through. When the source is unknown
4600        (no streamdetails) we fall back to F32 conservatively.
4601        """
4602        if streamdetails is None:
4603            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
4604        needs_headroom = (
4605            crossfade_enabled
4606            or overlay_active
4607            or streamdetails.volume_normalization_mode
4608            not in (VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.SOURCE)
4609            or any(self._resolve_player_dsp_config(player).enabled for player in players)
4610        )
4611        if needs_headroom:
4612            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
4613        # the depth the audio arrives in, not the one the source claims: a
4614        # provider that decoded on our behalf may advertise a narrower format
4615        # for display, and narrowing the stream to that would truncate it
4616        bit_depth = arriving_audio_format(streamdetails).bit_depth
4617        return ContentType.from_bit_depth(bit_depth), bit_depth
4618
4619    def _select_audio_source_pcm_format(
4620        self,
4621        player: Player,
4622        streamdetails: StreamDetails,
4623        supported_sample_rates: Iterable[int] | None = None,
4624    ) -> AudioFormat:
4625        """
4626        Return a passthrough PCM format for a realtime AudioSource item.
4627
4628        The format matches the source's native sample rate, bit depth and
4629        channel count whenever the player can accept them; if the player does
4630        not support the source's sample rate, it is snapped down to the
4631        closest supported rate. No F32 widening — realtime sources skip every
4632        processing stage that would otherwise need it. Surround sources are
4633        still folded down to stereo, which every output path requires anyway.
4634
4635        :param player: The player requesting the stream.
4636        :param streamdetails: Stream details for the AudioSource item.
4637        :param supported_sample_rates: Rates shared by every output player, if applicable.
4638        """
4639        resolved_sample_rates = (
4640            list(supported_sample_rates)
4641            if supported_sample_rates is not None
4642            else [sample_rate for sample_rate, _ in player.get_supported_sample_rates()]
4643        )
4644        # the format the audio arrives in, not the one the source claims: a provider
4645        # that decoded on our behalf may advertise a narrower format for display, and
4646        # narrowing the stream to that would truncate it
4647        source_format = arriving_audio_format(streamdetails)
4648        source_rate = source_format.sample_rate
4649        if source_rate in resolved_sample_rates:
4650            output_sample_rate = source_rate
4651        else:
4652            output_sample_rate = max(
4653                (rate for rate in resolved_sample_rates if rate <= source_rate),
4654                default=min(resolved_sample_rates),
4655            )
4656        return AudioFormat(
4657            content_type=ContentType.from_bit_depth(source_format.bit_depth),
4658            sample_rate=output_sample_rate,
4659            bit_depth=source_format.bit_depth,
4660            # a realtime source may announce more channels than anything downstream can
4661            # carry (a VBAN stream can be configured up to 8), and player handoff formats
4662            # copy this count straight through, so fold it here
4663            channels=min(source_format.channels, 2),
4664        )
4665
4666    def _flow_restart_context(
4667        self, queue_id: str, protocol_player: Player | None
4668    ) -> tuple[str, list[int]]:
4669        """
4670        Resolve the flow mode config and supported sample rates for restart decisions.
4671
4672        Prefers the protocol player actually consuming the flow stream over the
4673        queue's (wrapper) player, whose config may lack the audio specific entries.
4674        """
4675        if protocol_player is None:
4676            protocol_player = self.mass.players.get_player(queue_id)
4677        if protocol_player is None:
4678            flow_mode_sample_rate_conf = self.mass.config.get_raw_player_config_value(
4679                queue_id, CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
4680            )
4681            return flow_mode_sample_rate_conf, []
4682        flow_mode_sample_rate_conf = cast(
4683            "str",
4684            protocol_player.config.get_value(
4685                CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
4686            ),
4687        )
4688        supported_sample_rates = sorted(
4689            {sr for sr, _ in protocol_player.get_supported_sample_rates()}
4690        )
4691        return flow_mode_sample_rate_conf, supported_sample_rates
4692
4693    def _flow_stream_needs_restart(
4694        self,
4695        queue_track: QueueItem,
4696        pcm_format: AudioFormat,
4697        supported_sample_rates: list[int],
4698        flow_mode_sample_rate_conf: str,
4699        is_first_track: bool,
4700    ) -> bool:
4701        """
4702        Return True if the upcoming queue track requires exiting the flow stream.
4703
4704        Covers every case where the flow loop should break and hand control back to
4705        the queue controller for restart:
4706
4707        - Live media (radio, audio sources): cannot be played inside a flow,
4708          the controller will fall back to a single-item stream.
4709        - Sample rate mismatch ('smart' / 'bit_perfect' modes only): the next
4710          track's sample rate (snapped up to the closest supported player rate,
4711          mirroring select_flow_pcm_format's anchoring logic) is incompatible with
4712          the current flow rate, so a new flow must be opened.
4713
4714        The first (anchor) track is always allowed to continue for the sample
4715        rate check; select_flow_pcm_format has already snapped the flow rate to it.
4716
4717        :param queue_track: The upcoming queue item.
4718        :param pcm_format: The current flow stream's PCM format.
4719        :param supported_sample_rates: Sorted list of the player's supported rates.
4720        :param flow_mode_sample_rate_conf: The flow mode sample rate config value.
4721        :param is_first_track: Whether this is the first track of the flow stream.
4722        """
4723        # live audio (radio, plugin or audio source) cannot be flowed; let the
4724        # queue controller fall back to single-item streaming for this item
4725        if queue_track.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
4726            self.logger.info(
4727                "Live media item %s (%s, %s) encountered in flow stream "
4728                "- breaking out to single item stream",
4729                queue_track.queue_item_id,
4730                queue_track.name,
4731                queue_track.media_type,
4732            )
4733            return True
4734
4735        if is_first_track or queue_track.streamdetails is None:
4736            return False
4737        raw_next_rate = queue_track.streamdetails.audio_format.sample_rate
4738        if not raw_next_rate or not supported_sample_rates:
4739            return False
4740        effective_next_rate = _snap_supported_rate_up(raw_next_rate, supported_sample_rates)
4741
4742        # branch order mirrors select_flow_pcm_format: fixed-rate modes resample
4743        # everything to the chosen rate (no restart); bit_perfect restarts on any
4744        # mismatch; anything else falls through to smart-anchor behavior so
4745        # unknown/legacy config values don't silently pin the flow forever.
4746        if flow_mode_sample_rate_conf in (
4747            FLOW_MODE_SAMPLE_RATE_48000,
4748            FLOW_MODE_SAMPLE_RATE_96000,
4749            FLOW_MODE_SAMPLE_RATE_HIGHEST,
4750        ):
4751            needs_restart = False
4752        elif flow_mode_sample_rate_conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT:
4753            needs_restart = effective_next_rate != pcm_format.sample_rate
4754        else:
4755            needs_restart = effective_next_rate > pcm_format.sample_rate
4756
4757        if needs_restart:
4758            self.logger.info(
4759                "Track %s (%s) sample rate %s (snapped to %s) incompatible with flow rate %s "
4760                "(mode: %s) - breaking out to restart flow stream",
4761                queue_track.queue_item_id,
4762                queue_track.name,
4763                raw_next_rate,
4764                effective_next_rate,
4765                pcm_format.sample_rate,
4766                flow_mode_sample_rate_conf,
4767            )
4768        return needs_restart
4769
4770    @asynccontextmanager
4771    async def _connect_radio_stream(self, url: str, **kwargs: Any) -> AsyncGenerator[Any]:
4772        """
4773        Connect to a radio stream URL with fallback for legacy SSL/TLS configurations.
4774
4775        Some radio servers use outdated TLS configurations that reject modern
4776        cipher suites. Since radio streams are public broadcast content,
4777        relaxing cipher requirements is acceptable.
4778
4779        :param url: The radio stream URL to connect to.
4780        :param kwargs: Additional keyword arguments passed to aiohttp get().
4781        """
4782        request_url = encoded_request_url(url)
4783        try:
4784            async with self.mass.http_session_no_ssl.get(request_url, **kwargs) as resp:
4785                yield resp
4786        except ClientConnectorSSLError:
4787            self.logger.info(
4788                "SSL handshake failed for %s, retrying with permissive cipher configuration", url
4789            )
4790            insecure_ssl_context = ssl_util.client_context_no_verify(
4791                ssl_util.SSLCipherList.INSECURE
4792            )
4793            async with self.mass.http_session_no_ssl.get(
4794                request_url, ssl=insecure_ssl_context, **kwargs
4795            ) as resp:
4796                yield resp
4797
4798    async def _update_hls_radio_metadata(
4799        self,
4800        streamdetails: StreamDetails,
4801        elapsed_time: int,
4802    ) -> None:
4803        """
4804        Update HLS radio stream metadata by fetching the playlist.
4805
4806        Fetches the HLS playlist and extracts metadata from EXTINF lines.
4807
4808        :param streamdetails: StreamDetails object to update with metadata
4809        :param elapsed_time: Current playback position in seconds (unused for live radio)
4810        """
4811        mass = self.mass
4812        try:
4813            # Get the actual media playlist URL from cache or resolve it
4814            # We cache the media_playlist_url in streamdetails.data to avoid re-resolving
4815            if streamdetails.data is None:
4816                streamdetails.data = {}
4817            media_playlist_url = streamdetails.data.get("hls_media_playlist_url")
4818            if not media_playlist_url:
4819                try:
4820                    assert isinstance(streamdetails.path, str)  # for type checking
4821                    substream = await self.get_hls_substream(streamdetails.path)
4822                    media_playlist_url = substream.path
4823                    streamdetails.data["hls_media_playlist_url"] = media_playlist_url
4824                except Exception as err:
4825                    self.logger.warning(
4826                        "Failed to resolve HLS substream for metadata monitoring: %s", err
4827                    )
4828                    return
4829
4830            # Fetch the media playlist
4831            timeout = ClientTimeout(total=0, connect=10, sock_read=30)
4832            try:
4833                async with mass.http_session_no_ssl.get(
4834                    encoded_request_url(media_playlist_url), timeout=timeout
4835                ) as resp:
4836                    resp.raise_for_status()
4837                    playlist_content = await resp.text()
4838            except ClientResponseError as err:
4839                # Session token likely expired (410/403) — drop cache so next poll re-resolves
4840                if err.status in (403, 410):
4841                    streamdetails.data.pop("hls_media_playlist_url", None)
4842                raise
4843
4844            # Parse the playlist and look for EXTINF metadata
4845            # The most recent segment usually has the current metadata
4846            lines = playlist_content.strip().split("\n")
4847            for line in reversed(lines):
4848                if line.startswith("#EXTINF:"):
4849                    # Extract metadata from EXTINF line
4850                    metadata = parse_extinf_metadata(line)
4851
4852                    # Build stream title from title and artist
4853                    title = metadata.get("title", "")
4854                    artist = metadata.get("artist", "")
4855                    image_url = (
4856                        metadata.get("image") or metadata.get("artwork") or metadata.get("cover")
4857                    )
4858                    if not artist and " - " in title:
4859                        artist, title = title.split(" - ", 1)
4860                    if title or artist:
4861                        # Format as "Artist - Title"
4862                        if artist and title:
4863                            stream_title = f"{artist} - {title}"
4864                        elif title:
4865                            stream_title = title
4866                        else:
4867                            stream_title = artist
4868
4869                        # Clean the stream title
4870                        cleaned_title = clean_stream_title(stream_title)
4871
4872                        # Only update if changed
4873                        if cleaned_title != streamdetails.stream_title and cleaned_title:
4874                            self.logger.log(
4875                                VERBOSE_LOG_LEVEL, "HLS Radio metadata updated: %s", cleaned_title
4876                            )
4877                            streamdetails.stream_title = cleaned_title
4878                            self._update_radio_stream_metadata(
4879                                streamdetails,
4880                                artist=artist or None,
4881                                title=title or cleaned_title,
4882                                image_url=image_url,
4883                            )
4884
4885                    # Only check the most recent EXTINF
4886                    break
4887
4888        except Exception as err:
4889            self.logger.debug("Error fetching HLS metadata: %s", err)
4890
4891    @staticmethod
4892    def _normalize_reconnecting_urls(url: str | list[MultiPartPath]) -> list[str]:
4893        """Normalize a single URL or a sequence into a non-empty list."""
4894        if isinstance(url, str):
4895            return [url]
4896        if not url:
4897            msg = "Radio stream requires at least one URL"
4898            raise InvalidDataError(msg)
4899        return [part.path for part in url]
4900
4901    async def _resolve_overlay_input(self, queue: PlayerQueue) -> str | None:
4902        """
4903        Resolve the queue's overlay source to a file path or URL for ffmpeg.
4904
4905        Returns None (with a warning logged) when the source can not be resolved,
4906        so the caller can degrade to music-only playback.
4907        """
4908        if not (mapping := queue.overlay_source):
4909            return None
4910        try:
4911            provider = self.mass.get_provider(mapping.provider)
4912            if provider is None:
4913                raise MediaNotFoundError(f"Provider {mapping.provider} is not available")
4914            stream_prov = cast("MusicProvider | PluginProvider", provider)
4915            streamdetails = await stream_prov.get_stream_details(
4916                mapping.item_id, MediaType.SOUND_EFFECT
4917            )
4918        except Exception as err:
4919            self.logger.warning(
4920                "Audio overlay source %s is unavailable (%s) - continuing without overlay",
4921                mapping.uri,
4922                str(err) or err.__class__.__name__,
4923            )
4924            return None
4925        if streamdetails.stream_type not in (StreamType.LOCAL_FILE, StreamType.HTTP) or not (
4926            isinstance(streamdetails.path, str)
4927        ):
4928            self.logger.warning(
4929                "Audio overlay source %s uses unsupported stream type %s "
4930                "- continuing without overlay",
4931                mapping.uri,
4932                streamdetails.stream_type,
4933            )
4934            return None
4935        if streamdetails.stream_type == StreamType.LOCAL_FILE and not await aiofiles.os.path.isfile(
4936            streamdetails.path
4937        ):
4938            # guard against stale sources: feeding a missing file to the mixer would
4939            # kill the whole (music) stream instead of just the overlay
4940            self.logger.warning(
4941                "Audio overlay source %s does not exist - continuing without overlay",
4942                streamdetails.path,
4943            )
4944            return None
4945        return streamdetails.path
4946