music-assistant-server

1"""Various helpers for audio streaming and manipulation."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import re
8import struct
9import urllib.parse
10from collections.abc import AsyncGenerator, Iterable, Iterator
11from contextlib import aclosing
12from io import BytesIO
13from typing import TYPE_CHECKING, Final
14
15from music_assistant_models.enums import (
16    ContentType,
17    MediaType,
18    PlayerFeature,
19    PlayerType,
20    VolumeNormalizationMode,
21)
22from music_assistant_models.errors import InvalidDataError
23from music_assistant_models.streamdetails import MultiPartPath
24
25from music_assistant.constants import (
26    MASS_LOGGER_NAME,
27    VERBOSE_LOG_LEVEL,
28)
29from music_assistant.helpers.json import JSON_DECODE_EXCEPTIONS, json_loads
30
31from .ffmpeg import DEFAULT_MP3_BIT_RATE, get_ffmpeg_stream
32from .process import AsyncProcess, communicate
33
34if TYPE_CHECKING:
35    from music_assistant_models.media_items import AudioFormat
36    from music_assistant_models.streamdetails import StreamDetails
37
38    from music_assistant.mass import MusicAssistant
39    from music_assistant.models.player import Player
40
41LOGGER = logging.getLogger(f"{MASS_LOGGER_NAME}.helpers.audio")
42
43HTTP_HEADERS = {"User-Agent": "Lavf/60.16.100.MusicAssistant"}
44HTTP_HEADERS_ICY = {**HTTP_HEADERS, "Icy-MetaData": "1"}
45
46SLOW_PROVIDERS = ("tidal", "ytmusic", "apple_music")
47
48# Mapping of audio format identifiers to their correct IANA MIME types
49# where the format name differs from the MIME subtype.
50# Strict DLNA/UPnP devices reject non-standard MIME types (e.g. audio/mp3).
51_MIME_TYPE_OVERRIDES: Final[dict[str, str]] = {
52    "mp3": "audio/mpeg",
53}
54
55
56def get_mime_type(format_str: str) -> str:
57    """
58    Get the proper IANA MIME type for a given audio format string.
59
60    :param format_str: The audio format string (e.g. "mp3", "flac",
61        "pcm;codec=pcm;rate=44100;bitrate=16;channels=2").
62    """
63    base_format = format_str.split(";", maxsplit=1)[0]
64    if override := _MIME_TYPE_OVERRIDES.get(base_format):
65        return override
66    return f"audio/{format_str}"
67
68
69def parse_pcm_info(content_type: str) -> tuple[int, int, int]:
70    """
71    Parse PCM info from a codec/content_type string.
72
73    :param content_type: Content type string like "pcm;codec=pcm;rate=44100;bitrate=16;channels=2".
74    """
75    params = (
76        dict(urllib.parse.parse_qsl(content_type.replace(";", "&"))) if ";" in content_type else {}
77    )
78    sample_rate = int(params.get("rate", 44100))
79    sample_size = int(params.get("bitrate", 16))
80    channels = int(params.get("channels", 2))
81    return (sample_rate, sample_size, channels)
82
83
84CACHE_CATEGORY_RESOLVED_RADIO_URL: Final[int] = 100
85CACHE_PROVIDER: Final[str] = "audio"
86
87
88def iter_pcm_slices(
89    audio: bytes,
90    pcm_format: AudioFormat,
91    target_duration_ms: int = 100,
92) -> Iterator[bytes]:
93    """
94    Yield frame-aligned PCM slices of approximately ``target_duration_ms``.
95
96    Large PCM buffers (e.g. crossfade segments or full-track reads) are split
97    into fixed-size sub-chunks so that downstream consumers get predictable
98    chunk sizes for buffering, write-timeout management, and ring-buffer
99    bookkeeping.
100
101    :param audio: Raw PCM bytes to slice.
102    :param pcm_format: Format description (sample rate, bit depth, channels).
103    :param target_duration_ms: Desired slice length in milliseconds (default 100).
104    """
105    if not audio:
106        return
107    bytes_per_sample = max(1, pcm_format.bit_depth // 8)
108    frame_size = bytes_per_sample * pcm_format.channels
109    if frame_size <= 0:
110        yield audio
111        return
112    samples_per_slice = max(1, round((target_duration_ms / 1000) * pcm_format.sample_rate))
113    slice_size = max(frame_size, samples_per_slice * frame_size)
114    offset = 0
115    audio_len = len(audio)
116    while offset < audio_len:
117        end = min(audio_len, offset + slice_size)
118        # Align to frame boundary unless this is the tail of the buffer.
119        if end < audio_len:
120            aligned_end = end - (end % frame_size)
121            if aligned_end <= offset:
122                aligned_end = min(audio_len, offset + frame_size)
123            end = aligned_end
124        yield audio[offset:end]
125        offset = end
126
127
128def align_audio_to_frame_boundary(audio_data: bytes, pcm_format: AudioFormat) -> bytes:
129    """
130    Align audio data to frame boundaries by truncating incomplete frames.
131
132    :param audio_data: Raw PCM audio data to align.
133    :param pcm_format: AudioFormat of the audio data.
134    """
135    bytes_per_sample = pcm_format.bit_depth // 8
136    frame_size = bytes_per_sample * pcm_format.channels
137    valid_bytes = (len(audio_data) // frame_size) * frame_size
138    if valid_bytes != len(audio_data):
139        LOGGER.debug(
140            "Truncating %d bytes from audio buffer to align to frame boundary",
141            len(audio_data) - valid_bytes,
142        )
143        return audio_data[:valid_bytes]
144    return audio_data
145
146
147async def strip_silence(
148    audio_data: bytes,
149    pcm_format: AudioFormat,
150    reverse: bool = False,
151) -> bytes:
152    """
153    Strip silence from begin or end of pcm audio using ffmpeg.
154
155    :param audio_data: Raw PCM audio data.
156    :param pcm_format: AudioFormat of the audio data.
157    :param reverse: If True, strip from end instead of beginning.
158    """
159    args = ["ffmpeg", "-hide_banner", "-loglevel", "quiet"]
160    args += [
161        "-acodec",
162        pcm_format.content_type.name.lower(),
163        "-f",
164        pcm_format.content_type.value,
165        "-ac",
166        str(pcm_format.channels),
167        "-ar",
168        str(pcm_format.sample_rate),
169        "-i",
170        "-",
171    ]
172    if reverse:
173        args += [
174            "-af",
175            "areverse,atrim=start=0.2,silenceremove=start_periods=1"
176            ":start_silence=0.1:start_threshold=0.02,areverse",
177        ]
178    else:
179        args += [
180            "-af",
181            "atrim=start=0.2,silenceremove=start_periods=1:start_silence=0.1:start_threshold=0.02",
182        ]
183    args += ["-f", pcm_format.content_type.value, "-"]
184    _returncode, stripped_data, _stderr = await communicate(args, audio_data)
185
186    bytes_stripped = len(audio_data) - len(stripped_data)
187    if LOGGER.isEnabledFor(VERBOSE_LOG_LEVEL):
188        seconds_stripped = round(bytes_stripped / pcm_format.pcm_sample_size, 2)
189        location = "end" if reverse else "begin"
190        LOGGER.log(
191            VERBOSE_LOG_LEVEL,
192            "stripped %s seconds of silence from %s of pcm audio. bytes stripped: %s",
193            seconds_stripped,
194            location,
195            bytes_stripped,
196        )
197    return stripped_data
198
199
200def create_wave_header(
201    samplerate: int = 44100,
202    channels: int = 2,
203    bitspersample: int = 16,
204    duration: int | None = None,
205) -> bytes:
206    """Generate a wave header from given params."""
207    file = BytesIO()
208
209    # Generate format chunk
210    format_chunk_spec = b"<4sLHHLLHH"
211    format_chunk = struct.pack(
212        format_chunk_spec,
213        b"fmt ",  # Chunk id
214        16,  # Size of this chunk (excluding chunk id and this field)
215        1,  # Audio format, 1 for PCM
216        channels,  # Number of channels
217        int(samplerate),  # Samplerate, 44100, 48000, etc.
218        int(samplerate * channels * (bitspersample / 8)),  # Byterate
219        int(channels * (bitspersample / 8)),  # Blockalign
220        bitspersample,  # 16 bits for two byte samples, etc.
221    )
222    # Generate data chunk
223    # duration = 3600*6.7
224    data_chunk_spec = b"<4sL"
225    if duration is None:
226        # use max value possible
227        datasize = 4254768000  # = 6,7 hours at 44100/16
228    else:
229        # calculate from duration
230        numsamples = samplerate * duration
231        datasize = int(numsamples * channels * (bitspersample / 8))
232    data_chunk = struct.pack(
233        data_chunk_spec,
234        b"data",  # Chunk id
235        int(datasize),  # Chunk size (excluding chunk id and this field)
236    )
237    sum_items = [
238        # "WAVE" string following size field
239        4,
240        # "fmt " + chunk size field + chunk size
241        struct.calcsize(format_chunk_spec),
242        # Size of data chunk spec + data size
243        struct.calcsize(data_chunk_spec) + datasize,
244    ]
245    # Generate main header
246    all_chunks_size = int(sum(sum_items))
247    main_header_spec = b"<4sL4s"
248    main_header = struct.pack(main_header_spec, b"RIFF", all_chunks_size, b"WAVE")
249    # Write all the contents in
250    file.write(main_header)
251    file.write(format_chunk)
252    file.write(data_chunk)
253
254    # return file.getvalue(), all_chunks_size + 8
255    return file.getvalue()
256
257
258def create_streaming_wave_header(audio_format: AudioFormat) -> bytes:
259    """
260    Generate a wave header for a stream whose length is not known up front.
261
262    :param audio_format: The PCM format the audio behind the header is in.
263    """
264    channels = audio_format.channels
265    sample_rate = audio_format.sample_rate
266    bits_per_sample = audio_format.bit_depth
267    byte_rate = sample_rate * channels * (bits_per_sample // 8)
268    block_align = channels * (bits_per_sample // 8)
269    # RIFF size & data size both set to 0xFFFFFFFF so clients honoring the WAV
270    # length fields don't cut the stream off (create_wave_header hardcodes ~6.7h).
271    return (
272        b"RIFF"
273        + struct.pack("<L", 0xFFFFFFFF)
274        + b"WAVE"
275        + b"fmt "
276        + struct.pack(
277            "<LHHLLHH", 16, 1, channels, sample_rate, byte_rate, block_align, bits_per_sample
278        )
279        + b"data"
280        + struct.pack("<L", 0xFFFFFFFF)
281    )
282
283
284def parse_extinf_metadata(extinf_line: str) -> dict[str, str]:
285    """
286    Parse metadata from HLS EXTINF line.
287
288    Extracts structured metadata like title="...", artist="..." from EXTINF lines.
289    Common in iHeartRadio and other commercial radio HLS streams.
290
291    :param extinf_line: The EXTINF line containing metadata
292    """
293    metadata = {}
294
295    # Pattern to match key="value" pairs in the EXTINF line
296    # Handles nested quotes by matching everything until the closing quote
297    pattern = r'(\w+)="([^"]*)"'
298
299    matches = re.findall(pattern, extinf_line)
300    for key, value in matches:
301        metadata[key.lower()] = value
302
303    # Fallback: RFC 8216 plain title format `#EXTINF:<duration>,<title>`
304    if not metadata and "," in extinf_line:
305        title = extinf_line.split(",", 1)[1].strip()
306        if title:
307            metadata["title"] = title
308
309    return metadata
310
311
312def get_parts_from_position(
313    parts: list[MultiPartPath],
314    seek_position: int,
315) -> tuple[list[MultiPartPath], int]:
316    """
317    Get the remaining parts list from a timestamp.
318
319    Arguments:
320    parts: The list of  parts
321    seek_position: The seeking position in seconds of the tracklist
322
323    Returns:
324        In a tuple, A list of  parts, starting with the one at the requested
325        seek position and the position in seconds to seek to in the first
326        track.
327    """
328    skipped_duration = 0.0
329    for i, part in enumerate(parts):
330        if not isinstance(part, MultiPartPath):
331            raise InvalidDataError("Multi-file streamdetails requires a list of MultiPartPath")
332        if part.duration is None:
333            return parts, seek_position
334        if skipped_duration + part.duration < seek_position:
335            skipped_duration += part.duration
336            continue
337
338        position = seek_position - skipped_duration
339
340        # Seeking in some parts is inaccurate, making the seek to a chapter land on the end of
341        # the previous track. If we're within 2 second of the end, skip the current track
342        if position + 2 >= part.duration:
343            LOGGER.debug(
344                f"Skipping to the next part due to seek position being at the end: {position}",
345            )
346            if i + 1 < len(parts):
347                return parts[i + 1 :], 0
348            return parts[i:], int(position)  # last part, cannot skip
349
350        return parts[i:], int(position)
351
352    raise IndexError(f"Could not find any candidate part for position {seek_position}")
353
354
355def build_concat_filelist(paths: list[str]) -> str:
356    """
357    Build the file list content for ffmpeg's concat demuxer.
358
359    :param paths: The file paths to include, in playback order.
360    """
361    lines = []
362    for path in paths:
363        # The concat demuxer uses single quotes as delimiters, so a literal quote in the
364        # path must be written as '\'' to prevent the path being truncated at the quote.
365        escaped_path = path.replace("'", "'\\''")
366        lines.append(f"file '{escaped_path}'\n")
367    return "".join(lines)
368
369
370async def realtime_pcm_pacer(
371    inner: AsyncGenerator[bytes],
372    pcm_format: AudioFormat,
373) -> AsyncGenerator[bytes]:
374    """
375    Pace a PCM byte stream at the format's native rate.
376
377    Useful for live AudioSource streams whose producer is not realtime-paced
378    (e.g. librespot's pipe backend) — without rate-limiting the consumer would
379    buffer many seconds of audio ahead of playback, making skip/next laggy.
380
381    :param inner: Source generator yielding raw PCM bytes.
382    :param pcm_format: PCM format the inner generator emits.
383    """
384    bytes_per_second = pcm_format.sample_rate * pcm_format.channels * (pcm_format.bit_depth // 8)
385    if bytes_per_second <= 0 or not pcm_format.content_type.is_pcm():
386        # non-PCM or malformed format: pass through unchanged
387        async for chunk in inner:
388            yield chunk
389        return
390    loop = asyncio.get_running_loop()
391    start_time = loop.time()
392    total_bytes = 0
393    async for chunk in inner:
394        yield chunk
395        total_bytes += len(chunk)
396        expected_elapsed = total_bytes / bytes_per_second
397        actual_elapsed = loop.time() - start_time
398        if actual_elapsed < expected_elapsed:
399            await asyncio.sleep(expected_elapsed - actual_elapsed)
400
401
402async def audio_source_silence_keepalive(
403    inner: AsyncGenerator[bytes],
404    pcm_format: AudioFormat,
405    silence_chunk_ms: int = 100,
406    idle_threshold_s: float | None = None,
407) -> AsyncGenerator[bytes]:
408    """
409    Wrap a live AudioSource PCM stream and emit silence during idle gaps.
410
411    Plugin providers exposing an AudioSource may stop yielding bytes while the
412    upstream device is paused (e.g. user paused in the Spotify app). Without
413    bytes flowing the downstream consumer (ffmpeg / the player) may disconnect.
414    This wrapper inserts ``silence_chunk_ms`` worth of zero bytes whenever the
415    inner generator hasn't produced for ``idle_threshold_s`` seconds, while
416    relaying real bytes immediately when they arrive.
417
418    Only meaningful for PCM streams — injecting raw zero bytes into a compressed
419    stream (MP3/AAC/etc.) would corrupt the bitstream. For non-PCM ``pcm_format``
420    inputs the wrapper degrades to a transparent pass-through.
421
422    :param inner: The underlying async generator yielding raw PCM bytes.
423    :param pcm_format: PCM format the inner generator emits (used to size the
424        silence chunk so it lines up to a frame boundary).
425    :param silence_chunk_ms: Duration of each silence chunk in milliseconds.
426    :param idle_threshold_s: Seconds without input before silence is inserted.
427        Defaults to the chunk duration so silence flows at realtime — critical
428        for keeping HTTP consumers (Sonos, Chromecast) connected.
429    """
430    if idle_threshold_s is None:
431        idle_threshold_s = silence_chunk_ms / 1000
432    frame_size = pcm_format.channels * (pcm_format.bit_depth // 8)
433    bytes_per_second = (
434        pcm_format.sample_rate * frame_size if pcm_format.content_type.is_pcm() else 0
435    )
436    if bytes_per_second <= 0 or frame_size <= 0:
437        # non-PCM or malformed format: pass through unchanged, no silence injection
438        async for chunk in inner:
439            yield chunk
440        return
441
442    # Round the silence chunk size DOWN to a whole-frame multiple so emitted
443    # chunks line up to PCM frame boundaries for arbitrary silence_chunk_ms /
444    # sample-rate combinations.
445    raw_silence_bytes = bytes_per_second * silence_chunk_ms // 1000
446    silence_bytes = max(frame_size, (raw_silence_bytes // frame_size) * frame_size)
447    silence_chunk = b"\x00" * silence_bytes
448    # empty bytes is the end-of-stream sentinel; real PCM frames are never empty
449    queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=8)
450
451    async def _producer() -> None:
452        # aclosing ensures inner.aclose() runs on cancellation so the underlying
453        # generator's own finally (e.g. plugin lock release, fd cleanup) fires
454        # instead of leaking until GC.
455        try:
456            async with aclosing(inner) as managed_inner:
457                async for chunk in managed_inner:
458                    await queue.put(chunk)
459        finally:
460            await queue.put(b"")
461
462    producer_task = asyncio.create_task(_producer())
463    try:
464        while True:
465            try:
466                chunk = await asyncio.wait_for(queue.get(), timeout=idle_threshold_s)
467            except TimeoutError:
468                yield silence_chunk
469                continue
470            if not chunk:
471                break
472            yield chunk
473    finally:
474        producer_task.cancel()
475        try:
476            await producer_task
477        except asyncio.CancelledError:
478            pass
479        except Exception:
480            # log but don't re-raise: we're already in a finally and the
481            # downstream consumer has its own error handling for the outer stream.
482            LOGGER.exception("AudioSource producer task raised")
483
484
485async def get_silence(
486    duration: int,
487    output_format: AudioFormat,
488) -> AsyncGenerator[bytes]:
489    """Create stream of silence, encoded to format of choice."""
490    if output_format.content_type.is_pcm():
491        # pcm = just zeros
492        for _ in range(duration):
493            yield b"\0" * int(output_format.sample_rate * (output_format.bit_depth / 8) * 2)
494        return
495    if output_format.content_type == ContentType.WAV:
496        # wav silence = wave header + zero's
497        yield create_wave_header(
498            samplerate=output_format.sample_rate,
499            channels=2,
500            bitspersample=output_format.bit_depth,
501            duration=duration,
502        )
503        for _ in range(duration):
504            yield b"\0" * int(output_format.sample_rate * (output_format.bit_depth / 8) * 2)
505        return
506    # use ffmpeg for all other encodings
507    args = [
508        "ffmpeg",
509        "-hide_banner",
510        "-loglevel",
511        "quiet",
512        "-f",
513        "lavfi",
514        "-i",
515        f"anullsrc=r={output_format.sample_rate}:cl={'stereo'}",
516        "-t",
517        str(duration),
518        "-f",
519        output_format.output_format_str,
520        "-",
521    ]
522    async with AsyncProcess(args, stdout=True) as ffmpeg_proc:
523        async for chunk in ffmpeg_proc.iter_chunked():
524            yield chunk
525
526
527async def resample_pcm_audio(
528    input_audio: bytes | AsyncGenerator[bytes],
529    input_format: AudioFormat,
530    output_format: AudioFormat,
531    chunk_size: int | None = None,
532) -> AsyncGenerator[bytes]:
533    """
534    Resample PCM audio from input_format to output_format using ffmpeg.
535
536    Yields chunks of resampled audio as they become available.
537
538    :param input_audio: Raw PCM audio data or async generator of PCM chunks.
539    :param input_format: AudioFormat of the input audio.
540    :param output_format: Desired AudioFormat for the output audio.
541    :param chunk_size: Output chunk size in bytes. Defaults to 1 second of output PCM.
542    """
543    if chunk_size is None:
544        chunk_size = output_format.pcm_sample_size
545
546    async def _as_generator() -> AsyncGenerator[bytes]:
547        if isinstance(input_audio, bytes):
548            yield input_audio
549        else:
550            async for chunk in input_audio:
551                yield chunk
552
553    if input_format == output_format:
554        buffer = b""
555        async for chunk in _as_generator():
556            buffer += chunk
557            while len(buffer) >= chunk_size:
558                yield buffer[:chunk_size]
559                buffer = buffer[chunk_size:]
560        if buffer:
561            yield buffer
562        return
563
564    async for chunk in get_ffmpeg_stream(
565        audio_input=_as_generator(),
566        input_format=input_format,
567        output_format=output_format,
568        chunk_size=chunk_size,
569    ):
570        yield chunk
571
572
573def calculate_content_length(
574    fmt: AudioFormat,
575    seconds: float = 1,
576) -> int:
577    """
578    Calculate the estimated encoded size in bytes for a given format and duration.
579
580    For CBR lossy formats (MP3/AAC), the estimate is near-exact.
581    For lossless formats (FLAC), the estimate uses an empirical average
582    compression ratio and may differ from actual size by up to ~15%.
583    For uncompressed formats (PCM/WAV), the result is exact.
584
585    :param fmt: The audio format to estimate size for.
586    :param seconds: Duration in seconds.
587    """
588    pcm_size = int(fmt.sample_rate * (fmt.bit_depth / 8) * fmt.channels * seconds)
589    if fmt.content_type.is_pcm():
590        return pcm_size
591    if fmt.content_type in (ContentType.WAV, ContentType.AIFF, ContentType.DSF):
592        return pcm_size
593    if fmt.bit_rate and fmt.bit_rate < 10000:
594        return int(((fmt.bit_rate * 1000) / 8) * seconds)
595    if fmt.content_type in (ContentType.FLAC, ContentType.WAVPACK, ContentType.ALAC):
596        # FLAC compression_level 0: empirical ratio ~74.7% of PCM
597        # Source: https://z-issue.com/wp/flac-compression-level-comparison/
598        # Real-world variance: 65-85% depending on audio content.
599        return int(pcm_size * 0.747)
600    if fmt.content_type == ContentType.MP3:
601        return int(((DEFAULT_MP3_BIT_RATE * 1000) / 8) * seconds)
602    if fmt.content_type == ContentType.OGG:
603        return int((320000 / 8) * seconds)
604    if fmt.content_type in (ContentType.AAC, ContentType.M4A):
605        # CBR 256kbps as set in get_ffmpeg_args
606        return int((256000 / 8) * seconds)
607    return int((320000 / 8) * seconds)
608
609
610def get_output_format_key(fmt: AudioFormat) -> str:
611    """
612    Get a stable key representing the output encoding parameters.
613
614    :param fmt: The output audio format.
615    """
616    return f"{fmt.content_type.value}_{fmt.sample_rate}_{fmt.bit_depth}_{fmt.channels}"
617
618
619CONTENT_LENGTH_CACHE_CATEGORY = 50
620CONTENT_LENGTH_CACHE_PROVIDER = "audio"
621CONTENT_LENGTH_CACHE_EXPIRATION = 365 * 86400  # 1 year
622
623
624async def get_content_length(
625    mass: MusicAssistant,
626    uri: str,
627    output_format: AudioFormat,
628    seconds: float,
629) -> int:
630    """
631    Get the estimated encoded size, using cached actual measurement when available.
632
633    After a track has been fully streamed, its actual content size and duration
634    are cached. On subsequent plays this gives a near-exact content_length:
635    - Exact when the requested duration matches the cached duration.
636    - Very accurate when the duration differs (derived bytes-per-second).
637
638    Falls back to the static estimate from calculate_content_length() if no cache entry exists.
639
640    :param mass: The MusicAssistant instance (for cache access).
641    :param uri: The media URI (e.g. "qobuz://track/12345").
642    :param output_format: The output audio format.
643    :param seconds: Duration in seconds to estimate.
644    """
645    cache_key = f"{uri}/{get_output_format_key(output_format)}"
646    cached: dict[str, float] | None = await mass.cache.get(
647        cache_key,
648        provider=CONTENT_LENGTH_CACHE_PROVIDER,
649        category=CONTENT_LENGTH_CACHE_CATEGORY,
650    )
651    if cached is not None:
652        cached_size = cached["size"]
653        cached_duration = cached["duration"]
654        if abs(seconds - cached_duration) < 1:
655            # same duration: return the exact cached size
656            return int(cached_size)
657        # different duration: derive bytes-per-second from the cached measurement
658        return int((cached_size / cached_duration) * seconds)
659    return calculate_content_length(output_format, seconds)
660
661
662async def store_content_length_in_cache(
663    mass: MusicAssistant,
664    uri: str,
665    output_format: AudioFormat,
666    content_size: int,
667    seconds_streamed: float,
668) -> None:
669    """
670    Store the actual content size after a track has been fully streamed.
671
672    :param mass: The MusicAssistant instance (for cache access).
673    :param uri: The media URI (e.g. "qobuz://track/12345").
674    :param output_format: The output audio format used for encoding.
675    :param content_size: Total encoded bytes sent to the player.
676    :param seconds_streamed: Duration of audio streamed in seconds.
677    """
678    if seconds_streamed < 10 or content_size < 1000:
679        return
680    cache_key = f"{uri}/{get_output_format_key(output_format)}"
681    await mass.cache.set(
682        cache_key,
683        {"size": content_size, "duration": seconds_streamed},
684        expiration=CONTENT_LENGTH_CACHE_EXPIRATION,
685        provider=CONTENT_LENGTH_CACHE_PROVIDER,
686        category=CONTENT_LENGTH_CACHE_CATEGORY,
687        persistent=True,
688    )
689
690
691PROBED_DURATION_CACHE_CATEGORY = 51
692PROBED_DURATION_CACHE_PROVIDER = "audio"
693PROBED_DURATION_CACHE_EXPIRATION = 365 * 86400  # 1 year
694
695
696async def get_probed_duration(mass: MusicAssistant, uri: str) -> int | None:
697    """
698    Get the duration determined during an earlier playback of the given item, if any.
699
700    Use for items whose provider does not report a duration, such as podcast episodes
701    from a feed without itunes:duration.
702
703    :param mass: The MusicAssistant instance (for cache access).
704    :param uri: The media item URI (e.g. "overcast--1://podcast_episode/abc").
705    :return: The duration in seconds, or None if the item was never played.
706    """
707    duration: int | None = await mass.cache.get(
708        uri,
709        provider=PROBED_DURATION_CACHE_PROVIDER,
710        category=PROBED_DURATION_CACHE_CATEGORY,
711    )
712    return duration
713
714
715async def store_probed_duration(mass: MusicAssistant, uri: str, duration: int) -> None:
716    """
717    Store the duration of an item that was determined while streaming it.
718
719    A duration below a second is ignored.
720
721    :param mass: The MusicAssistant instance (for cache access).
722    :param uri: The media item URI (e.g. "overcast--1://podcast_episode/abc").
723    :param duration: The duration in seconds.
724    """
725    if duration < 1:
726        return
727    await mass.cache.set(
728        uri,
729        duration,
730        expiration=PROBED_DURATION_CACHE_EXPIRATION,
731        provider=PROBED_DURATION_CACHE_PROVIDER,
732        category=PROBED_DURATION_CACHE_CATEGORY,
733        persistent=True,
734    )
735
736
737def get_bit_rate(fmt: AudioFormat) -> int:
738    """Get the (estimated) bit rate for a given AudioFormat, if known."""
739    if fmt.bit_rate:
740        return int(fmt.bit_rate / 1000) if fmt.bit_rate >= 10000 else fmt.bit_rate
741    return int((calculate_content_length(fmt, seconds=1) / 1000) * 8)
742
743
744def resolve_output_player_ids(
745    mass: MusicAssistant,
746    player_ids: Iterable[str],
747) -> set[str]:
748    """
749    Resolve output destinations to their user-facing player identifiers.
750
751    :param mass: Music Assistant instance.
752    :param player_ids: Player or protocol-player identifiers to resolve.
753    :return: Deduplicated user-facing player identifiers.
754    """
755    resolved_ids: set[str] = set()
756    for player_id in player_ids:
757        player = mass.players.get_player(player_id)
758        resolved_ids.add(
759            player.protocol_parent_id if player and player.protocol_parent_id else player_id
760        )
761    return resolved_ids
762
763
764def is_grouping_preventing_dsp(player: Player) -> bool:
765    """
766    Check if grouping is preventing DSP from being applied to this leader/PlayerGroup.
767
768    If this returns True, no DSP should be applied to the player.
769    This function will not check if the Player is in a group, the caller should do that first.
770    """
771    # We require the caller to handle non-leader cases themselves since player.state.synced_to
772    # can be unreliable in some edge cases
773    multi_device_dsp_supported = PlayerFeature.MULTI_DEVICE_DSP in player.state.supported_features
774    child_count = len(player.state.group_members) if player.state.group_members else 0
775
776    is_multiple_devices: bool
777    if player.provider.domain == "player_group":
778        # PlayerGroups have no leader, so having a child count of 1 means
779        # the group actually contains only a single player.
780        is_multiple_devices = child_count > 1
781    elif player.state.type == PlayerType.GROUP:
782        # This is an group player external to Music Assistant.
783        is_multiple_devices = True
784    else:
785        is_multiple_devices = child_count > 0
786    return is_multiple_devices and not multi_device_dsp_supported
787
788
789def parse_loudnorm(raw_stderr: bytes | str) -> float | None:
790    """Parse Loudness measurement from ffmpeg stderr output."""
791    stderr_data = raw_stderr.decode() if isinstance(raw_stderr, bytes) else raw_stderr
792    if "[Parsed_loudnorm_0 @" not in stderr_data:
793        return None
794    for jsun_chunk in stderr_data.split(" { "):
795        try:
796            stderr_data = "{" + jsun_chunk.rsplit("}")[0].strip() + "}"
797            loudness_data = json_loads(stderr_data)
798            return float(loudness_data["input_i"])
799        except (*JSON_DECODE_EXCEPTIONS, KeyError, ValueError, IndexError):
800            continue
801    return None
802
803
804def get_normalization_mode(
805    preference: VolumeNormalizationMode,
806    volume_normalization_enabled: bool,
807    streamdetails: StreamDetails,
808) -> VolumeNormalizationMode:
809    """
810    Get the volume normalization mode for a given queue and stream.
811
812    :param preference: The configured normalization preference for the stream's media type
813        (tracks or radio), from the streams core config.
814    :param volume_normalization_enabled: Whether normalization is enabled for the queue, already
815        resolved from the per-queue setting and its global (queue controller) fallback.
816    :param streamdetails: The stream to evaluate.
817    """
818    if not volume_normalization_enabled:
819        # disabled for this queue
820        return VolumeNormalizationMode.DISABLED
821    if streamdetails.media_type == MediaType.AUDIO_SOURCE:
822        # live/realtime: upstream producer owns loudness, no measurement to converge on
823        return VolumeNormalizationMode.DISABLED
824    if streamdetails.media_type == MediaType.SOUND_EFFECT:
825        # never measured, and the dynamic fallback compresses short clips
826        return VolumeNormalizationMode.DISABLED
827    if streamdetails.target_loudness is None:
828        # no target loudness set, disable normalization
829        return VolumeNormalizationMode.DISABLED
830
831    # handle no measurement available but fallback to dynamic mode is allowed
832    if streamdetails.loudness is None and preference == VolumeNormalizationMode.FALLBACK_DYNAMIC:
833        return VolumeNormalizationMode.DYNAMIC
834
835    # handle no measurement available and no fallback allowed
836    if streamdetails.loudness is None and preference == VolumeNormalizationMode.MEASUREMENT_ONLY:
837        return VolumeNormalizationMode.DISABLED
838
839    # handle no measurement available and fallback to fixed gain is allowed
840    if streamdetails.loudness is None and preference == VolumeNormalizationMode.FALLBACK_FIXED_GAIN:
841        return VolumeNormalizationMode.FIXED_GAIN
842
843    # handle measurement available - chosen mode is measurement
844    if streamdetails.loudness is not None and preference not in (
845        VolumeNormalizationMode.DISABLED,
846        VolumeNormalizationMode.FIXED_GAIN,
847        VolumeNormalizationMode.DYNAMIC,
848    ):
849        return VolumeNormalizationMode.MEASUREMENT_ONLY
850
851    # simply return the preference
852    return preference
853