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