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