/
/
/
1"""
2Tests for skipping volume normalization when the source already did it.
3
4A provider that hands over audio at a loudness target of its own declares that
5with ``MusicProvider.delivers_normalized_audio``; correcting such a level again
6would mean normalizing twice, the second time against a measurement of the
7source's own output. That outcome is reported as ``SOURCE`` rather than
8``DISABLED`` - the audio is levelled, just not by us - so the gates that mean
9"Music Assistant applies nothing" have to accept both. Also verified: the
10loudness analyzer declines such a stream, so no measurement is stored for it.
11"""
12
13from __future__ import annotations
14
15from typing import Any, cast
16from unittest.mock import MagicMock
17
18import pytest
19from music_assistant_models.audio_processing import AudioNormalizationMeasurementSource
20from music_assistant_models.enums import ContentType, MediaType, VolumeNormalizationMode
21from music_assistant_models.media_items import AudioFormat
22from music_assistant_models.streamdetails import StreamDetails
23
24from music_assistant.controllers.streams.audio import StreamsAudio
25from music_assistant.controllers.streams.audio_processing import get_normalization_details
26from music_assistant.controllers.streams.constants import (
27 DEFAULT_VOLUME_NORMALIZATION_MODE,
28 OUTCOME_ONLY_NORMALIZATION_MODES,
29)
30from music_assistant.controllers.streams.controller import (
31 StreamsController,
32 _volume_normalization_preference_options,
33)
34from music_assistant.helpers.audio import get_normalization_mode
35from music_assistant.models.music_provider import MusicProvider
36
37
38class _NormalizingProvider(MusicProvider):
39 """A music provider that hands over audio it already levelled."""
40
41 def delivers_normalized_audio(self, streamdetails: StreamDetails) -> bool:
42 """Declare the source normalization."""
43 return True
44
45
46PCM_FORMAT = AudioFormat(
47 content_type=ContentType.PCM_S16LE,
48 codec_type=ContentType.PCM_S16LE,
49 sample_rate=44100,
50 bit_depth=16,
51 channels=2,
52)
53
54
55def test_a_normalized_source_is_left_alone() -> None:
56 """A source that normalizes its own output takes MA out of the loudness path."""
57 streamdetails = _streamdetails()
58 assert (
59 get_normalization_mode(
60 VolumeNormalizationMode.FALLBACK_DYNAMIC, True, streamdetails, source_normalized=True
61 )
62 == VolumeNormalizationMode.SOURCE
63 )
64
65
66def test_an_unnormalized_source_still_falls_back_to_dynamic() -> None:
67 """Without a measurement or a normalizing source, the dynamic fallback still applies."""
68 streamdetails = _streamdetails()
69 assert (
70 get_normalization_mode(
71 VolumeNormalizationMode.FALLBACK_DYNAMIC, True, streamdetails, source_normalized=False
72 )
73 == VolumeNormalizationMode.DYNAMIC
74 )
75
76
77def test_a_stored_measurement_does_not_override_a_normalized_source() -> None:
78 """A measurement left over from before must not pull MA back into correcting."""
79 streamdetails = _streamdetails()
80 # e.g. measured while the source was not normalizing, or on the other backend
81 streamdetails.loudness = -7.2
82 assert (
83 get_normalization_mode(
84 VolumeNormalizationMode.FALLBACK_DYNAMIC, True, streamdetails, source_normalized=True
85 )
86 == VolumeNormalizationMode.SOURCE
87 )
88 assert (
89 get_normalization_mode(
90 VolumeNormalizationMode.FALLBACK_DYNAMIC, True, streamdetails, source_normalized=False
91 )
92 == VolumeNormalizationMode.MEASUREMENT_ONLY
93 )
94
95
96def test_the_queue_setting_still_wins() -> None:
97 """Normalization disabled for the queue stays disabled either way."""
98 streamdetails = _streamdetails()
99 assert (
100 get_normalization_mode(
101 VolumeNormalizationMode.FALLBACK_DYNAMIC, False, streamdetails, source_normalized=True
102 )
103 == VolumeNormalizationMode.DISABLED
104 )
105
106
107def test_a_music_provider_declares_nothing_by_default() -> None:
108 """The declaration is opt-in: nothing downstream verifies it."""
109 assert object.__new__(MusicProvider).delivers_normalized_audio(_streamdetails()) is False
110
111
112@pytest.mark.parametrize("mode", [VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.SOURCE])
113async def test_the_analyzer_declines_a_stream_it_must_not_measure(
114 mode: VolumeNormalizationMode,
115) -> None:
116 """
117 A stream we do not normalize is not measured either, so no loudness is stored.
118
119 That is what keeps a value measured on one backend's output from being applied
120 to the other's, without any erase step. SOURCE has to decline for a second
121 reason: measuring there would record the source's own level as the track's.
122 """
123 from music_assistant.providers.loudness_analysis.provider import ( # noqa: PLC0415
124 LoudnessAnalysisProvider,
125 )
126
127 provider = object.__new__(LoudnessAnalysisProvider)
128 provider.logger = MagicMock()
129 streamdetails = _streamdetails()
130 streamdetails.volume_normalization_mode = mode
131 assert await provider._start_analysis("session-1", streamdetails, PCM_FORMAT) is False
132
133
134@pytest.mark.parametrize("mode", [VolumeNormalizationMode.DISABLED, VolumeNormalizationMode.SOURCE])
135def test_no_headroom_is_reserved_for_normalization_we_do_not_apply(
136 mode: VolumeNormalizationMode,
137) -> None:
138 """
139 F32 headroom is only paid for when Music Assistant itself touches the level.
140
141 A source-normalized stream keeps its native depth, exactly as a disabled one does.
142 """
143 audio = object.__new__(StreamsAudio)
144 streamdetails = _streamdetails()
145 streamdetails.volume_normalization_mode = mode
146
147 content_type, bit_depth = audio._pick_pcm_bit_depth(
148 [],
149 streamdetails,
150 crossfade_enabled=False,
151 overlay_active=False,
152 )
153
154 assert bit_depth == 16
155 assert content_type == ContentType.PCM_S16LE
156
157
158def test_source_normalized_audio_reports_a_mode_without_a_measurement() -> None:
159 """
160 A source-normalized stream reports who levelled it and nothing more.
161
162 The target and the measurement are ours, and neither describes what the source
163 did - a stale library measurement in particular must not be presented as the
164 level this audio was corrected against.
165 """
166 streamdetails = _streamdetails()
167 streamdetails.volume_normalization_mode = VolumeNormalizationMode.SOURCE
168 streamdetails.loudness = -7.2
169
170 details = get_normalization_details(streamdetails, None)
171
172 assert details is not None
173 assert details.mode == VolumeNormalizationMode.SOURCE
174 assert details.measurement_source == AudioNormalizationMeasurementSource.UNKNOWN
175 assert details.target_lufs is None
176 assert details.measured_lufs is None
177 assert details.applied_gain_db is None
178
179
180@pytest.mark.parametrize(
181 ("provider", "expected"),
182 [
183 (object.__new__(_NormalizingProvider), True),
184 (object.__new__(MusicProvider), False),
185 # a plugin provider, which never declares it
186 (MagicMock(), False),
187 ],
188)
189def test_only_a_music_provider_can_claim_it_levelled_the_audio(
190 provider: object, expected: bool
191) -> None:
192 """
193 A plugin provider serves playable items too, but never answers this.
194
195 Its live audio is taken out of the loudness path by its media type instead.
196 """
197 controller = cast("Any", object.__new__(StreamsController))
198 controller.mass = MagicMock()
199 controller.mass.get_provider.return_value = provider
200
201 assert controller.source_normalizes_audio(_streamdetails()) is expected
202
203
204def test_an_outcome_only_mode_is_not_offered_as_a_preference() -> None:
205 """
206 The setting says what Music Assistant should do, so outcomes are not choices.
207
208 SOURCE is set by a source that levels its own audio and UNKNOWN is what an
209 unrecognised value deserializes to; neither is something to ask a user for.
210 """
211 offered = {option.value for option in _volume_normalization_preference_options()}
212
213 assert offered == {
214 mode.value
215 for mode in VolumeNormalizationMode
216 if mode not in OUTCOME_ONLY_NORMALIZATION_MODES
217 }
218
219
220@pytest.mark.parametrize("stored", OUTCOME_ONLY_NORMALIZATION_MODES)
221def test_an_outcome_only_mode_stored_as_a_preference_is_not_honoured(
222 stored: VolumeNormalizationMode,
223) -> None:
224 """
225 Hiding an option does not make it unacceptable: nothing validates a saved value.
226
227 Handed back as the preference it would be applied as the mode, which for SOURCE
228 also means reporting that a source levelled audio nobody levelled.
229 """
230 audio = object.__new__(StreamsAudio)
231 audio.mass = MagicMock()
232 audio.mass.streams.get_config_value.return_value = stored.value
233
234 preference = audio._get_volume_normalization_preference(_streamdetails())
235
236 assert preference == DEFAULT_VOLUME_NORMALIZATION_MODE
237
238
239def _streamdetails() -> StreamDetails:
240 """Return StreamDetails for an ordinary track with a normalization target set."""
241 streamdetails = StreamDetails(
242 provider="test--1",
243 item_id="1",
244 audio_format=PCM_FORMAT,
245 media_type=MediaType.TRACK,
246 )
247 streamdetails.target_loudness = -14.0
248 return streamdetails
249