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