/
/
/
1"""Tests for the 1800-bin vocal-activity contract and vocal-collision math."""
2
3from __future__ import annotations
4
5import math
6
7import pytest
8
9from music_assistant.controllers.streams.smart_fades.vocal import (
10 VocalMask,
11 VocalTimeline,
12 build_vocal_windows,
13 collision_metrics,
14 merge_windows,
15 parse_vocal_probabilities,
16)
17from music_assistant.models.audio_analysis import AudioAnalysisData
18
19
20def _analysis(vocal_activity: object, duration: float | None = 240.0) -> AudioAnalysisData:
21 return AudioAnalysisData(duration=duration, extra_data={"vocal_activity": vocal_activity})
22
23
24# Used only by TestBuildVocalWindowsSustainedEvidenceGate below.
25FRAME = 0.1 # 10 bins per second
26
27
28def _timeline(spec: list[tuple[float, int]]) -> list[float]:
29 """Flat probability segments: [(prob, n_bins), ...]."""
30 probs: list[float] = []
31 for prob, bins in spec:
32 probs.extend([prob] * bins)
33 return probs
34
35
36class TestParseVocalProbabilities:
37 """The 1800-bin contract is validated in full; any defect disables vocal logic entirely."""
38
39 @pytest.mark.parametrize("duration", [90.0, 180.0, 240.0, 360.0])
40 def test_valid_contract_infers_timing_from_duration(self, duration: float) -> None:
41 """A valid list round-trips with exact per-bin timing for any canonical duration."""
42 probabilities = [0.1] * 1800
43 analysis = _analysis(probabilities, duration=duration)
44 assert parse_vocal_probabilities(analysis) == VocalTimeline(
45 probabilities=probabilities,
46 frame_duration=duration / 1800,
47 )
48
49 def test_frame_duration_is_derived_exactly_from_analysis_duration(self) -> None:
50 """Stored duration divided by 1800 is the timeline's only timing source."""
51 timeline = parse_vocal_probabilities(_analysis([0.1] * 1800, duration=241.7))
52 assert timeline is not None
53 assert timeline.frame_duration == 241.7 / 1800
54
55 def test_no_extra_data_returns_none(self) -> None:
56 """A row with no extra_data at all has no vocal timeline."""
57 assert parse_vocal_probabilities(AudioAnalysisData(duration=240.0)) is None
58
59 def test_missing_vocal_activity_key_returns_none(self) -> None:
60 """extra_data present but without a vocal_activity entry."""
61 analysis = AudioAnalysisData(duration=240.0, extra_data={"band_rms": {}})
62 assert parse_vocal_probabilities(analysis) is None
63
64 @pytest.mark.parametrize(
65 "vocal_activity",
66 [
67 "nope",
68 0.5,
69 tuple([0.1] * 1800),
70 {"model": "firered_aed", "frame_duration": 0.1, "probabilities": [0.1] * 1800},
71 ],
72 )
73 def test_non_list_or_old_contract_returns_none(self, vocal_activity: object) -> None:
74 """Only the direct list contract is accepted; old wrapped rows are stale."""
75 assert parse_vocal_probabilities(_analysis(vocal_activity)) is None
76
77 @pytest.mark.parametrize("length", [0, 1, 1799, 1801, 2400])
78 def test_wrong_probability_count_returns_none(self, length: int) -> None:
79 """The canonical timeline must contain exactly 1800 bins."""
80 assert parse_vocal_probabilities(_analysis([0.1] * length)) is None
81
82 @pytest.mark.parametrize(
83 "bad_value",
84 [-0.1, 1.1, float("nan"), float("inf"), "0.5", True],
85 )
86 def test_out_of_range_or_non_finite_value_returns_none(self, bad_value: object) -> None:
87 """Any bin outside 0..1, non-finite, non-numeric, or bool invalidates the row."""
88 probabilities: list[object] = [0.1] * 1800
89 probabilities[900] = bad_value
90 assert parse_vocal_probabilities(_analysis(probabilities)) is None
91
92 def test_huge_integer_returns_none_instead_of_raising(self) -> None:
93 """An arbitrary-size malformed integer is rejected before float conversion."""
94 probabilities: list[object] = [0.1] * 1800
95 probabilities[900] = 10**10000
96 assert parse_vocal_probabilities(_analysis(probabilities)) is None
97
98 def test_boundary_values_zero_and_one_are_accepted(self) -> None:
99 """0.0 and 1.0 are valid probabilities, not out-of-range."""
100 probabilities = [0.1] * 1800
101 probabilities[0] = 0.0
102 probabilities[1] = 1.0
103 assert parse_vocal_probabilities(_analysis(probabilities)) is not None
104
105 @pytest.mark.parametrize(
106 "duration",
107 [None, 0.0, -1.0, float("nan"), float("inf"), True],
108 )
109 def test_missing_or_invalid_duration_returns_none(self, duration: float | None) -> None:
110 """Bin timing requires a finite positive analysis duration."""
111 assert parse_vocal_probabilities(_analysis([0.1] * 1800, duration=duration)) is None
112
113
114class TestBuildVocalWindows:
115 """Hysteresis run detection, padding and gap bridging over a probability timeline."""
116
117 def test_single_run_is_padded_left_and_right(self) -> None:
118 """A run opens at OPEN and closes below CLOSE; the window gets padded both sides."""
119 probabilities = [0.0] * 100
120 for i in range(40, 60):
121 probabilities[i] = 0.9
122 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
123 assert len(mask.windows) == 1
124 left, right = mask.windows[0]
125 assert left == pytest.approx(4.0 - 0.25)
126 assert right == pytest.approx(6.0 + 0.75)
127
128 def test_probability_never_reaching_open_threshold_yields_no_window(self) -> None:
129 """A run must reach the OPEN threshold, not just exceed CLOSE."""
130 probabilities = [0.4] * 100
131 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
132 assert mask.windows == []
133
134 def test_sub_min_run_blip_yields_no_window(self) -> None:
135 """A one-off detector blip (crowd shout, vocal-formant one-shot) is not a phrase."""
136 probabilities = [0.0] * 100
137 probabilities[50] = 0.9
138 probabilities[51] = 0.9 # 0.2s: still under the 0.3s run floor
139 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
140 assert mask.windows == []
141
142 @pytest.mark.parametrize("start", [0, 1, 17, 50, 89])
143 def test_min_run_length_is_stable_at_every_timeline_position(self, start: int) -> None:
144 """Identical runs exactly at the floor survive regardless of their absolute index."""
145 probabilities = [0.0] * 100
146 for i in range(start, start + 3):
147 probabilities[i] = 0.9
148 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
149 assert len(mask.windows) == 1
150
151 @pytest.mark.parametrize(
152 ("frame_duration", "short_frames", "minimum_frames"),
153 [
154 (90.0 / 1800, 5, 6),
155 (240.0 / 1800, 2, 3),
156 (360.0 / 1800, 1, 2),
157 ],
158 )
159 def test_min_run_remains_seconds_based_for_variable_bin_durations(
160 self,
161 frame_duration: float,
162 short_frames: int,
163 minimum_frames: int,
164 ) -> None:
165 """The 0.3-second floor maps to the correct whole-bin count at each duration."""
166 short = [0.0] * 100
167 short[20 : 20 + short_frames] = [0.9] * short_frames
168 sufficient = [0.0] * 100
169 sufficient[20 : 20 + minimum_frames] = [0.9] * minimum_frames
170
171 end_s = len(short) * frame_duration
172 assert build_vocal_windows(short, frame_duration, 0.0, end_s).windows == []
173 assert len(build_vocal_windows(sufficient, frame_duration, 0.0, end_s).windows) == 1
174
175 def test_dip_below_open_but_above_close_does_not_split_the_run(self) -> None:
176 """Hysteresis: a dip that stays above CLOSE keeps the run open (no re-triggering)."""
177 probabilities = [0.0] * 100
178 for i in range(10, 20):
179 probabilities[i] = 0.9
180 for i in range(20, 25):
181 probabilities[i] = 0.4 # between CLOSE (0.3) and OPEN (0.5): stays open
182 for i in range(25, 35):
183 probabilities[i] = 0.9
184 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
185 assert len(mask.windows) == 1
186
187 def test_close_gap_bridges_into_one_window(self) -> None:
188 """Two runs closer than the gap bridge into a single window."""
189 probabilities = [0.0] * 200
190 for i in range(10, 20):
191 probabilities[i] = 0.9
192 for i in range(23, 33): # 0.3s gap after padding, under the 0.5s minimum gap
193 probabilities[i] = 0.9
194 mask = build_vocal_windows(probabilities, 0.1, 0.0, 20.0, beat_duration=0.5)
195 assert len(mask.windows) == 1
196
197 def test_far_apart_runs_stay_separate(self) -> None:
198 """Two runs well beyond the max gap remain distinct windows."""
199 probabilities = [0.0] * 200
200 for i in range(10, 20):
201 probabilities[i] = 0.9
202 for i in range(150, 160):
203 probabilities[i] = 0.9
204 mask = build_vocal_windows(probabilities, 0.1, 0.0, 20.0, beat_duration=0.5)
205 assert len(mask.windows) == 2
206
207 def test_gap_bridge_is_clamped_between_min_and_max(self) -> None:
208 """The bridge distance follows the beat length but is clamped to [min_gap, max_gap]."""
209 probabilities = [0.0] * 300
210 for i in range(10, 20):
211 probabilities[i] = 0.9
212 # padded gap of 0.7s: strictly between min_gap (0.5) and max_gap (1.0), so a tiny
213 # beat_duration (clamped up to min_gap) does not bridge but a long one does
214 for i in range(37, 47):
215 probabilities[i] = 0.9
216 tight = build_vocal_windows(probabilities, 0.1, 0.0, 30.0, beat_duration=0.1)
217 loose = build_vocal_windows(probabilities, 0.1, 0.0, 30.0, beat_duration=5.0)
218 assert len(tight.windows) == 2
219 assert len(loose.windows) == 1
220
221 def test_windows_clipped_to_range(self) -> None:
222 """A run whose padding would spill past start_s/end_s is clipped, not dropped."""
223 probabilities = [0.0] * 100
224 for i in range(5):
225 probabilities[i] = 0.9
226 mask = build_vocal_windows(probabilities, 0.1, 0.0, 10.0)
227 assert mask.windows[0][0] == 0.0 # left padding clipped at the range start
228
229 def test_trailing_run_without_a_close_frame_still_closes(self) -> None:
230 """A run still active at the end of the timeline closes at its last frame."""
231 probabilities = [0.0] * 50
232 for i in range(40, 50):
233 probabilities[i] = 0.9
234 mask = build_vocal_windows(probabilities, 0.1, 0.0, 5.0)
235 assert len(mask.windows) == 1
236 assert mask.windows[0][1] == pytest.approx(5.0)
237
238
239class TestBuildVocalWindowsSustainedEvidenceGate:
240 """A run only becomes a window when its peak OR its mean shows real confidence."""
241
242 def test_weak_backing_run_is_dropped(self) -> None:
243 """A 1.5s run peaking 0.69 (background 'aahh') never becomes a window."""
244 probs = _timeline([(0.0, 100), (0.55, 10), (0.69, 5), (0.0, 100)])
245 mask = build_vocal_windows(probs, FRAME, 0.0, len(probs) * FRAME)
246 assert mask.windows == []
247
248 def test_confident_peak_run_is_kept(self) -> None:
249 """A short run with a confident peak (real singing) still opens a window."""
250 probs = _timeline([(0.0, 100), (0.95, 15), (0.0, 100)])
251 mask = build_vocal_windows(probs, FRAME, 0.0, len(probs) * FRAME)
252 assert len(mask.windows) == 1
253
254 def test_sustained_medium_run_is_kept(self) -> None:
255 """A long medium-confidence run (soft verse, mean >= 0.65) is kept."""
256 # 0.70, not 0.65, so the assertion doesn't ride the mean-threshold boundary
257 probs = _timeline([(0.0, 50), (0.70, 80), (0.0, 50)])
258 mask = build_vocal_windows(probs, FRAME, 0.0, len(probs) * FRAME)
259 assert len(mask.windows) == 1
260
261 def test_moderate_mean_backing_run_is_dropped(self) -> None:
262 """A ~1.1s run just under both floors (peak 0.84, mean ~0.648) is dropped."""
263 probs = _timeline([(0.0, 100), (0.55, 5), (0.62, 3), (0.84, 3), (0.0, 100)])
264 mask = build_vocal_windows(probs, FRAME, 0.0, len(probs) * FRAME)
265 assert mask.windows == []
266
267
268class TestVocalMaskClampedTo:
269 """The hard boundary clamp: FireRed may never claim activity past a given bound."""
270
271 def test_window_entirely_past_bound_is_dropped(self) -> None:
272 """A window starting at/after the bound is removed outright."""
273 mask = VocalMask(windows=[(10.0, 12.0)])
274 assert mask.clamped_to(10.0).windows == []
275
276 def test_window_straddling_bound_is_clipped(self) -> None:
277 """A window that starts before the bound but reaches past it is clipped, not dropped."""
278 mask = VocalMask(windows=[(8.0, 12.0)])
279 assert mask.clamped_to(10.0).windows == [(8.0, 10.0)]
280
281 def test_window_fully_inside_bound_is_untouched(self) -> None:
282 """A window entirely within the bound passes through unchanged."""
283 mask = VocalMask(windows=[(1.0, 2.0)])
284 assert mask.clamped_to(10.0).windows == [(1.0, 2.0)]
285
286 def test_last_end_of_empty_mask_is_zero(self) -> None:
287 """An empty mask reports no activity."""
288 assert VocalMask(windows=[]).last_end() == 0.0
289
290 def test_last_end_returns_final_window_end(self) -> None:
291 """last_end reflects the rightmost window's end."""
292 assert VocalMask(windows=[(1.0, 2.0), (5.0, 8.5)]).last_end() == 8.5
293
294
295class TestMergeWindows:
296 """Interval merging that both collision_metrics and the planner's metrics rely on."""
297
298 def test_overlapping_windows_merge(self) -> None:
299 """Overlapping intervals combine into one."""
300 assert merge_windows([(0.0, 5.0), (3.0, 8.0)]) == [(0.0, 8.0)]
301
302 def test_touching_windows_merge(self) -> None:
303 """Intervals that exactly touch at the boundary also merge."""
304 assert merge_windows([(0.0, 5.0), (5.0, 8.0)]) == [(0.0, 8.0)]
305
306 def test_disjoint_windows_stay_separate(self) -> None:
307 """Non-overlapping intervals are returned independently, sorted."""
308 assert merge_windows([(5.0, 8.0), (0.0, 2.0)]) == [(0.0, 2.0), (5.0, 8.0)]
309
310 def test_empty_input_returns_empty(self) -> None:
311 """No intervals in, none out."""
312 assert merge_windows([]) == []
313
314
315class TestCollisionMetrics:
316 """The two-sided vocal-collision guard: simultaneous seconds and gain-weighted seconds."""
317
318 def test_no_overlap_yields_zero(self) -> None:
319 """Disjoint outgoing/incoming windows collide for zero seconds."""
320 seconds, weighted = collision_metrics([(0.0, 2.0)], [(5.0, 7.0)], duration=10.0)
321 assert seconds == 0.0
322 assert weighted == 0.0
323
324 def test_full_overlap_matches_analytic_integral(self) -> None:
325 """A full-duration overlap on both sides integrates to the exact analytic value."""
326 duration = 10.0
327 seconds, weighted = collision_metrics([(0.0, duration)], [(0.0, duration)], duration)
328 assert seconds == pytest.approx(duration)
329 # closed-form integral of 4p(1-p) over p in [0,1] is 2/3
330 assert weighted == pytest.approx(duration * 2.0 / 3.0)
331
332 def test_partial_overlap_is_clipped_to_the_render_window(self) -> None:
333 """Windows extending outside [0, duration] are clipped before scoring."""
334 seconds, _ = collision_metrics([(-5.0, 3.0)], [(1.0, 20.0)], duration=10.0)
335 assert seconds == pytest.approx(2.0) # overlap is [1,3]
336
337 def test_zero_duration_yields_zero(self) -> None:
338 """A zero-length crossfade collides for zero seconds, avoiding a division by zero."""
339 seconds, weighted = collision_metrics([(0.0, 1.0)], [(0.0, 1.0)], duration=0.0)
340 assert seconds == 0.0
341 assert weighted == 0.0
342
343 def test_overlapping_outgoing_windows_are_not_double_counted(self) -> None:
344 """Two outgoing windows both colliding with the same incoming window merge first."""
345 seconds, _ = collision_metrics([(0.0, 4.0), (3.0, 6.0)], [(0.0, 10.0)], duration=10.0)
346 assert seconds == pytest.approx(6.0)
347
348 def test_weight_peaks_at_the_midpoint(self) -> None:
349 """The simultaneous-power weight is highest at the crossfade's midpoint."""
350 duration = 10.0
351 _, edge = collision_metrics([(0.0, 1.0)], [(0.0, 1.0)], duration)
352 _, middle = collision_metrics([(4.5, 5.5)], [(4.5, 5.5)], duration)
353 assert middle > edge
354
355 def test_matches_manual_analytic_primitive(self) -> None:
356 """Cross-check against the hand-derived antiderivative for an arbitrary interval."""
357
358 def primitive(t: float, duration: float) -> float:
359 phase = t / duration
360 return duration * (2.0 * phase**2 - (4.0 / 3.0) * phase**3)
361
362 duration, left, right = 12.0, 3.2, 7.9
363 _, weighted = collision_metrics([(left, right)], [(left, right)], duration)
364 expected = primitive(right, duration) - primitive(left, duration)
365 assert weighted == pytest.approx(expected)
366 assert math.isfinite(weighted)
367