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