/
/
/
1"""
2Smart Fades - immutable per-transition context.
3
4A ``TransitionContext`` is every PER-TRANSITION FACT a planner needs: the two
5decks, their band profiles, the outgoing tail's energy/kick anchors and audible
6boundary, the chosen tier, the vocal-activity masks and the coda/fade-onset
7detections. None of it depends on a candidate's chosen overlap length or
8re-anchor - those anchor-DEPENDENT derivations (re-anchored grids, trims) are
9a candidate factory's job. Building this once, frozen, replaces the old
10planner's mutable ``self`` scratchpad (``_pristine_*`` snapshots, re-anchoring
11in place) with a value every candidate build reads from but never mutates.
12"""
13
14from __future__ import annotations
15
16import logging
17from dataclasses import dataclass, replace
18from typing import TYPE_CHECKING
19
20from music_assistant.constants import VERBOSE_LOG_LEVEL
21from music_assistant.controllers.streams.smart_fades.bands import (
22 build_band_profile,
23 detect_low_groove_entry,
24 detect_low_mix_out,
25)
26from music_assistant.controllers.streams.smart_fades.helpers import (
27 MIN_EFFECTIVE_FADE_BUFFER,
28 SMART_CROSSFADE_DURATION,
29 detect_effective_audio_end,
30 detect_groove_entry,
31 detect_mix_out_point,
32 extrapolate_downbeats,
33 keys_compatible,
34)
35from music_assistant.controllers.streams.smart_fades.models import (
36 Deck,
37 SmartFadeNotApplicable,
38 TransitionTier,
39)
40from music_assistant.controllers.streams.smart_fades.structure import (
41 detect_coda_zone,
42 detect_mastered_fadeout,
43)
44from music_assistant.controllers.streams.smart_fades.vocal import (
45 DEFAULT_VOCAL_CONFIG,
46 VocalMask,
47 build_vocal_windows,
48 mask_saturated,
49 parse_vocal_probabilities,
50)
51
52if TYPE_CHECKING:
53 import numpy as np
54 import numpy.typing as npt
55
56 from music_assistant.controllers.streams.smart_fades.models import BandProfile
57 from music_assistant.controllers.streams.smart_fades.structure import CodaZone
58 from music_assistant.controllers.streams.smart_fades.vocal import (
59 VocalHysteresisConfig,
60 VocalTimeline,
61 )
62 from music_assistant.models.audio_analysis import AudioAnalysisData
63
64# Only apply time stretching if BPM difference is < this %
65# (research: the tier-1 dance-cluster population triples at ±8 vs ±5)
66TIME_STRETCH_BPM_PERCENTAGE_THRESHOLD: float = 8.0
67
68# A track qualifies for kick-following anchors when its median active-bar
69# low-band power fraction reaches this share of its total power
70_LOW_TIMING_ELIGIBILITY: float = 0.10
71# Reference multiplier a bar's low power must reach to count as "kick present"
72_LOW_ANCHOR_BAR_FRACTION: float = 0.5
73# Beyond this many outgoing bars of disagreement with the full-band anchor,
74# the low anchor is untrustworthy and the full-band anchor wins
75_LOW_ANCHOR_DIVERGENCE_BARS: int = 16
76
77# D2 mastered-fadeout gates: engineered fades run 5-15s past -10dB with a
78# frozen spectrum (a post-mix gain ramp); musical decrescendos rarely
79# exceed ~6dB and re-orchestrate (band fractions move)
80_FADE_MIN_BARS: int = 4
81_FADE_DROP_DB: float = 10.0
82_FADE_MONOTONE_SHARE: float = 0.8
83_FADE_JITTER_DB: float = 0.5
84_FADE_FRAC_DRIFT: float = 0.15
85_FADE_AUDIBLE_FLOOR: float = 0.01
86_FADE_FRAC_FLOOR: float = 0.10
87# D4 coda gates: the bed must still carry audible program under an
88# equal-power fade; one musical gesture minimum; designed outros sustain
89# while any fade halves across the zone
90_CODA_TOTAL_FLOOR: float = 0.15
91_CODA_MIN_SECONDS: float = 4.0
92_CODA_MIN_BARS: int = 2
93_CODA_LEVEL_HOLD: float = 0.5
94
95
96@dataclass(frozen=True, slots=True)
97class TransitionContext:
98 """
99 Every per-transition fact a planner needs, computed once and never mutated.
100
101 All buffer-local times share one origin: the outgoing track's
102 ``buffer_offset`` media position. ``outgoing``/``incoming`` carry the
103 unmasked (only dropped-before-buffer beats removed) beat grids; a
104 candidate factory masks them to whichever anchor it is building.
105 """
106
107 outgoing: Deck
108 incoming: Deck
109 outgoing_profile: BandProfile | None
110 incoming_profile: BandProfile | None
111 buffer_duration: float
112 buffer_offset: float
113 audio_end: float
114 # the kick-folded, downbeat-snapped anchor (the old planner's pristine
115 # effective_end): the tier keyed on it, and it is where a candidate with
116 # no explicit anchor cues the tail
117 default_anchor: float
118 # the pure full-band mix-out anchor, kept as a diagnostic fact only: no
119 # generator anchors here (a full-band variant would defeat the kick fold;
120 # trim-closing may still anchor later when >=8s of audible tail would
121 # otherwise be stranded)
122 mix_out_anchor: float | None
123 kick_anchor: float | None
124 fade_onset: float | None
125 coda_zone: CodaZone | None
126 tier: TransitionTier
127 cross_meter: bool
128 bpm_diff_percent: float
129 vocal_out_placement: VocalMask | None
130 vocal_in_placement: VocalMask | None
131 vocal_out_scoring: VocalMask | None
132 vocal_in_scoring: VocalMask | None
133 natural_entry: float
134 protective_downbeats: tuple[float, ...]
135
136
137def build_transition_context(
138 fade_out_analysis: AudioAnalysisData,
139 fade_in_analysis: AudioAnalysisData,
140 buffer_duration: float,
141 logger: logging.Logger,
142) -> TransitionContext:
143 """
144 Build the immutable per-transition context from the two tracks' analysis.
145
146 Raises ``SmartFadeNotApplicable`` when the outgoing tail is too short
147 (mostly silent, or too short once anchored) for any candidate to be built.
148
149 :param fade_out_analysis: Analysis data for the outgoing track.
150 :param fade_in_analysis: Analysis data for the incoming track.
151 :param buffer_duration: Length in seconds of the available fade-out holdback.
152 :param logger: Logger for verbose per-transition diagnostics.
153 """
154 # numpy is imported inside the function to keep it off the server startup path
155 import numpy as np # noqa: PLC0415
156
157 if (
158 fade_out_analysis.bpm is None
159 or fade_in_analysis.bpm is None
160 or fade_out_analysis.beats is None
161 or fade_in_analysis.beats is None
162 ):
163 raise ValueError("AudioAnalysisData must have bpm and beats set for smart crossfade")
164
165 # AudioAnalysisData stores the grids as plain float lists (numpy-free model);
166 # the planner works in numpy, so convert once here.
167 incoming_beats = np.asarray(fade_in_analysis.beats, dtype=np.float32)
168 incoming_downbeats = (
169 np.asarray(fade_in_analysis.downbeats, dtype=np.float32)
170 if fade_in_analysis.downbeats is not None
171 else incoming_beats
172 )
173 incoming = Deck(
174 analysis=fade_in_analysis,
175 bpm=fade_in_analysis.bpm,
176 # Only beats within the buffered head are usable for alignment decisions
177 beats=incoming_beats[incoming_beats <= SMART_CROSSFADE_DURATION],
178 downbeats=incoming_downbeats[incoming_downbeats <= SMART_CROSSFADE_DURATION],
179 beats_per_bar=fade_in_analysis.beats_per_bar or 4,
180 )
181 outgoing = Deck(
182 analysis=fade_out_analysis,
183 bpm=fade_out_analysis.bpm,
184 # Raw full-track grids; the shift to buffer-local coordinates happens
185 # in _cue_outgoing_tail where the actual buffer length is known
186 beats=np.asarray(fade_out_analysis.beats, dtype=np.float32),
187 downbeats=(
188 np.asarray(fade_out_analysis.downbeats, dtype=np.float32)
189 if fade_out_analysis.downbeats is not None
190 else np.array([], dtype=np.float32)
191 ),
192 beats_per_bar=fade_out_analysis.beats_per_bar or 4,
193 )
194 outgoing_profile = build_band_profile(fade_out_analysis)
195 incoming_profile = build_band_profile(fade_in_analysis)
196
197 (
198 buffer_offset,
199 audio_end,
200 tier_anchor,
201 mix_out_anchor,
202 kick_anchor,
203 grid_beats,
204 grid_downbeats,
205 ) = _cue_outgoing_tail(outgoing, outgoing_profile, buffer_duration)
206 outgoing = replace(outgoing, beats=grid_beats, downbeats=grid_downbeats)
207
208 # Extrapolated up to the true RMS-audible boundary rather than the (possibly
209 # downbeat-snapped) mix_out_anchor, so a later vocal re-anchor can always
210 # find a real downbeat between the two
211 protective_downbeats = extrapolate_downbeats(
212 grid_downbeats,
213 buffer_size=audio_end,
214 bpm=outgoing.bpm,
215 beats_per_bar=outgoing.beats_per_bar,
216 )
217 # Computed once so every candidate's swap point and reciprocal decision windows agree
218 natural_entry = _detect_incoming_entry(incoming, incoming_profile)
219
220 vocal_out_placement, vocal_in_placement, vocal_out_scoring, vocal_in_scoring = (
221 _build_vocal_masks(
222 fade_out_analysis, fade_in_analysis, outgoing, incoming, buffer_offset, audio_end
223 )
224 )
225
226 # the tier reads the kick-folded anchor (the old planner's effective_end),
227 # never the pure full-band mix_out_anchor: a kick-timed track's blendability
228 # window ends where its kick dies, exactly as the old masked grid did
229 cross_meter, tier = choose_tier(outgoing, incoming, tier_anchor)
230 bpm_diff_percent = _bpm_diff_percent(outgoing.bpm, incoming.bpm)
231
232 # fade detection is a per-transition fact regardless of which anchor a
233 # candidate later picks; it never touches candidate state
234 fade_onset_media = (
235 detect_mastered_fadeout(
236 outgoing_profile,
237 buffer_offset,
238 fade_out_analysis.duration or 0.0,
239 min_bars=_FADE_MIN_BARS,
240 drop_db=_FADE_DROP_DB,
241 monotone_share=_FADE_MONOTONE_SHARE,
242 jitter_db=_FADE_JITTER_DB,
243 frac_drift=_FADE_FRAC_DRIFT,
244 audible_floor=_FADE_AUDIBLE_FLOOR,
245 frac_floor=_FADE_FRAC_FLOOR,
246 )
247 if outgoing_profile is not None
248 else None
249 )
250 fade_onset = fade_onset_media - buffer_offset if fade_onset_media is not None else None
251
252 coda_zone = _detect_coda_zone(
253 outgoing,
254 outgoing_profile,
255 vocal_out_placement,
256 fade_onset_media,
257 buffer_offset,
258 tier_anchor,
259 fade_out_analysis.duration or 0.0,
260 )
261
262 if vocal_out_scoring is not None and vocal_in_scoring is not None:
263 vocal_coverage = "both"
264 elif vocal_out_scoring is not None:
265 vocal_coverage = "out"
266 elif vocal_in_scoring is not None:
267 vocal_coverage = "in"
268 else:
269 vocal_coverage = "none"
270 logger.log(
271 VERBOSE_LOG_LEVEL,
272 "transition context: tier=%s bpm=%.1f->%.1f (diff=%.1f%%) cross_meter=%s buffer=%.1fs "
273 "offset=%.1fs audio_end=%.1fs anchor=%.2f mix_out=%.2f kick=%s fade_onset=%s coda=%s "
274 "natural_entry=%.2f vocals=%s",
275 tier,
276 outgoing.bpm,
277 incoming.bpm,
278 bpm_diff_percent,
279 cross_meter,
280 buffer_duration,
281 buffer_offset,
282 audio_end,
283 tier_anchor,
284 mix_out_anchor,
285 kick_anchor,
286 fade_onset,
287 coda_zone,
288 natural_entry,
289 vocal_coverage,
290 )
291
292 return TransitionContext(
293 outgoing=outgoing,
294 incoming=incoming,
295 outgoing_profile=outgoing_profile,
296 incoming_profile=incoming_profile,
297 buffer_duration=buffer_duration,
298 buffer_offset=buffer_offset,
299 audio_end=audio_end,
300 default_anchor=tier_anchor,
301 mix_out_anchor=mix_out_anchor,
302 kick_anchor=kick_anchor,
303 fade_onset=fade_onset,
304 coda_zone=coda_zone,
305 tier=tier,
306 cross_meter=cross_meter,
307 bpm_diff_percent=bpm_diff_percent,
308 vocal_out_placement=vocal_out_placement,
309 vocal_in_placement=vocal_in_placement,
310 vocal_out_scoring=vocal_out_scoring,
311 vocal_in_scoring=vocal_in_scoring,
312 natural_entry=natural_entry,
313 protective_downbeats=tuple(float(x) for x in protective_downbeats),
314 )
315
316
317def _cue_outgoing_tail(
318 outgoing: Deck, outgoing_profile: BandProfile | None, buffer_duration: float
319) -> tuple[
320 float, float, float, float, float | None, npt.NDArray[np.float32], npt.NDArray[np.float32]
321]:
322 """
323 Anchor the outgoing tail at its energy mix-out point, snapped to a downbeat.
324
325 Returns ``(buffer_offset, audio_end, tier_anchor, mix_out_anchor,
326 kick_anchor, grid_beats, grid_downbeats)``. ``tier_anchor`` is the
327 kick-FOLDED anchor the old planner called ``effective_end``: applicability
328 and the tier decision key on it, so a kick-timed track keeps its original
329 (shorter) blendability window. ``mix_out_anchor`` (pure full-band) and
330 ``kick_anchor`` (low-band, ``None`` if ineligible) stay separate facts -
331 which one a candidate should start from is a candidate-factory decision.
332 ``grid_beats``/``grid_downbeats`` are the unmasked (only dropping
333 pre-buffer beats) buffer-local grids, for a later candidate to mask to
334 whichever anchor it picks.
335 """
336 # ACTUAL buffer length, not the constant 45s: the holdback yield loop leaves
337 # up to ~1s less depending on chunk boundaries, and every buffer-local
338 # coordinate below (mix-out detection, grid shift) must agree on it
339 buffer_offset = max(0.0, (outgoing.analysis.duration or 0.0) - buffer_duration)
340 silence_end = detect_effective_audio_end(
341 outgoing.analysis.rms_energy, outgoing.analysis.duration, buffer_duration
342 )
343 raw_mix_out = detect_mix_out_point(
344 outgoing.analysis.rms_energy,
345 outgoing.analysis.duration,
346 buffer_duration,
347 outgoing.bpm,
348 beats_per_bar=outgoing.beats_per_bar,
349 )
350 kick_anchor = _apply_low_mix_out(outgoing, outgoing_profile, raw_mix_out, buffer_offset)
351
352 # the old planner folded the kick anchor into effective_end BEFORE snapping
353 # and masking the grids, so applicability and the blendability window (and
354 # thus the tier) are kick-aware; the fold stays tier-decision-local here
355 folded_mix_out = kick_anchor if kick_anchor is not None else raw_mix_out
356 tier_anchor = min(silence_end, folded_mix_out)
357 if tier_anchor < MIN_EFFECTIVE_FADE_BUFFER:
358 raise SmartFadeNotApplicable(f"outgoing tail is mostly silent ({tier_anchor:.1f}s audible)")
359
360 # Shift fade-out beats from full-track to buffer-local coordinates
361 beats = outgoing.beats - buffer_offset
362 downbeats = outgoing.downbeats - buffer_offset
363
364 # Unmasked (only dropping pre-buffer beats) grids: a later vocal/coda
365 # re-anchor reads from these rather than re-deriving buffer-local coordinates
366 grid_beats = beats[beats >= 0.0]
367 grid_downbeats = downbeats[downbeats >= 0.0]
368
369 # the RMS-audible boundary: a hard upper bound a later vocal/coda re-anchor
370 # may extend the (about to be downbeat-snapped) anchor back up to, never past
371 audio_end = min(silence_end, buffer_duration)
372
373 tier_anchor = _snap_anchor(tier_anchor, buffer_duration, grid_downbeats, outgoing)
374 if tier_anchor < MIN_EFFECTIVE_FADE_BUFFER:
375 raise SmartFadeNotApplicable(
376 f"outgoing tail too short after anchoring ({tier_anchor:.1f}s)"
377 )
378 mix_out_anchor = (
379 tier_anchor
380 if kick_anchor is None
381 else _snap_anchor(min(silence_end, raw_mix_out), buffer_duration, grid_downbeats, outgoing)
382 )
383
384 return (
385 buffer_offset,
386 audio_end,
387 tier_anchor,
388 mix_out_anchor,
389 kick_anchor,
390 grid_beats,
391 grid_downbeats,
392 )
393
394
395def _snap_anchor(
396 anchor: float,
397 buffer_duration: float,
398 grid_downbeats: npt.NDArray[np.float32],
399 outgoing: Deck,
400) -> float:
401 """Snap a tail anchor back to the last real downbeat within ~2 bars, or to the buffer end."""
402 # Sub-half-second slack is not worth trimming: RMS bin granularity is
403 # ~0.1-0.2s for typical track lengths, so finer precision is illusory
404 if anchor >= buffer_duration - 0.5:
405 # Without the trim the rendered stream still ends at buffer_duration,
406 # so the anchor must follow it or every schedule lands early
407 return buffer_duration
408 # Snap the anchor back to the last real downbeat within ~2 bars so the
409 # crossfade ends cleanly on the 1 rather than at an arbitrary RMS bin edge
410 bar_seconds = outgoing.beats_per_bar * 60.0 / outgoing.bpm
411 in_window = grid_downbeats[grid_downbeats <= anchor]
412 if len(in_window) and anchor - float(in_window[-1]) < 2 * bar_seconds:
413 return float(in_window[-1])
414 return anchor
415
416
417def _apply_low_mix_out(
418 outgoing: Deck,
419 outgoing_profile: BandProfile | None,
420 full_band_mix_out: float,
421 buffer_offset: float,
422) -> float | None:
423 """Return the low-band (kick) mix-out anchor when eligible and trustworthy, else None."""
424 if not _is_low_timing_eligible(outgoing_profile):
425 return None
426 assert outgoing_profile is not None # narrowed by the eligibility check
427 low_mix_out = detect_low_mix_out(outgoing_profile, _LOW_ANCHOR_BAR_FRACTION)
428 if low_mix_out is None:
429 return None
430 low_mix_out_local = low_mix_out - buffer_offset
431 bar_seconds = outgoing.beats_per_bar * 60.0 / outgoing.bpm
432 divergence_bars = abs(low_mix_out_local - full_band_mix_out) / bar_seconds
433 if divergence_bars > _LOW_ANCHOR_DIVERGENCE_BARS:
434 return None
435 if low_mix_out_local < MIN_EFFECTIVE_FADE_BUFFER:
436 return None
437 return low_mix_out_local
438
439
440def _is_low_timing_eligible(profile: BandProfile | None) -> bool:
441 """Return True when a track's low band carries enough of its power to time off."""
442 import numpy as np # noqa: PLC0415
443
444 if profile is None:
445 return False
446 f_low = profile.bar_power["low"][profile.active] / profile.total_power[profile.active]
447 return float(np.median(f_low)) >= _LOW_TIMING_ELIGIBILITY
448
449
450def _detect_incoming_entry(incoming: Deck, incoming_profile: BandProfile | None) -> float:
451 """Detect B's groove entry, preferring the kick when B is low-timing eligible."""
452 if _is_low_timing_eligible(incoming_profile):
453 assert incoming_profile is not None # narrowed by the eligibility check
454 low_entry = detect_low_groove_entry(incoming_profile, _LOW_ANCHOR_BAR_FRACTION)
455 if low_entry is not None and low_entry < SMART_CROSSFADE_DURATION:
456 return low_entry
457 return detect_groove_entry(
458 incoming.analysis.rms_energy, incoming.analysis.duration, incoming.downbeats
459 )
460
461
462def _build_vocal_masks(
463 fade_out_analysis: AudioAnalysisData,
464 fade_in_analysis: AudioAnalysisData,
465 outgoing: Deck,
466 incoming: Deck,
467 buffer_offset: float,
468 audio_end: float,
469) -> tuple[VocalMask | None, VocalMask | None, VocalMask | None, VocalMask | None]:
470 """
471 Build each deck's placement and scoring vocal masks, per deck.
472
473 :param fade_out_analysis: Analysis row for the outgoing track.
474 :param fade_in_analysis: Analysis row for the incoming track.
475 :param outgoing: The outgoing deck, for its BPM.
476 :param incoming: The incoming deck, for its BPM.
477 :param buffer_offset: Media time where the outgoing buffer starts.
478 :param audio_end: Buffer-local RMS-audible boundary.
479
480 Returns ``(out_placement, in_placement, out_scoring, in_scoring)``; a deck
481 without a validated FireRed timeline yields ``None`` for its two masks.
482 """
483 out_timeline = parse_vocal_probabilities(fade_out_analysis)
484 in_timeline = parse_vocal_probabilities(fade_in_analysis)
485 duration = fade_out_analysis.duration or buffer_offset
486 config = DEFAULT_VOCAL_CONFIG
487 # the scoring variant differs only in padding: padded edges are silence,
488 # so they place cuts and anchors but never count as collision
489 scoring_config = replace(config, left_padding=0.0, right_padding=0.0)
490 return (
491 _build_outgoing_mask(out_timeline, duration, config, outgoing, buffer_offset, audio_end),
492 _build_incoming_mask(in_timeline, config, incoming),
493 _build_outgoing_mask(
494 out_timeline, duration, scoring_config, outgoing, buffer_offset, audio_end
495 ),
496 _build_incoming_mask(in_timeline, scoring_config, incoming),
497 )
498
499
500def _build_outgoing_mask(
501 timeline: VocalTimeline | None,
502 duration: float,
503 config: VocalHysteresisConfig,
504 outgoing: Deck,
505 buffer_offset: float,
506 audio_end: float,
507) -> VocalMask | None:
508 """
509 Build the outgoing deck's buffer-local vocal mask for one config.
510
511 :param timeline: The outgoing deck's validated FireRed timeline, or ``None``.
512 :param duration: The outgoing track's media duration in seconds.
513 :param config: Hysteresis/padding thresholds for the mask.
514 :param outgoing: The outgoing deck, for its BPM.
515 :param buffer_offset: Media time where the buffer starts.
516 :param audio_end: Buffer-local RMS-audible boundary.
517 """
518 if timeline is None:
519 return None
520 media_time_mask = build_vocal_windows(
521 timeline.probabilities,
522 timeline.frame_duration,
523 buffer_offset,
524 duration,
525 beat_duration=60.0 / outgoing.bpm,
526 config=config,
527 )
528 # shift media time to buffer-local time, then clamp to the RMS-audible
529 # boundary: FireRed may never be trusted to extend it, so any window
530 # (or trailing sliver of one) past it is simply dropped
531 buffer_local = VocalMask(
532 windows=[
533 (left - buffer_offset, right - buffer_offset) for left, right in media_time_mask.windows
534 ]
535 )
536 return buffer_local.clamped_to(audio_end)
537
538
539def _build_incoming_mask(
540 timeline: VocalTimeline | None, config: VocalHysteresisConfig, incoming: Deck
541) -> VocalMask | None:
542 """
543 Build the incoming deck's head vocal mask for one config.
544
545 :param timeline: The incoming deck's validated FireRed timeline, or ``None``.
546 :param config: Hysteresis/padding thresholds for the mask.
547 :param incoming: The incoming deck, for its BPM.
548 """
549 if timeline is None:
550 return None
551 return build_vocal_windows(
552 timeline.probabilities,
553 timeline.frame_duration,
554 0.0,
555 float(SMART_CROSSFADE_DURATION),
556 beat_duration=60.0 / incoming.bpm,
557 config=config,
558 )
559
560
561def choose_tier(
562 outgoing: Deck,
563 incoming: Deck,
564 tier_anchor: float,
565) -> tuple[bool, TransitionTier]:
566 """Pick the transition tier; anything that casts doubt on a long blend picks a shorter one."""
567 cross_meter = outgoing.beats_per_bar != incoming.beats_per_bar
568 if cross_meter:
569 # no shared bar grid to beatmatch or blend across
570 return cross_meter, TransitionTier.QUICK_FADE
571 anchored_downbeats = outgoing.downbeats[outgoing.downbeats <= tier_anchor]
572 if not _tail_is_blendable(anchored_downbeats):
573 return cross_meter, TransitionTier.QUICK_FADE
574 if _bpm_diff_percent(outgoing.bpm, incoming.bpm) > TIME_STRETCH_BPM_PERCENTAGE_THRESHOLD:
575 return cross_meter, TransitionTier.QUICK_FADE
576 out_a, in_a = outgoing.analysis, incoming.analysis
577 # the 16-bar tier is earned by a verifiable energy anchor: without RMS data
578 # the blend could land on a mastered fade-out unnoticed
579 if out_a.rms_energy is not None and keys_compatible(out_a.key, out_a.mode, in_a.key, in_a.mode):
580 # a non-4/4 meter has no corpus evidence to support a 16-bar blend
581 if outgoing.beats_per_bar != 4:
582 return cross_meter, TransitionTier.TEMPO_BLEND
583 return cross_meter, TransitionTier.FULL_BLEND
584 return cross_meter, TransitionTier.TEMPO_BLEND
585
586
587def _tail_is_blendable(downbeats: npt.NDArray[np.float32]) -> bool:
588 """Return True when the anchored tail has enough regular downbeats for a blend."""
589 import numpy as np # noqa: PLC0415
590
591 if len(downbeats) < 8:
592 return False
593 # metronomic grid: research measured 74% of library tails under 0.1s interval std
594 return float(np.std(np.diff(downbeats))) < 0.1
595
596
597def _bpm_diff_percent(outgoing_bpm: float, incoming_bpm: float) -> float:
598 """Tempo difference between the two decks as a percentage."""
599 return abs(1.0 - incoming_bpm / outgoing_bpm) * 100
600
601
602def _detect_coda_zone(
603 outgoing: Deck,
604 outgoing_profile: BandProfile | None,
605 vocal_out_placement: VocalMask | None,
606 fade_onset_media: float | None,
607 buffer_offset: float,
608 tier_anchor: float,
609 duration: float,
610) -> CodaZone | None:
611 """Detect a validated outro/coda zone over the buffered tail, if any."""
612 if outgoing_profile is None:
613 return None
614 media_out = _vocal_out_media_mask(vocal_out_placement, buffer_offset)
615 tail_start = buffer_offset
616 out_saturated = mask_saturated(media_out, max(0.001, duration - tail_start))
617 if out_saturated:
618 # a saturated (near-continuous vocal) outro supplies no fine structure
619 # to distinguish a coda from the rest of the track
620 return None
621 earliest = _coda_earliest(outgoing, outgoing_profile, media_out, buffer_offset, tier_anchor)
622 return detect_coda_zone(
623 outgoing_profile,
624 media_out,
625 earliest,
626 fade_onset_media,
627 tail_start,
628 duration,
629 total_floor=_CODA_TOTAL_FLOOR,
630 min_seconds=_CODA_MIN_SECONDS,
631 min_bars=_CODA_MIN_BARS,
632 level_hold=_CODA_LEVEL_HOLD,
633 )
634
635
636def _coda_earliest(
637 outgoing: Deck,
638 outgoing_profile: BandProfile | None,
639 media_out: VocalMask,
640 buffer_offset: float,
641 tier_anchor: float,
642) -> float:
643 """Media time before which no coda bar may start: past A's vocal and its kick."""
644 lead_end = media_out.last_end()
645 kick_end: float | None = None
646 if _is_low_timing_eligible(outgoing_profile):
647 assert outgoing_profile is not None # narrowed by the eligibility check
648 kick_end = detect_low_mix_out(outgoing_profile, _LOW_ANCHOR_BAR_FRACTION)
649 if kick_end is not None:
650 # detect_low_mix_out returns the last kick bar's START; the coda
651 # begins after that bar rings out
652 kick_end += outgoing.beats_per_bar * 60.0 / outgoing.bpm
653 if kick_end is None:
654 kick_end = buffer_offset + tier_anchor
655 return max(lead_end, kick_end)
656
657
658def _vocal_out_media_mask(vocal_out_placement: VocalMask | None, buffer_offset: float) -> VocalMask:
659 """Return the outgoing vocal mask shifted from buffer-local back to media time (coda scope)."""
660 if vocal_out_placement is None:
661 return VocalMask(windows=[])
662 return VocalMask(
663 windows=[
664 (left + buffer_offset, right + buffer_offset)
665 for left, right in vocal_out_placement.windows
666 ]
667 )
668