/
/
/
1"""Tests for the smart fades candidate factory."""
2
3from __future__ import annotations
4
5import logging
6
7import numpy as np
8import pytest
9
10from music_assistant.controllers.streams.smart_fades.models import TransitionTier
11from music_assistant.controllers.streams.smart_fades.planner.candidates import (
12 Candidate,
13 CandidateFactory,
14 CandidateSpec,
15 bars_ladder,
16)
17from music_assistant.controllers.streams.smart_fades.planner.context import (
18 TransitionContext,
19 build_transition_context,
20)
21from music_assistant.models.audio_analysis import AudioAnalysisData
22
23from .conftest import _analysis_with_bands
24
25LOGGER = logging.getLogger(__name__)
26
27
28def _analysis(
29 bpm: float, duration: float = 240.0, key: str | None = "A", mode: str | None = "minor"
30) -> AudioAnalysisData:
31 interval = 60.0 / bpm
32 beats = np.arange(0.0, duration, interval, dtype=np.float32)
33 return AudioAnalysisData(
34 duration=duration,
35 bpm=bpm,
36 beats=beats.tolist(),
37 downbeats=beats[::4].tolist(),
38 rms_energy=np.full(1800, 0.5, dtype=np.float32).tolist(),
39 key=key,
40 mode=mode,
41 )
42
43
44def _ctx(
45 out: AudioAnalysisData, inc: AudioAnalysisData, buffer_duration: float = 45.0
46) -> TransitionContext:
47 return build_transition_context(out, inc, buffer_duration, LOGGER)
48
49
50def _spec(
51 ctx: TransitionContext,
52 bars: int,
53 anchor_s: float | None = None,
54 entry_s: float | None = None,
55) -> CandidateSpec:
56 return CandidateSpec(tier=ctx.tier, bars=bars, anchor_s=anchor_s, entry_s=entry_s)
57
58
59def _first_fitting(ctx: TransitionContext, factory: CandidateFactory) -> Candidate:
60 """Emulate the energy ladder: largest rung that builds (the old _energy_candidate)."""
61 for bars in bars_ladder(ctx, ctx.tier):
62 candidate = factory.build(_spec(ctx, bars))
63 if candidate is not None:
64 return candidate
65 raise AssertionError("the 1-bar rung must always yield a candidate")
66
67
68class TestFactoryGoldenTiming:
69 """
70 The factory's timed fields are pinned to golden values from the old planner.
71
72 The old monolithic planner (``_prepare_decks``/``_choose_tier``/
73 ``_energy_candidate``/``_build_candidate``) was deleted at the task-9
74 switchover; these values were captured from it, live, on these exact
75 fixtures, before deletion.
76 """
77
78 def test_energy_candidate_timing_matches_the_old_planner(self) -> None:
79 """The default-spec ladder walk reproduces the old planner's energy candidate timing."""
80 out, inc = _analysis(120.0, duration=240.0), _analysis(122.0, duration=240.0)
81
82 ctx = _ctx(out, inc)
83 candidate = _first_fitting(ctx, CandidateFactory(ctx, LOGGER))
84
85 assert candidate.plan.tier is TransitionTier.FULL_BLEND
86 assert candidate.plan.fade_out_window == 45.0
87 assert candidate.plan.crossfade_duration == pytest.approx(13.770492, abs=1e-5)
88 assert candidate.plan.fadein_trim_start == 0.0
89 assert len(candidate.plan.tempo_plan.steps) == 4
90 assert candidate.plan.fadeout_trim is None
91
92 def test_explicit_anchor_timing_matches_the_old_build_candidate(self) -> None:
93 """An explicitly re-anchored spec reproduces the old ``_build_candidate`` timing."""
94 out, inc = _analysis(80.0, duration=240.0), _analysis(83.2, duration=240.0)
95
96 ctx = _ctx(out, inc)
97 candidate = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 8, anchor_s=20.0, entry_s=0.0))
98
99 assert candidate is not None
100 assert candidate.plan.tier is TransitionTier.QUICK_FADE
101 assert candidate.plan.fade_out_window == 20.0
102 assert candidate.plan.crossfade_duration == pytest.approx(14.0, abs=1e-5)
103 assert candidate.plan.fadein_trim_start == 0.0
104
105
106class TestUnheardIntroClamp:
107 """A fade-in trim is bounded by what the overlap plays under the outgoing track."""
108
109 def test_a_trim_deeper_than_the_overlap_is_stripped(self) -> None:
110 """A cut skipping more of the incoming track than the blend plays is dropped."""
111 out, inc = _analysis(80.0, duration=240.0), _analysis(83.2, duration=240.0)
112 ctx = _ctx(out, inc)
113 deep = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 8, anchor_s=20.0, entry_s=30.0))
114 assert deep is not None
115 assert deep.plan.fadein_trim_start is None
116
117 def test_a_trim_the_overlap_covers_is_kept(self) -> None:
118 """An entry the blend actually plays under the outgoing track survives."""
119 out, inc = _analysis(80.0, duration=240.0), _analysis(83.2, duration=240.0)
120 ctx = _ctx(out, inc)
121 shallow = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 8, anchor_s=20.0, entry_s=10.0))
122 assert shallow is not None
123 assert shallow.plan.fadein_trim_start == pytest.approx(10.0, abs=2.0)
124
125
126class TestCandidateBuildGuards:
127 """Ports of the old candidate build guards, against the factory API."""
128
129 def test_dead_grid_before_energetic_buffer_end_does_not_inflate_quick_fade(self) -> None:
130 """A beatless-but-energetic outro keeps the quick fade at its intended bar count."""
131 out = _analysis(120.0, duration=240.0)
132 beats = np.asarray(out.beats, dtype=np.float32)
133 out.beats = beats[beats <= 230.0].tolist()
134 downbeats = np.asarray(out.downbeats, dtype=np.float32)
135 out.downbeats = downbeats[downbeats <= 230.0].tolist()
136 inc = _analysis(156.0, duration=240.0)
137
138 ctx = _ctx(out, inc)
139 candidate = _first_fitting(ctx, CandidateFactory(ctx, LOGGER))
140
141 assert candidate.plan.tier is TransitionTier.QUICK_FADE
142 bar_out = 4 * 60.0 / 120.0
143 assert candidate.plan.crossfade_duration <= 2 * bar_out + 0.1
144
145 def test_re_anchored_tier_downgrade_caps_the_bar_count(self) -> None:
146 """A re-anchor that downgrades the tier also caps the inherited bar count."""
147 ctx = _ctx(_analysis(80.0), _analysis(83.2))
148 candidate = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 8, anchor_s=20.0, entry_s=0.0))
149
150 assert candidate is not None
151 assert candidate.plan.tier is TransitionTier.QUICK_FADE
152 assert not candidate.plan.tempo_plan
153 # the built spec reflects the downgrade, so policies score reality
154 assert candidate.spec.tier is TransitionTier.QUICK_FADE
155 assert candidate.spec.bars <= 4
156 # 4-bar quick-fade cap at 80 BPM (3s bars) plus sub-bar anchor slack
157 assert candidate.plan.crossfade_duration <= 15.0
158
159 def test_protected_intro_before_late_drop_degrades_instead_of_asserting(self) -> None:
160 """A sung intro ahead of a late bass drop must yield a candidate, never raise."""
161
162 def env(value: float) -> np.ndarray:
163 return np.full(1800, value, dtype=np.float32)
164
165 t = np.linspace(0, 240.0, 1800)
166 low = np.where(t < 60.0, 0.02, 1.0).astype(np.float32)
167 inc = _analysis_with_bands(low, env(0.6), env(0.6), env(0.3))
168 rms = np.full(1800, 0.2, dtype=np.float32)
169 rms[t >= 35.0] = 1.0
170 inc.rms_energy = rms.tolist()
171 out = _analysis_with_bands(env(1.0), env(0.5), env(0.5), env(0.3))
172
173 ctx = _ctx(out, inc)
174 candidate = _first_fitting(ctx, CandidateFactory(ctx, LOGGER))
175
176 assert candidate.plan.crossfade_duration > 0.0
177
178 def test_oversized_entry_returns_none_for_multi_bar_rung(self) -> None:
179 """An entry too late for any multi-bar overlap makes the spec infeasible."""
180 ctx = _ctx(_analysis(120.0, duration=240.0), _analysis(122.0, duration=240.0))
181 # an entry this late leaves no room for any multi-bar overlap in the 45s head
182 candidate = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 8, entry_s=44.0))
183 assert candidate is None
184
185
186class TestFactoryPurity:
187 """Building candidates never leaks state between builds."""
188
189 def test_same_spec_builds_identical_plans(self) -> None:
190 """An intervening re-anchored build never contaminates the next build."""
191 ctx = _ctx(_analysis(120.0, duration=240.0), _analysis(122.0, duration=240.0))
192 factory = CandidateFactory(ctx, LOGGER)
193 spec = _spec(ctx, 8)
194
195 first = factory.build(spec)
196 # an intervening re-anchored build must not contaminate the next one
197 factory.build(_spec(ctx, 2, anchor_s=20.0))
198 second = factory.build(spec)
199
200 assert first is not None
201 assert second is not None
202 assert first.plan == second.plan
203 assert first.metrics == second.metrics
204
205
206class TestFactoryMetrics:
207 """Metrics are computed per candidate, vocal fields only with vocal data."""
208
209 def test_energy_only_metrics(self) -> None:
210 """Without vocal data the vocal metrics stay at their energy-only defaults."""
211 ctx = _ctx(_analysis(120.0, duration=240.0), _analysis(122.0, duration=240.0))
212 candidate = _first_fitting(ctx, CandidateFactory(ctx, LOGGER))
213
214 assert candidate.metrics.audible_outgoing_trim >= 0.0
215 assert candidate.metrics.collision_seconds == 0.0
216 assert candidate.metrics.weighted_collision_seconds == 0.0
217
218 def test_ideal_bars_defaults_to_spec_bars(self) -> None:
219 """A spec without an explicit ideal_bars falls back to its own bar count."""
220 ctx = _ctx(_analysis(120.0, duration=240.0), _analysis(122.0, duration=240.0))
221 candidate = CandidateFactory(ctx, LOGGER).build(_spec(ctx, 1))
222 assert candidate is not None
223 assert candidate.ideal_bars == 1
224