/
/
1"""Test that PCM formats follow the arriving audio, not the advertised source."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import cast
7
8from music_assistant_models.enums import ContentType, MediaType, VolumeNormalizationMode
9from music_assistant_models.media_items import AudioFormat
10from music_assistant_models.streamdetails import StreamDetails
11
12
13def _streamdetails(advertised: AudioFormat, arriving: AudioFormat | None) -> StreamDetails:
14 """Return StreamDetails advertising one format while delivering another."""
15 return StreamDetails(
16 provider="test--1",
17 item_id="1",
18 audio_format=advertised,
19 decoded_audio_format=arriving,
20 media_type=MediaType.TRACK,
21 )
22
23
24def test_the_buffer_follows_the_decoded_format() -> None:
25 """A provider that decoded the source for us must not have its depth taken literally."""
26 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
27 _buffer_pcm_format,
28 )
29
30 # advertised as 16-bit lossy, delivered as 32-bit PCM: taking the advert
31 # literally would truncate real audio to 16 bits
32 advertised = AudioFormat(
33 content_type=ContentType.OGG,
34 codec_type=ContentType.VORBIS,
35 sample_rate=44100,
36 bit_depth=16,
37 channels=2,
38 bit_rate=160,
39 )
40 arriving = AudioFormat(
41 content_type=ContentType.PCM_S32LE,
42 codec_type=ContentType.PCM_S32LE,
43 sample_rate=44100,
44 bit_depth=32,
45 channels=2,
46 )
47 pcm = _buffer_pcm_format(_streamdetails(advertised, arriving))
48 assert pcm.bit_depth == 32
49 assert pcm.content_type == ContentType.PCM_S32LE
50
51
52def test_the_buffer_uses_the_source_when_nothing_was_decoded() -> None:
53 """Without a separate handoff the advertised format describes the bytes too."""
54 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
55 _buffer_pcm_format,
56 )
57
58 advertised = AudioFormat(
59 content_type=ContentType.FLAC,
60 codec_type=ContentType.FLAC,
61 sample_rate=48000,
62 bit_depth=24,
63 channels=2,
64 )
65 pcm = _buffer_pcm_format(_streamdetails(advertised, None))
66 assert pcm.bit_depth == 24
67 assert pcm.sample_rate == 48000
68
69
70def test_a_surround_source_is_folded_to_stereo() -> None:
71 """The buffer holds the stereo fold, so analysis measures what is played."""
72 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
73 _buffer_pcm_format,
74 )
75
76 arriving = AudioFormat(
77 content_type=ContentType.PCM_S32LE,
78 codec_type=ContentType.PCM_S32LE,
79 sample_rate=44100,
80 bit_depth=32,
81 channels=6,
82 )
83 advertised = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=24)
84 assert _buffer_pcm_format(_streamdetails(advertised, arriving)).channels == 2
85
86
87def test_the_flow_depth_follows_the_arriving_audio() -> None:
88 """
89 The flow must not narrow a stream to a depth its audio never had.
90
91 Reusing the source's native depth is a passthrough optimisation, so it has
92 to read the depth the bytes arrive in - a provider that decoded for us may
93 advertise something narrower purely for display.
94 """
95 from music_assistant.controllers.streams.audio import StreamsAudio # noqa: PLC0415
96
97 advertised = AudioFormat(
98 content_type=ContentType.FLAC,
99 codec_type=ContentType.FLAC,
100 sample_rate=44100,
101 bit_depth=24,
102 channels=2,
103 )
104 arriving = AudioFormat(
105 content_type=ContentType.PCM_S32LE,
106 codec_type=ContentType.PCM_S32LE,
107 sample_rate=44100,
108 bit_depth=32,
109 channels=2,
110 )
111 streamdetails = _streamdetails(advertised, arriving)
112 streamdetails.volume_normalization_mode = VolumeNormalizationMode.DISABLED
113
114 content_type, bit_depth = StreamsAudio._pick_pcm_bit_depth(
115 cast("StreamsAudio", SimpleNamespace()),
116 players=(),
117 streamdetails=streamdetails,
118 crossfade_enabled=False,
119 )
120
121 assert bit_depth == 32
122 assert content_type == ContentType.PCM_S32LE
123