/
/
/
1"""Tests for the candidate generators' rung emission."""
2
3from __future__ import annotations
4
5import logging
6
7from music_assistant.controllers.streams.smart_fades.models import (
8 TransitionStrategy,
9 TransitionTier,
10)
11from music_assistant.controllers.streams.smart_fades.planner.candidates import (
12 _LAZY_OVERLAY_SECONDS,
13 EnergyLadderGenerator,
14 LazyOverlayGenerator,
15 TrimClosingAnchorGenerator,
16 _entry_options,
17 _vocal_duties,
18 _window_duties,
19 earns_instrumental_blend,
20)
21from music_assistant.controllers.streams.smart_fades.planner.context import (
22 TransitionContext,
23 build_transition_context,
24)
25from music_assistant.controllers.streams.smart_fades.planner.planner import SmartCrossFadePlanner
26from music_assistant.models.audio_analysis import AudioAnalysisData
27
28
29def _analysis(
30 bpm: float,
31 duration: float = 240.0,
32 grid_until: float | None = None,
33) -> AudioAnalysisData:
34 """Synthetic AudioAnalysisData with an even beat/downbeat grid, optionally truncated early."""
35 interval = 60.0 / bpm
36 count = int(duration / interval) + 1
37 beats = [i * interval for i in range(count)]
38 if grid_until is not None:
39 beats = [b for b in beats if b <= grid_until]
40 return AudioAnalysisData(
41 duration=duration,
42 bpm=bpm,
43 beats=beats,
44 downbeats=beats[::4],
45 beats_per_bar=4,
46 rms_energy=[0.8] * 1800,
47 key="C",
48 mode="minor",
49 extra_data={},
50 )
51
52
53def _instrumental_vs_vocal_ctx() -> TransitionContext:
54 """Build a transition context: outgoing instrumental, incoming vocal, both 128 BPM."""
55 beats = [i * 60 / 128 for i in range(int(180 * 128 / 60))]
56 downbeats = beats[::4]
57 aa_out = AudioAnalysisData(
58 duration=180.0,
59 bpm=128.0,
60 beats=beats,
61 downbeats=downbeats,
62 beats_per_bar=4,
63 rms_energy=[0.8] * 1800,
64 key="C",
65 mode="minor",
66 extra_data={"vocal_activity": [0.0] * 1800},
67 )
68 aa_in = AudioAnalysisData(
69 duration=180.0,
70 bpm=128.0,
71 beats=beats,
72 downbeats=downbeats,
73 beats_per_bar=4,
74 rms_energy=[0.8] * 1800,
75 key="C",
76 mode="minor",
77 extra_data={"vocal_activity": [0.9] * 1800},
78 )
79 return build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
80
81
82def _big_trim_gap_ctx() -> TransitionContext:
83 """
84 Build a context whose energy anchor lands early, stranding audible tail behind it.
85
86 rms_energy holds at 0.9 for the first 70% of the buffer, drops to a
87 still-audible 0.25 until 95%, then to silence - no vocal data, so the
88 gap can only be closed by an energy-path generator.
89 """
90 beats = [i * 60 / 128 for i in range(int(45 * 128 / 60))]
91 downbeats = beats[::4]
92 rms_energy = [0.9] * 1260 + [0.25] * 450 + [0.0] * 90
93 aa_out = AudioAnalysisData(
94 duration=45.0,
95 bpm=128.0,
96 beats=beats,
97 downbeats=downbeats,
98 beats_per_bar=4,
99 rms_energy=rms_energy,
100 key="C",
101 mode="minor",
102 extra_data={},
103 )
104 aa_in = AudioAnalysisData(
105 duration=45.0,
106 bpm=128.0,
107 beats=beats,
108 downbeats=downbeats,
109 beats_per_bar=4,
110 rms_energy=[0.8] * 1800,
111 key="C",
112 mode="minor",
113 extra_data={},
114 )
115 ctx = build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
116 assert ctx.audio_end - ctx.default_anchor >= 8.0
117 return ctx
118
119
120def _small_trim_gap_ctx() -> TransitionContext:
121 """Build a context with flat rms_energy, so the energy anchor already sits at the audible end."""
122 beats = [i * 60 / 128 for i in range(int(45 * 128 / 60))]
123 downbeats = beats[::4]
124 aa_out = AudioAnalysisData(
125 duration=45.0,
126 bpm=128.0,
127 beats=beats,
128 downbeats=downbeats,
129 beats_per_bar=4,
130 rms_energy=[0.8] * 1800,
131 key="C",
132 mode="minor",
133 extra_data={},
134 )
135 aa_in = AudioAnalysisData(
136 duration=45.0,
137 bpm=128.0,
138 beats=beats,
139 downbeats=downbeats,
140 beats_per_bar=4,
141 rms_energy=[0.8] * 1800,
142 key="C",
143 mode="minor",
144 extra_data={},
145 )
146 return build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
147
148
149def _small_positive_trim_gap_ctx() -> TransitionContext:
150 """
151 Build a context whose energy anchor lands a few seconds before the audible end.
152
153 Same shape as the big-gap fixture but with a short mid-tier energy
154 segment, giving a gap under the trim-closing generator's default min gap.
155 """
156 beats = [i * 60 / 128 for i in range(int(45 * 128 / 60))]
157 downbeats = beats[::4]
158 rms_energy = [0.9] * 1550 + [0.25] * 160 + [0.0] * 90
159 aa_out = AudioAnalysisData(
160 duration=45.0,
161 bpm=128.0,
162 beats=beats,
163 downbeats=downbeats,
164 beats_per_bar=4,
165 rms_energy=rms_energy,
166 key="C",
167 mode="minor",
168 extra_data={},
169 )
170 aa_in = AudioAnalysisData(
171 duration=45.0,
172 bpm=128.0,
173 beats=beats,
174 downbeats=downbeats,
175 beats_per_bar=4,
176 rms_energy=[0.8] * 1800,
177 key="C",
178 mode="minor",
179 extra_data={},
180 )
181 ctx = build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
182 assert 0.0 < ctx.audio_end - ctx.default_anchor < 8.0
183 return ctx
184
185
186def test_trim_closing_min_gap_zero_bypasses_the_gate() -> None:
187 """An ungated instance emits the ladder at the audible end even for a small trim gap."""
188 ctx = _small_positive_trim_gap_ctx()
189 assert list(TrimClosingAnchorGenerator().generate(ctx)) == []
190 specs = list(TrimClosingAnchorGenerator(min_gap=0.0).generate(ctx))
191 assert specs
192 assert any(
193 spec.anchor_s == ctx.audio_end and spec.source == "trim-closing-anchor" for spec in specs
194 )
195
196
197def test_energy_ladder_emits_only_plain_rungs() -> None:
198 """A one-instrumental/one-vocal pair gets the plain ladder, no 16-bar spec."""
199 instrumental_vs_vocal_ctx = _instrumental_vs_vocal_ctx()
200 specs = list(EnergyLadderGenerator().generate(instrumental_vs_vocal_ctx))
201 assert specs
202 assert all(spec.bars <= 8 for spec in specs)
203
204
205def test_trim_closing_ladder_emitted_for_big_trim_gap() -> None:
206 """An instrumental tail with a large audible gap past the energy anchor gets late-anchored rungs."""
207 ctx = _big_trim_gap_ctx()
208 specs = list(TrimClosingAnchorGenerator().generate(ctx))
209 assert specs
210 for spec in specs:
211 assert spec.anchor_s is not None
212 assert spec.anchor_s > ctx.default_anchor
213 assert spec.anchor_s <= ctx.audio_end
214 # the ladder is walked, not just one rung
215 assert len({spec.bars for spec in specs}) >= 2
216 # every rung shares the single late anchor, including the longest one
217 assert 8 in {spec.bars for spec in specs}
218
219
220def test_trim_closing_not_emitted_for_small_gap() -> None:
221 """A tail whose energy anchor already sits near the audible end emits nothing."""
222 specs = list(TrimClosingAnchorGenerator().generate(_small_trim_gap_ctx()))
223 assert specs == []
224
225
226def _late_blendable_only_ctx() -> TransitionContext:
227 """
228 Build a context whose early window is too sparse to blend but the late one qualifies.
229
230 A full 4/4 grid runs to the end of both 124 BPM decks with matching keys,
231 so the tier at the audible end is FULL_BLEND. The outgoing rms_energy
232 stays loud for only a few bars into the buffer before dropping to a
233 still-audible level and then real silence, so the early mix-out anchor
234 lands with too few downbeats behind it for the early window's tier check
235 to pass, while the audible tail runs on for many more bars past it.
236 """
237 beats = [i * 60 / 124 for i in range(int(240 * 124 / 60))]
238 downbeats = beats[::4]
239 rms_energy = [0.9] * 1550 + [0.25] * (1730 - 1550) + [0.0] * (1800 - 1730)
240 aa_out = AudioAnalysisData(
241 duration=240.0,
242 bpm=124.0,
243 beats=beats,
244 downbeats=downbeats,
245 beats_per_bar=4,
246 rms_energy=rms_energy,
247 key="A",
248 mode="minor",
249 extra_data={},
250 )
251 aa_in = AudioAnalysisData(
252 duration=240.0,
253 bpm=124.0,
254 beats=beats,
255 downbeats=downbeats,
256 beats_per_bar=4,
257 rms_energy=[0.8] * 1800,
258 key="A",
259 mode="minor",
260 extra_data={},
261 )
262 ctx = build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
263 assert ctx.audio_end - ctx.default_anchor >= 8.0
264 early_downbeats = [d for d in ctx.outgoing.downbeats if d <= ctx.default_anchor]
265 assert len(early_downbeats) < 8
266 return ctx
267
268
269def test_trim_closing_ladder_uses_the_tier_at_its_own_anchor() -> None:
270 """A grid that only becomes blendable at the audible end still earns the long rungs."""
271 ctx = _late_blendable_only_ctx()
272 assert ctx.tier is TransitionTier.QUICK_FADE # the early window has too few downbeats
273 specs = list(TrimClosingAnchorGenerator().generate(ctx))
274 assert specs
275 assert max(spec.bars for spec in specs) == 8
276 assert all(spec.tier is not TransitionTier.QUICK_FADE for spec in specs)
277 assert all(spec.ideal_bars == 8 for spec in specs)
278
279
280def _ctx_with_late_natural_entry() -> TransitionContext:
281 """Build a context where B grooves late: its natural entry lands deep in the 45s head."""
282 aa_out = _analysis(bpm=124.0, duration=200.0)
283 aa_in = _analysis(bpm=124.0, duration=45.0)
284 aa_in.rms_energy = [0.05] * 720 + [0.9] * 1080
285 ctx = build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
286 assert ctx.natural_entry > 10.0
287 return ctx
288
289
290def _ambient_unblendable_ctx() -> tuple[AudioAnalysisData, AudioAnalysisData]:
291 """
292 Build an outgoing/incoming pair whose grid is unusable but both decks are ambient.
293
294 The outgoing downbeat grid dies at 10s (rubato tail, like the 3.2 sparse-tail
295 fixture); its energy stays quiet-but-audible out to ~43s before real silence,
296 stranding a large gap past the energy anchor. Both decks carry a validated
297 all-zero vocal timeline, so both duties read 0.0 (ambient).
298 """
299 grid_beats = [i * 60 / 128 for i in range(int(10.0 * 128 / 60) + 1)]
300 rms_energy = [0.9] * 1260 + [0.25] * 450 + [0.0] * 90
301 aa_out = AudioAnalysisData(
302 duration=45.0,
303 bpm=128.0,
304 beats=grid_beats,
305 downbeats=grid_beats[::4],
306 beats_per_bar=4,
307 rms_energy=rms_energy,
308 key="C",
309 mode="minor",
310 extra_data={"vocal_activity": [0.0] * 1800},
311 )
312 full_beats = [i * 60 / 128 for i in range(int(45 * 128 / 60))]
313 aa_in = AudioAnalysisData(
314 duration=45.0,
315 bpm=128.0,
316 beats=full_beats,
317 downbeats=full_beats[::4],
318 beats_per_bar=4,
319 rms_energy=[0.8] * 1800,
320 key="C",
321 mode="minor",
322 extra_data={"vocal_activity": [0.0] * 1800},
323 )
324 return aa_out, aa_in
325
326
327def _vocal_unblendable_ctx() -> TransitionContext:
328 """Build the same ambient pair, but with the incoming deck fully sung: never qualifies."""
329 aa_out, aa_in = _ambient_unblendable_ctx()
330 aa_in.extra_data = {"vocal_activity": [0.9] * 1800}
331 return build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
332
333
334def _clean_full_blend_ctx() -> TransitionContext:
335 """Build a context with a full, evenly-spaced grid: earns the ordinary full-blend tier."""
336 aa_out = _analysis(bpm=124.0, duration=200.0)
337 aa_in = _analysis(bpm=124.0, duration=200.0)
338 return build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
339
340
341def test_lazy_overlay_wins_for_both_ambient_unblendable_pair() -> None:
342 """Quiet-tail + ambient incoming: the long overlay replaces the 2-bar rescue."""
343 ctx_out_aa, ctx_in_aa = _ambient_unblendable_ctx()
344 plan = SmartCrossFadePlanner(logging.getLogger("test")).plan(ctx_out_aa, ctx_in_aa, 45.0)
345 assert plan.metrics.strategy is TransitionStrategy.LAZY_OVERLAY
346 assert plan.crossfade_duration >= 12.0
347 assert plan.fadein_trim_start is None # B keeps its intro
348
349
350def test_lazy_overlay_not_emitted_for_vocal_material() -> None:
351 """A singing deck never gets the unphrased long overlay."""
352 specs = list(LazyOverlayGenerator().generate(_vocal_unblendable_ctx()))
353 assert specs == []
354
355
356def test_lazy_overlay_not_emitted_when_grid_blendable() -> None:
357 """A clean, blendable pair never falls back to the unphrased overlay."""
358 specs = list(LazyOverlayGenerator().generate(_clean_full_blend_ctx()))
359 assert specs == []
360
361
362def test_lazy_overlay_beats_trim_closing_on_a_qualifying_pair() -> None:
363 """
364 The overlay must win the tie against trim-closing's equally-cheap short rungs.
365
366 Both generators anchor near the audible end with ~zero trim on this
367 context, so this exercises the actual tie-break (generator order), not
368 just an absence of competition.
369 """
370 aa_out, aa_in = _ambient_unblendable_ctx()
371 ctx = build_transition_context(aa_out, aa_in, 45.0, logging.getLogger("test"))
372 # trim-closing must actually compete here, or this proves nothing
373 assert list(TrimClosingAnchorGenerator().generate(ctx))
374
375 plan = SmartCrossFadePlanner(logging.getLogger("test")).plan(aa_out, aa_in, 45.0)
376 assert plan.metrics.strategy is TransitionStrategy.LAZY_OVERLAY
377
378
379def _lazy_gate_outgoing() -> AudioAnalysisData:
380 """
381 Outgoing analysis shared by the lazy-gate vocal-window fixtures.
382
383 Same shape as ``_late_blendable_only_ctx``'s outgoing deck: a full 4/4
384 grid at 124 BPM, but the early mix-out anchor leaves fewer than 8
385 downbeats before it, so the pair reaches QUICK_FADE and the lazy gate.
386 The vocal timeline is all-zero, so the outgoing side never contributes duty.
387 """
388 beats = [i * 60 / 124 for i in range(int(240 * 124 / 60))]
389 downbeats = beats[::4]
390 rms_energy = [0.9] * 1550 + [0.25] * (1730 - 1550) + [0.0] * (1800 - 1730)
391 return AudioAnalysisData(
392 duration=240.0,
393 bpm=124.0,
394 beats=beats,
395 downbeats=downbeats,
396 beats_per_bar=4,
397 rms_energy=rms_energy,
398 key="A",
399 mode="minor",
400 extra_data={"vocal_activity": [0.0] * 1800},
401 )
402
403
404def _lazy_gate_incoming(vocal_run: tuple[float, float]) -> AudioAnalysisData:
405 """Incoming analysis for the lazy-gate fixtures: a 45s head with vocal only over ``vocal_run``."""
406 beats = [i * 60 / 124 for i in range(int(45 * 124 / 60))]
407 vocal_activity = [0.0] * 1800
408 frame_duration = 45.0 / 1800
409 start_bin = int(vocal_run[0] / frame_duration)
410 end_bin = int(vocal_run[1] / frame_duration)
411 for i in range(start_bin, end_bin):
412 vocal_activity[i] = 0.95
413 return AudioAnalysisData(
414 duration=45.0,
415 bpm=124.0,
416 beats=beats,
417 downbeats=beats[::4],
418 beats_per_bar=4,
419 rms_energy=[0.8] * 1800,
420 key="A",
421 mode="minor",
422 extra_data={"vocal_activity": vocal_activity},
423 )
424
425
426def _front_loaded_vocal_ctx() -> TransitionContext:
427 """
428 Build a lazy-gate context where B's vocal sits inside the overlay's first 16s.
429
430 B's vocal run covers media 4.0-7.2s: ~3.2s of a 16s overlay (~0.20 duty)
431 but only ~0.07 over the full 45s head, so the whole-window gate would
432 pass it while the windowed gate correctly blocks it.
433 """
434 ctx = build_transition_context(
435 _lazy_gate_outgoing(), _lazy_gate_incoming((4.0, 7.2)), 45.0, logging.getLogger("test")
436 )
437 assert ctx.tier is TransitionTier.QUICK_FADE
438 whole = _vocal_duties(ctx)
439 assert whole is not None
440 assert whole[1] <= 0.10
441 windowed = _window_duties(ctx, _LAZY_OVERLAY_SECONDS)
442 assert windowed is not None
443 assert windowed[1] > 0.10
444 return ctx
445
446
447def _late_vocal_ctx() -> TransitionContext:
448 """
449 Build a lazy-gate context where B's vocal sits entirely outside the overlay's first 16s.
450
451 B's vocal run covers media 20.0-30.0s: 0.0 duty inside a 16s overlay but
452 ~0.22 over the full 45s head, so the whole-window gate wrongly blocks it
453 while the windowed gate correctly allows it.
454 """
455 ctx = build_transition_context(
456 _lazy_gate_outgoing(), _lazy_gate_incoming((20.0, 30.0)), 45.0, logging.getLogger("test")
457 )
458 assert ctx.tier is TransitionTier.QUICK_FADE
459 whole = _vocal_duties(ctx)
460 assert whole is not None
461 assert whole[1] > 0.10
462 windowed = _window_duties(ctx, _LAZY_OVERLAY_SECONDS)
463 assert windowed is not None
464 assert windowed[1] <= 0.10
465 return ctx
466
467
468def test_lazy_overlay_denied_when_vocals_sit_inside_the_overlay() -> None:
469 """Vocals concentrated in B's first 16s block the overlay even when its 45s duty is low."""
470 ctx = _front_loaded_vocal_ctx()
471 assert list(LazyOverlayGenerator().generate(ctx)) == []
472
473
474def test_lazy_overlay_allowed_when_vocals_sit_outside_the_overlay() -> None:
475 """Vocals late in B's head leave the overlay window ambient, so the overlay still fires."""
476 ctx = _late_vocal_ctx()
477 specs = list(LazyOverlayGenerator().generate(ctx))
478 assert len(specs) == 1
479 assert specs[0].strategy is TransitionStrategy.LAZY_OVERLAY
480
481
482def test_instrumental_blend_gate_unchanged_by_window_duties() -> None:
483 """The both-instrumental 16-bar gate keeps reading whole-window duty."""
484 ctx = _front_loaded_vocal_ctx()
485 assert _vocal_duties(ctx) is not None
486 # the 16-bar gate's own verdict must not move when the lazy gate narrows its window
487 assert earns_instrumental_blend(ctx) is False
488
489
490def test_short_rungs_offer_intro_keeping_entry() -> None:
491 """At 1-2 bars an entry at 0.0 (keep B's intro) is offered alongside the natural entry."""
492 ctx = _ctx_with_late_natural_entry()
493 options = _entry_options(ctx, 2)
494 assert 0.0 in options
495 assert ctx.natural_entry in options
496 # 0.0 must precede the natural entry: the selector ties break to the
497 # earlier candidate, so order decides which one a tie actually prefers
498 assert options.index(0.0) < options.index(ctx.natural_entry)
499 assert 0.0 not in _entry_options(ctx, 8)
500