/
/
/
1"""Tests for the candidate scoring policies: rejections and soft penalties."""
2
3from __future__ import annotations
4
5import dataclasses
6
7import numpy as np
8import pytest
9
10from music_assistant.controllers.streams.smart_fades.models import (
11 Deck,
12 TransitionTier,
13)
14from music_assistant.controllers.streams.smart_fades.planner.context import (
15 TransitionContext,
16)
17from music_assistant.controllers.streams.smart_fades.planner.policies import (
18 AnchorAlignmentPolicy,
19 AudibleTrimPolicy,
20 DeadAirPolicy,
21 OverlapPreferencePolicy,
22 Verdict,
23 VocalCollisionPolicy,
24 VocalTruncationPolicy,
25 default_policies,
26)
27from music_assistant.controllers.streams.smart_fades.vocal import (
28 COLLISION_SECONDS_LIMIT,
29 SHORT_FADE_SECONDS,
30 WEIGHTED_COLLISION_LIMIT,
31 VocalMask,
32)
33from music_assistant.models.audio_analysis import AudioAnalysisData
34from tests.controllers.streams.smart_fades.conftest import build_test_candidate as _candidate
35
36
37def _ctx(
38 *,
39 tier: TransitionTier = TransitionTier.FULL_BLEND,
40 vocal_out_scoring: VocalMask | None = None,
41 vocal_in_scoring: VocalMask | None = None,
42 natural_entry: float = 0.0,
43 outgoing_analysis: AudioAnalysisData | None = None,
44 buffer_offset: float = 0.0,
45) -> TransitionContext:
46 deck = Deck(
47 analysis=AudioAnalysisData(),
48 bpm=120.0,
49 beats=np.array([], dtype=np.float32),
50 downbeats=np.array([], dtype=np.float32),
51 )
52 outgoing = dataclasses.replace(deck, analysis=outgoing_analysis) if outgoing_analysis else deck
53 return TransitionContext(
54 outgoing=outgoing,
55 incoming=deck,
56 outgoing_profile=None,
57 incoming_profile=None,
58 buffer_duration=45.0,
59 buffer_offset=buffer_offset,
60 audio_end=45.0,
61 default_anchor=45.0,
62 mix_out_anchor=None,
63 kick_anchor=None,
64 fade_onset=None,
65 coda_zone=None,
66 tier=tier,
67 cross_meter=False,
68 bpm_diff_percent=0.0,
69 vocal_out_placement=None,
70 vocal_in_placement=None,
71 vocal_out_scoring=vocal_out_scoring,
72 vocal_in_scoring=vocal_in_scoring,
73 natural_entry=natural_entry,
74 protective_downbeats=(),
75 )
76
77
78# a mask presence-only fixture: contents never matter to any policy, only
79# whether the field is None or not
80_MASK = VocalMask(windows=[(0.0, 1.0)])
81
82
83class TestVerdict:
84 """The Verdict value object's two constructors."""
85
86 def test_reject_sets_rejected_true(self) -> None:
87 """A rejecting verdict reports rejected=True with its reason."""
88 verdict = Verdict.reject("some reason")
89
90 assert verdict.rejected is True
91 assert verdict.reason == "some reason"
92
93 def test_ok_defaults_to_no_penalty_not_rejected(self) -> None:
94 """An accepted verdict defaults to zero penalty and rejected=False."""
95 verdict = Verdict.ok()
96
97 assert verdict.rejected is False
98 assert verdict.penalty == 0.0
99
100 def test_ok_carries_penalty(self) -> None:
101 """An accepted verdict can still carry a soft penalty."""
102 verdict = Verdict.ok(7.5, "reason")
103
104 assert verdict.rejected is False
105 assert verdict.penalty == 7.5
106 assert verdict.reason == "reason"
107
108
109class TestVocalCollisionPolicy:
110 """Reject/penalize simultaneous outgoing/incoming vocal overlap."""
111
112 policy = VocalCollisionPolicy()
113
114 def test_neutral_when_out_scoring_mask_missing(self) -> None:
115 """No outgoing vocal timeline disables the guard entirely."""
116 candidate = _candidate(collision=999.0, weighted=999.0)
117 ctx = _ctx(vocal_out_scoring=None, vocal_in_scoring=_MASK)
118
119 verdict = self.policy.evaluate(candidate, ctx)
120
121 assert verdict.rejected is False
122 assert verdict.penalty == 0.0
123
124 def test_neutral_when_in_scoring_mask_missing(self) -> None:
125 """No incoming vocal timeline disables the guard entirely."""
126 candidate = _candidate(collision=999.0, weighted=999.0)
127 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=None)
128
129 verdict = self.policy.evaluate(candidate, ctx)
130
131 assert verdict.rejected is False
132 assert verdict.penalty == 0.0
133
134 def test_rejects_at_collision_seconds_limit(self) -> None:
135 """Raw collision seconds at the limit rejects the candidate."""
136 candidate = _candidate(collision=COLLISION_SECONDS_LIMIT, weighted=0.0)
137 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
138
139 assert self.policy.evaluate(candidate, ctx).rejected is True
140
141 def test_no_rejection_just_below_collision_seconds_limit(self) -> None:
142 """Raw collision seconds just under the limit does not reject."""
143 candidate = _candidate(collision=COLLISION_SECONDS_LIMIT - 0.01, weighted=0.0)
144 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
145
146 assert self.policy.evaluate(candidate, ctx).rejected is False
147
148 def test_rejects_at_weighted_collision_limit(self) -> None:
149 """Weighted (gain-integrated) collision at the limit rejects the candidate."""
150 candidate = _candidate(collision=0.0, weighted=WEIGHTED_COLLISION_LIMIT)
151 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
152
153 assert self.policy.evaluate(candidate, ctx).rejected is True
154
155 def test_no_rejection_just_below_weighted_collision_limit(self) -> None:
156 """Weighted collision just under the limit does not reject."""
157 candidate = _candidate(collision=0.0, weighted=WEIGHTED_COLLISION_LIMIT - 0.01)
158 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
159
160 assert self.policy.evaluate(candidate, ctx).rejected is False
161
162 def test_penalty_quadratic_in_weighted_collision(self) -> None:
163 """Sub-limit penalty is quadratic: near-inaudible residue is cheap, near-rejection is steep."""
164 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
165 low = _candidate(weighted=0.1)
166 high = _candidate(weighted=0.2)
167
168 low_penalty = self.policy.evaluate(low, ctx).penalty
169 high_penalty = self.policy.evaluate(high, ctx).penalty
170
171 assert low_penalty == pytest.approx((0.1 / WEIGHTED_COLLISION_LIMIT) ** 2 * 20.0)
172 assert high_penalty == pytest.approx(4 * low_penalty)
173
174 def test_penalty_approaches_scale_at_the_rejection_boundary(self) -> None:
175 """Just under the rejection boundary the penalty approaches the full scale."""
176 ctx = _ctx(vocal_out_scoring=_MASK, vocal_in_scoring=_MASK)
177 near = _candidate(weighted=WEIGHTED_COLLISION_LIMIT - 1e-6)
178
179 assert self.policy.evaluate(near, ctx).penalty == pytest.approx(20.0, abs=0.01)
180
181
182class TestVocalTruncationPolicy:
183 """Reject a candidate that cuts off an audible outgoing vocal phrase."""
184
185 policy = VocalTruncationPolicy()
186
187 def test_neutral_when_out_scoring_mask_missing(self) -> None:
188 """No outgoing vocal timeline disables the guard entirely."""
189 candidate = _candidate(vocal_fade=999.0)
190 ctx = _ctx(vocal_out_scoring=None)
191
192 verdict = self.policy.evaluate(candidate, ctx)
193
194 assert verdict.rejected is False
195 assert verdict.penalty == 0.0
196
197 def test_rejects_when_audible_vocal_extends_past_the_anchor(self) -> None:
198 """A phrase cut off by the anchor trim (beyond 0.25s) rejects the candidate."""
199 candidate = _candidate(duration=20.0)
200 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[(18.0, 21.0)]))
201
202 assert self.policy.evaluate(candidate, ctx).rejected is True
203
204 def test_no_rejection_at_exactly_the_truncation_threshold(self) -> None:
205 """Exactly 0.25s of cut vocal does not reject (strictly-greater test)."""
206 candidate = _candidate(duration=20.0)
207 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[(20.0, 20.25)]))
208
209 assert self.policy.evaluate(candidate, ctx).rejected is False
210
211 def test_no_rejection_when_the_vocal_rides_the_fade_untruncated(self) -> None:
212 """A long phrase entirely inside the fade window is normal, never a truncation."""
213 candidate = _candidate(duration=20.0)
214 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[(5.0, 19.5)]))
215
216 assert self.policy.evaluate(candidate, ctx).rejected is False
217
218 def test_inaudible_vocal_past_the_rms_boundary_never_counts(self) -> None:
219 """Windows beyond audio_end are inaudible and cannot register as truncation."""
220 candidate = _candidate(duration=20.0)
221 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[(46.0, 50.0)]))
222
223 assert self.policy.evaluate(candidate, ctx).rejected is False
224
225
226class TestAudibleTrimPolicy:
227 """Reject (on a short fade) or penalize trimming audible outgoing material."""
228
229 policy = AudibleTrimPolicy()
230
231 def test_rejects_when_short_fade_trims_past_its_own_duration(self) -> None:
232 """A crossfade at the short-fade limit that trims more than it spans is rejected."""
233 candidate = _candidate(duration=SHORT_FADE_SECONDS, trim=SHORT_FADE_SECONDS + 0.5)
234 # the hard rule is vocal-protection-scoped: it needs a vocal timeline
235 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[]))
236
237 assert self.policy.evaluate(candidate, ctx).rejected is True
238
239 def test_rejects_even_when_vocal_data_is_missing(self) -> None:
240 """Missing vocal data must never disable the guard: a stale pre-vocal row still rejects."""
241 candidate = _candidate(duration=SHORT_FADE_SECONDS, trim=SHORT_FADE_SECONDS + 0.5)
242 ctx = _ctx(vocal_out_scoring=None)
243
244 assert self.policy.evaluate(candidate, ctx).rejected is True
245
246 def test_no_rejection_when_trim_equals_short_fade_duration(self) -> None:
247 """A short fade whose trim exactly equals its duration does not reject (strict >)."""
248 candidate = _candidate(duration=SHORT_FADE_SECONDS, trim=SHORT_FADE_SECONDS)
249 ctx = _ctx(vocal_out_scoring=VocalMask(windows=[]))
250
251 assert self.policy.evaluate(candidate, ctx).rejected is False
252
253 def test_no_rejection_when_fade_longer_than_short_fade_limit(self) -> None:
254 """A fade longer than the short-fade limit never rejects on trim, however large."""
255 candidate = _candidate(duration=SHORT_FADE_SECONDS + 0.01, trim=SHORT_FADE_SECONDS + 5.0)
256 ctx = _ctx()
257
258 assert self.policy.evaluate(candidate, ctx).rejected is False
259
260 def test_penalty_proportional_to_trim(self) -> None:
261 """Penalty scales linearly (1.0/s) with audible outgoing trim, doubling with it."""
262 ctx = _ctx()
263 low = _candidate(duration=20.0, trim=3.0)
264 high = _candidate(duration=20.0, trim=6.0)
265
266 low_penalty = self.policy.evaluate(low, ctx).penalty
267 high_penalty = self.policy.evaluate(high, ctx).penalty
268
269 assert low_penalty == pytest.approx(3.0)
270 assert high_penalty == pytest.approx(2 * low_penalty)
271
272
273class TestOverlapPreferencePolicy:
274 """Prefer the tier's top rung and the context's chosen tier."""
275
276 policy = OverlapPreferencePolicy()
277
278 def test_no_penalty_at_top_rung_and_matching_tier(self) -> None:
279 """A candidate at its ideal bar count and the context's tier earns no penalty."""
280 candidate = _candidate(bars=16, ideal=16, tier=TransitionTier.FULL_BLEND)
281 ctx = _ctx(tier=TransitionTier.FULL_BLEND)
282
283 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(0.0)
284
285 def test_penalty_scales_with_rung_gap(self) -> None:
286 """Each ladder rung below the ideal adds 10.0 penalty."""
287 ctx = _ctx(tier=TransitionTier.FULL_BLEND)
288 one_rung = _candidate(bars=8, ideal=16, tier=TransitionTier.FULL_BLEND)
289 two_rungs = _candidate(bars=4, ideal=16, tier=TransitionTier.FULL_BLEND)
290
291 assert self.policy.evaluate(one_rung, ctx).penalty == pytest.approx(10.0)
292 assert self.policy.evaluate(two_rungs, ctx).penalty == pytest.approx(20.0)
293
294 def test_penalty_scales_with_tier_steps_below_context(self) -> None:
295 """Each tier step below the context's chosen tier adds 15.0 penalty."""
296 ctx = _ctx(tier=TransitionTier.FULL_BLEND)
297 one_step = _candidate(bars=16, ideal=16, tier=TransitionTier.TEMPO_BLEND)
298 two_steps = _candidate(bars=16, ideal=16, tier=TransitionTier.QUICK_FADE)
299
300 assert self.policy.evaluate(one_step, ctx).penalty == pytest.approx(15.0)
301 assert self.policy.evaluate(two_steps, ctx).penalty == pytest.approx(30.0)
302
303 def test_no_negative_penalty_when_tier_exceeds_context(self) -> None:
304 """A candidate whose tier is more ambitious than the context's earns no tier penalty."""
305 candidate = _candidate(bars=16, ideal=16, tier=TransitionTier.FULL_BLEND)
306 ctx = _ctx(tier=TransitionTier.QUICK_FADE)
307
308 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(0.0)
309
310 def test_never_rejects(self) -> None:
311 """This is a pure soft-scoring policy: it never disqualifies a candidate."""
312 candidate = _candidate(bars=1, ideal=16, tier=TransitionTier.QUICK_FADE)
313 ctx = _ctx(tier=TransitionTier.FULL_BLEND)
314
315 assert self.policy.evaluate(candidate, ctx).rejected is False
316
317
318class TestAnchorAlignmentPolicy:
319 """Prefer downbeat-anchored fades and groove-aligned incoming entries."""
320
321 policy = AnchorAlignmentPolicy()
322
323 def test_no_penalty_on_downbeat_with_natural_entry(self) -> None:
324 """A downbeat-anchored fade with no pinned entry earns no penalty."""
325 candidate = _candidate(on_downbeat=True)
326 ctx = _ctx()
327
328 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(0.0)
329
330 def test_penalty_when_not_on_downbeat(self) -> None:
331 """A fade not anchored on a downbeat costs 4.0."""
332 candidate = _candidate(on_downbeat=False)
333 ctx = _ctx()
334
335 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(4.0)
336
337 def test_penalty_when_pinned_entry_not_groove_aligned(self) -> None:
338 """A pinned entry from a non-vocal-onset source costs 2.0."""
339 candidate = _candidate(on_downbeat=True)
340 candidate = dataclasses.replace(
341 candidate, spec=dataclasses.replace(candidate.spec, entry_s=12.3, source="some-other")
342 )
343 ctx = _ctx()
344
345 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(2.0)
346
347 def test_no_penalty_when_pinned_entry_is_vocal_onset(self) -> None:
348 """A pinned entry from the vocal-onset-entry generator counts as aligned."""
349 candidate = _candidate(on_downbeat=True)
350 candidate = dataclasses.replace(
351 candidate,
352 spec=dataclasses.replace(candidate.spec, entry_s=12.3, source="vocal-onset-entry"),
353 )
354 ctx = _ctx()
355
356 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(0.0)
357
358 def test_penalties_stack(self) -> None:
359 """Both the downbeat and entry-alignment penalties can apply together."""
360 candidate = _candidate(on_downbeat=False)
361 candidate = dataclasses.replace(
362 candidate, spec=dataclasses.replace(candidate.spec, entry_s=1.0, source="some-other")
363 )
364 ctx = _ctx()
365
366 assert self.policy.evaluate(candidate, ctx).penalty == pytest.approx(6.0)
367
368
369def test_default_policies_returns_the_six_standard_policies() -> None:
370 """The standard policy tuple contains one instance of each of the six policies."""
371 policies = default_policies()
372
373 assert [type(p) for p in policies] == [
374 VocalCollisionPolicy,
375 VocalTruncationPolicy,
376 AudibleTrimPolicy,
377 DeadAirPolicy,
378 OverlapPreferencePolicy,
379 AnchorAlignmentPolicy,
380 ]
381
382
383class TestDeadAirPolicy:
384 """Penalize a handover that strands the listener before B's groove entry."""
385
386 def _hot_outgoing(self, tail_level: float = 0.9) -> AudioAnalysisData:
387 # peak mid-track; the tail level decides whether the pre-fade window is hot
388 rms = [0.9] * 1500 + [tail_level] * 300
389 return AudioAnalysisData(duration=220.0, rms_energy=rms)
390
391 def test_dead_air_beyond_grace_is_penalized_per_second(self) -> None:
392 """A hot outgoing into a late groove entry pays for every second past the grace."""
393 ctx = _ctx(natural_entry=19.2, outgoing_analysis=self._hot_outgoing(), buffer_offset=175.0)
394 candidate = _candidate(duration=2.0, fade_end=42.0)
395
396 verdict = DeadAirPolicy().evaluate(candidate, ctx)
397
398 assert verdict.penalty == pytest.approx(19.2 - 2.0 - 4.0)
399 assert not verdict.rejected
400
401 def test_trimming_to_the_groove_entry_clears_the_penalty(self) -> None:
402 """A candidate that enters at B's groove has no dead air to pay for."""
403 ctx = _ctx(natural_entry=19.2, outgoing_analysis=self._hot_outgoing(), buffer_offset=175.0)
404 candidate = _candidate(duration=2.0, fade_end=42.0, fadein_trim=19.2)
405
406 assert DeadAirPolicy().evaluate(candidate, ctx).penalty == 0.0
407
408 def test_gap_within_grace_is_free(self) -> None:
409 """Short breathing room after the fade stays unpenalized."""
410 ctx = _ctx(natural_entry=5.5, outgoing_analysis=self._hot_outgoing(), buffer_offset=175.0)
411 candidate = _candidate(duration=2.0, fade_end=42.0)
412
413 assert DeadAirPolicy().evaluate(candidate, ctx).penalty == 0.0
414
415 def test_quiet_outgoing_keeps_the_incoming_intro(self) -> None:
416 """An outgoing already faded to a whisper welcomes a slow ambient intro."""
417 ctx = _ctx(
418 natural_entry=19.2,
419 outgoing_analysis=self._hot_outgoing(tail_level=0.02),
420 buffer_offset=175.0,
421 )
422 candidate = _candidate(duration=2.0, fade_end=42.0)
423
424 assert DeadAirPolicy().evaluate(candidate, ctx).penalty == 0.0
425
426 def test_missing_outgoing_rms_still_penalizes(self) -> None:
427 """Without RMS data the hot-outgoing gate cannot clear the penalty."""
428 ctx = _ctx(natural_entry=19.2, outgoing_analysis=AudioAnalysisData(), buffer_offset=175.0)
429 candidate = _candidate(duration=2.0, fade_end=42.0)
430
431 assert DeadAirPolicy().evaluate(candidate, ctx).penalty > 0.0
432