/
/
/
1"""
2Smart Fades - candidate value objects and the timed-candidate factory.
3
4A ``CandidateSpec`` is a generator's declared intent (which tier rung, which
5anchor/entry, which generator produced it) before any plan exists; a
6``Candidate`` is that spec paired with its timed ``TransitionPlan`` and
7computed ``PlanMetrics``, ready for policies to score. The factory builds
8TIMED candidates only - anchor, overlap timing, tempo ramp, trims and metrics.
9EQ is deliberately absent: scoring never needs it, so the winner-only
10``PlanAssembler`` applies it after selection.
11
12Every ``build()`` derives its anchored tail fresh from the immutable
13``TransitionContext``, which is what replaces the old planner's
14restore-pristine/re-anchor scratchpad: two builds of the same spec are
15guaranteed to yield identical candidates.
16"""
17
18from __future__ import annotations
19
20from abc import ABC, abstractmethod
21from dataclasses import dataclass, replace
22from typing import TYPE_CHECKING
23
24from music_assistant.constants import VERBOSE_LOG_LEVEL
25from music_assistant.controllers.streams.smart_fades.helpers import (
26 MIN_EFFECTIVE_FADE_BUFFER,
27 SMART_CROSSFADE_DURATION,
28 compute_gradual_tempo_steps,
29 generate_synthetic_timestamps,
30)
31from music_assistant.controllers.streams.smart_fades.models import (
32 FadeOutTrim,
33 PlanMetrics,
34 TempoPlan,
35 TransitionPlan,
36 TransitionStrategy,
37 TransitionTier,
38)
39from music_assistant.controllers.streams.smart_fades.structure import point_in_mask
40from music_assistant.controllers.streams.smart_fades.vocal import (
41 collision_metrics,
42 mask_saturated,
43 merge_windows,
44)
45
46from .context import TIME_STRETCH_BPM_PERCENTAGE_THRESHOLD, choose_tier
47
48if TYPE_CHECKING:
49 import logging
50 from collections.abc import Iterable
51
52 import numpy as np
53 import numpy.typing as npt
54
55 from music_assistant.controllers.streams.smart_fades.models import BandProfile
56
57 from .context import TransitionContext
58
59# Overlap length per tier, in bars of the outgoing grid (research: real DJ
60# transitions cluster at 32 beats = 8 bars; the doubled 16-bar blend is
61# earned only when FireRed shows both decks near-instrumental, where the
62# long exposure carries no vocal-collision risk)
63_FULL_BLEND_BARS: int = 8
64_INSTRUMENTAL_BLEND_BARS: int = 16
65# vocal duty (unpadded mask coverage) at or under this on BOTH decks
66# qualifies as near-instrumental
67_INSTRUMENTAL_DUTY_MAX: float = 0.05
68_TEMPO_BLEND_BARS: int = 8
69# QUICK_FADE bars by BPM incompatibility: (max diff %, bars); beyond -> 1 bar
70_QUICK_FADE_LADDER: tuple[tuple[float, int], ...] = ((12.0, 4), (20.0, 2))
71# phrase-aligned rung set every ladder walks, largest first
72RUNG_LADDER: tuple[int, ...] = (16, 8, 4, 2, 1)
73
74# Deep-trim guard: a bar is protected when its low band is silent but its
75# voice/melody bands are active (cutting there beheads a sung intro)
76_TRIM_GUARD_LOW_FLOOR: float = 0.25
77_TRIM_GUARD_VOICE_FLOOR: float = 0.4
78
79# Trim-closing anchors engage only when the audible end sits this much past
80# the energy anchor; below it the default anchor leads the main pass (the
81# rescue pass re-runs this generator ungated when that pass rejects everything)
82_TRIM_CLOSING_MIN_GAP_S: float = 8.0
83
84# Lazy-overlay length: a long unphrased equal-power blend, not a rung on any ladder
85_LAZY_OVERLAY_SECONDS: float = 16.0
86# both decks at or under this in-window vocal duty qualify as ambient
87_LAZY_DUTY_MAX: float = 0.10
88
89
90@dataclass(frozen=True, slots=True)
91class CandidateSpec:
92 """A candidate's declared shape: tier rung, anchor/entry choice, and generator provenance."""
93
94 tier: TransitionTier
95 bars: int
96 # buffer-local; None = pristine audible end
97 anchor_s: float | None
98 # None = natural entry
99 entry_s: float | None
100 strategy: TransitionStrategy = TransitionStrategy.ENERGY_ALIGNED
101 # generator name, for scoreboard + tie-break
102 source: str = ""
103 # the tier ladder's top rung; 0 = same as bars
104 ideal_bars: int = 0
105
106
107@dataclass(frozen=True, slots=True)
108class Candidate:
109 """One fully-built candidate: its spec, timed plan, and computed metrics."""
110
111 spec: CandidateSpec
112 # timed plan, eq_plan neutral until assembly
113 plan: TransitionPlan
114 metrics: PlanMetrics
115 # the tier ladder's top rung for this context
116 ideal_bars: int
117
118
119def earns_instrumental_blend(ctx: TransitionContext) -> bool:
120 """Whether verified near-instrumental decks earn the doubled full-blend overlap."""
121 duties = _vocal_duties(ctx)
122 if duties is None:
123 return False
124 out_duty, in_duty = duties
125 return out_duty <= _INSTRUMENTAL_DUTY_MAX and in_duty <= _INSTRUMENTAL_DUTY_MAX
126
127
128def bars_ladder(ctx: TransitionContext, tier: TransitionTier) -> list[int]:
129 """Candidate bar counts to try for a tier, largest first (shorter rungs fit smaller buffers)."""
130 if tier is TransitionTier.QUICK_FADE:
131 # a mismatched meter has no shared bar grid to blend across; cap short
132 # regardless of how close the tempos happen to be
133 ladder = ((0.0, 2),) if ctx.cross_meter else _QUICK_FADE_LADDER
134 ideal = next((bars for limit, bars in ladder if ctx.bpm_diff_percent <= limit), 1)
135 elif tier is TransitionTier.TEMPO_BLEND:
136 ideal = _TEMPO_BLEND_BARS
137 elif earns_instrumental_blend(ctx):
138 ideal = _INSTRUMENTAL_BLEND_BARS
139 else:
140 ideal = _FULL_BLEND_BARS
141 return [bars for bars in RUNG_LADDER if bars <= ideal]
142
143
144class CandidateGenerator(ABC):
145 """One source of candidate specs; generators only emit, the factory validates feasibility."""
146
147 name: str
148
149 @abstractmethod
150 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
151 """Emit this generator's candidate specs for one transition, best first."""
152
153
154class EnergyLadderGenerator(CandidateGenerator):
155 """Emits the tier's bar-count ladder at each viable anchor, across entry options."""
156
157 name = "energy-ladder"
158
159 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
160 """Emit one spec per (rung, entry option) at the default, full-band and pinned anchors."""
161 ladder = bars_ladder(ctx, ctx.tier)
162 ideal = ladder[0]
163 # the default anchor is already kick-folded; a pure full-band variant
164 # would let a longer blend win past the kick die-out, defeating the
165 # researched kick handover, so it is deliberately not emitted (trim-closing
166 # may still anchor later when >=8s of audible tail would otherwise be stranded)
167 anchors: list[float | None] = [None]
168 a_pin = _fade_onset_pin(ctx)
169 if a_pin < ctx.default_anchor:
170 anchors.append(a_pin)
171 for anchor in anchors:
172 for bars in ladder:
173 # the default anchor keeps the factory-chosen entry (grid
174 # alignment + rolling intro), exactly like the old energy
175 # candidate; explicit entry options are pin-scoped, like the
176 # old remediation rungs they came from
177 entries = [None] if anchor is None else _entry_options(ctx, bars)
178 for entry in entries:
179 yield CandidateSpec(
180 tier=ctx.tier,
181 bars=bars,
182 anchor_s=anchor,
183 entry_s=entry,
184 source=self.name,
185 ideal_bars=ideal,
186 )
187
188
189class CodaAnchorGenerator(CandidateGenerator):
190 """Emits candidates anchored inside the outgoing track's validated coda/outro zone."""
191
192 name = "coda-anchor"
193
194 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
195 """Emit the zone's fitting rung(s) at the coda anchor, or nothing without a valid zone."""
196 # coda shifting is a vocal-collision remediation: without a vocal
197 # timeline the zone was validated against an empty mask and the old
198 # planner never coda-shifted, so stay inactive on the energy-only path
199 if ctx.vocal_out_placement is None:
200 return
201 if ctx.coda_zone is None or ctx.outgoing_profile is None:
202 return
203 zone = ctx.coda_zone
204 bar_starts = ctx.outgoing_profile.bar_starts
205 in_zone = bar_starts[(bar_starts >= zone.start_s) & (bar_starts <= zone.end_s)]
206 if not len(in_zone):
207 return
208 anchor = float(in_zone[-1]) - ctx.buffer_offset
209 if anchor < MIN_EFFECTIVE_FADE_BUFFER:
210 return
211 bar_a = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
212 zone_bars = int((zone.end_s - zone.start_s) / bar_a)
213 top_rung = next((n for n in (16, 8, 4, 2) if n <= zone_bars), None)
214 if top_rung is None:
215 return
216 for bars in dict.fromkeys((top_rung, 2)):
217 for entry in _entry_options(ctx, bars):
218 yield CandidateSpec(
219 tier=ctx.tier,
220 bars=bars,
221 anchor_s=anchor,
222 entry_s=entry,
223 source=self.name,
224 ideal_bars=top_rung,
225 )
226
227
228class ProtectiveAnchorGenerator(CandidateGenerator):
229 """Emits the tier's ladder anchored to keep A's last outgoing vocal phrase intact."""
230
231 name = "protective-anchor"
232
233 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
234 """Emit every ladder rung at the nearest anchor that doesn't truncate A's last phrase."""
235 if ctx.vocal_out_placement is None or not ctx.vocal_out_placement.windows:
236 return
237 target = _outgoing_vocal_end(ctx)
238 anchor = _nearest_protective_anchor(ctx, target)
239 ladder = bars_ladder(ctx, ctx.tier)
240 ideal = ladder[0]
241 bar_seconds = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
242 for bars in ladder:
243 # the old protection had two re-anchor triggers: cover the last
244 # vocal phrase, and close a short fade's audible-trim gap; emit
245 # both anchors so each stays reachable as a candidate
246 anchors = [anchor]
247 trim_target = ctx.audio_end - bars * bar_seconds
248 if trim_target > anchor:
249 trim_anchor = _nearest_protective_anchor(ctx, trim_target, prefer_earliest=False)
250 if trim_anchor not in anchors:
251 anchors.append(trim_anchor)
252 # the factory-chosen (beat-aligned) entry leads, as the old
253 # protection rebuild pinned it; explicit groove/natural options
254 # follow so a remediated entry stays reachable too
255 for rung_anchor in anchors:
256 for entry in [None, *_entry_options(ctx, bars)]:
257 yield CandidateSpec(
258 tier=ctx.tier,
259 bars=bars,
260 anchor_s=rung_anchor,
261 entry_s=entry,
262 source=self.name,
263 ideal_bars=ideal,
264 )
265
266
267class VocalOnsetEntryGenerator(CandidateGenerator):
268 """Emits an entry that lands B's first vocal onset exactly at the overlap end."""
269
270 name = "vocal-onset-entry"
271
272 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
273 """Emit the vocal-onset-aligned entry at the tier's ideal rung, or nothing when illegal."""
274 if ctx.vocal_in_placement is None or not ctx.vocal_in_placement.windows:
275 return
276 if mask_saturated(ctx.vocal_in_placement, float(SMART_CROSSFADE_DURATION)):
277 return
278 ideal = bars_ladder(ctx, ctx.tier)[0]
279 bar_b = ctx.incoming.beats_per_bar * 60.0 / ctx.incoming.bpm
280 entry = ctx.vocal_in_placement.windows[0][0] - ideal * bar_b
281 if entry < 0.0 or point_in_mask(ctx.vocal_in_placement, entry):
282 return
283 a_pin = _fade_onset_pin(ctx)
284 anchors: list[float | None] = [a_pin if a_pin < ctx.default_anchor else None]
285 # the old planner's protection rebuild preserved a remediated entry at
286 # the protective anchor; emitting the combination keeps that reachable
287 if ctx.vocal_out_placement is not None and ctx.vocal_out_placement.windows:
288 protective = _nearest_protective_anchor(
289 ctx, min(ctx.vocal_out_placement.last_end(), ctx.audio_end)
290 )
291 if protective not in anchors and protective != ctx.default_anchor:
292 anchors.append(protective)
293 for anchor in anchors:
294 yield CandidateSpec(
295 tier=ctx.tier,
296 bars=ideal,
297 anchor_s=anchor,
298 entry_s=entry,
299 source=self.name,
300 ideal_bars=ideal,
301 )
302
303
304class RescueAnchorGenerator(CandidateGenerator):
305 """Emits a modest, late-anchored rung as a last resort before the emergency handoff."""
306
307 name = "rescue-anchor"
308
309 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
310 """Emit 1-2 bar rungs anchored as late as the tail allows, never past A's own vocal end."""
311 full_ladder = bars_ladder(ctx, ctx.tier)
312 bar_seconds = ctx.outgoing.beats_per_bar * 60.0 / ctx.outgoing.bpm
313 last_vocal_end = _outgoing_vocal_end(ctx)
314 for bars in [rung for rung in full_ladder if rung <= 2]:
315 target = min(ctx.audio_end, max(ctx.audio_end - bars * bar_seconds, last_vocal_end))
316 anchor = _nearest_protective_anchor(ctx, target, prefer_earliest=False)
317 for entry in [None, *_entry_options(ctx, bars)]:
318 yield CandidateSpec(
319 tier=ctx.tier,
320 bars=bars,
321 anchor_s=anchor,
322 entry_s=entry,
323 source=self.name,
324 ideal_bars=full_ladder[0],
325 )
326
327
328class TrimClosingAnchorGenerator(CandidateGenerator):
329 """Emits the tier's ladder at that anchor, at the audible end, when a large tail is stranded."""
330
331 name = "trim-closing-anchor"
332
333 def __init__(self, min_gap: float = _TRIM_CLOSING_MIN_GAP_S) -> None:
334 """Initialize the generator with the smallest stranded-tail gap that engages it."""
335 self._min_gap = min_gap
336
337 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
338 """Emit every ladder rung at the audible end, or nothing when the trim gap is small."""
339 anchor = ctx.audio_end
340 if anchor - ctx.default_anchor < self._min_gap:
341 return
342 # ctx.tier is decided at the early anchor; the grid can be blendable at the audible end
343 _, tier = choose_tier(ctx.outgoing, ctx.incoming, anchor)
344 ladder = bars_ladder(ctx, tier)
345 for bars in ladder:
346 yield CandidateSpec(
347 tier=tier,
348 bars=bars,
349 anchor_s=anchor,
350 entry_s=None,
351 source=self.name,
352 ideal_bars=ladder[0],
353 )
354
355
356class LazyOverlayGenerator(CandidateGenerator):
357 """Emits one long unphrased overlay when the grid is unusable but both decks are ambient."""
358
359 name = "lazy-overlay"
360
361 def generate(self, ctx: TransitionContext) -> Iterable[CandidateSpec]:
362 """Emit the overlay spec, or nothing when the pair doesn't qualify."""
363 if ctx.tier is not TransitionTier.QUICK_FADE or ctx.cross_meter:
364 return
365 if ctx.bpm_diff_percent > TIME_STRETCH_BPM_PERCENTAGE_THRESHOLD:
366 return
367 duties = _window_duties(ctx, _LAZY_OVERLAY_SECONDS)
368 if duties is None or duties[0] > _LAZY_DUTY_MAX or duties[1] > _LAZY_DUTY_MAX:
369 return
370 yield CandidateSpec(
371 tier=ctx.tier,
372 bars=1,
373 anchor_s=ctx.audio_end,
374 entry_s=None,
375 strategy=TransitionStrategy.LAZY_OVERLAY,
376 source=self.name,
377 ideal_bars=1,
378 )
379
380
381def default_generators() -> tuple[CandidateGenerator, ...]:
382 """Return the standard generator set, in preference order (best first)."""
383 return (
384 EnergyLadderGenerator(),
385 CodaAnchorGenerator(),
386 ProtectiveAnchorGenerator(),
387 VocalOnsetEntryGenerator(),
388 LazyOverlayGenerator(),
389 TrimClosingAnchorGenerator(),
390 )
391
392
393class CandidateFactory:
394 """Builds timed candidates from specs, purely over the transition context."""
395
396 def __init__(self, ctx: TransitionContext, logger: logging.Logger) -> None:
397 """Initialize the factory for one transition."""
398 self._ctx = ctx
399 self._logger = logger
400
401 def build(self, spec: CandidateSpec) -> Candidate | None:
402 """
403 Build one complete timed candidate for a spec, or ``None`` when it is infeasible.
404
405 Every timing, tempo and trim decision is derived fresh from the
406 context and the spec's anchor - a candidate never inherits state from
407 a previously built one. Infeasible means the spec's bar count needs
408 more room than the incoming buffer has, or its entry leaves no legal
409 alignment; a 1-bar spec never fails this way, matching the plan floor.
410 The returned candidate's spec reflects what was actually built: a
411 re-anchored tail can downgrade the tier and cap the bar count.
412 """
413 if spec.strategy is TransitionStrategy.LAZY_OVERLAY:
414 return self._build_lazy_overlay(spec)
415 tail = self._anchored_tail(spec.anchor_s)
416 # a re-anchored tail can downgrade the tier (shorter/irregular grid); the
417 # requested bar count still reflects the old tier, so cap it at the new
418 # tier's largest rung or a long overlap ships without its tempo ramp
419 _, tier = choose_tier(self._ctx.outgoing, self._ctx.incoming, tail.effective_end)
420 bars_cap = bars_ladder(self._ctx, tier)[0]
421 bars = min(spec.bars, bars_cap)
422
423 fadein_start_pos = (
424 spec.entry_s if spec.entry_s is not None else self._choose_fadein_entry(tail, bars)
425 )
426 if bars > 1 and fadein_start_pos is None:
427 self._logger.log(
428 VERBOSE_LOG_LEVEL,
429 "dropping spec source=%s tier=%s bars=%d anchor=%s: no beat-aligned incoming entry",
430 spec.source,
431 tier,
432 bars,
433 spec.anchor_s,
434 )
435 return None
436 crossfade_duration = self._calculate_crossfade_duration(tail, bars)
437
438 tempo_plan = self._choose_tempo_ramp(tier, tail, crossfade_duration)
439 crossfade_duration, fadein_trim_start = self._lock_in_timing(
440 tail, crossfade_duration, fadein_start_pos, tempo_plan
441 )
442 if bars > 1 and fadein_start_pos is not None and fadein_trim_start is None:
443 self._logger.log(
444 VERBOSE_LOG_LEVEL,
445 "dropping spec source=%s tier=%s bars=%d anchor=%s: "
446 "no legal timing lock for the pinned entry",
447 spec.source,
448 tier,
449 bars,
450 spec.anchor_s,
451 )
452 return None
453 # Rolling-intro alignment: on a full blend with no pinned entry, deepen B's
454 # trim so its groove entry lands at the overlap END (B's intro runs under A,
455 # its drop hits where A's music dies). A sung run that no legal cut clears
456 # is infeasible so the caller's ladder drops to a shorter overlap instead —
457 # except at the 1-bar floor, which must always yield a candidate: there the
458 # un-deepened trim ships as-is.
459 if tier is TransitionTier.FULL_BLEND and spec.entry_s is None:
460 feasible, aligned = self._align_rolling_intro(crossfade_duration, fadein_trim_start)
461 if not feasible:
462 if bars > 1:
463 self._logger.log(
464 VERBOSE_LOG_LEVEL,
465 "dropping spec source=%s tier=%s bars=%d anchor=%s: "
466 "rolling intro: no legal cut clears the sung run",
467 spec.source,
468 tier,
469 bars,
470 spec.anchor_s,
471 )
472 return None
473 else:
474 fadein_trim_start = aligned
475
476 plan = TransitionPlan(
477 tier=tier,
478 fade_out_window=tail.effective_end,
479 crossfade_duration=crossfade_duration,
480 tempo_plan=tempo_plan,
481 fadeout_trim=tail.fadeout_trim,
482 fadein_trim_start=fadein_trim_start,
483 )
484 built_spec = replace(spec, tier=tier, bars=bars)
485 return Candidate(
486 spec=built_spec,
487 plan=plan,
488 metrics=self._score(built_spec, plan),
489 ideal_bars=spec.ideal_bars or spec.bars,
490 )
491
492 def score(self, spec: CandidateSpec, plan: TransitionPlan) -> PlanMetrics:
493 """Score an arbitrary (spec, plan) pair against this context, for a plan edited post-build."""
494 return self._score(spec, plan)
495
496 @property
497 def _bpm_ratio(self) -> float:
498 """Tempo ratio between the incoming and outgoing track."""
499 return self._ctx.incoming.bpm / self._ctx.outgoing.bpm
500
501 def _anchored_tail(self, anchor_s: float | None) -> _AnchoredTail:
502 """Derive the tail state for an anchor, never later than the RMS-audible boundary."""
503 import numpy as np # noqa: PLC0415
504
505 ctx = self._ctx
506 anchor = anchor_s if anchor_s is not None else ctx.default_anchor
507 effective_end = min(anchor, ctx.audio_end)
508 # same sub-half-second slack rule as the tail cue: the rendered stream
509 # still ends at the buffer end, so the anchor must follow it
510 fadeout_trim: FadeOutTrim | None
511 if effective_end >= ctx.buffer_duration - 0.5:
512 effective_end = ctx.buffer_duration
513 fadeout_trim = None
514 else:
515 fadeout_trim = FadeOutTrim(
516 end_pos=effective_end,
517 trimmed_seconds=ctx.buffer_duration - effective_end,
518 )
519 protective = np.asarray(ctx.protective_downbeats, dtype=np.float32)
520 return _AnchoredTail(
521 effective_end=effective_end,
522 fadeout_trim=fadeout_trim,
523 beats=ctx.outgoing.beats[ctx.outgoing.beats <= effective_end],
524 downbeats=ctx.outgoing.downbeats[ctx.outgoing.downbeats <= effective_end],
525 # protective downbeats reach all the way to audio_end, so they cover
526 # any position an anchor could have chosen
527 extrapolated_downbeats=protective[protective <= effective_end],
528 )
529
530 def _choose_fadein_entry(self, tail: _AnchoredTail, crossfade_bars: int) -> float | None:
531 """Choose where the incoming track enters, aligned to its beat grid."""
532
533 def calculate_beat_positions(
534 fade_out_beats: npt.NDArray[np.float32],
535 fade_in_beats: npt.NDArray[np.float32],
536 num_beats: int,
537 ) -> float | None:
538 """Calculate start positions from beat arrays."""
539 if len(fade_out_beats) < num_beats or len(fade_in_beats) < num_beats:
540 return None
541
542 fade_in_slice = fade_in_beats[:num_beats]
543 return float(fade_in_slice[0])
544
545 # Try downbeats first for most musical timing
546 downbeat_positions = calculate_beat_positions(
547 tail.extrapolated_downbeats, self._ctx.incoming.downbeats, crossfade_bars
548 )
549 if downbeat_positions is not None:
550 return downbeat_positions
551
552 # Try regular beats if downbeats insufficient
553 required_beats = crossfade_bars * self._ctx.incoming.beats_per_bar
554 beat_positions = calculate_beat_positions(
555 tail.beats, self._ctx.incoming.beats, required_beats
556 )
557 if beat_positions is not None:
558 return beat_positions
559
560 # Fallback: No beat alignment possible
561 self._logger.log(VERBOSE_LOG_LEVEL, "No beat alignment possible (insufficient beats)")
562 return None
563
564 def _calculate_crossfade_duration(self, tail: _AnchoredTail, crossfade_bars: int) -> float:
565 """Calculate the crossfade duration for a bar count, capped to the audible tail."""
566 downbeats = tail.downbeats
567 bar_seconds = self._ctx.outgoing.beats_per_bar * 60.0 / self._ctx.outgoing.bpm
568 # the downbeat span assumes the anchor sits (near) the last downbeat;
569 # when the grid dies early (unsnapped anchor), the anchor gap would be
570 # added to EVERY rung — a "1-bar" quick fade could span the whole gap
571 if (
572 len(downbeats) > crossfade_bars
573 and tail.effective_end - float(downbeats[-1]) < bar_seconds
574 ):
575 # the real span between downbeats honors the track's own tempo/rubato
576 # more precisely than a constant-BPM estimate
577 musical_duration = float(
578 tail.effective_end - downbeats[len(downbeats) - 1 - crossfade_bars]
579 )
580 else:
581 seconds_per_beat = 60.0 / self._ctx.incoming.bpm
582 musical_duration = crossfade_bars * self._ctx.incoming.beats_per_bar * seconds_per_beat
583
584 # Cap at the audible fade-out room so crossfade_start never goes negative
585 # downstream (effective_end <= SMART_CROSSFADE_DURATION always)
586 actual_duration = min(musical_duration, tail.effective_end)
587
588 if musical_duration > actual_duration:
589 self._logger.log(
590 VERBOSE_LOG_LEVEL,
591 "Constraining crossfade duration from %.1fs to %.1fs (audible tail limit)",
592 musical_duration,
593 actual_duration,
594 )
595
596 return actual_duration
597
598 def _choose_tempo_ramp(
599 self, tier: TransitionTier, tail: _AnchoredTail, crossfade_duration: float
600 ) -> TempoPlan:
601 """Choose the gradual tempo ramp that beatmatches the outgoing track, if any."""
602 if tier is TransitionTier.QUICK_FADE:
603 return TempoPlan()
604 if not 0.1 < self._ctx.bpm_diff_percent <= TIME_STRETCH_BPM_PERCENTAGE_THRESHOLD:
605 return TempoPlan()
606 return TempoPlan(steps=self._compute_tempo_steps(tail, crossfade_duration))
607
608 def _compute_tempo_steps(
609 self, tail: _AnchoredTail, crossfade_duration: float
610 ) -> list[tuple[float, float]]:
611 """Compute the gradual tempo ramp in the 10s window before the crossfade."""
612 stretch_duration = 10.0
613 crossfade_start = tail.effective_end - crossfade_duration
614 # A crossfade consuming the whole audible tail leaves no room for a
615 # pre-fade tempo ramp
616 if crossfade_start <= 0:
617 return []
618 stretch_start = max(0.0, crossfade_start - stretch_duration)
619 stretch_end = crossfade_start
620
621 # Collect timing points within the stretch window
622 beats = tail.beats
623 beat_mask = (beats >= stretch_start) & (beats <= stretch_end)
624 db_mask = (tail.extrapolated_downbeats >= stretch_start) & (
625 tail.extrapolated_downbeats <= stretch_end
626 )
627 window_beats = beats[beat_mask] - stretch_start
628 window_downbeats = tail.extrapolated_downbeats[db_mask] - stretch_start
629
630 # >3% BPM diff: beat-level stepping (more steps = smoother)
631 # <=3%: downbeat-level stepping, fall back to beats if too few
632 if self._ctx.bpm_diff_percent > 3.0:
633 stretch_timestamps = window_beats
634 elif len(window_downbeats) >= 2:
635 stretch_timestamps = window_downbeats
636 else:
637 stretch_timestamps = window_beats
638
639 # Fall back to synthetic timestamps when < 2 real timestamps
640 if len(stretch_timestamps) < 2:
641 stretch_timestamps = generate_synthetic_timestamps(
642 stretch_end - stretch_start,
643 self._ctx.outgoing.bpm,
644 beats_per_bar=self._ctx.outgoing.beats_per_bar,
645 )
646
647 tempo_steps = compute_gradual_tempo_steps(
648 start_ratio=1.0,
649 end_ratio=self._bpm_ratio,
650 downbeats=stretch_timestamps,
651 )
652 if not tempo_steps:
653 tempo_steps = [(0.0, self._bpm_ratio)]
654
655 # Shift timestamps back to buffer-relative coordinates for FFmpeg
656 return [(ts + stretch_start, ratio) for ts, ratio in tempo_steps]
657
658 def _lock_in_timing(
659 self,
660 tail: _AnchoredTail,
661 crossfade_duration: float,
662 fadein_start_pos: float | None,
663 tempo_plan: TempoPlan,
664 ) -> tuple[float, float | None]:
665 """
666 Lock the overlap timing: confirm the fade-in entry and snap to downbeats.
667
668 Returns the final crossfade duration (downbeat-snapped and compensated
669 for time-stretch compression) and the fade-in trim position, or ``None``
670 when beat alignment is skipped.
671
672 :param tail: The anchored tail the candidate is built on.
673 :param crossfade_duration: Draft crossfade duration in seconds.
674 :param fadein_start_pos: Chosen entry point in the incoming track, if any.
675 :param tempo_plan: The tempo ramp chosen for this transition.
676 """
677 # Adjust crossfade duration to align with outgoing track's downbeats.
678 # When stretching, only consider downbeats after the stretch window
679 # to ensure the outgoing track has reached the target tempo.
680 crossfade_start = tail.effective_end - crossfade_duration
681 crossfade_duration = self._adjust_crossfade_to_downbeats(
682 tail,
683 crossfade_duration=crossfade_duration,
684 fadein_start_pos=fadein_start_pos,
685 min_downbeat_pos=crossfade_start if tempo_plan else 0.0,
686 render_ratio=self._bpm_ratio if tempo_plan else 1.0,
687 )
688
689 # Compensate crossfade duration for time-stretch compression.
690 # Gate on the tempo plan (not stretch eligibility) so a guard-skipped
691 # stretch doesn't apply a compensation for a stretch that never ran.
692 if tempo_plan:
693 crossfade_duration = crossfade_duration / self._bpm_ratio
694
695 fadein_trim_start: float | None = None
696 if (
697 fadein_start_pos is not None
698 and fadein_start_pos + crossfade_duration <= SMART_CROSSFADE_DURATION
699 ):
700 fadein_trim_start = fadein_start_pos
701 else:
702 self._logger.log(
703 VERBOSE_LOG_LEVEL,
704 "Skipping beat alignment: not enough audio after trim (%s + %.1fs > %.1fs)",
705 fadein_start_pos,
706 crossfade_duration,
707 SMART_CROSSFADE_DURATION,
708 )
709
710 return crossfade_duration, fadein_trim_start
711
712 def _adjust_crossfade_to_downbeats(
713 self,
714 tail: _AnchoredTail,
715 crossfade_duration: float,
716 fadein_start_pos: float | None,
717 min_downbeat_pos: float = 0.0,
718 render_ratio: float = 1.0,
719 ) -> float:
720 """Adjust crossfade duration to align with outgoing track's downbeats."""
721 # If we don't have downbeats or beat alignment is disabled, return original duration
722 if len(tail.extrapolated_downbeats) == 0 or fadein_start_pos is None:
723 return crossfade_duration
724
725 # Calculate where the crossfade would start in the buffer
726 ideal_start_pos = tail.effective_end - crossfade_duration
727
728 self._logger.log(
729 VERBOSE_LOG_LEVEL,
730 "Downbeat adjustment - ideal_start=%.2fs (effective_end=%.1fs - crossfade=%.2fs), "
731 "fadein_start=%.2fs",
732 ideal_start_pos,
733 tail.effective_end,
734 crossfade_duration,
735 fadein_start_pos,
736 )
737
738 # Find the closest downbeats (earlier and later)
739 earlier_downbeat = None
740 later_downbeat = None
741
742 for downbeat in tail.extrapolated_downbeats:
743 if downbeat < min_downbeat_pos:
744 continue
745 if downbeat <= ideal_start_pos:
746 earlier_downbeat = downbeat
747 elif downbeat > ideal_start_pos and later_downbeat is None:
748 later_downbeat = downbeat
749 break
750
751 # Try earlier downbeat first (longer crossfade)
752 if earlier_downbeat is not None:
753 adjusted_duration = float(tail.effective_end - earlier_downbeat)
754 if fadein_start_pos + adjusted_duration / render_ratio <= SMART_CROSSFADE_DURATION:
755 if abs(adjusted_duration - crossfade_duration) > 0.1:
756 self._logger.log(
757 VERBOSE_LOG_LEVEL,
758 "Adjusted crossfade duration from %.2fs to %.2fs to align with "
759 "downbeat at %.2fs (earlier)",
760 crossfade_duration,
761 adjusted_duration,
762 earlier_downbeat,
763 )
764 return adjusted_duration
765
766 # Try later downbeat (shorter crossfade)
767 if later_downbeat is not None:
768 adjusted_duration = float(tail.effective_end - later_downbeat)
769 if fadein_start_pos + adjusted_duration / render_ratio <= SMART_CROSSFADE_DURATION:
770 if abs(adjusted_duration - crossfade_duration) > 0.1:
771 self._logger.log(
772 VERBOSE_LOG_LEVEL,
773 "Adjusted crossfade duration from %.2fs to %.2fs to align with "
774 "downbeat at %.2fs (later)",
775 crossfade_duration,
776 adjusted_duration,
777 later_downbeat,
778 )
779 return adjusted_duration
780
781 # If no suitable downbeat found, return original duration
782 self._logger.log(
783 VERBOSE_LOG_LEVEL,
784 "Could not adjust crossfade duration to downbeats, using original %.2fs",
785 crossfade_duration,
786 )
787 return crossfade_duration
788
789 def _align_rolling_intro(
790 self, crossfade_duration: float, fadein_trim_start: float | None
791 ) -> tuple[bool, float | None]:
792 """
793 Deepen B's trim when the overlap can't cover its intro (else B's groove lands on dead air).
794
795 Returns ``(feasible, trim)``: ``(False, None)`` when a sung run leaves
796 no legal cut anywhere in the buffer, else the (possibly deepened) trim.
797 """
798 import numpy as np # noqa: PLC0415
799
800 entry = self._ctx.natural_entry
801 trim = fadein_trim_start or 0.0
802 if entry <= 0.0 or entry - trim <= crossfade_duration:
803 return True, fadein_trim_start
804 if entry > SMART_CROSSFADE_DURATION:
805 # groove enters beyond the buffered head — unreachable defensively
806 return True, fadein_trim_start
807 deep_trim = entry - crossfade_duration
808 downbeats = self._ctx.incoming.downbeats
809 if len(downbeats):
810 deep_trim = float(downbeats[np.argmin(np.abs(downbeats - deep_trim))])
811 guarded = self._guard_deep_trim(deep_trim, crossfade_duration)
812 if guarded is None:
813 self._logger.log(
814 VERBOSE_LOG_LEVEL,
815 "Rolling intro: no legal cut clears the sung run within the buffer; "
816 "dropping to a shorter overlap instead",
817 )
818 return False, None
819 self._logger.log(
820 VERBOSE_LOG_LEVEL,
821 "Rolling intro: trimming %.1fs of pre-groove intro to keep the handover anchored",
822 guarded,
823 )
824 return True, guarded
825
826 def _guard_deep_trim(self, deep_trim: float, crossfade_duration: float) -> float | None:
827 """Push a deep-trim cut off a protected run; ``None`` when no legal cut fits the buffer."""
828 import numpy as np # noqa: PLC0415
829
830 # the cut is a minimum, so only search later: first unprotected downbeat at/after wins
831 if self._ctx.incoming_profile is None:
832 return deep_trim
833 bar_starts = self._ctx.incoming_profile.bar_starts
834 protected = self._protected_bars(self._ctx.incoming_profile)
835 start_idx = int(np.searchsorted(bar_starts, deep_trim - 1e-6))
836 if start_idx >= len(bar_starts) or not protected[start_idx]:
837 return deep_trim
838 for i in range(start_idx, len(protected)):
839 if protected[i]:
840 continue
841 candidate = float(bar_starts[i])
842 if candidate + crossfade_duration <= SMART_CROSSFADE_DURATION:
843 return candidate
844 break
845 return None
846
847 @staticmethod
848 def _protected_bars(profile: BandProfile) -> npt.NDArray[np.bool_]:
849 """Mark each bar of ``profile`` as protected: low-silent but voice/melody-active."""
850 low = profile.bar_power["low"]
851 low_mid = profile.bar_power["low_mid"]
852 mid = profile.bar_power["mid"]
853 low_silent = low < _TRIM_GUARD_LOW_FLOOR * profile.reference["low"]
854 voice_active = (low_mid >= _TRIM_GUARD_VOICE_FLOOR * profile.reference["low_mid"]) | (
855 mid >= _TRIM_GUARD_VOICE_FLOOR * profile.reference["mid"]
856 )
857 return low_silent & voice_active
858
859 def _build_lazy_overlay(self, spec: CandidateSpec) -> Candidate:
860 """Build the unphrased long-overlay candidate: anchored at the audible end, no alignment."""
861 tail = self._anchored_tail(spec.anchor_s)
862 plan = TransitionPlan(
863 tier=spec.tier,
864 fade_out_window=tail.effective_end,
865 crossfade_duration=min(_LAZY_OVERLAY_SECONDS, tail.effective_end),
866 tempo_plan=TempoPlan(),
867 fadeout_trim=tail.fadeout_trim,
868 fadein_trim_start=None,
869 )
870 return Candidate(
871 spec=spec, plan=plan, metrics=self._score(spec, plan), ideal_bars=spec.ideal_bars
872 )
873
874 def _score(self, spec: CandidateSpec, plan: TransitionPlan) -> PlanMetrics:
875 """Score a candidate: trims, retained vocal time, downbeat alignment, collision."""
876 ctx = self._ctx
877 audible_outgoing_trim = max(0.0, ctx.audio_end - plan.fade_out_window)
878 anchor_on_downbeat = self._is_on_downbeat(plan.fade_out_window)
879 # deliberate extension over the old planner (which never scored the
880 # energy-only path): policies need real trim/downbeat facts on every
881 # candidate; each vocal-dependent field needs only its own deck's mask
882 outgoing_vocal_fade_seconds = 0.0
883 collision_seconds = weighted_collision = 0.0
884 if ctx.vocal_out_scoring is not None:
885 outgoing_windows = self._rendered_outgoing_windows(plan)
886 in_fade = [
887 (max(0.0, left), min(plan.crossfade_duration, right))
888 for left, right in outgoing_windows
889 if right > 0.0 and left < plan.crossfade_duration
890 ]
891 outgoing_vocal_fade_seconds = sum(
892 right - left for left, right in merge_windows(in_fade)
893 )
894 if ctx.vocal_in_scoring is not None:
895 collision_seconds, weighted_collision = collision_metrics(
896 outgoing_windows,
897 self._rendered_incoming_windows(plan),
898 plan.crossfade_duration,
899 )
900 return PlanMetrics(
901 strategy=spec.strategy,
902 audible_outgoing_trim=audible_outgoing_trim,
903 outgoing_vocal_fade_seconds=outgoing_vocal_fade_seconds,
904 anchor_on_downbeat=anchor_on_downbeat,
905 collision_seconds=collision_seconds,
906 weighted_collision_seconds=weighted_collision,
907 )
908
909 def _rendered_outgoing_windows(self, plan: TransitionPlan) -> list[tuple[float, float]]:
910 """Map the outgoing (unpadded) vocal scoring mask into rendered crossfade-local seconds."""
911 assert self._ctx.vocal_out_scoring is not None # narrowed by the caller
912 rendered_anchor = self._rendered_time(plan, plan.fade_out_window)
913 rendered_start = rendered_anchor - plan.crossfade_duration
914 return [
915 (
916 self._rendered_time(plan, left) - rendered_start,
917 self._rendered_time(plan, right) - rendered_start,
918 )
919 for left, right in self._ctx.vocal_out_scoring.windows
920 ]
921
922 def _rendered_incoming_windows(self, plan: TransitionPlan) -> list[tuple[float, float]]:
923 """Map the incoming (unpadded) scoring mask into the plan's fadein-trim-relative seconds."""
924 assert self._ctx.vocal_in_scoring is not None # narrowed by the caller
925 trim = plan.fadein_trim_start or 0.0
926 return [(left - trim, right - trim) for left, right in self._ctx.vocal_in_scoring.windows]
927
928 @staticmethod
929 def _rendered_time(plan: TransitionPlan, input_time: float) -> float:
930 """Map a buffer-local outgoing input-time position to its rendered-stream position."""
931 clamped = max(0.0, min(input_time, plan.fade_out_window))
932 return clamped - plan.tempo_plan.savings_until(clamped)
933
934 def _is_on_downbeat(self, position: float, tolerance: float = 0.05) -> bool:
935 """Whether a buffer-local position sits within tolerance of an outgoing downbeat."""
936 import numpy as np # noqa: PLC0415
937
938 downbeats = np.asarray(self._ctx.protective_downbeats, dtype=np.float32)
939 return bool(len(downbeats) and np.min(np.abs(downbeats - position)) <= tolerance)
940
941
942@dataclass(frozen=True, slots=True)
943class _AnchoredTail:
944 """The outgoing tail derived for one anchor: masked grids, trim, and effective end."""
945
946 effective_end: float
947 fadeout_trim: FadeOutTrim | None
948 beats: npt.NDArray[np.float32]
949 downbeats: npt.NDArray[np.float32]
950 extrapolated_downbeats: npt.NDArray[np.float32]
951
952
953def _vocal_duties(ctx: TransitionContext) -> tuple[float, float] | None:
954 """Outgoing/incoming vocal duty fractions the instrumental-blend/lazy-overlay gates key on."""
955 if ctx.vocal_out_scoring is None or ctx.vocal_in_scoring is None:
956 return None
957 out_duty = sum(right - left for left, right in ctx.vocal_out_scoring.windows) / max(
958 ctx.audio_end, 0.001
959 )
960 in_duty = sum(right - left for left, right in ctx.vocal_in_scoring.windows) / float(
961 SMART_CROSSFADE_DURATION
962 )
963 return out_duty, in_duty
964
965
966def _window_duties(ctx: TransitionContext, seconds: float) -> tuple[float, float] | None:
967 """
968 Vocal duty per deck over the window an unphrased overlay of ``seconds`` actually spans.
969
970 :param ctx: The transition context.
971 :param seconds: Requested overlay length; the outgoing window is the last
972 ``seconds`` before the audible end, the incoming window its first ``seconds``.
973 """
974 if ctx.vocal_out_scoring is None or ctx.vocal_in_scoring is None:
975 return None
976 # mirrors the anchored tail: a sub-half-second gap to the buffer end is not trimmed
977 effective_end = ctx.audio_end
978 if effective_end >= ctx.buffer_duration - 0.5:
979 effective_end = ctx.buffer_duration
980 span = min(seconds, effective_end)
981 if span <= 0.0:
982 return None
983 out_secs = sum(
984 max(0.0, min(right, effective_end) - max(left, effective_end - span))
985 for left, right in ctx.vocal_out_scoring.windows
986 )
987 in_secs = sum(
988 max(0.0, min(right, span) - max(left, 0.0)) for left, right in ctx.vocal_in_scoring.windows
989 )
990 return out_secs / span, in_secs / span
991
992
993def _fade_onset_pin(ctx: TransitionContext) -> float:
994 """Buffer-local anchor pin ahead of a detected mastered fade, else the default anchor."""
995 if ctx.fade_onset is None:
996 return ctx.default_anchor
997 # never pin inside A's own vocal window: exiting mid-phrase is the exact
998 # defect this pin fixes, so the vocal end floors the pin
999 lead_end = ctx.vocal_out_placement.last_end() if ctx.vocal_out_placement else 0.0
1000 onset_local = max(ctx.fade_onset, lead_end)
1001 if onset_local < MIN_EFFECTIVE_FADE_BUFFER:
1002 return ctx.default_anchor
1003 return min(ctx.default_anchor, onset_local)
1004
1005
1006def _entry_options(ctx: TransitionContext, bars: int) -> list[float]:
1007 """B entries for a rung: groove alignment, intro-keeping 0.0 (bars<=2), then natural entry."""
1008 import numpy as np # noqa: PLC0415
1009
1010 bar_b = ctx.incoming.beats_per_bar * 60.0 / ctx.incoming.bpm
1011 options: list[float] = []
1012 deep = ctx.natural_entry - bars * bar_b
1013 if deep > 0.0 and len(ctx.incoming.downbeats):
1014 downbeats = ctx.incoming.downbeats
1015 snapped = float(downbeats[np.argmin(np.abs(downbeats - deep))])
1016 in_mask = ctx.vocal_in_placement is not None and point_in_mask(
1017 ctx.vocal_in_placement, snapped
1018 )
1019 if snapped >= 0.0 and not in_mask:
1020 options.append(snapped)
1021 # at 1-2 bars the overlap is a handover, not a blend: keeping B's whole
1022 # intro (zero trim) lets it build naturally after A's tail rides out; it
1023 # goes before the natural entry so a scoring tie prefers keeping the intro
1024 if bars <= 2 and ctx.natural_entry > 0.0 and 0.0 not in options:
1025 options.append(0.0)
1026 if ctx.natural_entry not in options:
1027 options.append(ctx.natural_entry)
1028 return options
1029
1030
1031def _outgoing_vocal_end(ctx: TransitionContext) -> float:
1032 """Return the last outgoing vocal end within the audible tail, or 0.0 without vocal data."""
1033 mask = ctx.vocal_out_placement
1034 if mask is None or not mask.windows:
1035 return 0.0
1036 return min(mask.last_end(), ctx.audio_end)
1037
1038
1039def _nearest_protective_anchor(
1040 ctx: TransitionContext, target: float, *, prefer_earliest: bool = True
1041) -> float:
1042 """
1043 Protective downbeat at/after target within the RMS-audible boundary.
1044
1045 :param ctx: The transition context.
1046 :param target: Earliest buffer-local position the anchor may take.
1047 :param prefer_earliest: Use the first qualifying downbeat (protecting a
1048 vocal needs only just enough extra room) instead of the last (closing
1049 an audible-trim gap as tightly as possible).
1050 """
1051 candidates = [
1052 downbeat for downbeat in ctx.protective_downbeats if target <= downbeat <= ctx.audio_end
1053 ]
1054 if candidates:
1055 return candidates[0] if prefer_earliest else candidates[-1]
1056 return min(target, ctx.audio_end)
1057