ntr/maple
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

change bpm detect

Browse Source
This commit is contained in:
Nathan Rashleigh committed 2026-06-22 18:06:45 +10:00
1 parent b85cea496f
commit 096bce803a
6 files changed
+258 -359
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

maple-py/pyproject.toml
+1 -2
View File
@@ -9,12 +9,11 @@ authors = [
requires-python = ">=3.14"
dependencies = [
"aiohttp>=3.10",
"aubio>=0.4.9",
"kaitaistruct>=0.11",
"loguru>=0.7",
"mixxx-analyzer>=0.1.2",
"mutagen>=1.47.0",
"pillow>=12.2.0",
"polars>=1.0",
"pyrekordbox>=0.4.4",
"scipy>=1.10",
]
maple-py/src/maple/analyze.py
+152 -233
View File
@@ -1,12 +1,13 @@
"""Audio analysis: analytical BPM detection + waveform envelope extraction.
"""Audio analysis: BPM + beat grid via mixxx-analyzer, plus waveform envelopes.
ffmpeg decodes the source to 44.1 kHz mono float32 PCM on stdout. The
samples are then either streamed in hop-size chunks into aubio.tempo for
BPM detection, or fully buffered and binned to produce the per-column
amplitude envelopes that PWAV / PWV3 ANLZ sections embed.
BPM and the beat grid come from `mixxx-analyzer` — Python bindings around
Mixxx's own analysis (the Queen Mary tempo tracker) — which produces a
CDJ-grade beat grid. Its wheel bundles a native binary, so there's no extra
system dependency. ffmpeg still decodes the source to 44.1 kHz mono float32 PCM
for the per-column waveform envelopes (PWAV / PWV3 / PWV4 / PWV5).
The decoded audio never touches disk — ffmpeg's `-f f32le` output is
piped straight to numpy.
The decoded audio never touches disk — ffmpeg's `-f f32le` output is piped
straight to numpy.
"""
from __future__ import annotations
@@ -16,90 +17,74 @@ import subprocess
from base64 import b64decode, b64encode
from pathlib import Path
import aubio
import numpy as np
import polars as pl
from loguru import logger
HOP_SIZE = 512
WIN_SIZE = 1024
SAMPLE_RATE = 44100
MIN_BEATS = 8
# Waveform envelope dimensions.
# PWAV is stored as 200 col-pairs (height byte + intensity byte) = 400 bytes
# per Beat Link's WaveformPreview reader. Spec page says 400 cols × 1 byte;
# Beat Link reads 200 segments × 2 bytes — Beat Link is the production
# reference. PWV2 follows the same 2-bytes-per-col scheme (50 cols × 2B = 100B)
# but with 4-bit height (0-15) per the spec.
PWAV_COLS = 200 # paired-byte columns in PWAV (→ 400 bytes data)
PWV2_COLS = 50 # paired-byte columns in PWV2 (→ 100 bytes data)
# PWAV is 400 columns × 1 byte, each byte = (whiteness<<5)|height (5-bit height,
# 3-bit whiteness) — confirmed by byte-diffing a real export (the earlier
# "200 col-pairs" reading was wrong: it wrote height/intensity pairs the player
# decoded as garbled half-columns with no whiteness). PWV2 = 100 columns × 1
# byte, 4-bit height in the low nibble.
PWAV_COLS = 400 # PWAV: 400 single bytes (whiteness<<5 | 5-bit height)
PWV2_COLS = 100 # PWV2: 100 single bytes (4-bit height)
SCROLL_FPS = 150 # PWV3 / PWV5 / PWV7 = ~150 cols/sec
PWAV_HEIGHT_LEVELS = 31 # 5-bit (0-31), bits 0-4
PWV2_HEIGHT_LEVELS = 15 # 4-bit (0-15), bits 0-3
PWAV_LEVELS = 31 # legacy alias for callers
def _stream_pcm(path: Path):
"""Yield successive hop-sized float32 numpy arrays of mono PCM.
def _mixxx_analyze(path: Path):
"""Run mixxx-analyzer (Mixxx's BPM / beat-grid / key algorithms) on `path`.
Pads the final chunk with zeros so every yield is exactly HOP_SIZE samples
(aubio.tempo requires fixed-size inputs).
"""
proc = subprocess.Popen(
["ffmpeg", "-nostdin", "-v", "error",
"-i", str(path),
"-f", "f32le", "-ac", "1", "-ar", str(SAMPLE_RATE), "-"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
Returns the `AnalysisResult` (fields incl. `bpm` and `beatgrid` — beat
positions in seconds) or None on failure."""
try:
assert proc.stdout is not None
bytes_per_chunk = HOP_SIZE * 4 # float32
while True:
buf = proc.stdout.read(bytes_per_chunk)
if not buf:
break
arr = np.frombuffer(buf, dtype=np.float32)
if len(arr) < HOP_SIZE:
pad = np.zeros(HOP_SIZE - len(arr), dtype=np.float32)
arr = np.concatenate([arr, pad])
yield arr
finally:
if proc.poll() is None:
proc.terminate()
proc.wait()
import mixxx_analyzer
except ImportError:
logger.warning("mixxx-analyzer not installed; BPM/beat detection disabled")
return None
try:
return mixxx_analyzer.analyze(str(path))
except Exception as e:
logger.warning("mixxx-analyzer failed for {}: {}", path.name, e)
return None
def _beatgrid_to_pqtz(beatgrid: list[float], bpm: float) -> list[tuple[int, int, int]]:
"""Project a continuous PQTZ grid from mixxx's tempo + phase.
Returns `[(beat_number 1-4, tempo×100, time_ms), ...]`. mixxx gives an
accurate BPM and beat phase, but its raw beat list can have GAPS in quiet
sections (e.g. 907 detected beats where a continuous 160 BPM grid has
~1010). rekordbox/CDJs expect a gap-free grid — beat-sync and quantise lock
to it — so we anchor on mixxx's first beat, walk back to t≥0 to cover the
intro, and emit one beat every 60/bpm seconds through to the last detected
beat. beat_number cycles 1→4 from the anchor (no downbeat info)."""
if not beatgrid or bpm <= 0:
return []
interval_ms = 60000.0 / bpm
first_ms = beatgrid[0] * 1000.0
last_ms = beatgrid[-1] * 1000.0
n_before = int(first_ms / interval_ms) # preserve phase back to t≥0
anchor = first_ms - n_before * interval_ms
tempo_x100 = round(bpm * 100)
grid: list[tuple[int, int, int]] = []
i = 0
t = anchor
while t <= last_ms + interval_ms:
grid.append(((i % 4) + 1, tempo_x100, int(round(t))))
i += 1
t = anchor + i * interval_ms
return grid
def detect_bpm(path: Path) -> float | None:
"""Return the analytical BPM of `path`, or None on failure.
Uses median-of-intervals (via polars), which is robust against
aubio's "live" running-average swings on transient sections.
"""
if not shutil.which("ffmpeg"):
logger.warning("ffmpeg not on PATH; BPM detection disabled")
return None
tempo = aubio.tempo("default", WIN_SIZE, HOP_SIZE, SAMPLE_RATE)
beats: list[float] = []
try:
for chunk in _stream_pcm(path):
if tempo(chunk):
beats.append(tempo.get_last_s())
except Exception as e:
logger.warning("BPM detection failed for {}: {}", path.name, e)
return None
if len(beats) < MIN_BEATS:
return None
intervals = pl.Series(
[beats[i + 1] - beats[i] for i in range(len(beats) - 1)]
)
median_interval = intervals.median()
if not median_interval or median_interval <= 0:
return None
return 60.0 / median_interval
"""Return the BPM of `path` via mixxx-analyzer (Mixxx's algorithm), or None."""
res = _mixxx_analyze(path)
return float(res.bpm) if res is not None and res.bpm else None
def _decode_pcm(path: Path) -> np.ndarray:
@@ -123,67 +108,84 @@ def _decode_pcm(path: Path) -> np.ndarray:
return np.frombuffer(proc.stdout, dtype=np.float32)
# Real rekordbox waveforms have a sharp noise gate and an aggressive
# S-curve: silence reads as actual zero, peaks pin to max, mids cluster
# at the high end. Linear peak-normalisation produces mushy mid-range
# values that look "always loud, never silent" on the deck. These two
# numbers were tuned against the data/q1-rekordbox/ reference export.
SILENCE_GATE_PCT = 0.02 # < 2% of peak → 0 (matches ~13,600 silent cols on Q1)
QUANT_GAMMA = 0.55 # > sub-linear curve, pushes mid-values toward max
# Height: gate near-silence to 0, then a linear map to the level range
# (QUANT_GAMMA=1.0). Heights use per-track-normalised RMS loudness, which on the
# reference export spreads low (scroll ~5.7/31, overview ~13/31) with no
# saturation. Colour/whiteness use a sub-linear lift (COLOR_GAMMA) so quiet
# bands still read vivid mid-range instead of near-zero.
SILENCE_GATE_PCT = 0.02 # < 2% of peak → 0
QUANT_GAMMA = 1.0 # HEIGHT: linear; do NOT lift mids toward max
COLOR_GAMMA = 0.4 # COLOUR/whiteness: sub-linear lift
def _quantise(peaks: np.ndarray, levels: int) -> np.ndarray:
"""Gate + gamma + quantise. Returns uint8 array of length len(peaks)."""
peak_max = float(peaks.max()) if peaks.size else 0.0
if peak_max <= 0:
def _quantise(peaks: np.ndarray, levels: int, peak_max: float | None = None,
gamma: float | None = None) -> np.ndarray:
"""Gate + (optional gamma) + quantise to 0..levels uint8.
`peak_max` lets callers pass a SHARED normaliser (e.g. across bands) so the
relative energy between channels is preserved; defaults to this array's own
max. `gamma` overrides QUANT_GAMMA (use COLOR_GAMMA to lift colour)."""
g = QUANT_GAMMA if gamma is None else gamma
pm = float(peaks.max()) if peak_max is None else peak_max
if pm <= 0:
return np.zeros(peaks.size, dtype=np.uint8)
gated = np.where(peaks < peak_max * SILENCE_GATE_PCT, 0.0, peaks)
normalised = gated / peak_max
curved = np.power(normalised, QUANT_GAMMA, where=normalised > 0)
gated = np.where(peaks < pm * SILENCE_GATE_PCT, 0.0, peaks)
normalised = np.clip(gated / pm, 0.0, 1.0)
curved = normalised if g == 1.0 else np.power(normalised, g)
return np.clip(np.round(curved * levels), 0, levels).astype(np.uint8)
def _bin_amplitudes(samples: np.ndarray, n_bins: int, levels: int = PWAV_LEVELS) -> bytes:
"""Single-byte-per-bin amplitude (used for PWV3 detail waveform)."""
if n_bins <= 0 or len(samples) == 0:
return bytes(n_bins)
chunks = np.array_split(np.abs(samples), n_bins)
peaks = np.array([float(c.max()) if len(c) else 0.0 for c in chunks])
return _quantise(peaks, levels).tobytes()
def _col_rms(values: np.ndarray, n_cols: int) -> np.ndarray:
"""Per-column RMS over n_cols evenly-split chunks (loudness, not peak)."""
chunks = np.array_split(values.astype(np.float64), n_cols)
return np.array([float(np.sqrt(np.mean(c * c))) if len(c) else 0.0 for c in chunks])
def _bin_amplitudes_paired(samples: np.ndarray, n_cols: int, height_levels: int = PWAV_HEIGHT_LEVELS) -> bytes:
"""Pair-byte-per-column quantisation for PWAV / PWV2. Returns
`n_cols * 2` bytes — height in even bytes (bits 0..h_bits-1),
intensity in odd bytes (bits 0-2).
def _env_peaks(env: np.ndarray, n_cols: int) -> np.ndarray:
chunks = np.array_split(env, n_cols)
return np.array([float(c.max()) if len(c) else 0.0 for c in chunks])
Same gate + gamma curve as `_quantise` so silence reads as silence
and peaks pin to the channel max.
"""
def _packed_mono(samples: np.ndarray, hi_env: np.ndarray, n_cols: int,
height_levels: int = PWAV_HEIGHT_LEVELS) -> bytes:
"""One byte per column = (whiteness << 5) | height — the real PWAV/PWV3
on-disk format. height = RMS loudness (linear); whiteness = lifted treble
(high-band) peak, so hi-hats/snares read white and bass/tonal reads blue."""
if n_cols <= 0 or samples.size == 0:
return bytes(n_cols * 2)
abs_s = np.abs(samples)
chunks = np.array_split(abs_s, n_cols)
peaks = np.array([float(c.max()) if len(c) else 0.0 for c in chunks])
rms = np.array([float(np.sqrt(np.mean(c.astype(np.float64) ** 2))) if len(c) else 0.0
for c in chunks])
heights = _quantise(peaks, height_levels)
# Transients (peak > 1.5 × rms) get intensity 7. Heights already
# gated to 0 in silence — pair them: zero height → zero intensity.
intensities = np.where(
(heights > 0) & (peaks > 1.5 * np.maximum(rms, 1e-9)),
7, 0,
).astype(np.uint8)
out = np.empty(n_cols * 2, dtype=np.uint8)
out[0::2] = heights
out[1::2] = intensities
return out.tobytes()
return bytes(n_cols)
height = _quantise(_col_rms(np.abs(samples), n_cols), height_levels)
white = _quantise(_env_peaks(hi_env, n_cols), 7, gamma=COLOR_GAMMA) # lifted
white = np.where(height > 0, white, 0).astype(np.uint8) # silence → no white
return ((white << 5) | height).astype(np.uint8).tobytes()
def _packed_tiny(samples: np.ndarray, n_cols: int) -> bytes:
"""PWV2: one byte per column, 4-bit height (0-15) in the low nibble."""
if n_cols <= 0 or samples.size == 0:
return bytes(n_cols)
return _quantise(_col_rms(np.abs(samples), n_cols), PWV2_HEIGHT_LEVELS).astype(np.uint8).tobytes()
def _bin_bands(lo_env: np.ndarray, mid_env: np.ndarray, hi_env: np.ndarray,
n_cols: int, shared: bool, levels: int = 255) -> tuple[bytes, bytes, bytes]:
"""Bin three band envelopes (per-column PEAK) into n_cols, lifted by
COLOR_GAMMA. `shared=True` normalises all three by one SHARED peak → natural
bass→treble rolloff (PWV4 overview). `shared=False` normalises each band to
its OWN peak → vivid balanced colour where sustained bands read strong
(PWV5 colour scroll)."""
lp, mp, hp = _env_peaks(lo_env, n_cols), _env_peaks(mid_env, n_cols), _env_peaks(hi_env, n_cols)
if shared:
s = max(float(lp.max()), float(mp.max()), float(hp.max()), 1e-9)
nlo = nmd = nhi = s
else:
nlo, nmd, nhi = None, None, None
return (_quantise(lp, levels, nlo, gamma=COLOR_GAMMA).tobytes(),
_quantise(mp, levels, nmd, gamma=COLOR_GAMMA).tobytes(),
_quantise(hp, levels, nhi, gamma=COLOR_GAMMA).tobytes())
def _band_envelopes(samples: np.ndarray, sr: int) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Return per-sample low/mid/high envelopes (after band-pass + abs).
Uses a 4-pole Butterworth band-pass per band; scipy.signal.sosfilt
streams in a single pass so the whole-track envelope fits in RAM."""
"""Per-sample low/mid/high envelopes (4-pole Butterworth band-pass + abs)."""
from scipy.signal import butter, sosfilt
def env(lo: int, hi: int) -> np.ndarray:
sos = butter(4, [lo, hi], btype="band", fs=sr, output="sos")
@@ -191,96 +193,14 @@ def _band_envelopes(samples: np.ndarray, sr: int) -> tuple[np.ndarray, np.ndarra
return env(*BAND_LOW), env(*BAND_MID), env(*BAND_HIGH)
def _bin_to_bytes(values: np.ndarray, n_bins: int, levels: int = 255) -> bytes:
"""Bin `values` into `n_bins`, take per-bin max, normalise to `levels`
via the shared gate+gamma curve (matches what real rekordbox writes:
silence at zero, mids pulled toward max)."""
if n_bins <= 0 or values.size == 0:
return bytes(n_bins)
chunks = np.array_split(values, n_bins)
peaks = np.array([float(c.max()) if len(c) else 0.0 for c in chunks])
return _quantise(peaks, levels).tobytes()
CACHE_VERSION = 3 # mixxx-analyzer beat grid + RMS/lift waveform packing
def extract_waveform_envelope(path: Path) -> dict[str, bytes]:
"""Return the two byte streams a PWAV + PWV3 pair needs.
Output:
{
"preview": bytes of length PREVIEW_COLS (always 400),
"scroll": bytes of length ~ duration * SCROLL_FPS,
}
On decode failure both streams are empty.
"""
samples = _decode_pcm(path)
if samples.size == 0:
return {"preview": b"", "scroll": b""}
duration = samples.size / SAMPLE_RATE
scroll_cols = max(1, int(round(duration * SCROLL_FPS)))
return {
"preview": _bin_amplitudes(samples, PREVIEW_COLS),
"scroll": _bin_amplitudes(samples, scroll_cols),
}
def _detect_beats_in_samples(samples: np.ndarray) -> list[float]:
"""Run aubio's onset detector across a pre-loaded mono PCM buffer.
Returns the timestamps (seconds) of every detected beat.
"""
tempo = aubio.tempo("default", WIN_SIZE, HOP_SIZE, SAMPLE_RATE)
beats: list[float] = []
for i in range(0, len(samples), HOP_SIZE):
chunk = samples[i:i + HOP_SIZE]
if len(chunk) < HOP_SIZE:
chunk = np.concatenate(
[chunk, np.zeros(HOP_SIZE - len(chunk), dtype=np.float32)]
)
if tempo(chunk):
beats.append(tempo.get_last_s())
return beats
def _project_beat_grid(beats: list[float], bpm: float, duration_s: float) -> list[tuple[int, int, int]]:
"""Project a rigid `bpm` beat grid using the first detected beat as anchor.
Returns `[(beat_number 1-4, tempo*100, time_ms), ...]` covering t=0 to
`duration_s`. Rigid grids are what CDJs prefer over noisy per-beat
timing — the strongest detected beat fixes the phase and we project
forward + backward at the canonical tempo.
"""
if not beats or bpm <= 0:
return []
interval = 60.0 / bpm
first = beats[0]
# Walk back from the first detected beat to t≥0 so the grid covers the intro.
n_before = int(first / interval)
anchor = first - n_before * interval
if anchor < 0:
anchor += interval
n_before -= 1
tempo_x100 = round(bpm * 100)
grid: list[tuple[int, int, int]] = []
t = anchor
i = 0
while t <= duration_s + 1.0:
time_ms = int(round(t * 1000))
beat_number = (i % 4) + 1
grid.append((beat_number, tempo_x100, time_ms))
i += 1
t = anchor + i * interval
return grid
CACHE_VERSION = 2
# 3-band waveform split. Frequencies tuned to roughly match what
# rekordbox shows on the CDJ-3000 strip: bass (kick + sub), mid (snare,
# vocals, chords), high (hats, cymbals).
# 3-band waveform split: bass (kick + sub), mid (snare, vocals, chords),
# high (hats, cymbals).
BAND_LOW = (20, 200)
BAND_MID = (200, 2000)
BAND_HIGH = (2000, 15000)
PWV6_COLS = 1200 # 3-band preview width (fixed by spec)
PWV6_COLS = 1200 # 3-band / colour preview width (fixed by spec)
def _sanitise_isrc(isrc: str) -> str:
@@ -366,42 +286,41 @@ def analyze_audio(path: Path) -> dict:
"scroll_high": bytes,
}
"""
samples = _decode_pcm(path)
out: dict = {
"bpm": None, "beats": [],
"preview": b"", "tiny": b"", "scroll": b"",
"preview_low": b"", "preview_mid": b"", "preview_high": b"",
"scroll_low": b"", "scroll_mid": b"", "scroll_high": b"",
}
# BPM + beat grid from mixxx-analyzer (independent of the waveform decode,
# so a track still gets its grid even if ffmpeg waveform extraction fails).
res = _mixxx_analyze(path)
if res is not None and res.bpm and res.beatgrid:
out["bpm"] = float(res.bpm)
out["beats"] = _beatgrid_to_pqtz(list(res.beatgrid), out["bpm"])
samples = _decode_pcm(path)
if samples.size == 0:
return out
beat_times = _detect_beats_in_samples(samples)
if len(beat_times) >= MIN_BEATS:
intervals = pl.Series(
[beat_times[i + 1] - beat_times[i] for i in range(len(beat_times) - 1)]
)
median_interval = intervals.median()
if median_interval and median_interval > 0:
out["bpm"] = 60.0 / median_interval
duration_s = samples.size / SAMPLE_RATE
out["beats"] = _project_beat_grid(beat_times, out["bpm"], duration_s)
duration = samples.size / SAMPLE_RATE
scroll_cols = max(1, int(round(duration * SCROLL_FPS)))
# PWAV / PWV2 use paired bytes (height + intensity); PWV3 is 1B/col.
out["preview"] = _bin_amplitudes_paired(samples, PWAV_COLS, PWAV_HEIGHT_LEVELS)
out["tiny"] = _bin_amplitudes_paired(samples, PWV2_COLS, PWV2_HEIGHT_LEVELS)
out["scroll"] = _bin_amplitudes(samples, scroll_cols)
# 3-band envelopes for PWV4/5/6/7 — band-pass with scipy, then bin
# to (a) 1200 cols for PWV4/PWV6 previews and (b) the same scroll
# column count as PWV3 for PWV5/PWV7.
# Band envelopes first — the high band feeds the mono whiteness channel,
# and all three feed the colour sections.
lo_env, mid_env, hi_env = _band_envelopes(samples, SAMPLE_RATE)
out["preview_low"] = _bin_to_bytes(lo_env, PWV6_COLS)
out["preview_mid"] = _bin_to_bytes(mid_env, PWV6_COLS)
out["preview_high"] = _bin_to_bytes(hi_env, PWV6_COLS)
out["scroll_low"] = _bin_to_bytes(lo_env, scroll_cols)
out["scroll_mid"] = _bin_to_bytes(mid_env, scroll_cols)
out["scroll_high"] = _bin_to_bytes(hi_env, scroll_cols)
# Mono overview/scroll: one byte/col = (whiteness<<5)|height (PWAV/PWV3),
# 4-bit height for the tiny preview (PWV2).
out["preview"] = _packed_mono(samples, hi_env, PWAV_COLS, PWAV_HEIGHT_LEVELS) # PWAV 400B
out["tiny"] = _packed_tiny(samples, PWV2_COLS) # PWV2 100B
out["scroll"] = _packed_mono(samples, hi_env, scroll_cols, PWAV_HEIGHT_LEVELS) # PWV3
# 3-band colour data: SHARED-normalised preview (energy rolloff) for PWV4/
# PWV6, per-band-normalised scroll (vivid) for PWV5/PWV7.
out["preview_low"], out["preview_mid"], out["preview_high"] = _bin_bands(
lo_env, mid_env, hi_env, PWV6_COLS, shared=True)
out["scroll_low"], out["scroll_mid"], out["scroll_high"] = _bin_bands(
lo_env, mid_env, hi_env, scroll_cols, shared=False)
return out
maple-py/src/maple/anlz.py
+39 -58
View File
@@ -83,13 +83,12 @@ def _section_pqtz(beats: list[tuple[int, int, int]]) -> bytes:
def _section_pwav(preview: bytes) -> bytes:
"""PWAV: WavePreviewTag — 400-byte touch-strip preview.
Per Beat Link's WaveformPreview reader, PWAV stores 200 col-pairs:
byte 2N : bits 0-4 = height (0-31)
byte 2N+1 : bits 0-2 = intensity (0-7); ≥5 renders bright cyan
`preview` is expected to already be in that format (`maple.analyze.
_bin_amplitudes_paired`).
"""PWAV: WavePreviewTag — 400-byte touch-strip preview, **400 columns ×
1 byte**, each byte = (whiteness << 5) | height (3-bit whiteness, 5-bit
height). `preview` is produced already in that form by
`maple.analyze._packed_mono`. (The earlier "200 height/intensity col-pairs"
reading was wrong — the player decoded those as garbled half-columns with
no whiteness.)
"""
fourcc = b"PWAV"
len_header = 20
@@ -98,8 +97,9 @@ def _section_pwav(preview: bytes) -> bytes:
def _section_pwv2(tiny: bytes) -> bytes:
"""PWV2: 100-byte tiny preview. Same 2-bytes-per-col layout as PWAV
but with **4-bit** height (bits 0-3, 0-15 range) per spec."""
"""PWV2: 100-byte tiny preview — 100 columns × 1 byte, 4-bit height
(bits 0-3, 0-15) in the low nibble. `tiny` is produced by
`maple.analyze._packed_tiny`."""
fourcc = b"PWV2"
len_header = 20
len_tag = len_header + len(tiny)
@@ -149,76 +149,56 @@ def _section_pco2(cue_list_type: int) -> bytes:
def _section_pwv4(low: bytes, mid: bytes, high: bytes) -> bytes:
"""PWV4: WaveColorPreviewTag — fixed-width colour preview, 1200 cols ×
6 bytes/col. Per beatlink/deepsymmetry docs, the 6 bytes per column
interleave 6 "stripes" rendered top-to-bottom:
"""PWV4: WaveColorPreviewTag — colour overview, 1200 cols × 6 bytes/col:
byte 0 : whiteness 1 (transient highlight)
byte 1 : whiteness 2 (transient highlight)
byte 2 : energy lower-half of the frequency range
byte 3 : energy lower-third
byte 4 : energy middle
byte 5 : energy top
byte 0/1 : whiteness/brightness (= overall)
byte 2 : overall (full-spectrum) energy
byte 3 : bass (low band)
byte 4 : mid (mid band)
byte 5 : treble (high band)
The exact whiteness derivation isn't documented. We synthesise it as
the column's overall amplitude (peak of all bands), which gives the
same shape as the existing PWAV "intense column" highlight.
"""
`low/mid/high` arrive SHARED-normalised (0-255); `overall = max(bands)`."""
fourcc = b"PWV4"
len_header = 24
n = min(len(low), len(mid), len(high))
# Per-column max across bands → whiteness stand-in
lo = bytearray(low[:n])
mid_arr = bytearray(mid[:n])
hi = bytearray(high[:n])
overall = bytearray(max(lo[i], mid_arr[i], hi[i]) for i in range(n))
lo, md, hi = bytearray(low[:n]), bytearray(mid[:n]), bytearray(high[:n])
body = bytearray(n * 6)
for i in range(n):
body[i * 6 + 0] = overall[i] # whiteness 1
body[i * 6 + 1] = overall[i] # whiteness 2 (same — we don't have a separate channel)
body[i * 6 + 2] = lo[i] # bottom-half energy ≈ low band
body[i * 6 + 3] = lo[i] # bottom-third energy ≈ low band (closer to bottom)
body[i * 6 + 4] = mid_arr[i] # middle energy ≈ mid band
body[i * 6 + 5] = hi[i] # top energy ≈ high band
len_entry_bytes = 6
len_entries = n
overall = max(lo[i], md[i], hi[i])
body[i * 6 + 0] = overall
body[i * 6 + 1] = overall
body[i * 6 + 2] = overall
body[i * 6 + 3] = lo[i] # bass
body[i * 6 + 4] = md[i] # mid
body[i * 6 + 5] = hi[i] # treble
len_tag = len_header + len(body)
# Real export has `unknown = 0` here.
return fourcc + pack(">IIIII", len_header, len_tag, len_entry_bytes, len_entries, 0x00000000) + bytes(body)
return fourcc + pack(">IIIII", len_header, len_tag, 6, n, 0x00000000) + bytes(body)
def _section_pwv5(low: bytes, mid: bytes, high: bytes) -> bytes:
"""PWV5: WaveColorScrollTag — colour scroll, 2 bytes/col packed as
def _section_pwv5(scroll: bytes, low: bytes, mid: bytes, high: bytes) -> bytes:
"""PWV5: WaveColorScrollTag — colour scroll, u16 BE per col:
fedcba9876543210
RRRGGGBBBHHHHH--
(3 bits R, 3 bits G, 3 bits B, 5 bits height, 2 unused).
Map our (low, mid, high) band amplitudes (0-255) to (R, G, B) at
3-bit each. Height = peak across bands at 5-bit. Result: warm reds
for kick-heavy columns, blues for hat-heavy, etc. — matches how
rekordbox renders the NXS2 color waveform from low/mid/high content.
Colour **R=treble, G=mid, B=bass** (per spec). Bands arrive per-band
normalised 0-255 → 3-bit via >>5. Height is the broadband value from the
mono `scroll` byte's low 5 bits, so colour and mono scrolls share a height.
"""
fourcc = b"PWV5"
len_header = 24
n = min(len(low), len(mid), len(high))
n = min(len(scroll), len(low), len(mid), len(high))
body = bytearray(n * 2)
for i in range(n):
# Map 0-255 → 0-7 (3-bit per channel). Low band → red, mid → green,
# high → blue (matches the "RGB ≈ low/mid/high" convention rekordbox
# uses when reconstructing the NXS2 strip).
r = low[i] >> 5
g = mid[i] >> 5
b = high[i] >> 5
h_in = max(low[i], mid[i], high[i])
h = h_in >> 3 # 0-255 → 0-31
r = high[i] >> 5 # red = treble
g = mid[i] >> 5 # green = mid
b = low[i] >> 5 # blue = bass
h = scroll[i] & 0x1F # broadband height (0-31), shared with PWV3
word = (r & 0x7) << 13 | (g & 0x7) << 10 | (b & 0x7) << 7 | (h & 0x1F) << 2
body[i * 2] = (word >> 8) & 0xFF
body[i * 2 + 1] = word & 0xFF
len_entry_bytes = 2
len_entries = n
len_tag = len_header + len(body)
return fourcc + pack(">IIIII", len_header, len_tag, len_entry_bytes, len_entries, 0x00960305) + bytes(body)
return fourcc + pack(">IIIII", len_header, len_tag, 2, n, 0x00960305) + bytes(body)
def _section_pwv6(low: bytes, mid: bytes, high: bytes) -> bytes:
@@ -409,9 +389,10 @@ def build_ext(rb_path: str, analysis: dict | None = None) -> bytes:
# and (empirically) gates the CDJ's deck-side waveform/BPM display.
sections.append(_section_pqt2(analysis["beats"]))
if analysis:
scroll = analysis.get("scroll") or b""
s_lo, s_md, s_hi = analysis.get("scroll_low") or b"", analysis.get("scroll_mid") or b"", analysis.get("scroll_high") or b""
if s_lo and s_md and s_hi:
sections.append(_section_pwv5(s_lo, s_md, s_hi))
if scroll and s_lo and s_md and s_hi:
sections.append(_section_pwv5(scroll, s_lo, s_md, s_hi))
p_lo, p_md, p_hi = analysis.get("preview_low") or b"", analysis.get("preview_mid") or b"", analysis.get("preview_high") or b""
if p_lo and p_md and p_hi:
sections.append(_section_pwv4(p_lo, p_md, p_hi))
maple-py/src/maple/cli.py
+50 -24
View File
@@ -30,6 +30,36 @@ from maple.validate import TrackIssue, log_issues, validate_track
_GENRE_SPLIT_RE = re.compile(r"\s*[;,/]\s*")
def usbanlz_location(usb_path: str) -> tuple[int, int]:
"""rekordbox's ANLZ folder derivation from a USB-relative track path.
Returns ``(bucket, anlz_id)`` for ``/PIONEER/USBANLZ/P{bucket:03X}/{anlz_id:08X}/``.
`usb_path` is the forward-slashed, leading-slash USB-relative path that the
pdb stores as `file_path` (e.g. ``/Contents/Artist/Album/01 Title.mp3``).
CRITICAL: the CDJ/XDJ firmware recomputes this from `file_path` and reads a
track's analysis from the resulting folder, *ignoring* the pdb's
`analyze_path` string. So maple must write its ANLZ files here and nowhere
else, or the player shows no waveform/beatgrid. Verified 1002/1002 against a
genuine rekordbox export and confirmed on XDJ-RX2 hardware.
`anlz_id` is a polynomial hash of the path's UTF-16 code units mod 0x30D43
(200003 — hence ids never exceed 0x30D42); `bucket` is a fixed bit-extraction
from that id. (Algorithm: github.com/Radexito/DjManager protocol.md §2.)
"""
h = 0
for ch in usb_path:
h = (h * 0x34F5501D + ord(ch) * 0x93B6) & 0xFFFFFFFF
anlz_id = h % 0x30D43
bucket = (
((((((((((anlz_id >> 2) & 0x4000) | (anlz_id & 0x2000)) >> 3)
| (anlz_id & 0x200)) >> 1) | (anlz_id & 0xC0)) >> 3)
| (anlz_id & 0x4)) >> 1) | (anlz_id & 0x1)
)
return bucket, anlz_id
def split_genre_tags(raw: str) -> list[str]:
"""Tokenise a possibly multi-value genre string into normalised parts.
@@ -71,6 +101,13 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple
genre_to_tracks: dict[str, list[int]] = defaultdict(list)
track_by_src: dict[Path, int] = {}
artwork_cache: dict[str, int] = {}
# USBANLZ folder occupancy. Two distinct paths can hash to the same
# P{bucket}/{id} folder (a hash collision; also happens for genuine
# duplicate paths). rekordbox keeps both by bumping the filename to
# ANLZ0001.DAT, ANLZ0002.DAT, … inside the shared folder (verified: a
# real export had `Celestica` + `Soak It Up` co-located at P049/00030991).
# Mirror that so colliding tracks don't clobber each other's analysis.
anlz_folder_seen: dict[str, int] = defaultdict(int)
written: set[Path] = set()
audio_files = scan_audio(input_dir)
@@ -275,30 +312,19 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple
genre_id = primary_genre_id
track_id = len(db.tracks) + 1
# Reference exports always populate analyze_path even when the
# ANLZ files are real; tracks with an empty analyze_path appear
# to be hidden from the player's library. Synthesise a stable
# path keyed off the track id even if we don't write the .DAT.
import zlib
h = zlib.crc32(file_path.encode()) & 0xFFFFFFFF
# rekordbox writes 3-digit zero-padded buckets (e.g. P020 not P20) +
# 8-hex-digit ids. The bucket value range observed in real exports
# is 0..127 (P000..P07F) — wider masks here haven't shown up in any
# reference export and may push the CDJ outside whatever lookup
# range its firmware accepts. The 8-hex id is an opaque master.db
# value; we derive a stable-but-small value from crc32(path) capped
# to 20 bits so it lives in the same numeric range as real-export
# ids (Q1's id is 0x0002E10A ≈ 190K).
bucket = f"P{(h >> 20) & 0x7F:03X}"
analyze_id = f"{h & 0xFFFFF:08X}"
# DEBUG: force Q1.1 to use the exact ANLZ path the real rekordbox
# export uses — settles whether the CDJ's path-template parser
# cares about the bucket/id pattern.
if file_path == "/Contents/Basic Channel/Q 1.1/02 Q1.1 II.mp3":
bucket = "P020"
analyze_id = "0002E10A"
logger.warning("Q1.1 path forced to real rekordbox bucket: {}/{}", bucket, analyze_id)
analyze_path = f"/PIONEER/USBANLZ/{bucket}/{analyze_id}/ANLZ0000.DAT"
# The USBANLZ folder is derived from the USB-relative `file_path` by
# rekordbox's bespoke path hash (see `usbanlz_location`). The CDJ/XDJ
# firmware RE-DERIVES this same path from `file_path` and reads the
# analysis from there — it IGNORES the pdb's `analyze_path` string —
# so the folder MUST match rekordbox exactly or the player silently
# shows no waveform/beatgrid. Verified 1002/1002 vs a real export and
# on XDJ-RX2 hardware. (Earlier crc32 scheme matched 0/984 → grids
# missing.)
bucket_n, anlz_id = usbanlz_location(file_path)
anlz_folder = f"/PIONEER/USBANLZ/P{bucket_n:03X}/{anlz_id:08X}"
anlz_index = anlz_folder_seen[anlz_folder] # 0, then 1, 2, … on collision
anlz_folder_seen[anlz_folder] += 1
analyze_path = f"{anlz_folder}/ANLZ{anlz_index:04d}.DAT"
anlz_abs = (usb_root / analyze_path.lstrip("/")).resolve()
# `_analysis/` sits at the input collection root; ISRC-keyed cache
# populated during `just collection`. write_anlz reads it before
maple-py/uv.lock
Generated
+12 -41
View File
@@ -90,15 +90,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "aubio"
version = "0.4.9"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "numpy" },
]
sdist = { url = "https://files.pythonhosted.org/packages/cd/80/302d89240603e5347c7f8026c8b02c59f8dfaec66c91a743d82de7c86006/aubio-0.4.9.tar.gz", hash = "sha256:df1244f6c4cf5bea382c8c2d35aa43bc31f4cf631fe325ae3992c219546a4202", size = 479008, upload-time = "2019-02-08T11:21:02.438Z" }
[[package]]
name = "bidict"
version = "0.23.1"
@@ -252,12 +243,11 @@ version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
{ name = "aubio" },
{ name = "kaitaistruct" },
{ name = "loguru" },
{ name = "mixxx-analyzer" },
{ name = "mutagen" },
{ name = "pillow" },
{ name = "polars" },
{ name = "pyrekordbox" },
{ name = "scipy" },
]
@@ -270,12 +260,11 @@ dev = [
[package.metadata]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.10" },
{ name = "aubio", specifier = ">=0.4.9" },
{ name = "kaitaistruct", specifier = ">=0.11" },
{ name = "loguru", specifier = ">=0.7" },
{ name = "mixxx-analyzer", specifier = ">=0.1.2" },
{ name = "mutagen", specifier = ">=1.47.0" },
{ name = "pillow", specifier = ">=12.2.0" },
{ name = "polars", specifier = ">=1.0" },
{ name = "pyrekordbox", specifier = ">=0.4.4" },
{ name = "scipy", specifier = ">=1.10" },
]
@@ -283,6 +272,16 @@ requires-dist = [
[package.metadata.requires-dev]
dev = [{ name = "pytest", specifier = ">=9.0.3" }]
[[package]]
name = "mixxx-analyzer"
version = "0.1.2"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/53/b3/c4026eed48e7685a99212ede8a1ce22bd2442a7d677ec4b970a7718d5b23/mixxx_analyzer-0.1.2-py3-none-macosx_14_0_arm64.whl", hash = "sha256:cb509c06d630fad709f85dd07fbd21a029d337617de0866af74279ca0a1f596b", size = 13932797, upload-time = "2026-03-01T15:44:59.349Z" },
{ url = "https://files.pythonhosted.org/packages/a6/30/bf8afbb52d862d2976045b4f6d00456dcb6320bb5e449384a41a0ce40735/mixxx_analyzer-0.1.2-py3-none-manylinux_2_35_x86_64.whl", hash = "sha256:53d8ec329bae7a7714dfe1cd34cbcf3976ddb55fae0ed74c4872f47085a9d9cf", size = 56567723, upload-time = "2026-03-01T15:45:02.771Z" },
{ url = "https://files.pythonhosted.org/packages/2e/88/e22c9e7c3d862daec693de61c5f2843ce8e567cd4d5e866040c599b57e9d/mixxx_analyzer-0.1.2-py3-none-win_amd64.whl", hash = "sha256:cc6509acfc78f0f3e04ac9f0b1440e9cd31a3c389f888bbd91beae1f249c32b7", size = 44947103, upload-time = "2026-03-01T15:45:05.66Z" },
]
[[package]]
name = "multidict"
version = "6.7.1"
@@ -417,34 +416,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "polars"
version = "1.41.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "polars-runtime-32" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ff/f9/aeda46259b0669247a160315d2d51269de9504b9dd2f70acadbcb22f46b7/polars-1.41.2.tar.gz", hash = "sha256:256d6731162371b77f3f29a55eacb8c0fc740ddb1a293a01d2ef5b5393c5c708", size = 737996, upload-time = "2026-05-29T17:39:15.604Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/1f/22/28f62d24f7db56ac4343588f9362d49b7b4177e55ac47a466fe696b0099b/polars-1.41.2-py3-none-any.whl", hash = "sha256:23ce9a2910b6e3e8d4258770bf44aa17170958df7af6e85feedf4458a04d8d29", size = 833445, upload-time = "2026-05-29T17:37:05.576Z" },
]
[[package]]
name = "polars-runtime-32"
version = "1.41.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/56/54e3ea0e9b64f327179049e4742241cc6b1d3e8fa414b05a057dd26df367/polars_runtime_32-1.41.2.tar.gz", hash = "sha256:7af09ec1ab053da2c9669e8d15f809a4083a29be05db57111688b8051062af56", size = 2989474, upload-time = "2026-05-29T17:39:17.257Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d6/9b/fe72a3811c0357cdb06c67bdc7695fa1623ad47948fc523195f5ac31037f/polars_runtime_32-1.41.2-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:95a08346dac337357cdb825c8076df7d36da54c4caa59a5cb41d0a30691c5edd", size = 52265283, upload-time = "2026-05-29T17:37:09.407Z" },
{ url = "https://files.pythonhosted.org/packages/0a/93/fab9da803fd80d9e83ef88c20932f637a10bc611b20415fc322eec84bc44/polars_runtime_32-1.41.2-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:dedfaeec2c7f995298da7319dd9431d662e5dd1d0ec51b1459df4a0234ceff52", size = 46571222, upload-time = "2026-05-29T17:37:13.698Z" },
{ url = "https://files.pythonhosted.org/packages/c8/2a/8843f34a8ac57acd058a39b87b03b580dd352a490e9dae0415e02033bdd4/polars_runtime_32-1.41.2-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:18eea22c5cc34e27f8a60950458ad81e6a9ea75e89363ca1367e14e7e7f781fc", size = 50409372, upload-time = "2026-05-29T17:37:17.875Z" },
{ url = "https://files.pythonhosted.org/packages/6c/c6/92b352fe88cf51bd0a19fb99e1c0cbe46aa26c14dcf7995b89869cd932ae/polars_runtime_32-1.41.2-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2630540dfdfb0f36f9b04a07c7c2e3f50bf2ad384113263c1c812007ee9141e0", size = 56405484, upload-time = "2026-05-29T17:37:22.684Z" },
{ url = "https://files.pythonhosted.org/packages/74/c4/bae3174c3b02f6b441d2e58594387abcd509f67a098f682a83b195f08966/polars_runtime_32-1.41.2-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:20e969e08f9b137e233c04cc04de73d9795f89eb77d34854e40a025965a43763", size = 50603512, upload-time = "2026-05-29T17:37:27.422Z" },
{ url = "https://files.pythonhosted.org/packages/f4/ed/f2d26ae02d92c2689056838ed59e2a626326ad23c2831d58637d25f6c82a/polars_runtime_32-1.41.2-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:e7016a3deb641b64a31447abbbee0f34bd020a6a9ae34ee6b743837def15e2a4", size = 54328561, upload-time = "2026-05-29T17:37:32.587Z" },
{ url = "https://files.pythonhosted.org/packages/9b/c4/9c3831cc885dc7769e59abf8f583821a5fb4403fd0e4eba0ccc6d47a3d4b/polars_runtime_32-1.41.2-cp310-abi3-win_amd64.whl", hash = "sha256:1e5e5377c315e0dcafdfb2a31adc546abbaeb3f9cb1864e6536523d2af473265", size = 51978643, upload-time = "2026-05-29T17:37:37.443Z" },
{ url = "https://files.pythonhosted.org/packages/cd/c6/79e9f3f270270d7ed5575d92b7bfef49f01abd9275447161275b23b553a8/polars_runtime_32-1.41.2-cp310-abi3-win_arm64.whl", hash = "sha256:843d96f69d18eca53429c1198e58891db7f18111f83b9c419bb45ad9d73eaed5", size = 46006901, upload-time = "2026-05-29T17:37:42.522Z" },
]
[[package]]
name = "propcache"
version = "0.5.2"