music-assistant-server

1"""
2Controller to stream audio to players.
3
4The streams controller hosts a basic, unprotected HTTP-only webserver
5purely to stream audio packets to players and some control endpoints such as
6the upnp callbacks and json rpc api for slimproto clients.
7"""
8
9from __future__ import annotations
10
11import asyncio
12import gc
13import logging
14import os
15import urllib.parse
16from collections.abc import AsyncGenerator
17from dataclasses import dataclass
18from typing import TYPE_CHECKING, Final, cast
19
20from aiofiles.os import wrap
21from aiohttp import web
22from music_assistant_models.config_entries import ConfigEntry, ConfigValueOption, ConfigValueType
23from music_assistant_models.enums import (
24    ConfigEntryType,
25    ContentType,
26    MediaType,
27    PlayerFeature,
28    StreamType,
29    VolumeNormalizationMode,
30)
31from music_assistant_models.errors import (
32    AudioError,
33    InvalidDataError,
34    ProviderUnavailableError,
35    QueueEmpty,
36)
37from music_assistant_models.media_items import AudioFormat, Track
38from music_assistant_models.player_queue import PlayLogEntry
39
40from music_assistant.constants import (
41    ANNOUNCE_ALERT_FILE,
42    CONF_BIND_IP,
43    CONF_BIND_PORT,
44    CONF_CROSSFADE_DURATION,
45    CONF_ENTRY_ENABLE_ICY_METADATA,
46    CONF_ENTRY_LOG_LEVEL,
47    CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES,
48    CONF_ENTRY_ZEROCONF_INTERFACES,
49    CONF_HTTP_PROFILE,
50    CONF_OUTPUT_CHANNELS,
51    CONF_OUTPUT_CODEC,
52    CONF_PUBLISH_IP,
53    CONF_SAMPLE_RATES,
54    CONF_SMART_FADES_MODE,
55    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
56    CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
57    CONF_VOLUME_NORMALIZATION_RADIO,
58    CONF_VOLUME_NORMALIZATION_TRACKS,
59    DEFAULT_STREAM_HEADERS,
60    ICY_HEADERS,
61    INTERNAL_PCM_FORMAT,
62    SILENCE_FILE,
63    VERBOSE_LOG_LEVEL,
64)
65from music_assistant.controllers.players.player_controller import AnnounceData
66from music_assistant.controllers.streams.smart_fades import SmartFadesMixer
67from music_assistant.controllers.streams.smart_fades.analyzer import SmartFadesAnalyzer
68from music_assistant.controllers.streams.smart_fades.fades import SMART_CROSSFADE_DURATION
69from music_assistant.helpers.audio import LOGGER as AUDIO_LOGGER
70from music_assistant.helpers.audio import (
71    get_buffered_media_stream,
72    get_chunksize,
73    get_media_stream,
74    get_player_filter_params,
75    get_stream_details,
76    resample_pcm_audio,
77)
78from music_assistant.helpers.buffered_generator import buffered, use_buffer
79from music_assistant.helpers.ffmpeg import LOGGER as FFMPEG_LOGGER
80from music_assistant.helpers.ffmpeg import check_ffmpeg_version, get_ffmpeg_stream
81from music_assistant.helpers.util import (
82    divide_chunks,
83    get_ip_addresses,
84    get_total_system_memory,
85    select_free_port,
86)
87from music_assistant.helpers.webserver import Webserver
88from music_assistant.models.core_controller import CoreController
89from music_assistant.models.music_provider import MusicProvider
90from music_assistant.models.plugin import PluginProvider, PluginSource
91from music_assistant.models.smart_fades import SmartFadesMode
92from music_assistant.providers.universal_group.constants import UGP_PREFIX
93from music_assistant.providers.universal_group.player import UniversalGroupPlayer
94
95if TYPE_CHECKING:
96    from music_assistant_models.config_entries import CoreConfig
97    from music_assistant_models.player import PlayerMedia
98    from music_assistant_models.player_queue import PlayerQueue
99    from music_assistant_models.queue_item import QueueItem
100    from music_assistant_models.streamdetails import StreamDetails
101
102    from music_assistant.mass import MusicAssistant
103    from music_assistant.models.player import Player
104
105
106isfile = wrap(os.path.isfile)
107
108CONF_ALLOW_BUFFER: Final[str] = "allow_buffering"
109CONF_ALLOW_CROSSFADE_SAME_ALBUM: Final[str] = "allow_crossfade_same_album"
110CONF_SMART_FADES_LOG_LEVEL: Final[str] = "smart_fades_log_level"
111
112# Calculate total system memory once at module load time
113TOTAL_SYSTEM_MEMORY_GB: Final[float] = get_total_system_memory()
114CONF_ALLOW_BUFFER_DEFAULT = TOTAL_SYSTEM_MEMORY_GB >= 8.0
115
116
117def parse_pcm_info(content_type: str) -> tuple[int, int, int]:
118    """Parse PCM info from a codec/content_type string."""
119    params = (
120        dict(urllib.parse.parse_qsl(content_type.replace(";", "&"))) if ";" in content_type else {}
121    )
122    sample_rate = int(params.get("rate", 44100))
123    sample_size = int(params.get("bitrate", 16))
124    channels = int(params.get("channels", 2))
125    return (sample_rate, sample_size, channels)
126
127
128@dataclass
129class CrossfadeData:
130    """Data class to hold crossfade data."""
131
132    data: bytes
133    fade_in_size: int
134    pcm_format: AudioFormat  # Format of the 'data' bytes (current/previous track's format)
135    fade_in_pcm_format: AudioFormat  # Format for 'fade_in_size' (next track's format)
136    queue_item_id: str
137
138
139class StreamsController(CoreController):
140    """Webserver Controller to stream audio to players."""
141
142    domain: str = "streams"
143
144    def __init__(self, mass: MusicAssistant) -> None:
145        """Initialize instance."""
146        super().__init__(mass)
147        self._server = Webserver(self.logger, enable_dynamic_routes=True)
148        self.register_dynamic_route = self._server.register_dynamic_route
149        self.unregister_dynamic_route = self._server.unregister_dynamic_route
150        self.manifest.name = "Streamserver"
151        self.manifest.description = (
152            "Music Assistant's core controller that is responsible for "
153            "streaming audio to players on the local network."
154        )
155        self.manifest.icon = "cast-audio"
156        self.announcements: dict[str, AnnounceData] = {}
157        self._crossfade_data: dict[str, CrossfadeData] = {}
158        self._bind_ip: str = "0.0.0.0"
159        self._smart_fades_mixer = SmartFadesMixer(self)
160        self._smart_fades_analyzer = SmartFadesAnalyzer(self)
161
162    @property
163    def base_url(self) -> str:
164        """Return the base_url for the streamserver."""
165        return self._server.base_url
166
167    @property
168    def bind_ip(self) -> str:
169        """Return the IP address this streamserver is bound to."""
170        return self._bind_ip
171
172    @property
173    def smart_fades_mixer(self) -> SmartFadesMixer:
174        """Return the SmartFadesMixer instance."""
175        return self._smart_fades_mixer
176
177    @property
178    def smart_fades_analyzer(self) -> SmartFadesAnalyzer:
179        """Return the SmartFadesAnalyzer instance."""
180        return self._smart_fades_analyzer
181
182    async def get_config_entries(
183        self,
184        action: str | None = None,
185        values: dict[str, ConfigValueType] | None = None,
186    ) -> tuple[ConfigEntry, ...]:
187        """Return all Config Entries for this core module (if any)."""
188        ip_addresses = await get_ip_addresses()
189        default_port = await select_free_port(8097, 9200)
190        return (
191            ConfigEntry(
192                key=CONF_ALLOW_BUFFER,
193                type=ConfigEntryType.BOOLEAN,
194                default_value=CONF_ALLOW_BUFFER_DEFAULT,
195                label="Allow (in-memory) buffering of (track) audio",
196                description="By default, Music Assistant tries to be as resource "
197                "efficient as possible when streaming audio, especially considering "
198                "low-end devices such as Raspberry Pi's. This means that audio "
199                "buffering is disabled by default to reduce memory usage. \n\n"
200                "Enabling this option allows for in-memory buffering of audio, "
201                "which (massively) improves playback (and seeking) performance but it comes "
202                "at the cost of increased memory usage. "
203                "If you run Music Assistant on a capable device with enough memory, "
204                "enabling this option is strongly recommended.",
205                required=False,
206                category="audio",
207            ),
208            ConfigEntry(
209                key=CONF_VOLUME_NORMALIZATION_RADIO,
210                type=ConfigEntryType.STRING,
211                default_value=VolumeNormalizationMode.FALLBACK_FIXED_GAIN,
212                label="Volume normalization method for radio streams",
213                options=[
214                    ConfigValueOption(x.value.replace("_", " ").title(), x.value)
215                    for x in VolumeNormalizationMode
216                ],
217                category="audio",
218            ),
219            ConfigEntry(
220                key=CONF_VOLUME_NORMALIZATION_TRACKS,
221                type=ConfigEntryType.STRING,
222                default_value=VolumeNormalizationMode.FALLBACK_DYNAMIC,
223                label="Volume normalization method for tracks",
224                options=[
225                    ConfigValueOption(x.value.replace("_", " ").title(), x.value)
226                    for x in VolumeNormalizationMode
227                ],
228                category="audio",
229            ),
230            ConfigEntry(
231                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO,
232                type=ConfigEntryType.FLOAT,
233                range=(-20, 10),
234                default_value=-6,
235                label="Fixed/fallback gain adjustment for radio streams",
236                category="audio",
237            ),
238            ConfigEntry(
239                key=CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS,
240                type=ConfigEntryType.FLOAT,
241                range=(-20, 10),
242                default_value=-6,
243                label="Fixed/fallback gain adjustment for tracks",
244                category="audio",
245            ),
246            ConfigEntry(
247                key=CONF_ALLOW_CROSSFADE_SAME_ALBUM,
248                type=ConfigEntryType.BOOLEAN,
249                default_value=False,
250                label="Allow crossfade between tracks from the same album",
251                description="Enabling this option allows for crossfading between tracks "
252                "that are part of the same album.",
253                category="audio",
254            ),
255            ConfigEntry(
256                key=CONF_PUBLISH_IP,
257                type=ConfigEntryType.STRING,
258                default_value=ip_addresses[0],
259                label="Published IP address",
260                description="This IP address is communicated to players where to find this server."
261                "\nMake sure that this IP can be reached by players on the local network, "
262                "otherwise audio streaming will not work.",
263                required=False,
264                category="advanced",
265            ),
266            ConfigEntry(
267                key=CONF_BIND_PORT,
268                type=ConfigEntryType.INTEGER,
269                default_value=default_port,
270                label="TCP Port",
271                description="The TCP port to run the server. "
272                "Make sure that this server can be reached "
273                "on the given IP and TCP port by players on the local network.",
274                category="advanced",
275            ),
276            ConfigEntry(
277                key=CONF_BIND_IP,
278                type=ConfigEntryType.STRING,
279                default_value="0.0.0.0",
280                options=[ConfigValueOption(x, x) for x in {"0.0.0.0", *ip_addresses}],
281                label="Bind to IP/interface",
282                description="Start the stream server on this specific interface. \n"
283                "Use 0.0.0.0 to bind to all interfaces, which is the default. \n"
284                "This is an advanced setting that should normally "
285                "not be adjusted in regular setups.",
286                category="advanced",
287                required=False,
288            ),
289            ConfigEntry(
290                key=CONF_SMART_FADES_LOG_LEVEL,
291                type=ConfigEntryType.STRING,
292                label="Smart Fades Log level",
293                description="Log level for the Smart Fades mixer and analyzer.",
294                options=CONF_ENTRY_LOG_LEVEL.options,
295                default_value="GLOBAL",
296                category="advanced",
297            ),
298            CONF_ENTRY_ZEROCONF_INTERFACES,
299        )
300
301    async def setup(self, config: CoreConfig) -> None:
302        """Async initialize of module."""
303        # copy log level to audio/ffmpeg loggers
304        AUDIO_LOGGER.setLevel(self.logger.level)
305        FFMPEG_LOGGER.setLevel(self.logger.level)
306        self._setup_smart_fades_logger(config)
307        # perform check for ffmpeg version
308        await check_ffmpeg_version()
309        # start the webserver
310        self.publish_port = config.get_value(CONF_BIND_PORT)
311        self.publish_ip = config.get_value(CONF_PUBLISH_IP)
312        self._bind_ip = bind_ip = str(config.get_value(CONF_BIND_IP))
313        # print a big fat message in the log where the streamserver is running
314        # because this is a common source of issues for people with more complex setups
315        self.logger.log(
316            logging.INFO if self.mass.config.onboard_done else logging.WARNING,
317            "\n\n################################################################################\n"
318            "Starting streamserver on  %s:%s\n"
319            "This is the IP address that is communicated to players.\n"
320            "If this is incorrect, audio will not play!\n"
321            "See the documentation how to configure the publish IP for the Streamserver\n"
322            "in Settings --> Core modules --> Streamserver\n"
323            "################################################################################\n",
324            self.publish_ip,
325            self.publish_port,
326        )
327        await self._server.setup(
328            bind_ip=bind_ip,
329            bind_port=cast("int", self.publish_port),
330            base_url=f"http://{self.publish_ip}:{self.publish_port}",
331            static_routes=[
332                (
333                    "*",
334                    "/flow/{session_id}/{queue_id}/{queue_item_id}.{fmt}",
335                    self.serve_queue_flow_stream,
336                ),
337                (
338                    "*",
339                    "/single/{session_id}/{queue_id}/{queue_item_id}.{fmt}",
340                    self.serve_queue_item_stream,
341                ),
342                (
343                    "*",
344                    "/command/{queue_id}/{command}.mp3",
345                    self.serve_command_request,
346                ),
347                (
348                    "*",
349                    "/announcement/{player_id}.{fmt}",
350                    self.serve_announcement_stream,
351                ),
352                (
353                    "*",
354                    "/pluginsource/{plugin_source}/{player_id}.{fmt}",
355                    self.serve_plugin_source_stream,
356                ),
357            ],
358        )
359        # Start periodic garbage collection task
360        # This ensures memory from audio buffers and streams is cleaned up regularly
361        self.mass.call_later(900, self._periodic_garbage_collection)  # 15 minutes
362
363    async def close(self) -> None:
364        """Cleanup on exit."""
365        await self._server.close()
366
367    async def resolve_stream_url(
368        self,
369        session_id: str,
370        queue_item: QueueItem,
371        flow_mode: bool = False,
372        player_id: str | None = None,
373    ) -> str:
374        """Resolve the stream URL for the given QueueItem."""
375        if not player_id:
376            player_id = queue_item.queue_id
377        conf_output_codec = await self.mass.config.get_player_config_value(
378            player_id, CONF_OUTPUT_CODEC, default="flac", return_type=str
379        )
380        output_codec = ContentType.try_parse(conf_output_codec or "flac")
381        fmt = output_codec.value
382        # handle raw pcm without exact format specifiers
383        if output_codec.is_pcm() and ";" not in fmt:
384            fmt += f";codec=pcm;rate={44100};bitrate={16};channels={2}"
385        base_path = "flow" if flow_mode else "single"
386        return f"{self._server.base_url}/{base_path}/{session_id}/{queue_item.queue_id}/{queue_item.queue_item_id}.{fmt}"  # noqa: E501
387
388    async def get_plugin_source_url(
389        self,
390        plugin_source: PluginSource,
391        player_id: str,
392    ) -> str:
393        """Get the url for the Plugin Source stream/proxy."""
394        if plugin_source.audio_format.content_type.is_pcm():
395            fmt = ContentType.WAV.value
396        else:
397            fmt = plugin_source.audio_format.content_type.value
398        return f"{self._server.base_url}/pluginsource/{plugin_source.id}/{player_id}.{fmt}"
399
400    async def serve_queue_item_stream(self, request: web.Request) -> web.StreamResponse:
401        """Stream single queueitem audio to a player."""
402        self._log_request(request)
403        queue_id = request.match_info["queue_id"]
404        queue = self.mass.player_queues.get(queue_id)
405        if not queue:
406            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
407        session_id = request.match_info["session_id"]
408        if queue.session_id and session_id != queue.session_id:
409            raise web.HTTPNotFound(reason=f"Unknown (or invalid) session: {session_id}")
410        queue_player = self.mass.players.get(queue_id)
411        queue_item_id = request.match_info["queue_item_id"]
412        queue_item = self.mass.player_queues.get_item(queue_id, queue_item_id)
413        if not queue_item:
414            raise web.HTTPNotFound(reason=f"Unknown Queue item: {queue_item_id}")
415        if not queue_item.streamdetails:
416            try:
417                queue_item.streamdetails = await get_stream_details(
418                    mass=self.mass, queue_item=queue_item
419                )
420            except Exception as e:
421                self.logger.error(
422                    "Failed to get streamdetails for QueueItem %s: %s", queue_item_id, e
423                )
424                queue_item.available = False
425                raise web.HTTPNotFound(reason=f"No streamdetails for Queue item: {queue_item_id}")
426
427        # pick output format based on the streamdetails and player capabilities
428        if not queue_player:
429            raise web.HTTPNotFound(reason=f"Unknown Player: {queue_id}")
430
431        # work out pcm format based on streamdetails
432        pcm_format = await self._select_pcm_format(
433            player=queue_player,
434            streamdetails=queue_item.streamdetails,
435            smartfades_enabled=True,
436        )
437        output_format = await self.get_output_format(
438            output_format_str=request.match_info["fmt"],
439            player=queue_player,
440            content_sample_rate=pcm_format.sample_rate,
441            content_bit_depth=pcm_format.bit_depth,
442        )
443
444        # prepare request, add some DLNA/UPNP compatible headers
445        headers = {
446            **DEFAULT_STREAM_HEADERS,
447            "icy-name": queue_item.name,
448            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01500000000000000000000000000000",  # noqa: E501
449            "Accept-Ranges": "none",
450            "Content-Type": f"audio/{output_format.output_format_str}",
451        }
452        resp = web.StreamResponse(
453            status=200,
454            reason="OK",
455            headers=headers,
456        )
457        resp.content_type = f"audio/{output_format.output_format_str}"
458        http_profile = await self.mass.config.get_player_config_value(
459            queue_id, CONF_HTTP_PROFILE, default="default", return_type=str
460        )
461        if http_profile == "forced_content_length" and not queue_item.duration:
462            # just set an insane high content length to make sure the player keeps playing
463            resp.content_length = get_chunksize(output_format, 12 * 3600)
464        elif http_profile == "forced_content_length" and queue_item.duration:
465            # guess content length based on duration
466            resp.content_length = get_chunksize(output_format, queue_item.duration)
467        elif http_profile == "chunked":
468            resp.enable_chunked_encoding()
469
470        await resp.prepare(request)
471
472        # return early if this is not a GET request
473        if request.method != "GET":
474            return resp
475
476        if queue_item.media_type != MediaType.TRACK:
477            # no crossfade on non-tracks
478            smart_fades_mode = SmartFadesMode.DISABLED
479        else:
480            smart_fades_mode = await self.mass.config.get_player_config_value(
481                queue.queue_id, CONF_SMART_FADES_MODE, return_type=SmartFadesMode
482            )
483            standard_crossfade_duration = self.mass.config.get_raw_player_config_value(
484                queue.queue_id, CONF_CROSSFADE_DURATION, 10
485            )
486        if (
487            smart_fades_mode != SmartFadesMode.DISABLED
488            and PlayerFeature.GAPLESS_PLAYBACK not in queue_player.supported_features
489        ):
490            # crossfade is not supported on this player due to missing gapless playback
491            self.logger.warning(
492                "Crossfade disabled: Player %s does not support gapless playback, "
493                "consider enabling flow mode to enable crossfade on this player.",
494                queue_player.display_name if queue_player else "Unknown Player",
495            )
496            smart_fades_mode = SmartFadesMode.DISABLED
497
498        if smart_fades_mode != SmartFadesMode.DISABLED:
499            # crossfade is enabled, use special crossfaded single item stream
500            # where the crossfade of the next track is present in the stream of
501            # a single track. This only works if the player supports gapless playback!
502            audio_input = self.get_queue_item_stream_with_smartfade(
503                queue_item=queue_item,
504                pcm_format=pcm_format,
505                smart_fades_mode=smart_fades_mode,
506                standard_crossfade_duration=standard_crossfade_duration,
507            )
508        else:
509            # no crossfade, just a regular single item stream
510            audio_input = self.get_queue_item_stream(
511                queue_item=queue_item,
512                pcm_format=pcm_format,
513                seek_position=queue_item.streamdetails.seek_position,
514            )
515        # stream the audio
516        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
517        # the desired output format for the player including any player specific filter params
518        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
519        if queue_item.media_type == MediaType.RADIO:
520            # keep very short buffer for radio streams
521            # to keep them (more or less) realtime and prevent time outs
522            read_rate_input_args = ["-readrate", "1.0", "-readrate_initial_burst", "2"]
523        else:
524            # just allow the player to buffer whatever it wants for single item streams
525            read_rate_input_args = None
526
527        first_chunk_received = False
528        bytes_sent = 0
529        async for chunk in get_ffmpeg_stream(
530            audio_input=audio_input,
531            input_format=pcm_format,
532            output_format=output_format,
533            filter_params=get_player_filter_params(
534                self.mass,
535                player_id=queue_player.player_id,
536                input_format=pcm_format,
537                output_format=output_format,
538            ),
539            extra_input_args=read_rate_input_args,
540        ):
541            try:
542                await resp.write(chunk)
543                bytes_sent += len(chunk)
544                if not first_chunk_received:
545                    first_chunk_received = True
546                    # inform the queue that the track is now loaded in the buffer
547                    # so for example the next track can be enqueued
548                    self.mass.player_queues.track_loaded_in_buffer(
549                        queue_item.queue_id, queue_item.queue_item_id
550                    )
551            except (BrokenPipeError, ConnectionResetError, ConnectionError) as err:
552                if first_chunk_received and not queue_player.stop_called:
553                    # Player disconnected (unexpected) after receiving at least some data
554                    # This could indicate buffering issues, network problems,
555                    # or player-specific issues
556                    bytes_expected = get_chunksize(output_format, queue_item.duration or 3600)
557                    self.logger.warning(
558                        "Player %s disconnected prematurely from stream for %s (%s) - "
559                        "error: %s, sent %d bytes, expected (approx) bytes=%d",
560                        queue.display_name,
561                        queue_item.name,
562                        queue_item.uri,
563                        err.__class__.__name__,
564                        bytes_sent,
565                        bytes_expected,
566                    )
567                break
568        if queue_item.streamdetails.stream_error:
569            self.logger.error(
570                "Error streaming QueueItem %s (%s) to %s - will try to skip to next item",
571                queue_item.name,
572                queue_item.uri,
573                queue.display_name,
574            )
575            # try to skip to the next item in the queue after a short delay
576            self.mass.call_later(5, self.mass.player_queues.next(queue_id))
577        return resp
578
579    async def serve_queue_flow_stream(self, request: web.Request) -> web.StreamResponse:
580        """Stream Queue Flow audio to player."""
581        self._log_request(request)
582        queue_id = request.match_info["queue_id"]
583        queue = self.mass.player_queues.get(queue_id)
584        if not queue:
585            raise web.HTTPNotFound(reason=f"Unknown Queue: {queue_id}")
586        if not (queue_player := self.mass.players.get(queue_id)):
587            raise web.HTTPNotFound(reason=f"Unknown Player: {queue_id}")
588        start_queue_item_id = request.match_info["queue_item_id"]
589        start_queue_item = self.mass.player_queues.get_item(queue_id, start_queue_item_id)
590        if not start_queue_item:
591            raise web.HTTPNotFound(reason=f"Unknown Queue item: {start_queue_item_id}")
592
593        queue.flow_mode_stream_log = []
594
595        # select the highest possible PCM settings for this player
596        flow_pcm_format = await self._select_flow_format(queue_player)
597
598        # work out output format/details
599        output_format = await self.get_output_format(
600            output_format_str=request.match_info["fmt"],
601            player=queue_player,
602            content_sample_rate=flow_pcm_format.sample_rate,
603            content_bit_depth=flow_pcm_format.bit_depth,
604        )
605        # work out ICY metadata support
606        icy_preference = self.mass.config.get_raw_player_config_value(
607            queue_id,
608            CONF_ENTRY_ENABLE_ICY_METADATA.key,
609            CONF_ENTRY_ENABLE_ICY_METADATA.default_value,
610        )
611        enable_icy = request.headers.get("Icy-MetaData", "") == "1" and icy_preference != "disabled"
612        icy_meta_interval = 256000 if icy_preference == "full" else 16384
613
614        # prepare request, add some DLNA/UPNP compatible headers
615        headers = {
616            **DEFAULT_STREAM_HEADERS,
617            **ICY_HEADERS,
618            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01700000000000000000000000000000",  # noqa: E501
619            "Accept-Ranges": "none",
620            "Content-Type": f"audio/{output_format.output_format_str}",
621        }
622        if enable_icy:
623            headers["icy-metaint"] = str(icy_meta_interval)
624
625        resp = web.StreamResponse(
626            status=200,
627            reason="OK",
628            headers=headers,
629        )
630        http_profile = await self.mass.config.get_player_config_value(
631            queue_id, CONF_HTTP_PROFILE, default="default", return_type=str
632        )
633        if http_profile == "forced_content_length":
634            # just set an insane high content length to make sure the player keeps playing
635            resp.content_length = get_chunksize(output_format, 12 * 3600)
636        elif http_profile == "chunked":
637            resp.enable_chunked_encoding()
638
639        await resp.prepare(request)
640
641        # return early if this is not a GET request
642        if request.method != "GET":
643            return resp
644
645        # all checks passed, start streaming!
646        # this final ffmpeg process in the chain will convert the raw, lossless PCM audio into
647        # the desired output format for the player including any player specific filter params
648        # such as channels mixing, DSP, resampling and, only if needed, encoding to lossy formats
649        self.logger.debug("Start serving Queue flow audio stream for %s", queue.display_name)
650
651        async for chunk in get_ffmpeg_stream(
652            audio_input=self.get_queue_flow_stream(
653                queue=queue,
654                start_queue_item=start_queue_item,
655                pcm_format=flow_pcm_format,
656            ),
657            input_format=flow_pcm_format,
658            output_format=output_format,
659            filter_params=get_player_filter_params(
660                self.mass, queue_player.player_id, flow_pcm_format, output_format
661            ),
662            # we need to slowly feed the music to avoid the player stopping and later
663            # restarting (or completely failing) the audio stream by keeping the buffer short.
664            # this is reported to be an issue especially with Chromecast players.
665            # see for example: https://github.com/music-assistant/support/issues/3717
666            # allow buffer ahead of 6 seconds and read rest in realtime
667            extra_input_args=["-readrate", "1.0", "-readrate_initial_burst", "6"],
668            chunk_size=icy_meta_interval if enable_icy else get_chunksize(output_format),
669        ):
670            try:
671                await resp.write(chunk)
672            except (BrokenPipeError, ConnectionResetError, ConnectionError):
673                # race condition
674                break
675
676            if not enable_icy:
677                continue
678
679            # if icy metadata is enabled, send the icy metadata after the chunk
680            if (
681                # use current item here and not buffered item, otherwise
682                # the icy metadata will be too much ahead
683                (current_item := queue.current_item)
684                and current_item.streamdetails
685                and current_item.streamdetails.stream_title
686            ):
687                title = current_item.streamdetails.stream_title
688            elif queue and current_item and current_item.name:
689                title = current_item.name
690            else:
691                title = "Music Assistant"
692            metadata = f"StreamTitle='{title}';".encode()
693            if icy_preference == "full" and current_item and current_item.image:
694                metadata += f"StreamURL='{current_item.image.path}'".encode()
695            while len(metadata) % 16 != 0:
696                metadata += b"\x00"
697            length = len(metadata)
698            length_b = chr(int(length / 16)).encode()
699            await resp.write(length_b + metadata)
700
701        return resp
702
703    async def serve_command_request(self, request: web.Request) -> web.FileResponse:
704        """Handle special 'command' request for a player."""
705        self._log_request(request)
706        queue_id = request.match_info["queue_id"]
707        command = request.match_info["command"]
708        if command == "next":
709            self.mass.create_task(self.mass.player_queues.next(queue_id))
710        return web.FileResponse(SILENCE_FILE, headers={"icy-name": "Music Assistant"})
711
712    async def serve_announcement_stream(self, request: web.Request) -> web.StreamResponse:
713        """Stream announcement audio to a player."""
714        self._log_request(request)
715        player_id = request.match_info["player_id"]
716        player = self.mass.player_queues.get(player_id)
717        if not player:
718            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
719        if not (announce_data := self.announcements.get(player_id)):
720            raise web.HTTPNotFound(reason=f"No pending announcements for Player: {player_id}")
721
722        # work out output format/details
723        fmt = request.match_info["fmt"]
724        audio_format = AudioFormat(content_type=ContentType.try_parse(fmt))
725
726        http_profile = await self.mass.config.get_player_config_value(
727            player_id, CONF_HTTP_PROFILE, default="default", return_type=str
728        )
729        if http_profile == "forced_content_length":
730            # given the fact that an announcement is just a short audio clip,
731            # just send it over completely at once so we have a fixed content length
732            data = b""
733            async for chunk in self.get_announcement_stream(
734                announcement_url=announce_data["announcement_url"],
735                output_format=audio_format,
736                pre_announce=announce_data["pre_announce"],
737                pre_announce_url=announce_data["pre_announce_url"],
738            ):
739                data += chunk
740            return web.Response(
741                body=data,
742                content_type=f"audio/{audio_format.output_format_str}",
743                headers=DEFAULT_STREAM_HEADERS,
744            )
745
746        resp = web.StreamResponse(
747            status=200,
748            reason="OK",
749            headers=DEFAULT_STREAM_HEADERS,
750        )
751        resp.content_type = f"audio/{audio_format.output_format_str}"
752        if http_profile == "chunked":
753            resp.enable_chunked_encoding()
754
755        await resp.prepare(request)
756
757        # return early if this is not a GET request
758        if request.method != "GET":
759            return resp
760
761        # all checks passed, start streaming!
762        self.logger.debug(
763            "Start serving audio stream for Announcement %s to %s",
764            announce_data["announcement_url"],
765            player.display_name,
766        )
767        async for chunk in self.get_announcement_stream(
768            announcement_url=announce_data["announcement_url"],
769            output_format=audio_format,
770            pre_announce=announce_data["pre_announce"],
771            pre_announce_url=announce_data["pre_announce_url"],
772        ):
773            try:
774                await resp.write(chunk)
775            except (BrokenPipeError, ConnectionResetError):
776                break
777
778        self.logger.debug(
779            "Finished serving audio stream for Announcement %s to %s",
780            announce_data["announcement_url"],
781            player.display_name,
782        )
783
784        return resp
785
786    async def serve_plugin_source_stream(self, request: web.Request) -> web.StreamResponse:
787        """Stream PluginSource audio to a player."""
788        self._log_request(request)
789        plugin_source_id = request.match_info["plugin_source"]
790        provider = cast("PluginProvider", self.mass.get_provider(plugin_source_id))
791        if not provider:
792            raise ProviderUnavailableError(f"Unknown PluginSource: {plugin_source_id}")
793        # work out output format/details
794        player_id = request.match_info["player_id"]
795        player = self.mass.players.get(player_id)
796        if not player:
797            raise web.HTTPNotFound(reason=f"Unknown Player: {player_id}")
798        plugin_source = provider.get_source()
799        output_format = await self.get_output_format(
800            output_format_str=request.match_info["fmt"],
801            player=player,
802            content_sample_rate=plugin_source.audio_format.sample_rate,
803            content_bit_depth=plugin_source.audio_format.bit_depth,
804        )
805        headers = {
806            **DEFAULT_STREAM_HEADERS,
807            "contentFeatures.dlna.org": "DLNA.ORG_OP=01;DLNA.ORG_FLAGS=01700000000000000000000000000000",  # noqa: E501
808            "icy-name": plugin_source.name,
809            "Accept-Ranges": "none",
810            "Content-Type": f"audio/{output_format.output_format_str}",
811        }
812
813        resp = web.StreamResponse(
814            status=200,
815            reason="OK",
816            headers=headers,
817        )
818        resp.content_type = f"audio/{output_format.output_format_str}"
819        http_profile = await self.mass.config.get_player_config_value(
820            player_id, CONF_HTTP_PROFILE, default="default", return_type=str
821        )
822        if http_profile == "forced_content_length":
823            # just set an insanely high content length to make sure the player keeps playing
824            resp.content_length = get_chunksize(output_format, 12 * 3600)
825        elif http_profile == "chunked":
826            resp.enable_chunked_encoding()
827
828        await resp.prepare(request)
829
830        # return early if this is not a GET request
831        if request.method != "GET":
832            return resp
833
834        # all checks passed, start streaming!
835        if not plugin_source.audio_format:
836            raise InvalidDataError(f"No audio format for plugin source {plugin_source_id}")
837        async for chunk in self.get_plugin_source_stream(
838            plugin_source_id=plugin_source_id,
839            output_format=output_format,
840            player_id=player_id,
841            player_filter_params=get_player_filter_params(
842                self.mass, player_id, plugin_source.audio_format, output_format
843            ),
844        ):
845            try:
846                await resp.write(chunk)
847            except (BrokenPipeError, ConnectionResetError, ConnectionError):
848                break
849        return resp
850
851    def get_command_url(self, player_or_queue_id: str, command: str) -> str:
852        """Get the url for the special command stream."""
853        return f"{self.base_url}/command/{player_or_queue_id}/{command}.mp3"
854
855    def get_announcement_url(
856        self,
857        player_id: str,
858        announce_data: AnnounceData,
859        content_type: ContentType = ContentType.MP3,
860    ) -> str:
861        """Get the url for the special announcement stream."""
862        self.announcements[player_id] = announce_data
863        # use stream server to host announcement on local network
864        # this ensures playback on all players, including ones that do not
865        # like https hosts and it also offers the pre-announce 'bell'
866        return f"{self.base_url}/announcement/{player_id}.{content_type.value}"
867
868    def get_stream(
869        self, media: PlayerMedia, pcm_format: AudioFormat, force_flow_mode: bool = False
870    ) -> AsyncGenerator[bytes, None]:
871        """
872        Get a stream of the given media as raw PCM audio.
873
874        This is used as helper for player providers that can consume the raw PCM
875        audio stream directly (e.g. AirPlay) and not rely on HTTP transport.
876        """
877        # select audio source
878        if media.media_type == MediaType.ANNOUNCEMENT:
879            # special case: stream announcement
880            assert media.custom_data
881            audio_source = self.get_announcement_stream(
882                media.custom_data["announcement_url"],
883                output_format=pcm_format,
884                pre_announce=media.custom_data["pre_announce"],
885                pre_announce_url=media.custom_data["pre_announce_url"],
886            )
887        elif media.media_type == MediaType.PLUGIN_SOURCE:
888            # special case: plugin source stream
889            assert media.custom_data
890            audio_source = self.get_plugin_source_stream(
891                plugin_source_id=media.custom_data["source_id"],
892                output_format=pcm_format,
893                # need to pass player_id from the PlayerMedia object
894                # because this could have been a group
895                player_id=media.custom_data["player_id"],
896            )
897        elif (
898            media.media_type == MediaType.FLOW_STREAM
899            and media.source_id
900            and media.source_id.startswith(UGP_PREFIX)
901            and media.uri
902            and "/ugp/" in media.uri
903        ):
904            # special case: member player accessing UGP stream
905            # Check URI to distinguish from the UGP accessing its own stream
906            ugp_player = cast("UniversalGroupPlayer", self.mass.players.get(media.source_id))
907            ugp_stream = ugp_player.stream
908            assert ugp_stream is not None  # for type checker
909            if ugp_stream.base_pcm_format == pcm_format:
910                # no conversion needed
911                audio_source = ugp_stream.subscribe_raw()
912            else:
913                audio_source = ugp_stream.get_stream(output_format=pcm_format)
914        elif (
915            media.source_id
916            and media.queue_item_id
917            and (media.media_type == MediaType.FLOW_STREAM or force_flow_mode)
918        ):
919            # regular queue (flow) stream request
920            queue = self.mass.player_queues.get(media.source_id)
921            assert queue
922            start_queue_item = self.mass.player_queues.get_item(
923                media.source_id, media.queue_item_id
924            )
925            assert start_queue_item
926            audio_source = self.mass.streams.get_queue_flow_stream(
927                queue=queue,
928                start_queue_item=start_queue_item,
929                pcm_format=pcm_format,
930            )
931        elif media.source_id and media.queue_item_id:
932            # single item stream (e.g. radio)
933            queue_item = self.mass.player_queues.get_item(media.source_id, media.queue_item_id)
934            assert queue_item
935            audio_source = buffered(
936                self.get_queue_item_stream(
937                    queue_item=queue_item,
938                    pcm_format=pcm_format,
939                ),
940                buffer_size=10,
941                min_buffer_before_yield=2,
942            )
943        else:
944            # assume url or some other direct path
945            # NOTE: this will fail if its an uri not playable by ffmpeg
946            audio_source = get_ffmpeg_stream(
947                audio_input=media.uri,
948                input_format=AudioFormat(content_type=ContentType.try_parse(media.uri)),
949                output_format=pcm_format,
950            )
951        return audio_source
952
953    @use_buffer(buffer_size=30, min_buffer_before_yield=2)
954    async def get_queue_flow_stream(
955        self,
956        queue: PlayerQueue,
957        start_queue_item: QueueItem,
958        pcm_format: AudioFormat,
959    ) -> AsyncGenerator[bytes, None]:
960        """
961        Get a flow stream of all tracks in the queue as raw PCM audio.
962
963        yields chunks of exactly 1 second of audio in the given pcm_format.
964        """
965        # ruff: noqa: PLR0915
966        assert pcm_format.content_type.is_pcm()
967        queue_track = None
968        last_fadeout_part: bytes = b""
969        last_streamdetails: StreamDetails | None = None
970        last_play_log_entry: PlayLogEntry | None = None
971        queue.flow_mode = True
972        if not start_queue_item:
973            # this can happen in some (edge case) race conditions
974            return
975        pcm_sample_size = pcm_format.pcm_sample_size
976        if start_queue_item.media_type != MediaType.TRACK:
977            # no crossfade on non-tracks
978            smart_fades_mode = SmartFadesMode.DISABLED
979            standard_crossfade_duration = 0
980        else:
981            smart_fades_mode = await self.mass.config.get_player_config_value(
982                queue.queue_id, CONF_SMART_FADES_MODE, return_type=SmartFadesMode
983            )
984            standard_crossfade_duration = self.mass.config.get_raw_player_config_value(
985                queue.queue_id, CONF_CROSSFADE_DURATION, 10
986            )
987        self.logger.info(
988            "Start Queue Flow stream for Queue %s - crossfade: %s %s",
989            queue.display_name,
990            smart_fades_mode,
991            f"({standard_crossfade_duration}s)"
992            if smart_fades_mode == SmartFadesMode.STANDARD_CROSSFADE
993            else "",
994        )
995        total_bytes_sent = 0
996        total_chunks_received = 0
997
998        while True:
999            # get (next) queue item to stream
1000            if queue_track is None:
1001                queue_track = start_queue_item
1002            else:
1003                try:
1004                    queue_track = await self.mass.player_queues.load_next_queue_item(
1005                        queue.queue_id, queue_track.queue_item_id
1006                    )
1007                except QueueEmpty:
1008                    break
1009
1010            if queue_track.streamdetails is None:
1011                raise InvalidDataError(
1012                    "No Streamdetails known for queue item %s",
1013                    queue_track.queue_item_id,
1014                )
1015
1016            self.logger.debug(
1017                "Start Streaming queue track: %s (%s) for queue %s",
1018                queue_track.streamdetails.uri,
1019                queue_track.name,
1020                queue.display_name,
1021            )
1022            # append to play log so the queue controller can work out which track is playing
1023            play_log_entry = PlayLogEntry(queue_track.queue_item_id)
1024            queue.flow_mode_stream_log.append(play_log_entry)
1025            # calculate crossfade buffer size
1026            crossfade_buffer_duration = (
1027                SMART_CROSSFADE_DURATION
1028                if smart_fades_mode == SmartFadesMode.SMART_CROSSFADE
1029                else standard_crossfade_duration
1030            )
1031            crossfade_buffer_duration = min(
1032                crossfade_buffer_duration,
1033                int(queue_track.streamdetails.duration / 2)
1034                if queue_track.streamdetails.duration
1035                else crossfade_buffer_duration,
1036            )
1037            # Ensure crossfade buffer size is aligned to frame boundaries
1038            # Frame size = bytes_per_sample * channels
1039            bytes_per_sample = pcm_format.bit_depth // 8
1040            frame_size = bytes_per_sample * pcm_format.channels
1041            crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1042            # Round down to nearest frame boundary
1043            crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1044
1045            bytes_written = 0
1046            buffer = b""
1047            # handle incoming audio chunks
1048            first_chunk_received = False
1049            # buffer size needs to be big enough to include the crossfade part
1050
1051            async for chunk in self.get_queue_item_stream(
1052                queue_track,
1053                pcm_format=pcm_format,
1054                seek_position=queue_track.streamdetails.seek_position,
1055                raise_on_error=False,
1056            ):
1057                total_chunks_received += 1
1058                if not first_chunk_received:
1059                    first_chunk_received = True
1060                    # inform the queue that the track is now loaded in the buffer
1061                    # so the next track can be preloaded
1062                    self.mass.player_queues.track_loaded_in_buffer(
1063                        queue.queue_id, queue_track.queue_item_id
1064                    )
1065                if total_chunks_received < 10 and smart_fades_mode != SmartFadesMode.DISABLED:
1066                    # we want a stream to start as quickly as possible
1067                    # so for the first 10 chunks we keep a very short buffer
1068                    req_buffer_size = pcm_format.pcm_sample_size
1069                else:
1070                    req_buffer_size = (
1071                        pcm_sample_size
1072                        if smart_fades_mode == SmartFadesMode.DISABLED
1073                        else crossfade_buffer_size
1074                    )
1075
1076                # ALWAYS APPEND CHUNK TO BUFFER
1077                buffer += chunk
1078                del chunk
1079                if len(buffer) < req_buffer_size:
1080                    # buffer is not full enough, move on
1081                    # yield control to event loop with 10ms delay
1082                    await asyncio.sleep(0.01)
1083                    continue
1084
1085                ####  HANDLE CROSSFADE OF PREVIOUS TRACK AND NEW TRACK
1086                if last_fadeout_part and last_streamdetails:
1087                    # perform crossfade
1088                    fadein_part = buffer[:crossfade_buffer_size]
1089                    remaining_bytes = buffer[crossfade_buffer_size:]
1090                    # Use the mixer to handle all crossfade logic
1091                    crossfade_part = await self._smart_fades_mixer.mix(
1092                        fade_in_part=fadein_part,
1093                        fade_out_part=last_fadeout_part,
1094                        fade_in_streamdetails=queue_track.streamdetails,
1095                        fade_out_streamdetails=last_streamdetails,
1096                        pcm_format=pcm_format,
1097                        standard_crossfade_duration=standard_crossfade_duration,
1098                        mode=smart_fades_mode,
1099                    )
1100                    # because the crossfade exists of both the fadein and fadeout part
1101                    # we need to correct the bytes_written accordingly so the duration
1102                    # calculations at the end of the track are correct
1103                    crossfade_part_len = len(crossfade_part)
1104                    bytes_written += int(crossfade_part_len / 2)
1105                    if last_play_log_entry:
1106                        assert last_play_log_entry.seconds_streamed is not None
1107                        last_play_log_entry.seconds_streamed += (
1108                            crossfade_part_len / 2 / pcm_sample_size
1109                        )
1110                    # yield crossfade_part (in pcm_sample_size chunks)
1111                    for _chunk in divide_chunks(crossfade_part, pcm_sample_size):
1112                        yield _chunk
1113                        del _chunk
1114                    del crossfade_part
1115                    # also write the leftover bytes from the crossfade action
1116                    if remaining_bytes:
1117                        yield remaining_bytes
1118                        bytes_written += len(remaining_bytes)
1119                        del remaining_bytes
1120                    # clear vars
1121                    last_fadeout_part = b""
1122                    last_streamdetails = None
1123                    buffer = b""
1124
1125                #### OTHER: enough data in buffer, feed to output
1126                while len(buffer) > req_buffer_size:
1127                    yield buffer[:pcm_sample_size]
1128                    bytes_written += pcm_sample_size
1129                    buffer = buffer[pcm_sample_size:]
1130
1131            #### HANDLE END OF TRACK
1132            if last_fadeout_part:
1133                # edge case: we did not get enough data to make the crossfade
1134                for _chunk in divide_chunks(last_fadeout_part, pcm_sample_size):
1135                    yield _chunk
1136                    del _chunk
1137                bytes_written += len(last_fadeout_part)
1138                last_fadeout_part = b""
1139            if self._crossfade_allowed(
1140                queue_track, smart_fades_mode=smart_fades_mode, flow_mode=True
1141            ):
1142                # if crossfade is enabled, save fadeout part to pickup for next track
1143                last_fadeout_part = buffer[-crossfade_buffer_size:]
1144                last_streamdetails = queue_track.streamdetails
1145                last_play_log_entry = play_log_entry
1146                remaining_bytes = buffer[:-crossfade_buffer_size]
1147                if remaining_bytes:
1148                    yield remaining_bytes
1149                    bytes_written += len(remaining_bytes)
1150                del remaining_bytes
1151            elif buffer:
1152                # no crossfade enabled, just yield the buffer last part
1153                bytes_written += len(buffer)
1154                for _chunk in divide_chunks(buffer, pcm_sample_size):
1155                    yield _chunk
1156                    del _chunk
1157            # make sure the buffer gets cleaned up
1158            del buffer
1159
1160            # update duration details based on the actual pcm data we sent
1161            # this also accounts for crossfade and silence stripping
1162            seconds_streamed = bytes_written / pcm_sample_size
1163            queue_track.streamdetails.seconds_streamed = seconds_streamed
1164            queue_track.streamdetails.duration = int(
1165                queue_track.streamdetails.seek_position + seconds_streamed
1166            )
1167            play_log_entry.seconds_streamed = seconds_streamed
1168            play_log_entry.duration = queue_track.streamdetails.duration
1169            total_bytes_sent += bytes_written
1170            self.logger.debug(
1171                "Finished Streaming queue track: %s (%s) on queue %s",
1172                queue_track.streamdetails.uri,
1173                queue_track.name,
1174                queue.display_name,
1175            )
1176        #### HANDLE END OF QUEUE FLOW STREAM
1177        # end of queue flow: make sure we yield the last_fadeout_part
1178        if last_fadeout_part:
1179            for _chunk in divide_chunks(last_fadeout_part, pcm_sample_size):
1180                yield _chunk
1181                del _chunk
1182            # correct seconds streamed/duration
1183            last_part_seconds = len(last_fadeout_part) / pcm_sample_size
1184            streamdetails = queue_track.streamdetails
1185            assert streamdetails is not None
1186            streamdetails.seconds_streamed = (
1187                streamdetails.seconds_streamed or 0
1188            ) + last_part_seconds
1189            streamdetails.duration = int((streamdetails.duration or 0) + last_part_seconds)
1190            last_fadeout_part = b""
1191        total_bytes_sent += bytes_written
1192        self.logger.info("Finished Queue Flow stream for Queue %s", queue.display_name)
1193
1194    async def get_announcement_stream(
1195        self,
1196        announcement_url: str,
1197        output_format: AudioFormat,
1198        pre_announce: bool | str = False,
1199        pre_announce_url: str = ANNOUNCE_ALERT_FILE,
1200    ) -> AsyncGenerator[bytes, None]:
1201        """Get the special announcement stream."""
1202        announcement_data: asyncio.Queue[bytes | None] = asyncio.Queue(10)
1203        # we are doing announcement in PCM first to avoid multiple encodings
1204        # when mixing pre-announce and announcement
1205        # also we have to deal with some TTS sources being super slow in delivering audio
1206        # so we take an approach where we start fetching the announcement in the background
1207        # while we can already start playing the pre-announce sound (if any)
1208
1209        pcm_format = (
1210            output_format
1211            if output_format.content_type.is_pcm()
1212            else AudioFormat(
1213                sample_rate=output_format.sample_rate,
1214                content_type=ContentType.PCM_S16LE,
1215                bit_depth=16,
1216                channels=output_format.channels,
1217            )
1218        )
1219
1220        async def fetch_announcement() -> None:
1221            fmt = announcement_url.rsplit(".")[-1]
1222            async for chunk in get_ffmpeg_stream(
1223                audio_input=announcement_url,
1224                input_format=AudioFormat(content_type=ContentType.try_parse(fmt)),
1225                output_format=pcm_format,
1226                chunk_size=get_chunksize(pcm_format, 1),
1227            ):
1228                await announcement_data.put(chunk)
1229            await announcement_data.put(None)  # signal end of stream
1230
1231        self.mass.create_task(fetch_announcement())
1232
1233        async def _announcement_stream() -> AsyncGenerator[bytes, None]:
1234            """Generate the PCM audio stream for the announcement + optional pre-announce."""
1235            if pre_announce:
1236                async for chunk in get_ffmpeg_stream(
1237                    audio_input=pre_announce_url,
1238                    input_format=AudioFormat(content_type=ContentType.try_parse(pre_announce_url)),
1239                    output_format=pcm_format,
1240                    chunk_size=get_chunksize(pcm_format, 1),
1241                ):
1242                    yield chunk
1243            # pad silence while we're waiting for the announcement to be ready
1244            while announcement_data.empty():
1245                yield b"\0" * int(
1246                    pcm_format.sample_rate * (pcm_format.bit_depth / 8) * pcm_format.channels * 0.1
1247                )
1248                await asyncio.sleep(0.1)
1249            # stream announcement
1250            while True:
1251                announcement_chunk = await announcement_data.get()
1252                if announcement_chunk is None:
1253                    break
1254                yield announcement_chunk
1255
1256        if output_format == pcm_format:
1257            # no need to re-encode, just yield the raw PCM stream
1258            async for chunk in _announcement_stream():
1259                yield chunk
1260            return
1261
1262        # stream final announcement in requested output format
1263        async for chunk in get_ffmpeg_stream(
1264            audio_input=_announcement_stream(),
1265            input_format=pcm_format,
1266            output_format=output_format,
1267        ):
1268            yield chunk
1269
1270    async def get_plugin_source_stream(
1271        self,
1272        plugin_source_id: str,
1273        output_format: AudioFormat,
1274        player_id: str,
1275        player_filter_params: list[str] | None = None,
1276    ) -> AsyncGenerator[bytes, None]:
1277        """Get the special plugin source stream."""
1278        plugin_prov = cast("PluginProvider", self.mass.get_provider(plugin_source_id))
1279        if not plugin_prov:
1280            raise ProviderUnavailableError(f"Unknown PluginSource: {plugin_source_id}")
1281
1282        plugin_source = plugin_prov.get_source()
1283        self.logger.debug(
1284            "Start streaming PluginSource %s to %s using output format %s",
1285            plugin_source_id,
1286            player_id,
1287            output_format,
1288        )
1289        # this should already be set by the player controller, but just to be sure
1290        plugin_source.in_use_by = player_id
1291
1292        try:
1293            async for chunk in get_ffmpeg_stream(
1294                audio_input=cast(
1295                    "str | AsyncGenerator[bytes, None]",
1296                    plugin_prov.get_audio_stream(player_id)
1297                    if plugin_source.stream_type == StreamType.CUSTOM
1298                    else plugin_source.path,
1299                ),
1300                input_format=plugin_source.audio_format,
1301                output_format=output_format,
1302                filter_params=player_filter_params,
1303                extra_input_args=["-y", "-re"],
1304            ):
1305                if plugin_source.in_use_by != player_id:
1306                    # another player took over or the stream ended, stop streaming
1307                    break
1308                yield chunk
1309        finally:
1310            self.logger.debug(
1311                "Finished streaming PluginSource %s to %s", plugin_source_id, player_id
1312            )
1313            await asyncio.sleep(1)  # prevent race conditions when selecting source
1314            if plugin_source.in_use_by == player_id:
1315                # release control
1316                plugin_source.in_use_by = None
1317
1318    async def get_queue_item_stream(
1319        self,
1320        queue_item: QueueItem,
1321        pcm_format: AudioFormat,
1322        seek_position: int = 0,
1323        raise_on_error: bool = True,
1324    ) -> AsyncGenerator[bytes, None]:
1325        """Get the (PCM) audio stream for a single queue item."""
1326        # collect all arguments for ffmpeg
1327        streamdetails = queue_item.streamdetails
1328        assert streamdetails
1329        filter_params: list[str] = []
1330
1331        # handle volume normalization
1332        gain_correct: float | None = None
1333        if streamdetails.volume_normalization_mode == VolumeNormalizationMode.DYNAMIC:
1334            # volume normalization using loudnorm filter (in dynamic mode)
1335            # which also collects the measurement on the fly during playback
1336            # more info: https://k.ylo.ph/2016/04/04/loudnorm.html
1337            filter_rule = f"loudnorm=I={streamdetails.target_loudness}:TP=-2.0:LRA=10.0:offset=0.0"
1338            filter_rule += ":print_format=json"
1339            filter_params.append(filter_rule)
1340        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.FIXED_GAIN:
1341            # apply user defined fixed volume/gain correction
1342            config_key = (
1343                CONF_VOLUME_NORMALIZATION_FIXED_GAIN_TRACKS
1344                if streamdetails.media_type == MediaType.TRACK
1345                else CONF_VOLUME_NORMALIZATION_FIXED_GAIN_RADIO
1346            )
1347            gain_value = await self.mass.config.get_core_config_value(
1348                self.domain, config_key, default=0.0, return_type=float
1349            )
1350            gain_correct = round(gain_value, 2)
1351            filter_params.append(f"volume={gain_correct}dB")
1352        elif streamdetails.volume_normalization_mode == VolumeNormalizationMode.MEASUREMENT_ONLY:
1353            # volume normalization with known loudness measurement
1354            # apply volume/gain correction
1355            target_loudness = (
1356                float(streamdetails.target_loudness)
1357                if streamdetails.target_loudness is not None
1358                else 0.0
1359            )
1360            if streamdetails.prefer_album_loudness and streamdetails.loudness_album is not None:
1361                gain_correct = target_loudness - float(streamdetails.loudness_album)
1362            elif streamdetails.loudness is not None:
1363                gain_correct = target_loudness - float(streamdetails.loudness)
1364            else:
1365                gain_correct = 0.0
1366            gain_correct = round(gain_correct, 2)
1367            filter_params.append(f"volume={gain_correct}dB")
1368        streamdetails.volume_normalization_gain_correct = gain_correct
1369
1370        allow_buffer = bool(
1371            self.mass.config.get_raw_core_config_value(
1372                self.domain, CONF_ALLOW_BUFFER, CONF_ALLOW_BUFFER_DEFAULT
1373            )
1374            and streamdetails.duration
1375        )
1376
1377        self.logger.debug(
1378            "Starting queue item stream for %s (%s)"
1379            " - using buffer: %s"
1380            " - using fade-in: %s"
1381            " - using volume normalization: %s",
1382            queue_item.name,
1383            streamdetails.uri,
1384            allow_buffer,
1385            streamdetails.fade_in,
1386            streamdetails.volume_normalization_mode,
1387        )
1388        if allow_buffer:
1389            media_stream_gen = get_buffered_media_stream(
1390                self.mass,
1391                streamdetails=streamdetails,
1392                pcm_format=pcm_format,
1393                seek_position=int(seek_position),
1394                filter_params=filter_params,
1395            )
1396        else:
1397            media_stream_gen = get_media_stream(
1398                self.mass,
1399                streamdetails=streamdetails,
1400                pcm_format=pcm_format,
1401                seek_position=int(seek_position),
1402                filter_params=filter_params,
1403            )
1404
1405        first_chunk_received = False
1406        fade_in_buffer = b""
1407        bytes_received = 0
1408        finished = False
1409        stream_started_at = asyncio.get_event_loop().time()
1410        try:
1411            async for chunk in media_stream_gen:
1412                bytes_received += len(chunk)
1413                if not first_chunk_received:
1414                    first_chunk_received = True
1415                    self.logger.debug(
1416                        "First audio chunk received for %s (%s) after %.2f seconds",
1417                        queue_item.name,
1418                        streamdetails.uri,
1419                        asyncio.get_event_loop().time() - stream_started_at,
1420                    )
1421                # handle optional fade-in
1422                if streamdetails.fade_in:
1423                    if len(fade_in_buffer) < pcm_format.pcm_sample_size * 4:
1424                        fade_in_buffer += chunk
1425                    elif fade_in_buffer:
1426                        async for fade_chunk in get_ffmpeg_stream(
1427                            # NOTE: get_ffmpeg_stream signature says str | AsyncGenerator
1428                            # but FFMpeg class actually accepts bytes too. This works at
1429                            # runtime but needs type: ignore for mypy.
1430                            audio_input=fade_in_buffer + chunk,  # type: ignore[arg-type]
1431                            input_format=pcm_format,
1432                            output_format=pcm_format,
1433                            filter_params=["afade=type=in:start_time=0:duration=3"],
1434                        ):
1435                            yield fade_chunk
1436                    fade_in_buffer = b""
1437                    streamdetails.fade_in = False
1438                else:
1439                    yield chunk
1440                # help garbage collection by explicitly deleting chunk
1441                del chunk
1442            finished = True
1443        except AudioError as err:
1444            streamdetails.stream_error = True
1445            queue_item.available = False
1446            if raise_on_error:
1447                raise
1448            # yes, we swallow the error here after logging it
1449            # so the outer stream can handle it gracefully
1450            self.logger.error(
1451                "AudioError while streaming queue item %s (%s): %s",
1452                queue_item.name,
1453                streamdetails.uri,
1454                err,
1455            )
1456        finally:
1457            # determine how many seconds we've streamed
1458            # for pcm output we can calculate this easily
1459            seconds_streamed = bytes_received / pcm_format.pcm_sample_size
1460            streamdetails.seconds_streamed = seconds_streamed
1461            self.logger.debug(
1462                "stream %s for %s in %.2f seconds - seconds streamed/buffered: %.2f",
1463                "aborted" if not finished else "finished",
1464                streamdetails.uri,
1465                asyncio.get_event_loop().time() - stream_started_at,
1466                seconds_streamed,
1467            )
1468            # report stream to provider
1469            if (finished or seconds_streamed >= 90) and (
1470                music_prov := self.mass.get_provider(streamdetails.provider)
1471            ):
1472                if TYPE_CHECKING:  # avoid circular import
1473                    assert isinstance(music_prov, MusicProvider)
1474                self.mass.create_task(music_prov.on_streamed(streamdetails))
1475
1476    @use_buffer(buffer_size=30, min_buffer_before_yield=2)
1477    async def get_queue_item_stream_with_smartfade(
1478        self,
1479        queue_item: QueueItem,
1480        pcm_format: AudioFormat,
1481        smart_fades_mode: SmartFadesMode = SmartFadesMode.SMART_CROSSFADE,
1482        standard_crossfade_duration: int = 10,
1483    ) -> AsyncGenerator[bytes, None]:
1484        """Get the audio stream for a single queue item with (smart) crossfade to the next item."""
1485        queue = self.mass.player_queues.get(queue_item.queue_id)
1486        if not queue:
1487            raise RuntimeError(f"Queue {queue_item.queue_id} not found")
1488
1489        streamdetails = queue_item.streamdetails
1490        assert streamdetails
1491        crossfade_data = self._crossfade_data.pop(queue.queue_id, None)
1492
1493        if crossfade_data and streamdetails.seek_position > 0:
1494            # don't do crossfade when seeking into track
1495            crossfade_data = None
1496        if crossfade_data and (crossfade_data.queue_item_id != queue_item.queue_item_id):
1497            # edge case alert: the next item changed just while we were preloading/crossfading
1498            self.logger.warning(
1499                "Skipping crossfade data for queue %s - next item changed!", queue.display_name
1500            )
1501            crossfade_data = None
1502
1503        self.logger.debug(
1504            "Start Streaming queue track: %s (%s) for queue %s "
1505            "- crossfade mode: %s "
1506            "- crossfading from previous track: %s ",
1507            queue_item.streamdetails.uri if queue_item.streamdetails else "Unknown URI",
1508            queue_item.name,
1509            queue.display_name,
1510            smart_fades_mode,
1511            "true" if crossfade_data else "false",
1512        )
1513
1514        buffer = b""
1515        bytes_written = 0
1516        # calculate crossfade buffer size
1517        crossfade_buffer_duration = (
1518            SMART_CROSSFADE_DURATION
1519            if smart_fades_mode == SmartFadesMode.SMART_CROSSFADE
1520            else standard_crossfade_duration
1521        )
1522        crossfade_buffer_duration = min(
1523            crossfade_buffer_duration,
1524            int(streamdetails.duration / 2)
1525            if streamdetails.duration
1526            else crossfade_buffer_duration,
1527        )
1528        # Ensure crossfade buffer size is aligned to frame boundaries
1529        # Frame size = bytes_per_sample * channels
1530        bytes_per_sample = pcm_format.bit_depth // 8
1531        frame_size = bytes_per_sample * pcm_format.channels
1532        crossfade_buffer_size = int(pcm_format.pcm_sample_size * crossfade_buffer_duration)
1533        # Round down to nearest frame boundary
1534        crossfade_buffer_size = (crossfade_buffer_size // frame_size) * frame_size
1535        fade_out_data: bytes | None = None
1536
1537        if crossfade_data:
1538            # Calculate discard amount in seconds (format-independent)
1539            # Use fade_in_pcm_format because fade_in_size is in the next track's original format
1540            fade_in_duration_seconds = (
1541                crossfade_data.fade_in_size / crossfade_data.fade_in_pcm_format.pcm_sample_size
1542            )
1543            discard_seconds = int(fade_in_duration_seconds) - 1
1544            # Calculate discard amounts in CURRENT track's format
1545            discard_bytes = int(discard_seconds * pcm_format.pcm_sample_size)
1546            # Convert fade_in_size to current track's format for correct leftover calculation
1547            fade_in_size_in_current_format = int(
1548                fade_in_duration_seconds * pcm_format.pcm_sample_size
1549            )
1550            discard_leftover = fade_in_size_in_current_format - discard_bytes
1551        else:
1552            discard_seconds = streamdetails.seek_position
1553            discard_leftover = 0
1554        total_chunks_received = 0
1555        req_buffer_size = crossfade_buffer_size
1556        async for chunk in self.get_queue_item_stream(
1557            queue_item, pcm_format, seek_position=discard_seconds
1558        ):
1559            total_chunks_received += 1
1560            if discard_leftover:
1561                # discard leftover bytes from crossfade data
1562                chunk = chunk[discard_leftover:]  # noqa: PLW2901
1563                discard_leftover = 0
1564
1565            if total_chunks_received < 10:
1566                # we want a stream to start as quickly as possible
1567                # so for the first 10 chunks we keep a very short buffer
1568                req_buffer_size = pcm_format.pcm_sample_size
1569            else:
1570                req_buffer_size = crossfade_buffer_size
1571
1572            # ALWAYS APPEND CHUNK TO BUFFER
1573            buffer += chunk
1574            del chunk
1575            if len(buffer) < req_buffer_size:
1576                # buffer is not full enough, move on
1577                continue
1578
1579            ####  HANDLE CROSSFADE DATA FROM PREVIOUS TRACK
1580            if crossfade_data:
1581                # send the (second half of the) crossfade data
1582                if crossfade_data.pcm_format != pcm_format:
1583                    # edge case: pcm format mismatch, we need to resample
1584                    self.logger.debug(
1585                        "Resampling crossfade data from %s to %s for queue %s",
1586                        crossfade_data.pcm_format.sample_rate,
1587                        pcm_format.sample_rate,
1588                        queue.display_name,
1589                    )
1590                    resampled_data = await resample_pcm_audio(
1591                        crossfade_data.data,
1592                        crossfade_data.pcm_format,
1593                        pcm_format,
1594                    )
1595                    if resampled_data:
1596                        for _chunk in divide_chunks(resampled_data, pcm_format.pcm_sample_size):
1597                            yield _chunk
1598                        bytes_written += len(resampled_data)
1599                    else:
1600                        # Resampling failed, error already logged in resample_pcm_audio
1601                        # Skip crossfade data entirely - stream continues without it
1602                        self.logger.warning(
1603                            "Skipping crossfade data for queue %s due to resampling failure",
1604                            queue.display_name,
1605                        )
1606                else:
1607                    for _chunk in divide_chunks(crossfade_data.data, pcm_format.pcm_sample_size):
1608                        yield _chunk
1609                    bytes_written += len(crossfade_data.data)
1610                # clear vars
1611                crossfade_data = None
1612
1613            #### OTHER: enough data in buffer, feed to output
1614            while len(buffer) > req_buffer_size:
1615                yield buffer[: pcm_format.pcm_sample_size]
1616                bytes_written += pcm_format.pcm_sample_size
1617                buffer = buffer[pcm_format.pcm_sample_size :]
1618
1619        #### HANDLE END OF TRACK
1620
1621        if crossfade_data:
1622            # edge case: we did not get enough data to send the crossfade data
1623            # send the (second half of the) crossfade data
1624            if crossfade_data.pcm_format != pcm_format:
1625                # (yet another) edge case: pcm format mismatch, we need to resample
1626                self.logger.debug(
1627                    "Resampling remaining crossfade data from %s to %s for queue %s",
1628                    crossfade_data.pcm_format.sample_rate,
1629                    pcm_format.sample_rate,
1630                    queue.display_name,
1631                )
1632                resampled_crossfade_data = await resample_pcm_audio(
1633                    crossfade_data.data,
1634                    crossfade_data.pcm_format,
1635                    pcm_format,
1636                )
1637                if resampled_crossfade_data:
1638                    crossfade_data.data = resampled_crossfade_data
1639                else:
1640                    # Resampling failed, error already logged in resample_pcm_audio
1641                    # Skip the crossfade data entirely
1642                    self.logger.warning(
1643                        "Skipping remaining crossfade data for queue %s due to resampling failure",
1644                        queue.display_name,
1645                    )
1646                    crossfade_data = None
1647            if crossfade_data:
1648                for _chunk in divide_chunks(crossfade_data.data, pcm_format.pcm_sample_size):
1649                    yield _chunk
1650                bytes_written += len(crossfade_data.data)
1651                crossfade_data = None
1652
1653        # get next track for crossfade
1654        next_queue_item: QueueItem | None
1655        try:
1656            self.logger.debug(
1657                "Preloading NEXT track for crossfade for queue %s",
1658                queue.display_name,
1659            )
1660            next_queue_item = await self.mass.player_queues.load_next_queue_item(
1661                queue.queue_id, queue_item.queue_item_id
1662            )
1663            # set index_in_buffer to prevent our next track is overwritten while preloading
1664            if next_queue_item.streamdetails is None:
1665                raise InvalidDataError(
1666                    f"No streamdetails for next queue item {next_queue_item.queue_item_id}"
1667                )
1668            queue.index_in_buffer = self.mass.player_queues.index_by_id(
1669                queue.queue_id, next_queue_item.queue_item_id
1670            )
1671            queue_player = self.mass.players.get(queue.queue_id)
1672            assert queue_player is not None
1673            next_queue_item_pcm_format = await self._select_pcm_format(
1674                player=queue_player,
1675                streamdetails=next_queue_item.streamdetails,
1676                smartfades_enabled=True,
1677            )
1678        except QueueEmpty:
1679            # end of queue reached, no next item
1680            next_queue_item = None
1681
1682        if not next_queue_item or not self._crossfade_allowed(
1683            queue_item,
1684            smart_fades_mode=smart_fades_mode,
1685            flow_mode=False,
1686            next_queue_item=next_queue_item,
1687            sample_rate=pcm_format.sample_rate,
1688            next_sample_rate=next_queue_item_pcm_format.sample_rate,
1689        ):
1690            # no crossfade enabled/allowed, just yield the buffer last part
1691            bytes_written += len(buffer)
1692            for _chunk in divide_chunks(buffer, pcm_format.pcm_sample_size):
1693                yield _chunk
1694        else:
1695            # if crossfade is enabled, save fadeout part in buffer to pickup for next track
1696            fade_out_data = buffer
1697            buffer = b""
1698            try:
1699                async for chunk in self.get_queue_item_stream(
1700                    next_queue_item, next_queue_item_pcm_format
1701                ):
1702                    # append to buffer until we reach crossfade size
1703                    # we only need the first X seconds of the NEXT track so we can
1704                    # perform the crossfade.
1705                    # the crossfaded audio of the previous and next track will be
1706                    # sent in two equal parts: first half now, second half
1707                    # when the next track starts. We use CrossfadeData to store
1708                    # the second half to be picked up by the next track's stream generator.
1709                    # Note that we more or less expect the user to have enabled the in-memory
1710                    # buffer so we can keep the next track's audio data in memory.
1711                    buffer += chunk
1712                    del chunk
1713                    if len(buffer) >= crossfade_buffer_size:
1714                        break
1715                ####  HANDLE CROSSFADE OF PREVIOUS TRACK AND NEW TRACK
1716                # Store original buffer size before any resampling for fade_in_size calculation
1717                # This size is in the next track's original format which is what we need
1718                original_buffer_size = len(buffer)
1719                if next_queue_item_pcm_format != pcm_format:
1720                    # edge case: pcm format mismatch, we need to resample the next track's
1721                    # beginning part before crossfading
1722                    self.logger.debug(
1723                        "Resampling next track's crossfade from %s to %s for queue %s",
1724                        next_queue_item_pcm_format.sample_rate,
1725                        pcm_format.sample_rate,
1726                        queue.display_name,
1727                    )
1728                    buffer = await resample_pcm_audio(
1729                        buffer,
1730                        next_queue_item_pcm_format,
1731                        pcm_format,
1732                    )
1733                # perform actual (smart fades) crossfade using mixer
1734                crossfade_bytes = await self._smart_fades_mixer.mix(
1735                    fade_in_part=buffer,
1736                    fade_out_part=fade_out_data,
1737                    fade_in_streamdetails=cast("StreamDetails", next_queue_item.streamdetails),
1738                    fade_out_streamdetails=streamdetails,
1739                    pcm_format=pcm_format,
1740                    standard_crossfade_duration=standard_crossfade_duration,
1741                    mode=smart_fades_mode,
1742                )
1743                # send half of the crossfade_part (= approx the fadeout part)
1744                split_point = (len(crossfade_bytes) + 1) // 2
1745                crossfade_first = crossfade_bytes[:split_point]
1746                crossfade_second = crossfade_bytes[split_point:]
1747                del crossfade_bytes
1748                bytes_written += len(crossfade_first)
1749                for _chunk in divide_chunks(crossfade_first, pcm_format.pcm_sample_size):
1750                    yield _chunk
1751                # store the other half for the next track
1752                # IMPORTANT: crossfade_second data is in CURRENT track's format (pcm_format)
1753                # because it was created from the resampled buffer used for mixing.
1754                # BUT fade_in_size represents bytes in NEXT track's original format
1755                # (next_queue_item_pcm_format) because that's how much of the next track
1756                # was consumed during the crossfade. We need both formats to correctly
1757                # handle the crossfade data when the next track starts.
1758                self._crossfade_data[queue_item.queue_id] = CrossfadeData(
1759                    data=crossfade_second,
1760                    fade_in_size=original_buffer_size,
1761                    pcm_format=pcm_format,  # Format of the data (current track)
1762                    fade_in_pcm_format=next_queue_item_pcm_format,  # Format for fade_in_size
1763                    queue_item_id=next_queue_item.queue_item_id,
1764                )
1765            except AudioError:
1766                # no crossfade possible, just yield the fade_out_data
1767                next_queue_item = None
1768                yield fade_out_data
1769                bytes_written += len(fade_out_data)
1770                del fade_out_data
1771        # make sure the buffer gets cleaned up
1772        del buffer
1773        # update duration details based on the actual pcm data we sent
1774        # this also accounts for crossfade and silence stripping
1775        seconds_streamed = bytes_written / pcm_format.pcm_sample_size
1776        streamdetails.seconds_streamed = seconds_streamed
1777        streamdetails.duration = int(streamdetails.seek_position + seconds_streamed)
1778        self.logger.debug(
1779            "Finished Streaming queue track: %s (%s) on queue %s "
1780            "- crossfade data prepared for next track: %s",
1781            streamdetails.uri,
1782            queue_item.name,
1783            queue.display_name,
1784            next_queue_item.name if next_queue_item else "N/A",
1785        )
1786
1787    def _log_request(self, request: web.Request) -> None:
1788        """Log request."""
1789        if self.logger.isEnabledFor(VERBOSE_LOG_LEVEL):
1790            self.logger.log(
1791                VERBOSE_LOG_LEVEL,
1792                "Got %s request to %s from %s\nheaders: %s\n",
1793                request.method,
1794                request.path,
1795                request.remote,
1796                request.headers,
1797            )
1798        else:
1799            self.logger.debug(
1800                "Got %s request to %s from %s",
1801                request.method,
1802                request.path,
1803                request.remote,
1804            )
1805
1806    async def get_output_format(
1807        self,
1808        output_format_str: str,
1809        player: Player,
1810        content_sample_rate: int,
1811        content_bit_depth: int,
1812    ) -> AudioFormat:
1813        """Parse (player specific) output format details for given format string."""
1814        content_type: ContentType = ContentType.try_parse(output_format_str)
1815        supported_rates_conf = cast(
1816            "list[tuple[str, str]]",
1817            await self.mass.config.get_player_config_value(
1818                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1819            ),
1820        )
1821        output_channels_str = self.mass.config.get_raw_player_config_value(
1822            player.player_id, CONF_OUTPUT_CHANNELS, "stereo"
1823        )
1824        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1825        supported_bit_depths = tuple(int(x[1]) for x in supported_rates_conf)
1826
1827        player_max_bit_depth = max(supported_bit_depths)
1828        output_bit_depth = min(content_bit_depth, player_max_bit_depth)
1829        if content_sample_rate in supported_sample_rates:
1830            output_sample_rate = content_sample_rate
1831        else:
1832            output_sample_rate = max(supported_sample_rates)
1833
1834        if not content_type.is_lossless():
1835            # no point in having a higher bit depth for lossy formats
1836            output_bit_depth = 16
1837            output_sample_rate = min(48000, output_sample_rate)
1838        if output_format_str == "pcm":
1839            content_type = ContentType.from_bit_depth(output_bit_depth)
1840        return AudioFormat(
1841            content_type=content_type,
1842            sample_rate=output_sample_rate,
1843            bit_depth=output_bit_depth,
1844            channels=1 if output_channels_str != "stereo" else 2,
1845        )
1846
1847    async def _select_flow_format(
1848        self,
1849        player: Player,
1850    ) -> AudioFormat:
1851        """Parse (player specific) flow stream PCM format."""
1852        supported_rates_conf = cast(
1853            "list[tuple[str, str]]",
1854            await self.mass.config.get_player_config_value(
1855                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1856            ),
1857        )
1858        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1859        output_sample_rate = INTERNAL_PCM_FORMAT.sample_rate
1860        for sample_rate in (192000, 96000, 48000, 44100):
1861            if sample_rate in supported_sample_rates:
1862                output_sample_rate = sample_rate
1863                break
1864        return AudioFormat(
1865            content_type=INTERNAL_PCM_FORMAT.content_type,
1866            sample_rate=output_sample_rate,
1867            bit_depth=INTERNAL_PCM_FORMAT.bit_depth,
1868            channels=2,
1869        )
1870
1871    async def _select_pcm_format(
1872        self,
1873        player: Player,
1874        streamdetails: StreamDetails,
1875        smartfades_enabled: bool,
1876    ) -> AudioFormat:
1877        """Parse (player specific) stream internal PCM format."""
1878        supported_rates_conf = cast(
1879            "list[tuple[str, str]]",
1880            await self.mass.config.get_player_config_value(
1881                player.player_id, CONF_SAMPLE_RATES, unpack_splitted_values=True
1882            ),
1883        )
1884        supported_sample_rates = tuple(int(x[0]) for x in supported_rates_conf)
1885        # use highest supported rate within content rate
1886        output_sample_rate = max(
1887            (r for r in supported_sample_rates if r <= streamdetails.audio_format.sample_rate),
1888            default=48000,  # sane/safe default
1889        )
1890        # work out pcm format based on streamdetails
1891        pcm_format = AudioFormat(
1892            sample_rate=output_sample_rate,
1893            # always use f32 internally for extra headroom for filters etc
1894            content_type=INTERNAL_PCM_FORMAT.content_type,
1895            bit_depth=INTERNAL_PCM_FORMAT.bit_depth,
1896            channels=streamdetails.audio_format.channels,
1897        )
1898        if smartfades_enabled:
1899            pcm_format.channels = 2  # force stereo for crossfading
1900
1901        return pcm_format
1902
1903    def _crossfade_allowed(
1904        self,
1905        queue_item: QueueItem,
1906        smart_fades_mode: SmartFadesMode,
1907        flow_mode: bool = False,
1908        next_queue_item: QueueItem | None = None,
1909        sample_rate: int | None = None,
1910        next_sample_rate: int | None = None,
1911    ) -> bool:
1912        """Get the crossfade config for a queue item."""
1913        if smart_fades_mode == SmartFadesMode.DISABLED:
1914            return False
1915        if not (self.mass.players.get(queue_item.queue_id)):
1916            return False  # just a guard
1917        if queue_item.media_type != MediaType.TRACK:
1918            self.logger.debug("Skipping crossfade: current item is not a track")
1919            return False
1920        # check if the next item is part of the same album
1921        next_item = next_queue_item or self.mass.player_queues.get_next_item(
1922            queue_item.queue_id, queue_item.queue_item_id
1923        )
1924        if not next_item:
1925            # there is no next item!
1926            return False
1927        # check if next item is a track
1928        if next_item.media_type != MediaType.TRACK:
1929            self.logger.debug("Skipping crossfade: next item is not a track")
1930            return False
1931        if (
1932            isinstance(queue_item.media_item, Track)
1933            and isinstance(next_item.media_item, Track)
1934            and queue_item.media_item.album
1935            and next_item.media_item.album
1936            and queue_item.media_item.album == next_item.media_item.album
1937            and not self.mass.config.get_raw_core_config_value(
1938                self.domain, CONF_ALLOW_CROSSFADE_SAME_ALBUM, False
1939            )
1940        ):
1941            # in general, crossfade is not desired for tracks of the same (gapless) album
1942            # because we have no accurate way to determine if the album is gapless or not,
1943            # for now we just never crossfade between tracks of the same album
1944            self.logger.debug("Skipping crossfade: next item is part of the same album")
1945            return False
1946
1947        # check if we're allowed to crossfade on different sample rates
1948        if (
1949            not flow_mode
1950            and sample_rate
1951            and next_sample_rate
1952            and sample_rate != next_sample_rate
1953            and not self.mass.config.get_raw_player_config_value(
1954                queue_item.queue_id,
1955                CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES.key,
1956                CONF_ENTRY_SUPPORT_GAPLESS_DIFFERENT_SAMPLE_RATES.default_value,
1957            )
1958        ):
1959            self.logger.debug(
1960                "Skipping crossfade: player does not support gapless playback "
1961                "with different sample rates (%s vs %s)",
1962                sample_rate,
1963                next_sample_rate,
1964            )
1965            return False
1966
1967        return True
1968
1969    async def _periodic_garbage_collection(self) -> None:
1970        """Periodic garbage collection to free up memory from audio buffers and streams."""
1971        self.logger.log(
1972            VERBOSE_LOG_LEVEL,
1973            "Running periodic garbage collection...",
1974        )
1975        # Run garbage collection in executor to avoid blocking the event loop
1976        # Since this runs periodically (not in response to subprocess cleanup),
1977        # it's safe to run in a thread without causing thread-safety issues
1978        loop = asyncio.get_running_loop()
1979        collected = await loop.run_in_executor(None, gc.collect)
1980        self.logger.log(
1981            VERBOSE_LOG_LEVEL,
1982            "Garbage collection completed, collected %d objects",
1983            collected,
1984        )
1985        # Schedule next run in 15 minutes
1986        self.mass.call_later(900, self._periodic_garbage_collection)
1987
1988    def _setup_smart_fades_logger(self, config: CoreConfig) -> None:
1989        """Set up smart fades logger level."""
1990        log_level = str(config.get_value(CONF_SMART_FADES_LOG_LEVEL))
1991        if log_level == "GLOBAL":
1992            self.smart_fades_analyzer.logger.setLevel(self.logger.level)
1993            self.smart_fades_mixer.logger.setLevel(self.logger.level)
1994        else:
1995            self.smart_fades_analyzer.logger.setLevel(log_level)
1996            self.smart_fades_mixer.logger.setLevel(log_level)
1997