/
/
/
1"""Tests for PlanAssembler and EmergencyHandoffFactory."""
2
3from __future__ import annotations
4
5import logging
6from dataclasses import replace
7
8import numpy as np
9import pytest
10
11from music_assistant.controllers.streams.smart_fades.models import TransitionStrategy
12from music_assistant.controllers.streams.smart_fades.planner import SmartCrossFadePlanner
13from music_assistant.controllers.streams.smart_fades.planner.assembly import (
14 EmergencyHandoffFactory,
15 PlanAssembler,
16)
17from music_assistant.controllers.streams.smart_fades.planner.candidates import (
18 Candidate,
19 CandidateFactory,
20 CandidateSpec,
21 bars_ladder,
22)
23from music_assistant.controllers.streams.smart_fades.planner.context import (
24 TransitionContext,
25 build_transition_context,
26)
27from music_assistant.models.audio_analysis import AudioAnalysisData
28
29from .conftest import _analysis_with_bands
30
31LOGGER = logging.getLogger(__name__)
32
33
34def _analysis(
35 bpm: float, duration: float = 240.0, key: str | None = "A", mode: str | None = "minor"
36) -> AudioAnalysisData:
37 interval = 60.0 / bpm
38 beats = np.arange(0.0, duration, interval, dtype=np.float32)
39 return AudioAnalysisData(
40 duration=duration,
41 bpm=bpm,
42 beats=beats.tolist(),
43 downbeats=beats[::4].tolist(),
44 rms_energy=np.full(1800, 0.5, dtype=np.float32).tolist(),
45 key=key,
46 mode=mode,
47 )
48
49
50def _ctx(
51 out: AudioAnalysisData, inc: AudioAnalysisData, buffer_duration: float = 45.0
52) -> TransitionContext:
53 return build_transition_context(out, inc, buffer_duration, LOGGER)
54
55
56def _first_fitting_candidate(ctx: TransitionContext, factory: CandidateFactory) -> Candidate:
57 """Emulate the energy ladder: largest rung that builds (mirrors test_candidates.py)."""
58 for bars in bars_ladder(ctx, ctx.tier):
59 candidate = factory.build(
60 CandidateSpec(tier=ctx.tier, bars=bars, anchor_s=None, entry_s=None)
61 )
62 if candidate is not None:
63 return candidate
64 raise AssertionError("the 1-bar rung must always yield a candidate")
65
66
67def _bands_pair(f_low_out: float, f_low_in: float) -> tuple[AudioAnalysisData, AudioAnalysisData]:
68 """Mirrors test_planner.py's ``_bands_pair``: an out/in pair with the given low-band fractions."""
69
70 def _levels(f_low: float) -> tuple[float, float, float, float]:
71 low = 1.0
72 rest = ((1.0 - f_low) / (3.0 * f_low)) ** 0.5 * low
73 return low, rest, rest, rest
74
75 out = _analysis_with_bands(*_levels(f_low_out), duration=240.0)
76 inc = _analysis_with_bands(*_levels(f_low_in), duration=240.0)
77 return out, inc
78
79
80def _mastered_fade_pair() -> tuple[AudioAnalysisData, AudioAnalysisData]:
81 """Build a -14dB frozen-spectrum ramp from media 210s, mirroring test_generators.py's fixture."""
82 t = np.linspace(0.0, 240.0, 1800)
83 gain_db = np.where(t < 210.0, 0.0, -(t - 210.0) / 30.0 * 14.0)
84 band = (0.3 * 10.0 ** (gain_db / 20.0)).astype(np.float32)
85 out = _analysis_with_bands(band, band, band, band, duration=240.0)
86 inc = _analysis(120.0, duration=240.0)
87 return out, inc
88
89
90def _rich_pair() -> tuple[AudioAnalysisData, AudioAnalysisData]:
91 """Mirrors test_planner.py's ``_rich_pair``: bass-rich AND mid-heavy on both sides."""
92 out = _analysis_with_bands(1.0, 0.3, 1.0, 0.3, duration=240.0)
93 inc = _analysis_with_bands(1.0, 0.3, 1.0, 0.3, duration=240.0)
94 return out, inc
95
96
97def _wash_mid_stack_pair() -> tuple[AudioAnalysisData, AudioAnalysisData]:
98 """Mirrors test_planner.py's ``_wash_mid_stack_pair``: triggers mid-depth dip-guard remediation."""
99 amps = (0.065**0.5, 0.065**0.5, 0.42**0.5, 0.45**0.5)
100 out = _analysis_with_bands(*amps, duration=240.0)
101 inc = _analysis_with_bands(*amps, duration=240.0)
102 t = np.linspace(0, 240.0, 1800)
103 inc.rms_energy = np.where(t < 14.0, 0.05, 0.5).astype(np.float32).tolist()
104 return out, inc
105
106
107def _vocal_probabilities(duration: float, active_windows: list[tuple[float, float]]) -> list[float]:
108 """Build a probability timeline that is quiet except inside ``active_windows``."""
109 n_frames = 1800
110 frame_duration = duration / n_frames
111 probabilities = [0.05] * n_frames
112 for start, end in active_windows:
113 start_index = max(0, int(start / frame_duration))
114 end_index = min(n_frames, int(end / frame_duration) + 1)
115 for i in range(start_index, end_index):
116 probabilities[i] = 0.9
117 return probabilities
118
119
120def _with_vocal_activity(
121 analysis: AudioAnalysisData, active_windows: list[tuple[float, float]]
122) -> AudioAnalysisData:
123 """Attach a valid vocal_activity list, active only inside ``active_windows``."""
124 assert analysis.duration is not None
125 analysis.extra_data = {
126 "vocal_activity": _vocal_probabilities(analysis.duration, active_windows)
127 }
128 return analysis
129
130
131class TestFinalizeEqSelfConsistency:
132 """``finalize()`` must yield the same EqPlan the old planner computes for the equivalent plan."""
133
134 def test_full_blend_bass_and_mid_swap_parity(self) -> None:
135 """A bass-rich and mid-heavy pair: low, mid and high schedules all match the old planner."""
136 out, inc = _rich_pair()
137 reference_plan = SmartCrossFadePlanner(LOGGER).plan(out, inc, 45.0)
138
139 ctx = _ctx(out, inc)
140 factory = CandidateFactory(ctx, LOGGER)
141 candidate = _first_fitting_candidate(ctx, factory)
142 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
143
144 assert new_plan.eq_plan.swap_at == pytest.approx(reference_plan.eq_plan.swap_at)
145 for attr in ("low_out", "low_in", "high_out", "high_in", "mid_out", "mid_in"):
146 old_sched = getattr(reference_plan.eq_plan, attr)
147 new_sched = getattr(new_plan.eq_plan, attr)
148 assert (old_sched is None) == (new_sched is None), attr
149 if old_sched is not None:
150 assert new_sched.steps == pytest.approx(old_sched.steps), attr
151 assert new_sched.shelf_type == old_sched.shelf_type
152 assert new_sched.frequency == old_sched.frequency
153
154 def test_bass_only_swap_parity(self) -> None:
155 """A bass-rich, mid-light pair: only the low/high schedules engage, matching the old planner."""
156 out, inc = _bands_pair(0.4, 0.4)
157 reference_plan = SmartCrossFadePlanner(LOGGER).plan(out, inc, 45.0)
158
159 ctx = _ctx(out, inc)
160 factory = CandidateFactory(ctx, LOGGER)
161 candidate = _first_fitting_candidate(ctx, factory)
162 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
163
164 assert new_plan.eq_plan.mid_out is None
165 assert new_plan.eq_plan.mid_in is None
166 assert reference_plan.eq_plan.low_out is not None
167 assert new_plan.eq_plan.low_out is not None
168 assert new_plan.eq_plan.low_out.steps == pytest.approx(reference_plan.eq_plan.low_out.steps)
169 assert reference_plan.eq_plan.low_in is not None
170 assert new_plan.eq_plan.low_in is not None
171 assert new_plan.eq_plan.low_in.steps == pytest.approx(reference_plan.eq_plan.low_in.steps)
172
173
174class TestFinalizeCarriesMetrics:
175 """The finalized plan exposes the winning candidate's metrics, not defaults."""
176
177 def test_finalized_plan_carries_the_candidate_metrics(self) -> None:
178 """finalize() must copy the scored metrics onto the returned plan."""
179 out, inc = _rich_pair()
180 ctx = build_transition_context(out, inc, 45.0, LOGGER)
181 factory = CandidateFactory(ctx, LOGGER)
182 candidate = factory.build(CandidateSpec(tier=ctx.tier, bars=8, anchor_s=None, entry_s=None))
183 assert candidate is not None
184
185 plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
186
187 assert plan.metrics == candidate.metrics
188
189
190class TestFinalizeChoosesFadeoutCurve:
191 """The crossfade degrades to ``nofade`` only when the overlap sits fully inside a mastered fade."""
192
193 def _candidate(self, ctx: TransitionContext) -> Candidate:
194 factory = CandidateFactory(ctx, LOGGER)
195 candidate = factory.build(CandidateSpec(tier=ctx.tier, bars=1, anchor_s=None, entry_s=None))
196 assert candidate is not None # the 1-bar rung always yields a candidate
197 return candidate
198
199 def test_overlap_entirely_inside_the_fade_uses_nofade(self) -> None:
200 """A crossfade that starts after the fade onset and runs to the audible end gets nofade."""
201 out, inc = _mastered_fade_pair()
202 ctx = _ctx(out, inc)
203 assert ctx.fade_onset is not None
204 candidate = self._candidate(ctx)
205 plan = replace(candidate.plan, fade_out_window=44.0, crossfade_duration=20.0)
206 candidate = replace(candidate, plan=plan)
207
208 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
209
210 assert new_plan.fadeout_curve == "nofade"
211
212 def test_no_detected_fade_keeps_qsin(self) -> None:
213 """Without a detected mastered fade, the crossfade curve stays the qsin default."""
214 out, inc = _rich_pair()
215 ctx = _ctx(out, inc)
216 assert ctx.fade_onset is None
217 candidate = self._candidate(ctx)
218
219 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
220
221 assert new_plan.fadeout_curve == "qsin"
222
223 def test_overlap_straddling_the_fade_onset_keeps_qsin(self) -> None:
224 """A crossfade that starts before the fade onset (flat-then-fade) cannot use nofade."""
225 out, inc = _mastered_fade_pair()
226 ctx = _ctx(out, inc)
227 assert ctx.fade_onset is not None
228 candidate = self._candidate(ctx)
229 plan = replace(candidate.plan, fade_out_window=44.0, crossfade_duration=30.0)
230 candidate = replace(candidate, plan=plan)
231
232 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
233
234 assert new_plan.fadeout_curve == "qsin"
235
236
237class TestFinalizeDipGuardBehavior:
238 """The dip-guard repair must match the old planner's remediation exactly."""
239
240 def test_wash_mid_stack_remediation_parity(self) -> None:
241 """A stacked wash+mid dip: the mid-depth remediation matches the old planner's shrink."""
242 out, inc = _wash_mid_stack_pair()
243 reference_plan = SmartCrossFadePlanner(LOGGER).plan(out, inc, 45.0)
244 # the fixture is only useful once remediation actually engaged
245 gate_depth = -8.0
246 assert reference_plan.eq_plan.mid_out is None or reference_plan.eq_plan.mid_out.steps[-1][
247 1
248 ] != pytest.approx(gate_depth, abs=0.01)
249
250 ctx = _ctx(out, inc)
251 factory = CandidateFactory(ctx, LOGGER)
252 candidate = _first_fitting_candidate(ctx, factory)
253 new_plan = PlanAssembler(ctx, LOGGER).finalize(candidate)
254
255 assert (new_plan.eq_plan.mid_out is None) == (reference_plan.eq_plan.mid_out is None)
256 if new_plan.eq_plan.mid_out is not None:
257 assert reference_plan.eq_plan.mid_out is not None
258 assert new_plan.eq_plan.mid_out.steps == pytest.approx(
259 reference_plan.eq_plan.mid_out.steps
260 )
261 assert (new_plan.eq_plan.low_out is None) == (reference_plan.eq_plan.low_out is None)
262 if new_plan.eq_plan.low_out is not None:
263 assert reference_plan.eq_plan.low_out is not None
264 assert new_plan.eq_plan.low_out.steps == pytest.approx(
265 reference_plan.eq_plan.low_out.steps
266 )
267
268
269class TestEmergencyHandoff:
270 """The emergency handoff never fails and always lands in the incoming track's vocal gap."""
271
272 def test_duration_lands_in_the_incoming_vocal_gap(self) -> None:
273 """A mid-range incoming vocal onset yields a duration just short of it, with neutral EQ."""
274 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 239.9)])
275 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(1.05, 40.0)])
276 ctx = _ctx(out, inc)
277 factory = CandidateFactory(ctx, LOGGER)
278 assert ctx.vocal_in_placement is not None
279 assert ctx.vocal_in_placement.windows
280 incoming_onset = ctx.vocal_in_placement.windows[0][0]
281
282 plan = EmergencyHandoffFactory(ctx, factory, LOGGER).build()
283
284 assert 0.4 <= plan.crossfade_duration <= 1.0
285 # the handoff ends just short of the incoming track's own vocal onset
286 assert plan.crossfade_duration == pytest.approx(incoming_onset - 0.1, abs=1e-6)
287 assert plan.metrics.strategy is TransitionStrategy.SHORT_VOCAL_HANDOFF
288 assert not plan.tempo_plan.steps
289 assert plan.fadein_trim_start is None
290 eq = plan.eq_plan
291 assert eq.low_out is None
292 assert eq.low_in is None
293 assert eq.high_out is None
294 assert eq.high_in is None
295 assert eq.mid_out is None
296 assert eq.mid_in is None
297 assert eq.swap_at == pytest.approx(plan.crossfade_duration / 2.0)
298
299 def test_duration_floors_at_min_handoff_seconds(self) -> None:
300 """An incoming vocal onset right at the head clamps the handoff to the 0.4s floor."""
301 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 239.9)])
302 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(0.0, 40.0)])
303 ctx = _ctx(out, inc)
304 factory = CandidateFactory(ctx, LOGGER)
305
306 plan = EmergencyHandoffFactory(ctx, factory, LOGGER).build()
307
308 assert plan.crossfade_duration == pytest.approx(0.4)
309 assert plan.metrics.strategy is TransitionStrategy.SHORT_VOCAL_HANDOFF
310
311 def test_duration_caps_at_max_handoff_seconds(self) -> None:
312 """A comfortably late incoming vocal onset clamps the handoff to the 1.0s ceiling."""
313 out = _with_vocal_activity(_analysis(120.0, duration=240.0), [(200.0, 239.9)])
314 inc = _with_vocal_activity(_analysis(120.0, duration=240.0), [(5.0, 40.0)])
315 ctx = _ctx(out, inc)
316 factory = CandidateFactory(ctx, LOGGER)
317
318 plan = EmergencyHandoffFactory(ctx, factory, LOGGER).build()
319
320 assert plan.crossfade_duration == pytest.approx(1.0)
321 assert plan.metrics.strategy is TransitionStrategy.SHORT_VOCAL_HANDOFF
322