/
/
/
1"""Tests for the DSP helpers."""
2
3import math
4
5from music_assistant_models.dsp import (
6 AudioChannel,
7 BalanceFilter,
8 CompressorFilter,
9 ConvolutionFilter,
10 CrossfeedFilter,
11 GainFilter,
12 HighLowPassFilter,
13 HighLowPassMode,
14 HighLowPassSlope,
15 ParametricEQBand,
16 ParametricEQBandType,
17 ParametricEQFilter,
18 SafetyLimiterFilter,
19 StereoWidthFilter,
20 ToneControlFilter,
21 TransposeFilter,
22)
23from music_assistant_models.media_items.audio_format import AudioFormat
24
25from music_assistant.helpers.dsp import (
26 ComplexFilter,
27 ComplexFilterInput,
28 filter_to_ffmpeg_params,
29)
30
31INPUT_FORMAT = AudioFormat(sample_rate=48000)
32IR_DIR = "/irs"
33
34
35def test_gain_filter() -> None:
36 """Test that a gain filter maps to a volume filter in dB."""
37 dsp_filter = GainFilter(enabled=True, gain=5.5)
38 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == ["volume=5.5dB"]
39 dsp_filter = GainFilter(enabled=True, gain=-15.0)
40 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == ["volume=-15.0dB"]
41
42
43def test_gain_filter_zero_is_passthrough() -> None:
44 """Test that a gain filter with 0 dB gain emits no ffmpeg filter."""
45 dsp_filter = GainFilter(enabled=True, gain=0.0)
46 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
47
48
49def test_balance_filter_right() -> None:
50 """Test that balance towards the right attenuates only the left channel."""
51 dsp_filter = BalanceFilter(enabled=True, balance=40)
52 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
53 "pan=stereo|FL=0.6*FL|FR=FR"
54 ]
55
56
57def test_balance_filter_left() -> None:
58 """Test that balance towards the left attenuates only the right channel."""
59 dsp_filter = BalanceFilter(enabled=True, balance=-40)
60 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
61 "pan=stereo|FL=FL|FR=0.6*FR"
62 ]
63
64
65def test_balance_filter_full_deflection_mutes_opposite_channel() -> None:
66 """Test that full balance deflection fully mutes the opposite channel."""
67 dsp_filter = BalanceFilter(enabled=True, balance=100)
68 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
69 "pan=stereo|FL=0.0*FL|FR=FR"
70 ]
71
72
73def test_balance_filter_zero_is_passthrough() -> None:
74 """Test that a centered balance filter emits no ffmpeg filter."""
75 dsp_filter = BalanceFilter(enabled=True, balance=0)
76 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
77
78
79def test_balance_filter_mono_is_skipped() -> None:
80 """Test that balance is skipped on a mono source (stereo-only operation)."""
81 dsp_filter = BalanceFilter(enabled=True, balance=40)
82 mono_format = AudioFormat(sample_rate=48000, channels=1)
83 assert filter_to_ffmpeg_params(dsp_filter, mono_format, ir_dir=IR_DIR) == []
84
85
86def test_transpose_filter_octaves() -> None:
87 """Test that a full octave transpose halves or doubles the pitch ratio."""
88 assert filter_to_ffmpeg_params(
89 TransposeFilter(enabled=True, semitones=12.0), INPUT_FORMAT, ir_dir=IR_DIR
90 ) == ["rubberband=pitch=2.0:formant=preserved:pitchq=quality:window=long"]
91 assert filter_to_ffmpeg_params(
92 TransposeFilter(enabled=True, semitones=-12.0), INPUT_FORMAT, ir_dir=IR_DIR
93 ) == ["rubberband=pitch=0.5:formant=preserved:pitchq=quality:window=long"]
94
95
96def test_transpose_filter_fractional_semitones() -> None:
97 """Test that a fractional transpose is supported (e.g. A=432Hz concert pitch)."""
98 dsp_filter = TransposeFilter(enabled=True, semitones=-0.318)
99 params = filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR)
100 assert len(params) == 1
101 assert isinstance(params[0], str)
102 pitch, _, options = params[0].removeprefix("rubberband=pitch=").partition(":")
103 assert math.isclose(float(pitch), 0.98183, rel_tol=1e-4)
104 assert options == "formant=preserved:pitchq=quality:window=long"
105
106
107def test_transpose_filter_zero_is_passthrough() -> None:
108 """Test that a transpose filter of 0 semitones emits no ffmpeg filter."""
109 dsp_filter = TransposeFilter(enabled=True, semitones=0.0)
110 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
111
112
113def test_stereo_width_filter_wide() -> None:
114 """Test that a widened stereo width maps to an extrastereo filter."""
115 dsp_filter = StereoWidthFilter(enabled=True, width=1.5)
116 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
117 "extrastereo=m=1.5:c=0"
118 ]
119
120
121def test_stereo_width_filter_does_not_clip_internally() -> None:
122 """Test that widening never uses the internal clipping extrastereo enables by default."""
123 for width in (0.0, 0.5, 1.5, 2.0):
124 dsp_filter = StereoWidthFilter(enabled=True, width=width)
125 params = filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR)
126 assert params == [f"extrastereo=m={width}:c=0"]
127
128
129def test_stereo_width_filter_mono() -> None:
130 """Test that zero width collapses the side signal to mono."""
131 dsp_filter = StereoWidthFilter(enabled=True, width=0.0)
132 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
133 "extrastereo=m=0.0:c=0"
134 ]
135
136
137def test_stereo_width_filter_neutral_is_passthrough() -> None:
138 """Test that a unity stereo width emits no ffmpeg filter."""
139 dsp_filter = StereoWidthFilter(enabled=True, width=1.0)
140 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
141
142
143def test_stereo_width_filter_mono_is_skipped() -> None:
144 """Test that stereo width is skipped on a mono source (stereo-only operation)."""
145 dsp_filter = StereoWidthFilter(enabled=True, width=1.5)
146 mono_format = AudioFormat(sample_rate=48000, channels=1)
147 assert filter_to_ffmpeg_params(dsp_filter, mono_format, ir_dir=IR_DIR) == []
148
149
150def test_crossfeed_filter() -> None:
151 """Test that a crossfeed filter maps to an ffmpeg crossfeed filter."""
152 dsp_filter = CrossfeedFilter(enabled=True, strength=0.35, soundstage=0.6)
153 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
154 "crossfeed=strength=0.35:range=0.6:level_in=1"
155 ]
156
157
158def test_crossfeed_filter_mono_is_skipped() -> None:
159 """Test that crossfeed is skipped on a mono source (stereo-only operation)."""
160 dsp_filter = CrossfeedFilter(enabled=True, strength=0.2, soundstage=0.5)
161 mono_format = AudioFormat(sample_rate=48000, channels=1)
162 assert filter_to_ffmpeg_params(dsp_filter, mono_format, ir_dir=IR_DIR) == []
163
164
165def test_tone_control_filter() -> None:
166 """Test that tone control levels map to equalizer filters, omitting zero levels."""
167 dsp_filter = ToneControlFilter(enabled=True, bass_level=4.0, mid_level=0.0, treble_level=-2.0)
168 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
169 "equalizer=frequency=100:width=200:width_type=h:gain=4.0",
170 "equalizer=frequency=9000:width=18000:width_type=h:gain=-2.0",
171 ]
172
173
174def test_tone_control_filter_neutral_is_passthrough() -> None:
175 """Test that a tone control filter with all levels at 0 emits no ffmpeg filter."""
176 dsp_filter = ToneControlFilter(enabled=True, bass_level=0.0, mid_level=0.0, treble_level=0.0)
177 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
178
179
180def test_parametric_eq_preamp() -> None:
181 """Test that a parametric EQ preamp maps to a volume filter."""
182 dsp_filter = ParametricEQFilter(enabled=True, preamp=-3.0, per_channel_preamp={}, bands=[])
183 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == ["volume=-3.0dB"]
184
185
186def test_parametric_eq_per_channel_preamp() -> None:
187 """Test that per-channel preamp maps to a pan filter with linear per-channel gains."""
188 dsp_filter = ParametricEQFilter(
189 enabled=True,
190 preamp=0.0,
191 per_channel_preamp={AudioChannel.FL: -6.0},
192 bands=[],
193 )
194 expected_gain = 10 ** (-6.0 / 20)
195 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
196 f"pan=stereo|FL={expected_gain}*FL|FR=FR"
197 ]
198
199
200def test_parametric_eq_peak_band() -> None:
201 """Test that an enabled peak band maps to a biquad filter with cookbook coefficients."""
202 band = ParametricEQBand(
203 frequency=1000.0,
204 q=1.0,
205 gain=3.0,
206 type=ParametricEQBandType.PEAK,
207 enabled=True,
208 channel=AudioChannel.FL,
209 )
210 dsp_filter = ParametricEQFilter(enabled=True, per_channel_preamp={}, bands=[band])
211
212 a = math.sqrt(10 ** (3.0 / 20))
213 w_0 = 2 * math.pi * 1000.0 / 48000
214 alpha = math.sin(w_0) / 2
215 b0 = 1 + alpha * a
216 b1 = -2 * math.cos(w_0)
217 b2 = 1 - alpha * a
218 a0 = 1 + alpha / a
219 a1 = -2 * math.cos(w_0)
220 a2 = 1 - alpha / a
221 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
222 f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}:c=FL"
223 ]
224
225
226def test_parametric_eq_disabled_band_skipped() -> None:
227 """Test that disabled parametric EQ bands are skipped."""
228 band = ParametricEQBand(enabled=False, gain=6.0)
229 dsp_filter = ParametricEQFilter(enabled=True, per_channel_preamp={}, bands=[band])
230 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
231
232
233def _expected_pass_section(mode: HighLowPassMode, frequency: float, q: float) -> str:
234 """Build the expected cookbook biquad string for one high/low-pass section."""
235 w_0 = 2 * math.pi * frequency / INPUT_FORMAT.sample_rate
236 alpha = math.sin(w_0) / (2 * q)
237 if mode == HighLowPassMode.HIGH_PASS:
238 b0 = (1 + math.cos(w_0)) / 2
239 b1 = -(1 + math.cos(w_0))
240 b2 = (1 + math.cos(w_0)) / 2
241 else:
242 b0 = (1 - math.cos(w_0)) / 2
243 b1 = 1 - math.cos(w_0)
244 b2 = (1 - math.cos(w_0)) / 2
245 a0 = 1 + alpha
246 a1 = -2 * math.cos(w_0)
247 a2 = 1 - alpha
248 return f"biquad=b0={b0}:b1={b1}:b2={b2}:a0={a0}:a1={a1}:a2={a2}"
249
250
251def test_high_low_pass_12db_single_section() -> None:
252 """Test that a 12 dB/octave high-pass is a single Butterworth biquad at Q=1/sqrt(2)."""
253 dsp_filter = HighLowPassFilter(
254 enabled=True, mode=HighLowPassMode.HIGH_PASS, frequency=100.0, slope=HighLowPassSlope.DB12
255 )
256 q = 1 / (2 * math.cos(math.pi / 4))
257 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
258 _expected_pass_section(HighLowPassMode.HIGH_PASS, 100.0, q)
259 ]
260
261
262def test_high_low_pass_24db_two_sections() -> None:
263 """Test that a 24 dB/octave low-pass is two sections with the 4th-order Butterworth Qs."""
264 dsp_filter = HighLowPassFilter(
265 enabled=True, mode=HighLowPassMode.LOW_PASS, frequency=8000.0, slope=HighLowPassSlope.DB24
266 )
267 qs = [1 / (2 * math.cos(math.pi * (2 * k + 1) / 8)) for k in range(2)]
268 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
269 _expected_pass_section(HighLowPassMode.LOW_PASS, 8000.0, q) for q in qs
270 ]
271
272
273def test_high_low_pass_48db_has_four_sections() -> None:
274 """Test that a 48 dB/octave filter is a cascade of four biquad sections."""
275 dsp_filter = HighLowPassFilter(
276 enabled=True, mode=HighLowPassMode.HIGH_PASS, frequency=40.0, slope=HighLowPassSlope.DB48
277 )
278 params = filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR)
279 assert len(params) == 4
280 assert all(isinstance(p, str) and p.startswith("biquad=") for p in params)
281
282
283def test_high_low_pass_mode_changes_coefficients() -> None:
284 """Test that high-pass and low-pass at identical settings produce different coefficients."""
285 high = HighLowPassFilter(
286 enabled=True, mode=HighLowPassMode.HIGH_PASS, frequency=1000.0, slope=HighLowPassSlope.DB12
287 )
288 low = HighLowPassFilter(
289 enabled=True, mode=HighLowPassMode.LOW_PASS, frequency=1000.0, slope=HighLowPassSlope.DB12
290 )
291 assert filter_to_ffmpeg_params(high, INPUT_FORMAT, ir_dir=IR_DIR) != filter_to_ffmpeg_params(
292 low, INPUT_FORMAT, ir_dir=IR_DIR
293 )
294
295
296def test_safety_limiter_filter() -> None:
297 """Test that a safety limiter maps to an in-chain alimiter at the given ceiling."""
298 dsp_filter = SafetyLimiterFilter(enabled=True, ceiling=-2.0)
299 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
300 "alimiter=limit=-2.0dB:level=false:asc=true:latency=true"
301 ]
302
303
304def test_compressor_filter() -> None:
305 """Test that a compressor maps to acompressor with dB/ms/ratio values."""
306 dsp_filter = CompressorFilter(
307 enabled=True,
308 threshold=-18.0,
309 ratio=2.0,
310 attack=20.0,
311 release=250.0,
312 knee=9.0,
313 makeup=0.0,
314 )
315 expected = (
316 "acompressor=threshold=-18.0dB:ratio=2.0:attack=20.0:release=250.0"
317 f":knee={10 ** (9.0 / 20)}:makeup=0.0dB"
318 )
319 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [expected]
320
321
322def test_compressor_unity_ratio_is_passthrough() -> None:
323 """Test that a compressor which compresses nothing and adds no gain emits no filter."""
324 dsp_filter = CompressorFilter(enabled=True, ratio=1.0, makeup=0.0)
325 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
326
327
328def test_compressor_unity_ratio_keeps_makeup_gain() -> None:
329 """Test that make-up gain still applies at unity ratio, rather than being dropped."""
330 dsp_filter = CompressorFilter(enabled=True, ratio=1.0, makeup=6.0)
331 params = filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR)
332 assert len(params) == 1
333 assert isinstance(params[0], str)
334 assert ":makeup=6.0dB" in params[0]
335
336
337def test_compressor_knee_db_maps_to_linear_factor() -> None:
338 """Test that a knee width in dB maps to acompressor's linear knee factor."""
339 # 0 dB is a hard knee, acompressor's minimum knee factor of 1.0
340 hard_knee = CompressorFilter(enabled=True, knee=0.0)
341 hard_params = filter_to_ffmpeg_params(hard_knee, INPUT_FORMAT, ir_dir=IR_DIR)
342 assert isinstance(hard_params[0], str)
343 assert ":knee=1.0:" in hard_params[0]
344 # 18 dB maps to ~7.94, just under acompressor's maximum knee factor of 8
345 soft_knee = CompressorFilter(enabled=True, knee=18.0)
346 soft_params = filter_to_ffmpeg_params(soft_knee, INPUT_FORMAT, ir_dir=IR_DIR)
347 assert isinstance(soft_params[0], str)
348 assert f":knee={10 ** (18.0 / 20)}:" in soft_params[0]
349
350
351def test_convolution_filter() -> None:
352 """Test that a convolution filter maps to an afir fragment pulling in the IR."""
353 dsp_filter = ConvolutionFilter(enabled=True, ir_id="abc123")
354 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
355 ComplexFilter(
356 body="afir=irnorm=1",
357 inputs=[ComplexFilterInput(path="/irs/abc123.wav", filters="aresample=48000")],
358 )
359 ]
360
361
362def test_convolution_filter_with_gain() -> None:
363 """Test that a non-zero convolution gain appends a trailing volume filter."""
364 dsp_filter = ConvolutionFilter(enabled=True, ir_id="abc123", gain=3.0)
365 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == [
366 ComplexFilter(
367 body="afir=irnorm=1",
368 inputs=[ComplexFilterInput(path="/irs/abc123.wav", filters="aresample=48000")],
369 ),
370 "volume=3.0dB",
371 ]
372
373
374def test_convolution_filter_ir_path_is_not_escaped() -> None:
375 """Test that the IR path is passed through verbatim, quoting characters and all."""
376 dsp_filter = ConvolutionFilter(enabled=True, ir_id="abc123")
377 params = filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir="/it's a dir")
378 assert isinstance(params[0], ComplexFilter)
379 assert params[0].inputs[0].path == "/it's a dir/abc123.wav"
380
381
382def test_convolution_filter_empty_ir_id_skipped() -> None:
383 """Test that a convolution filter with no impulse response selected is a no-op."""
384 dsp_filter = ConvolutionFilter(enabled=True, ir_id="")
385 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
386
387
388def test_convolution_filter_unsafe_ir_id_skipped() -> None:
389 """Test that an ir_id outside the id format (path traversal attempt) is skipped."""
390 # uppercase, accented and full-width characters all pass str.isalnum()
391 for ir_id in ("../../etc/passwd", "ABC123", "abc\xe9", "\uff11\uff12\uff13"):
392 dsp_filter = ConvolutionFilter(enabled=True, ir_id=ir_id)
393 assert filter_to_ffmpeg_params(dsp_filter, INPUT_FORMAT, ir_dir=IR_DIR) == []
394