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