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