/
/
/
1"""
2Smart Fades - EQ assembly and the emergency vocal handoff.
3
4The candidate factory builds every candidate with a neutral EQ plan (scoring
5never needs it), so the bass/mid/high handover EQ is computed exactly once,
6for the winner only, by ``PlanAssembler``. ``EmergencyHandoffFactory`` builds
7the click-free equal-power fallback used when every phrased candidate still
8collides with the incoming track's vocal.
9"""
10
11from __future__ import annotations
12
13from dataclasses import replace
14from typing import TYPE_CHECKING
15
16from music_assistant.controllers.streams.smart_fades.bands import (
17 loudness_referenced_level,
18 smoothstep,
19 window_duty,
20 window_fraction,
21 window_level,
22)
23from music_assistant.controllers.streams.smart_fades.filters import ShelfType
24from music_assistant.controllers.streams.smart_fades.helpers import db_ramp
25from music_assistant.controllers.streams.smart_fades.models import (
26 BAND_RMS_BANDS,
27 EqPlan,
28 ShelfSchedule,
29 TempoPlan,
30 TransitionPlan,
31 TransitionStrategy,
32 TransitionTier,
33)
34from music_assistant.controllers.streams.smart_fades.vocal import (
35 COLLISION_SECONDS_LIMIT,
36 MAX_HANDOFF_SECONDS,
37 MIN_HANDOFF_SECONDS,
38 SHORT_FADE_SECONDS,
39 WEIGHTED_COLLISION_LIMIT,
40)
41
42from .candidates import CandidateSpec, _nearest_protective_anchor, _outgoing_vocal_end
43
44if TYPE_CHECKING:
45 import logging
46
47 from music_assistant.controllers.streams.smart_fades.models import BandProfile
48
49 from .candidates import Candidate, CandidateFactory
50 from .context import TransitionContext
51
52# Bass-swap EQ: shelf corners/depths as on real club mixers
53_LOW_SHELF_FREQ: int = 100
54_HIGH_SHELF_FREQ: int = 13000
55_EQ_KILL_DB: float = -26.0
56_HIGH_EASE_DB: float = -20.0
57# bass handover spans half the overlap; <2 bars reads as an event, >8 bars is masked
58_BASS_SWAP_FRACTION: float = 0.5
59_BASS_SWAP_MIN_BARS: int = 2
60_BASS_SWAP_MAX_BARS: int = 8
61# decision window for the reciprocal swap depth gates, in A/B-bars
62_BASS_SWAP_WINDOW_BARS: int = 8
63
64# Reciprocal bass-swap gate: smoothstep each side's low-band fraction (over the
65# OTHER deck's window) to scale its own kill depth; dropped below _EQ_BYPASS_BELOW_DB.
66# Corridor sits deliberately below the published dance-master figures (~0.34-0.45
67# of power under 120Hz; pop ~0.14 â Elowsson & Friberg 2017, Pestana et al. 2013):
68# it reads an 8-bar transition window, not a whole-track LTAS, and putting lo under
69# the pop mean lets ordinary pop still earn a partial swap. Final values corpus-tuned.
70_LOW_GATE_LO: float = 0.10
71_LOW_GATE_HI: float = 0.25
72_EQ_BYPASS_BELOW_DB: float = -6.0
73
74# Reciprocal high-ease gate, same shape on the high band. Mean-music LTAS is ~0.02 in
75# 4-11kHz (Elowsson & Friberg 2017): lo = an average track earns no duck, hi = clearly
76# brighter than average earns the full ease. A deck whose own-window high level is below
77# _HIGH_OWN_SIDE_FLOOR of its own reference gets no shelf at all.
78_HIGH_GATE_LO: float = 0.02
79_HIGH_GATE_HI: float = 0.06
80_HIGH_OWN_SIDE_FLOOR: float = 0.25
81# Cymbal-wash mode: when both own-windows read bright and comparably loud a plain
82# reciprocal ease would stack their highs, so duck A complementary with B's restore.
83_WASH_DUTY: float = 0.8
84_WASH_DEPTH_DB: float = -26.0
85_WASH_LEVEL_TOLERANCE_DB: float = 6.0
86_WASH_MIN_BLEND_BARS: int = 8
87
88# Measured mid-band (vocal) swap: bass-swap gate shape on the mid band, gated also on
89# duty so one loud mid bar can't unlock a swap over otherwise instrumental material.
90_MID_FREQ: int = 1200
91_MID_WIDTH_OCT: float = 2.5
92# Corridor sits in the gap between vocal-forward mixes (~0.25-0.45 mid fraction) and
93# instrumental dance (~0.10-0.15); absolute placement verified on our unweighted
94# pipeline (LTAS: Elowsson & Friberg 2017, Pestana et al. 2013).
95_MID_GATE_LO: float = 0.18
96_MID_GATE_HI: float = 0.30
97_MID_DUTY_LO: float = 0.60
98_MID_DUTY_HI: float = 0.85
99# the mid (vocal) swap is capped shallower on a tempo blend than a full blend
100_MID_CAP_FULL_DB: float = -8.0
101_MID_CAP_TEMPO_DB: float = -6.0
102_MID_BYPASS_BELOW_DB: float = -1.0
103
104# Dip guard: combined qsin-weighted power of both decks may never sag more than this
105# below its plateau across the overlap (outside the intentional bass-handover notch).
106_MAX_PREDICTED_DIP_DB: float = 3.0
107
108
109class PlanAssembler:
110 """Applies the full EQ handover and its dip-guard repair to the selected candidate's plan."""
111
112 def __init__(self, ctx: TransitionContext, logger: logging.Logger) -> None:
113 """Initialize the assembler for one transition."""
114 self._ctx = ctx
115 self._logger = logger
116
117 def finalize(self, candidate: Candidate) -> TransitionPlan:
118 """
119 Return the candidate's plan with the full EQ schedule and dip guard applied.
120
121 EQ is deliberately absent from every factory-built candidate (scoring
122 never needs it); this is where the bass/mid/high handover for the
123 selected winner is computed and folded in.
124 """
125 # the winner's metrics ride along: consumers read them off the plan
126 return replace(
127 candidate.plan,
128 eq_plan=self._choose_eq(candidate.plan),
129 fadeout_curve=self._choose_fadeout_curve(candidate.plan),
130 metrics=candidate.metrics,
131 )
132
133 def _choose_eq(self, plan: TransitionPlan) -> EqPlan:
134 """Plan the low/mid/high EQ handover, centered on the swap point."""
135 ctx = self._ctx
136 effective_end = plan.fade_out_window
137 crossfade_duration = plan.crossfade_duration
138 tempo_plan = plan.tempo_plan
139 # A-side schedules are in A-input time (rendered before the tempo stretch);
140 # B-side schedules are in B's post-trim time, where t=0 is the crossfade start
141 bar_in = ctx.incoming.beats_per_bar * 60.0 / ctx.incoming.bpm
142 swap_at = self._choose_swap_point(crossfade_duration, plan.fadein_trim_start)
143 swap_len = min(
144 max(_BASS_SWAP_FRACTION * crossfade_duration, _BASS_SWAP_MIN_BARS * bar_in),
145 _BASS_SWAP_MAX_BARS * bar_in,
146 crossfade_duration,
147 )
148 # pull a late swap point back so the centered window still fits the overlap
149 swap_at = min(swap_at, crossfade_duration - swap_len / 2)
150 ease = 0.25 * crossfade_duration
151
152 # the ramp completes before the crossfade, so rendered-to-A-input mapping is linear
153 ratio = tempo_plan.steps[-1][1] if tempo_plan else 1.0
154 cf_start_input = effective_end - crossfade_duration * ratio
155 swap_at_input = cf_start_input + swap_at * ratio
156 swap_len_input = swap_len * ratio
157 start_in = max(0.0, swap_at - swap_len / 2)
158 start_out = max(cf_start_input, swap_at_input - swap_len_input / 2)
159
160 depth_a, depth_b = self._choose_swap_depths(effective_end)
161
162 low_out = (
163 ShelfSchedule(
164 ShelfType.LOW,
165 _LOW_SHELF_FREQ,
166 [(0.0, 0.0), *db_ramp(start_out, swap_len_input, 0.0, depth_a)],
167 )
168 if abs(depth_a) >= abs(_EQ_BYPASS_BELOW_DB)
169 else None
170 )
171 low_in = (
172 ShelfSchedule(
173 ShelfType.LOW,
174 _LOW_SHELF_FREQ,
175 [(0.0, depth_b), *db_ramp(start_in, swap_len, depth_b, 0.0)],
176 )
177 if abs(depth_b) >= abs(_EQ_BYPASS_BELOW_DB)
178 else None
179 )
180 high_out, high_in = self._choose_high_swap(
181 effective_end,
182 start_out,
183 start_in,
184 cf_start_input,
185 swap_len_input,
186 ease,
187 ratio,
188 crossfade_duration,
189 )
190
191 depth_mid_a, depth_mid_b = self._choose_mid_swap_depths(effective_end, plan.tier)
192 mid_out = (
193 ShelfSchedule(
194 ShelfType.PEAK,
195 _MID_FREQ,
196 [(0.0, 0.0), *db_ramp(start_out, swap_len_input, 0.0, depth_mid_a)],
197 width_oct=_MID_WIDTH_OCT,
198 )
199 if depth_mid_a is not None
200 else None
201 )
202 mid_in = (
203 ShelfSchedule(
204 ShelfType.PEAK,
205 _MID_FREQ,
206 [(0.0, depth_mid_b), *db_ramp(start_in, swap_len, depth_mid_b, 0.0)],
207 width_oct=_MID_WIDTH_OCT,
208 )
209 if depth_mid_b is not None
210 else None
211 )
212
213 eq_plan = EqPlan(
214 swap_at=swap_at,
215 low_out=low_out,
216 low_in=low_in,
217 high_out=high_out,
218 high_in=high_in,
219 mid_out=mid_out,
220 mid_in=mid_in,
221 )
222 # the low crossover's rendered ramp window; the dip guard exempts it as
223 # the intentional bass-handover gesture, but ONLY when a low swap is
224 # actually engaged â a bass-light pair has no handover to exempt, so
225 # the notch collapses to an interval that never matches a sample time
226 notch = (
227 (start_in, start_in + swap_len)
228 if (low_out is not None or low_in is not None)
229 else (-1.0, -1.0)
230 )
231 return self._apply_dip_guard(
232 eq_plan,
233 effective_end=effective_end,
234 crossfade_duration=crossfade_duration,
235 cf_start_input=cf_start_input,
236 ratio=ratio,
237 swap_at=swap_at,
238 bar_in=bar_in,
239 notch=notch,
240 )
241
242 def _choose_swap_point(
243 self, crossfade_duration: float, fadein_trim_start: float | None
244 ) -> float:
245 """Pick the bass-swap moment: B's groove entry when inside the overlap, else 60% through."""
246 import numpy as np # noqa: PLC0415
247
248 trim = fadein_trim_start or 0.0
249 candidate = self._ctx.natural_entry - trim
250 if not 0.0 < candidate <= crossfade_duration:
251 candidate = 0.6 * crossfade_duration
252 # snap to the incoming grid so the new bassline lands on its own 1
253 post_trim = self._ctx.incoming.downbeats - trim
254 post_trim = post_trim[(post_trim > 0.0) & (post_trim < crossfade_duration)]
255 if len(post_trim):
256 candidate = float(post_trim[np.argmin(np.abs(post_trim - candidate))])
257 return candidate
258
259 def _swap_windows(
260 self, effective_end: float, window_bars: int
261 ) -> tuple[tuple[float, float], tuple[float, float]]:
262 """Reciprocal decision windows for the swap gates: A's outgoing tail, B's incoming head."""
263 ctx = self._ctx
264 bar_out = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
265 bar_in = ctx.incoming.beats_per_bar * 60.0 / ctx.incoming.bpm
266 anchor_media = ctx.buffer_offset + effective_end
267 w_a_out = (anchor_media - window_bars * bar_out, anchor_media)
268 entry = ctx.natural_entry
269 w_b_in = (entry, entry + window_bars * bar_in)
270 return w_a_out, w_b_in
271
272 def _choose_swap_depths(self, effective_end: float) -> tuple[float, float]:
273 """Reciprocally scale each side's bass-kill depth to the other deck's measured bass."""
274 ctx = self._ctx
275 # missing band data on either side keeps the shipped full-depth kill (bit-identical)
276 if ctx.outgoing_profile is None or ctx.incoming_profile is None:
277 return _EQ_KILL_DB, _EQ_KILL_DB
278 w_a_out, w_b_in = self._swap_windows(effective_end, _BASS_SWAP_WINDOW_BARS)
279 f_low_b_in = window_fraction(ctx.incoming_profile, "low", *w_b_in)
280 f_low_a_out = window_fraction(ctx.outgoing_profile, "low", *w_a_out)
281 depth_a = _EQ_KILL_DB * smoothstep(f_low_b_in, _LOW_GATE_LO, _LOW_GATE_HI)
282 depth_b = _EQ_KILL_DB * smoothstep(f_low_a_out, _LOW_GATE_LO, _LOW_GATE_HI)
283 return depth_a, depth_b
284
285 def _choose_high_swap(
286 self,
287 effective_end: float,
288 start_out: float,
289 start_in: float,
290 cf_start_input: float,
291 swap_len_input: float,
292 ease: float,
293 ratio: float,
294 crossfade_duration: float,
295 ) -> tuple[ShelfSchedule | None, ShelfSchedule | None]:
296 """Build the high-ease shelves: reciprocal depths, own-side no-op skip, cymbal-wash mode."""
297 ctx = self._ctx
298 wash = False
299 if ctx.outgoing_profile is None or ctx.incoming_profile is None:
300 depth_a = depth_b = _HIGH_EASE_DB
301 else:
302 w_a_out, w_b_in = self._swap_windows(effective_end, _BASS_SWAP_WINDOW_BARS)
303 f_high_b_in = window_fraction(ctx.incoming_profile, "high", *w_b_in)
304 f_high_a_out = window_fraction(ctx.outgoing_profile, "high", *w_a_out)
305 depth_a = _HIGH_EASE_DB * smoothstep(f_high_b_in, _HIGH_GATE_LO, _HIGH_GATE_HI)
306 depth_b = _HIGH_EASE_DB * smoothstep(f_high_a_out, _HIGH_GATE_LO, _HIGH_GATE_HI)
307
308 own_dark_a = window_level(ctx.outgoing_profile, "high", *w_a_out) < (
309 _HIGH_OWN_SIDE_FLOOR * ctx.outgoing_profile.reference["high"]
310 )
311 own_dark_b = window_level(ctx.incoming_profile, "high", *w_b_in) < (
312 _HIGH_OWN_SIDE_FLOOR * ctx.incoming_profile.reference["high"]
313 )
314 if own_dark_a:
315 depth_a = 0.0
316 if own_dark_b:
317 depth_b = 0.0
318
319 wash = self._wash_mode_engages(w_a_out, w_b_in, crossfade_duration)
320 if wash:
321 depth_a = _WASH_DEPTH_DB
322
323 high_out = (
324 ShelfSchedule(
325 ShelfType.HIGH,
326 _HIGH_SHELF_FREQ,
327 self._high_out_steps(
328 depth_a,
329 start_out,
330 start_in,
331 cf_start_input,
332 swap_len_input,
333 ease,
334 ratio,
335 wash=wash,
336 ),
337 )
338 if abs(depth_a) >= abs(_EQ_BYPASS_BELOW_DB)
339 else None
340 )
341 high_in = (
342 ShelfSchedule(
343 ShelfType.HIGH,
344 _HIGH_SHELF_FREQ,
345 [
346 (0.0, depth_b),
347 *db_ramp(max(0.0, start_in - ease), ease, depth_b, 0.0),
348 ],
349 )
350 if abs(depth_b) >= abs(_EQ_BYPASS_BELOW_DB)
351 else None
352 )
353 return high_out, high_in
354
355 @staticmethod
356 def _high_out_steps(
357 depth_a: float,
358 start_out: float,
359 start_in: float,
360 cf_start_input: float,
361 swap_len_input: float,
362 ease: float,
363 ratio: float,
364 *,
365 wash: bool,
366 ) -> list[tuple[float, float]]:
367 """Build A's high-duck ramp: shipped post-swap placement, or wash mode's mirrored duck."""
368 if wash:
369 # mirror B's restore window; it ends at start_in inside the overlap,
370 # so the duck can never overrun the crossfade end
371 start = cf_start_input + max(0.0, start_in - ease) * ratio
372 else:
373 start = start_out + swap_len_input
374 return [(0.0, 0.0), *db_ramp(start, ease * ratio, 0.0, depth_a)]
375
376 def _wash_mode_engages(
377 self,
378 w_a_out: tuple[float, float],
379 w_b_in: tuple[float, float],
380 crossfade_duration: float,
381 ) -> bool:
382 """Return True when both decks read bright, comparably loud, over a long enough blend."""
383 import numpy as np # noqa: PLC0415
384
385 ctx = self._ctx
386 assert ctx.outgoing_profile is not None # narrowed by the caller
387 assert ctx.incoming_profile is not None
388 bar_out = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
389 if crossfade_duration < _WASH_MIN_BLEND_BARS * bar_out:
390 return False
391 # absolute brightness floor: duty vs a track's OWN reference measures
392 # consistency, not brightness â a dark steady track has duty 1.0
393 f_high_a = window_fraction(ctx.outgoing_profile, "high", *w_a_out)
394 f_high_b = window_fraction(ctx.incoming_profile, "high", *w_b_in)
395 if f_high_a < _HIGH_GATE_HI or f_high_b < _HIGH_GATE_HI:
396 return False
397 duty_a = window_duty(ctx.outgoing_profile, "high", *w_a_out, k=0.5)
398 duty_b = window_duty(ctx.incoming_profile, "high", *w_b_in, k=0.5)
399 if duty_a < _WASH_DUTY or duty_b < _WASH_DUTY:
400 return False
401 level_a = loudness_referenced_level(ctx.outgoing_profile, "high", *w_a_out)
402 level_b = loudness_referenced_level(ctx.incoming_profile, "high", *w_b_in)
403 if level_a <= 0.0 or level_b <= 0.0:
404 return False
405 # the planner has no access to playback state, so this presumes loudness-
406 # normalized playback; that assumption is strictly more conservative than
407 # a duty-only fallback, which would engage wash mode more often
408 level_gap_db = abs(10.0 * float(np.log10(level_a / level_b)))
409 return level_gap_db <= _WASH_LEVEL_TOLERANCE_DB
410
411 def _choose_mid_swap_depths(
412 self, effective_end: float, tier: TransitionTier
413 ) -> tuple[float | None, float | None]:
414 """Gate and scale the measured mid-band (vocal) swap depth; ``None`` means bypass."""
415 ctx = self._ctx
416 cap = _MID_CAP_FULL_DB if tier is TransitionTier.FULL_BLEND else _MID_CAP_TEMPO_DB
417 if tier is TransitionTier.QUICK_FADE:
418 return None, None
419 if ctx.outgoing_profile is None or ctx.incoming_profile is None:
420 return None, None
421 w_a_out, w_b_in = self._swap_windows(effective_end, _BASS_SWAP_WINDOW_BARS)
422
423 def _score(profile: BandProfile, window: tuple[float, float]) -> float:
424 f_mid = window_fraction(profile, "mid", *window)
425 duty_mid = window_duty(profile, "mid", *window, k=0.5)
426 return smoothstep(f_mid, _MID_GATE_LO, _MID_GATE_HI) * smoothstep(
427 duty_mid, _MID_DUTY_LO, _MID_DUTY_HI
428 )
429
430 score_a = _score(ctx.outgoing_profile, w_a_out)
431 score_b = _score(ctx.incoming_profile, w_b_in)
432 # the weaker side rules: a swap only reads as a handover when both decks carry a mid element
433 depth = cap * min(score_a, score_b)
434 if abs(depth) < abs(_MID_BYPASS_BELOW_DB):
435 return None, None
436 return depth, depth
437
438 def _apply_dip_guard(
439 self,
440 eq_plan: EqPlan,
441 *,
442 effective_end: float,
443 crossfade_duration: float,
444 cf_start_input: float,
445 ratio: float,
446 swap_at: float,
447 bar_in: float,
448 notch: tuple[float, float],
449 ) -> EqPlan:
450 """Remediate a predicted outside-notch dip: shrink mid depth, then steepen low ramps."""
451 ctx = self._ctx
452 if ctx.outgoing_profile is None or ctx.incoming_profile is None:
453 return eq_plan
454 w_a_out, w_b_in = self._swap_windows(effective_end, _BASS_SWAP_WINDOW_BARS)
455 f_a = {
456 band: window_fraction(ctx.outgoing_profile, band, *w_a_out) for band in BAND_RMS_BANDS
457 }
458 f_b = {
459 band: window_fraction(ctx.incoming_profile, band, *w_b_in) for band in BAND_RMS_BANDS
460 }
461
462 def dip_db(plan: EqPlan) -> float:
463 return self._predicted_dip_db(
464 plan, crossfade_duration, cf_start_input, ratio, f_a, f_b, notch
465 )
466
467 if dip_db(eq_plan) <= _MAX_PREDICTED_DIP_DB:
468 return eq_plan
469
470 # (1) reduce mid depth toward bypass, in steps, until the dip clears
471 # or mid is fully bypassed
472 shrink_steps = (0.75, 0.5, 0.25, 0.0)
473 for shrink in shrink_steps:
474 candidate = eq_plan.with_mid_depth_scaled(shrink, bypass_below_db=_MID_BYPASS_BELOW_DB)
475 if dip_db(candidate) <= _MAX_PREDICTED_DIP_DB or shrink == 0.0:
476 eq_plan = candidate
477 break
478 if dip_db(eq_plan) <= _MAX_PREDICTED_DIP_DB:
479 return eq_plan
480
481 # (2) steepen the low ramps to the 2-bar floor; endpoints untouched.
482 # This is the last remediation step (never shallow the low depth â
483 # there is no further knob beyond the floor), so its result is
484 # returned whether or not it fully clears the budget.
485 if eq_plan.low_out is None and eq_plan.low_in is None:
486 # a bass-light pair has no low handover to tighten and no notch to
487 # narrow; mid-scaling was the only lever
488 return eq_plan
489 floor_len = _BASS_SWAP_MIN_BARS * bar_in
490 return eq_plan.with_low_ramps_steepened(swap_at, floor_len, ratio, cf_start_input)
491
492 def _predicted_dip_db(
493 self,
494 eq_plan: EqPlan,
495 crossfade_duration: float,
496 cf_start_input: float,
497 ratio: float,
498 f_a: dict[str, float],
499 f_b: dict[str, float],
500 notch: tuple[float, float],
501 n_samples: int = 64,
502 ) -> float:
503 """Max plateau-to-valley drop, in dB, of the predicted combined qsin-weighted power."""
504 import numpy as np # noqa: PLC0415
505
506 # qsin weights match the renderer's acrossfade=...:c1=qsin:c2=qsin curve
507 schedules_a = {"low": eq_plan.low_out, "mid": eq_plan.mid_out, "high": eq_plan.high_out}
508 schedules_b = {"low": eq_plan.low_in, "mid": eq_plan.mid_in, "high": eq_plan.high_in}
509 running_max = 0.0
510 max_drop_db = 0.0
511 for i in range(n_samples + 1):
512 t = crossfade_duration * i / n_samples
513 if notch[0] <= t <= notch[1]:
514 continue
515 w_a = np.cos(np.pi / 2 * t / crossfade_duration) ** 2 if crossfade_duration else 1.0
516 w_b = np.sin(np.pi / 2 * t / crossfade_duration) ** 2 if crossfade_duration else 0.0
517 p_a = sum(
518 f_a[band]
519 * 10.0
520 ** (_band_gain(schedules_a.get(band), t, cf_start_input, ratio, side="A") / 10.0)
521 for band in BAND_RMS_BANDS
522 )
523 p_b = sum(
524 f_b[band]
525 * 10.0
526 ** (_band_gain(schedules_b.get(band), t, cf_start_input, ratio, side="B") / 10.0)
527 for band in BAND_RMS_BANDS
528 )
529 power = float(w_a * p_a + w_b * p_b)
530 running_max = max(running_max, power)
531 if running_max > 0.0 and power > 0.0:
532 max_drop_db = max(max_drop_db, 10.0 * float(np.log10(running_max / power)))
533 return max_drop_db
534
535 def _choose_fadeout_curve(self, plan: TransitionPlan) -> str:
536 """Pick ``nofade`` when the overlap sits entirely inside a detected mastered fade."""
537 ctx = self._ctx
538 if ctx.fade_onset is None:
539 return "qsin"
540 crossfade_start = plan.fade_out_window - plan.crossfade_duration
541 if crossfade_start < ctx.fade_onset:
542 return "qsin"
543 bar_out = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
544 if plan.fade_out_window < ctx.audio_end - bar_out:
545 return "qsin"
546 # the record already fades itself here; don't double it with a second curve
547 return "nofade"
548
549
550class EmergencyHandoffFactory:
551 """Builds the click-free equal-power SHORT_VOCAL_HANDOFF fallback plan."""
552
553 def __init__(
554 self, ctx: TransitionContext, factory: CandidateFactory, logger: logging.Logger
555 ) -> None:
556 """Initialize the handoff factory for one transition."""
557 self._ctx = ctx
558 self._factory = factory
559 self._logger = logger
560
561 def build(self) -> TransitionPlan:
562 """
563 Build the never-fail click-free equal-power handoff plan.
564
565 Used only when every phrased candidate still collides: keeps the
566 outgoing vocal's protective anchor and shrinks the overlap to the
567 auditioned click-free window, favoring the incoming track's own
568 vocal onset so the handoff needs no EQ to hide anything.
569 """
570 ctx = self._ctx
571 base_spec = CandidateSpec(
572 tier=ctx.tier,
573 bars=1,
574 anchor_s=None,
575 entry_s=ctx.natural_entry,
576 strategy=TransitionStrategy.SHORT_VOCAL_HANDOFF,
577 source="emergency-handoff",
578 ideal_bars=1,
579 )
580 candidate = self._factory.build(base_spec)
581 assert candidate is not None # the 1-bar rung always yields a candidate
582 spec = base_spec
583
584 # old-planner protection semantics: extend the anchor (never past the
585 # RMS-audible boundary) far enough to cover any outgoing vocal the
586 # handoff would cut short, AND to keep a short fade's audible trim
587 # within its own overlap length
588 last_vocal_end = _outgoing_vocal_end(ctx)
589 plan0 = candidate.plan
590 vocal_would_be_cut = last_vocal_end > plan0.fade_out_window + 1e-9
591 overtrims_short_fade = (
592 plan0.crossfade_duration <= SHORT_FADE_SECONDS
593 and ctx.audio_end - plan0.fade_out_window > plan0.crossfade_duration + 1e-9
594 )
595 if vocal_would_be_cut or overtrims_short_fade:
596 target = max(plan0.fade_out_window, last_vocal_end)
597 if overtrims_short_fade:
598 target = max(target, ctx.audio_end - plan0.crossfade_duration)
599 target = min(target, ctx.audio_end)
600 anchor = _nearest_protective_anchor(ctx, target, prefer_earliest=vocal_would_be_cut)
601 reasons = [
602 reason
603 for reason, fired in (
604 ("vocal_would_be_cut", vocal_would_be_cut),
605 ("overtrims_short_fade", overtrims_short_fade),
606 )
607 if fired
608 ]
609 self._logger.debug(
610 "extending emergency handoff anchor (%s): old_anchor=%.2f new_anchor=%.2f",
611 ",".join(reasons),
612 plan0.fade_out_window,
613 anchor,
614 )
615 spec = replace(base_spec, anchor_s=anchor)
616 rebuilt = self._factory.build(spec)
617 assert rebuilt is not None # the 1-bar rung always yields a candidate
618 candidate = rebuilt
619
620 in_mask = ctx.vocal_in_placement
621 incoming_onset = in_mask.windows[0][0] if in_mask is not None and in_mask.windows else 0.0
622 duration = max(MIN_HANDOFF_SECONDS, min(MAX_HANDOFF_SECONDS, incoming_onset - 0.1))
623 handoff = self._as_handoff(candidate.plan, duration)
624 metrics = self._factory.score(spec, handoff)
625 shrunk_to_min = False
626 if (
627 metrics.collision_seconds >= COLLISION_SECONDS_LIMIT
628 or metrics.weighted_collision_seconds >= WEIGHTED_COLLISION_LIMIT
629 ) and duration > MIN_HANDOFF_SECONDS:
630 handoff = self._as_handoff(candidate.plan, MIN_HANDOFF_SECONDS)
631 metrics = self._factory.score(spec, handoff)
632 shrunk_to_min = True
633 self._logger.debug(
634 "emergency handoff: duration=%.2f anchor=%.2f collision=%.2f weighted_collision=%.2f%s",
635 handoff.crossfade_duration,
636 handoff.fade_out_window,
637 metrics.collision_seconds,
638 metrics.weighted_collision_seconds,
639 " (onset-sized window collided; shrunk to the minimum duration)"
640 if shrunk_to_min
641 else "",
642 )
643 return replace(handoff, metrics=metrics)
644
645 @staticmethod
646 def _as_handoff(protected: TransitionPlan, duration: float) -> TransitionPlan:
647 """Return a copy of ``protected`` shrunk to a tempo/EQ-free equal-power handoff."""
648 return replace(
649 protected,
650 crossfade_duration=duration,
651 fadein_trim_start=None,
652 tempo_plan=TempoPlan(),
653 eq_plan=EqPlan.neutral(swap_at=duration / 2.0),
654 )
655
656
657def _band_gain(
658 schedule: ShelfSchedule | None,
659 rendered_t: float,
660 cf_start_input: float,
661 ratio: float,
662 *,
663 side: str,
664) -> float:
665 """Gain (dB) of a band schedule at rendered overlap time; ``None`` is 0dB."""
666 if schedule is None:
667 return 0.0
668 # A-side schedules live in pre-stretch input time; B-side already in rendered time
669 schedule_time = cf_start_input + rendered_t * ratio if side == "A" else rendered_t
670 return schedule.gain_at(schedule_time)
671