/
/
/
1"""
2Tests for the vocal-aware Smart Fades planner.
3
4Covers the FireRed-driven layer on top of the energy/downbeat-aligned
5planner: collision-avoiding candidate search, outgoing-vocal retention,
6the click-free equal-power handoff, and equivalence with the energy-only
7planner when vocal data is missing or invalid.
8"""
9
10from __future__ import annotations
11
12import logging
13
14import numpy as np
15import pytest
16from music_assistant_models.enums import ContentType
17from music_assistant_models.media_items import AudioFormat
18
19from music_assistant.controllers.streams.smart_fades.filters import StreamingCrossfadeFilter
20from music_assistant.controllers.streams.smart_fades.mixer import SmartFadesMixer
21from music_assistant.controllers.streams.smart_fades.models import (
22 TransitionPlan,
23 TransitionStrategy,
24)
25from music_assistant.controllers.streams.smart_fades.planner import SmartCrossFadePlanner
26from music_assistant.controllers.streams.smart_fades.planner.candidates import (
27 CandidateFactory,
28 CandidateSpec,
29 bars_ladder,
30)
31from music_assistant.controllers.streams.smart_fades.planner.context import (
32 build_transition_context,
33)
34from music_assistant.controllers.streams.smart_fades.renderer import TransitionRenderer
35from music_assistant.controllers.streams.smart_fades.vocal import (
36 COLLISION_SECONDS_LIMIT,
37 WEIGHTED_COLLISION_LIMIT,
38)
39from music_assistant.models.audio_analysis import AudioAnalysisData
40
41LOGGER = logging.getLogger(__name__)
42
43
44def _beats(start: float, count: int, interval: float) -> np.ndarray:
45 return np.arange(count, dtype=np.float32) * interval + start
46
47
48def _analysis(
49 bpm: float,
50 duration: float = 240.0,
51 rms_energy: np.ndarray | None = None,
52 key: str | None = "A",
53 mode: str | None = "minor",
54) -> AudioAnalysisData:
55 interval = 60.0 / bpm
56 count = int(duration / interval) + 1
57 beats = _beats(0.0, count, interval)
58 # a normal (non-silent) track with a known key earns the full-blend tier, so a
59 # colliding 16-bar candidate genuinely exercises the remediation ladder
60 energy = rms_energy if rms_energy is not None else np.full(1800, 0.5, dtype=np.float32)
61 return AudioAnalysisData(
62 duration=duration,
63 bpm=bpm,
64 beats=beats.tolist(),
65 downbeats=beats[::4].tolist(),
66 rms_energy=energy.tolist(),
67 key=key,
68 mode=mode,
69 )
70
71
72def _vocal_probabilities(
73 duration: float,
74 active_windows: list[tuple[float, float]],
75 level: float = 0.9,
76) -> list[float]:
77 """Build a probability timeline that is quiet except inside ``active_windows``."""
78 n_frames = 1800
79 frame_duration = duration / n_frames
80 probabilities = [0.05] * n_frames
81 for start, end in active_windows:
82 start_index = max(0, int(start / frame_duration))
83 end_index = min(n_frames, int(end / frame_duration) + 1)
84 for i in range(start_index, end_index):
85 probabilities[i] = level
86 return probabilities
87
88
89def _with_vocal_activity(
90 analysis: AudioAnalysisData,
91 active_windows: list[tuple[float, float]],
92 level: float = 0.9,
93) -> AudioAnalysisData:
94 """Attach a valid vocal_activity list, active only inside ``active_windows``."""
95 assert analysis.duration is not None
96 analysis.extra_data = {
97 "vocal_activity": _vocal_probabilities(analysis.duration, active_windows, level)
98 }
99 return analysis
100
101
102def _rms_with_silence(duration: float, silence_start: float) -> np.ndarray:
103 """1800-bin RMS envelope that drops to near-silence at ``silence_start``."""
104 bins = np.full(1800, 0.5, dtype=np.float32)
105 t = np.linspace(0, duration, 1800)
106 bins[t >= silence_start] = 0.001
107 return bins
108
109
110def _plan(
111 fade_out: AudioAnalysisData, fade_in: AudioAnalysisData, buffer: float = 45.0
112) -> TransitionPlan:
113 return SmartCrossFadePlanner(LOGGER).plan(fade_out, fade_in, buffer)
114
115
116class TestCleanBlendWithoutCollision:
117 """When no vocal collision exists, planning matches the plain energy-only ladder."""
118
119 def test_clean_blend_with_real_vocals_matches_the_energy_ladder(self) -> None:
120 """Tempo-compatible tracks with non-colliding (real-duty) vocals keep the 8-bar blend."""
121 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
122
123 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(220.0, 226.0)])
124 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(35.0, 41.0)])
125 plan = _plan(out, inc)
126
127 assert plan.crossfade_duration == pytest.approx(baseline.crossfade_duration)
128 assert plan.metrics.strategy is TransitionStrategy.ENERGY_ALIGNED
129 assert plan.metrics.collision_seconds == 0.0
130 assert plan.metrics.weighted_collision_seconds == 0.0
131
132 def test_verified_instrumental_pair_earns_the_16_bar_blend(self) -> None:
133 """Near-zero vocal duty on BOTH decks earns the doubled 16-bar overlap."""
134 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [])
135 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [])
136 plan = _plan(out, inc)
137
138 bar = 4 * 60.0 / 120.0
139 assert round(plan.crossfade_duration / bar) == 16
140 assert plan.metrics.strategy is TransitionStrategy.ENERGY_ALIGNED
141
142 def test_one_vocal_deck_denies_the_16_bar_blend(self) -> None:
143 """Real vocal duty on either side keeps the corpus-backed 8-bar default."""
144 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 230.0)])
145 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [])
146 plan = _plan(out, inc)
147
148 bar = 4 * 60.0 / 120.0
149 assert round(plan.crossfade_duration / bar) <= 8
150
151
152class TestVocalCollisionAvoidance:
153 """A colliding candidate is rejected in favor of a smaller, collision-free one."""
154
155 def test_collision_on_the_largest_candidate_falls_back_to_a_smaller_one(self) -> None:
156 """
157 Near-instrumental decks whose lone phrases collide only at 16 bars drop a rung.
158
159 Both decks' vocal duty is under the instrumental threshold, so the pair
160 earns the 16-bar overlap â where A's phrase (media 228s â rendered ~20s)
161 stacks exactly on B's phrase at 20s. The 8-bar overlap misses both.
162 """
163 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(228.0, 230.05)])
164 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(20.0, 22.05)])
165 plan = _plan(out, inc)
166
167 # collision resolved either by a shorter rung or by an anchor move;
168 # the invariant is a collision-free plan, not the specific lever
169 assert plan.metrics.collision_seconds < COLLISION_SECONDS_LIMIT
170 assert plan.metrics.weighted_collision_seconds < WEIGHTED_COLLISION_LIMIT
171 assert plan.metrics.strategy is TransitionStrategy.ENERGY_ALIGNED
172 assert plan.metrics.collision_seconds == 0.0
173
174 def test_the_rejected_larger_candidate_would_indeed_have_collided(self) -> None:
175 """Confirm the 16-bar candidate this scenario skips really does breach the guard."""
176 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(228.0, 230.05)])
177 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(20.0, 22.05)])
178 ctx = build_transition_context(out, inc, 45.0, LOGGER)
179 candidate = CandidateFactory(ctx, LOGGER).build(
180 CandidateSpec(tier=ctx.tier, bars=16, anchor_s=None, entry_s=None)
181 )
182 assert candidate is not None
183 assert (
184 candidate.metrics.collision_seconds >= COLLISION_SECONDS_LIMIT
185 or candidate.metrics.weighted_collision_seconds >= WEIGHTED_COLLISION_LIMIT
186 )
187
188
189class TestShortVocalHandoff:
190 """When every phrased candidate collides, ship the click-free equal-power fallback."""
191
192 def test_short_handoff_when_every_candidate_collides(self) -> None:
193 """Vocals spanning almost the whole tail/head collide at every ladder rung."""
194 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 239.9)])
195 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(0.0, 40.0)])
196 plan = _plan(out, inc)
197
198 assert plan.metrics.strategy is TransitionStrategy.SHORT_VOCAL_HANDOFF
199 assert 0.4 <= plan.crossfade_duration <= 1.0
200 assert not plan.tempo_plan.steps
201 assert plan.fadein_trim_start is None
202 eq = plan.eq_plan
203 assert eq.low_out is None
204 assert eq.low_in is None
205 assert eq.high_out is None
206 assert eq.high_in is None
207 assert eq.mid_out is None
208 assert eq.mid_in is None
209
210 def test_handoff_still_uses_the_qsin_crossfade_filter(self) -> None:
211 """The handoff plan renders through the same equal-power StreamingCrossfadeFilter as any other."""
212 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 239.9)])
213 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(0.0, 40.0)])
214 plan = _plan(out, inc)
215 assert plan.metrics.strategy is TransitionStrategy.SHORT_VOCAL_HANDOFF
216
217 pcm_format = AudioFormat(
218 content_type=ContentType.PCM_F32LE, sample_rate=48000, bit_depth=32, channels=2
219 )
220 filters, _timing = TransitionRenderer(LOGGER).render(
221 plan, pcm_format, fade_in_bytes_len=int(45.0 * pcm_format.pcm_sample_size)
222 )
223 crossfade_filters = [f for f in filters if isinstance(f, StreamingCrossfadeFilter)]
224 assert len(crossfade_filters) == 1
225
226
227class TestOutgoingVocalRetention:
228 """The outgoing track's own vocal is never truncated, even at the cost of a longer tail."""
229
230 def test_retention_extends_the_anchor_into_the_downbeat_snap_gap(self) -> None:
231 """
232 A vocal ending just past the downbeat-snapped anchor pulls the anchor back out.
233
234 The RMS-audible boundary sits slightly after the nearest downbeat the
235 energy-only planner would normally snap to; when FireRed shows the
236 outgoing vocal running into that gap, the anchor must move to cover it,
237 never past the RMS boundary itself.
238 """
239 duration = 200.0
240 bpm = 100.0
241 probe_out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
242 probe_ctx = build_transition_context(probe_out, _analysis(bpm, duration), 45.0, LOGGER)
243 assert probe_ctx.audio_end > probe_ctx.default_anchor + 0.1, "fixture needs a real snap gap"
244
245 buffer_offset = duration - 45.0
246 vocal_end_buffer_local = (probe_ctx.default_anchor + probe_ctx.audio_end) / 2
247 vocal_end_media = vocal_end_buffer_local + buffer_offset
248
249 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
250 out = _with_vocal_activity(out, [(vocal_end_media - 0.5, vocal_end_media)])
251 inc = _with_vocal_activity(_analysis(bpm, duration), [])
252
253 plan = _plan(out, inc, 45.0)
254 assert plan.fade_out_window >= vocal_end_buffer_local - 1e-6
255 assert plan.fade_out_window > probe_ctx.default_anchor + 1e-6
256
257 def test_retention_never_exceeds_the_rms_audible_boundary(self) -> None:
258 """Even a vocal claiming to run past the RMS boundary caps the anchor at that boundary."""
259 duration = 200.0
260 bpm = 100.0
261 probe_ctx = build_transition_context(
262 _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3)),
263 _analysis(bpm, duration),
264 45.0,
265 LOGGER,
266 )
267
268 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
269 # a vocal window reaching all the way to the buffer end (past audio_end)
270 out = _with_vocal_activity(out, [(duration - 5.0, duration - 0.05)])
271 inc = _with_vocal_activity(_analysis(bpm, duration), [])
272
273 plan = _plan(out, inc, 45.0)
274 assert plan.fade_out_window <= probe_ctx.audio_end + 1e-6
275
276 def test_retention_bytes_counts_a_weak_trailing_run(self) -> None:
277 """
278 A trailing run too weak for the planner's veto gate still protects retention.
279
280 The gate exists so a spurious window can't veto an entire blend; here a
281 window only decides how much outgoing audio survives silence removal, so
282 a weak-but-real run (peak and mean both below the planner's confidence
283 floors) must still count instead of yielding zero retention.
284 """
285 duration = 200.0
286 bpm = 100.0
287 buffer_seconds = 45.0
288 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
289 out = _with_vocal_activity(out, [(185.0, 187.0)], level=0.6)
290
291 pcm_format = AudioFormat(
292 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=2
293 )
294 fade_out_bytes_len = int(buffer_seconds * pcm_format.pcm_sample_size)
295
296 retention = SmartFadesMixer._get_vocal_retention_bytes(out, fade_out_bytes_len, pcm_format)
297 assert retention > 0
298
299
300class TestScheduleRecomputationAfterAnchorMovement:
301 """A re-anchored candidate is rebuilt whole, not patched â every schedule reflects it."""
302
303 def test_eq_and_trim_schedules_match_the_new_anchor(self) -> None:
304 """After retention moves the anchor, EQ/trim bounds are consistent with the new plan."""
305 duration = 200.0
306 bpm = 100.0
307 probe_out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
308 probe_ctx = build_transition_context(probe_out, _analysis(bpm, duration), 45.0, LOGGER)
309 assert probe_ctx.audio_end > probe_ctx.default_anchor + 0.1, "fixture needs a real snap gap"
310
311 buffer_offset = duration - 45.0
312 vocal_end_buffer_local = (probe_ctx.default_anchor + probe_ctx.audio_end) / 2
313 vocal_end_media = vocal_end_buffer_local + buffer_offset
314
315 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
316 out = _with_vocal_activity(out, [(vocal_end_media - 0.5, vocal_end_media)])
317 inc = _with_vocal_activity(_analysis(bpm, duration), [])
318
319 plan = _plan(out, inc, 45.0)
320 # the anchor really did move, so this exercises the rebuild, not a no-op
321 assert plan.fade_out_window > probe_ctx.default_anchor + 1e-6
322
323 assert 0.0 <= plan.eq_plan.swap_at <= plan.crossfade_duration
324 if plan.eq_plan.low_out is not None:
325 assert plan.eq_plan.low_out.steps[-1][0] <= plan.fade_out_window + 1e-6
326 if plan.eq_plan.low_in is not None:
327 assert plan.eq_plan.low_in.steps[-1][0] <= plan.crossfade_duration + 1e-6
328 if plan.tempo_plan:
329 crossfade_start = plan.fade_out_window - plan.crossfade_duration
330 assert plan.tempo_plan.steps[-1][0] <= crossfade_start + 1e-6
331
332 def test_rebuilt_candidate_is_not_a_patched_copy_of_the_pristine_one(self) -> None:
333 """The rebuilt plan's timing differs from the pristine (unprotected) anchor entirely."""
334 duration = 200.0
335 bpm = 100.0
336 probe_out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
337 probe_ctx = build_transition_context(probe_out, _analysis(bpm, duration), 45.0, LOGGER)
338
339 buffer_offset = duration - 45.0
340 vocal_end_buffer_local = (probe_ctx.default_anchor + probe_ctx.audio_end) / 2
341 vocal_end_media = vocal_end_buffer_local + buffer_offset
342
343 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
344 out = _with_vocal_activity(out, [(vocal_end_media - 0.5, vocal_end_media)])
345 inc = _with_vocal_activity(_analysis(bpm, duration), [])
346
347 plan = SmartCrossFadePlanner(LOGGER).plan(out, inc, 45.0)
348 ctx = build_transition_context(out, inc, 45.0, LOGGER)
349 # the pristine (unprotected) candidate always anchors at the energy-only
350 # default, ignoring the vocal retention the winning plan applied
351 pristine_candidate = CandidateFactory(ctx, LOGGER).build(
352 CandidateSpec(tier=ctx.tier, bars=1, anchor_s=None, entry_s=None)
353 )
354 assert pristine_candidate is not None
355 assert pristine_candidate.plan.fade_out_window != plan.fade_out_window
356 assert pristine_candidate.plan.fadeout_trim != plan.fadeout_trim
357
358 def test_rebuild_preserves_a_remediated_incoming_entry(self) -> None:
359 """Moving the outgoing anchor keeps the incoming entry selected by remediation."""
360 duration = 200.0
361 bpm = 100.0
362 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 190.3))
363 out = _with_vocal_activity(out, [(189.0, 190.0)])
364 inc = _with_vocal_activity(_analysis(bpm, duration), [])
365
366 ctx = build_transition_context(out, inc, 45.0, LOGGER)
367 factory = CandidateFactory(ctx, LOGGER)
368 pristine = factory.build(CandidateSpec(tier=ctx.tier, bars=2, anchor_s=None, entry_s=4.8))
369 assert pristine is not None
370 assert pristine.plan.fadein_trim_start == pytest.approx(4.8)
371
372 # the old planner's protection rebuilt the candidate at the protective
373 # anchor with its entry pinned; the factory must honor the same spec
374 # combination instead of silently resetting the entry
375 vocal_end = min(ctx.vocal_out_placement.last_end(), ctx.audio_end) # type: ignore[union-attr]
376 anchor = next(db for db in ctx.protective_downbeats if db >= vocal_end)
377 protected = factory.build(
378 CandidateSpec(tier=ctx.tier, bars=2, anchor_s=anchor, entry_s=4.8)
379 )
380 assert protected is not None
381 assert protected.plan.fade_out_window > pristine.plan.fade_out_window
382 assert protected.plan.fadein_trim_start == pytest.approx(4.8)
383
384
385class TestShortFadeAudibleTrimBound:
386 """For crossfades at or under 8s, dropped audible material never exceeds the overlap."""
387
388 def test_audible_trim_never_exceeds_the_overlap_length(self) -> None:
389 """
390 A tiny forced overlap must not drop more audible material than it covers.
391
392 A sparse outgoing downbeat grid combined with a fast incoming track forces a
393 tiny synthetic overlap far shorter than the downbeat-snap gap; the planner
394 must re-anchor so the audible material dropped never exceeds that overlap.
395 """
396 duration = 200.0
397 buffer_duration = 45.0
398 out_bpm = 100.0
399 in_bpm = 300.0
400
401 dense_beats = np.arange(0.0, 150.0, 0.6, dtype=np.float32)
402 sparse_downbeat_media = 191.5 # one real downbeat, 3.5s before the RMS boundary
403 out = AudioAnalysisData(
404 duration=duration,
405 bpm=out_bpm,
406 beats=np.concatenate([dense_beats, [sparse_downbeat_media]]).tolist(),
407 downbeats=sorted([sparse_downbeat_media, *dense_beats[::4].tolist()]),
408 rms_energy=_rms_with_silence(duration, 195.0).tolist(),
409 )
410 out = _with_vocal_activity(out, [])
411 inc = AudioAnalysisData(
412 duration=duration,
413 bpm=in_bpm,
414 beats=np.arange(0.0, 60.0, 60.0 / in_bpm, dtype=np.float32).tolist(),
415 downbeats=np.arange(0.0, 60.0, 4 * 60.0 / in_bpm, dtype=np.float32).tolist(),
416 )
417 inc = _with_vocal_activity(inc, [])
418
419 ctx = build_transition_context(out, inc, buffer_duration, LOGGER)
420 naive_candidate = CandidateFactory(ctx, LOGGER).build(
421 CandidateSpec(tier=ctx.tier, bars=1, anchor_s=None, entry_s=None)
422 )
423 assert naive_candidate is not None
424 naive_gap = ctx.audio_end - naive_candidate.plan.fade_out_window
425 assert naive_gap > naive_candidate.plan.crossfade_duration, "fixture needs a real violation"
426
427 protected = SmartCrossFadePlanner(LOGGER).plan(out, inc, buffer_duration)
428 assert (
429 protected.metrics.audible_outgoing_trim
430 <= naive_candidate.plan.crossfade_duration + 1e-6
431 )
432
433
434class TestHighHopesStyleFalsePositive:
435 """FireRed activity in a genuinely silent tail must never extend the audible boundary."""
436
437 def test_false_positive_past_the_rms_boundary_does_not_move_the_anchor(self) -> None:
438 """A high-confidence FireRed run entirely inside the RMS-silent tail is ignored."""
439 duration = 240.0
440 bpm = 120.0
441 baseline = _plan(
442 _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 225.0)),
443 _analysis(bpm, duration),
444 )
445
446 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 225.0))
447 # FireRed hallucinates vocal-like activity in the low-energy tail, past the boundary
448 out = _with_vocal_activity(out, [(229.5, 238.5)])
449 inc = _with_vocal_activity(_analysis(bpm, duration), [])
450
451 plan = _plan(out, inc)
452 assert plan.fade_out_window == pytest.approx(baseline.fade_out_window)
453 assert plan.crossfade_duration == pytest.approx(baseline.crossfade_duration)
454
455 def test_false_positive_window_is_dropped_entirely_from_the_mask(self) -> None:
456 """The clamp drops the false-positive window outright rather than clipping a sliver."""
457 duration = 240.0
458 bpm = 120.0
459 out = _analysis(bpm, duration, rms_energy=_rms_with_silence(duration, 225.0))
460 out = _with_vocal_activity(out, [(229.5, 238.5)])
461 inc = _with_vocal_activity(_analysis(bpm, duration), [])
462
463 ctx = build_transition_context(out, inc, 45.0, LOGGER)
464 assert ctx.vocal_out_placement is not None
465 assert ctx.vocal_out_placement.windows == []
466
467
468class TestRemediationAltersTheCandidate:
469 """A colliding candidate 0 is remediated deterministically (bar rung / entry / anchor)."""
470
471 def test_collision_remediation_drops_the_bar_rung_deterministically(self) -> None:
472 """The colliding 16-bar candidate 0 is remediated to a shorter, collision-free overlap."""
473 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(228.0, 230.05)])
474 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(20.0, 22.05)])
475
476 ctx = build_transition_context(out, inc, 45.0, LOGGER)
477 bars0 = bars_ladder(ctx, ctx.tier)[0]
478 cand0 = CandidateFactory(ctx, LOGGER).build(
479 CandidateSpec(tier=ctx.tier, bars=bars0, anchor_s=None, entry_s=None)
480 )
481 assert cand0 is not None
482 # the full-blend candidate 0 spans 16 bars and genuinely collides
483 assert bars0 == 16
484 assert round(cand0.plan.crossfade_duration / (4 * 60.0 / 120.0)) == 16
485 assert (
486 cand0.metrics.collision_seconds >= COLLISION_SECONDS_LIMIT
487 or cand0.metrics.weighted_collision_seconds >= WEIGHTED_COLLISION_LIMIT
488 )
489
490 plan = _plan(out, inc)
491 # the shipped plan resolves the collision deterministically: either a
492 # shorter overlap or a moved anchor, never the colliding candidate 0
493 assert plan.metrics.collision_seconds < COLLISION_SECONDS_LIMIT
494 assert plan.metrics.weighted_collision_seconds < WEIGHTED_COLLISION_LIMIT
495 assert (
496 plan.crossfade_duration < cand0.plan.crossfade_duration
497 or plan.fade_out_window != cand0.plan.fade_out_window
498 )
499 assert plan.metrics.strategy is TransitionStrategy.ENERGY_ALIGNED
500 assert plan.metrics.collision_seconds < COLLISION_SECONDS_LIMIT
501 assert plan.metrics.weighted_collision_seconds < WEIGHTED_COLLISION_LIMIT
502 # the search is a pure function of the inputs
503 assert _plan(out, inc) == plan
504
505
506class TestMissingOrInvalidVocalDataFallsBackToEnergyOnly:
507 """Any defect in either side's vocal timeline disables vocal logic entirely."""
508
509 def test_missing_vocal_activity_matches_the_energy_only_plan(self) -> None:
510 """No vocal_activity at all on either side yields exactly the energy-only plan."""
511 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
512 plan = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
513 assert plan == baseline
514
515 def test_old_wrapped_contract_matches_the_energy_only_plan(self) -> None:
516 """A row using the old wrapped contract disables vocal logic."""
517 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
518 out = _analysis(120.0, duration=240.0)
519 out.extra_data = {
520 "vocal_activity": {
521 "model": "some_other_model",
522 "frame_duration": 0.1,
523 "probabilities": [0.9] * 2400,
524 }
525 }
526 plan = _plan(out, _analysis(120.0, duration=240.0))
527 assert plan == baseline
528
529 def test_non_list_timeline_matches_the_energy_only_plan(self) -> None:
530 """A tuple timeline is malformed because the provider contract stores a list."""
531 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
532 out = _analysis(120.0, duration=240.0)
533 out.extra_data = {"vocal_activity": tuple([0.9] * 1800)}
534 plan = _plan(out, _analysis(120.0, duration=240.0))
535 assert plan == baseline
536
537 def test_partial_timeline_matches_the_energy_only_plan(self) -> None:
538 """A timeline with fewer than 1800 bins is rejected."""
539 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
540 out = _analysis(120.0, duration=240.0)
541 out.extra_data = {"vocal_activity": [0.9] * 1000}
542 plan = _plan(out, _analysis(120.0, duration=240.0))
543 assert plan == baseline
544
545 def test_one_sided_valid_data_engages_that_sides_protection(self) -> None:
546 """Valid data on one side engages that side's protections independently."""
547 baseline = _plan(_analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0))
548 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(238.0, 239.0)])
549 plan = _plan(out, _analysis(120.0, duration=240.0))
550 # this vocal rides the fade rather than forcing a re-anchor, so the
551 # geometry matches the energy-only baseline while the outgoing vocal
552 # is now measured; cross-deck collision needs both sides and stays 0
553 assert plan.fade_out_window == baseline.fade_out_window
554 assert plan.crossfade_duration == baseline.crossfade_duration
555 assert plan.metrics.outgoing_vocal_fade_seconds > 0.0
556 assert plan.metrics.collision_seconds == 0.0
557
558
559def test_outgoing_vocal_metrics_survive_a_missing_incoming_timeline() -> None:
560 """Outgoing-only vocal data still yields outgoing vocal metrics; collision stays neutral."""
561 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(228.0, 230.05)])
562 inc = _analysis(120.0, duration=240.0)
563 ctx = build_transition_context(out, inc, 45.0, LOGGER)
564 candidate = CandidateFactory(ctx, LOGGER).build(
565 CandidateSpec(tier=ctx.tier, bars=8, anchor_s=None, entry_s=None)
566 )
567 assert candidate is not None
568 assert candidate.metrics.outgoing_vocal_fade_seconds > 0.0
569 assert candidate.metrics.collision_seconds == 0.0
570