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.abc import AsyncGenerator, Iterable
17from contextlib import asynccontextmanager
18from dataclasses import dataclass
19from functools import partial
20from typing import TYPE_CHECKING, Any, cast
21from urllib.parse import urlparse
22
23import aiofiles
24import aiofiles.os
25import aiohttp
26import shortuuid
27from aiohttp import ClientConnectorSSLError, ClientResponseError, ClientTimeout
28from music_assistant_models.audio_processing import (
29    AudioDSPDetails,
30    AudioOutputDetails,
31    AudioQueueProcessing,
32)
33from music_assistant_models.dsp import (
34    AudioChannel,
35    ConvolutionFilter,
36    DSPConfig,
37    DSPFilter,
38    DSPState,
39)
40from music_assistant_models.enums import (
41    ContentType,
42    CrossfadeMode,
43    MediaType,
44    PlayerFeature,
45    ProviderType,
46    StreamType,
47    VolumeNormalizationMode,
48)
49from music_assistant_models.errors import (
50    AudioError,
51    InvalidDataError,
52    MediaNotFoundError,
53    MusicAssistantError,
54    ProviderPermissionDenied,
55    ProviderUnavailableError,
56    QueueEmpty,
57    RetriesExhausted,
58)
59from music_assistant_models.media_items import AudioFormat, Track
60from music_assistant_models.player_queue import PlayLogEntry
61from music_assistant_models.streamdetails import MultiPartPath, StreamMetadata
62
63from music_assistant.constants import (
64    CONF_CROSSFADE_DURATION,
65    CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES,
66    CONF_ENTRY_VOLUME_NORMALIZATION_TARGET,
67    CONF_FLOW_MODE_SAMPLE_RATE,
68    CONF_OUTPUT_CHANNELS,
69    CONF_PLAYER_QUEUES,
70    CONF_VALUE_DISABLED,
71    CONF_VALUE_ENABLED,
72    CONF_VOLUME_NORMALIZATION,
73    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
74    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
75    CONF_VOLUME_NORMALIZATION_RADIO,
76    CONF_VOLUME_NORMALIZATION_TARGET,
77    CONF_VOLUME_NORMALIZATION_TRACKS,
78    DSP_IRS_DIRNAME,
79    FLOW_MODE_SAMPLE_RATE_48000,
80    FLOW_MODE_SAMPLE_RATE_96000,
81    FLOW_MODE_SAMPLE_RATE_BIT_PERFECT,
82    FLOW_MODE_SAMPLE_RATE_HIGHEST,
83    FLOW_MODE_SAMPLE_RATE_SMART,
84    INTERNAL_PCM_FORMAT,
85    MASS_LOGGER_NAME,
86    STREAM_STALL_TIMEOUT,
87    STREAM_START_TIMEOUT,
88    VERBOSE_LOG_LEVEL,
89)
90from music_assistant.controllers.streams.audio_analysis import (
91    LOUDNESS_ANALYSIS_DOMAIN,
92)
93from music_assistant.controllers.streams.audio_buffer import AudioBuffer
94from music_assistant.controllers.streams.audio_processing import (
95    AudioOutputPlan,
96    get_normalization_details,
97)
98from music_assistant.controllers.streams.constants import (
99    CACHE_CATEGORY_RESOLVED_RADIO_URL,
100    CACHE_PROVIDER,
101    CONF_ALLOW_CROSSFADE_SAME_ALBUM,
102)
103from music_assistant.controllers.streams.ogg_handler import get_chained_ogg_stream
104from music_assistant.controllers.streams.smart_fades import SmartFadesMixer
105from music_assistant.controllers.streams.smart_fades.fades import SmartFade
106from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
107from music_assistant.helpers import ssl as ssl_util
108from music_assistant.helpers.aiohttp_client import encoded_request_url
109from music_assistant.helpers.audio import (
110    HTTP_HEADERS,
111    HTTP_HEADERS_ICY,
112    build_concat_filelist,
113    calculate_content_length,
114    get_bit_rate,
115    get_normalization_mode,
116    get_parts_from_position,
117    is_grouping_preventing_dsp,
118    iter_pcm_slices,
119    parse_extinf_metadata,
120    realtime_pcm_pacer,
121    resample_pcm_audio,
122    resolve_output_player_ids,
123)
124from music_assistant.helpers.dsp import ComplexFilter, filter_to_ffmpeg_params
125from music_assistant.helpers.ffmpeg import (
126    FFMpeg,
127    get_ffmpeg_overlay_stream,
128    get_ffmpeg_stream,
129)
130from music_assistant.helpers.named_pipe import read_named_pipe
131from music_assistant.helpers.playlists import IsHLSPlaylist, PlaylistItem, fetch_playlist, parse_m3u
132from music_assistant.helpers.throttle_retry import BYPASS_THROTTLER
133from music_assistant.helpers.util import (
134    clean_stream_title,
135    detect_charset,
136    parse_quoted_stream_title,
137    parse_title_and_version,
138    remove_file,
139)
140
141if TYPE_CHECKING:
142    from music_assistant_models.player_queue import PlayerQueue
143    from music_assistant_models.queue_item import QueueItem
144    from music_assistant_models.streamdetails import StreamDetails
145
146    from music_assistant.mass import MusicAssistant
147    from music_assistant.models.music_provider import MusicProvider
148    from music_assistant.models.player import Player
149    from music_assistant.models.plugin import PluginProvider
150
151# ruff: noqa: PLR0915
152
153# Seconds of PCM yielded directly to the player before the crossfade holdback starts buffering.
154WARMUP_DURATION = 8
155
156# Chunk size for the realtime AudioSource path; small enough to keep ffmpeg→consumer
157# latency below ~50 ms while still amortising per-chunk overhead.
158AUDIO_SOURCE_CHUNK_SECONDS = 0.02
159
160# Terminal errors get_icy_radio_stream raises once a single mirror is exhausted; the
161# multi-mirror reader treats these as the signal to fail over to the next URL.
162RADIO_MIRROR_FAILOVER_ERRORS = (
163    MediaNotFoundError,
164    ProviderPermissionDenied,
165    ProviderUnavailableError,
166    RetriesExhausted,
167    InvalidDataError,
168)
169
170
171@dataclass
172class CrossfadeData:
173    """Data class to hold crossfade data."""
174
175    data: bytes
176    fade_in_size: int
177    pcm_format: AudioFormat  # Format of the 'data' bytes (current/previous track's format)
178    fade_in_pcm_format: AudioFormat  # Format for 'fade_in_size' (next track's format)
179    queue_item_id: str
180    # Offset for the fade_in track's elapsed time calculation, to account for crossfade duration and trim
181    elapsed_time_offset: float = 0.0
182    # Normalization mode the intro PCM was baked with, used to pin the next track's body to the same mode
183    normalization_mode: VolumeNormalizationMode | None = None
184
185
186def _snap_supported_rate_up(target: int, supported_sample_rates: list[int]) -> int:
187    """Snap target up, falling back to its highest supported divisor or the maximum."""
188    if target in supported_sample_rates:
189        return target
190    higher = [r for r in supported_sample_rates if r > target]
191    if higher:
192        return min(higher)
193    same_family = [r for r in supported_sample_rates if target % r == 0]
194    return max(same_family) if same_family else max(supported_sample_rates)
195
196
197def _snap_supported_rate_down(target: int, supported_sample_rates: list[int]) -> int:
198    """Snap target down to the highest supported rate <= target, falling back to min."""
199    if target in supported_sample_rates:
200        return target
201    lower = [r for r in supported_sample_rates if r < target]
202    return max(lower) if lower else min(supported_sample_rates)
203
204
205def _pcm_formats_match(a: AudioFormat, b: AudioFormat) -> bool:
206    """Return True if two PCM formats describe identical raw bytes."""
207    return (
208        a.content_type == b.content_type
209        and a.sample_rate == b.sample_rate
210        and a.bit_depth == b.bit_depth
211        and a.channels == b.channels
212    )
213
214
215def overlay_active(queue: PlayerQueue) -> bool:
216    """Return True if the given queue has an audio overlay enabled and a source selected."""
217    return queue.overlay_enabled and queue.overlay_source is not None
218
219
220class StreamsAudio:
221    """Audio stream acquisition and processing for the streams controller."""
222
223    def __init__(self, mass: MusicAssistant) -> None:
224        """
225        Initialize StreamsAudio.
226
227        :param mass: The MusicAssistant instance.
228        """
229        self.mass = mass
230        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.streams.audio")
231        self._crossfade_data: dict[str, CrossfadeData] = {}
232        self._smart_fades_mixer: SmartFadesMixer | None = None
233
234    def setup(self) -> None:
235        """Set up the audio sub-controller (called after all core controllers are created)."""
236        self._smart_fades_mixer = SmartFadesMixer(self.mass.streams)
237
238    @property
239    def smart_fades_mixer(self) -> SmartFadesMixer:
240        """Return the smart fades mixer."""
241        assert self._smart_fades_mixer is not None, "StreamsAudio.setup() not called"
242        return self._smart_fades_mixer
243
244    # --- Public methods ---
245
246    async def get_stream_details(
247        self,
248        queue_item: QueueItem,
249        seek_position: int = 0,
250        fade_in: bool = False,
251        prefer_album_loudness: bool = False,
252    ) -> StreamDetails:
253        """
254        Get streamdetails for the given QueueItem.
255
256        This is called just-in-time when a PlayerQueue wants a MediaItem to be played.
257        Do not try to request streamdetails too much in advance as this is expiring data.
258        """
259        mass = self.mass
260        streamdetails: StreamDetails | None = None
261        time_start = time.time()
262        self.logger.debug("Getting streamdetails for %s", queue_item.uri)
263
264        if not queue_item.media_item and not queue_item.streamdetails:
265            # in case of a non-media item queue item, the streamdetails should already be provided
266            # this should not happen, but guard it just in case
267            raise MediaNotFoundError(
268                f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
269            )
270
271        if queue_item.streamdetails and (
272            # reuse if the buffer can serve this seek position (fast seek path)
273            (
274                queue_item.streamdetails.buffer
275                and queue_item.streamdetails.buffer.is_valid(int(seek_position * 1000))
276            )
277            # or reuse if streamdetails hasn't expired yet (new buffer will be created)
278            or (queue_item.streamdetails.created_at + queue_item.streamdetails.expiration)
279            > time.time()
280        ):
281            streamdetails = queue_item.streamdetails
282        else:
283            # need to (re)create streamdetails
284            # retrieve streamdetails from provider
285
286            media_item = queue_item.media_item
287            assert media_item is not None  # for type checking
288            preferred_providers: list[str] = []
289            if (
290                (pq_data := mass.player_queues.queue_data_or_none(queue_item.queue_id))
291                and pq_data.userid
292                and (playback_user := await mass.webserver.auth.get_user(pq_data.userid))
293                and playback_user.provider_filter
294            ):
295                # handle steering into user preferred providerinstance
296                preferred_providers = playback_user.provider_filter
297            else:
298                preferred_providers = [x.provider_instance for x in media_item.provider_mappings]
299            # Remember the last AudioError so we can re-raise its (actionable)
300            # message instead of the generic MediaNotFoundError below.
301            last_audio_error: AudioError | None = None
302            for allow_other_provider in (False, True):
303                if streamdetails:
304                    break
305                # sort by quality and check item's availability
306                for prov_media in sorted(
307                    media_item.provider_mappings, key=lambda x: x.quality or 0, reverse=True
308                ):
309                    if not prov_media.available:
310                        self.logger.debug(f"Skipping unavailable {prov_media}")
311                        continue
312                    if (
313                        not allow_other_provider
314                        and prov_media.provider_instance not in preferred_providers
315                    ):
316                        continue
317                    # guard that provider is available
318                    provider = mass.get_provider(prov_media.provider_instance)
319                    if not provider:
320                        self.logger.debug(f"Skipping {prov_media} - provider not available")
321                        continue  # provider not available ?
322                    # get streamdetails from provider; music and plugin providers
323                    # share this signature, so either type can own the item.
324                    try:
325                        BYPASS_THROTTLER.set(True)
326                        stream_prov = cast("MusicProvider | PluginProvider", provider)
327                        streamdetails = await stream_prov.get_stream_details(
328                            prov_media.item_id, media_item.media_type
329                        )
330                    except AudioError as err:
331                        last_audio_error = err
332                        self.logger.warning(str(err))
333                    except MusicAssistantError as err:
334                        self.logger.warning(str(err))
335                    else:
336                        break
337                    finally:
338                        BYPASS_THROTTLER.set(False)
339
340            if not streamdetails:
341                if last_audio_error is not None:
342                    raise last_audio_error
343                msg = f"Unable to retrieve streamdetails for {queue_item.name} ({queue_item.uri})"
344                raise MediaNotFoundError(msg)
345
346            # work out how to handle radio stream
347            if (
348                streamdetails.stream_type in (StreamType.ICY, StreamType.HLS, StreamType.HTTP)
349                and streamdetails.media_type == MediaType.RADIO
350                and isinstance(streamdetails.path, str)
351            ):
352                resolved_url, stream_type = await self.resolve_radio_stream(streamdetails.path)
353                streamdetails.path = resolved_url
354                streamdetails.stream_type = stream_type
355                # Set up metadata monitoring callback for HLS radio streams, if not already set
356                if (
357                    stream_type == StreamType.HLS
358                    and not streamdetails.stream_metadata_update_callback
359                ):
360                    streamdetails.stream_metadata_update_callback = partial(
361                        self._update_hls_radio_metadata
362                    )
363                    streamdetails.stream_metadata_update_interval = 5
364
365        # providers report an unknown duration as either None or 0
366        if not streamdetails.duration:
367            if queue_item.media_item and queue_item.media_item.duration:
368                streamdetails.duration = queue_item.media_item.duration
369            elif queue_item.duration:
370                streamdetails.duration = queue_item.duration
371        if seek_position and not streamdetails.allow_seek:
372            self.logger.warning("seeking is not possible on this stream!")
373            seek_position = 0
374        elif seek_position and not streamdetails.duration:
375            self.logger.warning("seeking is not possible on duration-less streams!")
376            seek_position = 0
377
378        # set queue_id on the streamdetails so we know what is being streamed
379        streamdetails.queue_id = queue_item.queue_id
380        # handle skip/fade_in details
381        streamdetails.seek_position = seek_position
382        streamdetails.fade_in = fade_in
383
384        streamdetails.prefer_album_loudness = prefer_album_loudness
385        conf_volume_normalization_target = float(
386            mass.streams.get_config_value(CONF_VOLUME_NORMALIZATION_TARGET, return_type=int)
387        )
388        # guard against invalid volume normalization values
389        # range and default_value are guaranteed to be set for this constant
390        volume_range = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.range
391        assert volume_range is not None
392        if (
393            conf_volume_normalization_target < volume_range[0]
394            or conf_volume_normalization_target >= volume_range[1]
395        ):
396            default_val = CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value
397            assert isinstance(default_val, (int, float))
398            conf_volume_normalization_target = float(default_val)
399            self.logger.warning(
400                "Invalid volume normalization target configured, resetting to default of %s LUFS",
401                CONF_ENTRY_VOLUME_NORMALIZATION_TARGET.default_value,
402            )
403        streamdetails.target_loudness = conf_volume_normalization_target
404        volume_normalization_enabled = (
405            mass.config.get_effective_player_queue_config_value(
406                streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
407            )
408            != CONF_VALUE_DISABLED
409        )
410        streamdetails.volume_normalization_mode = get_normalization_mode(
411            self._get_volume_normalization_preference(streamdetails),
412            volume_normalization_enabled,
413            streamdetails,
414        )
415
416        self.logger.debug(
417            "Retrieved streamdetails for %s in %s milliseconds",
418            queue_item.uri,
419            int((time.time() - time_start) * 1000),
420        )
421        return streamdetails
422
423    async def get_media_stream(
424        self,
425        streamdetails: StreamDetails,
426        pcm_format: AudioFormat,
427        seek_position: int = 0,
428        filter_params: list[str] | None = None,
429        chunk_seconds: float = 1.0,
430    ) -> AsyncGenerator[bytes]:
431        """
432        Get audio stream for given media details as raw PCM.
433
434        :param streamdetails: Details of the stream to fetch.
435        :param pcm_format: Target PCM format the consumer expects.
436        :param seek_position: Seek offset in seconds (only honoured when the
437            source allows seeking; ignored for live AudioSources).
438        :param filter_params: Optional ffmpeg filter expressions.
439        :param chunk_seconds: Size of each yielded chunk in seconds of audio.
440            Defaults to 1 s for track-like sources; callers streaming live
441            AudioSources should pass a much smaller value (e.g. 0.02) to keep
442            end-to-end latency low.
443        """
444        mass = self.mass
445        logger = self.logger.getChild("media_stream")
446        logger.log(VERBOSE_LOG_LEVEL, "Starting media stream for %s", streamdetails.uri)
447        # copy: the args below are appended per call, while the StreamDetails is cached on
448        # the queue item and reused across calls (retry, seek, background analysis)
449        extra_input_args = list(streamdetails.extra_input_args or [])
450
451        # work out audio source for these streamdetails
452        audio_source: str | AsyncGenerator[bytes]
453        stream_type = streamdetails.stream_type
454        if stream_type == StreamType.CUSTOM:
455            # MusicProvider and PluginProvider both expose get_audio_stream with the same shape
456            provider = mass.get_provider(streamdetails.provider)
457            if provider is None:
458                raise ProviderUnavailableError(
459                    f"Provider {streamdetails.provider} for stream is no longer available"
460                )
461            provider = cast("MusicProvider | PluginProvider", provider)
462            audio_source = provider.get_audio_stream(
463                streamdetails, seek_position=seek_position if streamdetails.can_seek else 0
464            )
465            seek_position = 0 if streamdetails.can_seek else seek_position
466        elif stream_type == StreamType.ICY:
467            assert streamdetails.path is not None
468            assert isinstance(streamdetails.path, (str, list))
469            audio_source = self.get_reconnecting_icy_radio_stream(streamdetails.path, streamdetails)
470            seek_position = 0
471        elif stream_type == StreamType.SHOUTCAST:
472            assert isinstance(streamdetails.path, str)
473            audio_source = self.get_shoutcast_stream(streamdetails.path, streamdetails)
474            seek_position = 0
475        elif stream_type == StreamType.IN_BAND:
476            assert isinstance(streamdetails.path, str)  # for type checking
477
478            # For IN_BAND (OGG/Opus) radio streams, use chained OGG handler.
479            # This handles the chained OGG format by stitching logical bitstreams together
480            # so FFmpeg sees a single continuous stream. Metadata is extracted in-band.
481            def _on_inband_metadata(metadata: dict[str, str]) -> None:
482                """Handle metadata extracted from the OGG stream."""
483                title = metadata.get("title", "")
484                artist = metadata.get("artist", "")
485                album = metadata.get("album", "")
486                if not artist and " - " in title:
487                    artist, title = title.split(" - ", 1)
488                if title or artist:
489                    if artist and title:
490                        stream_title = f"{artist} - {title}"
491                    elif title:
492                        stream_title = title
493                    else:
494                        stream_title = artist
495                    cleaned_title = clean_stream_title(stream_title)
496                    if cleaned_title and cleaned_title != streamdetails.stream_title:
497                        self.logger.log(VERBOSE_LOG_LEVEL, "In-band metadata: %s", cleaned_title)
498                        streamdetails.stream_title = cleaned_title
499                        self._update_radio_stream_metadata(
500                            streamdetails,
501                            artist=artist or None,
502                            title=title or cleaned_title,
503                            album=album or None,
504                        )
505
506            audio_source = get_chained_ogg_stream(
507                mass, streamdetails.path, metadata_callback=_on_inband_metadata
508            )
509            seek_position = 0  # seeking not possible on radio streams
510        elif stream_type == StreamType.HLS:
511            assert isinstance(streamdetails.path, str)  # for type checking
512            substream = await self.get_hls_substream(streamdetails.path)
513            audio_source = substream.path
514            if streamdetails.media_type == MediaType.RADIO:
515                # HLS streams (especially the BBC) struggle when they're played directly
516                # with ffmpeg, where they just stop after some minutes,
517                # so we tell ffmpeg to loop around in this case.
518                extra_input_args += ["-stream_loop", "-1", "-re"]
519        else:
520            # all other stream types (HTTP, FILE, etc)
521            if stream_type == StreamType.ENCRYPTED_HTTP:
522                assert streamdetails.decryption_key is not None  # for type checking
523                extra_input_args += ["-decryption_key", streamdetails.decryption_key]
524            if isinstance(streamdetails.path, list):
525                # multi part stream
526                audio_source = self.get_multi_file_stream(streamdetails, seek_position)
527                seek_position = 0  # handled by get_multi_file_stream
528            else:
529                # regular single file/url stream
530                assert isinstance(streamdetails.path, str)  # for type checking
531                audio_source = streamdetails.path
532
533        # pace ffmpeg at native rate for live sources; the producer (e.g.
534        # librespot's pipe backend) may otherwise write faster than realtime
535        if streamdetails.media_type == MediaType.AUDIO_SOURCE and "-re" not in extra_input_args:
536            extra_input_args += ["-re"]
537
538        # handle seek support
539        if seek_position and streamdetails.duration and streamdetails.allow_seek:
540            extra_input_args += ["-ss", str(int(seek_position))]
541
542        bytes_sent = 0
543        finished = False
544        cancelled = False
545        first_chunk_received = False
546        ffmpeg_loglevel = "debug" if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL) else "info"
547        # When a provider hands us already-decoded audio (e.g. Spotify Connect /
548        # AirPlay receivers piping PCM after their own decode), audio_format is
549        # the original source format meant for display while decoded_audio_format
550        # is what ffmpeg actually needs to read off the wire.
551        ffmpeg_input_format = streamdetails.decoded_audio_format or streamdetails.audio_format
552        ffmpeg_proc = FFMpeg(
553            audio_input=audio_source,
554            input_format=ffmpeg_input_format,
555            output_format=pcm_format,
556            filter_params=filter_params,
557            extra_input_args=extra_input_args,
558            collect_log_history=True,
559            loglevel=ffmpeg_loglevel,
560        )
561
562        try:
563            await ffmpeg_proc.start()
564            assert ffmpeg_proc.proc is not None  # for type checking
565            if logger.isEnabledFor(VERBOSE_LOG_LEVEL):
566                logger.log(
567                    VERBOSE_LOG_LEVEL,
568                    "Started media stream for %s - using streamtype: %s "
569                    "- pcm format: %s - ffmpeg PID: %s",
570                    streamdetails.uri,
571                    streamdetails.stream_type,
572                    pcm_format.content_type.value,
573                    ffmpeg_proc.proc.pid,
574                )
575            else:
576                logger.debug(
577                    "Started media stream for %s - using streamtype: %s",
578                    streamdetails.uri,
579                    streamdetails.stream_type,
580                )
581            stream_start = mass.loop.time()
582            chunk_size = calculate_content_length(pcm_format, chunk_seconds)
583            chunk_iter = ffmpeg_proc.iter_chunked(chunk_size)
584            while True:
585                # Time the read, not the yield: catches a stalled source, ignores backpressure.
586                read_timeout = (
587                    STREAM_START_TIMEOUT if not first_chunk_received else STREAM_STALL_TIMEOUT
588                )
589                try:
590                    async with asyncio.timeout(read_timeout):
591                        chunk = await anext(chunk_iter)
592                except StopAsyncIteration:
593                    break
594                except TimeoutError as err:
595                    raise AudioError(f"Source stalled: no audio for {read_timeout}s") from err
596                if not first_chunk_received:
597                    # At this point ffmpeg has started and should now know the codec used
598                    # for encoding the audio.
599                    # Note: ffmpeg_proc.input_format is the same object as
600                    # ffmpeg_input_format, so sample_rate / bit_depth / bit_rate
601                    # parsed from the ffmpeg log already live on streamdetails too.
602                    first_chunk_received = True
603                    # Skip the codec_type writeback when the provider declared a
604                    # decoded format: audio_format already holds the authoritative
605                    # source codec and the probed value would just be the
606                    # post-decode wire format (e.g. PCM for Spotify Connect).
607                    if streamdetails.decoded_audio_format is None:
608                        streamdetails.audio_format.codec_type = ffmpeg_proc.input_format.codec_type
609                    # Some providers omit (or report 0 for) the item duration; ffmpeg can
610                    # usually probe it from the source. Only apply when missing so we
611                    # don't clobber an accurate provider value with a rounded one.
612                    if ffmpeg_proc.parsed_duration is not None and not streamdetails.duration:
613                        streamdetails.duration = ffmpeg_proc.parsed_duration
614                    logger.debug(
615                        "First chunk received after %.2f seconds (codec detected: %s)",
616                        mass.loop.time() - stream_start,
617                        ffmpeg_proc.input_format.codec_type,
618                    )
619                yield chunk
620                bytes_sent += len(chunk)
621
622            # end of audio/track reached
623            logger.debug("End of media stream reached for %s", streamdetails.uri)
624            # wait until stderr also completed reading
625            await ffmpeg_proc.wait_with_timeout(5)
626            logger.log(
627                VERBOSE_LOG_LEVEL,
628                "FFmpeg process ended with return code %s for %s",
629                ffmpeg_proc.returncode,
630                streamdetails.uri,
631            )
632            if ffmpeg_proc.returncode not in (0, None):
633                log_trail = "\n".join(list(ffmpeg_proc.log_history)[-5:])
634                raise AudioError(f"FFMpeg exited with code {ffmpeg_proc.returncode}: {log_trail}")
635            if bytes_sent == 0:
636                # edge case: no audio data was received at all
637                raise AudioError("No audio was received")
638            finished = True
639        except (Exception, GeneratorExit, asyncio.CancelledError) as err:
640            if isinstance(err, asyncio.CancelledError | GeneratorExit):
641                # we were cancelled, just raise
642                cancelled = True
643                raise
644            # dump the last 10 lines of the log in case of an unclean exit
645            logger.warning("\n".join(list(ffmpeg_proc.log_history)[-10:]))
646            raise AudioError(f"Error while streaming: {err}") from err
647        finally:
648            # always ensure close is called which also handles all cleanup
649            await ffmpeg_proc.close()
650            # determine how many seconds we've received
651            # for pcm output we can calculate this easily
652            seconds_received = bytes_sent / pcm_format.pcm_sample_size if bytes_sent else 0
653            # store accurate duration
654            if finished and not seek_position and seconds_received:
655                streamdetails.duration = int(seconds_received)
656
657            logger.log(
658                VERBOSE_LOG_LEVEL,
659                "stream %s (with code %s) for %s",
660                "cancelled" if cancelled else "finished" if finished else "aborted",
661                ffmpeg_proc.returncode,
662                streamdetails.uri,
663            )
664
665    async def resolve_radio_stream(self, url: str) -> tuple[str, StreamType]:
666        """
667        Resolve a streaming radio URL.
668
669        Unwraps playlists and determines stream type (ICY, HLS, SHOUTCAST, IN_BAND, HTTP).
670
671        :param url: Radio stream URL to resolve
672        """
673        mass = self.mass
674        if cache := await mass.cache.get(
675            key=url, provider=CACHE_PROVIDER, category=CACHE_CATEGORY_RESOLVED_RADIO_URL
676        ):
677            if TYPE_CHECKING:
678                cache = cast("tuple[str, str]", cache)
679            return (cache[0], StreamType(cache[1]))
680
681        stream_type = StreamType.HTTP
682        timeout = ClientTimeout(total=None, connect=10, sock_read=5)
683
684        try:
685            async with self._connect_radio_stream(
686                url, headers=HTTP_HEADERS_ICY, allow_redirects=True, timeout=timeout
687            ) as resp:
688                headers = resp.headers
689                resp.raise_for_status()
690                if not resp.headers:
691                    raise InvalidDataError("no headers found")
692
693            if headers.get("icy-metaint") is not None:
694                stream_type = StreamType.ICY
695            elif headers.get("content-type", "") in ("application/ogg", "audio/ogg"):
696                # Ogg streams (Opus/Vorbis) have in-band metadata via Vorbis comments
697                stream_type = StreamType.IN_BAND
698
699            if (
700                url.endswith((".m3u", ".m3u8", ".pls"))
701                or ".m3u?" in url
702                or ".m3u8?" in url
703                or ".pls?" in url
704                or "audio/x-mpegurl" in headers.get("content-type", "")
705                or "audio/x-scpls" in headers.get("content-type", "")
706            ):
707                try:
708                    substreams = await fetch_playlist(mass, url)
709                    if not any(x for x in substreams if x.length):
710                        for line in substreams:
711                            if not line.is_url:
712                                continue
713                            return await self.resolve_radio_stream(line.path)
714                        raise InvalidDataError("No content found in playlist")
715                except IsHLSPlaylist:
716                    stream_type = StreamType.HLS
717
718        except TimeoutError as err:
719            self.logger.warning("Timeout while parsing radio URL %s", url)
720            raise InvalidDataError(f"Timeout connecting to {url}") from err
721
722        except aiohttp.ClientResponseError as err:
723            if err.status == 404:
724                raise MediaNotFoundError(f"Radio stream not found: {url}") from err
725            if err.status == 403:
726                raise InvalidDataError(f"Access denied to radio stream: {url}") from err
727            if err.status >= 500:
728                raise InvalidDataError(
729                    f"Radio stream server error (HTTP {err.status}): {url}"
730                ) from err
731            if err.status == 400:
732                # 400 errors might be from legacy Shoutcast servers
733                return await self._handle_client_error_for_radio_stream(url, err, stream_type)
734            raise InvalidDataError(f"HTTP error {err.status} from {url}") from err
735
736        except aiohttp.ClientError as err:
737            return await self._handle_client_error_for_radio_stream(url, err, stream_type)
738
739        return await self._cache_radio_result(url, stream_type)
740
741    async def get_icy_radio_stream(
742        self, url: str, streamdetails: StreamDetails
743    ) -> AsyncGenerator[bytes]:
744        """
745        Stream radio audio with ICY metadata support, reconnecting on disconnect.
746
747        Requires icy-metaint header support. Stream type should be validated
748        by resolve_radio_stream() before calling this function.
749
750        :param url: Radio stream URL
751        :param streamdetails: StreamDetails to update with metadata
752        """
753        self.logger.debug("Start streaming radio with ICY metadata from url %s", url)
754        timeout = ClientTimeout(total=0, connect=30, sock_read=5 * 60)
755        # Budget for *consecutive* reconnects that delivered no audio. A connection
756        # that actually streamed data resets it, so a healthy long-running stream can
757        # reconnect indefinitely while a dead/looping one bails out instead of spinning.
758        failed_reconnects = 0
759        max_failed_reconnects = 25
760
761        while True:
762            streamed_data = False
763            try:
764                async with self._connect_radio_stream(
765                    url, allow_redirects=True, headers=HTTP_HEADERS_ICY, timeout=timeout
766                ) as resp:
767                    # surface a non-200 (e.g. on reconnect) as a ClientResponseError so the
768                    # terminal/HTTP handling below applies instead of failing on the header
769                    resp.raise_for_status()
770                    meta_int_str = resp.headers.get("icy-metaint")
771                    if not meta_int_str:
772                        raise InvalidDataError(f"No icy-metaint header for radio stream: {url}")
773                    try:
774                        meta_int = int(meta_int_str)
775                    except ValueError as err:
776                        raise InvalidDataError(
777                            f"Invalid icy-metaint value for radio stream: {url}"
778                        ) from err
779                    if meta_int <= 0:
780                        raise InvalidDataError(f"Invalid icy-metaint value for radio stream: {url}")
781                    # readexactly raises IncompleteReadError when the server closes the
782                    # connection mid-frame; that (and the network errors below) drops us
783                    # out to the reconnect handler so a live stream survives the blip.
784                    while True:
785                        chunk = await resp.content.readexactly(meta_int)
786                        streamed_data = True
787                        yield chunk
788                        meta_byte = await resp.content.readexactly(1)
789                        if meta_byte == b"\x00":
790                            continue
791                        meta_length = ord(meta_byte) * 16
792                        meta_data = await resp.content.readexactly(meta_length)
793                        self._parse_icy_metadata(meta_data, streamdetails)
794            except asyncio.CancelledError:
795                self.logger.debug("ICY radio stream cancelled for %s", url)
796                raise
797            except aiohttp.ClientResponseError as err:
798                if err.status == 404:
799                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
800                if err.status == 403:
801                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
802                raise ProviderUnavailableError(
803                    f"Radio stream returned HTTP {err.status}: {err}"
804                ) from err
805            except (
806                asyncio.IncompleteReadError,
807                aiohttp.ClientConnectionError,
808                aiohttp.ClientPayloadError,
809                aiohttp.ServerDisconnectedError,
810            ) as err:
811                if streamed_data:
812                    # a healthy session that dropped - reconnect without spending budget
813                    failed_reconnects = 0
814                    self.logger.debug("ICY radio stream dropped, reconnecting: %s", err)
815                else:
816                    failed_reconnects += 1
817                    if failed_reconnects > max_failed_reconnects:
818                        raise RetriesExhausted(
819                            f"ICY radio stream failed after {max_failed_reconnects} "
820                            f"reconnects without data: {err}"
821                        ) from err
822                    self.logger.warning(
823                        "ICY radio stream reconnect produced no data (%d/%d): %s",
824                        failed_reconnects,
825                        max_failed_reconnects,
826                        err,
827                    )
828                await asyncio.sleep(0.5)
829
830    async def get_reconnecting_icy_radio_stream(
831        self, url: str | list[MultiPartPath], streamdetails: StreamDetails
832    ) -> AsyncGenerator[bytes]:
833        """
834        Yield ICY radio audio with metadata, failing over across mirror URLs.
835
836        A single URL is delegated to :meth:`get_icy_radio_stream`, which already reconnects
837        on disconnect. Multiple URLs are treated as interchangeable mirrors and tried in turn;
838        a mirror that delivers audio resets the failover budget, so a healthy mirror keeps
839        streaming while a set of unreachable mirrors raises the last error instead of spinning.
840
841        :param url: One stream URL, or a list of mirror URLs to fail over between.
842        :param streamdetails: StreamDetails to update with metadata.
843        """
844        urls = self._normalize_reconnecting_urls(url)
845        if len(urls) == 1:
846            async for chunk in self.get_icy_radio_stream(urls[0], streamdetails):
847                yield chunk
848            return
849
850        url_index = 0
851        failed_rotations = 0
852        max_failed_rotations = len(urls) * 2
853        last_err: MusicAssistantError | None = None
854        while failed_rotations <= max_failed_rotations:
855            current_url = urls[url_index % len(urls)]
856            url_index += 1
857            delivered_audio = False
858            try:
859                async for chunk in self.get_icy_radio_stream(current_url, streamdetails):
860                    delivered_audio = True
861                    failed_rotations = 0
862                    # release the previous failure while healthy: it pins the full
863                    # exception traceback (with frames) for the lifetime of the stream
864                    last_err = None
865                    yield chunk
866                return
867            except RADIO_MIRROR_FAILOVER_ERRORS as err:
868                last_err = err
869                if not delivered_audio:
870                    failed_rotations += 1
871                self.logger.warning(
872                    "ICY radio mirror %s failed, trying next url (%d/%d): %s",
873                    current_url,
874                    failed_rotations,
875                    max_failed_rotations,
876                    err,
877                )
878        if last_err is not None:
879            raise last_err
880
881    async def get_reconnecting_radio_stream(self, url: str) -> AsyncGenerator[bytes]:
882        """
883        Yield continuous radio stream data, automatically reconnecting on disconnect.
884
885        :param url: URL of the radio stream.
886        """
887        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
888        reconnect_count = 0
889        max_reconnects = 1000  # Allow many reconnects for long-running radio
890
891        while reconnect_count <= max_reconnects:
892            try:
893                async with self._connect_radio_stream(
894                    url, allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
895                ) as resp:
896                    chunk_count = 0
897                    async for chunk in resp.content.iter_any():
898                        chunk_count += 1
899                        yield chunk
900
901                    # Connection closed normally - reconnect
902                    self.logger.debug(
903                        "Radio stream connection closed after %d chunks, reconnecting... "
904                        "(reconnect #%d)",
905                        chunk_count,
906                        reconnect_count,
907                    )
908                    reconnect_count += 1
909                    await asyncio.sleep(0.1)  # Brief delay before reconnect
910
911            except asyncio.CancelledError:
912                self.logger.debug("Radio stream cancelled for %s", url)
913                raise
914            except (
915                aiohttp.ClientConnectionError,
916                aiohttp.ClientPayloadError,
917                aiohttp.ServerDisconnectedError,
918            ) as err:
919                # Transient network errors - retry
920                self.logger.warning("Radio stream error (reconnect #%d): %s", reconnect_count, err)
921                reconnect_count += 1
922                if reconnect_count > max_reconnects:
923                    raise RetriesExhausted(
924                        f"Radio stream failed after {max_reconnects} reconnects: {err}"
925                    ) from err
926                await asyncio.sleep(0.5)
927            except aiohttp.ClientResponseError as err:
928                if err.status == 404:
929                    raise MediaNotFoundError(f"Radio stream not found: {url}") from err
930                if err.status == 403:
931                    raise ProviderPermissionDenied(f"Radio stream access denied: {url}") from err
932                # Other HTTP errors (5xx etc) - could be temporary
933                raise ProviderUnavailableError(
934                    f"Radio stream returned HTTP {err.status}: {err}"
935                ) from err
936
937        self.logger.warning("Radio stream reached max reconnects (%d) for %s", max_reconnects, url)
938
939    async def get_hls_substream(self, url: str) -> PlaylistItem:
940        """Select the (highest quality) HLS substream for given HLS playlist/URL."""
941        mass = self.mass
942        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
943        # fetch master playlist and select (best) child playlist
944        # https://datatracker.ietf.org/doc/html/draft-pantos-http-live-streaming-19#section-10
945        async with mass.http_session_no_ssl.get(
946            encoded_request_url(url), allow_redirects=True, headers=HTTP_HEADERS, timeout=timeout
947        ) as resp:
948            resp.raise_for_status()
949            raw_data = await resp.read()
950            encoding = resp.charset or await detect_charset(raw_data)
951            master_m3u_data = raw_data.decode(encoding)
952        substreams = parse_m3u(master_m3u_data)
953        # There is a chance that we did not get a master playlist with subplaylists
954        # but just a single master/sub playlist with the actual audio stream(s)
955        # so we need to detect if the playlist child's contain audio streams or
956        # sub-playlists.
957        if any(
958            x
959            for x in substreams
960            if (x.length or x.path.endswith((".mp4", ".aac")))
961            and not x.path.endswith((".m3u", ".m3u8"))
962        ):
963            return PlaylistItem(path=url, key=substreams[0].key)
964        # sort substreams on best quality (highest bandwidth) when available
965        if any(x for x in substreams if x.stream_info):
966            substreams.sort(
967                key=lambda x: int(
968                    x.stream_info.get("BANDWIDTH", "0") if x.stream_info is not None else 0
969                ),
970                reverse=True,
971            )
972        substream = substreams[0]
973        if not substream.path.startswith("http"):
974            # path is relative, stitch it together
975            base_path = url.rsplit("/", 1)[0]
976            substream.path = base_path + "/" + substream.path
977        return substream
978
979    async def get_http_stream(
980        self,
981        url: str,
982        streamdetails: StreamDetails,
983        seek_position: int = 0,
984        verify_ssl: bool = True,
985    ) -> AsyncGenerator[bytes]:
986        """Get audio stream from HTTP."""
987        mass = self.mass
988        self.logger.debug(
989            "Start HTTP stream for %s (seek_position %s)", streamdetails.uri, seek_position
990        )
991        if seek_position:
992            assert streamdetails.duration, "Duration required for seek requests"
993        http_session = mass.http_session if verify_ssl else mass.http_session_no_ssl
994        # try to get filesize with a head request
995        seek_supported = streamdetails.can_seek
996        if seek_position or not streamdetails.size:
997            async with http_session.head(
998                encoded_request_url(url), allow_redirects=True, headers=HTTP_HEADERS
999            ) as resp:
1000                resp.raise_for_status()
1001                if size := resp.headers.get("Content-Length"):
1002                    streamdetails.size = int(size)
1003                seek_supported = resp.headers.get("Accept-Ranges") == "bytes"
1004        # headers
1005        headers = {**HTTP_HEADERS}
1006        timeout = ClientTimeout(total=None, connect=30, sock_read=5 * 60)
1007        skip_bytes = 0
1008        if seek_position and streamdetails.size:
1009            assert streamdetails.duration is not None  # for type checking
1010            skip_bytes = int(streamdetails.size / streamdetails.duration * seek_position)
1011            headers["Range"] = f"bytes={skip_bytes}-{streamdetails.size}"
1012
1013        # seeking an unknown or container format is not supported due to the (moov) headers
1014        if seek_position and (
1015            not seek_supported
1016            or streamdetails.audio_format.content_type
1017            in (ContentType.UNKNOWN, ContentType.M4A, ContentType.M4B)
1018        ):
1019            self.logger.warning(
1020                "Seeking in %s (%s) not possible.",
1021                streamdetails.uri,
1022                streamdetails.audio_format.output_format_str,
1023            )
1024            seek_position = 0
1025            streamdetails.seek_position = 0
1026
1027        # start the streaming from http
1028        bytes_received = 0
1029        async with http_session.get(
1030            encoded_request_url(url), allow_redirects=True, headers=headers, timeout=timeout
1031        ) as resp:
1032            is_partial = resp.status == 206
1033            if seek_position and not is_partial:
1034                raise InvalidDataError("HTTP source does not support seeking!")
1035            resp.raise_for_status()
1036            async for chunk in resp.content.iter_any():
1037                bytes_received += len(chunk)
1038                yield chunk
1039
1040        # store size on streamdetails for later use
1041        if not streamdetails.size:
1042            streamdetails.size = bytes_received
1043        self.logger.debug(
1044            "Finished HTTP stream for %s (transferred %s/%s bytes)",
1045            streamdetails.uri,
1046            bytes_received,
1047            streamdetails.size,
1048        )
1049
1050    async def get_file_stream(
1051        self,
1052        filename: str,
1053        streamdetails: StreamDetails,
1054        seek_position: int = 0,
1055    ) -> AsyncGenerator[bytes]:
1056        """Get audio stream from local accessible file."""
1057        if seek_position:
1058            assert streamdetails.duration, "Duration required for seek requests"
1059        if not streamdetails.size:
1060            stat = await asyncio.to_thread(os.stat, filename)
1061            streamdetails.size = stat.st_size
1062
1063        # seeking an unknown or container format is not supported due to the (moov) headers
1064        if seek_position and (
1065            streamdetails.audio_format.content_type
1066            in (ContentType.UNKNOWN, ContentType.M4A, ContentType.M4B, ContentType.MP4)
1067        ):
1068            self.logger.warning(
1069                "Seeking in %s (%s) not possible.",
1070                streamdetails.uri,
1071                streamdetails.audio_format.output_format_str,
1072            )
1073            seek_position = 0
1074            streamdetails.seek_position = 0
1075
1076        chunk_size = calculate_content_length(streamdetails.audio_format)
1077        async with aiofiles.open(streamdetails.data, "rb") as _file:
1078            if seek_position:
1079                assert streamdetails.duration is not None  # for type checking
1080                seek_pos = int((streamdetails.size / streamdetails.duration) * seek_position)
1081                await _file.seek(seek_pos)
1082            # yield chunks of data from file
1083            while True:
1084                data = await _file.read(chunk_size)
1085                if not data:
1086                    break
1087                yield data
1088
1089    async def get_multi_file_stream(
1090        self,
1091        streamdetails: StreamDetails,
1092        seek_position: int = 0,
1093    ) -> AsyncGenerator[bytes]:
1094        """
1095        Return audio stream for a concatenation of multiple files.
1096
1097        Arguments:
1098        seek_position: The position to seek to in seconds
1099        """
1100        if not isinstance(streamdetails.path, list):
1101            raise InvalidDataError("Multi-file streamdetails requires a list of MultiPartPath")
1102        parts, seek_position = get_parts_from_position(streamdetails.path, seek_position)
1103        files_list = [part.path for part in parts]
1104
1105        # concat input files
1106        temp_file = f"/tmp/{shortuuid.random(20)}.txt"  # noqa: S108
1107        async with aiofiles.open(temp_file, "w") as f:
1108            await f.write(build_concat_filelist(files_list))
1109
1110        try:
1111            async for chunk in get_ffmpeg_stream(
1112                audio_input=temp_file,
1113                input_format=streamdetails.audio_format,
1114                output_format=AudioFormat(
1115                    content_type=ContentType.NUT,
1116                    sample_rate=streamdetails.audio_format.sample_rate,
1117                    bit_depth=streamdetails.audio_format.bit_depth,
1118                    channels=streamdetails.audio_format.channels,
1119                ),
1120                extra_input_args=[
1121                    "-safe",
1122                    "0",
1123                    "-f",
1124                    "concat",
1125                    "-i",
1126                    temp_file,
1127                    "-ss",
1128                    str(seek_position),
1129                ],
1130            ):
1131                yield chunk
1132        finally:
1133            await remove_file(temp_file)
1134
1135    def get_player_output_plan(
1136        self,
1137        player_id: str,
1138        input_format: AudioFormat,
1139        output_format: AudioFormat,
1140        *,
1141        shared_player_ids: Iterable[str] | None = None,
1142        handoff_format: AudioFormat | None = None,
1143        queue_id: str | None = None,
1144        session_id: str | None = None,
1145        queue_item_id: str | None = None,
1146    ) -> AudioOutputPlan:
1147        """
1148        Return executable filters and matching output details for a player.
1149
1150        :param player_id: Destination player identifier.
1151        :param input_format: PCM format entering player-specific processing.
1152        :param output_format: Furthest downstream output format known to the server.
1153        :param shared_player_ids: Additional players receiving this identical output path.
1154            An empty iterable marks a path that can gain shared destinations later.
1155        :param handoff_format: Earlier provider handoff format when it differs.
1156        :param queue_id: Explicit queue identifier for the processing snapshot.
1157        :param session_id: Explicit queue session identifier for the processing snapshot.
1158        :param queue_item_id: Queue item for a single-item output path.
1159        """
1160        filter_params: list[str | ComplexFilter] = []
1161        player = self.mass.players.get_player(player_id)
1162        destination_player_id = (
1163            player.protocol_parent_id if player and player.protocol_parent_id else player_id
1164        )
1165        resolved_shared_player_ids = (
1166            resolve_output_player_ids(self.mass, shared_player_ids) - {destination_player_id}
1167            if shared_player_ids is not None
1168            else None
1169        )
1170        destination_player_ids = {destination_player_id, *(resolved_shared_player_ids or ())}
1171        if player:
1172            dsp_config_id = self._resolve_player_dsp_config_id(player)
1173            dsp = self._resolve_player_dsp_config(player)
1174            configured_dsp = self.mass.config.get_player_dsp_config(dsp_config_id)
1175            if configured_dsp.enabled and not dsp.enabled and is_grouping_preventing_dsp(player):
1176                dsp_state = DSPState.DISABLED_BY_UNSUPPORTED_GROUP
1177            else:
1178                dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1179        else:
1180            dsp_config_id = player_id
1181            dsp = self.mass.config.get_player_dsp_config(player_id)
1182            dsp_state = DSPState.ENABLED if dsp.enabled else DSPState.DISABLED
1183
1184        enabled_filters = [dsp_filter for dsp_filter in dsp.filters if dsp_filter.enabled]
1185        # a neutral filter (0 dB gain, centered balance) emits no params; exclude
1186        # it so it is not reported as an active, non-bit-perfect stage
1187        effective_filters: list[DSPFilter] = []
1188        if dsp.enabled:
1189            if dsp.input_gain != 0:
1190                filter_params.append(f"volume={dsp.input_gain}dB")
1191            ir_dir = os.path.join(self.mass.storage_path, DSP_IRS_DIRNAME)
1192            known_ir_ids = {record["ir_id"] for record in self.mass.config.get_dsp_irs()}
1193            for dsp_filter in enabled_filters:
1194                if isinstance(dsp_filter, ConvolutionFilter) and dsp_filter.ir_id:
1195                    # ffmpeg fails to open the graph if the impulse response file is gone,
1196                    # which costs the player all audio, so drop the filter instead
1197                    if dsp_filter.ir_id not in known_ir_ids:
1198                        self.logger.warning(
1199                            "Skipping the convolution filter of player %s: "
1200                            "impulse response %s is not stored",
1201                            player_id,
1202                            dsp_filter.ir_id,
1203                        )
1204                        continue
1205                params = filter_to_ffmpeg_params(dsp_filter, input_format, ir_dir=ir_dir)
1206                if not params:
1207                    continue
1208                filter_params.extend(params)
1209                effective_filters.append(dsp_filter)
1210            if dsp.output_gain != 0:
1211                filter_params.append(f"volume={dsp.output_gain}dB")
1212
1213        channel_value = self._get_output_channels(player, player_id)
1214        source_channel = None
1215        channel_mix = ""
1216        # a single channel source is already the downmix and holds no FL/FR to select
1217        # from, where a pan would silently resolve every gain to zero
1218        if input_format.channels > 1:
1219            if channel_value == "left":
1220                source_channel = AudioChannel.FL
1221                channel_mix = "FL"
1222            elif channel_value == "right":
1223                source_channel = AudioChannel.FR
1224                channel_mix = "FR"
1225            elif channel_value == "mono":
1226                # both source channels feed the downmix, so report ALL to keep the
1227                # output from ever being presented as bit perfect
1228                source_channel = AudioChannel.ALL
1229                channel_mix = "0.5*FL+0.5*FR"
1230        if channel_mix:
1231            # the pan runs in the command that emits the handoff format, and it feeds
1232            # every channel of it explicitly: leaving ffmpeg to upmix from a single
1233            # channel costs 3 dB through its rematrix
1234            if (handoff_format or output_format).channels == 1:
1235                filter_params.append(f"pan=mono|c0={channel_mix}")
1236            else:
1237                filter_params.append(f"pan=stereo|c0={channel_mix}|c1={channel_mix}")
1238
1239        output_details = AudioOutputDetails(
1240            player_ids=sorted(destination_player_ids),
1241            dsp=AudioDSPDetails(
1242                state=dsp_state,
1243                input_gain=dsp.input_gain if dsp.enabled else 0.0,
1244                filters=effective_filters,
1245                output_gain=dsp.output_gain if dsp.enabled else 0.0,
1246                preset_id=dsp.preset_id,
1247            ),
1248            source_channel=source_channel,
1249            output_format=output_format,
1250        )
1251        output_plan = AudioOutputPlan(
1252            filter_params=filter_params,
1253            output_details=output_details,
1254            input_format=input_format,
1255            handoff_format=handoff_format,
1256            dsp_config_id=dsp_config_id,
1257        )
1258        if queue_id is not None and session_id is not None:
1259            self.mass.streams.audio_processing.update_output(
1260                destination_player_id,
1261                output_plan,
1262                shared_player_ids=resolved_shared_player_ids,
1263                queue_id=queue_id,
1264                session_id=session_id,
1265                queue_item_id=queue_item_id,
1266            )
1267        self.logger.log(
1268            VERBOSE_LOG_LEVEL,
1269            "Generated ffmpeg params for player %s: %s",
1270            player_id,
1271            filter_params,
1272        )
1273        return output_plan
1274
1275    async def get_output_format(
1276        self,
1277        output_format_str: str,
1278        player: Player,
1279        content_sample_rate: int,
1280        content_bit_depth: int,
1281        media_type: MediaType = MediaType.UNKNOWN,
1282    ) -> AudioFormat:
1283        """Parse (player specific) output format details for given format string."""
1284        content_type: ContentType = ContentType.try_parse(output_format_str)
1285        player_supported_rates = player.get_supported_sample_rates()
1286        supported_sample_rates = [sr for sr, _ in player_supported_rates]
1287        if content_sample_rate in supported_sample_rates:
1288            output_sample_rate = content_sample_rate
1289        else:
1290            output_sample_rate = max(supported_sample_rates)
1291        # only consider bit depths that are actually paired with the chosen sample rate
1292        bit_depths_for_rate = [
1293            bd for (sr, bd) in player_supported_rates if sr == output_sample_rate
1294        ]
1295        output_bit_depth = min(content_bit_depth, max(bit_depths_for_rate, default=16))
1296
1297        if not content_type.is_lossless():
1298            # no point in having a higher bit depth for lossy formats
1299            output_bit_depth = 16
1300            output_sample_rate = min(48000, output_sample_rate)
1301        if media_type not in (MediaType.TRACK, MediaType.AUDIO_SOURCE, MediaType.FLOW_STREAM):
1302            # no point in having a higher bit depth for non-track media types (e.g. TTS, radio)
1303            output_bit_depth = min(output_bit_depth, 16)
1304        if output_format_str == "pcm":
1305            content_type = ContentType.from_bit_depth(output_bit_depth)
1306
1307        output_channels_str = self._get_output_channels(player, player.player_id)
1308        fmt = AudioFormat(
1309            content_type=content_type,
1310            sample_rate=output_sample_rate,
1311            bit_depth=output_bit_depth,
1312            channels=1 if output_channels_str != "stereo" else 2,
1313        )
1314        fmt.bit_rate = get_bit_rate(fmt)
1315        return fmt
1316
1317    async def select_pcm_format(
1318        self,
1319        player: Player,
1320        streamdetails: StreamDetails,
1321        crossfade_enabled: bool,
1322        overlay_active: bool = False,
1323    ) -> AudioFormat:
1324        """
1325        Select the internal PCM format for streaming a single queue item.
1326
1327        Used by the per-item (non-flow) stream path. The sample rate is the highest
1328        rate the player supports that is <= the source rate, so the source is never
1329        upsampled. The bit depth follows the source unless audio processing
1330        (crossfade, volume normalization, DSP) is active — those need F32 headroom
1331        to avoid clipping/precision loss. Surround sources are folded down to stereo.
1332        Realtime AudioSource items skip all processing and get a pure passthrough
1333        format (source rate/bit depth when the player supports them).
1334
1335        :param player: The player requesting the stream.
1336        :param streamdetails: Stream details for the current item.
1337        :param crossfade_enabled: Whether crossfade is enabled for this stream.
1338        :param overlay_active: Whether an audio overlay will be mixed into this stream.
1339        """
1340        if streamdetails.media_type == MediaType.AUDIO_SOURCE:
1341            return self._select_audio_source_pcm_format(player, streamdetails)
1342        supported_sample_rates = [sr for sr, _ in player.get_supported_sample_rates()]
1343        # snap-down: pick the highest supported rate <= source. when the source rate
1344        # is below every supported rate (e.g. 22 kHz content on a 44.1k-only player),
1345        # fall back to the lowest supported rate instead of a hardcoded 48 kHz that
1346        # the player may not actually support.
1347        output_sample_rate = max(
1348            (r for r in supported_sample_rates if r <= streamdetails.audio_format.sample_rate),
1349            default=min(supported_sample_rates),
1350        )
1351        content_type, bit_depth = self._pick_pcm_bit_depth(
1352            (player,),
1353            streamdetails,
1354            crossfade_enabled,
1355            overlay_active,
1356        )
1357        pcm_format = AudioFormat(
1358            sample_rate=output_sample_rate,
1359            content_type=content_type,
1360            bit_depth=bit_depth,
1361            # fold surround sources down to stereo right at the decode step: no
1362            # output format carries more than two channels, so a wider PCM format
1363            # only makes every bytes-to-seconds sum on the stream come out short
1364            channels=min(streamdetails.audio_format.channels, 2),
1365        )
1366        if crossfade_enabled or overlay_active:
1367            pcm_format.channels = 2
1368        return pcm_format
1369
1370    async def select_flow_pcm_format(
1371        self,
1372        player: Player,
1373        start_streamdetails: StreamDetails | None = None,
1374        crossfade_enabled: bool = False,
1375        overlay_active: bool = False,
1376        fallback_sample_rate: int | None = None,
1377        output_players: Iterable[Player] | None = None,
1378    ) -> AudioFormat:
1379        """
1380        Select the internal PCM format for a Queue Flow Mode stream.
1381
1382        Used by the gapless flow path that stitches multiple queue items into one
1383        continuous PCM stream. The sample rate is driven by the player's
1384        ``CONF_FLOW_MODE_SAMPLE_RATE`` setting (smart/bit_perfect/48k/96k/highest)
1385        — for the anchored modes it follows the first track's rate, for the fixed
1386        modes it snaps to the configured rate. The bit depth follows the first
1387        track's source unless audio processing is active (then F32 for headroom),
1388        avoiding an unnecessary up-convert to 32-bit when none of the consumers
1389        will benefit from it. When the first item is a realtime AudioSource, the
1390        flow mode config is ignored and a pure passthrough format is used so the
1391        source audio is delivered with minimum overhead and latency.
1392
1393        :param player: The player the flow stream is being prepared for.
1394        :param start_streamdetails: Stream details of the first track in the flow.
1395            Required for the anchored modes ('smart' / 'bit_perfect') and for the
1396            bit-depth optimization. May be omitted for the fixed-rate modes — when
1397            omitted the bit depth defaults to F32.
1398        :param crossfade_enabled: Whether the queue will use crossfade transitions.
1399        :param overlay_active: Whether an audio overlay will be mixed into the stream.
1400        :param fallback_sample_rate: Preferred rate when the first item format is unknown.
1401        :param output_players: All players consuming the shared PCM stream. Their common
1402            sample rates and processing requirements determine the session format.
1403        """
1404        players = tuple(output_players) if output_players is not None else (player,)
1405        if not players:
1406            raise AudioError("At least one output player is required")
1407        supported_sample_rates = sorted(
1408            set.intersection(
1409                *(
1410                    {sample_rate for sample_rate, _ in item.get_supported_sample_rates()}
1411                    for item in players
1412                )
1413            )
1414        )
1415        if not supported_sample_rates:
1416            raise AudioError("Output players do not share a supported sample rate")
1417        if start_streamdetails is not None and (
1418            start_streamdetails.media_type == MediaType.AUDIO_SOURCE
1419        ):
1420            return self._select_audio_source_pcm_format(
1421                player,
1422                start_streamdetails,
1423                supported_sample_rates=supported_sample_rates,
1424            )
1425        flow_mode_conf = cast(
1426            "str",
1427            player.config.get_value(CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART),
1428        )
1429
1430        if flow_mode_conf == FLOW_MODE_SAMPLE_RATE_HIGHEST:
1431            output_sample_rate = max(supported_sample_rates)
1432        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_48000:
1433            # for the fixed-rate modes, the user picked a specific bandwidth/quality
1434            # ceiling; prefer the highest supported rate <= target
1435            output_sample_rate = _snap_supported_rate_down(48000, supported_sample_rates)
1436        elif flow_mode_conf == FLOW_MODE_SAMPLE_RATE_96000:
1437            output_sample_rate = _snap_supported_rate_down(96000, supported_sample_rates)
1438        else:
1439            # smart or bit_perfect (default): anchor the flow at the starting track's
1440            # sample rate; if the player doesn't natively support it, upsample to the
1441            # closest higher supported rate
1442            target_rate = (
1443                start_streamdetails.audio_format.sample_rate
1444                if start_streamdetails
1445                else (
1446                    fallback_sample_rate
1447                    if fallback_sample_rate is not None
1448                    else max(supported_sample_rates)
1449                )
1450            )
1451            output_sample_rate = _snap_supported_rate_up(target_rate, supported_sample_rates)
1452
1453        content_type, bit_depth = self._pick_pcm_bit_depth(
1454            players, start_streamdetails, crossfade_enabled, overlay_active
1455        )
1456        return AudioFormat(
1457            content_type=content_type,
1458            sample_rate=output_sample_rate,
1459            bit_depth=bit_depth,
1460            channels=2,
1461        )
1462
1463    async def get_audio_source_stream(
1464        self,
1465        queue_item: QueueItem,
1466        pcm_format: AudioFormat,
1467        raise_on_error: bool = True,
1468    ) -> AsyncGenerator[bytes]:
1469        """
1470        Get the realtime PCM stream for an AudioSource queue item.
1471
1472        AudioSources are live/realtime: bytes flow at the producer's pace, with
1473        no pre-buffering, no loudness hydration, no volume normalization, no
1474        crossfade/fade-in, no playback-speed shift, no next-track preload. The
1475        path stays as small as possible to keep end-to-end latency low.
1476
1477        Fast path: when the source PCM format already matches the consumer's
1478        ``pcm_format``, the provider's bytes are paced in Python and forwarded
1479        directly — no ffmpeg in the data path.
1480
1481        Slow path: when formats differ, ffmpeg resamples/recodes the stream
1482        (with ``-re`` for rate pacing) via ``get_media_stream``.
1483
1484        :param queue_item: The AudioSource queue item to stream.
1485        :param pcm_format: Output PCM format the consumer wants.
1486        :param raise_on_error: Re-raise stream errors instead of swallowing them.
1487        """
1488        streamdetails = queue_item.streamdetails
1489        assert streamdetails
1490        logger = self.logger.getChild("audio_source_stream")
1491        bytes_received = 0
1492        try:
1493            async for chunk in self._iter_audio_source_pcm(streamdetails, pcm_format):
1494                bytes_received += len(chunk)
1495                yield chunk
1496        except AudioError as err:
1497            streamdetails.stream_error = True
1498            # revoke availability when the stream never produced any audio
1499            if bytes_received == 0:
1500                queue_item.available = False
1501            if raise_on_error:
1502                raise
1503            logger.error(
1504                "AudioError while streaming AudioSource %s (%s): %s",
1505                queue_item.name,
1506                streamdetails.uri,
1507                err,
1508            )
1509        except asyncio.CancelledError:
1510            raise
1511        except Exception as err:
1512            streamdetails.stream_error = True
1513            if raise_on_error:
1514                raise
1515            logger.exception(
1516                "Unexpected error while streaming AudioSource %s (%s): %s",
1517                queue_item.name,
1518                streamdetails.uri,
1519                err,
1520            )
1521        finally:
1522            streamdetails.seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1523
1524    async def get_queue_item_stream(
1525        self,
1526        queue_item: QueueItem,
1527        pcm_format: AudioFormat,
1528        seek_position: int = 0,
1529        playback_speed: float = 1.0,
1530        raise_on_error: bool = True,
1531        normalization_override: VolumeNormalizationMode | None = None,
1532        session_id: str | None = None,
1533    ) -> AsyncGenerator[bytes]:
1534        """
1535        Get the (PCM) audio stream for a single queue item.
1536
1537        Audio is always served from the AudioBuffer which stores raw decoded PCM.
1538        Volume normalization and other filters are applied on-the-fly when reading
1539        from the buffer.
1540
1541        AudioSource items dispatch to ``get_audio_source_stream`` instead: they
1542        are realtime and bypass the buffering/normalization/filter machinery.
1543
1544        :param normalization_override: Force this volume normalization mode instead of
1545            re-evaluating it from the (possibly just-updated) loudness measurement. Used by
1546            the crossfade path to keep a track's replayed intro and its body on the same mode.
1547        :param session_id: Queue session that owns processing-detail updates.
1548        """
1549        if queue_item.media_type == MediaType.AUDIO_SOURCE:
1550            async for chunk in self.get_audio_source_stream(
1551                queue_item=queue_item,
1552                pcm_format=pcm_format,
1553                raise_on_error=raise_on_error,
1554            ):
1555                yield chunk
1556            return
1557
1558        streamdetails = queue_item.streamdetails
1559        assert streamdetails
1560        filter_params: list[str] = []
1561
1562        logger = self.logger.getChild("queue_item_stream")
1563
1564        if normalization_override is not None:
1565            # crossfade path pins the body to the intro's mode; skip hydration/re-eval that could flip it
1566            streamdetails.volume_normalization_mode = normalization_override
1567        else:
1568            # hydrate loudness from audio analysis (just-in-time, so that a measurement
1569            # completed during a previous play is picked up here). A live analyzer run
1570            # may have already populated streamdetails.loudness in memory — don't clobber
1571            # that, and don't clobber a value set upstream by the music provider.
1572            if streamdetails.loudness is None:
1573                if analysis := await self.mass.streams.audio_analysis.get_audio_analysis(
1574                    streamdetails.item_id,
1575                    streamdetails.provider,
1576                    media_type=streamdetails.media_type,
1577                    # use the authoritative EBU R128 value, not another provider's loudness proxy
1578                    priority=(LOUDNESS_ANALYSIS_DOMAIN,),
1579                ):
1580                    if analysis.loudness_integrated is not None:
1581                        streamdetails.loudness = round(analysis.loudness_integrated, 2)
1582                    if analysis.loudness_album is not None and streamdetails.loudness_album is None:
1583                        streamdetails.loudness_album = round(analysis.loudness_album, 2)
1584
1585            # re-evaluate normalization mode: the background loudness analyzer may have
1586            # updated streamdetails.loudness since get_stream_details was called
1587            if streamdetails.queue_id:
1588                volume_normalization_enabled = (
1589                    self.mass.config.get_effective_player_queue_config_value(
1590                        streamdetails.queue_id, CONF_VOLUME_NORMALIZATION, CONF_VALUE_ENABLED
1591                    )
1592                    != CONF_VALUE_DISABLED
1593                )
1594                streamdetails.volume_normalization_mode = get_normalization_mode(
1595                    self._get_volume_normalization_preference(streamdetails),
1596                    volume_normalization_enabled,
1597                    streamdetails,
1598                )
1599
1600        # handle volume normalization
1601        gain_correct: float | None = None
1602        if streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC:
1603            filter_rule = (
1604                f"loudnorm=I={streamdetails.target_loudness}"
1605                ":TP=-2.0:LRA=10.0:offset=0.0:print_format=json"
1606            )
1607            filter_params.append(filter_rule)
1608        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.FIXED_GAIN:
1609            config_key = (
1610                CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS
1611                if streamdetails.media_type == MediaType.TRACK
1612                else CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO
1613            )
1614            gain_value = self.mass.streams.get_config_value(config_key, return_type=float)
1615            gain_correct = round(gain_value, 2)
1616            filter_params.append(f"volume={gain_correct}dB")
1617        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.MEASUREMENT_ONLY:
1618            target_loudness = (
1619                float(streamdetails.target_loudness)
1620                if streamdetails.target_loudness is not None
1621                else 0.0
1622            )
1623            if streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
1624                gain_correct = target_loudness - float(streamdetails.loudness_album)
1625            elif streamdetails.loudness is not None:
1626                gain_correct = target_loudness - float(streamdetails.loudness)
1627            else:
1628                gain_correct = 0.0
1629            gain_correct = round(gain_correct, 2)
1630            filter_params.append(f"volume={gain_correct}dB")
1631        streamdetails.volume_normalization_gain_correct = gain_correct
1632
1633        # handle playback speed
1634        if playback_speed != 1.0:
1635            filter_params.append(f"atempo={playback_speed}")
1636
1637        # handle optional fade-in
1638        if streamdetails.fade_in:
1639            filter_params.insert(0, "afade=type=in:start_time=0:duration=3")
1640
1641        logger.log(
1642            VERBOSE_LOG_LEVEL,
1643            "Starting queue item stream for %s (%s)"
1644            " - using fade-in: %s"
1645            " - using volume normalization: %s"
1646            " - using playback speed: %s",
1647            queue_item.name,
1648            streamdetails.uri,
1649            streamdetails.fade_in,
1650            streamdetails.volume_normalization_mode,
1651            playback_speed,
1652        )
1653
1654        # get or create the AudioBuffer (stores raw decoded PCM)
1655        seek_position_ms = int(seek_position * 1000)
1656        audio_buffer = await AudioBuffer.get_buffer(
1657            mass=self.mass,
1658            streamdetails=streamdetails,
1659            seek_position_ms=seek_position_ms,
1660            reason="streaming",
1661        )
1662        if (
1663            streamdetails.queue_id
1664            and (queue_data := self.mass.player_queues.queue_data_or_none(streamdetails.queue_id))
1665            and (processing_session_id := session_id or queue_data.session_id)
1666        ):
1667            self.mass.streams.audio_processing.update_item_runtime(
1668                queue_id=streamdetails.queue_id,
1669                session_id=processing_session_id,
1670                queue_item_id=queue_item.queue_item_id,
1671                input_format=audio_buffer.pcm_format,
1672                pcm_format=pcm_format,
1673                normalization=get_normalization_details(streamdetails, gain_correct),
1674                playback_speed=playback_speed,
1675                alters_audio=streamdetails.fade_in,
1676            )
1677        # read from buffer with filters applied (volume normalization, speed, fade-in, etc.)
1678        # if no processing needed, this yields directly from the buffer
1679        media_stream_gen = audio_buffer.get_stream(
1680            output_format=pcm_format,
1681            seek_position_ms=seek_position_ms,
1682            filter_params=filter_params or None,
1683        )
1684
1685        first_chunk_received = False
1686        bytes_received = 0
1687        finished = False
1688        next_buffer_triggered = False
1689        stream_started_at = asyncio.get_event_loop().time()
1690        try:
1691            async for chunk in media_stream_gen:
1692                bytes_received += len(chunk)
1693                if not first_chunk_received:
1694                    first_chunk_received = True
1695                    logger.log(
1696                        VERBOSE_LOG_LEVEL,
1697                        "First audio chunk received for %s (%s) after %.2f seconds",
1698                        queue_item.name,
1699                        streamdetails.uri,
1700                        asyncio.get_event_loop().time() - stream_started_at,
1701                    )
1702                # trigger pre-buffering of the next item well before end
1703                # to ensure the raw PCM is ready when the next item needs to be streamed.
1704                # tracks and sound effects are finite files that fill and close immediately;
1705                # live sources (radio, audio_source) open an upstream connection that would
1706                # sit idle and likely time out before the player actually consumes it.
1707                if (
1708                    not next_buffer_triggered
1709                    and streamdetails.duration
1710                    and (queue := self.mass.player_queues.get_active_queue(queue_item.queue_id))
1711                    and queue.next_item
1712                    and queue.next_item.queue_item_id != queue_item.queue_item_id
1713                    and queue.next_item.media_type in (MediaType.TRACK, MediaType.SOUND_EFFECT)
1714                    and (bytes_received / pcm_format.pcm_sample_size + seek_position)
1715                    >= streamdetails.duration - 60
1716                ):
1717                    next_buffer_triggered = True
1718                    self.mass.player_queues.prepare_next_audio_buffer(queue_item.queue_id)
1719                yield chunk
1720                del chunk
1721            # if we received no audio and the buffer has a producer error,
1722            # the error was swallowed by the FFmpeg stdin feeder - re-raise it
1723            if bytes_received == 0 and audio_buffer.has_error:
1724                raise AudioError("Failed to stream audio") from audio_buffer._producer_error
1725            finished = True
1726        except AudioError as err:
1727            streamdetails.stream_error = True
1728            # revoke availability when the stream never produced any audio
1729            if bytes_received == 0:
1730                queue_item.available = False
1731            if raise_on_error:
1732                raise
1733            logger.error(
1734                "AudioError while streaming queue item %s (%s): %s",
1735                queue_item.name,
1736                streamdetails.uri,
1737                err,
1738            )
1739        except asyncio.CancelledError:
1740            raise
1741        except Exception as err:
1742            streamdetails.stream_error = True
1743            if raise_on_error:
1744                raise
1745            logger.exception(
1746                "Unexpected error while streaming queue item %s (%s): %s",
1747                queue_item.name,
1748                streamdetails.uri,
1749                err,
1750            )
1751        finally:
1752            seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1753            streamdetails.seconds_streamed = seconds_streamed
1754            logger.log(
1755                VERBOSE_LOG_LEVEL,
1756                "stream %s for %s in %.2f seconds - seconds streamed/buffered: %.2f",
1757                "aborted" if not finished else "finished",
1758                streamdetails.uri,
1759                asyncio.get_event_loop().time() - stream_started_at,
1760                seconds_streamed,
1761            )
1762            self._notify_provider_streamed(streamdetails, finished, seconds_streamed)
1763
1764    async def get_queue_item_stream_with_smartfade(
1765        self,
1766        player: Player,
1767        queue_item: QueueItem,
1768        pcm_format: AudioFormat,
1769        crossfade_mode: CrossfadeMode = CrossfadeMode.SMART_CROSSFADE,
1770        standard_crossfade_duration: int = 10,
1771        session_id: str | None = None,
1772    ) -> AsyncGenerator[bytes]:
1773        """
1774        Return one queue item with a crossfade into the next item.
1775
1776        :param player: Player consuming the stream.
1777        :param queue_item: Queue item to stream.
1778        :param pcm_format: Shared PCM format.
1779        :param crossfade_mode: Effective crossfade mode.
1780        :param standard_crossfade_duration: Configured standard crossfade duration.
1781        :param session_id: Queue session that owns processing-detail updates.
1782        """
1783        queue = self.mass.player_queues.get(queue_item.queue_id)
1784        if not queue:
1785            raise RuntimeError(f"Queue {queue_item.queue_id} not found")
1786
1787        streamdetails = queue_item.streamdetails
1788        assert streamdetails
1789        crossfade_data = self._crossfade_data.get(queue.queue_id)
1790
1791        if crossfade_data and streamdetails.seek_position > 0:
1792            # don't do crossfade when seeking into track
1793            self.logger.debug(
1794                "Discarding crossfade data for queue %s - seeking into track (pos=%s)",
1795                queue.display_name,
1796                streamdetails.seek_position,
1797            )
1798            crossfade_data = None
1799        if crossfade_data and (crossfade_data.queue_item_id != queue_item.queue_item_id):
1800            # edge case alert: the next item changed just while we were preloading/crossfading
1801            self.logger.warning(
1802                "Skipping crossfade data for queue %s - next item changed!"
1803                " (expected queue_item_id=%s, got=%s)",
1804                queue.display_name,
1805                crossfade_data.queue_item_id,
1806                queue_item.queue_item_id,
1807            )
1808            crossfade_data = None
1809            self._crossfade_data.pop(queue.queue_id, None)
1810        elif not crossfade_data:
1811            self.logger.debug(
1812                "No crossfade data available for queue %s (queue_item_id=%s)",
1813                queue.display_name,
1814                queue_item.queue_item_id,
1815            )
1816
1817        self.logger.debug(
1818            "Start Streaming queue track: %s (%s) for queue %s on player %s"
1819            "- crossfade mode: %s "
1820            "- crossfading from previous track: %s ",
1821            queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
1822            queue_item.name,
1823            queue.display_name,
1824            player.name,
1825            crossfade_mode,
1826            "true" if crossfade_data else "false",
1827        )
1828
1829        buffer = bytearray()
1830        bytes_written = 0
1831        # calculate crossfade buffer size
1832        crossfade_buffer_duration = (
1833            SMART_CROSSFADE_DURATION
1834            if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
1835            else standard_crossfade_duration
1836        )
1837        crossfade_buffer_duration = min(
1838            crossfade_buffer_duration,
1839            int(streamdetails.duration / 2)
1840            if streamdetails.duration
1841            else crossfade_buffer_duration,
1842        )
1843        # skip crossfade if buffer would be too small to be meaningful
1844        if crossfade_buffer_duration < 5:
1845            crossfade_buffer_duration = 0
1846        # Ensure crossfade buffer size is aligned to frame boundaries
1847        # Frame size = bytes_per_sample * channels
1848        bytes_per_sample = pcm_format.bit_depth // 8
1849        frame_size = bytes_per_sample * pcm_format.channels
1850        crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1851        # Round down to nearest frame boundary
1852        crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1853        fade_out_data: bytes | None = None
1854        uncredited_tail_bytes = 0
1855
1856        # pin the body to DYNAMIC when the intro was baked DYNAMIC,
1857        # else a late measurement flips it and causes a volume jump
1858        norm_override: VolumeNormalizationMode | None = None
1859        if crossfade_data and crossfade_data.normalization_mode == VolumeNormalizationMode.DYNAMIC:
1860            norm_override = VolumeNormalizationMode.DYNAMIC
1861
1862        if crossfade_data:
1863            # reported media-time (TRIM + CF) is decoupled from the raw buffer seek below (X)
1864            streamdetails.seek_position = crossfade_data.elapsed_time_offset
1865            # yield the POST portion (resample if previous track's format differs)
1866            if crossfade_data.pcm_format != pcm_format:
1867                async for _chunk in resample_pcm_audio(
1868                    crossfade_data.data, crossfade_data.pcm_format, pcm_format
1869                ):
1870                    yield _chunk
1871                    bytes_written += len(_chunk)
1872            else:
1873                for pcm_slice in iter_pcm_slices(crossfade_data.data, pcm_format, 1000):
1874                    yield pcm_slice
1875                    await asyncio.sleep(0)
1876                bytes_written += len(crossfade_data.data)
1877            # skip past the audio already consumed by the crossfade
1878            fade_in_duration_seconds = (
1879                crossfade_data.fade_in_size / crossfade_data.fade_in_pcm_format.pcm_sample_size
1880            )
1881            discard_seconds = int(fade_in_duration_seconds)
1882            fractional_seconds = fade_in_duration_seconds - discard_seconds
1883            discard_leftover = int(fractional_seconds * pcm_format.pcm_sample_size)
1884            discard_leftover = (discard_leftover // frame_size) * frame_size
1885            crossfade_data = None
1886            self._crossfade_data.pop(queue.queue_id, None)
1887        else:
1888            discard_seconds = int(streamdetails.seek_position)
1889            discard_leftover = 0
1890
1891        # Yield the first WARMUP_DURATION worth of audio immediately so playback starts
1892        # right away. After that, start accumulating the crossfade holdback buffer.
1893        warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
1894        warmup_bytes = 0
1895        total_chunks_received = 0
1896        playback_speed = cast("float", queue_item.extra_attributes.get("playback_speed", 1.0))
1897        async for chunk in self.get_queue_item_stream(
1898            queue_item,
1899            pcm_format,
1900            seek_position=discard_seconds,
1901            playback_speed=playback_speed,
1902            normalization_override=norm_override,
1903            session_id=session_id,
1904        ):
1905            total_chunks_received += 1
1906            if discard_leftover:
1907                chunk = chunk[discard_leftover:]  # noqa: PLW2901
1908                discard_leftover = 0
1909
1910            if warmup_bytes < warmup_size:
1911                # warmup: yield directly, don't buffer
1912                yield chunk
1913                warmup_bytes += len(chunk)
1914                bytes_written += len(chunk)
1915                del chunk
1916                continue
1917
1918            buffer.extend(chunk)
1919            del chunk
1920            if len(buffer) < crossfade_buffer_size:
1921                await asyncio.sleep(0)
1922                continue
1923            # yield everything above the crossfade buffer
1924            while len(buffer) > crossfade_buffer_size:
1925                yield bytes(buffer[: pcm_format.pcm_sample_size])
1926                bytes_written += pcm_format.pcm_sample_size
1927                del buffer[: pcm_format.pcm_sample_size]
1928                await asyncio.sleep(0)
1929
1930        #### HANDLE END OF TRACK
1931
1932        # get next track for crossfade
1933        crossfade_start_time = asyncio.get_event_loop().time()
1934        next_queue_item: QueueItem | None
1935        try:
1936            self.logger.debug(
1937                "Preloading NEXT track for crossfade for queue %s", queue.display_name
1938            )
1939            next_queue_item = await self.mass.player_queues.load_next_queue_item(
1940                queue.queue_id, queue_item.queue_item_id
1941            )
1942            # set index_in_buffer to prevent our next track is overwritten while preloading
1943            if next_queue_item.streamdetails is None:
1944                raise InvalidDataError(
1945                    f"No streamdetails for next queue item {next_queue_item.queue_item_id}"
1946                )
1947            queue.index_in_buffer = self.mass.player_queues.index_by_id(
1948                queue.queue_id, next_queue_item.queue_item_id
1949            )
1950        except QueueEmpty:
1951            # end of queue reached, no next item
1952            next_queue_item = None
1953
1954        crossfade_allowed = False
1955        if next_queue_item and next_queue_item.streamdetails:
1956            next_pcm = await self.select_pcm_format(
1957                player=player,
1958                streamdetails=next_queue_item.streamdetails,
1959                crossfade_enabled=True,
1960            )
1961            crossfade_allowed = self.crossfade_allowed(
1962                queue_item,
1963                crossfade_mode=crossfade_mode,
1964                player_id=player.player_id,
1965                flow_mode=False,
1966                next_queue_item=next_queue_item,
1967                sample_rate=pcm_format.sample_rate,
1968                next_sample_rate=next_pcm.sample_rate,
1969            )
1970        if not crossfade_allowed:
1971            # no crossfade enabled/allowed, just yield the buffer last part
1972            bytes_written += len(buffer)
1973            for pcm_slice in iter_pcm_slices(bytes(buffer), pcm_format, 1000):
1974                yield pcm_slice
1975                await asyncio.sleep(0)
1976        else:
1977            assert next_queue_item is not None
1978            # the remaining buffer is the fade-out tail of the current track
1979            fade_out_data = bytes(buffer)
1980            buffer = bytearray()
1981            # initialized before the try block — the except handler reads these
1982            first_part_written = 0
1983            second_part_buf = bytearray()
1984            try:
1985                # wrap the next track's stream in a counting generator that caps
1986                # at crossfade_buffer_size and tracks how many bytes were consumed
1987                fade_in_bytes_consumed = 0
1988
1989                _next_item = next_queue_item
1990
1991                async def _limited_fade_in() -> AsyncGenerator[bytes]:
1992                    nonlocal fade_in_bytes_consumed
1993                    async for chunk in self.get_queue_item_stream(
1994                        _next_item,
1995                        pcm_format,
1996                        playback_speed=cast(
1997                            "float", _next_item.extra_attributes.get("playback_speed", 1.0)
1998                        ),
1999                        session_id=session_id,
2000                    ):
2001                        remaining = crossfade_buffer_size - fade_in_bytes_consumed
2002                        if remaining <= 0:
2003                            break
2004                        if len(chunk) > remaining:
2005                            fade_in_bytes_consumed += remaining
2006                            yield chunk[:remaining]
2007                            break
2008                        fade_in_bytes_consumed += len(chunk)
2009                        yield chunk
2010
2011                smart_fade = await self.smart_fades_mixer.build(
2012                    fade_in_streamdetails=cast("StreamDetails", next_queue_item.streamdetails),
2013                    fade_out_streamdetails=streamdetails,
2014                    pcm_format=pcm_format,
2015                    standard_crossfade_duration=standard_crossfade_duration,
2016                    mode=crossfade_mode,
2017                    fade_out_data=fade_out_data,
2018                    fade_in_bytes_len=crossfade_buffer_size,
2019                )
2020                crossfade_timing = smart_fade.timing_info
2021                # Split mix output at end-of-overlap: PRE+CF to A, POST to B's intro.
2022                fadeout_share_bytes = int(
2023                    (crossfade_timing.pre_crossfade_duration + crossfade_timing.crossfade_duration)
2024                    * pcm_format.pcm_sample_size
2025                )
2026                fadeout_share_bytes = (fadeout_share_bytes // frame_size) * frame_size
2027                async for mix_chunk in self.smart_fades_mixer.mix(
2028                    smart_fade,
2029                    fade_in_part=_limited_fade_in(),
2030                    fade_out_part=fade_out_data,
2031                    pcm_format=pcm_format,
2032                ):
2033                    if first_part_written < fadeout_share_bytes:
2034                        # split this chunk so A gets exactly fadeout_share_bytes
2035                        remaining = fadeout_share_bytes - first_part_written
2036                        if len(mix_chunk) > remaining:
2037                            yield mix_chunk[:remaining]
2038                            first_part_written += remaining
2039                            bytes_written += remaining
2040                            second_part_buf.extend(mix_chunk[remaining:])
2041                        else:
2042                            yield mix_chunk
2043                            first_part_written += len(mix_chunk)
2044                            bytes_written += len(mix_chunk)
2045                    else:
2046                        second_part_buf.extend(mix_chunk)
2047                # tail consumed by the mix but not credited to bytes_written
2048                uncredited_tail_bytes = len(fade_out_data) - first_part_written
2049                self._crossfade_data[queue_item.queue_id] = CrossfadeData(
2050                    data=bytes(second_part_buf),
2051                    fade_in_size=fade_in_bytes_consumed,
2052                    pcm_format=pcm_format,
2053                    fade_in_pcm_format=pcm_format,
2054                    queue_item_id=next_queue_item.queue_item_id,
2055                    elapsed_time_offset=(
2056                        crossfade_timing.fadein_trimmed_duration
2057                        + crossfade_timing.crossfade_duration
2058                    ),
2059                    normalization_mode=cast(
2060                        "StreamDetails", next_queue_item.streamdetails
2061                    ).volume_normalization_mode,
2062                )
2063                crossfade_elapsed = asyncio.get_event_loop().time() - crossfade_start_time
2064                self.logger.debug(
2065                    "Stored crossfade data for queue %s"
2066                    " - next queue_item_id: %s (preparation took %.1fs)",
2067                    queue.display_name,
2068                    next_queue_item.queue_item_id,
2069                    crossfade_elapsed,
2070                )
2071            except Exception as err:
2072                if first_part_written or second_part_buf:
2073                    # partial mix already played — concat'd fade_out_data would duplicate audio
2074                    raise
2075                # crossfade failed, fall back to just yielding the fade_out_data
2076                self.logger.warning(
2077                    "Crossfade failed for queue %s: %s",
2078                    queue.display_name,
2079                    err,
2080                )
2081                next_queue_item = None
2082                for pcm_slice in iter_pcm_slices(fade_out_data, pcm_format, 1000):
2083                    yield pcm_slice
2084                    await asyncio.sleep(0)
2085                bytes_written += len(fade_out_data)
2086                del fade_out_data
2087        # make sure the buffer gets cleaned up
2088        del buffer
2089        # update duration details based on the actual pcm data we sent
2090        # this also accounts for crossfade and silence stripping
2091        seconds_streamed = bytes_written / pcm_format.pcm_sample_size
2092        streamdetails.seconds_streamed = seconds_streamed
2093        uncredited_tail_seconds = uncredited_tail_bytes / pcm_format.pcm_sample_size
2094        # streamdetails.duration is in media-time; seconds_streamed is stream-time
2095        # (post-atempo), so we scale by playback_speed to recover media-time.
2096        streamdetails.duration = int(
2097            streamdetails.seek_position
2098            + (seconds_streamed + uncredited_tail_seconds) * playback_speed
2099        )
2100        # propagate accurate duration to queue_item so UI displays it
2101        queue_item.duration = streamdetails.duration
2102        self.logger.debug(
2103            "Finished Streaming queue track: %s (%s) on queue %s "
2104            "- crossfade data prepared for next track: %s",
2105            streamdetails.uri,
2106            queue_item.name,
2107            queue.display_name,
2108            next_queue_item.name if next_queue_item else "N/A",
2109        )
2110
2111    async def get_queue_flow_stream(
2112        self,
2113        queue: PlayerQueue,
2114        start_queue_item: QueueItem,
2115        pcm_format: AudioFormat,
2116        session_id: str | None = None,
2117        protocol_player: Player | None = None,
2118    ) -> AsyncGenerator[bytes]:
2119        """
2120        Get a flow stream of all tracks in the queue as raw PCM audio.
2121
2122        yields chunks of exactly 1 second of audio in the given pcm_format.
2123
2124        :param queue: Queue being streamed.
2125        :param start_queue_item: First queue item in the flow stream.
2126        :param pcm_format: Shared PCM format for the complete flow stream.
2127        :param session_id: Queue session that owns processing-detail updates.
2128        :param protocol_player: The protocol player actually consuming the flow stream.
2129            Must be the same player that was used to select ``pcm_format`` so
2130            restart decisions are made against the correct supported sample rates
2131            and flow mode configuration. Falls back to the queue's player when omitted.
2132        """
2133        # ruff: noqa: PLR0915
2134        assert pcm_format.content_type.is_pcm()
2135        queue_track = None
2136        last_fadeout_part: bytes = b""
2137        last_streamdetails: StreamDetails | None = None
2138        last_play_log_entry: PlayLogEntry | None = None
2139        # Snapshot the queue's current session_id. PlayerQueues rotates this on
2140        # every new stream session, so if a newer producer takes over the queue
2141        # (rapid track switch, sync-group reform, dynamic leader handoff) the
2142        # snapshot will no longer match and we exit cleanly on the next yield or
2143        # playlog append — preventing two producers from writing to the same
2144        # pq_data.flow_mode_stream_log.
2145        pq_data = self.mass.player_queues.queue_data(queue.queue_id)
2146        flow_session_id = session_id or pq_data.session_id
2147        if flow_session_id is None or pq_data.session_id != flow_session_id:
2148            self.logger.debug(
2149                "Ignoring stale flow stream for queue %s (session %s, active %s)",
2150                queue.display_name,
2151                flow_session_id,
2152                pq_data.session_id,
2153            )
2154            return
2155        queue.flow_mode = True
2156        pq_data.flow_mode_stream_log = []
2157        if not start_queue_item:
2158            # this can happen in some (edge case) race conditions
2159            return
2160        pcm_sample_size = pcm_format.pcm_sample_size
2161        if start_queue_item.media_type != MediaType.TRACK:
2162            # no crossfade on non-tracks
2163            crossfade_mode = CrossfadeMode.DISABLED
2164            standard_crossfade_duration = 0
2165        else:
2166            crossfade_mode = self.mass.streams.get_crossfade_mode(queue)
2167            # crossfade duration is a global (queue controller) setting; fallback matches
2168            # CONF_ENTRY_CROSSFADE_DURATION's default
2169            standard_crossfade_duration = self.mass.config.get_raw_core_config_value(
2170                CONF_PLAYER_QUEUES, CONF_CROSSFADE_DURATION, 8
2171            )
2172        flow_mode_sample_rate_conf, flow_supported_sample_rates = self._flow_restart_context(
2173            queue.queue_id, protocol_player
2174        )
2175        # note: get_crossfade_mode() already falls back to standard when smart fades aren't
2176        # available (no analysis provider / minimal buffer), so crossfade_mode is safe to use.
2177        self.logger.info(
2178            "Start Queue Flow stream for Queue %s - crossfade: %s %s",
2179            queue.display_name,
2180            crossfade_mode,
2181            f"({standard_crossfade_duration}s)"
2182            if crossfade_mode == CrossfadeMode.STANDARD_CROSSFADE
2183            else "",
2184        )
2185        total_chunks_received = 0
2186
2187        def _superseded() -> bool:
2188            """Return True if a newer stream session has taken over this queue."""
2189            return pq_data.session_id != flow_session_id
2190
2191        queue_exhausted = False
2192        while True:
2193            # bail out early if a newer producer has taken over this queue,
2194            # so we don't append another entry to a stream log we no longer own
2195            if _superseded():
2196                self.logger.debug(
2197                    "Flow stream for queue %s superseded (session %s -> %s) "
2198                    "- exiting before next track",
2199                    queue.display_name,
2200                    flow_session_id,
2201                    pq_data.session_id,
2202                )
2203                return
2204            # get (next) queue item to stream
2205            if queue_track is None:
2206                queue_track = start_queue_item
2207            else:
2208                try:
2209                    queue_track = await self.mass.player_queues.load_next_queue_item(
2210                        queue.queue_id, queue_track.queue_item_id
2211                    )
2212                except QueueEmpty:
2213                    queue_exhausted = True
2214                    break
2215
2216            if self._flow_stream_needs_restart(
2217                queue_track,
2218                pcm_format,
2219                flow_supported_sample_rates,
2220                flow_mode_sample_rate_conf,
2221                is_first_track=queue_track is start_queue_item,
2222            ):
2223                break
2224
2225            if queue_track.streamdetails is None:
2226                self.logger.error(
2227                    "No StreamDetails for queue item %s (%s) on queue %s - skipping track",
2228                    queue_track.queue_item_id,
2229                    queue_track.name,
2230                    queue.display_name,
2231                )
2232                continue
2233            if flow_session_id is not None:
2234                self.mass.streams.audio_processing.update_item_context(
2235                    queue_id=queue.queue_id,
2236                    session_id=flow_session_id,
2237                    queue_item_id=queue_track.queue_item_id,
2238                    queue_processing=AudioQueueProcessing(
2239                        pcm_format=pcm_format,
2240                        playback_speed=cast(
2241                            "float",
2242                            queue_track.extra_attributes.get("playback_speed", 1.0),
2243                        ),
2244                        crossfade_mode=crossfade_mode,
2245                        overlay_active=overlay_active(queue),
2246                    ),
2247                    alters_audio=queue_track.streamdetails.fade_in,
2248                )
2249
2250            self.logger.debug(
2251                "Start Streaming queue track: %s (%s) for queue %s",
2252                queue_track.streamdetails.uri,
2253                queue_track.name,
2254                queue.display_name,
2255            )
2256            # last chance to bail before mutating the stream log: a newer producer
2257            # may have taken over while we were awaiting load_next_queue_item
2258            if _superseded():
2259                self.logger.debug(
2260                    "Flow stream for queue %s superseded - exiting before playlog append",
2261                    queue.display_name,
2262                )
2263                return
2264            track_playback_speed = cast(
2265                "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2266            )
2267            # calculate crossfade buffer size
2268            crossfade_buffer_duration = (
2269                SMART_CROSSFADE_DURATION
2270                if crossfade_mode == CrossfadeMode.SMART_CROSSFADE
2271                else standard_crossfade_duration
2272            )
2273            crossfade_buffer_duration = min(
2274                crossfade_buffer_duration,
2275                int(queue_track.streamdetails.duration / 2)
2276                if queue_track.streamdetails.duration
2277                else crossfade_buffer_duration,
2278            )
2279            # skip crossfade if buffer would be too small to be meaningful
2280            if crossfade_buffer_duration < 5:
2281                crossfade_buffer_duration = 0
2282            # Ensure crossfade buffer size is aligned to frame boundaries
2283            # Frame size = bytes_per_sample * channels
2284            bytes_per_sample = pcm_format.bit_depth // 8
2285            frame_size = bytes_per_sample * pcm_format.channels
2286            crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
2287            # Round down to nearest frame boundary
2288            crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
2289            warmup_size = int(pcm_format.pcm_sample_size * WARMUP_DURATION)
2290
2291            # raw_seek_position feeds the PCM buffer; streamdetails.seek_position
2292            # (overwritten below) only drives reported elapsed time.
2293            raw_seek_position = queue_track.streamdetails.seek_position
2294            # Build eagerly so seek_position is set before PlayLogEntry is appended —
2295            # consumer-paced mix() would otherwise let the queue briefly report 0.
2296            crossfade_smart_fade: SmartFade | None = None
2297            if (
2298                last_fadeout_part
2299                and last_streamdetails
2300                and crossfade_buffer_size > 0
2301                and crossfade_mode != CrossfadeMode.DISABLED
2302            ):
2303                crossfade_smart_fade = await self.smart_fades_mixer.build(
2304                    fade_in_streamdetails=queue_track.streamdetails,
2305                    fade_out_streamdetails=last_streamdetails,
2306                    pcm_format=pcm_format,
2307                    standard_crossfade_duration=standard_crossfade_duration,
2308                    mode=crossfade_mode,
2309                    fade_out_data=last_fadeout_part,
2310                    fade_in_bytes_len=crossfade_buffer_size,
2311                )
2312                timing_info = crossfade_smart_fade.timing_info
2313                queue_track.streamdetails.seek_position = (
2314                    raw_seek_position
2315                    + timing_info.fadein_trimmed_duration
2316                    + timing_info.crossfade_duration
2317                )
2318            # append to play log so the queue controller can work out which track is playing
2319            play_log_entry = PlayLogEntry(queue_track.queue_item_id)
2320            pq_data.flow_mode_stream_log.append(play_log_entry)
2321
2322            bytes_written = 0
2323            crossfade_buffer = bytearray()
2324            warmup_bytes = 0
2325            first_chunk_received = False
2326
2327            async for chunk in self.get_queue_item_stream(
2328                queue_track,
2329                pcm_format=pcm_format,
2330                seek_position=int(raw_seek_position),
2331                playback_speed=cast(
2332                    "float", queue_track.extra_attributes.get("playback_speed", 1.0)
2333                ),
2334                raise_on_error=False,
2335                session_id=flow_session_id,
2336            ):
2337                # if a newer producer has taken over this queue, stop sending
2338                # audio and exit cleanly before the outer-loop end-of-track
2339                # bookkeeping mutates seconds_streamed / duration on the log
2340                if _superseded():
2341                    self.logger.debug(
2342                        "Flow stream for queue %s superseded - stopping chunk yield",
2343                        queue.display_name,
2344                    )
2345                    return
2346                total_chunks_received += 1
2347                if not first_chunk_received:
2348                    first_chunk_received = True
2349                    # inform the queue that the track is now loaded in the buffer
2350                    # so the next track can be preloaded
2351                    self.mass.player_queues.track_loaded_in_buffer(
2352                        queue.queue_id, queue_track.queue_item_id
2353                    )
2354
2355                if crossfade_mode == CrossfadeMode.DISABLED:
2356                    # no cross/smart fade: yield chunks directly without intermediate buffer
2357                    yield chunk
2358                    bytes_written += len(chunk)
2359                    del chunk
2360                    continue
2361
2362                # Warmup: yield chunks directly until we have streamed WARMUP_DURATION
2363                # worth of audio, so playback starts immediately. Skip warmup when
2364                # crossfade data from the previous track is pending — we need a full
2365                # buffer for the mix.
2366                if warmup_bytes < warmup_size and not last_fadeout_part:
2367                    yield chunk
2368                    warmup_bytes += len(chunk)
2369                    bytes_written += len(chunk)
2370                    del chunk
2371                    continue
2372
2373                # smart fades enabled: accumulate chunks in crossfade buffer
2374                crossfade_buffer.extend(chunk)
2375                del chunk
2376                if len(crossfade_buffer) < crossfade_buffer_size:
2377                    await asyncio.sleep(0)
2378                    continue
2379
2380                # handle crossfade of previous track and new track
2381                if (
2382                    last_fadeout_part
2383                    and last_streamdetails
2384                    and crossfade_smart_fade is not None
2385                    and last_play_log_entry is not None
2386                ):
2387                    fadein_part = bytes(crossfade_buffer[:crossfade_buffer_size])
2388                    remaining_bytes = bytes(crossfade_buffer[crossfade_buffer_size:])
2389                    try:
2390                        crossfade_bytes_written = 0
2391                        async for mix_chunk in self.smart_fades_mixer.mix(
2392                            crossfade_smart_fade,
2393                            fade_in_part=fadein_part,
2394                            fade_out_part=last_fadeout_part,
2395                            pcm_format=pcm_format,
2396                        ):
2397                            yield mix_chunk
2398                            crossfade_bytes_written += len(mix_chunk)
2399                    except Exception as mix_err:
2400                        if crossfade_bytes_written:
2401                            # partial mix already played — concat'd tail would duplicate audio
2402                            raise
2403                        self.logger.warning(
2404                            "Crossfade mixer failed for %s, falling back to simple concat: %s",
2405                            queue_track.name,
2406                            mix_err,
2407                        )
2408                        for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2409                            yield pcm_slice
2410                            await asyncio.sleep(0)
2411                        # full tail was pre-counted and is now yielded as-is
2412                        crossfade_bytes_written = 0
2413                        remaining_bytes = bytes(crossfade_buffer)
2414                        # mix failed — undo the eager seek_position
2415                        queue_track.streamdetails.seek_position = raw_seek_position
2416                    if crossfade_bytes_written:
2417                        # Split mix output at end-of-overlap: PRE+CF to A, POST to B.
2418                        fadeout_share_seconds = (
2419                            timing_info.pre_crossfade_duration + timing_info.crossfade_duration
2420                        )
2421                        fadeout_share = int(fadeout_share_seconds * pcm_sample_size)
2422                        fadeout_share = (fadeout_share // frame_size) * frame_size
2423                        fadeout_share = min(fadeout_share, crossfade_bytes_written)
2424                        fadein_share = crossfade_bytes_written - fadeout_share
2425                        bytes_written += fadein_share
2426                        if last_play_log_entry:
2427                            assert last_play_log_entry.seconds_streamed is not None
2428                            # correct pre-counted full tail to the timing-based share
2429                            last_play_log_entry.seconds_streamed += (
2430                                fadeout_share - len(last_fadeout_part)
2431                            ) / pcm_sample_size
2432                    if remaining_bytes:
2433                        for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2434                            yield pcm_slice
2435                            await asyncio.sleep(0)
2436                        bytes_written += len(remaining_bytes)
2437                        del remaining_bytes
2438                    last_fadeout_part = b""
2439                    last_streamdetails = None
2440                    crossfade_buffer = bytearray()
2441                    warmup_bytes = 0
2442
2443                # yield everything above the crossfade buffer size
2444                while len(crossfade_buffer) > crossfade_buffer_size:
2445                    yield bytes(crossfade_buffer[:pcm_sample_size])
2446                    bytes_written += pcm_sample_size
2447                    del crossfade_buffer[:pcm_sample_size]
2448                    await asyncio.sleep(0)
2449
2450            #### HANDLE END OF TRACK
2451            if not first_chunk_received:
2452                self.logger.warning(
2453                    "Track %s (%s) on queue %s produced no audio data - skipping",
2454                    queue_track.name,
2455                    queue_track.streamdetails.uri if queue_track.streamdetails else "unknown",
2456                    queue.display_name,
2457                )
2458                queue_track.streamdetails.stream_error = True
2459                play_log_entry.seconds_streamed = 0
2460                continue
2461            if last_fadeout_part:
2462                # edge case: we did not get enough data to make the crossfade
2463                # attribute these bytes to the previous track (they are its tail)
2464                for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2465                    yield pcm_slice
2466                    await asyncio.sleep(0)
2467                # no crossfade happened — undo the eager seek_position
2468                queue_track.streamdetails.seek_position = raw_seek_position
2469                # full tail was pre-counted and is now yielded as-is
2470                last_fadeout_part = b""
2471            if self.crossfade_allowed(
2472                queue_track,
2473                crossfade_mode=crossfade_mode,
2474                player_id=queue.queue_id,
2475                flow_mode=True,
2476            ):
2477                last_fadeout_part = bytes(crossfade_buffer[-crossfade_buffer_size:])
2478                last_streamdetails = queue_track.streamdetails
2479                last_play_log_entry = play_log_entry
2480                remaining_bytes = bytes(crossfade_buffer[:-crossfade_buffer_size])
2481                if remaining_bytes:
2482                    for pcm_slice in iter_pcm_slices(remaining_bytes, pcm_format, 1000):
2483                        yield pcm_slice
2484                        await asyncio.sleep(0)
2485                    bytes_written += len(remaining_bytes)
2486                del remaining_bytes
2487            elif crossfade_mode != CrossfadeMode.DISABLED and crossfade_buffer:
2488                bytes_written += len(crossfade_buffer)
2489                for pcm_slice in iter_pcm_slices(bytes(crossfade_buffer), pcm_format, 1000):
2490                    yield pcm_slice
2491                    await asyncio.sleep(0)
2492            crossfade_buffer = bytearray()
2493
2494            # update duration details based on the actual pcm data we sent
2495            # this also accounts for crossfade and silence stripping
2496            seconds_streamed = bytes_written / pcm_sample_size
2497            queue_track.streamdetails.seconds_streamed = seconds_streamed
2498            # the held-back crossfade tail still counts as this track's media-time
2499            tail_seconds = len(last_fadeout_part) / pcm_sample_size
2500            # streamdetails.duration is in media-time; seconds_streamed is stream-time
2501            # (post-atempo), so we scale by the track's playback_speed to recover media-time.
2502            queue_track.streamdetails.duration = int(
2503                queue_track.streamdetails.seek_position
2504                + (seconds_streamed + tail_seconds) * track_playback_speed
2505            )
2506            # propagate accurate duration to queue_item so UI displays it
2507            queue_track.duration = queue_track.streamdetails.duration
2508            play_log_entry.seconds_streamed = seconds_streamed
2509            play_log_entry.duration = queue_track.streamdetails.duration
2510            if last_play_log_entry is play_log_entry and last_fadeout_part:
2511                # Pre-count the full crossfade tail so the queue index calculation
2512                # doesn't undercount while waiting for the next track's crossfade mix.
2513                # This will be corrected to crossfade_total/2 once the mix completes.
2514                assert play_log_entry.seconds_streamed is not None
2515                play_log_entry.seconds_streamed += len(last_fadeout_part) / pcm_sample_size
2516            self.logger.debug(
2517                "Finished Streaming queue track: %s (%s) on queue %s",
2518                queue_track.streamdetails.uri,
2519                queue_track.name,
2520                queue.display_name,
2521            )
2522        #### HANDLE END OF QUEUE FLOW STREAM
2523        # skip end-of-queue bookkeeping if a newer producer has superseded us;
2524        # the new producer owns queue_buffer_completed and the play log now
2525        if _superseded():
2526            self.logger.debug(
2527                "Flow stream for queue %s superseded - skipping end-of-queue handling",
2528                queue.display_name,
2529            )
2530            return
2531        # end of queue flow: make sure we yield the last_fadeout_part
2532        if last_fadeout_part:
2533            for pcm_slice in iter_pcm_slices(last_fadeout_part, pcm_format, 1000):
2534                yield pcm_slice
2535                await asyncio.sleep(0)
2536            # correct seconds streamed - the duration already includes the tail
2537            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
2538            streamdetails = queue_track.streamdetails
2539            assert streamdetails is not None
2540            streamdetails.seconds_streamed = (
2541                streamdetails.seconds_streamed or 0
2542            ) + last_part_seconds
2543            # also update the play log entry so elapsed time tracking stays in sync
2544            if last_play_log_entry:
2545                assert last_play_log_entry.seconds_streamed is not None
2546                # full tail was pre-counted and is now yielded as-is
2547                last_play_log_entry.duration = streamdetails.duration
2548            last_fadeout_part = b""
2549        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
2550        # only signal completion if we are still the active producer — a later
2551        # producer would (incorrectly) see this as its own completion otherwise
2552        if not _superseded():
2553            # inform the queue controller that all audio data has been generated
2554            # so it can handle the case where new items were added after the flow stream ended
2555            self.mass.player_queues.queue_buffer_completed(queue.queue_id, queue_exhausted)
2556
2557    async def get_overlay_mixed_stream(
2558        self,
2559        queue: PlayerQueue,
2560        audio_input: AsyncGenerator[bytes],
2561        pcm_format: AudioFormat,
2562    ) -> AsyncGenerator[bytes]:
2563        """
2564        Mix the queue's audio overlay (looping sound effect) into the given PCM stream.
2565
2566        The mixed output has the exact same PCM format, duration and chunking as the
2567        input stream. If the overlay source can not be resolved, the original stream
2568        is passed through unchanged so playback is never interrupted.
2569
2570        :param queue: The PlayerQueue holding the overlay source and volume.
2571        :param audio_input: The audio stream (raw PCM in ``pcm_format``) to mix into.
2572        :param pcm_format: PCM format of both the input and the mixed output.
2573        """
2574        overlay_input = await self._resolve_overlay_input(queue)
2575        if overlay_input is None:
2576            # overlay source unavailable: degrade gracefully to music-only
2577            async for chunk in audio_input:
2578                yield chunk
2579            return
2580        async for chunk in get_ffmpeg_overlay_stream(
2581            audio_input=audio_input,
2582            overlay_input=overlay_input,
2583            pcm_format=pcm_format,
2584            overlay_volume=queue.overlay_volume,
2585            chunk_size=pcm_format.pcm_sample_size,
2586        ):
2587            yield chunk
2588
2589    def crossfade_allowed(
2590        self,
2591        queue_item: QueueItem,
2592        crossfade_mode: CrossfadeMode,
2593        player_id: str,
2594        flow_mode: bool = False,
2595        next_queue_item: QueueItem | None = None,
2596        sample_rate: int | None = None,
2597        next_sample_rate: int | None = None,
2598    ) -> bool:
2599        """Get the crossfade config for a queue item."""
2600        if crossfade_mode == CrossfadeMode.DISABLED:
2601            return False
2602        if not (self.mass.player_queues.get(queue_item.queue_id)):
2603            return False  # just a guard
2604        if not (self.mass.players.get_player(player_id)):
2605            return False  # just a guard
2606        if queue_item.media_type != MediaType.TRACK:
2607            self.logger.debug("Skipping crossfade: current item is not a track")
2608            return False
2609        # check if the next item is part of the same album
2610        next_item = next_queue_item or self.mass.player_queues.get_next_item(
2611            queue_item.queue_id, queue_item.queue_item_id
2612        )
2613        if not next_item:
2614            # there is no next item!
2615            return False
2616        # check if next item is a track
2617        if next_item.media_type != MediaType.TRACK:
2618            self.logger.debug("Skipping crossfade: next item is not a track")
2619            return False
2620        if (
2621            isinstance(queue_item.media_item, Track)
2622            and isinstance(next_item.media_item, Track)
2623            and queue_item.media_item.album
2624            and next_item.media_item.album
2625            and queue_item.media_item.album == next_item.media_item.album
2626            and not self.mass.config.get_raw_core_config_value(
2627                "streams", CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
2628            )
2629        ):
2630            # in general, crossfade is not desired for tracks of the same (gapless) album
2631            # because we have no accurate way to determine if the album is gapless or not,
2632            # for now we just never crossfade between tracks of the same album
2633            self.logger.debug("Skipping crossfade: next item is part of the same album")
2634            return False
2635
2636        # check if we're allowed to crossfade on different sample rates
2637        if (
2638            not flow_mode
2639            and sample_rate
2640            and next_sample_rate
2641            and sample_rate != next_sample_rate
2642            and not self.mass.config.get_raw_player_config_value(
2643                player_id,
2644                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key,
2645                CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.default_value,
2646            )
2647        ):
2648            self.logger.debug(
2649                "Skipping crossfade: player(protocol) does not support gapless playback "
2650                "with different sample rates (%s vs %s)",
2651                sample_rate,
2652                next_sample_rate,
2653            )
2654            return False
2655
2656        return True
2657
2658    def clear_crossfade_data(self, queue_id: str) -> None:
2659        """
2660        Clear any pending crossfade data for a queue.
2661
2662        :param queue_id: The queue ID to clear crossfade data for.
2663        """
2664        if queue_id in self._crossfade_data:
2665            self.logger.debug("Clearing crossfade data for queue %s", queue_id)
2666            del self._crossfade_data[queue_id]
2667
2668    async def get_shoutcast_stream(
2669        self, url: str, streamdetails: StreamDetails
2670    ) -> AsyncGenerator[bytes]:
2671        """
2672        Yield audio from a legacy Shoutcast server, with ICY metadata parsed inline.
2673
2674        :param url: Shoutcast stream URL.
2675        :param streamdetails: StreamDetails to update with ICY metadata as it arrives.
2676        """
2677        self.logger.debug("Start streaming from legacy Shoutcast server: %s", url)
2678
2679        parsed = urlparse(url)
2680        host = parsed.hostname
2681        port = parsed.port or 80
2682        path = parsed.path or "/"
2683        if parsed.query:
2684            path = f"{path}?{parsed.query}"
2685
2686        try:
2687            # Open raw socket connection
2688            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=30)
2689        except TimeoutError as err:
2690            raise AudioError(f"Timeout connecting to Shoutcast stream {url}") from err
2691        except (OSError, ConnectionError) as err:
2692            raise AudioError(f"Failed to connect to Shoutcast stream {url}") from err
2693
2694        try:
2695            # Send HTTP request with ICY metadata header
2696            request = (
2697                f"GET {path} HTTP/1.1\r\n"
2698                f"Host: {host}\r\n"
2699                f"User-Agent: {HTTP_HEADERS['User-Agent']}\r\n"
2700                f"Icy-MetaData: 1\r\n\r\n"
2701            )
2702            writer.write(request.encode())
2703            await writer.drain()
2704
2705            # Read and parse response line
2706            try:
2707                response_line = await asyncio.wait_for(reader.readline(), timeout=10)
2708            except TimeoutError as err:
2709                raise AudioError("Timeout reading Shoutcast response") from err
2710
2711            if not response_line.startswith(b"ICY"):
2712                raise InvalidDataError("Invalid Shoutcast response")
2713
2714            # Read headers until empty line
2715            headers: dict[str, str] = {}
2716            while True:
2717                try:
2718                    line = await asyncio.wait_for(reader.readline(), timeout=5)
2719                except TimeoutError as err:
2720                    raise AudioError("Timeout reading Shoutcast headers") from err
2721
2722                if line in (b"\r\n", b"\n", b""):
2723                    break
2724
2725                if b":" in line:
2726                    try:
2727                        key, value = line.decode("latin-1", errors="ignore").split(":", 1)
2728                        headers[key.strip().lower()] = value.strip()
2729                    except UnicodeDecodeError, ValueError:
2730                        continue
2731
2732            # Get metadata interval
2733            meta_int_str = headers.get("icy-metaint")
2734            if not meta_int_str:
2735                raise InvalidDataError("No icy-metaint header in Shoutcast response")
2736
2737            try:
2738                meta_int = int(meta_int_str)
2739            except ValueError as err:
2740                raise InvalidDataError("Invalid icy-metaint value") from err
2741
2742            self.logger.debug("Connected to Shoutcast stream %s (icy-metaint: %s)", url, meta_int)
2743
2744            # Stream audio data with metadata parsing
2745            while True:
2746                try:
2747                    # Read audio chunk
2748                    audio_chunk = await reader.readexactly(meta_int)
2749                    yield audio_chunk
2750
2751                    # Read metadata length
2752                    meta_byte = await reader.readexactly(1)
2753                    if meta_byte == b"\x00":
2754                        continue
2755
2756                    meta_length = ord(meta_byte) * 16
2757                    meta_data = await reader.readexactly(meta_length)
2758                    self._parse_icy_metadata(meta_data, streamdetails)
2759
2760                except asyncio.exceptions.IncompleteReadError:
2761                    # End of stream
2762                    break
2763
2764        finally:
2765            writer.close()
2766            await writer.wait_closed()
2767
2768    # --- Private methods ---
2769
2770    def _notify_provider_streamed(
2771        self, streamdetails: StreamDetails, finished: bool, seconds_streamed: float
2772    ) -> None:
2773        """Report a (mostly) streamed item back to the provider that owns it."""
2774        if not finished and seconds_streamed < 90:
2775            return
2776        provider = self.mass.get_provider(streamdetails.provider)
2777        # plugin providers serve playable items too, but on_streamed is MusicProvider-only
2778        if provider is None or provider.type != ProviderType.MUSIC:
2779            return
2780        music_prov = cast("MusicProvider", provider)
2781        self.mass.create_task(music_prov.on_streamed(streamdetails))
2782
2783    def _get_volume_normalization_preference(
2784        self, streamdetails: StreamDetails
2785    ) -> VolumeNormalizationMode:
2786        """Return the configured normalization preference for the stream's media type."""
2787        conf_key = (
2788            CONF_VOLUME_NORMALIZATION_RADIO
2789            if streamdetails.media_type == MediaType.RADIO
2790            else CONF_VOLUME_NORMALIZATION_TRACKS
2791        )
2792        return VolumeNormalizationMode(
2793            self.mass.streams.get_config_value(conf_key, return_type=str)
2794        )
2795
2796    def _update_radio_stream_metadata(
2797        self,
2798        streamdetails: StreamDetails,
2799        artist: str | None,
2800        title: str,
2801        image_url: str | None = None,
2802        album: str | None = None,
2803    ) -> None:
2804        """
2805        Update radio stream metadata and trigger artwork lookup.
2806
2807        :param streamdetails: The stream details to update.
2808        :param artist: Artist name (will be normalized).
2809        :param title: Track title (will be cleaned for display).
2810        :param image_url: Optional image URL from stream metadata.
2811        :param album: Optional album name.
2812        """
2813        station_image_url = image_url or self.mass.metadata.get_radio_stream_station_image(
2814            streamdetails
2815        )
2816        artist_normalized = (
2817            self.mass.metadata.normalize_radio_artist_name(artist) if artist else None
2818        )
2819        display_title, _ = parse_title_and_version(title, strip_for_display=True)
2820
2821        streamdetails.stream_metadata = StreamMetadata(
2822            title=display_title,
2823            artist=artist_normalized,
2824            album=album,
2825            image_url=station_image_url,
2826        )
2827        streamdetails.stream_metadata_last_updated = time.time()
2828        if streamdetails.queue_id:
2829            self.mass.player_queues.signal_update(streamdetails.queue_id)
2830
2831        # Fetch artwork in background (track, album then artist)
2832        if artist and title and not image_url:
2833            self.mass.call_later(
2834                0.2,
2835                self.mass.metadata.update_radio_stream_artwork,
2836                streamdetails,
2837                task_id=f"update_radio_artwork_{streamdetails.queue_id}",
2838            )
2839
2840    async def _cache_radio_result(
2841        self,
2842        url: str,
2843        stream_type: StreamType,
2844        resolved_url: str | None = None,
2845    ) -> tuple[str, StreamType]:
2846        """Cache and return a radio stream resolution result."""
2847        result = (resolved_url or url, stream_type)
2848        await self.mass.cache.set(
2849            url,
2850            result,
2851            expiration=3600 * 3,
2852            provider=CACHE_PROVIDER,
2853            category=CACHE_CATEGORY_RESOLVED_RADIO_URL,
2854        )
2855        return result
2856
2857    async def _handle_client_error_for_radio_stream(
2858        self, url: str, err: aiohttp.ClientError, fallback_stream_type: StreamType
2859    ) -> tuple[str, StreamType]:
2860        """Handle aiohttp client errors during radio stream resolution."""
2861        # Prefer the final post-redirect URL: aiohttp follows redirects before raising,
2862        # but the original url may just point at a redirector rather than the ICY endpoint.
2863        request_info = getattr(err, "request_info", None)
2864        validate_url = str(request_info.url) if request_info is not None else url
2865
2866        # Check if this is a Shoutcast/ICY response that aiohttp can't parse
2867        if isinstance(err, aiohttp.ClientResponseError) and "ICY" in str(err).upper():
2868            self.logger.debug(
2869                "ICY response detected for %s, validating Shoutcast stream", validate_url
2870            )
2871            if await self._validate_shoutcast_stream(validate_url):
2872                return await self._cache_radio_result(
2873                    url, StreamType.SHOUTCAST, resolved_url=validate_url
2874                )
2875            self.logger.warning(
2876                "ICY response detected but Shoutcast validation failed for %s", validate_url
2877            )
2878            return await self._cache_radio_result(
2879                url, fallback_stream_type, resolved_url=validate_url
2880            )
2881
2882        # Other aiohttp errors - might still be Shoutcast, check it
2883        self.logger.debug("aiohttp error for %s, checking if legacy Shoutcast stream", validate_url)
2884        if await self._validate_shoutcast_stream(validate_url):
2885            return await self._cache_radio_result(
2886                url, StreamType.SHOUTCAST, resolved_url=validate_url
2887            )
2888
2889        # Unknown error - still try to stream
2890        self.logger.warning(
2891            "Failed to parse radio URL %s: %s - attempting direct stream", validate_url, str(err)
2892        )
2893        return await self._cache_radio_result(url, fallback_stream_type, resolved_url=validate_url)
2894
2895    async def _iter_audio_source_pcm(
2896        self,
2897        streamdetails: StreamDetails,
2898        pcm_format: AudioFormat,
2899    ) -> AsyncGenerator[bytes]:
2900        """Yield PCM for an AudioSource, bypassing ffmpeg when formats match."""
2901        if _pcm_formats_match(streamdetails.audio_format, pcm_format):
2902            source_gen = self._open_audio_source_generator(streamdetails)
2903            async for chunk in realtime_pcm_pacer(source_gen, pcm_format):
2904                yield chunk
2905            return
2906        # format mismatch → fall back to ffmpeg for resampling (still small chunks)
2907        async for chunk in self.get_media_stream(
2908            streamdetails=streamdetails,
2909            pcm_format=pcm_format,
2910            filter_params=None,
2911            chunk_seconds=AUDIO_SOURCE_CHUNK_SECONDS,
2912        ):
2913            yield chunk
2914
2915    def _open_audio_source_generator(self, streamdetails: StreamDetails) -> AsyncGenerator[bytes]:
2916        """Open the raw PCM generator for an AudioSource (CUSTOM or NAMED_PIPE)."""
2917        if streamdetails.stream_type == StreamType.CUSTOM:
2918            provider = self.mass.get_provider(streamdetails.provider)
2919            if provider is None:
2920                raise ProviderUnavailableError(
2921                    f"Provider {streamdetails.provider} for stream is no longer available"
2922                )
2923            provider = cast("MusicProvider | PluginProvider", provider)
2924            return provider.get_audio_stream(streamdetails)
2925        if streamdetails.stream_type == StreamType.NAMED_PIPE:
2926            assert isinstance(streamdetails.path, str)  # for type checking
2927            return read_named_pipe(streamdetails.path)
2928        raise AudioError(f"Unsupported stream_type {streamdetails.stream_type} for AudioSource")
2929
2930    def _parse_icy_metadata(self, meta_data: bytes, streamdetails: StreamDetails) -> None:
2931        """
2932        Parse ICY metadata and update streamdetails.
2933
2934        Sets the cleaned stream title and, when the title parses as "Artist - Track",
2935        triggers a radio-artwork metadata update.
2936
2937        :param meta_data: Raw metadata bytes from an ICY stream chunk.
2938        :param streamdetails: StreamDetails to update with parsed title and metadata.
2939        """
2940        if not meta_data:
2941            return
2942
2943        meta_data = meta_data.rstrip(b"\0")
2944        # Match StreamTitle, handling apostrophes in titles
2945        stream_title_re = re.search(rb"StreamTitle='(.*?)';", meta_data)
2946
2947        if not stream_title_re:
2948            self.logger.log(
2949                VERBOSE_LOG_LEVEL,
2950                "ICY metadata does not contain StreamTitle field. Raw: %s",
2951                meta_data.decode("utf-8", errors="replace")[:200],
2952            )
2953            return
2954
2955        try:
2956            # in 99% of the cases the stream title is utf-8 encoded
2957            stream_title = stream_title_re.group(1).decode("utf-8")
2958        except UnicodeDecodeError:
2959            # fallback to iso-8859-1
2960            stream_title = stream_title_re.group(1).decode("iso-8859-1", errors="replace")
2961
2962        cleaned_stream_title = clean_stream_title(stream_title)
2963
2964        if not cleaned_stream_title:
2965            return
2966
2967        if cleaned_stream_title == streamdetails.stream_title:
2968            return
2969
2970        self.logger.log(VERBOSE_LOG_LEVEL, "ICY Radio streamtitle original: %s", stream_title)
2971        self.logger.log(
2972            VERBOSE_LOG_LEVEL, "ICY Radio streamtitle cleaned: %s", cleaned_stream_title
2973        )
2974        streamdetails.stream_title = cleaned_stream_title
2975
2976        # Prefer station-provided cover art from the ICY 'StreamUrl' field (when it is
2977        # an image) over the MusicBrainz artwork lookup in _update_radio_stream_metadata.
2978        image_url = self._parse_icy_image_url(meta_data)
2979
2980        # Parse the original title for structured fields first so stations that announce
2981        # an album can refine the artwork lookup; fall back to the "Artist - Track" split.
2982        album: str | None = None
2983        if parsed := parse_quoted_stream_title(stream_title):
2984            track_name, artist_name_raw, album = parsed
2985        elif " - " in cleaned_stream_title:
2986            artist_name_raw, track_name = (
2987                part.strip() for part in cleaned_stream_title.split(" - ", 1)
2988            )
2989        else:
2990            return
2991
2992        if artist_name_raw and track_name:
2993            self.logger.debug(
2994                "ICY metadata: artist='%s', track='%s', album='%s'",
2995                artist_name_raw,
2996                track_name,
2997                album,
2998            )
2999            self._update_radio_stream_metadata(
3000                streamdetails,
3001                artist=artist_name_raw,
3002                title=track_name,
3003                album=album,
3004                image_url=image_url,
3005            )
3006
3007    def _parse_icy_image_url(self, meta_data: bytes) -> str | None:
3008        """
3009        Return a PNG or JPEG cover-art URL from the ICY 'StreamUrl' field, if present.
3010
3011        :param meta_data: Raw metadata bytes from an ICY stream chunk.
3012        """
3013        # The trailing semicolon is optional to match sources that omit it.
3014        stream_url_re = re.search(rb"StreamUrl='([^']*)'", meta_data)
3015        if not stream_url_re:
3016            return None
3017        try:
3018            image_url = stream_url_re.group(1).decode("utf-8").strip()
3019        except UnicodeDecodeError:
3020            return None
3021        if not image_url:
3022            return None
3023        # StreamUrl is not a standardized artwork field (reference clients such as VLC
3024        # ignore it and it conventionally holds a station website link), so only accept
3025        # values that point at a PNG or JPEG image.
3026        parsed = urlparse(image_url)
3027        if parsed.scheme not in ("http", "https"):
3028            return None
3029        if not parsed.path.lower().endswith((".png", ".jpg", ".jpeg")):
3030            return None
3031        self.logger.debug("ICY metadata: StreamUrl image='%s'", image_url)
3032        return image_url
3033
3034    async def _validate_shoutcast_stream(self, url: str) -> bool:
3035        """
3036        Return True if the URL responds with a legacy Shoutcast "ICY 200 OK" line.
3037
3038        :param url: The URL to validate.
3039        """
3040        try:
3041            parsed = urlparse(url)
3042            host = parsed.hostname
3043            port = parsed.port or 80
3044            path = parsed.path or "/"
3045            if parsed.query:
3046                path = f"{path}?{parsed.query}"
3047
3048            # Open raw socket connection with timeout
3049            reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=10)
3050            try:
3051                # Send minimal HTTP request with ICY metadata header
3052                request = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nIcy-MetaData: 1\r\n\r\n"
3053                writer.write(request.encode())
3054                await writer.drain()
3055
3056                # Read just the response line
3057                response_line = await asyncio.wait_for(reader.readline(), timeout=5)
3058            finally:
3059                writer.close()
3060                await writer.wait_closed()
3061
3062            # Check if response starts with "ICY"
3063            decoded_line = response_line.decode("latin-1", errors="ignore").strip()
3064            return decoded_line.startswith("ICY")
3065
3066        except TimeoutError:
3067            self.logger.debug("Timeout during Shoutcast validation for %s", url)
3068            return False
3069        except OSError, ConnectionError:
3070            self.logger.debug("Connection failed during Shoutcast validation for %s", url)
3071            return False
3072        except UnicodeDecodeError:
3073            self.logger.debug("Invalid response encoding during Shoutcast validation for %s", url)
3074            return False
3075
3076    def _resolve_player_dsp_config(self, player: Player) -> DSPConfig:
3077        """
3078        Resolve the effective DSP config for a player.
3079
3080        Single source of truth shared by every code path that needs to know
3081        whether DSP will run for this player. Protocol wrappers defer to their
3082        parent player; single-leg ``player_group`` instances that don't expose
3083        ``MULTI_DEVICE_DSP`` defer to their first member; players whose grouping
3084        context prevents DSP get a disabled config back regardless.
3085
3086        :param player: The player to resolve DSP config for.
3087        """
3088        dsp_player_id = self._resolve_player_dsp_config_id(player)
3089        dsp = self.mass.config.get_player_dsp_config(dsp_player_id)
3090        if is_grouping_preventing_dsp(player):
3091            dsp.enabled = False
3092        elif player.provider.domain == "player_group" and (
3093            PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3094        ):
3095            if not player.state.group_members:
3096                dsp.enabled = False
3097        return dsp
3098
3099    def _resolve_player_dsp_config_id(self, player: Player) -> str:
3100        """
3101        Return the player identifier that supplies the effective DSP config.
3102
3103        :param player: Player whose DSP config source should be resolved.
3104        """
3105        dsp_player_id = player.protocol_parent_id or player.player_id
3106        if (
3107            not is_grouping_preventing_dsp(player)
3108            and player.provider.domain == "player_group"
3109            and PlayerFeature.MULTI_DEVICE_DSP not in player.state.supported_features
3110            and player.state.group_members
3111        ):
3112            child_player = self.mass.players.get_player(player.state.group_members[0])
3113            assert child_player is not None
3114            dsp_player_id = child_player.player_id
3115        return dsp_player_id
3116
3117    def _get_output_channels(self, player: Player | None, player_id: str) -> str:
3118        """
3119        Return the configured output channels for the rendering player.
3120
3121        The value may be stored on the rendering player(protocol) itself (the
3122        protocol section of the config UI) or on its visible parent player (the
3123        native section); the rendering player's own stored value wins.
3124        """
3125        parent_id = player.protocol_parent_id if player and player.protocol_parent_id else player_id
3126        parent_value = self.mass.config.get_raw_player_config_value(
3127            parent_id, CONF_OUTPUT_CHANNELS, "stereo"
3128        )
3129        return self.mass.config.get_raw_player_config_value(
3130            player.player_id if player else player_id, CONF_OUTPUT_CHANNELS, parent_value
3131        )
3132
3133    def _pick_pcm_bit_depth(
3134        self,
3135        players: Iterable[Player],
3136        streamdetails: StreamDetails | None,
3137        crossfade_enabled: bool,
3138        overlay_active: bool = False,
3139    ) -> tuple[ContentType, int]:
3140        """
3141        Return ``(content_type, bit_depth)`` for an internal PCM stream.
3142
3143        F32 is chosen when audio processing (crossfade, audio overlay, volume
3144        normalization, DSP) will run on the stream — those need the extra
3145        headroom to avoid clipping and precision loss. Otherwise the source's
3146        native bit depth is reused so we don't waste memory upcasting a 16-bit
3147        stream to 32-bit just to pass it through. When the source is unknown
3148        (no streamdetails) we fall back to F32 conservatively.
3149        """
3150        if streamdetails is None:
3151            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3152        needs_headroom = (
3153            crossfade_enabled
3154            or overlay_active
3155            or streamdetails.volume_normalization_mode != VolumeNormalizationMode.DISABLED
3156            or any(self._resolve_player_dsp_config(player).enabled for player in players)
3157        )
3158        if needs_headroom:
3159            return INTERNAL_PCM_FORMAT.content_type, INTERNAL_PCM_FORMAT.bit_depth
3160        bit_depth = streamdetails.audio_format.bit_depth
3161        return ContentType.from_bit_depth(bit_depth), bit_depth
3162
3163    def _select_audio_source_pcm_format(
3164        self,
3165        player: Player,
3166        streamdetails: StreamDetails,
3167        supported_sample_rates: Iterable[int] | None = None,
3168    ) -> AudioFormat:
3169        """
3170        Return a passthrough PCM format for a realtime AudioSource item.
3171
3172        The format matches the source's native sample rate, bit depth and
3173        channel count whenever the player can accept them; if the player does
3174        not support the source's sample rate, it is snapped down to the
3175        closest supported rate. No F32 widening — realtime sources skip every
3176        processing stage that would otherwise need it. Surround sources are
3177        still folded down to stereo, which every output path requires anyway.
3178
3179        :param player: The player requesting the stream.
3180        :param streamdetails: Stream details for the AudioSource item.
3181        :param supported_sample_rates: Rates shared by every output player, if applicable.
3182        """
3183        resolved_sample_rates = (
3184            list(supported_sample_rates)
3185            if supported_sample_rates is not None
3186            else [sample_rate for sample_rate, _ in player.get_supported_sample_rates()]
3187        )
3188        source_rate = streamdetails.audio_format.sample_rate
3189        if source_rate in resolved_sample_rates:
3190            output_sample_rate = source_rate
3191        else:
3192            output_sample_rate = max(
3193                (rate for rate in resolved_sample_rates if rate <= source_rate),
3194                default=min(resolved_sample_rates),
3195            )
3196        bit_depth = streamdetails.audio_format.bit_depth
3197        return AudioFormat(
3198            content_type=ContentType.from_bit_depth(bit_depth),
3199            sample_rate=output_sample_rate,
3200            bit_depth=bit_depth,
3201            # a realtime source may announce more channels than anything downstream can
3202            # carry (a VBAN stream can be configured up to 8), and player handoff formats
3203            # copy this count straight through, so fold it here
3204            channels=min(streamdetails.audio_format.channels, 2),
3205        )
3206
3207    def _flow_restart_context(
3208        self, queue_id: str, protocol_player: Player | None
3209    ) -> tuple[str, list[int]]:
3210        """
3211        Resolve the flow mode config and supported sample rates for restart decisions.
3212
3213        Prefers the protocol player actually consuming the flow stream over the
3214        queue's (wrapper) player, whose config may lack the audio specific entries.
3215        """
3216        if protocol_player is None:
3217            protocol_player = self.mass.players.get_player(queue_id)
3218        if protocol_player is None:
3219            flow_mode_sample_rate_conf = self.mass.config.get_raw_player_config_value(
3220                queue_id, CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3221            )
3222            return flow_mode_sample_rate_conf, []
3223        flow_mode_sample_rate_conf = cast(
3224            "str",
3225            protocol_player.config.get_value(
3226                CONF_FLOW_MODE_SAMPLE_RATE, FLOW_MODE_SAMPLE_RATE_SMART
3227            ),
3228        )
3229        supported_sample_rates = sorted(
3230            {sr for sr, _ in protocol_player.get_supported_sample_rates()}
3231        )
3232        return flow_mode_sample_rate_conf, supported_sample_rates
3233
3234    def _flow_stream_needs_restart(
3235        self,
3236        queue_track: QueueItem,
3237        pcm_format: AudioFormat,
3238        supported_sample_rates: list[int],
3239        flow_mode_sample_rate_conf: str,
3240        is_first_track: bool,
3241    ) -> bool:
3242        """
3243        Return True if the upcoming queue track requires exiting the flow stream.
3244
3245        Covers every case where the flow loop should break and hand control back to
3246        the queue controller for restart:
3247
3248        - Live media (radio, audio sources): cannot be played inside a flow,
3249          the controller will fall back to a single-item stream.
3250        - Sample rate mismatch ('smart' / 'bit_perfect' modes only): the next
3251          track's sample rate (snapped up to the closest supported player rate,
3252          mirroring select_flow_pcm_format's anchoring logic) is incompatible with
3253          the current flow rate, so a new flow must be opened.
3254
3255        The first (anchor) track is always allowed to continue for the sample
3256        rate check; select_flow_pcm_format has already snapped the flow rate to it.
3257
3258        :param queue_track: The upcoming queue item.
3259        :param pcm_format: The current flow stream's PCM format.
3260        :param supported_sample_rates: Sorted list of the player's supported rates.
3261        :param flow_mode_sample_rate_conf: The flow mode sample rate config value.
3262        :param is_first_track: Whether this is the first track of the flow stream.
3263        """
3264        # live audio (radio, plugin or audio source) cannot be flowed; let the
3265        # queue controller fall back to single-item streaming for this item
3266        if queue_track.media_type in (MediaType.RADIO, MediaType.AUDIO_SOURCE):
3267            self.logger.info(
3268                "Live media item %s (%s, %s) encountered in flow stream "
3269                "- breaking out to single item stream",
3270                queue_track.queue_item_id,
3271                queue_track.name,
3272                queue_track.media_type,
3273            )
3274            return True
3275
3276        if is_first_track or queue_track.streamdetails is None:
3277            return False
3278        raw_next_rate = queue_track.streamdetails.audio_format.sample_rate
3279        if not raw_next_rate or not supported_sample_rates:
3280            return False
3281        effective_next_rate = _snap_supported_rate_up(raw_next_rate, supported_sample_rates)
3282
3283        # branch order mirrors select_flow_pcm_format: fixed-rate modes resample
3284        # everything to the chosen rate (no restart); bit_perfect restarts on any
3285        # mismatch; anything else falls through to smart-anchor behavior so
3286        # unknown/legacy config values don't silently pin the flow forever.
3287        if flow_mode_sample_rate_conf in (
3288            FLOW_MODE_SAMPLE_RATE_48000,
3289            FLOW_MODE_SAMPLE_RATE_96000,
3290            FLOW_MODE_SAMPLE_RATE_HIGHEST,
3291        ):
3292            needs_restart = False
3293        elif flow_mode_sample_rate_conf == FLOW_MODE_SAMPLE_RATE_BIT_PERFECT:
3294            needs_restart = effective_next_rate != pcm_format.sample_rate
3295        else:
3296            needs_restart = effective_next_rate > pcm_format.sample_rate
3297
3298        if needs_restart:
3299            self.logger.info(
3300                "Track %s (%s) sample rate %s (snapped to %s) incompatible with flow rate %s "
3301                "(mode: %s) - breaking out to restart flow stream",
3302                queue_track.queue_item_id,
3303                queue_track.name,
3304                raw_next_rate,
3305                effective_next_rate,
3306                pcm_format.sample_rate,
3307                flow_mode_sample_rate_conf,
3308            )
3309        return needs_restart
3310
3311    @asynccontextmanager
3312    async def _connect_radio_stream(self, url: str, **kwargs: Any) -> AsyncGenerator[Any]:
3313        """
3314        Connect to a radio stream URL with fallback for legacy SSL/TLS configurations.
3315
3316        Some radio servers use outdated TLS configurations that reject modern
3317        cipher suites. Since radio streams are public broadcast content,
3318        relaxing cipher requirements is acceptable.
3319
3320        :param url: The radio stream URL to connect to.
3321        :param kwargs: Additional keyword arguments passed to aiohttp get().
3322        """
3323        request_url = encoded_request_url(url)
3324        try:
3325            async with self.mass.http_session_no_ssl.get(request_url, **kwargs) as resp:
3326                yield resp
3327        except ClientConnectorSSLError:
3328            self.logger.info(
3329                "SSL handshake failed for %s, retrying with permissive cipher configuration", url
3330            )
3331            insecure_ssl_context = ssl_util.client_context_no_verify(
3332                ssl_util.SSLCipherList.INSECURE
3333            )
3334            async with self.mass.http_session_no_ssl.get(
3335                request_url, ssl=insecure_ssl_context, **kwargs
3336            ) as resp:
3337                yield resp
3338
3339    async def _update_hls_radio_metadata(
3340        self,
3341        streamdetails: StreamDetails,
3342        elapsed_time: int,
3343    ) -> None:
3344        """
3345        Update HLS radio stream metadata by fetching the playlist.
3346
3347        Fetches the HLS playlist and extracts metadata from EXTINF lines.
3348
3349        :param streamdetails: StreamDetails object to update with metadata
3350        :param elapsed_time: Current playback position in seconds (unused for live radio)
3351        """
3352        mass = self.mass
3353        try:
3354            # Get the actual media playlist URL from cache or resolve it
3355            # We cache the media_playlist_url in streamdetails.data to avoid re-resolving
3356            if streamdetails.data is None:
3357                streamdetails.data = {}
3358            media_playlist_url = streamdetails.data.get("hls_media_playlist_url")
3359            if not media_playlist_url:
3360                try:
3361                    assert isinstance(streamdetails.path, str)  # for type checking
3362                    substream = await self.get_hls_substream(streamdetails.path)
3363                    media_playlist_url = substream.path
3364                    streamdetails.data["hls_media_playlist_url"] = media_playlist_url
3365                except Exception as err:
3366                    self.logger.warning(
3367                        "Failed to resolve HLS substream for metadata monitoring: %s", err
3368                    )
3369                    return
3370
3371            # Fetch the media playlist
3372            timeout = ClientTimeout(total=0, connect=10, sock_read=30)
3373            try:
3374                async with mass.http_session_no_ssl.get(
3375                    encoded_request_url(media_playlist_url), timeout=timeout
3376                ) as resp:
3377                    resp.raise_for_status()
3378                    playlist_content = await resp.text()
3379            except ClientResponseError as err:
3380                # Session token likely expired (410/403) — drop cache so next poll re-resolves
3381                if err.status in (403, 410):
3382                    streamdetails.data.pop("hls_media_playlist_url", None)
3383                raise
3384
3385            # Parse the playlist and look for EXTINF metadata
3386            # The most recent segment usually has the current metadata
3387            lines = playlist_content.strip().split("\n")
3388            for line in reversed(lines):
3389                if line.startswith("#EXTINF:"):
3390                    # Extract metadata from EXTINF line
3391                    metadata = parse_extinf_metadata(line)
3392
3393                    # Build stream title from title and artist
3394                    title = metadata.get("title", "")
3395                    artist = metadata.get("artist", "")
3396                    image_url = (
3397                        metadata.get("image") or metadata.get("artwork") or metadata.get("cover")
3398                    )
3399                    if not artist and " - " in title:
3400                        artist, title = title.split(" - ", 1)
3401                    if title or artist:
3402                        # Format as "Artist - Title"
3403                        if artist and title:
3404                            stream_title = f"{artist} - {title}"
3405                        elif title:
3406                            stream_title = title
3407                        else:
3408                            stream_title = artist
3409
3410                        # Clean the stream title
3411                        cleaned_title = clean_stream_title(stream_title)
3412
3413                        # Only update if changed
3414                        if cleaned_title != streamdetails.stream_title and cleaned_title:
3415                            self.logger.log(
3416                                VERBOSE_LOG_LEVEL, "HLS Radio metadata updated: %s", cleaned_title
3417                            )
3418                            streamdetails.stream_title = cleaned_title
3419                            self._update_radio_stream_metadata(
3420                                streamdetails,
3421                                artist=artist or None,
3422                                title=title or cleaned_title,
3423                                image_url=image_url,
3424                            )
3425
3426                    # Only check the most recent EXTINF
3427                    break
3428
3429        except Exception as err:
3430            self.logger.debug("Error fetching HLS metadata: %s", err)
3431
3432    @staticmethod
3433    def _normalize_reconnecting_urls(url: str | list[MultiPartPath]) -> list[str]:
3434        """Normalize a single URL or a sequence into a non-empty list."""
3435        if isinstance(url, str):
3436            return [url]
3437        if not url:
3438            msg = "Radio stream requires at least one URL"
3439            raise InvalidDataError(msg)
3440        return [part.path for part in url]
3441
3442    async def _resolve_overlay_input(self, queue: PlayerQueue) -> str | None:
3443        """
3444        Resolve the queue's overlay source to a file path or URL for ffmpeg.
3445
3446        Returns None (with a warning logged) when the source can not be resolved,
3447        so the caller can degrade to music-only playback.
3448        """
3449        if not (mapping := queue.overlay_source):
3450            return None
3451        try:
3452            provider = self.mass.get_provider(mapping.provider)
3453            if provider is None:
3454                raise MediaNotFoundError(f"Provider {mapping.provider} is not available")
3455            stream_prov = cast("MusicProvider | PluginProvider", provider)
3456            streamdetails = await stream_prov.get_stream_details(
3457                mapping.item_id, MediaType.SOUND_EFFECT
3458            )
3459        except Exception as err:
3460            self.logger.warning(
3461                "Audio overlay source %s is unavailable (%s) - continuing without overlay",
3462                mapping.uri,
3463                str(err) or err.__class__.__name__,
3464            )
3465            return None
3466        if streamdetails.stream_type not in (StreamType.LOCAL_FILE, StreamType.HTTP) or not (
3467            isinstance(streamdetails.path, str)
3468        ):
3469            self.logger.warning(
3470                "Audio overlay source %s uses unsupported stream type %s "
3471                "- continuing without overlay",
3472                mapping.uri,
3473                streamdetails.stream_type,
3474            )
3475            return None
3476        if streamdetails.stream_type == StreamType.LOCAL_FILE and not await aiofiles.os.path.isfile(
3477            streamdetails.path
3478        ):
3479            # guard against stale sources: feeding a missing file to the mixer would
3480            # kill the whole (music) stream instead of just the overlay
3481            self.logger.warning(
3482                "Audio overlay source %s does not exist - continuing without overlay",
3483                streamdetails.path,
3484            )
3485            return None
3486        return streamdetails.path
3487