/
/
/
1"""Tests for the smart fades candidate generators."""
2
3from __future__ import annotations
4
5import dataclasses
6import logging
7from typing import Any
8
9import numpy as np
10import pytest
11
12from music_assistant.controllers.streams.smart_fades.bands import build_band_profile
13from music_assistant.controllers.streams.smart_fades.models import Deck, TransitionTier
14from music_assistant.controllers.streams.smart_fades.planner.candidates import (
15 _INSTRUMENTAL_BLEND_BARS,
16 CodaAnchorGenerator,
17 EnergyLadderGenerator,
18 ProtectiveAnchorGenerator,
19 RescueAnchorGenerator,
20 VocalOnsetEntryGenerator,
21 _entry_options,
22 default_generators,
23)
24from music_assistant.controllers.streams.smart_fades.planner.context import (
25 TransitionContext,
26 build_transition_context,
27)
28from music_assistant.controllers.streams.smart_fades.structure import CodaZone
29from music_assistant.controllers.streams.smart_fades.vocal import VocalMask
30from music_assistant.models.audio_analysis import AudioAnalysisData
31
32from .conftest import _analysis_with_bands
33
34LOGGER = logging.getLogger(__name__)
35
36
37def _analysis(
38 bpm: float, duration: float = 240.0, key: str | None = "A", mode: str | None = "minor"
39) -> AudioAnalysisData:
40 """Build a plain (no band_rms) analysis row for a track."""
41 interval = 60.0 / bpm
42 beats = np.arange(0.0, duration, interval, dtype=np.float32)
43 return AudioAnalysisData(
44 duration=duration,
45 bpm=bpm,
46 beats=beats.tolist(),
47 downbeats=beats[::4].tolist(),
48 rms_energy=np.full(1800, 0.5, dtype=np.float32).tolist(),
49 key=key,
50 mode=mode,
51 )
52
53
54def _vocal_probabilities(duration: float, active_windows: list[tuple[float, float]]) -> list[float]:
55 """Build a probability timeline that is quiet except inside ``active_windows``."""
56 n_frames = 1800
57 frame_duration = duration / n_frames
58 probabilities = [0.05] * n_frames
59 for start, end in active_windows:
60 start_index = max(0, int(start / frame_duration))
61 end_index = min(n_frames, int(end / frame_duration) + 1)
62 for i in range(start_index, end_index):
63 probabilities[i] = 0.9
64 return probabilities
65
66
67def _with_vocal_activity(
68 analysis: AudioAnalysisData, active_windows: list[tuple[float, float]]
69) -> AudioAnalysisData:
70 """Attach a valid vocal_activity list, merged into any existing extra_data."""
71 assert analysis.duration is not None
72 extra = dict(analysis.extra_data or {})
73 extra["vocal_activity"] = _vocal_probabilities(analysis.duration, active_windows)
74 analysis.extra_data = extra
75 return analysis
76
77
78def _base_ctx(**overrides: Any) -> TransitionContext:
79 """Build a manually-controlled TransitionContext; only the fields under test vary."""
80 deck = Deck(
81 analysis=AudioAnalysisData(),
82 bpm=120.0,
83 beats=np.array([], dtype=np.float32),
84 downbeats=np.array([], dtype=np.float32),
85 )
86 base = TransitionContext(
87 outgoing=deck,
88 incoming=deck,
89 outgoing_profile=None,
90 incoming_profile=None,
91 buffer_duration=45.0,
92 buffer_offset=0.0,
93 audio_end=45.0,
94 default_anchor=45.0,
95 mix_out_anchor=None,
96 kick_anchor=None,
97 fade_onset=None,
98 coda_zone=None,
99 tier=TransitionTier.FULL_BLEND,
100 cross_meter=False,
101 bpm_diff_percent=0.0,
102 vocal_out_placement=None,
103 vocal_in_placement=None,
104 vocal_out_scoring=None,
105 vocal_in_scoring=None,
106 natural_entry=0.0,
107 protective_downbeats=(),
108 )
109 return dataclasses.replace(base, **overrides)
110
111
112def _out_analysis_with_quiet_tail() -> AudioAnalysisData:
113 """Outgoing analysis: loud mid before media 195s, quiet mid after (the blend region)."""
114 t = np.linspace(0.0, 240.0, 1800)
115 mid = np.where(t < 195.0, 0.5, 0.02).astype(np.float32)
116 low = np.full(1800, 0.05, dtype=np.float32)
117 analysis = _analysis_with_bands(low, low, mid, low, duration=240.0)
118 return _with_vocal_activity(analysis, [])
119
120
121class TestEnergyLadderGenerator:
122 """The primary energy ladder: default/kick anchor selection and the instrumental 16-bar rung."""
123
124 def test_emits_full_ladder_at_default_anchor_when_energy_only(self) -> None:
125 """Without vocal data, every rung is emitted once at the default anchor."""
126 ctx = build_transition_context(
127 _analysis(120.0, duration=240.0), _analysis(120.0, duration=240.0), 45.0, LOGGER
128 )
129 specs = list(EnergyLadderGenerator().generate(ctx))
130
131 assert {s.bars for s in specs} == {8, 4, 2, 1}
132 assert {s.anchor_s for s in specs} == {None}
133
134 def test_no_full_band_anchor_variant_on_a_kick_timed_track(self) -> None:
135 """The kick-folded default anchor is authoritative: no pure full-band variant is emitted."""
136 ctx = _base_ctx(kick_anchor=10.0, default_anchor=10.0, mix_out_anchor=30.0)
137 specs = list(EnergyLadderGenerator().generate(ctx))
138
139 # a full-band variant would let a longer blend win past the kick
140 # die-out, defeating the researched kick handover
141 assert {s.anchor_s for s in specs} == {None}
142 assert {s.bars for s in specs} == {8, 4, 2, 1}
143 assert all(s.source == "energy-ladder" for s in specs)
144
145 def test_no_full_band_variant_when_it_equals_the_default_anchor(self) -> None:
146 """A full-band anchor equal to the (kick-folded) default emits no duplicate rungs."""
147 ctx = _base_ctx(kick_anchor=45.0, default_anchor=45.0, mix_out_anchor=45.0)
148 specs = list(EnergyLadderGenerator().generate(ctx))
149
150 assert {s.anchor_s for s in specs} == {None}
151
152 def test_no_full_band_variant_without_a_kick_anchor(self) -> None:
153 """A track with no kick anchor keeps the single (default) anchor."""
154 ctx = _base_ctx(kick_anchor=None, default_anchor=45.0, mix_out_anchor=45.0)
155 specs = list(EnergyLadderGenerator().generate(ctx))
156
157 assert {s.anchor_s for s in specs} == {None}
158
159 def test_both_instrumental_decks_earn_16_bars(self) -> None:
160 """Near-zero vocal duty on both decks earns the 16-bar instrumental-blend rung."""
161 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [])
162 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [])
163 ctx = build_transition_context(out, inc, 45.0, LOGGER)
164
165 specs = list(EnergyLadderGenerator().generate(ctx))
166 sixteens = [s for s in specs if s.bars == _INSTRUMENTAL_BLEND_BARS]
167
168 assert sixteens
169 assert all(s.ideal_bars == _INSTRUMENTAL_BLEND_BARS for s in sixteens)
170
171 def test_mixed_vocal_instrumental_pair_never_earns_16_bars(self) -> None:
172 """A near-instrumental deck paired with a clearly vocal one earns no 16-bar rung."""
173 out = _out_analysis_with_quiet_tail()
174 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(5.0, 25.0)])
175 ctx = build_transition_context(out, inc, 45.0, LOGGER)
176
177 specs = list(EnergyLadderGenerator().generate(ctx))
178
179 assert not [s for s in specs if s.bars == _INSTRUMENTAL_BLEND_BARS]
180
181
182class TestFadeOnsetPin:
183 """The mastered-fade pin: the ladder is also emitted just ahead of a detected fade."""
184
185 def test_ladder_also_emitted_at_the_fade_onset_pin(self) -> None:
186 """A fade onset ahead of the default anchor pins a second full rung set there."""
187 ctx = _base_ctx(fade_onset=20.0, default_anchor=45.0)
188 specs = list(EnergyLadderGenerator().generate(ctx))
189
190 assert {s.anchor_s for s in specs} == {None, 20.0}
191 assert {s.bars for s in specs if s.anchor_s == 20.0} == {8, 4, 2, 1}
192
193 def test_pin_is_floored_at_the_outgoing_vocal_end(self) -> None:
194 """A fade onset inside A's own vocal window pins at the vocal end instead."""
195 ctx = _base_ctx(
196 fade_onset=20.0,
197 default_anchor=45.0,
198 vocal_out_placement=VocalMask(windows=[(0.0, 30.0)]),
199 )
200 specs = list(EnergyLadderGenerator().generate(ctx))
201
202 assert {s.anchor_s for s in specs} == {None, 30.0}
203
204 def test_no_pin_when_onset_leaves_too_little_tail(self) -> None:
205 """A fade onset under the minimum effective buffer emits no pinned rungs."""
206 ctx = _base_ctx(fade_onset=5.0, default_anchor=45.0)
207 specs = list(EnergyLadderGenerator().generate(ctx))
208
209 assert {s.anchor_s for s in specs} == {None}
210
211 def test_vocal_onset_entry_is_built_at_the_pin(self) -> None:
212 """With a detected fade, the vocal-onset entry spec anchors at the pin, not the default."""
213 ctx = _base_ctx(
214 fade_onset=20.0,
215 default_anchor=45.0,
216 vocal_in_placement=VocalMask(windows=[(20.0, 25.0)]),
217 )
218 specs = list(VocalOnsetEntryGenerator().generate(ctx))
219
220 assert len(specs) == 1
221 assert specs[0].anchor_s == 20.0
222
223 def test_real_mastered_fade_fixture_produces_the_pin(self) -> None:
224 """An end-to-end mastered-fade fixture yields ctx.fade_onset and the pinned rung set."""
225 t = np.linspace(0.0, 240.0, 1800)
226 # a -14dB monotone ramp from media 210s with a frozen spectrum:
227 # the D2 mastered-fadeout signature (>=10dB drop, still audible)
228 gain_db = np.where(t < 210.0, 0.0, -(t - 210.0) / 30.0 * 14.0)
229 band = (0.3 * 10.0 ** (gain_db / 20.0)).astype(np.float32)
230 out = _analysis_with_bands(band, band, band, band, duration=240.0)
231 inc = _analysis(120.0, duration=240.0)
232
233 ctx = build_transition_context(out, inc, 45.0, LOGGER)
234 assert ctx.fade_onset is not None, "fixture must trigger the mastered-fade detector"
235 assert ctx.fade_onset < ctx.default_anchor
236
237 specs = list(EnergyLadderGenerator().generate(ctx))
238 pinned = [s for s in specs if s.anchor_s == ctx.fade_onset]
239 assert {s.bars for s in pinned} == {8, 4, 2, 1}
240
241
242class TestCodaAnchorGenerator:
243 """Anchors and rungs derived from the outgoing track's validated coda zone."""
244
245 def _profile_ctx(self, zone: CodaZone, **overrides: Any) -> TransitionContext:
246 """Build a context with a real outgoing band profile and the given coda zone."""
247 profile = _analysis_with_bands(0.3, 0.3, 0.3, 0.3, duration=240.0)
248 merged: dict[str, object] = {
249 "coda_zone": zone,
250 "outgoing_profile": build_band_profile(profile),
251 "buffer_offset": 195.0,
252 # coda shifting is vocal-remediation-scoped: it needs a timeline
253 "vocal_out_placement": VocalMask(windows=[(1.0, 2.0)]),
254 **overrides,
255 }
256 return _base_ctx(**merged)
257
258 def test_emits_nothing_without_a_vocal_timeline(self) -> None:
259 """Coda shifting is a vocal remediation: the energy-only path never coda-shifts."""
260 zone = CodaZone(start_s=200.0, end_s=216.0)
261 ctx = self._profile_ctx(zone, vocal_out_placement=None)
262 assert list(CodaAnchorGenerator().generate(ctx)) == []
263
264 def test_emits_nothing_when_zone_is_none(self) -> None:
265 """No coda zone means no coda-anchored candidates at all."""
266 ctx = _base_ctx(coda_zone=None)
267 assert list(CodaAnchorGenerator().generate(ctx)) == []
268
269 def test_emits_nothing_when_outgoing_profile_missing(self) -> None:
270 """A zone without a band profile to derive bar starts from emits nothing."""
271 zone = CodaZone(start_s=200.0, end_s=216.0)
272 ctx = _base_ctx(coda_zone=zone, outgoing_profile=None)
273 assert list(CodaAnchorGenerator().generate(ctx)) == []
274
275 def test_emits_top_rung_and_2_bar_floor_anchored_inside_the_zone(self) -> None:
276 """An 8-bar-wide zone emits the 8-bar and 2-bar rungs, anchored at the zone's last bar."""
277 zone = CodaZone(start_s=200.0, end_s=216.0)
278 ctx = self._profile_ctx(zone)
279
280 specs = list(CodaAnchorGenerator().generate(ctx))
281
282 assert {s.bars for s in specs} == {8, 2}
283 assert {s.anchor_s for s in specs} == {216.0 - 195.0}
284 assert all(s.ideal_bars == 8 for s in specs)
285 assert all(s.source == "coda-anchor" for s in specs)
286
287 def test_emits_nothing_when_zone_shorter_than_smallest_rung(self) -> None:
288 """A zone narrower than 2 bars fits no rung, so nothing is emitted."""
289 zone = CodaZone(start_s=200.0, end_s=201.0)
290 ctx = self._profile_ctx(zone)
291 assert list(CodaAnchorGenerator().generate(ctx)) == []
292
293 def test_emits_nothing_when_anchor_below_min_effective_buffer(self) -> None:
294 """An anchor too close to the buffer offset leaves no usable overlap room."""
295 zone = CodaZone(start_s=200.0, end_s=216.0)
296 ctx = self._profile_ctx(zone, buffer_offset=210.0)
297 assert list(CodaAnchorGenerator().generate(ctx)) == []
298
299
300class TestProtectiveAnchorGenerator:
301 """Anchors chosen to keep A's last outgoing vocal phrase intact."""
302
303 def test_emits_nothing_when_out_placement_is_none(self) -> None:
304 """No outgoing placement mask at all emits nothing."""
305 ctx = _base_ctx(vocal_out_placement=None)
306 assert list(ProtectiveAnchorGenerator().generate(ctx)) == []
307
308 def test_emits_nothing_when_out_placement_has_no_windows(self) -> None:
309 """An empty (no-vocal) outgoing placement mask emits nothing."""
310 ctx = _base_ctx(vocal_out_placement=VocalMask(windows=[]))
311 assert list(ProtectiveAnchorGenerator().generate(ctx)) == []
312
313 def test_emits_ladder_at_earliest_qualifying_protective_downbeat(self) -> None:
314 """Every ladder rung is anchored at the earliest protective downbeat past the phrase."""
315 ctx = _base_ctx(
316 vocal_out_placement=VocalMask(windows=[(0.0, 30.0)]),
317 protective_downbeats=(10.0, 20.0, 32.0, 35.0, 40.0),
318 audio_end=45.0,
319 )
320 specs = list(ProtectiveAnchorGenerator().generate(ctx))
321
322 assert {s.bars for s in specs} == {8, 4, 2, 1}
323 # every rung anchors at the vocal-end downbeat; short rungs add a
324 # trim-closing anchor near audio_end (asserted separately)
325 assert all(
326 any(s.anchor_s == 32.0 for s in specs if s.bars == bars) for bars in (8, 4, 2, 1)
327 )
328 # entry options mirror the energy ladder EXACTLY, so a protected anchor
329 # can keep an aligned entry instead of forcing the natural one
330 for bars in (8, 4, 2, 1):
331 emitted = {s.entry_s for s in specs if s.bars == bars}
332 assert emitted == {None, *_entry_options(ctx, bars)}
333 assert all(s.ideal_bars == 8 for s in specs)
334
335 def test_factory_chosen_entry_is_first_among_protective_entries(self) -> None:
336 """Each protective rung leads with entry_s=None so beat alignment stays reachable."""
337 ctx = _base_ctx(
338 vocal_out_placement=VocalMask(windows=[(0.0, 30.0)]),
339 protective_downbeats=(10.0, 20.0, 32.0, 35.0, 40.0),
340 audio_end=45.0,
341 )
342 specs = list(ProtectiveAnchorGenerator().generate(ctx))
343
344 for bars in (8, 4, 2, 1):
345 entries = [s.entry_s for s in specs if s.bars == bars]
346 # None (factory beat alignment) comes first: with equal penalties
347 # the selector tie-breaks by emission order
348 assert entries[0] is None
349 assert len(entries) > 1
350
351 def test_emits_trim_closing_anchor_for_short_rungs(self) -> None:
352 """Short rungs also anchor near audio_end to close the audible-trim gap."""
353 ctx = _base_ctx(
354 vocal_out_placement=VocalMask(windows=[(0.0, 18.0)]),
355 protective_downbeats=(10.0, 20.0, 32.0, 40.0, 42.0, 44.0),
356 audio_end=45.0,
357 )
358 specs = list(ProtectiveAnchorGenerator().generate(ctx))
359
360 # 120 BPM 4/4 -> a 1-bar (2s) rung's trim-closing target is 43.0; the
361 # LAST qualifying downbeat closes the gap as tightly as possible
362 one_bar_anchors = {s.anchor_s for s in specs if s.bars == 1}
363 assert 44.0 in one_bar_anchors
364 # the vocal-end anchor stays available too
365 assert 20.0 in one_bar_anchors
366
367 def test_falls_back_to_target_when_no_downbeat_qualifies(self) -> None:
368 """No qualifying protective downbeat falls back to the (clamped) target itself."""
369 ctx = _base_ctx(
370 vocal_out_placement=VocalMask(windows=[(0.0, 30.0)]),
371 protective_downbeats=(5.0, 10.0),
372 audio_end=45.0,
373 )
374 specs = list(ProtectiveAnchorGenerator().generate(ctx))
375
376 assert 30.0 in {s.anchor_s for s in specs}
377
378
379class TestVocalOnsetEntryGenerator:
380 """An entry that lands B's first vocal onset exactly at the overlap end."""
381
382 def test_emits_nothing_when_in_placement_is_none(self) -> None:
383 """No incoming placement mask at all emits nothing."""
384 ctx = _base_ctx(vocal_in_placement=None)
385 assert list(VocalOnsetEntryGenerator().generate(ctx)) == []
386
387 def test_emits_nothing_when_in_placement_has_no_windows(self) -> None:
388 """An empty (no-vocal) incoming placement mask emits nothing."""
389 ctx = _base_ctx(vocal_in_placement=VocalMask(windows=[]))
390 assert list(VocalOnsetEntryGenerator().generate(ctx)) == []
391
392 def test_emits_nothing_when_placement_mask_is_saturated(self) -> None:
393 """A near-continuous incoming vocal mask supplies no legal onset entry."""
394 ctx = _base_ctx(vocal_in_placement=VocalMask(windows=[(0.0, 44.0)]))
395 assert list(VocalOnsetEntryGenerator().generate(ctx)) == []
396
397 def test_emits_onset_aligned_entry_at_the_ideal_rung(self) -> None:
398 """The entry lands B's first onset exactly ``ideal_bars`` bars before the overlap end."""
399 ctx = _base_ctx(vocal_in_placement=VocalMask(windows=[(20.0, 25.0)]))
400 specs = list(VocalOnsetEntryGenerator().generate(ctx))
401
402 assert len(specs) == 1
403 (spec,) = specs
404 assert spec.bars == 8
405 assert spec.ideal_bars == 8
406 assert spec.anchor_s is None
407 assert spec.entry_s == 20.0 - 8 * (4 * 60.0 / 120.0)
408 assert spec.source == "vocal-onset-entry"
409
410 def test_emits_nothing_when_the_derived_entry_is_negative(self) -> None:
411 """An onset too early in the buffer leaves no legal (non-negative) entry."""
412 ctx = _base_ctx(vocal_in_placement=VocalMask(windows=[(5.0, 10.0)]))
413 assert list(VocalOnsetEntryGenerator().generate(ctx)) == []
414
415 def test_emits_nothing_when_the_derived_entry_falls_inside_another_window(self) -> None:
416 """An onset-aligned entry landing inside a vocal window is not a legal cut."""
417 ctx = _base_ctx(vocal_in_placement=VocalMask(windows=[(20.0, 25.0), (2.0, 5.0)]))
418 assert list(VocalOnsetEntryGenerator().generate(ctx)) == []
419
420
421class TestRescueAnchorGenerator:
422 """The last-resort rescue rung: a modest (<=2-bar) late-anchored candidate."""
423
424 def test_emits_only_rungs_of_two_bars_or_fewer(self) -> None:
425 """The rescue ladder never proposes more than a 2-bar overlap."""
426 ctx = _base_ctx(
427 protective_downbeats=(20.0, 30.0, 39.0, 41.0),
428 audio_end=45.0,
429 vocal_out_placement=VocalMask(windows=[(0.0, 10.0)]),
430 )
431 specs = list(RescueAnchorGenerator().generate(ctx))
432
433 assert {s.bars for s in specs} == {2, 1}
434 assert all(s.source == "rescue-anchor" for s in specs)
435
436 def test_anchor_lands_near_audio_end_minus_bars_times_bar_seconds(self) -> None:
437 """Each rung's anchor snaps to the protective downbeat nearest its own late target."""
438 ctx = _base_ctx(
439 protective_downbeats=(20.0, 30.0, 39.0, 41.0),
440 audio_end=45.0,
441 vocal_out_placement=VocalMask(windows=[(0.0, 10.0)]),
442 )
443 specs = list(RescueAnchorGenerator().generate(ctx))
444
445 bar_seconds = 2.0
446 anchors_by_bars = {s.bars: s.anchor_s for s in specs}
447 assert anchors_by_bars[2] == pytest.approx(45.0 - 2 * bar_seconds)
448 assert anchors_by_bars[1] == pytest.approx(45.0 - 1 * bar_seconds)
449
450 def test_anchor_never_falls_below_the_outgoing_vocal_end(self) -> None:
451 """A late outgoing vocal floors every rung's target, so it is never truncated."""
452 ctx = _base_ctx(
453 protective_downbeats=(20.0, 30.0, 39.0, 41.0, 44.0),
454 audio_end=45.0,
455 vocal_out_placement=VocalMask(windows=[(0.0, 43.5)]),
456 )
457 specs = list(RescueAnchorGenerator().generate(ctx))
458
459 for spec in specs:
460 assert spec.anchor_s is not None
461 assert spec.anchor_s >= 43.5 - 1e-9
462
463 def test_not_part_of_default_generators(self) -> None:
464 """The rescue generator is a planner-only fallback, never part of the default set."""
465 assert RescueAnchorGenerator not in [type(g) for g in default_generators()]
466
467
468class TestDefaultGenerators:
469 """The standard generator set and its preference order."""
470
471 def test_default_generators_preference_order(self) -> None:
472 """
473 Generators run best-first.
474
475 Order: energy ladder, coda anchor, protective anchor, onset entry,
476 lazy overlay, trim closing.
477 """
478 names = [g.name for g in default_generators()]
479 assert names == [
480 "energy-ladder",
481 "coda-anchor",
482 "protective-anchor",
483 "vocal-onset-entry",
484 "lazy-overlay",
485 "trim-closing-anchor",
486 ]
487