/
/
/
1"""
2Smart Fades - candidate scoring policies.
3
4Each ``Policy`` independently judges one built ``Candidate`` against the
5shared ``TransitionContext``, returning a ``Verdict``: either an outright
6rejection (the candidate is disqualified) or a soft penalty folded into
7ranking against other surviving candidates. Keeping each rule its own class lets selection
8compose/reorder/disable them without touching the scoring math itself.
9"""
10
11from __future__ import annotations
12
13from abc import ABC, abstractmethod
14from dataclasses import dataclass
15
16from music_assistant.controllers.streams.smart_fades.models import (
17 TransitionPlan,
18 TransitionStrategy,
19 TransitionTier,
20)
21from music_assistant.controllers.streams.smart_fades.vocal import (
22 COLLISION_SECONDS_LIMIT,
23 SHORT_FADE_SECONDS,
24 WEIGHTED_COLLISION_LIMIT,
25)
26
27from .candidates import RUNG_LADDER, Candidate, VocalOnsetEntryGenerator
28from .context import TransitionContext
29
30# Ambition ordering of the transition tiers, most ambitious first
31_TIER_ORDER: tuple[TransitionTier, ...] = (
32 TransitionTier.FULL_BLEND,
33 TransitionTier.TEMPO_BLEND,
34 TransitionTier.QUICK_FADE,
35)
36
37
38@dataclass(frozen=True, slots=True)
39class Verdict:
40 """One policy's judgment on a candidate: a rejection, or an accept with an optional penalty."""
41
42 penalty: float = 0.0
43 rejected: bool = False
44 reason: str = ""
45
46 @classmethod
47 def reject(cls, reason: str) -> Verdict:
48 """Disqualify a candidate outright, with a human-readable reason."""
49 return cls(rejected=True, reason=reason)
50
51 @classmethod
52 def ok(cls, penalty: float = 0.0, reason: str = "") -> Verdict:
53 """Accept a candidate, optionally carrying a soft penalty."""
54 return cls(penalty=penalty, reason=reason)
55
56
57class Policy(ABC):
58 """One independent scoring rule applied to a built candidate."""
59
60 @abstractmethod
61 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
62 """Judge one candidate against the shared per-transition context."""
63
64
65class VocalCollisionPolicy(Policy):
66 """Reject or penalize simultaneous outgoing/incoming vocal overlap inside the crossfade."""
67
68 collision_seconds_limit: float = COLLISION_SECONDS_LIMIT
69 weighted_collision_limit: float = WEIGHTED_COLLISION_LIMIT
70 weighted_penalty_scale: float = 20.0
71
72 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
73 """Judge one candidate against the shared per-transition context."""
74 if ctx.vocal_out_scoring is None or ctx.vocal_in_scoring is None:
75 return Verdict.ok()
76 metrics = candidate.metrics
77 if (
78 metrics.collision_seconds >= self.collision_seconds_limit
79 or metrics.weighted_collision_seconds >= self.weighted_collision_limit
80 ):
81 return Verdict.reject("vocal collision exceeds the guard limit")
82 # quadratic in the normalized residue: overlap is a threshold percept,
83 # so near-inaudible low-gain residue stays cheap while the cost climbs
84 # steeply toward the rejection boundary (panel-recommended shape)
85 normalized = metrics.weighted_collision_seconds / self.weighted_collision_limit
86 return Verdict.ok(normalized**2 * self.weighted_penalty_scale)
87
88
89class VocalTruncationPolicy(Policy):
90 """Reject a candidate that cuts off an audible outgoing vocal phrase."""
91
92 # An audible outgoing phrase cut by more than this many seconds reads as a
93 # truncation rather than an inaudible tail sliver
94 max_truncated_vocal: float = 0.25
95
96 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
97 """Judge one candidate against the shared per-transition context."""
98 if ctx.vocal_out_scoring is None:
99 return Verdict.ok()
100 # truncation = audible vocal BEYOND the candidate's anchor (cut off by the
101 # trim), not vocal inside the fade - a phrase riding the fade is normal
102 anchor = candidate.plan.fade_out_window
103 truncated = sum(
104 min(right, ctx.audio_end) - max(left, anchor)
105 for left, right in ctx.vocal_out_scoring.windows
106 if min(right, ctx.audio_end) > max(left, anchor)
107 )
108 if truncated > self.max_truncated_vocal:
109 return Verdict.reject("truncates an audible outgoing vocal phrase")
110 return Verdict.ok()
111
112
113class AudibleTrimPolicy(Policy):
114 """Reject a short fade that trims more audible outgoing material than it spans, else penalize."""
115
116 short_fade_seconds: float = SHORT_FADE_SECONDS
117 trim_penalty_per_second: float = 1.0
118
119 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
120 """Judge one candidate against the shared per-transition context."""
121 plan = candidate.plan
122 trim = candidate.metrics.audible_outgoing_trim
123 # missing vocal data must never disable this guard: a stale
124 # pre-vocal-analysis row would otherwise ship a huge, unprotected cut
125 if plan.crossfade_duration <= self.short_fade_seconds and trim > plan.crossfade_duration:
126 return Verdict.reject("audible trim exceeds a short fade's own duration")
127 return Verdict.ok(trim * self.trim_penalty_per_second)
128
129
130class DeadAirPolicy(Policy):
131 """Penalize a handover that strands the listener in silence before B's groove entry."""
132
133 grace_seconds: float = 4.0
134 dead_air_penalty_per_second: float = 1.0
135 # pre-fade outgoing level (relative to its own track peak) below which the
136 # outgoing is already a whisper and a slow incoming intro is welcome
137 hot_outgoing_floor: float = 0.178 # -15 dB
138 lookback_seconds: float = 8.0
139
140 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
141 """Judge one candidate against the shared per-transition context."""
142 plan = candidate.plan
143 gap = (
144 ctx.natural_entry
145 - (plan.fadein_trim_start or 0.0)
146 - plan.crossfade_duration
147 - self.grace_seconds
148 )
149 if gap <= 0.0 or not self._outgoing_is_hot(plan, ctx):
150 return Verdict.ok()
151 return Verdict.ok(gap * self.dead_air_penalty_per_second)
152
153 def _outgoing_is_hot(self, plan: TransitionPlan, ctx: TransitionContext) -> bool:
154 """Whether the outgoing still plays at level just before the candidate's fade."""
155 analysis = ctx.outgoing.analysis
156 rms = analysis.rms_energy
157 duration = analysis.duration
158 if rms is None or len(rms) == 0 or not duration:
159 # without energy data the gate cannot clear the penalty; dead air
160 # after an unknown outgoing is worse than a trimmed intro
161 return True
162 peak = max(rms)
163 if peak <= 0.0:
164 return False
165 bin_seconds = duration / len(rms)
166 # sample the source RMS over the last lookback window ending at the cut
167 # anchor; the stored energy is unfaded, so this reads the outgoing's own
168 # level right before the cut with no crossfade-ramp or tempo-stretch mixed in
169 anchor = ctx.buffer_offset + plan.fade_out_window
170 low = max(0, int((anchor - self.lookback_seconds) / bin_seconds))
171 high = max(low + 1, min(len(rms), int(anchor / bin_seconds)))
172 window = rms[low:high]
173 return sum(window) / len(window) >= peak * self.hot_outgoing_floor
174
175
176class OverlapPreferencePolicy(Policy):
177 """Prefer the tier's top rung and the context's chosen tier."""
178
179 rung_penalty_per_step: float = 10.0
180 tier_penalty_per_step: float = 15.0
181
182 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
183 """Judge one candidate against the shared per-transition context."""
184 spec = candidate.spec
185 if spec.strategy is TransitionStrategy.LAZY_OVERLAY:
186 return Verdict.ok() # the overlay has no rung/tier notion to score
187 rung_gap = RUNG_LADDER.index(spec.bars) - RUNG_LADDER.index(candidate.ideal_bars)
188 tier_steps = max(0, _TIER_ORDER.index(spec.tier) - _TIER_ORDER.index(ctx.tier))
189 penalty = self.rung_penalty_per_step * rung_gap
190 penalty += self.tier_penalty_per_step * tier_steps
191 return Verdict.ok(penalty)
192
193
194class AnchorAlignmentPolicy(Policy):
195 """Prefer downbeat-anchored fades and groove-aligned incoming entries."""
196
197 downbeat_penalty: float = 4.0
198 entry_misalignment_penalty: float = 2.0
199
200 def evaluate(self, candidate: Candidate, ctx: TransitionContext) -> Verdict:
201 """Judge one candidate against the shared per-transition context."""
202 if candidate.spec.strategy is TransitionStrategy.LAZY_OVERLAY:
203 return Verdict.ok() # an unphrased overlay doesn't pretend beat alignment
204 penalty = 0.0
205 if not candidate.metrics.anchor_on_downbeat:
206 penalty += self.downbeat_penalty
207 spec = candidate.spec
208 # a generator-pinned entry counts as groove-aligned only when the
209 # generator itself already lands it on the vocal onset
210 if spec.entry_s is not None and spec.source != VocalOnsetEntryGenerator.name:
211 penalty += self.entry_misalignment_penalty
212 return Verdict.ok(penalty)
213
214
215def default_policies() -> tuple[Policy, ...]:
216 """Return the standard policy set applied to every candidate, in evaluation order."""
217 return (
218 VocalCollisionPolicy(),
219 VocalTruncationPolicy(),
220 AudibleTrimPolicy(),
221 DeadAirPolicy(),
222 OverlapPreferencePolicy(),
223 AnchorAlignmentPolicy(),
224 )
225