music-assistant-server

1"""End-to-end render check: the built chain actually swaps the bass in ffmpeg."""
2
3from __future__ import annotations
4
5import logging
6
7import numpy as np
8import pytest
9from music_assistant_models.enums import ContentType
10from music_assistant_models.media_items import AudioFormat
11
12from music_assistant.controllers.streams.smart_fades.fades import SmartCrossFade
13from music_assistant.models.audio_analysis import AudioAnalysisData
14
15PCM = AudioFormat(content_type=ContentType.PCM_F32LE, sample_rate=44100, bit_depth=32, channels=2)
16SR = 44100
17
18
19def _tone(freq: float, seconds: float, level: float = 0.2) -> np.ndarray:
20    """Return a stereo-interleaved sine tone."""
21    t = np.arange(int(SR * seconds)) / SR
22    mono = (level * np.sin(2 * np.pi * freq * t)).astype(np.float32)
23    return np.repeat(mono, 2)
24
25
26def _analysis(bpm: float, duration: float) -> AudioAnalysisData:
27    """Synthetic flat-energy analysis with a steady beat grid."""
28    interval = 60.0 / bpm
29    beats = np.arange(0.0, duration, interval, dtype=np.float32)
30    return AudioAnalysisData(
31        duration=duration,
32        bpm=bpm,
33        beats=beats.tolist(),
34        downbeats=beats[::4].tolist(),
35        rms_energy=np.full(1800, 0.5, dtype=np.float32).tolist(),
36        key="A",
37        mode="minor",
38    )
39
40
41def _with_bands(
42    analysis: AudioAnalysisData, low: float, low_mid: float, mid: float, high: float
43) -> AudioAnalysisData:
44    """Attach flat ``band_rms`` envelopes at the given amplitudes."""
45    analysis.extra_data = {
46        "band_rms": {
47            "low": np.full(1800, low, dtype=np.float32).tolist(),
48            "low_mid": np.full(1800, low_mid, dtype=np.float32).tolist(),
49            "mid": np.full(1800, mid, dtype=np.float32).tolist(),
50            "high": np.full(1800, high, dtype=np.float32).tolist(),
51        }
52    }
53    return analysis
54
55
56def _analysis_with_mid_bands(bpm: float, duration: float) -> AudioAnalysisData:
57    """Analysis with a mid-heavy, bass-light ``band_rms`` profile that clears the mid gate."""
58    # bass-light so the low swap stays out of the way; mid-heavy and constant
59    # so duty_mid saturates to 1.0 and F_mid clears the 0.18-0.30 gate corridor
60    return _with_bands(_analysis(bpm, duration), 0.05, 0.3, 0.7, 0.3)
61
62
63def _analysis_with_instrumental_bands(bpm: float, duration: float) -> AudioAnalysisData:
64    """Analysis with a bass-light, mid-light profile: every measured EQ gate bypasses."""
65    # f_low ~0.014 and f_mid ~0.13 sit below their gate corridors, so both the
66    # low and mid swap bypass while anchors/entry stay on the full-band paths
67    return _with_bands(_analysis(bpm, duration), 0.1, 0.55, 0.3, 0.55)
68
69
70def _band_rms(x: np.ndarray, lo: float, hi: float) -> float:
71    """RMS of one frequency band of the (interleaved stereo) signal's left channel."""
72    mono = x[0::2]
73    spec = np.abs(np.fft.rfft(mono))
74    freqs = np.fft.rfftfreq(len(mono), 1 / SR)
75    mask = (freqs >= lo) & (freqs < hi)
76    return float(np.sqrt(np.mean(spec[mask] ** 2)))
77
78
79async def _render(
80    out_analysis: AudioAnalysisData,
81    in_analysis: AudioAnalysisData,
82    fade_out: bytes,
83    fade_in: bytes,
84) -> tuple[np.ndarray, SmartCrossFade]:
85    """Build and apply a SmartCrossFade, returning the rendered mix and the fade."""
86    fade = SmartCrossFade(logging.getLogger(), out_analysis, in_analysis)
87    fade.build(len(fade_out), len(fade_in), PCM)
88    chunks = [chunk async for chunk in fade.apply(fade_out, fade_in, PCM)]
89    return np.frombuffer(b"".join(chunks), dtype=np.float32), fade
90
91
92def _cf_slice(mix: np.ndarray, fade: SmartCrossFade, frac0: float, frac1: float) -> np.ndarray:
93    """Slice the rendered crossfade window between two fractions of its span."""
94    timing = fade.timing_info
95    start_s = timing.pre_crossfade_duration + frac0 * timing.crossfade_duration
96    end_s = timing.pre_crossfade_duration + frac1 * timing.crossfade_duration
97    return mix[int(start_s * SR) * 2 : int(end_s * SR) * 2]
98
99
100@pytest.mark.asyncio
101async def test_bass_swaps_between_tracks() -> None:
102    """The low shelves attenuate A's bass and duck B's entrance vs an EQ-bypassed render."""
103    fade_out = (_tone(60.0, 45.0) + _tone(3000.0, 45.0)).tobytes()  # A: 60Hz bass
104    fade_in = (_tone(90.0, 45.0) + _tone(5000.0, 45.0)).tobytes()  # B: 90Hz bass
105    # differential render: identical PCM, one plan with the shipped full-depth
106    # kill (no band data) and one whose measured gates bypass all low shelves --
107    # any energy difference is then attributable to the low EQ, not acrossfade
108    killed_mix, killed = await _render(
109        _analysis(120.0, 240.0), _analysis(120.0, 240.0), fade_out, fade_in
110    )
111    open_mix, open_ = await _render(
112        _analysis_with_instrumental_bands(120.0, 240.0),
113        _analysis_with_instrumental_bands(120.0, 240.0),
114        fade_out,
115        fade_in,
116    )
117    assert killed.plan is not None
118    assert killed.plan.eq_plan.low_out is not None
119    assert open_.plan is not None
120    assert open_.plan.eq_plan.low_out is None
121    assert open_.plan.eq_plan.low_in is None
122    # identical geometry: the band data must only change EQ, never the timing
123    assert len(killed_mix) == len(open_mix)
124    # measure inside the crossfade window itself: A's bass is killed where the
125    # swap completes (late); B enters bass-ducked (early); -26dB kill leaves
126    # well under 30% of the bypassed render's energy
127    killed_late = _cf_slice(killed_mix, killed, 0.7, 0.95)
128    open_late = _cf_slice(open_mix, open_, 0.7, 0.95)
129    killed_early = _cf_slice(killed_mix, killed, 0.05, 0.3)
130    open_early = _cf_slice(open_mix, open_, 0.05, 0.3)
131    assert _band_rms(killed_late, 55, 65) < 0.3 * _band_rms(open_late, 55, 65)
132    assert _band_rms(killed_early, 85, 95) < 0.3 * _band_rms(open_early, 85, 95)
133    # sanity on the killed render alone: A's bass dominates early, B's late
134    assert _band_rms(killed_early, 55, 65) > 3 * _band_rms(killed_early, 85, 95)
135    assert _band_rms(killed_late, 85, 95) > 3 * _band_rms(killed_late, 55, 65)
136
137
138@pytest.mark.asyncio
139async def test_mid_swaps_between_tracks() -> None:
140    """The mid peaks trade A's 1kHz for B's 2kHz vs an EQ-bypassed render of the same PCM."""
141    fade_out = _tone(1000.0, 45.0).tobytes()  # A: 1kHz "vocal"
142    fade_in = _tone(2000.0, 45.0).tobytes()  # B: 2kHz "vocal"
143    # differential render: identical PCM, one plan whose band data engages the
144    # mid gate and one whose band data bypasses every measured EQ gate -- the
145    # 1k/2k energy difference is then attributable to the mid EQ alone
146    gated_mix, gated = await _render(
147        _analysis_with_mid_bands(120.0, 240.0),
148        _analysis_with_mid_bands(120.0, 240.0),
149        fade_out,
150        fade_in,
151    )
152    open_mix, open_ = await _render(
153        _analysis_with_instrumental_bands(120.0, 240.0),
154        _analysis_with_instrumental_bands(120.0, 240.0),
155        fade_out,
156        fade_in,
157    )
158    assert gated.plan is not None
159    assert gated.plan.eq_plan.mid_out is not None
160    assert gated.plan.eq_plan.mid_in is not None
161    assert open_.plan is not None
162    assert open_.plan.eq_plan.mid_out is None
163    assert open_.plan.eq_plan.mid_in is None
164    # identical geometry: the band data must only change EQ, never the timing
165    assert len(gated_mix) == len(open_mix)
166    # the -8dB depth is modest, so assert a measurable drop (not dominance):
167    # A's 1kHz is attenuated where the swap completes (late); B's 2kHz enters
168    # ducked (early); both measured against the EQ-bypassed render, inside
169    # the crossfade window itself
170    gated_late = _cf_slice(gated_mix, gated, 0.7, 0.95)
171    open_late = _cf_slice(open_mix, open_, 0.7, 0.95)
172    gated_early = _cf_slice(gated_mix, gated, 0.05, 0.3)
173    open_early = _cf_slice(open_mix, open_, 0.05, 0.3)
174    assert _band_rms(gated_late, 950, 1050) < 0.7 * _band_rms(open_late, 950, 1050)
175    assert _band_rms(gated_early, 1950, 2050) < 0.7 * _band_rms(open_early, 1950, 2050)
176