/
/
/
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 TYPE_CHECKING, 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
12if TYPE_CHECKING:
13 from music_assistant.models.player import Player
14
15
16def _streamdetails(
17 advertised: AudioFormat,
18 arriving: AudioFormat | None,
19 media_type: MediaType = MediaType.TRACK,
20) -> StreamDetails:
21 """Return StreamDetails advertising one format while delivering another."""
22 return StreamDetails(
23 provider="test--1",
24 item_id="1",
25 audio_format=advertised,
26 decoded_audio_format=arriving,
27 media_type=media_type,
28 )
29
30
31def test_the_buffer_follows_the_decoded_format() -> None:
32 """A provider that decoded the source for us must not have its depth taken literally."""
33 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
34 _buffer_pcm_format,
35 )
36
37 # advertised as 16-bit lossy, delivered as 32-bit PCM: taking the advert
38 # literally would truncate real audio to 16 bits
39 advertised = AudioFormat(
40 content_type=ContentType.OGG,
41 codec_type=ContentType.VORBIS,
42 sample_rate=44100,
43 bit_depth=16,
44 channels=2,
45 bit_rate=160,
46 )
47 arriving = AudioFormat(
48 content_type=ContentType.PCM_S32LE,
49 codec_type=ContentType.PCM_S32LE,
50 sample_rate=44100,
51 bit_depth=32,
52 channels=2,
53 )
54 pcm = _buffer_pcm_format(_streamdetails(advertised, arriving))
55 assert pcm.bit_depth == 32
56 assert pcm.content_type == ContentType.PCM_S32LE
57
58
59def test_the_buffer_uses_the_source_when_nothing_was_decoded() -> None:
60 """Without a separate handoff the advertised format describes the bytes too."""
61 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
62 _buffer_pcm_format,
63 )
64
65 advertised = AudioFormat(
66 content_type=ContentType.FLAC,
67 codec_type=ContentType.FLAC,
68 sample_rate=48000,
69 bit_depth=24,
70 channels=2,
71 )
72 pcm = _buffer_pcm_format(_streamdetails(advertised, None))
73 assert pcm.bit_depth == 24
74 assert pcm.sample_rate == 48000
75
76
77def test_a_surround_source_is_folded_to_stereo() -> None:
78 """The buffer holds the stereo fold, so analysis measures what is played."""
79 from music_assistant.controllers.streams.audio_buffer import ( # noqa: PLC0415
80 _buffer_pcm_format,
81 )
82
83 arriving = AudioFormat(
84 content_type=ContentType.PCM_S32LE,
85 codec_type=ContentType.PCM_S32LE,
86 sample_rate=44100,
87 bit_depth=32,
88 channels=6,
89 )
90 advertised = AudioFormat(content_type=ContentType.FLAC, sample_rate=44100, bit_depth=24)
91 assert _buffer_pcm_format(_streamdetails(advertised, arriving)).channels == 2
92
93
94def test_the_flow_depth_follows_the_arriving_audio() -> None:
95 """
96 The flow must not narrow a stream to a depth its audio never had.
97
98 Reusing the source's native depth is a passthrough optimisation, so it has
99 to read the depth the bytes arrive in - a provider that decoded for us may
100 advertise something narrower purely for display.
101 """
102 from music_assistant.controllers.streams.audio import StreamsAudio # noqa: PLC0415
103
104 advertised = AudioFormat(
105 content_type=ContentType.FLAC,
106 codec_type=ContentType.FLAC,
107 sample_rate=44100,
108 bit_depth=24,
109 channels=2,
110 )
111 arriving = AudioFormat(
112 content_type=ContentType.PCM_S32LE,
113 codec_type=ContentType.PCM_S32LE,
114 sample_rate=44100,
115 bit_depth=32,
116 channels=2,
117 )
118 streamdetails = _streamdetails(advertised, arriving)
119 streamdetails.volume_normalization_mode = VolumeNormalizationMode.DISABLED
120
121 content_type, bit_depth = StreamsAudio._pick_pcm_bit_depth(
122 cast("StreamsAudio", SimpleNamespace()),
123 players=(),
124 streamdetails=streamdetails,
125 crossfade_enabled=False,
126 )
127
128 assert bit_depth == 32
129 assert content_type == ContentType.PCM_S32LE
130
131
132def test_the_audio_source_passthrough_follows_the_arriving_audio() -> None:
133 """
134 A live source's passthrough format must not narrow the audio it passes through.
135
136 The passthrough exists to hand the player the source's own samples, so it
137 has to read the format the bytes arrive in - an engine that decoded for us
138 (Spotify Connect) advertises the quality tier it asked for, which is
139 narrower than the PCM it hands over.
140 """
141 from music_assistant.controllers.streams.audio import StreamsAudio # noqa: PLC0415
142
143 advertised = AudioFormat(
144 content_type=ContentType.OGG,
145 codec_type=ContentType.VORBIS,
146 sample_rate=44100,
147 bit_depth=16,
148 channels=2,
149 bit_rate=160,
150 )
151 arriving = AudioFormat(
152 content_type=ContentType.PCM_S32LE,
153 codec_type=ContentType.PCM_S32LE,
154 sample_rate=44100,
155 bit_depth=32,
156 channels=2,
157 )
158
159 pcm_format = StreamsAudio._select_audio_source_pcm_format(
160 cast("StreamsAudio", SimpleNamespace()),
161 player=cast("Player", SimpleNamespace()),
162 streamdetails=_streamdetails(advertised, arriving, MediaType.AUDIO_SOURCE),
163 supported_sample_rates=(44100, 48000),
164 )
165
166 assert pcm_format.bit_depth == 32
167 assert pcm_format.content_type == ContentType.PCM_S32LE
168 assert pcm_format.sample_rate == 44100
169