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                # re-read the effective crossfade settings so a change applies at the
2485                # next track transition instead of the next stream session; a realtime
2486                # source still gets its fade decided from what its boundary can
2487                # actually deliver (see _select_buffered_crossfade)
2488                if queue_track.media_type != MediaType.TRACK:
2489                    item_crossfade_mode = CrossfadeMode.DISABLED
2490                else:
2491                    item_crossfade_mode = self.mass.streams.get_crossfade_mode(queue)
2492                    standard_crossfade_duration = self.mass.config.get_raw_core_config_value(
2493                        CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8
2494                    )
2495                if item_crossfade_mode != crossfade_mode:
2496                    self.logger.debug(
2497                        "Crossfade mode for queue %s changed mid-session: %s -> %s",
2498                        queue.display_name,
2499                        crossfade_mode,
2500                        item_crossfade_mode,
2501                    )
2502                    crossfade_mode = item_crossfade_mode
2503                self.logger.debug(
2504                    "Start Streaming queue track: %s (%s) for queue %s",
2505                    queue_track.streamdetails.uri,
2506                    queue_track.name,
2507                    queue.display_name,
2508                )
2509                # last chance to bail before mutating the stream log: a newer producer
2510                # may have taken over while we were awaiting load_next_queue_item
2511                if _superseded():
2512                    self.logger.debug(
2513                        "Flow stream for queue %s superseded - exiting before playlog append",
2514                        queue.display_name,
2515                    )
2516                    return
2517                track_playback_speed = cast(
2518                    "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2519                )
2520                # calculate crossfade buffer size: the ceiling for this item's holdback
2521                crossfade_buffer_duration = (
2522                    SMART_CROSSFADE_DURATION
2523                    if item_crossfade_mode == CrossfadeMode.SMART_CROSSFADE
2524                    else standard_crossfade_duration
2525                )
2526                crossfade_buffer_duration = min(
2527                    crossfade_buffer_duration,
2528                    int(queue_track.streamdetails.duration / 2)
2529                    if queue_track.streamdetails.duration
2530                    else crossfade_buffer_duration,
2531                )
2532                # skip crossfade if buffer would be too small to be meaningful
2533                if crossfade_buffer_duration < MIN_CROSSFADE_DURATION:
2534                    crossfade_buffer_duration = 0
2535                # Ensure crossfade buffer size is aligned to frame boundaries
2536                # Frame size = bytes_per_sample * channels
2537                bytes_per_sample = pcm_format.bit_depth // 8
2538                frame_size = bytes_per_sample * pcm_format.channels
2539                crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
2540                # Round down to nearest frame boundary
2541                crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
2542
2543                # raw_seek_position feeds the PCM buffer; streamdetails.seek_position
2544                # (overwritten below) only drives reported elapsed time.
2545                raw_seek_position = queue_track.streamdetails.seek_position
2546                # Build eagerly so seek_position is set before PlayLogEntry is appended —
2547                # consumer-paced mix() would otherwise let the queue briefly report 0.
2548                crossfade_smart_fade: SmartFade | None = None
2549                collect_resident = 0.0
2550                incoming_crossfade_size = crossfade_buffer_size
2551                incoming_audio_buffer: AudioBuffer | None = None
2552                build_seconds = 0.0
2553                transition_mode = CrossfadeMode.DISABLED
2554                applied_mode = CrossfadeMode.DISABLED
2555                outgoing_queue_track = last_queue_track
2556                if last_fadeout_part and last_streamdetails:
2557                    incoming_duration = 0.0
2558                    if crossfade_buffer_size > 0 and item_crossfade_mode != CrossfadeMode.DISABLED:
2559                        # a realtime incoming track has audio to read only once its
2560                        # session produces; give it a bounded chance to show up
2561                        await self._await_realtime_fade_source(queue_track.streamdetails)
2562                        transition_mode, incoming_duration = self._select_buffered_crossfade(
2563                            queue_track.streamdetails,
2564                            item_crossfade_mode,
2565                            standard_crossfade_duration,
2566                            fade_out_seconds=len(last_fadeout_part) / pcm_sample_size,
2567                            playback_speed=track_playback_speed,
2568                        )
2569                    if transition_mode == CrossfadeMode.DISABLED:
2570                        # nothing to fade into: flush the held-back tail of the previous track
2571                        for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2572                            yield pcm_slice
2573                            await asyncio.sleep(0)
2574                        last_fadeout_part = b""
2575                        last_streamdetails = None
2576                        last_play_log_entry = None
2577                        last_queue_track = None
2578                    else:
2579                        assert queue_track.streamdetails.buffer is not None
2580                        incoming_audio_buffer = cast(
2581                            "AudioBuffer", queue_track.streamdetails.buffer
2582                        )
2583                        incoming_crossfade_size = int(
2584                            pcm_format.pcm_sample_size * incoming_duration
2585                        )
2586                        incoming_crossfade_size = (
2587                            incoming_crossfade_size // frame_size
2588                        ) * frame_size
2589                        collect_resident = incoming_audio_buffer.duration_available
2590                        applied_mode = transition_mode
2591                        build_started = asyncio.get_event_loop().time()
2592                        crossfade_smart_fade = await self.smart_fades_mixer.build(
2593                            fade_in_streamdetails=queue_track.streamdetails,
2594                            fade_out_streamdetails=last_streamdetails,
2595                            pcm_format=pcm_format,
2596                            standard_crossfade_duration=standard_crossfade_duration,
2597                            mode=transition_mode,
2598                            fade_out_data=last_fadeout_part,
2599                            fade_in_bytes_len=incoming_crossfade_size,
2600                        )
2601                        build_seconds = asyncio.get_event_loop().time() - build_started
2602                        timing_info = crossfade_smart_fade.timing_info
2603                        if isinstance(crossfade_smart_fade, StandardCrossFade):
2604                            # the mixer degrades to a standard fade when the smart one
2605                            # cannot be planned, so that is what will really be applied
2606                            applied_mode = CrossfadeMode.STANDARD_CROSSFADE
2607                            # A standard fade blends its overlap and passes everything after it
2608                            # through untouched, so only the overlap has to be in hand before
2609                            # the transition can start. Holding back the rest buys nothing and
2610                            # keeps the player waiting - a smart fade does need its full window,
2611                            # which is only chosen when the analysis it needs is already there.
2612                            blended_seconds = (
2613                                timing_info.fadein_trimmed_duration + timing_info.crossfade_duration
2614                            )
2615                            blended_size = int(pcm_format.pcm_sample_size * blended_seconds)
2616                            incoming_crossfade_size = min(
2617                                incoming_crossfade_size,
2618                                (blended_size // frame_size) * frame_size,
2619                            )
2620                        queue_track.streamdetails.seek_position = (
2621                            raw_seek_position
2622                            + (timing_info.fadein_trimmed_duration + timing_info.crossfade_duration)
2623                            * track_playback_speed
2624                        )
2625                # no fade is credited to this track until one is really rendered below
2626                self._report_crossfade_mode(
2627                    queue.queue_id,
2628                    queue_track,
2629                    pcm_format,
2630                    CrossfadeMode.DISABLED,
2631                    flow_session_id,
2632                    overlay_enabled=overlay_active(queue),
2633                )
2634                # append to play log so the queue controller can work out which track is playing
2635                play_log_entry = PlayLogEntry(queue_track.queue_item_id)
2636                flow_log.append(play_log_entry)
2637
2638                bytes_written = 0
2639                crossfade_buffer = bytearray()
2640                first_chunk_received = False
2641                holding_back = item_crossfade_mode != CrossfadeMode.DISABLED
2642
2643                item_stream = await incoming_prefetcher.take(queue_track, int(raw_seek_position))
2644                prefetched_size = incoming_prefetcher.collected_at_handover if item_stream else 0
2645                if item_stream is None:
2646                    item_stream = self.get_queue_item_stream(
2647                        queue_track,
2648                        pcm_format=pcm_format,
2649                        seek_position=int(raw_seek_position),
2650                        playback_speed=cast(
2651                            "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2652                        ),
2653                        raise_on_error=False,
2654                        session_id=flow_session_id,
2655                        prepared_buffer=incoming_audio_buffer,
2656                    )
2657
2658                # closing here releases the decoders on an early exit,
2659                # instead of leaving them to the garbage collector
2660                async with aclosing(item_stream):
2661                    async for chunk in item_stream:
2662                        # if a newer producer has taken over this queue, stop sending
2663                        # audio and exit cleanly before the outer-loop end-of-track
2664                        # bookkeeping mutates seconds_streamed / duration on the log
2665                        if _superseded():
2666                            self.logger.debug(
2667                                "Flow stream for queue %s superseded - stopping chunk yield",
2668                                queue.display_name,
2669                            )
2670                            return
2671                        total_chunks_received += 1
2672                        if not first_chunk_received:
2673                            first_chunk_received = True
2674                            # inform the queue that the track is now loaded in the buffer
2675                            # so the next track can be preloaded
2676                            self.mass.player_queues.track_loaded_in_buffer(
2677                                queue.queue_id, queue_track.queue_item_id
2678                            )
2679
2680                        if item_crossfade_mode == CrossfadeMode.DISABLED:
2681                            # no cross/smart fade: yield chunks directly without intermediate buffer
2682                            yield chunk
2683                            bytes_written += len(chunk)
2684                            del chunk
2685                            continue
2686
2687                        if not last_fadeout_part:
2688                            # the tail is being held back, so the audio the next transition
2689                            # blends in can be gathered alongside it instead of after it
2690                            incoming_prefetcher.ensure_started(
2691                                queue,
2692                                queue_track,
2693                                item_crossfade_mode,
2694                                standard_crossfade_duration,
2695                            )
2696
2697                        # accumulate chunks in the crossfade buffer: the outgoing tail
2698                        # window, or (at a boundary) whatever of the incoming overlap
2699                        # arrived before the mix starts
2700                        crossfade_buffer.extend(chunk)
2701                        del chunk
2702                        hold_target = (
2703                            tail_hold_target(queue_track, crossfade_buffer_size, frame_size)
2704                            if holding_back
2705                            else 0
2706                        )
2707                        if not last_fadeout_part and len(crossfade_buffer) <= hold_target:
2708                            await asyncio.sleep(0)
2709                            continue
2710                        # handle crossfade of previous track and new track
2711                        if (
2712                            last_fadeout_part
2713                            and last_streamdetails
2714                            and crossfade_smart_fade is not None
2715                            and last_play_log_entry is not None
2716                        ):
2717                            self.logger.debug(
2718                                "Starting the transition into %s with %.1fs of its overlap"
2719                                " in hand (%.1fs prefetched, %.1fs build,"
2720                                " %.1fs was resident at the boundary)",
2721                                queue_track.name,
2722                                len(crossfade_buffer) / pcm_sample_size,
2723                                prefetched_size / pcm_sample_size,
2724                                build_seconds,
2725                                collect_resident,
2726                            )
2727                            # The mixer consumes the incoming overlap as it arrives and
2728                            # emits the blend at that same pace, so the transition
2729                            # streams instead of first collecting the whole overlap.
2730                            overlap_overshoot = bytearray()
2731                            mix_start_collected = len(crossfade_buffer)
2732                            overlap_pulled = 0
2733
2734                            def _note_overlap_bytes(count: int) -> None:
2735                                # the mixer reads the incoming stream itself for the
2736                                # length of the overlap, so the loop below cannot see
2737                                # those bytes; the overshoot bookkeeping needs them
2738                                nonlocal overlap_pulled
2739                                overlap_pulled += count
2740
2741                            overlap_stream = _incoming_overlap_stream(
2742                                bytes(crossfade_buffer),
2743                                item_stream,
2744                                incoming_crossfade_size,
2745                                overlap_overshoot,
2746                                _note_overlap_bytes,
2747                            )
2748                            crossfade_buffer = bytearray()
2749                            # The mix output is split live as it flows: the first
2750                            # fadeout_share bytes are the outgoing track's (its held
2751                            # tail, processed), the rest belong to this one. Credited
2752                            # per chunk, because a single correction afterwards would
2753                            # leave the queue's position mapping on the wrong track
2754                            # for the whole (source-paced) duration of the blend. The
2755                            # pre-counted tail makes way for that live credit.
2756                            fadeout_share_seconds = (
2757                                timing_info.pre_crossfade_duration + timing_info.crossfade_duration
2758                            )
2759                            fadeout_share = int(fadeout_share_seconds * pcm_sample_size)
2760                            fadeout_share = (fadeout_share // frame_size) * frame_size
2761                            assert last_play_log_entry.seconds_streamed is not None
2762                            last_play_log_entry.seconds_streamed -= (
2763                                len(last_fadeout_part) / pcm_sample_size
2764                            )
2765                            mix_stream = self.smart_fades_mixer.mix(
2766                                crossfade_smart_fade,
2767                                fade_in_part=overlap_stream,
2768                                fade_out_part=last_fadeout_part,
2769                                pcm_format=pcm_format,
2770                            )
2771                            try:
2772                                crossfade_bytes_written = 0
2773                                # closed before item_stream on an aborted flow: its
2774                                # teardown stops the feeder that still holds a read
2775                                # on item_stream, which must not be closed mid-read
2776                                async with aclosing(mix_stream):
2777                                    async for mix_chunk in mix_stream:
2778                                        yield mix_chunk
2779                                        outgoing_part = min(
2780                                            len(mix_chunk),
2781                                            max(0, fadeout_share - crossfade_bytes_written),
2782                                        )
2783                                        last_play_log_entry.seconds_streamed += (
2784                                            outgoing_part / pcm_sample_size
2785                                        )
2786                                        bytes_written += len(mix_chunk) - outgoing_part
2787                                        crossfade_bytes_written += len(mix_chunk)
2788                                remaining_bytes = bytes(overlap_overshoot)
2789                            except Exception as mix_err:
2790                                if crossfade_bytes_written:
2791                                    # partial mix already played — concat'd tail would duplicate audio
2792                                    raise
2793                                self.logger.warning(
2794                                    "Crossfade mixer failed for %s, falling back to simple concat: %s",
2795                                    queue_track.name,
2796                                    mix_err,
2797                                )
2798                                # the tail was un-counted for the live credit above;
2799                                # it now plays as ordinary outgoing audio
2800                                last_play_log_entry.seconds_streamed += (
2801                                    len(last_fadeout_part) / pcm_sample_size
2802                                )
2803                                for pcm_slice in iter_pcm_slices(
2804                                    last_fadeout_part, pcm_format, 1000
2805                                ):
2806                                    yield pcm_slice
2807                                    await asyncio.sleep(0)
2808                                crossfade_bytes_written = 0
2809                                remaining_bytes = b""
2810                                # mix failed — undo the eager seek_position
2811                                queue_track.streamdetails.seek_position = raw_seek_position
2812                                # The mixer teardown cancels its feeder, which was
2813                                # likely parked reading item_stream - that ends the
2814                                # stream itself. Play the track from a fresh stream
2815                                # (its buffer still holds what the mixer consumed)
2816                                # rather than silently losing its body.
2817                                await item_stream.aclose()
2818                                fallback_stream = self.get_queue_item_stream(
2819                                    queue_track,
2820                                    pcm_format=pcm_format,
2821                                    seek_position=int(raw_seek_position),
2822                                    playback_speed=track_playback_speed,
2823                                    raise_on_error=False,
2824                                    session_id=flow_session_id,
2825                                )
2826                                async with aclosing(fallback_stream):
2827                                    async for fallback_chunk in fallback_stream:
2828                                        if _superseded():
2829                                            return
2830                                        yield fallback_chunk
2831                                        bytes_written += len(fallback_chunk)
2832                            if crossfade_bytes_written:
2833                                # the blend really played, so credit both of its sides with it
2834                                for faded_item in (queue_track, outgoing_queue_track):
2835                                    if faded_item is None:
2836                                        continue
2837                                    self._report_crossfade_mode(
2838                                        queue.queue_id,
2839                                        faded_item,
2840                                        pcm_format,
2841                                        applied_mode,
2842                                        flow_session_id,
2843                                        overlay_enabled=overlay_active(queue),
2844                                    )
2845                            if remaining_bytes:
2846                                for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2847                                    yield pcm_slice
2848                                    await asyncio.sleep(0)
2849                                bytes_written += len(remaining_bytes)
2850                                del remaining_bytes
2851                            # the position was reported for the planned overlap; an
2852                            # incoming stream that ended short blended less than that,
2853                            # and the track must not be reported past its own audio
2854                            blended = max(
2855                                0, mix_start_collected + overlap_pulled - len(overlap_overshoot)
2856                            )
2857                            queue_track.streamdetails.seek_position = min(
2858                                queue_track.streamdetails.seek_position,
2859                                raw_seek_position
2860                                + blended / pcm_sample_size * track_playback_speed,
2861                            )
2862                            last_fadeout_part = b""
2863                            last_streamdetails = None
2864                            last_queue_track = None
2865                            crossfade_buffer = bytearray()
2866
2867                        # yield everything above the current holdback window; the
2868                        # slice can run short of a whole second when the window is
2869                        # small, so credit what is actually yielded - a nominal
2870                        # full-second credit inflates the play log and pins the
2871                        # queue's position mapping to the wrong track
2872                        while len(crossfade_buffer) > hold_target:
2873                            pcm_slice = bytes(crossfade_buffer[:pcm_sample_size])
2874                            yield pcm_slice
2875                            bytes_written += len(pcm_slice)
2876                            del crossfade_buffer[: len(pcm_slice)]
2877                            await asyncio.sleep(0)
2878
2879                # A source error after partial audio must not look like a completed item.
2880                # Progress reporting skips items with stream_error, so the item is not
2881                # marked played; move on to the next queue item like the zero-audio path.
2882                if first_chunk_received and queue_track.streamdetails.stream_error:
2883                    if _superseded():
2884                        return
2885                    self.logger.warning(
2886                        "Track %s (%s) on queue %s aborted by a stream error - skipping",
2887                        queue_track.name,
2888                        queue_track.streamdetails.uri,
2889                        queue.display_name,
2890                    )
2891                    # the audio sent so far will still play out; keep the play log entry
2892                    # honest about how much of this item was actually streamed
2893                    play_log_entry.seconds_streamed = bytes_written / pcm_sample_size
2894                    if last_fadeout_part:
2895                        # crossfade into this item never happened — undo the eager seek_position
2896                        queue_track.streamdetails.seek_position = raw_seek_position
2897                    continue
2898
2899                #### HANDLE END OF TRACK
2900                if not first_chunk_received:
2901                    self.logger.warning(
2902                        "Track %s (%s) on queue %s produced no audio data - skipping",
2903                        queue_track.name,
2904                        queue_track.streamdetails.uri if queue_track.streamdetails else "unknown",
2905                        queue.display_name,
2906                    )
2907                    queue_track.streamdetails.stream_error = True
2908                    play_log_entry.seconds_streamed = 0
2909                    if last_fadeout_part:
2910                        queue_track.streamdetails.seek_position = raw_seek_position
2911                    continue
2912                if last_fadeout_part:
2913                    # edge case: we did not get enough data to make the crossfade
2914                    # attribute these bytes to the previous track (they are its tail)
2915                    for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2916                        yield pcm_slice
2917                        await asyncio.sleep(0)
2918                    # no crossfade happened — undo the eager seek_position
2919                    queue_track.streamdetails.seek_position = raw_seek_position
2920                    # full tail was pre-counted and is now yielded as-is
2921                    last_fadeout_part = b""
2922                # a fade needs enough of the outgoing track to overlap with; a holdback that
2923                # armed late (or not at all) leaves less than that
2924                min_fade_out_size = int(pcm_sample_size * MIN_CROSSFADE_DURATION)
2925                if len(crossfade_buffer) >= min_fade_out_size and self.crossfade_allowed(
2926                    queue_track,
2927                    crossfade_mode=item_crossfade_mode,
2928                    player_id=queue.queue_id,
2929                    flow_mode=True,
2930                ):
2931                    last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:])
2932                    last_streamdetails = queue_track.streamdetails
2933                    last_queue_track = queue_track
2934                    last_play_log_entry = play_log_entry
2935                    remaining_bytes = bytes(crossfade_buffer[:-crossfade_buffer_size])
2936                    if remaining_bytes:
2937                        for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2938                            yield pcm_slice
2939                            await asyncio.sleep(0)
2940                        bytes_written += len(remaining_bytes)
2941                    del remaining_bytes
2942                elif item_crossfade_mode != CrossfadeMode.DISABLED and crossfade_buffer:
2943                    bytes_written += len(crossfade_buffer)
2944                    for pcm_slice in iter_pcm_slices(bytes(crossfade_buffer), pcm_format, 1000):
2945                        yield pcm_slice
2946                        await asyncio.sleep(0)
2947                crossfade_buffer = bytearray()
2948
2949                # update duration details based on the actual pcm data we sent
2950                # this also accounts for crossfade and silence stripping
2951                seconds_streamed = bytes_written / pcm_sample_size
2952                queue_track.streamdetails.seconds_streamed = seconds_streamed
2953                play_log_entry.seconds_streamed = seconds_streamed
2954                # an externally aborted source ends in a clean EOF mid-track, so the
2955                # streamed length must not be written back as the item's duration
2956                source_buffer = queue_track.streamdetails.buffer
2957                source_aborted = source_buffer is not None and source_buffer.cancelled
2958                if not source_aborted:
2959                    # the held-back crossfade tail still counts as this track's media-time
2960                    tail_seconds = len(last_fadeout_part) / pcm_sample_size
2961                    # streamdetails.duration is in media-time; seconds_streamed is stream-time
2962                    # (post-atempo), so we scale by the track's playback_speed to recover media-time.
2963                    queue_track.streamdetails.duration = int(
2964                        queue_track.streamdetails.seek_position
2965                        + (seconds_streamed + tail_seconds) * track_playback_speed
2966                    )
2967                    # propagate accurate duration to queue_item so UI displays it
2968                    queue_track.duration = queue_track.streamdetails.duration
2969                    play_log_entry.duration = queue_track.streamdetails.duration
2970                if last_play_log_entry is play_log_entry and last_fadeout_part:
2971                    # Pre-count the full crossfade tail so the queue index calculation
2972                    # doesn't undercount while waiting for the next track's crossfade mix.
2973                    # This will be corrected to crossfade_total/2 once the mix completes.
2974                    assert play_log_entry.seconds_streamed is not None
2975                    play_log_entry.seconds_streamed += len(last_fadeout_part) / pcm_sample_size
2976                self.logger.debug(
2977                    "Finished Streaming queue track: %s (%s) on queue %s",
2978                    queue_track.streamdetails.uri,
2979                    queue_track.name,
2980                    queue.display_name,
2981                )
2982        finally:
2983            await incoming_prefetcher.close()
2984        #### HANDLE END OF QUEUE FLOW STREAM
2985        # skip end-of-queue bookkeeping if a newer producer has superseded us;
2986        # the new producer owns queue_buffer_completed and the play log now
2987        if _superseded():
2988            self.logger.debug(
2989                "Flow stream for queue %s superseded - skipping end-of-queue handling",
2990                queue.display_name,
2991            )
2992            return
2993        # end of queue flow: make sure we yield the last_fadeout_part
2994        if last_fadeout_part:
2995            for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2996                yield pcm_slice
2997                await asyncio.sleep(0)
2998            # correct seconds streamed - the duration already includes the tail
2999            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
3000            streamdetails = queue_track.streamdetails
3001            assert streamdetails is not None
3002            streamdetails.seconds_streamed = (
3003                streamdetails.seconds_streamed or 0
3004            ) + last_part_seconds
3005            # also update the play log entry so elapsed time tracking stays in sync
3006            if last_play_log_entry:
3007                assert last_play_log_entry.seconds_streamed is not None
3008                # full tail was pre-counted and is now yielded as-is
3009                last_play_log_entry.duration = streamdetails.duration
3010            last_fadeout_part = b""
3011        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
3012        # only signal completion if we are still the active producer — a later
3013        # producer would (incorrectly) see this as its own completion otherwise
3014        if not _superseded():
3015            # inform the queue controller that all audio data has been generated
3016            # so it can handle the case where new items were added after the flow stream ended
3017            self.mass.player_queues.queue_buffer_completed(queue.queue_id, queue_exhausted)
3018
3019    async def get_overlay_mixed_stream(
3020        self,
3021        queue: PlayerQueue,
3022        audio_input: AsyncGenerator[bytes],
3023        pcm_format: AudioFormat,
3024    ) -> AsyncGenerator[bytes]:
3025        """
3026        Mix the queue's audio overlay (looping sound effect) into the given PCM stream.
3027
3028        The mixed output has the exact same PCM format, duration and chunking as the
3029        input stream. If the overlay source can not be resolved, the original stream
3030        is passed through unchanged so playback is never interrupted.
3031
3032        :param queue: The PlayerQueue holding the overlay source and volume.
3033        :param audio_input: The audio stream (raw PCM in ``pcm_format``) to mix into.
3034        :param pcm_format: PCM format of both the input and the mixed output.
3035        """
3036        overlay_input = await self._resolve_overlay_input(queue)
3037        if overlay_input is None:
3038            # overlay source unavailable: degrade gracefully to music-only
3039            async for chunk in audio_input:
3040                yield chunk
3041            return
3042        async for chunk in get_ffmpeg_overlay_stream(
3043            audio_input=audio_input,
3044            overlay_input=overlay_input,
3045            pcm_format=pcm_format,
3046            overlay_volume=queue.overlay_volume,
3047            chunk_size=pcm_format.pcm_sample_size,
3048        ):
3049            yield chunk
3050
3051    def crossfade_allowed(
3052        self,
3053        queue_item: QueueItem,
3054        crossfade_mode: CrossfadeMode,
3055        player_id: str,
3056        flow_mode: bool = False,
3057        next_queue_item: QueueItem | None = None,
3058        sample_rate: int | None = None,
3059        next_sample_rate: int | None = None,
3060    ) -> bool:
3061        """Get the crossfade config for a queue item."""
3062        if crossfade_mode == CrossfadeMode.DISABLED:
3063            return False
3064        if not (self.mass.player_queues.get(queue_item.queue_id)):
3065            return False  # just a guard
3066        if not (self.mass.players.get_player(player_id)):
3067            return False  # just a guard
3068        if queue_item.media_type != MediaType.TRACK:
3069            self.logger.debug("Skipping crossfade: current item is not a track")
3070            return False
3071        # check if the next item is part of the same album
3072        next_item = next_queue_item or self.mass.player_queues.get_next_item(
3073            queue_item.queue_id, queue_item.queue_item_id
3074        )
3075        if not next_item:
3076            # there is no next item!
3077            return False
3078        # check if next item is a track
3079        if next_item.media_type != MediaType.TRACK:
3080            self.logger.debug("Skipping crossfade: next item is not a track")
3081            return False
3082        # an item picks up its library album only once it is loaded, so a queue fed straight
3083        # from a provider can hold the provider album on the side that is not loaded yet.
3084        # Matching on the provider mappings recognises both shapes as the same album; the
3085        # uri-based equality of the album objects does not.
3086        if (
3087            isinstance(queue_item.media_item, Track)
3088            and isinstance(next_item.media_item, Track)
3089            and queue_item.media_item.album
3090            and next_item.media_item.album
3091            and compare_item_ids(queue_item.media_item.album, next_item.media_item.album)
3092            and not self.mass.config.get_raw_core_config_value(
3093                "streams", CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
3094            )
3095        ):
3096            # in general, crossfade is not desired for tracks of the same (gapless) album
3097            # because we have no accurate way to determine if the album is gapless or not,
3098            # for now we just never crossfade between tracks of the same album
3099            self.logger.debug("Skipping crossfade: next item is part of the same album")
3100            return False
3101
3102        # check if we're allowed to crossfade on different sample rates
3103        if (
3104            not flow_mode
3105            and sample_rate
3106            and next_sample_rate
3107            and sample_rate != next_sample_rate
3108            and not self.mass.config.get_raw_player_config_value(
3109                player_id,
3110                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key,
3111                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.default_value,
3112            )
3113        ):
3114            self.logger.debug(
3115                "Skipping crossfade: player(protocol) does not support gapless playback "
3116                "with different sample rates (%s vs %s)",
3117                sample_rate,
3118                next_sample_rate,
3119            )
3120            return False
3121
3122        return True
3123
3124    def clear_crossfade_handover(self, queue_id: str) -> None:
3125        """
3126        Clear any pending crossfade data for a queue.
3127
3128        :param queue_id: The queue ID to clear crossfade data for.
3129        """
3130        if (stale := self._crossfade_handover.pop(queue_id, None)) is not None:
3131            self.logger.debug("Clearing crossfade data for queue %s", queue_id)
3132            if stale.stream is not None:
3133                self.mass.create_task(stale.close())
3134        # and release anything waiting on a fade this queue will never finish mixing
3135        if pending := self._crossfade_pending.pop(queue_id, None):
3136            pending[1].set()
3137
3138    async def get_shoutcast_stream(
3139        self, url: str, streamdetails: StreamDetails
3140    ) -> AsyncGenerator[bytes]:
3141        """
3142        Yield audio from a legacy Shoutcast server, with ICY metadata parsed inline.
3143
3144        :param url: Shoutcast stream URL.
3145        :param streamdetails: StreamDetails to update with ICY metadata as it arrives.
3146        """
3147        self.logger.debug("Start streaming from legacy Shoutcast server: %s", url)
3148
3149        parsed = urlparse(url)
3150        host = parsed.hostname
3151        port = parsed.port or 80
3152        path = parsed.path or "/"
3153        if parsed.query:
3154            path = f"{path}?{parsed.query}"
3155
3156        try:
3157            # Open raw socket connection
3158            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=30)
3159        except TimeoutError as err:
3160            raise AudioError(f"Timeout connecting to Shoutcast stream {url}") from err
3161        except (OSError, ConnectionError) as err:
3162            raise AudioError(f"Failed to connect to Shoutcast stream {url}") from err
3163
3164        try:
3165            # Send HTTP request with ICY metadata header
3166            request = (
3167                f"GET {path} HTTP/1.1\r\n"
3168                f"Host: {host}\r\n"
3169                f"User-Agent: {HTTP_HEADERS['User-Agent']}\r\n"
3170                f"Icy-MetaData: 1\r\n\r\n"
3171            )
3172            writer.write(request.encode())
3173            await writer.drain()
3174
3175            # Read and parse response line
3176            try:
3177                response_line = await asyncio.wait_for(reader.readline(), timeout=10)
3178            except TimeoutError as err:
3179                raise AudioError("Timeout reading Shoutcast response") from err
3180
3181            if not response_line.startswith(b"ICY"):
3182                raise InvalidDataError("Invalid Shoutcast response")
3183
3184            # Read headers until empty line
3185            headers: dict[str, str] = {}
3186            while True:
3187                try:
3188                    line = await asyncio.wait_for(reader.readline(), timeout=5)
3189                except TimeoutError as err:
3190                    raise AudioError("Timeout reading Shoutcast headers") from err
3191
3192                if line in (b"\r\n", b"\n", b""):
3193                    break
3194
3195                if b":" in line:
3196                    try:
3197                        key, value = line.decode("latin-1", errors="ignore").split(":", 1)
3198                        headers[key.strip().lower()] = value.strip()
3199                    except UnicodeDecodeError, ValueError:
3200                        continue
3201
3202            # Get metadata interval
3203            meta_int_str = headers.get("icy-metaint")
3204            if not meta_int_str:
3205                raise InvalidDataError("No icy-metaint header in Shoutcast response")
3206
3207            try:
3208                meta_int = int(meta_int_str)
3209            except ValueError as err:
3210                raise InvalidDataError("Invalid icy-metaint value") from err
3211
3212            self.logger.debug("Connected to Shoutcast stream %s (icy-metaint: %s)", url, meta_int)
3213
3214            # Stream audio data with metadata parsing
3215            while True:
3216                try:
3217                    # Read audio chunk
3218                    audio_chunk = await reader.readexactly(meta_int)
3219                    yield audio_chunk
3220
3221                    # Read metadata length
3222                    meta_byte = await reader.readexactly(1)
3223                    if meta_byte == b"\x00":
3224                        continue
3225
3226                    meta_length = ord(meta_byte) * 16
3227                    meta_data = await reader.readexactly(meta_length)
3228                    self._parse_icy_metadata(meta_data, streamdetails)
3229
3230                except asyncio.exceptions.IncompleteReadError:
3231                    # End of stream
3232                    break
3233
3234        finally:
3235            writer.close()
3236            await writer.wait_closed()
3237
3238    # --- Private methods ---
3239
3240    def _notify_provider_streamed(
3241        self, streamdetails: StreamDetails, finished: bool, seconds_streamed: float
3242    ) -> None:
3243        """Report a (mostly) streamed item back to the provider that owns it."""
3244        if not finished and seconds_streamed < 90:
3245            return
3246        provider = self.mass.get_provider(streamdetails.provider)
3247        # plugin providers serve playable items too, but on_streamed is MusicProvider-only
3248        if provider is None or provider.type != ProviderType.MUSIC:
3249            return
3250        music_prov = cast("MusicProvider", provider)
3251        self.mass.create_task(music_prov.on_streamed(streamdetails))
3252
3253    def _get_volume_normalization_preference(
3254        self, streamdetails: StreamDetails
3255    ) -> VolumeNormalizationMode:
3256        """Return the configured normalization preference for the stream's media type."""
3257        conf_key = (
3258            CONF_VOLUME_NORMALIZATION_RADIO
3259            if streamdetails.media_type == MediaType.RADIO
3260            else CONF_VOLUME_NORMALIZATION_TRACKS
3261        )
3262        preference = VolumeNormalizationMode(
3263            self.mass.streams.get_config_value(conf_key, return_type=str)
3264        )
3265        # a stored value the options never offered is not a preference: nothing
3266        # validates a saved config value against them
3267        if preference in OUTCOME_ONLY_NORMALIZATION_MODES:
3268            return DEFAULT_VOLUME_NORMALIZATION_MODE
3269        return preference
3270
3271    def _update_radio_stream_metadata(
3272        self,
3273        streamdetails: StreamDetails,
3274        artist: str | None,
3275        title: str,
3276        image_url: str | None = None,
3277        album: str | None = None,
3278    ) -> None:
3279        """
3280        Update radio stream metadata and trigger artwork lookup.
3281
3282        :param streamdetails: The stream details to update.
3283        :param artist: Artist name (will be normalized).
3284        :param title: Track title (will be cleaned for display).
3285        :param image_url: Optional image URL from stream metadata.
3286        :param album: Optional album name.
3287        """
3288        station_image_url = image_url or self.mass.metadata.get_radio_stream_station_image(
3289            streamdetails
3290        )
3291        artist_normalized = (
3292            self.mass.metadata.normalize_radio_artist_name(artist) if artist else None
3293        )
3294        display_title, _ = parse_title_and_version(title, strip_for_display=True)
3295
3296        streamdetails.stream_metadata = StreamMetadata(
3297            title=display_title,
3298            artist=artist_normalized,
3299            album=album,
3300            image_url=station_image_url,
3301        )
3302        streamdetails.stream_metadata_last_updated = time.time()
3303        if streamdetails.queue_id:
3304            self.mass.player_queues.signal_update(streamdetails.queue_id)
3305
3306        # Fetch artwork in background (track, album then artist)
3307        if artist and title and not image_url:
3308            self.mass.call_later(
3309                0.2,
3310                self.mass.metadata.update_radio_stream_artwork,
3311                streamdetails,
3312                task_id=f"update_radio_artwork_{streamdetails.queue_id}",
3313            )
3314
3315    async def _cache_radio_result(
3316        self,
3317        url: str,
3318        stream_type: StreamType,
3319        resolved_url: str | None = None,
3320    ) -> tuple[str, StreamType]:
3321        """Cache and return a radio stream resolution result."""
3322        result = (resolved_url or url, stream_type)
3323        await self.mass.cache.set(
3324            url,
3325            result,
3326            expiration=3600 * 3,
3327            provider=CACHE_PROVIDER,
3328            category=CACHE_CATEGORY_RESOLVED_RADIO_URL,
3329        )
3330        return result
3331
3332    async def _handle_client_error_for_radio_stream(
3333        self, url: str, err: aiohttp.ClientError, fallback_stream_type: StreamType
3334    ) -> tuple[str, StreamType]:
3335        """Handle aiohttp client errors during radio stream resolution."""
3336        # Prefer the final post-redirect URL: aiohttp follows redirects before raising,
3337        # but the original url may just point at a redirector rather than the ICY endpoint.
3338        request_info = getattr(err, "request_info", None)
3339        validate_url = str(request_info.url) if request_info is not None else url
3340
3341        # Check if this is a Shoutcast/ICY response that aiohttp can't parse
3342        if isinstance(err, aiohttp.ClientResponseError) and "ICY" in str(err).upper():
3343            self.logger.debug(
3344                "ICY response detected for %s, validating Shoutcast stream", validate_url
3345            )
3346            if await self._validate_shoutcast_stream(validate_url):
3347                return await self._cache_radio_result(
3348                    url, StreamType.SHOUTCAST, resolved_url=validate_url
3349                )
3350            self.logger.warning(
3351                "ICY response detected but Shoutcast validation failed for %s", validate_url
3352            )
3353            return await self._cache_radio_result(
3354                url, fallback_stream_type, resolved_url=validate_url
3355            )
3356
3357        # Other aiohttp errors - might still be Shoutcast, check it
3358        self.logger.debug("aiohttp error for %s, checking if legacy Shoutcast stream", validate_url)
3359        if await self._validate_shoutcast_stream(validate_url):
3360            return await self._cache_radio_result(
3361                url, StreamType.SHOUTCAST, resolved_url=validate_url
3362            )
3363
3364        # Unknown error - still try to stream
3365        self.logger.warning(
3366            "Failed to parse radio URL %s: %s - attempting direct stream", validate_url, str(err)
3367        )
3368        return await self._cache_radio_result(url, fallback_stream_type, resolved_url=validate_url)
3369
3370    async def _get_audio_buffer(
3371        self,
3372        queue_item: QueueItem,
3373        seek_position_ms: int,
3374        reason: str,
3375        capacity_wait_timeout: float,
3376        allow_provider_match: bool,
3377    ) -> AudioBuffer:
3378        """
3379        Create or reuse a ready AudioBuffer within one queue-item preparation lock.
3380
3381        :param queue_item: Queue item whose source should be buffered.
3382        :param seek_position_ms: Position in milliseconds to start from.
3383        :param reason: Caller context for logging.
3384        :param capacity_wait_timeout: Total seconds to spend waiting for source capacity.
3385        :param allow_provider_match: Whether an on-demand cross-provider match may widen
3386            the candidates when all are saturated.
3387        """
3388        loop = asyncio.get_running_loop()
3389        # the playback intent lives on the details we start from; keep it across a reselection
3390        initial_streamdetails = queue_item.streamdetails
3391        seek_position = (
3392            int(initial_streamdetails.seek_position)
3393            if initial_streamdetails
3394            else seek_position_ms // 1000
3395        )
3396        fade_in = bool(initial_streamdetails and initial_streamdetails.fade_in)
3397        prefer_album_loudness = bool(
3398            initial_streamdetails and initial_streamdetails.prefer_album_loudness
3399        )
3400        all_candidate_instances = {
3401            provider.instance_id
3402            for mapping in (
3403                queue_item.media_item.provider_mappings if queue_item.media_item else ()
3404            )
3405            if mapping.available
3406            for provider in self._get_mapping_providers(mapping)
3407        }
3408        if initial_streamdetails is not None:
3409            all_candidate_instances.add(initial_streamdetails.provider)
3410        # a track may also exist on streaming providers it has no mapping for yet; such a
3411        # match is only searched once, and only when every known candidate is saturated
3412        match_pending = (
3413            allow_provider_match
3414            and isinstance(queue_item.media_item, Track)
3415            and self._has_alternative_match_providers(queue_item.media_item)
3416        )
3417
3418        deadline = loop.time() + capacity_wait_timeout
3419        busy_instances: set[str] = set()
3420        final_pass = False
3421        last_capacity_error: ProviderStreamLimitError | None = None
3422        last_failed_streamdetails: StreamDetails | None = None
3423        while True:
3424            if queue_item.streamdetails is None or (
3425                queue_item.streamdetails.provider in busy_instances and not final_pass
3426            ):
3427                try:
3428                    queue_item.streamdetails = await self.get_stream_details(
3429                        queue_item,
3430                        seek_position=seek_position,
3431                        fade_in=fade_in,
3432                        prefer_album_loudness=prefer_album_loudness,
3433                        excluded_provider_instances=busy_instances,
3434                    )
3435                except (AudioError, MediaNotFoundError) as err:
3436                    if last_capacity_error is None:
3437                        raise
3438                    if final_pass:
3439                        # capacity was the root cause, surface the typed (actionable) error
3440                        raise last_capacity_error from err
3441                    # no usable alternative mapping: restore the capacity-blocked details
3442                    # and spend the remaining budget blocking on that provider's slot
3443                    final_pass = True
3444                    continue
3445                finally:
3446                    if queue_item.streamdetails is None:
3447                        # never leave the queue item without streamdetails on any exit,
3448                        # including a cancellation or a non-audio provider failure
3449                        queue_item.streamdetails = last_failed_streamdetails
3450            streamdetails = queue_item.streamdetails
3451            assert streamdetails is not None  # for type checking
3452            remaining = max(deadline - loop.time(), 0)
3453            alternatives_left = bool(
3454                all_candidate_instances - busy_instances - {streamdetails.provider}
3455            )
3456            # probe (0s) whenever a reselection can still follow: a free slot is still
3457            # acquired instantly, while a busy one fails fast instead of spending the
3458            # whole budget on this candidate. Block only on the last resort.
3459            source_wait = (
3460                0.0
3461                if (not final_pass and (alternatives_left or busy_instances or match_pending))
3462                else remaining
3463            )
3464            # record whose audio this is before it exists: a queue stop releases only the
3465            # buffers of the session it is tearing down, and details resolved by an earlier
3466            # session are reused as they are, so the claim has to be made where the buffer
3467            # is attached rather than where the details came from. The queue's own session
3468            # is the owner rather than the one a caller asks for: a superseded request that
3469            # reuses a live buffer must not take it from the session still playing it
3470            streamdetails.queue_session_id = (
3471                queue_data.session_id
3472                if (queue_data := self.mass.player_queues.queue_data_or_none(queue_item.queue_id))
3473                else None
3474            )
3475            try:
3476                return await AudioBuffer.get_buffer(
3477                    mass=self.mass,
3478                    streamdetails=streamdetails,
3479                    seek_position_ms=seek_position_ms,
3480                    wait_ready=True,
3481                    reason=reason,
3482                    source_wait_timeout=source_wait,
3483                )
3484            except ProviderStreamLimitError as err:
3485                last_capacity_error = err
3486                last_failed_streamdetails = streamdetails
3487                busy_instances.add(err.provider_instance)
3488                if final_pass or loop.time() >= deadline:
3489                    raise
3490                if all_candidate_instances.issubset(busy_instances):
3491                    discovered: set[str] = set()
3492                    if match_pending:
3493                        match_pending = False
3494                        try:
3495                            discovered = await self._discover_alternative_provider_mappings(
3496                                queue_item, busy_instances, max(deadline - loop.time(), 0)
3497                            )
3498                        except Exception as err:
3499                            # discovery is best-effort: any failure falls back to the
3500                            # final blocking wait instead of replacing the typed error
3501                            self.logger.warning(
3502                                "Alternative provider search for %s failed: %s",
3503                                queue_item.name,
3504                                err,
3505                            )
3506                    if discovered:
3507                        all_candidate_instances.update(discovered)
3508                    else:
3509                        # every candidate is saturated: one last blocking wait on the best one
3510                        busy_instances.clear()
3511                        final_pass = True
3512                queue_item.streamdetails = None
3513            except AudioError:
3514                if last_capacity_error is None or final_pass:
3515                    raise
3516                # a broken alternate must not turn a transient capacity miss into a hard
3517                # failure: restore the blocked details and spend the rest of the budget there
3518                queue_item.streamdetails = last_failed_streamdetails
3519                final_pass = True
3520
3521    def _get_streamdetail_candidates(
3522        self,
3523        provider_mappings: Iterable[ProviderMapping],
3524        preferred_providers: list[str],
3525        excluded_provider_instances: set[str],
3526    ) -> list[tuple[ProviderMapping, Provider]]:
3527        """
3528        Return mapping candidates in steering, quality, and instance-fallback order.
3529
3530        :param provider_mappings: Mappings attached to the media item.
3531        :param preferred_providers: Provider instances tried before widening to the rest.
3532        :param excluded_provider_instances: Provider instances unavailable to this attempt.
3533        :return: Ordered provider mapping candidates.
3534        """
3535        ordered_mappings = sorted(
3536            provider_mappings, key=lambda mapping: mapping.quality or 0, reverse=True
3537        )
3538        preferred_candidates: list[tuple[ProviderMapping, Provider]] = []
3539        fallback_candidates: list[tuple[ProviderMapping, Provider]] = []
3540        seen_candidates: set[tuple[str, str]] = set()
3541        for mapping in ordered_mappings:
3542            if not mapping.available:
3543                self.logger.debug("Skipping unavailable %s", mapping)
3544                continue
3545            for provider in self._get_mapping_providers(mapping):
3546                candidate_id = (provider.instance_id, mapping.item_id)
3547                if (
3548                    candidate_id in seen_candidates
3549                    or provider.instance_id in excluded_provider_instances
3550                ):
3551                    continue
3552                seen_candidates.add(candidate_id)
3553                candidate = (mapping, provider)
3554                if provider.instance_id in preferred_providers:
3555                    preferred_candidates.append(candidate)
3556                else:
3557                    fallback_candidates.append(candidate)
3558        return [*preferred_candidates, *fallback_candidates]
3559
3560    def _get_mapping_providers(self, mapping: ProviderMapping) -> list[Provider]:
3561        """
3562        Return the mapped provider followed by compatible instances of its streaming catalog.
3563
3564        :param mapping: Provider mapping whose item ID will be requested.
3565        :return: Loaded provider instances that can resolve the mapping.
3566        """
3567        providers: list[Provider] = []
3568        if (
3569            primary_provider := self.mass.get_provider(
3570                mapping.provider_instance, return_unavailable=True
3571            )
3572        ) and primary_provider.available:
3573            providers.append(primary_provider)
3574        # another account of the same streaming catalog serves the same item ID,
3575        # so it can stand in when the mapped instance can not
3576        for provider in self.mass.providers:
3577            if (
3578                not isinstance(provider, MusicProvider)
3579                or not provider.available
3580                or not provider.is_streaming_provider
3581                or provider.domain != mapping.provider_domain
3582                or provider in providers
3583            ):
3584                continue
3585            providers.append(provider)
3586        if not providers:
3587            self.logger.debug("Skipping %s - provider not available", mapping)
3588        return providers
3589
3590    def _is_match_candidate_provider(
3591        self, provider: MusicProvider, known_domains: set[str]
3592    ) -> bool:
3593        """
3594        Return whether a provider is eligible to search a track match on.
3595
3596        :param provider: Music provider to check.
3597        :param known_domains: Provider domains the track already has mappings for.
3598        """
3599        return (
3600            provider.available
3601            and provider.is_streaming_provider
3602            and ProviderFeature.SEARCH in provider.supported_features
3603            and provider.domain not in known_domains
3604            and MediaType.TRACK in provider.supported_media_types
3605        )
3606
3607    def _has_alternative_match_providers(self, media_item: Track) -> bool:
3608        """
3609        Return whether any configured streaming provider could carry an unmapped match.
3610
3611        :param media_item: Track whose existing mappings define the known provider domains.
3612        """
3613        known_domains = {mapping.provider_domain for mapping in media_item.provider_mappings}
3614        return any(
3615            self._is_match_candidate_provider(provider, known_domains)
3616            for provider in self.mass.music.providers
3617        )
3618
3619    async def _discover_alternative_provider_mappings(
3620        self, queue_item: QueueItem, busy_instances: set[str], remaining: float
3621    ) -> set[str]:
3622        """
3623        Search other streaming providers for the queue item's track and widen its mappings.
3624
3625        A found mapping is added to the media item (and persisted for library items) so the
3626        capacity reselection can continue on the discovered provider.
3627
3628        :param queue_item: Queue item whose track should be matched on another provider.
3629        :param busy_instances: Provider instances already known to be saturated.
3630        :param remaining: Seconds left of the caller's capacity budget.
3631        :return: Provider instances able to serve the discovered mappings.
3632        """
3633        media_item = queue_item.media_item
3634        if not isinstance(media_item, Track):
3635            return set()
3636        known_domains = {mapping.provider_domain for mapping in media_item.provider_mappings}
3637        eligible = [
3638            provider
3639            for provider in self.mass.music.providers
3640            if self._is_match_candidate_provider(provider, known_domains)
3641            and provider.instance_id not in busy_instances
3642            and provider.has_available_stream_slot
3643        ]
3644        if not eligible:
3645            return set()
3646        # mirror the playback user's provider steering for the search order
3647        if (
3648            (pq_data := self.mass.player_queues.queue_data_or_none(queue_item.queue_id))
3649            and pq_data.userid
3650            and (playback_user := await self.mass.webserver.auth.get_user(pq_data.userid))
3651            and playback_user.provider_filter
3652        ):
3653            preferred = set(playback_user.provider_filter)
3654            eligible.sort(key=lambda provider: provider.instance_id not in preferred)
3655        # one instance per domain: a found mapping widens to sibling instances anyway
3656        candidates: list[MusicProvider] = []
3657        for provider in eligible:
3658            if provider.domain in known_domains:
3659                continue
3660            known_domains.add(provider.domain)
3661            candidates.append(provider)
3662        # the track's own album is free, sufficient evidence for the strict compare and
3663        # avoids match_provider's multi-provider album lookup on every call
3664        ref_albums = [media_item.album] if isinstance(media_item.album, Album) else []
3665        matches: list[ProviderMapping] = []
3666        try:
3667            async with asyncio.timeout(min(STREAM_SLOT_MATCH_TIMEOUT, remaining)):
3668                for provider in candidates:
3669                    # one failing provider must not end the search on the others
3670                    try:
3671                        matches = await self.mass.music.tracks.match_provider(
3672                            media_item, provider, strict=True, ref_albums=ref_albums
3673                        )
3674                    except Exception as err:
3675                        self.logger.debug("Searching a match on %s failed: %s", provider.name, err)
3676                        continue
3677                    if matches:
3678                        break
3679        except TimeoutError:
3680            self.logger.debug("Searching an alternative provider for %s timed out", media_item.name)
3681        if not matches:
3682            return set()
3683        media_item.provider_mappings.update(matches)
3684        if media_item.provider == "library":
3685            # persist in the background so future plays have the mapping ahead of time;
3686            # cancellation of this playback must never interrupt the library write
3687            self.mass.create_task(
3688                self.mass.music.tracks.add_provider_mappings(media_item.item_id, matches)
3689            )
3690        self.logger.info(
3691            "All known sources for %s are at their stream limit, "
3692            "using a matching track found on %s",
3693            media_item.name,
3694            matches[0].provider_domain,
3695        )
3696        return {
3697            provider.instance_id
3698            for mapping in matches
3699            for provider in self._get_mapping_providers(mapping)
3700        }
3701
3702    async def _request_streamdetails(
3703        self,
3704        candidates: Iterable[tuple[ProviderMapping, Provider]],
3705        media_type: MediaType,
3706    ) -> StreamDetails | None:
3707        """
3708        Request stream details from ordered provider mapping candidates.
3709
3710        :param candidates: Candidates in mapping and compatible-instance order.
3711        :param media_type: Media type requested from each provider.
3712        :return: The first resolved stream details, or None when every candidate failed.
3713        :raises AudioError: The last (actionable) audio error when no candidate resolved.
3714        """
3715        last_audio_error: AudioError | None = None
3716        for mapping, provider in candidates:
3717            # music and plugin providers share this signature, so either type can own the item
3718            token = BYPASS_THROTTLER.set(True)
3719            try:
3720                stream_prov = cast("MusicProvider | PluginProvider", provider)
3721                return await stream_prov.get_stream_details(mapping.item_id, media_type)
3722            except AudioError as err:
3723                # remember the last one so its (actionable) message can be re-raised
3724                last_audio_error = err
3725                self.logger.warning("%s", err)
3726            except MusicAssistantError as err:
3727                self.logger.warning("%s", err)
3728            finally:
3729                BYPASS_THROTTLER.reset(token)
3730        if last_audio_error is not None:
3731            raise last_audio_error
3732        return None
3733
3734    async def _get_media_stream(
3735        self,
3736        streamdetails: StreamDetails,
3737        pcm_format: AudioFormat,
3738        seek_position: int,
3739        filter_params: list[str] | None,
3740        chunk_seconds: float,
3741    ) -> AsyncGenerator[bytes]:
3742        """
3743        Stream one provider source as raw PCM.
3744
3745        :param streamdetails: Details of the stream to fetch.
3746        :param pcm_format: Target PCM format the consumer expects.
3747        :param seek_position: Requested seek offset in seconds.
3748        :param filter_params: Optional ffmpeg filter expressions.
3749        :param chunk_seconds: Size of each yielded chunk in seconds of audio.
3750        """
3751        mass = self.mass
3752        logger = self.logger.getChild("media_stream")
3753        logger.log(VERBOSE_LOG_LEVEL, "Starting media stream for %s", streamdetails.uri)
3754        # copy: the args below are appended per call, while the StreamDetails is cached on
3755        # the queue item and reused across calls (retry, seek, background analysis)
3756        extra_input_args = list(streamdetails.extra_input_args or [])
3757        # the resolver below zeroes out seek_position where the seek is delegated to the
3758        # source itself, so keep the requested position for the duration writeback
3759        requested_seek_position = seek_position
3760
3761        # work out audio source for these streamdetails
3762        audio_source, seek_position, extra_input_args = await self._resolve_media_stream_source(
3763            streamdetails, seek_position, extra_input_args
3764        )
3765
3766        # pace ffmpeg at native rate for live sources; the producer (e.g.
3767        # librespot's pipe backend) may otherwise write faster than realtime.
3768        # The initial burst grants a small bounded read-ahead so downstream
3769        # jitter does not immediately underrun the player. Providers that need
3770        # different pacing can pass their own -re/-readrate args to override.
3771        if (
3772            streamdetails.media_type == MediaType.AUDIO_SOURCE
3773            and "-re" not in extra_input_args
3774            and "-readrate" not in extra_input_args
3775        ):
3776            extra_input_args += ["-readrate", "1", "-readrate_initial_burst", "0.5"]
3777
3778        # handle seek support
3779        if seek_position and streamdetails.duration and streamdetails.allow_seek:
3780            extra_input_args += ["-ss", str(int(seek_position))]
3781
3782        bytes_sent = 0
3783        finished = False
3784        cancelled = False
3785        first_chunk_received = False
3786        ffmpeg_loglevel = "debug" if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) else "info"
3787        ffmpeg_input_format = arriving_audio_format(streamdetails)
3788        ffmpeg_proc = FFMpeg(
3789            audio_input=audio_source,
3790            input_format=ffmpeg_input_format,
3791            output_format=pcm_format,
3792            filter_params=filter_params,
3793            extra_input_args=extra_input_args,
3794            collect_log_history=True,
3795            loglevel=ffmpeg_loglevel,
3796        )
3797
3798        try:
3799            await ffmpeg_proc.start()
3800            assert ffmpeg_proc.proc is not None  # for type checking
3801            if logger.isEnabledFor(VERBOSE_LOG_LEVEL):
3802                logger.log(
3803                    VERBOSE_LOG_LEVEL,
3804                    "Started media stream for %s - using streamtype: %s "
3805                    "- pcm format: %s - ffmpeg PID: %s",
3806                    streamdetails.uri,
3807                    streamdetails.stream_type,
3808                    pcm_format.content_type.value,
3809                    ffmpeg_proc.proc.pid,
3810                )
3811            else:
3812                logger.debug(
3813                    "Started media stream for %s - using streamtype: %s",
3814                    streamdetails.uri,
3815                    streamdetails.stream_type,
3816                )
3817            stream_start = mass.loop.time()
3818            chunk_size = calculate_content_length(pcm_format, chunk_seconds)
3819            chunk_iter = ffmpeg_proc.iter_chunked(chunk_size)
3820            while True:
3821                # Time the read, not the yield: catches a stalled source, ignores backpressure.
3822                read_timeout = (
3823                    STREAM_START_TIMEOUT if not first_chunk_received else STREAM_STALL_TIMEOUT
3824                )
3825                try:
3826                    async with asyncio.timeout(read_timeout):
3827                        chunk = await anext(chunk_iter)
3828                except StopAsyncIteration:
3829                    break
3830                except TimeoutError as err:
3831                    raise AudioError(f"Source stalled: no audio for {read_timeout}s") from err
3832                if not first_chunk_received:
3833                    # At this point ffmpeg has started and should now know the codec used
3834                    # for encoding the audio.
3835                    # Note: ffmpeg_proc.input_format is the same object as
3836                    # ffmpeg_input_format, so sample_rate / bit_depth / bit_rate
3837                    # parsed from the ffmpeg log already live on streamdetails too.
3838                    first_chunk_received = True
3839                    # Skip the codec_type writeback when the provider declared a
3840                    # decoded format: audio_format already holds the authoritative
3841                    # source codec and the probed value would just be the
3842                    # post-decode wire format (e.g. PCM for Spotify Connect).
3843                    if streamdetails.decoded_audio_format is None:
3844                        streamdetails.audio_format.codec_type = ffmpeg_proc.input_format.codec_type
3845                    # Some providers omit (or report 0 for) the item duration; ffmpeg can
3846                    # usually probe it from the source. Only apply when missing so we
3847                    # don't clobber an accurate provider value with a rounded one.
3848                    if ffmpeg_proc.parsed_duration is not None and not streamdetails.duration:
3849                        streamdetails.duration = ffmpeg_proc.parsed_duration
3850                    logger.debug(
3851                        "First chunk received after %.2f seconds (codec detected: %s)",
3852                        mass.loop.time() - stream_start,
3853                        ffmpeg_proc.input_format.codec_type,
3854                    )
3855                yield chunk
3856                bytes_sent += len(chunk)
3857
3858            # end of audio/track reached
3859            logger.debug("End of media stream reached for %s", streamdetails.uri)
3860            # wait until stderr also completed reading
3861            await ffmpeg_proc.wait_with_timeout(5)
3862            logger.log(
3863                VERBOSE_LOG_LEVEL,
3864                "FFmpeg process ended with return code %s for %s",
3865                ffmpeg_proc.returncode,
3866                streamdetails.uri,
3867            )
3868            # a nested source raises through the stdin feeder, where ffmpeg's own exit
3869            # would otherwise flatten it into a generic AudioError
3870            if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
3871                raise feeder_exception
3872            if ffmpeg_proc.returncode not in (0, None):
3873                log_trail = "\n".join(list(ffmpeg_proc.log_history)[-5:])
3874                raise AudioError(f"FFMpeg exited with code {ffmpeg_proc.returncode}: {log_trail}")
3875            if bytes_sent == 0:
3876                # edge case: no audio data was received at all
3877                raise AudioError("No audio was received")
3878            finished = True
3879        except (Exception, GeneratorExit, asyncio.CancelledError) as err:
3880            if isinstance(err, asyncio.CancelledError | GeneratorExit):
3881                # we were cancelled, just raise
3882                cancelled = True
3883                raise
3884            if feeder_exception := ffmpeg_proc.stdin_feeder_exception:
3885                if isinstance(feeder_exception, ProviderStreamLimitError):
3886                    raise ffmpeg_proc.stdin_feeder_exception
3887                err = feeder_exception
3888            if isinstance(err, ProviderStreamLimitError):
3889                raise
3890            # dump the last 10 lines of the log in case of an unclean exit
3891            logger.warning("\n".join(list(ffmpeg_proc.log_history)[-10:]))
3892            raise AudioError(f"Error while streaming: {err}") from err
3893        finally:
3894            # An ffmpeg wedged on an input that will never deliver again pays close()'s
3895            # full drain - some 12 seconds in practice - under a held player lock,
3896            # before the SIGKILL that was always coming. Once the process has exited
3897            # close() is free, and it is what cleans up the stdin feeder behind it.
3898            if ffmpeg_proc.returncode is not None:
3899                await ffmpeg_proc.close()
3900            else:
3901                await ffmpeg_proc.kill()
3902            # determine how many seconds we've received
3903            # for pcm output we can calculate this easily
3904            seconds_received = bytes_sent / pcm_format.pcm_sample_size if bytes_sent else 0
3905            # store accurate duration, but only for a playthrough from the very start:
3906            # a seeked stream yields the remaining audio, not the item's full length
3907            if finished and not requested_seek_position and seconds_received:
3908                streamdetails.duration = int(seconds_received)
3909
3910            logger.log(
3911                VERBOSE_LOG_LEVEL,
3912                "stream %s (with code %s) for %s",
3913                "cancelled" if cancelled else "finished" if finished else "aborted",
3914                ffmpeg_proc.returncode,
3915                streamdetails.uri,
3916            )
3917
3918    def _report_crossfade_mode(
3919        self,
3920        queue_id: str,
3921        queue_item: QueueItem,
3922        pcm_format: AudioFormat,
3923        crossfade_mode: CrossfadeMode,
3924        session_id: str | None,
3925        *,
3926        overlay_enabled: bool,
3927    ) -> None:
3928        """
3929        Publish the crossfade that is actually applied to a queue item's audio.
3930
3931        :param queue_id: Queue the item is streamed from.
3932        :param queue_item: Queue item the fade touches.
3933        :param pcm_format: Shared PCM format leaving queue processing.
3934        :param crossfade_mode: Mode of the applied fade, SOURCE when the item's own
3935            source applies it, or DISABLED when none is applied.
3936        :param session_id: Queue session that owns processing-detail updates.
3937        :param overlay_enabled: Whether an overlay is mixed into this stream.
3938        """
3939        if session_id is None or queue_item.streamdetails is None:
3940            return
3941        self.mass.streams.audio_processing.update_item_context(
3942            queue_id=queue_id,
3943            session_id=session_id,
3944            queue_item_id=queue_item.queue_item_id,
3945            queue_processing=AudioQueueProcessing(
3946                pcm_format=pcm_format,
3947                playback_speed=cast(
3948                    "float", queue_item.extra_attributes.get("playback_speed", 1.0)
3949                ),
3950                crossfade_mode=crossfade_mode,
3951                overlay_active=overlay_enabled,
3952            ),
3953            alters_audio=queue_item.streamdetails.fade_in,
3954        )
3955
3956    async def _await_pending_crossfade(
3957        self, queue: PlayerQueue, queue_item: QueueItem
3958    ) -> CrossfadeHandover | None:
3959        """
3960        Wait briefly for a fade into this item that is still being mixed.
3961
3962        :param queue: The queue this request belongs to.
3963        :param queue_item: The item whose stream is starting.
3964        :return: The fade data if it landed in time, else None.
3965        """
3966        pending = self._crossfade_pending.get(queue.queue_id)
3967        if pending is None or pending[0] != queue_item.queue_item_id:
3968            return None
3969        handoff = pending[1]
3970        self.logger.debug(
3971            "Waiting up to %.1fs for the fade into %s being mixed for queue %s",
3972            CROSSFADE_HANDOFF_WAIT,
3973            queue_item.name,
3974            queue.display_name,
3975        )
3976        waited_from = asyncio.get_event_loop().time()
3977        with suppress(TimeoutError):
3978            await asyncio.wait_for(handoff.wait(), CROSSFADE_HANDOFF_WAIT)
3979        # claimed (popped) only when it is this item's own boundary; a stale entry
3980        # for another item stays for its own request
3981        handover = self._crossfade_handover.get(queue.queue_id)
3982        if handover is not None and handover.queue_item_id == queue_item.queue_item_id:
3983            self._crossfade_handover.pop(queue.queue_id, None)
3984        else:
3985            handover = None
3986        if handover is None and self._crossfade_pending.get(queue.queue_id) is pending:
3987            # the wait was given up: un-claim the boundary so it is not published
3988            # later, playing an intro this request is about to play itself
3989            self._crossfade_pending.pop(queue.queue_id, None)
3990        self.logger.debug(
3991            "Waited %.1fs for the fade into %s on queue %s - %s",
3992            asyncio.get_event_loop().time() - waited_from,
3993            queue_item.name,
3994            queue.display_name,
3995            "landed" if handover else "gave up",
3996        )
3997        return handover
3998
3999    async def _await_realtime_fade_source(self, streamdetails: StreamDetails) -> None:
4000        """
4001        Give a realtime incoming track a bounded chance to start delivering.
4002
4003        :param streamdetails: Stream details of the incoming (fade-in) track.
4004        """
4005        if not streamdetails.is_realtime:
4006            return
4007        loop = asyncio.get_event_loop()
4008        deadline = loop.time() + REALTIME_FADE_SOURCE_WAIT
4009        while True:
4010            audio_buffer = cast("AudioBuffer | None", streamdetails.buffer)
4011            if audio_buffer is not None:
4012                if audio_buffer.has_error:
4013                    return
4014                with suppress(TimeoutError):
4015                    await asyncio.wait_for(audio_buffer.ready.wait(), deadline - loop.time())
4016                return
4017            if loop.time() >= deadline:
4018                return
4019            # the buffer appears when the source's session starts producing
4020            await asyncio.sleep(0.1)
4021
4022    def _select_buffered_crossfade(
4023        self,
4024        streamdetails: StreamDetails,
4025        crossfade_mode: CrossfadeMode,
4026        standard_crossfade_duration: int,
4027        fade_out_seconds: float,
4028        playback_speed: float = 1.0,
4029    ) -> tuple[CrossfadeMode, float]:
4030        """
4031        Select the crossfade this boundary can carry.
4032
4033        The configured mode picks the fade; the held-back outgoing tail sizes its
4034        window, up to that mode's ceiling and to what the incoming track can supply.
4035        Too short a tail to blend at all means no fade rather than a different one.
4036
4037        :param streamdetails: Incoming track stream details.
4038        :param crossfade_mode: Requested crossfade mode.
4039        :param standard_crossfade_duration: Configured standard overlap in seconds.
4040        :param fade_out_seconds: Held-back outgoing tail in seconds.
4041        :param playback_speed: Incoming track playback-speed multiplier.
4042        :return: Effective mode and fade-in duration in seconds.
4043        """
4044        audio_buffer = streamdetails.buffer
4045        if (
4046            crossfade_mode == CrossfadeMode.DISABLED
4047            or playback_speed <= 0
4048            or audio_buffer is None
4049            or audio_buffer.has_error
4050            or not audio_buffer.is_valid()
4051            or not audio_buffer.ready.is_set()
4052        ):
4053            return CrossfadeMode.DISABLED, 0
4054
4055        # The blend streams, so the incoming window does not have to be resident:
4056        # it arrives while the blend plays. The tail we held back is what bounds it.
4057        window = min(
4058            SMART_CROSSFADE_DURATION
4059            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
4060            else standard_crossfade_duration,
4061            fade_out_seconds,
4062        )
4063        if audio_buffer.eof:
4064            # the source is done, so what is resident is all there will ever be
4065            window = min(window, audio_buffer.duration_available / playback_speed)
4066        if streamdetails.duration:
4067            # a short incoming track cannot supply a long overlap, and blending into
4068            # more than half of it would leave the listener no clean part of it. The
4069            # window is stream time, the track's remaining audio is media time.
4070            remaining_media = max(0.0, streamdetails.duration - streamdetails.seek_position)
4071            window = min(window, remaining_media / playback_speed / 2)
4072        if window < MIN_CROSSFADE_DURATION:
4073            return CrossfadeMode.DISABLED, 0
4074        self.logger.debug(
4075            "Using a %.1f second %s for %s",
4076            window,
4077            crossfade_mode.value,
4078            streamdetails.uri,
4079        )
4080        return crossfade_mode, window
4081
4082    async def _resolve_media_stream_source(
4083        self,
4084        streamdetails: StreamDetails,
4085        seek_position: int,
4086        extra_input_args: list[str],
4087    ) -> tuple[str | AsyncGenerator[bytes], int, list[str]]:
4088        """
4089        Resolve the input consumed by ffmpeg for the given stream details.
4090
4091        :param streamdetails: Details of the stream to fetch.
4092        :param seek_position: Requested seek offset in seconds.
4093        :param extra_input_args: Provider-supplied ffmpeg input arguments.
4094        :return: The ffmpeg input, the remaining seek offset and the ffmpeg input arguments.
4095        """
4096        stream_type = streamdetails.stream_type
4097        if stream_type == StreamType.CUSTOM:
4098            if streamdetails.media_type == MediaType.AUDIO_SOURCE:
4099                audio_source = self._open_audio_source_generator(
4100                    streamdetails,
4101                    seek_position=seek_position if streamdetails.can_seek else 0,
4102                )
4103            else:
4104                # MusicProvider and PluginProvider both expose get_audio_stream with the same
4105                # shape. Pin the exact instance: a domain fallback would stream from a sibling
4106                # account while the source-stream slot is charged to the issuing instance.
4107                provider = self.mass.get_provider(streamdetails.provider, return_unavailable=True)
4108                if provider is None or not provider.available:
4109                    raise ProviderUnavailableError(
4110                        f"Provider {streamdetails.provider} for stream is no longer available"
4111                    )
4112                provider = cast("MusicProvider | PluginProvider", provider)
4113                audio_source = provider.get_audio_stream(
4114                    streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
4115                )
4116            return audio_source, 0 if streamdetails.can_seek else seek_position, extra_input_args
4117        if stream_type == StreamType.ICY:
4118            assert streamdetails.path is not None
4119            assert isinstance(streamdetails.path, (str, list))
4120            audio_source = self.get_reconnecting_icy_radio_stream(streamdetails.path, streamdetails)
4121            return audio_source, 0, extra_input_args
4122        if stream_type == StreamType.SHOUTCAST:
4123            assert isinstance(streamdetails.path, str)
4124            return self.get_shoutcast_stream(streamdetails.path, streamdetails), 0, extra_input_args
4125        if stream_type == StreamType.IN_BAND:
4126            assert isinstance(streamdetails.path, str)  # for type checking
4127
4128            # For IN_BAND (OGG/Opus) radio streams, use chained OGG handler.
4129            # This handles the chained OGG format by stitching logical bitstreams together
4130            # so FFmpeg sees a single continuous stream. Metadata is extracted in-band.
4131            audio_source = get_chained_ogg_stream(
4132                self.mass,
4133                streamdetails.path,
4134                metadata_callback=partial(self._handle_inband_metadata, streamdetails),
4135            )
4136            # seeking not possible on radio streams
4137            return audio_source, 0, extra_input_args
4138        if stream_type == StreamType.HLS:
4139            assert isinstance(streamdetails.path, str)  # for type checking
4140            substream = await self.get_hls_substream(streamdetails.path)
4141            if streamdetails.media_type == MediaType.RADIO:
4142                # HLS streams (especially the BBC) struggle when they're played directly
4143                # with ffmpeg, where they just stop after some minutes,
4144                # so we tell ffmpeg to loop around in this case.
4145                extra_input_args += ["-stream_loop", "-1", "-re"]
4146            return substream.path, seek_position, extra_input_args
4147
4148        # all other stream types (HTTP, FILE, etc)
4149        if stream_type == StreamType.ENCRYPTED_HTTP:
4150            assert streamdetails.decryption_key is not None  # for type checking
4151            extra_input_args += ["-decryption_key", streamdetails.decryption_key]
4152        if isinstance(streamdetails.path, list):
4153            # multi part stream, which handles the seek itself
4154            return self.get_multi_file_stream(streamdetails, seek_position), 0, extra_input_args
4155        # regular single file/url stream
4156        assert isinstance(streamdetails.path, str)  # for type checking
4157        return streamdetails.path, seek_position, extra_input_args
4158
4159    async def _iter_audio_source_pcm(
4160        self,
4161        streamdetails: StreamDetails,
4162        pcm_format: AudioFormat,
4163    ) -> AsyncGenerator[bytes]:
4164        """Yield PCM for an AudioSource, bypassing ffmpeg when formats match."""
4165        # deliberately the advertised format: an AudioSource provider states the
4166        # PCM it delivers here, and providers that advertise a codec instead rely
4167        # on the ffmpeg path below to notice their source ending
4168        if streamdetails.audio_format == pcm_format:
4169            source_gen = self._open_audio_source_generator(streamdetails)
4170            async for chunk in realtime_pcm_pacer(source_gen, pcm_format):
4171                yield chunk
4172            return
4173        # format mismatch → fall back to ffmpeg for resampling (still small chunks)
4174        async for chunk in self.get_media_stream(
4175            streamdetails=streamdetails,
4176            pcm_format=pcm_format,
4177            filter_params=None,
4178            chunk_seconds=AUDIO_SOURCE_CHUNK_SECONDS,
4179        ):
4180            yield chunk
4181
4182    def _open_audio_source_generator(
4183        self,
4184        streamdetails: StreamDetails,
4185        seek_position: int = 0,
4186    ) -> AsyncGenerator[bytes]:
4187        """
4188        Open the raw PCM generator for an AudioSource.
4189
4190        :param streamdetails: Details of the AudioSource to stream.
4191        :param seek_position: Requested seek offset in seconds.
4192        """
4193        if streamdetails.stream_type == StreamType.CUSTOM:
4194            # pin the exact instance, see _resolve_media_stream_source
4195            provider = self.mass.get_provider(streamdetails.provider, return_unavailable=True)
4196            if provider is None or not provider.available:
4197                raise ProviderUnavailableError(
4198                    f"Provider {streamdetails.provider} for stream is no longer available"
4199                )
4200            provider = cast("MusicProvider | PluginProvider", provider)
4201            audio_source = provider.get_audio_stream(
4202                streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
4203            )
4204            return audio_source_silence_keepalive(
4205                audio_source, arriving_audio_format(streamdetails)
4206            )
4207        if streamdetails.stream_type == StreamType.NAMED_PIPE:
4208            assert isinstance(streamdetails.path, str)  # for type checking
4209            return read_named_pipe(streamdetails.path)
4210        raise AudioError(f"Unsupported stream_type {streamdetails.stream_type} for AudioSource")
4211
4212    def _handle_inband_metadata(
4213        self, streamdetails: StreamDetails, metadata: dict[str, str]
4214    ) -> None:
4215        """Handle metadata extracted from a chained Ogg stream."""
4216        title = metadata.get("title", "")
4217        artist = metadata.get("artist", "")
4218        album = metadata.get("album", "")
4219        if not artist and " - " in title:
4220            artist, title = title.split(" - ", 1)
4221        if not (title or artist):
4222            return
4223
4224        stream_title = f"{artist} - {title}" if artist and title else title or artist
4225        cleaned_title = clean_stream_title(stream_title)
4226        if not cleaned_title:
4227            return
4228        if self._record_inband_stream_title(streamdetails, cleaned_title):
4229            return
4230        if cleaned_title != streamdetails.stream_title:
4231            self.logger.log(VERBOSE_LOG_LEVEL, "In-band metadata: %s", cleaned_title)
4232            streamdetails.stream_title = cleaned_title
4233            self._update_radio_stream_metadata(
4234                streamdetails,
4235                artist=artist or None,
4236                title=title or cleaned_title,
4237                album=album or None,
4238            )
4239
4240    def _record_inband_stream_title(self, streamdetails: StreamDetails, cleaned_title: str) -> bool:
4241        """
4242        Record an in-band stream title for provider-owned metadata, if applicable.
4243
4244        When a provider opts into owning stream_metadata (and stream_title is only
4245        a derived view of it), writing either from the stream reader would fight the
4246        provider. The cleaned in-band title is recorded on StreamDetails.data instead,
4247        as the identity signal for the provider callback.
4248
4249        :param streamdetails: StreamDetails carrying the stream.
4250        :param cleaned_title: Cleaned in-band stream title.
4251        :returns: True when recorded (the caller must not write stream metadata);
4252            False when no provider callback exists and normal handling applies.
4253        """
4254        if (
4255            streamdetails.stream_metadata_update_callback is None
4256            or streamdetails.data is None
4257            or not streamdetails.data.get(STREAMDETAILS_INBAND_TITLE_HANDOFF_KEY)
4258        ):
4259            return False
4260        if streamdetails.data.get(STREAMDETAILS_INBAND_TITLE_KEY) != cleaned_title:
4261            # occupancy approximates how far this detection leads audible playback
4262            buffer = streamdetails.buffer
4263            self.logger.debug(
4264                "In-band stream title: %s (buffer occupancy: %ss)",
4265                cleaned_title,
4266                buffer.size_seconds if buffer is not None else "unknown",
4267            )
4268            streamdetails.data[STREAMDETAILS_INBAND_TITLE_KEY] = cleaned_title
4269        return True
4270
4271    def _parse_icy_metadata(self, meta_data: bytes, streamdetails: StreamDetails) -> None:
4272        """
4273        Parse ICY metadata and update streamdetails.
4274
4275        Sets the cleaned stream title and, when the title parses as "Artist - Track",
4276        triggers a radio-artwork metadata update.
4277
4278        :param meta_data: Raw metadata bytes from an ICY stream chunk.
4279        :param streamdetails: StreamDetails to update with parsed title and metadata.
4280        """
4281        if not meta_data:
4282            return
4283
4284        meta_data = meta_data.rstrip(b"\0")
4285        # Match StreamTitle, handling apostrophes in titles
4286        stream_title_re = re.search(rb"StreamTitle='(.*?)';", meta_data)
4287
4288        if not stream_title_re:
4289            self.logger.log(
4290                VERBOSE_LOG_LEVEL,
4291                "ICY metadata does not contain StreamTitle field. Raw: %s",
4292                meta_data.decode("utf-8", errors="replace")[:200],
4293            )
4294            return
4295
4296        try:
4297            # in 99% of the cases the stream title is utf-8 encoded
4298            stream_title = stream_title_re.group(1).decode("utf-8")
4299        except UnicodeDecodeError:
4300            # fallback to iso-8859-1
4301            stream_title = stream_title_re.group(1).decode("iso-8859-1", errors="replace")
4302
4303        cleaned_stream_title = clean_stream_title(stream_title)
4304
4305        if not cleaned_stream_title:
4306            return
4307
4308        if self._record_inband_stream_title(streamdetails, cleaned_stream_title):
4309            return
4310
4311        if cleaned_stream_title == streamdetails.stream_title:
4312            return
4313
4314        self.logger.log(VERBOSE_LOG_LEVEL, "ICY Radio streamtitle original: %s", stream_title)
4315        self.logger.log(
4316            VERBOSE_LOG_LEVEL, "ICY Radio streamtitle cleaned: %s", cleaned_stream_title
4317        )
4318        streamdetails.stream_title = cleaned_stream_title
4319
4320        # Prefer station-provided cover art from the ICY 'StreamUrl' field (when it is
4321        # an image) over the MusicBrainz artwork lookup in _update_radio_stream_metadata.
4322        image_url = self._parse_icy_image_url(meta_data)
4323
4324        # Parse the original title for structured fields first so stations that announce
4325        # an album can refine the artwork lookup; fall back to the "Artist - Track" split.
4326        album: str | None = None
4327        if parsed := parse_quoted_stream_title(stream_title):
4328            track_name, artist_name_raw, album = parsed
4329        elif " - " in cleaned_stream_title:
4330            artist_name_raw, track_name = (
4331                part.strip() for part in cleaned_stream_title.split(" - ", 1)
4332            )
4333        else:
4334            return
4335
4336        if artist_name_raw and track_name:
4337            self.logger.debug(
4338                "ICY metadata: artist='%s', track='%s', album='%s'",
4339                artist_name_raw,
4340                track_name,
4341                album,
4342            )
4343            self._update_radio_stream_metadata(
4344                streamdetails,
4345                artist=artist_name_raw,
4346                title=track_name,
4347                album=album,
4348                image_url=image_url,
4349            )
4350
4351    def _parse_icy_image_url(self, meta_data: bytes) -> str | None:
4352        """
4353        Return a PNG or JPEG cover-art URL from the ICY 'StreamUrl' field, if present.
4354
4355        :param meta_data: Raw metadata bytes from an ICY stream chunk.
4356        """
4357        # The trailing semicolon is optional to match sources that omit it.
4358        stream_url_re = re.search(rb"StreamUrl='([^']*)'", meta_data)
4359        if not stream_url_re:
4360            return None
4361        try:
4362            image_url = stream_url_re.group(1).decode("utf-8").strip()
4363        except UnicodeDecodeError:
4364            return None
4365        if not image_url:
4366            return None
4367        # StreamUrl is not a standardized artwork field (reference clients such as VLC
4368        # ignore it and it conventionally holds a station website link), so only accept
4369        # values that point at a PNG or JPEG image.
4370        parsed = urlparse(image_url)
4371        if parsed.scheme not in ("http", "https"):
4372            return None
4373        if not parsed.path.lower().endswith((".png", ".jpg", ".jpeg")):
4374            return None
4375        self.logger.debug("ICY metadata: StreamUrl image='%s'", image_url)
4376        return image_url
4377
4378    async def _validate_shoutcast_stream(self, url: str) -> bool:
4379        """
4380        Return True if the URL responds with a legacy Shoutcast "ICY 200 OK" line.
4381
4382        :param url: The URL to validate.
4383        """
4384        try:
4385            parsed = urlparse(url)
4386            host = parsed.hostname
4387            port = parsed.port or 80
4388            path = parsed.path or "/"
4389            if parsed.query:
4390                path = f"{path}?{parsed.query}"
4391
4392            # Open raw socket connection with timeout
4393            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=10)
4394            try:
4395                # Send minimal HTTP request with ICY metadata header
4396                request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nIcy-MetaData: 1\r\n\r\n"
4397                writer.write(request.encode())
4398                await writer.drain()
4399
4400                # Read just the response line
4401                response_line = await asyncio.wait_for(reader.readline(), timeout=5)
4402            finally:
4403                writer.close()
4404                await writer.wait_closed()
4405
4406            # Check if response starts with "ICY"
4407            decoded_line = response_line.decode("latin-1", errors="ignore").strip()
4408            return decoded_line.startswith("ICY")
4409
4410        except TimeoutError:
4411            self.logger.debug("Timeout during Shoutcast validation for %s", url)
4412            return False
4413        except OSError, ConnectionError:
4414            self.logger.debug("Connection failed during Shoutcast validation for %s", url)
4415            return False
4416        except UnicodeDecodeError:
4417            self.logger.debug("Invalid response encoding during Shoutcast validation for %s", url)
4418            return False
4419
4420    def _resolve_player_dsp_config(self, player: Player) -> DSPConfig:
4421        """
4422        Resolve the effective DSP config for a player.
4423
4424        Single source of truth shared by every code path that needs to know
4425        whether DSP will run for this player. Protocol wrappers defer to their
4426        parent player; single-leg ``player_group`` instances that don't expose
4427        ``MULTI_DEVICE_DSP`` defer to their first member; players whose grouping
4428        context prevents DSP get a disabled config back regardless.
4429
4430        :param player: The player to resolve DSP config for.
4431        """
4432        dsp_player_id = self._resolve_player_dsp_config_id(player)
4433        dsp = self.mass.config.get_player_dsp_config(dsp_player_id)
4434        if is_grouping_preventing_dsp(player):
4435            dsp.enabled = False
4436        elif player.provider.domain == "player_group" and (
4437            PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
4438        ):
4439            if not player.state.group_members:
4440                dsp.enabled = False
4441        return dsp
4442
4443    def _resolve_player_dsp_config_id(self, player: Player) -> str:
4444        """
4445        Return the player identifier that supplies the effective DSP config.
4446
4447        :param player: Player whose DSP config source should be resolved.
4448        """
4449        dsp_player_id = player.protocol_parent_id or player.player_id
4450        if (
4451            not is_grouping_preventing_dsp(player)
4452            and player.provider.domain == "player_group"
4453            and PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
4454            and player.state.group_members
4455        ):
4456            child_player = self.mass.players.get_player(player.state.group_members[0])
4457            assert child_player is not None
4458            dsp_player_id = child_player.player_id
4459        return dsp_player_id
4460
4461    def _get_output_channels(self, player: Player | None, player_id: str) -> str:
4462        """
4463        Return the configured output channels for the rendering player.
4464
4465        The value may be stored on the rendering player(protocol) itself (the
4466        protocol section of the config UI) or on its visible parent player (the
4467        native section); the rendering player's own stored value wins.
4468        """
4469        parent_id = player.protocol_parent_id if player and player.protocol_parent_id else player_id
4470        parent_value = self.mass.config.get_raw_player_config_value(
4471            parent_id, CONF_OUTPUT_CHANNELS, "stereo"
4472        )
4473        return self.mass.config.get_raw_player_config_value(
4474            player.player_id if player else player_id, CONF_OUTPUT_CHANNELS, parent_value
4475        )
4476
4477    def _pick_pcm_bit_depth(
4478        self,
4479        players: Iterable[Player],
4480        streamdetails: StreamDetails | None,
4481        crossfade_enabled: bool,
4482        overlay_active: bool = False,
4483    ) -> tuple[ContentType, int]:
4484        """
4485        Return ``(content_type, bit_depth)`` for an internal PCM stream.
4486
4487        F32 is chosen when audio processing (crossfade, audio overlay, volume
4488        normalization, DSP) will run on the stream — those need the extra
4489        headroom to avoid clipping and precision loss. Otherwise the source's
4490        native bit depth is reused so we don't waste memory upcasting a 16-bit
4491        stream to 32-bit just to pass it through. When the source is unknown
4492        (no streamdetails) we fall back to F32 conservatively.
4493        """
4494        if streamdetails is None:
4495            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
4496        needs_headroom = (
4497            crossfade_enabled
4498            or overlay_active
4499            or streamdetails.volume_normalization_mode
4500            not in (VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.SOURCE)
4501            or any(self._resolve_player_dsp_config(player).enabled for player in players)
4502        )
4503        if needs_headroom:
4504            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
4505        # the depth the audio arrives in, not the one the source claims: a
4506        # provider that decoded on our behalf may advertise a narrower format
4507        # for display, and narrowing the stream to that would truncate it
4508        bit_depth = arriving_audio_format(streamdetails).bit_depth
4509        return ContentType.from_bit_depth(bit_depth), bit_depth
4510
4511    def _select_audio_source_pcm_format(
4512        self,
4513        player: Player,
4514        streamdetails: StreamDetails,
4515        supported_sample_rates: Iterable[int] | None = None,
4516    ) -> AudioFormat:
4517        """
4518        Return a passthrough PCM format for a realtime AudioSource item.
4519
4520        The format matches the source's native sample rate, bit depth and
4521        channel count whenever the player can accept them; if the player does
4522        not support the source's sample rate, it is snapped down to the
4523        closest supported rate. No F32 widening — realtime sources skip every
4524        processing stage that would otherwise need it. Surround sources are
4525        still folded down to stereo, which every output path requires anyway.
4526
4527        :param player: The player requesting the stream.
4528        :param streamdetails: Stream details for the AudioSource item.
4529        :param supported_sample_rates: Rates shared by every output player, if applicable.
4530        """
4531        resolved_sample_rates = (
4532            list(supported_sample_rates)
4533            if supported_sample_rates is not None
4534            else [sample_rate for sample_rate, _ in player.get_supported_sample_rates()]
4535        )
4536        # the format the audio arrives in, not the one the source claims: a provider
4537        # that decoded on our behalf may advertise a narrower format for display, and
4538        # narrowing the stream to that would truncate it
4539        source_format = arriving_audio_format(streamdetails)
4540        source_rate = source_format.sample_rate
4541        if source_rate in resolved_sample_rates:
4542            output_sample_rate = source_rate
4543        else:
4544            output_sample_rate = max(
4545                (rate for rate in resolved_sample_rates if rate <= source_rate),
4546                default=min(resolved_sample_rates),
4547            )
4548        return AudioFormat(
4549            content_type=ContentType.from_bit_depth(source_format.bit_depth),
4550            sample_rate=output_sample_rate,
4551            bit_depth=source_format.bit_depth,
4552            # a realtime source may announce more channels than anything downstream can
4553            # carry (a VBAN stream can be configured up to 8), and player handoff formats
4554            # copy this count straight through, so fold it here
4555            channels=min(source_format.channels, 2),
4556        )
4557
4558    def _flow_restart_context(
4559        self, queue_id: str, protocol_player: Player | None
4560    ) -> tuple[str, list[int]]:
4561        """
4562        Resolve the flow mode config and supported sample rates for restart decisions.
4563
4564        Prefers the protocol player actually consuming the flow stream over the
4565        queue's (wrapper) player, whose config may lack the audio specific entries.
4566        """
4567        if protocol_player is None:
4568            protocol_player = self.mass.players.get_player(queue_id)
4569        if protocol_player is None:
4570            flow_mode_sample_rate_conf = self.mass.config.get_raw_player_config_value(
4571                queue_id, CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
4572            )
4573            return flow_mode_sample_rate_conf, []
4574        flow_mode_sample_rate_conf = cast(
4575            "str",
4576            protocol_player.config.get_value(
4577                CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
4578            ),
4579        )
4580        supported_sample_rates = sorted(
4581            {sr for sr, _ in protocol_player.get_supported_sample_rates()}
4582        )
4583        return flow_mode_sample_rate_conf, supported_sample_rates
4584
4585    def _flow_stream_needs_restart(
4586        self,
4587        queue_track: QueueItem,
4588        pcm_format: AudioFormat,
4589        supported_sample_rates: list[int],
4590        flow_mode_sample_rate_conf: str,
4591        is_first_track: bool,
4592    ) -> bool:
4593        """
4594        Return True if the upcoming queue track requires exiting the flow stream.
4595
4596        Covers every case where the flow loop should break and hand control back to
4597        the queue controller for restart:
4598
4599        - Live media (radio, audio sources): cannot be played inside a flow,
4600          the controller will fall back to a single-item stream.
4601        - Sample rate mismatch ('smart' / 'bit_perfect' modes only): the next
4602          track's sample rate (snapped up to the closest supported player rate,
4603          mirroring select_flow_pcm_format's anchoring logic) is incompatible with
4604          the current flow rate, so a new flow must be opened.
4605
4606        The first (anchor) track is always allowed to continue for the sample
4607        rate check; select_flow_pcm_format has already snapped the flow rate to it.
4608
4609        :param queue_track: The upcoming queue item.
4610        :param pcm_format: The current flow stream's PCM format.
4611        :param supported_sample_rates: Sorted list of the player's supported rates.
4612        :param flow_mode_sample_rate_conf: The flow mode sample rate config value.
4613        :param is_first_track: Whether this is the first track of the flow stream.
4614        """
4615        # live audio (radio, plugin or audio source) cannot be flowed; let the
4616        # queue controller fall back to single-item streaming for this item
4617        if queue_track.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
4618            self.logger.info(
4619                "Live media item %s (%s, %s) encountered in flow stream "
4620                "- breaking out to single item stream",
4621                queue_track.queue_item_id,
4622                queue_track.name,
4623                queue_track.media_type,
4624            )
4625            return True
4626
4627        if is_first_track or queue_track.streamdetails is None:
4628            return False
4629        raw_next_rate = queue_track.streamdetails.audio_format.sample_rate
4630        if not raw_next_rate or not supported_sample_rates:
4631            return False
4632        effective_next_rate = _snap_supported_rate_up(raw_next_rate, supported_sample_rates)
4633
4634        # branch order mirrors select_flow_pcm_format: fixed-rate modes resample
4635        # everything to the chosen rate (no restart); bit_perfect restarts on any
4636        # mismatch; anything else falls through to smart-anchor behavior so
4637        # unknown/legacy config values don't silently pin the flow forever.
4638        if flow_mode_sample_rate_conf in (
4639            FLOW_MODE_SAMPLE_RATE_48000,
4640            FLOW_MODE_SAMPLE_RATE_96000,
4641            FLOW_MODE_SAMPLE_RATE_HIGHEST,
4642        ):
4643            needs_restart = False
4644        elif flow_mode_sample_rate_conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT:
4645            needs_restart = effective_next_rate != pcm_format.sample_rate
4646        else:
4647            needs_restart = effective_next_rate > pcm_format.sample_rate
4648
4649        if needs_restart:
4650            self.logger.info(
4651                "Track %s (%s) sample rate %s (snapped to %s) incompatible with flow rate %s "
4652                "(mode: %s) - breaking out to restart flow stream",
4653                queue_track.queue_item_id,
4654                queue_track.name,
4655                raw_next_rate,
4656                effective_next_rate,
4657                pcm_format.sample_rate,
4658                flow_mode_sample_rate_conf,
4659            )
4660        return needs_restart
4661
4662    @asynccontextmanager
4663    async def _connect_radio_stream(self, url: str, **kwargs: Any) -> AsyncGenerator[Any]:
4664        """
4665        Connect to a radio stream URL with fallback for legacy SSL/TLS configurations.
4666
4667        Some radio servers use outdated TLS configurations that reject modern
4668        cipher suites. Since radio streams are public broadcast content,
4669        relaxing cipher requirements is acceptable.
4670
4671        :param url: The radio stream URL to connect to.
4672        :param kwargs: Additional keyword arguments passed to aiohttp get().
4673        """
4674        request_url = encoded_request_url(url)
4675        try:
4676            async with self.mass.http_session_no_ssl.get(request_url, **kwargs) as resp:
4677                yield resp
4678        except ClientConnectorSSLError:
4679            self.logger.info(
4680                "SSL handshake failed for %s, retrying with permissive cipher configuration", url
4681            )
4682            insecure_ssl_context = ssl_util.client_context_no_verify(
4683                ssl_util.SSLCipherList.INSECURE
4684            )
4685            async with self.mass.http_session_no_ssl.get(
4686                request_url, ssl=insecure_ssl_context, **kwargs
4687            ) as resp:
4688                yield resp
4689
4690    async def _update_hls_radio_metadata(
4691        self,
4692        streamdetails: StreamDetails,
4693        elapsed_time: int,
4694    ) -> None:
4695        """
4696        Update HLS radio stream metadata by fetching the playlist.
4697
4698        Fetches the HLS playlist and extracts metadata from EXTINF lines.
4699
4700        :param streamdetails: StreamDetails object to update with metadata
4701        :param elapsed_time: Current playback position in seconds (unused for live radio)
4702        """
4703        mass = self.mass
4704        try:
4705            # Get the actual media playlist URL from cache or resolve it
4706            # We cache the media_playlist_url in streamdetails.data to avoid re-resolving
4707            if streamdetails.data is None:
4708                streamdetails.data = {}
4709            media_playlist_url = streamdetails.data.get("hls_media_playlist_url")
4710            if not media_playlist_url:
4711                try:
4712                    assert isinstance(streamdetails.path, str)  # for type checking
4713                    substream = await self.get_hls_substream(streamdetails.path)
4714                    media_playlist_url = substream.path
4715                    streamdetails.data["hls_media_playlist_url"] = media_playlist_url
4716                except Exception as err:
4717                    self.logger.warning(
4718                        "Failed to resolve HLS substream for metadata monitoring: %s", err
4719                    )
4720                    return
4721
4722            # Fetch the media playlist
4723            timeout = ClientTimeout(total=0, connect=10, sock_read=30)
4724            try:
4725                async with mass.http_session_no_ssl.get(
4726                    encoded_request_url(media_playlist_url), timeout=timeout
4727                ) as resp:
4728                    resp.raise_for_status()
4729                    playlist_content = await resp.text()
4730            except ClientResponseError as err:
4731                # Session token likely expired (410/403) — drop cache so next poll re-resolves
4732                if err.status in (403, 410):
4733                    streamdetails.data.pop("hls_media_playlist_url", None)
4734                raise
4735
4736            # Parse the playlist and look for EXTINF metadata
4737            # The most recent segment usually has the current metadata
4738            lines = playlist_content.strip().split("\n")
4739            for line in reversed(lines):
4740                if line.startswith("#EXTINF:"):
4741                    # Extract metadata from EXTINF line
4742                    metadata = parse_extinf_metadata(line)
4743
4744                    # Build stream title from title and artist
4745                    title = metadata.get("title", "")
4746                    artist = metadata.get("artist", "")
4747                    image_url = (
4748                        metadata.get("image") or metadata.get("artwork") or metadata.get("cover")
4749                    )
4750                    if not artist and " - " in title:
4751                        artist, title = title.split(" - ", 1)
4752                    if title or artist:
4753                        # Format as "Artist - Title"
4754                        if artist and title:
4755                            stream_title = f"{artist} - {title}"
4756                        elif title:
4757                            stream_title = title
4758                        else:
4759                            stream_title = artist
4760
4761                        # Clean the stream title
4762                        cleaned_title = clean_stream_title(stream_title)
4763
4764                        # Only update if changed
4765                        if cleaned_title != streamdetails.stream_title and cleaned_title:
4766                            self.logger.log(
4767                                VERBOSE_LOG_LEVEL, "HLS Radio metadata updated: %s", cleaned_title
4768                            )
4769                            streamdetails.stream_title = cleaned_title
4770                            self._update_radio_stream_metadata(
4771                                streamdetails,
4772                                artist=artist or None,
4773                                title=title or cleaned_title,
4774                                image_url=image_url,
4775                            )
4776
4777                    # Only check the most recent EXTINF
4778                    break
4779
4780        except Exception as err:
4781            self.logger.debug("Error fetching HLS metadata: %s", err)
4782
4783    @staticmethod
4784    def _normalize_reconnecting_urls(url: str | list[MultiPartPath]) -> list[str]:
4785        """Normalize a single URL or a sequence into a non-empty list."""
4786        if isinstance(url, str):
4787            return [url]
4788        if not url:
4789            msg = "Radio stream requires at least one URL"
4790            raise InvalidDataError(msg)
4791        return [part.path for part in url]
4792
4793    async def _resolve_overlay_input(self, queue: PlayerQueue) -> str | None:
4794        """
4795        Resolve the queue's overlay source to a file path or URL for ffmpeg.
4796
4797        Returns None (with a warning logged) when the source can not be resolved,
4798        so the caller can degrade to music-only playback.
4799        """
4800        if not (mapping := queue.overlay_source):
4801            return None
4802        try:
4803            provider = self.mass.get_provider(mapping.provider)
4804            if provider is None:
4805                raise MediaNotFoundError(f"Provider {mapping.provider} is not available")
4806            stream_prov = cast("MusicProvider | PluginProvider", provider)
4807            streamdetails = await stream_prov.get_stream_details(
4808                mapping.item_id, MediaType.SOUND_EFFECT
4809            )
4810        except Exception as err:
4811            self.logger.warning(
4812                "Audio overlay source %s is unavailable (%s) - continuing without overlay",
4813                mapping.uri,
4814                str(err) or err.__class__.__name__,
4815            )
4816            return None
4817        if streamdetails.stream_type not in (StreamType.LOCAL_FILE, StreamType.HTTP) or not (
4818            isinstance(streamdetails.path, str)
4819        ):
4820            self.logger.warning(
4821                "Audio overlay source %s uses unsupported stream type %s "
4822                "- continuing without overlay",
4823                mapping.uri,
4824                streamdetails.stream_type,
4825            )
4826            return None
4827        if streamdetails.stream_type == StreamType.LOCAL_FILE and not await aiofiles.os.path.isfile(
4828            streamdetails.path
4829        ):
4830            # guard against stale sources: feeding a missing file to the mixer would
4831            # kill the whole (music) stream instead of just the overlay
4832            self.logger.warning(
4833                "Audio overlay source %s does not exist - continuing without overlay",
4834                streamdetails.path,
4835            )
4836            return None
4837        return streamdetails.path
4838