/
/
/
1"""Shared helpers for smart fades."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING
6
7if TYPE_CHECKING:
8 import numpy as np
9 import numpy.typing as npt
10
11# Buffer size in seconds for crossfade analysis
12SMART_CROSSFADE_DURATION = 45
13
14# Below this many seconds of audible tail there is no room to place a musical
15# blend; the planner degrades such a boundary to a standard fade.
16MIN_EFFECTIVE_FADE_BUFFER = 8.0
17
18# Fraction of sustained (median-active) energy below which the outro no longer
19# carries the groove; the crossfade should end at or before this point.
20MIX_OUT_ENERGY_FRACTION = 0.70
21
22
23def detect_effective_audio_end(
24 rms_energy: npt.NDArray[np.float32] | list[float] | None,
25 track_duration: float | None,
26 buffer_duration: float,
27) -> float:
28 """
29 Return the buffer-local time where the outgoing track's audible content ends.
30
31 Returns ``buffer_duration`` when no usable energy data exists or there is no
32 trailing silence, and ``0.0`` when the entire tail is silent.
33
34 :param rms_energy: Peak-normalized RMS energy bins spanning the full track.
35 :param track_duration: Full track duration in seconds.
36 :param buffer_duration: Length in seconds of the fade-out holdback buffer.
37 """
38 # numpy is imported inside the functions here to keep it off the server startup path
39 import numpy as np # noqa: PLC0415
40
41 if rms_energy is None or not track_duration:
42 return buffer_duration
43 rms_energy = np.asarray(rms_energy, dtype=np.float32)
44 if len(rms_energy) < 2 or not np.any(np.isfinite(rms_energy)):
45 return buffer_duration
46 bin_duration = track_duration / len(rms_energy)
47 start_bin = max(0, int((track_duration - buffer_duration) / bin_duration))
48 tail = rms_energy[start_bin:]
49 # floor relative to sustained track energy, so hiss/noise tails count as
50 # silence but intentionally quiet outros do not
51 sustained = rms_energy[rms_energy > 0.01]
52 floor = max(0.02, 0.05 * float(np.median(sustained))) if len(sustained) else 0.02
53 audible = np.nonzero(tail > floor)[0]
54 if len(audible) == 0:
55 return 0.0
56 return min(
57 float((start_bin + audible[-1] + 1) * bin_duration)
58 - max(0.0, track_duration - buffer_duration),
59 buffer_duration,
60 )
61
62
63def extrapolate_downbeats(
64 downbeats: npt.NDArray[np.float32],
65 buffer_size: float = SMART_CROSSFADE_DURATION,
66 bpm: float | None = None,
67 beats_per_bar: int = 4,
68) -> npt.NDArray[np.float32]:
69 """
70 Extrapolate downbeats based on actual intervals when detection is incomplete.
71
72 This is needed when we want to perform beat alignment in an 'atmospheric' outro
73 that does not have any detected downbeats.
74
75 :param downbeats: Array of detected downbeat positions in seconds.
76 :param buffer_size: Maximum buffer size in seconds.
77 :param bpm: Optional BPM for validation when extrapolating with only 2 downbeats.
78 :param beats_per_bar: Track meter, used to predict the bar interval from BPM.
79 """
80 import numpy as np # noqa: PLC0415
81
82 # Handle case with exactly 2 downbeats (with BPM validation)
83 if len(downbeats) == 2 and bpm is not None:
84 interval = float(downbeats[1] - downbeats[0])
85
86 # Expected interval for this BPM and meter
87 expected_interval = (60.0 / bpm) * beats_per_bar
88
89 # Only extrapolate if interval matches BPM within 15% tolerance
90 if abs(interval - expected_interval) / expected_interval < 0.15:
91 last_downbeat = float(downbeats[-1])
92
93 # If the last downbeat is close to the buffer end, no extrapolation needed
94 if last_downbeat >= buffer_size - 5:
95 return downbeats
96
97 # Extrapolate forward from last downbeat
98 extrapolated = []
99 current_pos = last_downbeat + interval
100 max_extrapolation_distance = 25.0 # Don't extrapolate more than 25s
101
102 while (
103 current_pos < buffer_size
104 and (current_pos - last_downbeat) <= max_extrapolation_distance
105 ):
106 extrapolated.append(current_pos)
107 current_pos += interval
108
109 if extrapolated:
110 return np.concatenate([downbeats, np.array(extrapolated, dtype=np.float32)])
111
112 return downbeats
113 # else: interval doesn't match BPM, fall through to return original
114
115 if len(downbeats) < 2:
116 return downbeats
117
118 last_downbeat = float(downbeats[-1])
119
120 # If the last downbeat is close to the buffer end, no extrapolation needed
121 if last_downbeat >= buffer_size - 5:
122 return downbeats
123
124 # Calculate intervals between downbeats
125 intervals = np.diff(downbeats)
126 median_interval = float(np.median(intervals))
127 std_interval = float(np.std(intervals))
128
129 # Only extrapolate if intervals are consistent (low standard deviation)
130 if std_interval > 0.2:
131 return downbeats
132
133 # Extrapolate forward from last downbeat using median interval
134 extrapolated = []
135 current_pos = last_downbeat + median_interval
136 max_extrapolation_distance = 25.0 # Don't extrapolate more than 25s
137
138 while current_pos < buffer_size and (current_pos - last_downbeat) <= max_extrapolation_distance:
139 extrapolated.append(current_pos)
140 current_pos += median_interval
141
142 if extrapolated:
143 return np.concatenate([downbeats, np.array(extrapolated, dtype=np.float32)])
144
145 return downbeats
146
147
148def compute_gradual_tempo_steps(
149 start_ratio: float,
150 end_ratio: float,
151 downbeats: npt.NDArray[np.float32],
152 max_step_pct: float = 0.005,
153) -> list[tuple[float, float]]:
154 """
155 Compute S-curve tempo steps aligned to downbeats.
156
157 :param start_ratio: Starting tempo ratio (e.g., 1.0).
158 :param end_ratio: Target tempo ratio (e.g., 1.05).
159 :param downbeats: Downbeat timestamps to align steps to.
160 :param max_step_pct: Maximum tempo change per step as a fraction.
161 :return: List of (timestamp_seconds, tempo_ratio) tuples.
162 """
163 import numpy as np # noqa: PLC0415
164
165 total_change = abs(end_ratio - start_ratio)
166 if total_change < 1e-6:
167 return []
168
169 min_steps = max(1, int(np.ceil(total_change / max_step_pct)))
170 n_steps = min(min_steps, len(downbeats))
171 if n_steps < 1:
172 return [(0.0, end_ratio)]
173
174 # Evenly sample timestamps across the full window when we have more than needed
175 if len(downbeats) > n_steps:
176 indices = np.round(np.linspace(0, len(downbeats) - 1, n_steps)).astype(int)
177 selected_downbeats = downbeats[indices]
178 else:
179 selected_downbeats = downbeats[:n_steps]
180
181 # S-curve (sigmoid) with steepness adapted to keep max step within budget
182 if n_steps == 1:
183 sigmoid_values = np.array([1.0])
184 else:
185 # Binary search for the steepest k where max step <= max_step_pct
186 k_lo, k_hi = 0.1, 10.0
187 for _ in range(20):
188 k_mid = (k_lo + k_hi) / 2.0
189 x = np.linspace(-1, 1, n_steps)
190 s = 1.0 / (1.0 + np.exp(-k_mid * x))
191 s = (s - s[0]) / (s[-1] - s[0])
192 deltas = np.diff(s) * total_change
193 if float(np.max(deltas)) <= max_step_pct:
194 k_lo = k_mid
195 else:
196 k_hi = k_mid
197 k = k_lo
198 x = np.linspace(-1, 1, n_steps)
199 sigmoid_values = 1.0 / (1.0 + np.exp(-k * x))
200 sigmoid_values = (sigmoid_values - sigmoid_values[0]) / (
201 sigmoid_values[-1] - sigmoid_values[0]
202 )
203
204 steps: list[tuple[float, float]] = []
205 for i in range(n_steps):
206 timestamp = float(selected_downbeats[i])
207 ratio = start_ratio + (end_ratio - start_ratio) * float(sigmoid_values[i])
208 steps.append((timestamp, round(ratio, 6)))
209
210 return steps
211
212
213def generate_synthetic_timestamps(
214 stretch_duration: float,
215 bpm: float,
216 n_min: int = 4,
217 beats_per_bar: int = 4,
218) -> npt.NDArray[np.float32]:
219 """
220 Generate evenly-spaced synthetic timing points for gradual stretch.
221
222 Used when real beat/downbeat detection provides fewer than 2 timestamps
223 in the stretch window.
224
225 :param stretch_duration: Duration of the stretch window in seconds.
226 :param bpm: BPM of the track (used to approximate bar-level spacing).
227 :param n_min: Minimum number of timing points.
228 :param beats_per_bar: Track meter, used to approximate bar-level spacing.
229 """
230 import numpy as np # noqa: PLC0415
231
232 bar_duration = beats_per_bar * (60.0 / bpm)
233 n_points = max(n_min, int(stretch_duration / bar_duration))
234 return np.linspace(0, stretch_duration, n_points, dtype=np.float32)
235
236
237def sustained_energy_floor(rms_energy: npt.NDArray[np.float32]) -> float:
238 """
239 Median energy of the track's active (non-silent) bins.
240
241 :param rms_energy: Peak-normalized RMS energy bins spanning the full track.
242 """
243 import numpy as np # noqa: PLC0415
244
245 active = rms_energy[rms_energy > 0.01]
246 return float(np.median(active)) if len(active) else 0.0
247
248
249def detect_mix_out_point(
250 rms_energy: npt.NDArray[np.float32] | list[float] | None,
251 track_duration: float | None,
252 buffer_duration: float,
253 bpm: float,
254 fraction: float = MIX_OUT_ENERGY_FRACTION,
255 beats_per_bar: int = 4,
256) -> float:
257 """
258 Return the buffer-local time where the outro's energy last drops below the floor.
259
260 The floor is ``fraction`` of the track's sustained energy; the curve is
261 smoothed over ~1 bar so isolated quiet bins don't move the anchor. Returns
262 ``buffer_duration`` when no usable data exists or there is no decay, and
263 ``0.0`` when the entire tail sits below the floor.
264
265 :param rms_energy: Peak-normalized RMS energy bins spanning the full track.
266 :param track_duration: Full track duration in seconds.
267 :param buffer_duration: Length in seconds of the fade-out holdback buffer.
268 :param bpm: Track tempo, used for the one-bar smoothing window.
269 :param fraction: Energy floor as a fraction of the sustained level.
270 :param beats_per_bar: Track meter, used for the one-bar smoothing window.
271 """
272 import numpy as np # noqa: PLC0415
273
274 if rms_energy is None or not track_duration:
275 return buffer_duration
276 rms_energy = np.asarray(rms_energy, dtype=np.float32)
277 if len(rms_energy) < 2 or not np.any(np.isfinite(rms_energy)):
278 return buffer_duration
279 floor = fraction * sustained_energy_floor(rms_energy)
280 if floor <= 0.0:
281 return buffer_duration
282 bin_duration = track_duration / len(rms_energy)
283 bins_per_bar = max(1, round((beats_per_bar * 60.0 / bpm) / bin_duration))
284 # a median (not mean) over the bar: isolated quiet bins don't move the anchor,
285 # but a real silence cliff stays exactly where it is instead of smearing early
286 if bins_per_bar > 1:
287 padded = np.pad(rms_energy, bins_per_bar // 2, mode="edge")
288 windows = np.lib.stride_tricks.sliding_window_view(padded, bins_per_bar)
289 smoothed = np.median(windows, axis=1)[: len(rms_energy)]
290 else:
291 smoothed = rms_energy
292 start_bin = max(0, int((track_duration - buffer_duration) / bin_duration))
293 tail = smoothed[start_bin:]
294 above = np.nonzero(tail >= floor)[0]
295 if len(above) == 0:
296 return 0.0
297 return min(
298 float((start_bin + above[-1] + 1) * bin_duration)
299 - max(0.0, track_duration - buffer_duration),
300 buffer_duration,
301 )
302
303
304def detect_groove_entry(
305 rms_energy: npt.NDArray[np.float32] | list[float] | None,
306 track_duration: float | None,
307 downbeats: npt.NDArray[np.float32],
308 k_sigma: float = 1.5,
309) -> float:
310 """
311 Return the media time where the track's groove enters (first sustained energy step).
312
313 Detects the first bar whose energy steps up by more than ``k_sigma`` standard
314 deviations over the preceding bars AND holds for the following 4 bars â
315 typically the drums/kick entering after an intro. Returns ``0.0`` when the
316 track opens at full energy or no usable data exists.
317
318 :param rms_energy: Peak-normalized RMS energy bins spanning the full track.
319 :param track_duration: Full track duration in seconds.
320 :param downbeats: Downbeat grid in media time.
321 :param k_sigma: Step threshold in standard deviations of bar-to-bar changes.
322 """
323 import numpy as np # noqa: PLC0415
324
325 if rms_energy is None or not track_duration or len(downbeats) < 12:
326 return 0.0
327 rms_energy = np.asarray(rms_energy, dtype=np.float32)
328 t = (np.arange(len(rms_energy)) + 0.5) * (track_duration / len(rms_energy))
329 bar_rms = np.array(
330 [
331 float(rms_energy[(t >= downbeats[i]) & (t < downbeats[i + 1])].mean())
332 if ((t >= downbeats[i]) & (t < downbeats[i + 1])).any()
333 else 0.0
334 for i in range(len(downbeats) - 1)
335 ]
336 )
337 diffs = np.diff(bar_rms)
338 sigma = float(np.std(diffs)) or 1e-6
339 floor = 0.5 * sustained_energy_floor(rms_energy)
340 for i in range(1, len(bar_rms) - 4):
341 pre = bar_rms[max(0, i - 4) : i].mean()
342 post = bar_rms[i : i + 4].mean()
343 if bar_rms[i] - pre > k_sigma * sigma and post > max(pre * 1.5, floor):
344 return float(downbeats[i])
345 return 0.0
346
347
348def db_ramp(
349 start: float,
350 duration: float,
351 from_db: float,
352 to_db: float,
353 step_interval: float = 0.1,
354) -> list[tuple[float, float]]:
355 """
356 Build a linear-in-dB gain schedule for asendcmd-driven filters.
357
358 :param start: Schedule start time in seconds.
359 :param duration: Ramp length in seconds.
360 :param from_db: Gain at the start of the ramp.
361 :param to_db: Gain at the end of the ramp.
362 :param step_interval: Seconds between steps (small enough to avoid zipper noise).
363 """
364 n_steps = max(2, int(duration / step_interval))
365 return [
366 (start + (i / n_steps) * duration, from_db + (to_db - from_db) * (i / n_steps))
367 for i in range(n_steps + 1)
368 ]
369
370
371def keys_compatible(
372 key_a: str | None, mode_a: str | None, key_b: str | None, mode_b: str | None
373) -> bool:
374 """
375 Return True when two keys mix harmonically on the Camelot wheel.
376
377 Compatible = same slot, one step along the wheel in the same mode, or
378 relative major/minor. Unknown or missing keys are treated as incompatible
379 so key gating only ever shortens a blend (a short fade never sounds wrong).
380
381 :param key_a: Pitch class of the first track's key, e.g. "C", "F#", "Bb".
382 :param mode_a: "major" or "minor".
383 :param key_b: Pitch class of the second track's key.
384 :param mode_b: "major" or "minor".
385 """
386 a = _camelot_code(key_a, mode_a)
387 b = _camelot_code(key_b, mode_b)
388 if a is None or b is None:
389 return False
390 num_a, major_a = a
391 num_b, major_b = b
392 if major_a == major_b:
393 return min((num_a - num_b) % 12, (num_b - num_a) % 12) <= 1
394 return num_a == num_b
395
396
397_FLAT_TO_SHARP = {"Db": "C#", "Eb": "D#", "Gb": "F#", "Ab": "G#", "Bb": "A#"}
398# Camelot wheel numbers; relative major/minor share a number (Am=8A, C=8B)
399_CAMELOT_MAJOR = {
400 "C": 8,
401 "G": 9,
402 "D": 10,
403 "A": 11,
404 "E": 12,
405 "B": 1,
406 "F#": 2,
407 "C#": 3,
408 "G#": 4,
409 "D#": 5,
410 "A#": 6,
411 "F": 7,
412}
413_CAMELOT_MINOR = {
414 "A": 8,
415 "E": 9,
416 "B": 10,
417 "F#": 11,
418 "C#": 12,
419 "G#": 1,
420 "D#": 2,
421 "A#": 3,
422 "F": 4,
423 "C": 5,
424 "G": 6,
425 "D": 7,
426}
427
428
429def _camelot_code(key: str | None, mode: str | None) -> tuple[int, bool] | None:
430 """Return (wheel number, is_major) or None for unknown input."""
431 if not key or mode not in ("major", "minor"):
432 return None
433 key = _FLAT_TO_SHARP.get(key, key)
434 table = _CAMELOT_MAJOR if mode == "major" else _CAMELOT_MINOR
435 num = table.get(key)
436 return (num, mode == "major") if num else None
437