music-assistant-server

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