/
/
/
1"""
2Smart Fades - vocal-activity contract and vocal-collision math.
3
4Parses the optional 1800-bin ``vocal_activity`` list stored in
5``AudioAnalysisData.extra_data``, turns it into hysteresis-gated vocal windows,
6and scores how much two tracks' vocals would collide inside a candidate
7crossfade. Every function here works over plain floats and lists - no NumPy -
8so a missing or malformed timeline never pulls it onto a path that would
9otherwise stay numpy-free.
10"""
11
12from __future__ import annotations
13
14import math
15from dataclasses import dataclass
16from typing import TYPE_CHECKING
17
18if TYPE_CHECKING:
19 from music_assistant.models.audio_analysis import AudioAnalysisData
20
21# The analysis provider aggregates every canonical analysis timeline to this
22# fixed number of bins, independent of track duration
23VOCAL_ACTIVITY_BINS = 1800
24
25# Hysteresis thresholds: a run opens once probability reaches OPEN and only
26# closes once it drops below CLOSE, so a phrase's quieter syllables don't
27# fragment its window into several
28VOCAL_OPEN_THRESHOLD = 0.5
29VOCAL_CLOSE_THRESHOLD = 0.3
30# Padding absorbs the detector's attack/release lag around a real phrase
31VOCAL_LEFT_PADDING = 0.25
32VOCAL_RIGHT_PADDING = 0.75
33# A raw run shorter than this is a detector blip (crowd shout, vocal-formant
34# one-shot), not a phrase: it never becomes a window
35MIN_VOCAL_RUN = 0.3
36# Two padded windows closer than this bridge into one; the gap itself is
37# clamped to one beat, since a shorter silence is almost certainly a breath
38# inside the same phrase rather than a real gap between two of them
39MIN_VOCAL_GAP = 0.5
40MAX_VOCAL_GAP = 1.0
41# Sustained-evidence gate: a run only becomes a window when its peak OR its
42# mean shows real confidence; weak short runs are backing 'aahhs'/detector
43# noise, and a window here can veto an entire 8-bar blend downstream
44VOCAL_MIN_RUN_PEAK = 0.85
45VOCAL_MIN_RUN_MEAN = 0.65
46
47# Two-sided vocal-collision guard, in rendered-crossfade seconds. Collisions
48# are scored on UNPADDED windows (padding is silence, not vocal), and the
49# gain-weighted integral is the binding test: the raw limit only backstops
50# pathological cases the weight can't see
51COLLISION_SECONDS_LIMIT = 2.0
52WEIGHTED_COLLISION_LIMIT = 0.35
53
54# Click-free equal-power fallback bounds, used when no phrased candidate can
55# avoid a vocal collision; the floor keeps the handoff reading as intentional
56# rather than as a radio edit
57MIN_HANDOFF_SECONDS = 0.4
58MAX_HANDOFF_SECONDS = 1.0
59# crossfades at or under this length must not drop more audible outgoing
60# material than the overlap itself covers
61SHORT_FADE_SECONDS = 8.0
62
63
64@dataclass(slots=True)
65class VocalMask:
66 """Hysteresis-gated vocal-activity windows over one buffer, in buffer-local seconds."""
67
68 windows: list[tuple[float, float]]
69
70 def last_end(self) -> float:
71 """End of the last window, or 0.0 when the mask has none."""
72 return self.windows[-1][1] if self.windows else 0.0
73
74 def clamped_to(self, upper_bound: float) -> VocalMask:
75 """
76 Return a copy whose windows never reach past ``upper_bound``.
77
78 Windows starting at or beyond the bound are dropped entirely rather
79 than clipped to a zero-length sliver at the edge.
80
81 :param upper_bound: Latest buffer-local second any window may reach.
82 """
83 clipped = [
84 (max(0.0, left), min(upper_bound, right))
85 for left, right in self.windows
86 if left < upper_bound
87 ]
88 return VocalMask(windows=clipped)
89
90
91@dataclass(frozen=True, slots=True)
92class VocalHysteresisConfig:
93 """
94 Tunable hysteresis/padding/gap-bridging thresholds for ``build_vocal_windows``.
95
96 A run also needs a confident peak or mean probability to survive as a window.
97 """
98
99 open_threshold: float = VOCAL_OPEN_THRESHOLD
100 close_threshold: float = VOCAL_CLOSE_THRESHOLD
101 left_padding: float = VOCAL_LEFT_PADDING
102 right_padding: float = VOCAL_RIGHT_PADDING
103 min_run: float = MIN_VOCAL_RUN
104 min_gap: float = MIN_VOCAL_GAP
105 max_gap: float = MAX_VOCAL_GAP
106 min_run_peak: float = VOCAL_MIN_RUN_PEAK
107 min_run_mean: float = VOCAL_MIN_RUN_MEAN
108
109
110DEFAULT_VOCAL_CONFIG = VocalHysteresisConfig()
111# retention keeps audio, so it wants recall; only the planner's veto needs the gate
112PROTECTIVE_VOCAL_CONFIG = VocalHysteresisConfig(min_run_peak=0.0, min_run_mean=0.0)
113
114
115@dataclass(frozen=True, slots=True)
116class VocalTimeline:
117 """A validated vocal-activity timeline: per-bin probabilities and their spacing."""
118
119 probabilities: list[float]
120 frame_duration: float
121
122
123def parse_vocal_probabilities(analysis: AudioAnalysisData) -> VocalTimeline | None:
124 """
125 Validate and return the stored 1800-bin vocal-activity timeline.
126
127 Returns ``None`` when ``extra_data["vocal_activity"]`` is absent or fails
128 any part of the contract: it is not a list of exactly 1800 finite numeric
129 probabilities in the inclusive range 0..1, or the analysis has no finite
130 positive duration. Callers must treat that as "vocal-aware logic is
131 unavailable for this track".
132
133 :param analysis: Stored analysis row to read ``vocal_activity`` from.
134 """
135 duration = analysis.duration
136 if (
137 isinstance(duration, bool)
138 or not isinstance(duration, (int, float))
139 or not math.isfinite(duration)
140 or duration <= 0.0
141 ):
142 return None
143 vocal_activity = (analysis.extra_data or {}).get("vocal_activity")
144 if not isinstance(vocal_activity, list) or len(vocal_activity) != VOCAL_ACTIVITY_BINS:
145 return None
146 values: list[float] = []
147 for raw_value in vocal_activity:
148 if isinstance(raw_value, bool) or not isinstance(raw_value, (int, float)):
149 return None
150 if raw_value < 0.0 or raw_value > 1.0:
151 return None
152 value = float(raw_value)
153 if not math.isfinite(value):
154 return None
155 values.append(value)
156 return VocalTimeline(
157 probabilities=values,
158 frame_duration=float(duration) / VOCAL_ACTIVITY_BINS,
159 )
160
161
162def build_vocal_windows(
163 probabilities: list[float],
164 frame_duration: float,
165 start_s: float,
166 end_s: float,
167 *,
168 beat_duration: float | None = None,
169 config: VocalHysteresisConfig = DEFAULT_VOCAL_CONFIG,
170) -> VocalMask:
171 """
172 Turn a validated vocal-activity timeline into hysteresis-gated windows over a time range.
173
174 A run opens once probability reaches ``config.open_threshold`` and only
175 closes once it drops below ``config.close_threshold``, then gets padded
176 left/right and bridged across short gaps so one phrase doesn't fragment
177 into several windows - and dropped if it never reaches a confident peak
178 or mean. Probabilities are indexed from the same time origin as
179 ``start_s``/``end_s``; windows outside that range are dropped.
180
181 :param probabilities: Validated per-bin vocal probabilities (0..1).
182 :param frame_duration: Seconds spanned by each probability bin.
183 :param start_s: Range start (inclusive); windows are clipped to it.
184 :param end_s: Range end (exclusive); windows are clipped to it.
185 :param beat_duration: The track's beat length in seconds, used to size the
186 gap bridge; ``None`` keeps the bridge at its minimum.
187 :param config: Hysteresis/padding/gap-bridging thresholds.
188 """
189 if frame_duration <= 0.0 or not math.isfinite(frame_duration):
190 return VocalMask(windows=[])
191
192 runs: list[tuple[int, int]] = []
193 run_start: int | None = None
194 for index, probability in enumerate(probabilities):
195 if probability >= config.open_threshold and run_start is None:
196 run_start = index
197 elif probability < config.close_threshold and run_start is not None:
198 runs.append((run_start, index))
199 run_start = None
200 if run_start is not None:
201 runs.append((run_start, len(probabilities)))
202 # a sub-min_run blip would still grow to over a second once padded, enough
203 # to fail a candidate on its own â drop it before padding can inflate it
204 min_run_frames = max(
205 1,
206 math.ceil(math.nextafter(config.min_run / frame_duration, -math.inf)),
207 )
208 timed_runs: list[tuple[float, float]] = []
209 for run_start_frame, run_end_frame in runs:
210 if run_end_frame - run_start_frame < min_run_frames:
211 continue
212 run = probabilities[run_start_frame:run_end_frame]
213 if max(run) < config.min_run_peak and sum(run) / len(run) < config.min_run_mean:
214 continue
215 timed_runs.append((run_start_frame * frame_duration, run_end_frame * frame_duration))
216
217 gap = min(config.max_gap, max(config.min_gap, beat_duration or config.min_gap))
218 windows: list[tuple[float, float]] = []
219 for left, right in timed_runs:
220 padded_left = max(start_s, left - config.left_padding)
221 padded_right = min(end_s, right + config.right_padding)
222 if padded_right <= start_s or padded_left >= end_s:
223 continue
224 if windows and padded_left - windows[-1][1] <= gap:
225 windows[-1] = (windows[-1][0], max(windows[-1][1], padded_right))
226 else:
227 windows.append((padded_left, padded_right))
228 return VocalMask(windows=windows)
229
230
231def mask_saturated(mask: VocalMask, span: float) -> bool:
232 """
233 Whether a mask's windows cover >=90% of a span (near-continuous vocal, no fine structure).
234
235 :param mask: Vocal-activity mask to measure.
236 :param span: Length in seconds of the range the mask was built over.
237 """
238 covered = sum(right - left for left, right in mask.windows)
239 return covered >= 0.9 * max(0.001, span)
240
241
242def merge_windows(windows: list[tuple[float, float]]) -> list[tuple[float, float]]:
243 """
244 Merge overlapping or touching (left, right) intervals into their minimal covering set.
245
246 :param windows: Intervals to merge, in any order.
247 """
248 merged: list[tuple[float, float]] = []
249 for left, right in sorted(windows):
250 if merged and left <= merged[-1][1]:
251 merged[-1] = (merged[-1][0], max(merged[-1][1], right))
252 else:
253 merged.append((left, right))
254 return merged
255
256
257def collision_metrics(
258 outgoing_windows: list[tuple[float, float]],
259 incoming_windows: list[tuple[float, float]],
260 duration: float,
261) -> tuple[float, float]:
262 """
263 Simultaneous and gain-weighted vocal overlap, in rendered-crossfade-local seconds.
264
265 The weight models the acrossfade curve's simultaneous audible power at a
266 rendered position (``phase = position / duration``): ``4 * phase * (1 -
267 phase)``. It is integrated analytically (exact antiderivative, not a
268 sampled sum) so the score is deterministic and independent of any step size.
269
270 :param outgoing_windows: Outgoing vocal windows, in rendered crossfade-local seconds.
271 :param incoming_windows: Incoming vocal windows, in the same rendered timeline.
272 :param duration: Rendered crossfade duration in seconds.
273 """
274 if duration <= 0.0:
275 return 0.0, 0.0
276 overlaps: list[tuple[float, float]] = []
277 for out_left, out_right in outgoing_windows:
278 for in_left, in_right in incoming_windows:
279 left = max(0.0, out_left, in_left)
280 right = min(duration, out_right, in_right)
281 if right > left:
282 overlaps.append((left, right))
283 merged = merge_windows(overlaps)
284 collision_seconds = sum(right - left for left, right in merged)
285 weighted_collision = sum(_weighted_overlap(left, right, duration) for left, right in merged)
286 return collision_seconds, weighted_collision
287
288
289def _weighted_overlap(left: float, right: float, duration: float) -> float:
290 """Exact integral of the simultaneous-power weight 4p(1-p) over [left, right]."""
291
292 def primitive(t: float) -> float:
293 phase = t / duration
294 return duration * (2.0 * phase * phase - (4.0 / 3.0) * phase * phase * phase)
295
296 return primitive(right) - primitive(left)
297