diff --git a/maple-py/CLAUDE.md b/maple-py/CLAUDE.md new file mode 100644 index 0000000..0205596 --- /dev/null +++ b/maple-py/CLAUDE.md @@ -0,0 +1,55 @@ +# maple-py conventions + +Python conventions for the `maple-py/` package. These OVERRIDE default behavior; +follow them exactly. (General repo guidance is in the root `CLAUDE.md`; 4-space +indentation applies here too.) + +The package is pyright-checked (basic mode, Python 3.14). `Track` +(`src/maple/track.py`) and the `Analysis` family (`src/maple/analyze.py`) are the +reference examples for all of the below. + +## Imports at the top of the file + +All imports go at module top. No lazy/local imports inside functions. Never wrap +an import in `try/except ImportError` to degrade gracefully — if an integral +dependency (e.g. `mixxx_analyzer`, `scipy`) is missing, the process must **die at +startup** via the top-level import failing. The only acceptable indented import +is an `if TYPE_CHECKING:` block to break a circular type-only reference. + +## Fail loud — never swallow into None + +"Don't swallow into None" is a general principle at every level, including call +sites. An operation that fails must **let the exception propagate** — never catch +it and return `None` / an empty / degraded result. BPM/beat analysis is integral: +`_mixxx_analyze` raises rather than returning None, and every `Analysis.analyze` +call site (in `anlz.py`, `cli.py`, `collection/core.py`) lets failures abort the +run (including not catching around `ThreadPoolExecutor` `fut.result()`). + +Reserve `None` returns for *legitimate absence* (e.g. a cache miss in +`Analysis.load`), never for *failure*. The one tolerated catch is a genuinely +*optional* side-effect failing (e.g. writing the optimization cache via +`analysis.save()`), which logs and continues because it doesn't affect +correctness. + +## Specific types over generic ones + +- **Dataclasses, not bare dicts**, for structured values passed around. +- **`pathlib.Path`, not `str`**, for local filesystem paths (strings are fine for + serialization / rekordbox-USB paths that are inherently string data). +- **`typing.NamedTuple` (or a small dataclass), not bare tuples**, for fixed + records (e.g. `Beat(number, tempo_x100, time_ms)`, `Bands(low, mid, high)`). +- **No `None`-valued fields.** Absent values resolve at instantiation to a + concrete sentinel — `0.0` (0.0 = unknown), `""`, `[]`, `b""` — per `Track`. + `None` is acceptable only as a *factory return* signalling absence, never as a + stored field. + +Idiomatic dataclass shape: plain `@dataclass` (no frozen/slots), `@property` for +computed views, `@classmethod from_*()` factories, concrete field defaults. + +## Type every function parameter + +Always annotate every parameter, including nested/inner functions and +`@classmethod` factory args (`self`/`cls` exempt). No bare `def f(res):`. Use the +real type even from a third-party lib — e.g. `mixxx_analyzer.AnalysisResult` +(referenced directly; no extra import needed under +`from __future__ import annotations`). diff --git a/maple-py/src/maple/analyze.py b/maple-py/src/maple/analyze.py index 2e896e8..8cce00d 100644 --- a/maple-py/src/maple/analyze.py +++ b/maple-py/src/maple/analyze.py @@ -8,6 +8,12 @@ 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. + +Everything an analysis produces is held by the `Analysis` dataclass, composed of +`BeatGrid` (BPM + PQTZ beats), `MonoWaveforms` (PWAV/PWV2/PWV3) and +`BandWaveforms` (the 3-band colour streams). Build one with +`Analysis.analyze(path)` (or `Analysis.from_track(track)`); persist it with +`.save(cache_dir)` and read it back with `Analysis.load(cache_dir, isrc)`. """ from __future__ import annotations @@ -15,10 +21,17 @@ import json import shutil import subprocess from base64 import b64decode, b64encode +from dataclasses import dataclass, field from pathlib import Path +from typing import TYPE_CHECKING, NamedTuple +import mixxx_analyzer import numpy as np from loguru import logger +from scipy.signal import butter, sosfilt + +if TYPE_CHECKING: + from maple.track import Track SAMPLE_RATE = 44100 @@ -33,30 +46,56 @@ 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 +PWV6_COLS = 1200 # 3-band / colour preview width (fixed by spec) + +CACHE_VERSION = 9 # PWAV paired (2B/seg); treble-based brightness (not crest) -def _mixxx_analyze(path: Path): +class Band(NamedTuple): + """A band-pass frequency range, in Hz.""" + lo_hz: int + hi_hz: int + + +# 3-band waveform split: bass (kick + sub), mid (snare, vocals, chords), +# high (hats, cymbals). +BAND_LOW = Band(20, 200) +BAND_MID = Band(200, 2000) +BAND_HIGH = Band(2000, 15000) + + +class Beat(NamedTuple): + """One PQTZ beat-grid row. Unpacks straight into `struct.pack(">HHI", *beat)`.""" + number: int # 1..4 (cycles from the anchor; no downbeat info) + tempo_x100: int # BPM × 100 + time_ms: int # position in milliseconds + + +class Bands(NamedTuple): + """A low/mid/high triple of per-column waveform byte-streams.""" + low: bytes = b"" + mid: bytes = b"" + high: bytes = b"" + + +# --------------------------------------------------------------------------- # +# mixxx-analyzer + ffmpeg decode # +# --------------------------------------------------------------------------- # + + +def _mixxx_analyze(path: Path) -> mixxx_analyzer.AnalysisResult: """Run mixxx-analyzer (Mixxx's BPM / beat-grid / key algorithms) on `path`. Returns the `AnalysisResult` (fields incl. `bpm` and `beatgrid` — beat - positions in seconds) or None on failure.""" - try: - 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 + positions in seconds). BPM/beat detection is integral to an export, so a + failure here propagates rather than degrading to an analysis with no grid.""" + return mixxx_analyzer.analyze(str(path)) -def _beatgrid_to_pqtz(beatgrid: list[float], bpm: float) -> list[tuple[int, int, int]]: +def _project_pqtz(beatgrid: list[float], bpm: float) -> list[Beat]: """Project a continuous PQTZ grid from mixxx's tempo + phase. - Returns `[(beat_number 1-4, tempo×100, time_ms), ...]`. mixxx gives an + 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 @@ -71,22 +110,16 @@ def _beatgrid_to_pqtz(beatgrid: list[float], bpm: float) -> list[tuple[int, int, 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]] = [] + grid: list[Beat] = [] i = 0 t = anchor while t <= last_ms + interval_ms: - grid.append(((i % 4) + 1, tempo_x100, int(round(t)))) + grid.append(Beat((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 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: """Decode `path` to a 1D float32 numpy array of mono PCM at SAMPLE_RATE. @@ -108,6 +141,10 @@ def _decode_pcm(path: Path) -> np.ndarray: return np.frombuffer(proc.stdout, dtype=np.float32) +# --------------------------------------------------------------------------- # +# Quantisation + per-column binning (the waveform DSP) # +# --------------------------------------------------------------------------- # + # 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 @@ -218,7 +255,7 @@ def _packed_tiny(samples: np.ndarray, n_cols: int) -> bytes: 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]: + n_cols: int, shared: bool, levels: int = 255) -> Bands: """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 @@ -230,28 +267,24 @@ def _bin_bands(lo_env: np.ndarray, mid_env: np.ndarray, hi_env: np.ndarray, 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()) + return Bands( + _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]: """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") + def env(band: Band) -> np.ndarray: + sos = butter(4, [band.lo_hz, band.hi_hz], btype="band", fs=sr, output="sos") return np.abs(sosfilt(sos, samples)).astype(np.float32) - return env(*BAND_LOW), env(*BAND_MID), env(*BAND_HIGH) + return env(BAND_LOW), env(BAND_MID), env(BAND_HIGH) -CACHE_VERSION = 9 # PWAV paired (2B/seg); treble-based brightness (not crest) - -# 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 / colour preview width (fixed by spec) +# --------------------------------------------------------------------------- # +# ISRC-keyed cache paths # +# --------------------------------------------------------------------------- # def _sanitise_isrc(isrc: str) -> str: @@ -267,111 +300,199 @@ def cache_path_for(cache_dir: Path, isrc: str) -> Path | None: return cache_dir / f"{s}.json" -def save_analysis(cache_dir: Path, isrc: str, analysis: dict) -> Path | None: - """Persist `analysis` for `isrc` under `cache_dir`. No-op if isrc empty.""" - target = cache_path_for(cache_dir, isrc) - if target is None: - return None - target.parent.mkdir(parents=True, exist_ok=True) - payload = { - "version": CACHE_VERSION, - "isrc": _sanitise_isrc(isrc), - "bpm": analysis.get("bpm"), - "beats": analysis.get("beats", []), - "preview_b64": b64encode(analysis.get("preview") or b"").decode("ascii"), - "tiny_b64": b64encode(analysis.get("tiny") or b"").decode("ascii"), - "scroll_b64": b64encode(analysis.get("scroll") or b"").decode("ascii"), - # 3-band envelopes (cache v2). Kept separate from the mono streams - # so callers can ask for whichever they need without re-decoding. - "preview_low_b64": b64encode(analysis.get("preview_low") or b"").decode("ascii"), - "preview_mid_b64": b64encode(analysis.get("preview_mid") or b"").decode("ascii"), - "preview_high_b64": b64encode(analysis.get("preview_high") or b"").decode("ascii"), - "scroll_low_b64": b64encode(analysis.get("scroll_low") or b"").decode("ascii"), - "scroll_mid_b64": b64encode(analysis.get("scroll_mid") or b"").decode("ascii"), - "scroll_high_b64": b64encode(analysis.get("scroll_high") or b"").decode("ascii"), - } - target.write_text(json.dumps(payload, separators=(",", ":"))) - return target +# --------------------------------------------------------------------------- # +# Composed analysis dataclasses # +# --------------------------------------------------------------------------- # -def load_analysis(cache_dir: Path, isrc: str) -> dict | None: - """Return cached analysis for `isrc`, or None if absent/stale/unreadable.""" - target = cache_path_for(cache_dir, isrc) - if target is None or not target.exists(): - return None - try: - d = json.loads(target.read_text()) +@dataclass +class BeatGrid: + """BPM + the projected PQTZ beat grid. `bpm == 0.0` means unknown — there is + no `None` here, an absent grid is simply empty.""" + + bpm: float = 0.0 + beats: list[Beat] = field(default_factory=list) + + @property + def tempo_x100(self) -> int: + """BPM × 100, the integer tempo every PQTZ/PQT2 row carries.""" + return round(self.bpm * 100) + + @classmethod + def from_mixxx(cls, res: mixxx_analyzer.AnalysisResult) -> "BeatGrid": + """Build from a mixxx `AnalysisResult`. `_project_pqtz` tolerates an + empty beat list, so a result with no usable tempo/phase yields an empty + grid (bpm 0.0) rather than raising.""" + bpm = float(res.bpm) if res.bpm else 0.0 + beats = _project_pqtz(list(res.beatgrid), bpm) if res.beatgrid else [] + return cls(bpm=bpm, beats=beats) + + +@dataclass +class MonoWaveforms: + """The single-channel waveform streams. + + `preview` = PWAV (200 segments × 2 bytes), `tiny` = PWV2 (100 × 1 byte, + 4-bit height), `scroll` = PWV3 (~150 cols/sec, packed whiteness|height).""" + + preview: bytes = b"" + tiny: bytes = b"" + scroll: bytes = b"" + + @classmethod + def from_audio(cls, samples: np.ndarray, hi_env: np.ndarray, scroll_cols: int) -> "MonoWaveforms": + return cls( + preview=_packed_preview(samples, hi_env, PWAV_SEGMENTS), + tiny=_packed_tiny(samples, PWV2_COLS), + scroll=_packed_mono(samples, hi_env, scroll_cols, PWAV_HEIGHT_LEVELS), + ) + + +@dataclass +class BandWaveforms: + """The 3-band colour streams. `preview` is SHARED-normalised (energy rolloff, + feeds PWV4/PWV6); `scroll` is per-band-normalised (vivid, feeds PWV5/PWV7).""" + + preview: Bands = Bands() + scroll: Bands = Bands() + + @classmethod + def from_envelopes(cls, lo_env: np.ndarray, mid_env: np.ndarray, + hi_env: np.ndarray, scroll_cols: int) -> "BandWaveforms": + return cls( + preview=_bin_bands(lo_env, mid_env, hi_env, PWV6_COLS, shared=True), + scroll=_bin_bands(lo_env, mid_env, hi_env, scroll_cols, shared=False), + ) + + +@dataclass +class Analysis: + """A track's full audio analysis: BPM + beat grid and every waveform stream + the rekordbox ANLZ writer needs. Build with `Analysis.analyze(path)` or + `Analysis.from_track(track)`; cache with `.save()` / `Analysis.load()`.""" + + isrc: str = "" + grid: BeatGrid = field(default_factory=BeatGrid) + mono: MonoWaveforms = field(default_factory=MonoWaveforms) + bands: BandWaveforms = field(default_factory=BandWaveforms) + + # ---- views ----------------------------------------------------------- # + + @property + def bpm(self) -> float: + return self.grid.bpm + + @property + def beats(self) -> list[Beat]: + return self.grid.beats + + # ---- construction ---------------------------------------------------- # + + @classmethod + def analyze(cls, path: Path, *, isrc: str = "") -> "Analysis": + """One-pass decode + analysis of an audio file. + + BPM + beat grid come from mixxx-analyzer (independent of the waveform + decode, so a track still gets its grid even if ffmpeg fails). Always + returns a populated `Analysis` — empty waveforms when ffmpeg can't + decode the source.""" + grid = BeatGrid.from_mixxx(_mixxx_analyze(path)) + samples = _decode_pcm(path) + if samples.size == 0: + return cls(isrc=isrc, grid=grid) + scroll_cols = max(1, int(round(samples.size / SAMPLE_RATE * SCROLL_FPS))) + # 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) + return cls( + isrc=isrc, + grid=grid, + mono=MonoWaveforms.from_audio(samples, hi_env, scroll_cols), + bands=BandWaveforms.from_envelopes(lo_env, mid_env, hi_env, scroll_cols), + ) + + @classmethod + def from_track(cls, track: "Track") -> "Analysis": + """Seed `isrc` from the track and compute the rest from `track.path`. + + Returns a bare (isrc-only) `Analysis` when the track has no audio + downloaded yet (`track.path is None`).""" + if track.path is None: + return cls(isrc=track.isrc) + return cls.analyze(track.path, isrc=track.isrc) + + # ---- cache I/O ------------------------------------------------------- # + + def to_cache(self) -> dict: + """Serialise to the on-disk JSON payload (bytes → base64). Mirrors the + `CACHE_VERSION` v9 layout so existing caches stay readable.""" + return { + "version": CACHE_VERSION, + "isrc": _sanitise_isrc(self.isrc), + "bpm": self.grid.bpm or None, + "beats": [list(b) for b in self.grid.beats], + "preview_b64": b64encode(self.mono.preview).decode("ascii"), + "tiny_b64": b64encode(self.mono.tiny).decode("ascii"), + "scroll_b64": b64encode(self.mono.scroll).decode("ascii"), + # 3-band envelopes — kept separate from the mono streams so callers + # can ask for whichever they need without re-decoding. + "preview_low_b64": b64encode(self.bands.preview.low).decode("ascii"), + "preview_mid_b64": b64encode(self.bands.preview.mid).decode("ascii"), + "preview_high_b64": b64encode(self.bands.preview.high).decode("ascii"), + "scroll_low_b64": b64encode(self.bands.scroll.low).decode("ascii"), + "scroll_mid_b64": b64encode(self.bands.scroll.mid).decode("ascii"), + "scroll_high_b64": b64encode(self.bands.scroll.high).decode("ascii"), + } + + @classmethod + def from_cache(cls, d: dict) -> "Analysis | None": + """Rebuild from a `to_cache()` payload, or None if the version is stale.""" if d.get("version") != CACHE_VERSION: return None - return { - "bpm": d.get("bpm"), - "beats": [tuple(b) for b in d.get("beats", [])], - "preview": b64decode(d.get("preview_b64") or ""), - "tiny": b64decode(d.get("tiny_b64") or ""), - "scroll": b64decode(d.get("scroll_b64") or ""), - "preview_low": b64decode(d.get("preview_low_b64") or ""), - "preview_mid": b64decode(d.get("preview_mid_b64") or ""), - "preview_high": b64decode(d.get("preview_high_b64") or ""), - "scroll_low": b64decode(d.get("scroll_low_b64") or ""), - "scroll_mid": b64decode(d.get("scroll_mid_b64") or ""), - "scroll_high": b64decode(d.get("scroll_high_b64") or ""), - } - except (json.JSONDecodeError, OSError, ValueError) as e: - logger.warning("analysis cache read failed for {}: {}", isrc, e) - return None + return cls( + isrc=d.get("isrc") or "", + grid=BeatGrid( + bpm=float(d.get("bpm") or 0.0), + beats=[Beat(*b) for b in d.get("beats", [])], + ), + mono=MonoWaveforms( + preview=b64decode(d.get("preview_b64") or ""), + tiny=b64decode(d.get("tiny_b64") or ""), + scroll=b64decode(d.get("scroll_b64") or ""), + ), + bands=BandWaveforms( + preview=Bands( + b64decode(d.get("preview_low_b64") or ""), + b64decode(d.get("preview_mid_b64") or ""), + b64decode(d.get("preview_high_b64") or ""), + ), + scroll=Bands( + b64decode(d.get("scroll_low_b64") or ""), + b64decode(d.get("scroll_mid_b64") or ""), + b64decode(d.get("scroll_high_b64") or ""), + ), + ), + ) + def save(self, cache_dir: Path) -> Path | None: + """Persist this analysis under `cache_dir`, keyed by ISRC. No-op + (returns None) when the ISRC is empty.""" + target = cache_path_for(cache_dir, self.isrc) + if target is None: + return None + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(self.to_cache(), separators=(",", ":"))) + return target -def analyze_audio(path: Path) -> dict: - """One-pass decode + analysis. Returns: - { - "bpm": float | None, - "beats": list[(beat_n, tempo_x100, time_ms)], # for PQTZ - "preview": bytes, # 400 cols, 5-bit height (PWAV) - "scroll": bytes, # ~150 cols/sec, 5-bit (PWV3) - # 3-band envelopes: - "preview_low": bytes, # PWV6_COLS × 1B (0-255) - "preview_mid": bytes, # PWV6_COLS × 1B - "preview_high": bytes, # PWV6_COLS × 1B - "scroll_low": bytes, # ~150 cols/sec × 1B - "scroll_mid": bytes, - "scroll_high": bytes, - } - """ - 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 - - duration = samples.size / SAMPLE_RATE - scroll_cols = max(1, int(round(duration * SCROLL_FPS))) - - # 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) - - # Mono overview/scroll: one byte/col = (whiteness<<5)|height (PWAV/PWV3), - # 4-bit height for the tiny preview (PWV2). - out["preview"] = _packed_preview(samples, hi_env, PWAV_SEGMENTS) # PWAV 200 seg × 2B (paired) - out["tiny"] = _packed_tiny(samples, PWV2_COLS) # PWV2 100B - out["scroll"] = _packed_mono(samples, hi_env, scroll_cols, PWAV_HEIGHT_LEVELS) # PWV3 (unchanged) - - # 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 + @classmethod + def load(cls, cache_dir: Path, isrc: str) -> "Analysis | None": + """Return the cached analysis for `isrc`, or None if + absent / stale / unreadable.""" + target = cache_path_for(cache_dir, isrc) + if target is None or not target.exists(): + return None + try: + return cls.from_cache(json.loads(target.read_text())) + except (json.JSONDecodeError, OSError, ValueError) as e: + logger.warning("analysis cache read failed for {}: {}", isrc, e) + return None diff --git a/maple-py/src/maple/anlz.py b/maple-py/src/maple/anlz.py index cf07e78..481cbb8 100644 --- a/maple-py/src/maple/anlz.py +++ b/maple-py/src/maple/anlz.py @@ -29,7 +29,7 @@ from struct import pack from loguru import logger -from maple.analyze import analyze_audio, load_analysis, save_analysis +from maple.analyze import Analysis, Beat # Real rekordbox PMAI header carries three non-zero u4 markers in the bytes # after the standard PMAI/len_header/len_file triple. Verified by hex-dump @@ -64,7 +64,7 @@ def _section_pvbr() -> bytes: return fourcc + pack(">II", len_header, len_tag) + unnamed0 + body_var -def _section_pqtz(beats: list[tuple[int, int, int]]) -> bytes: +def _section_pqtz(beats: list[Beat]) -> bytes: """PQTZ: BeatGridTag — per-beat (beat_number, tempo×100, time_ms). The CDJ's loaded-deck BPM readout, tempo slider and sync engine all @@ -235,7 +235,7 @@ def _section_pwv7(low: bytes, mid: bytes, high: bytes) -> bytes: return fourcc + pack(">IIIII", len_header, len_tag, len_entry_bytes, len_entries, 0x00960000) + bytes(body) -def _section_pqt2(beats: list[tuple[int, int, int]]) -> bytes: +def _section_pqt2(beats: list[Beat]) -> bytes: """PQT2: extended beat grid (CDJ-3000+). Layout is only partially documented; the format isn't in the kaitai schema or the deepsymmetry spec page. The hex dump of a real Q1 PQT2 shows a 56-byte header @@ -347,29 +347,27 @@ def _pmai_header(file_len: int) -> bytes: return b"PMAI" + pack(">II", ANLZ_HEADER_LEN, file_len) + PMAI_HEADER_TRAILER -def build_dat(rb_path: str, analysis: dict | None) -> bytes: +def build_dat(rb_path: str, analysis: Analysis | None) -> bytes: """Build ANLZ0000.DAT — the file every CDJ reads on track load. Section order matches rekordbox's own output: PPTH, PVBR, PQTZ, PWAV, PWV2, PCOB(memory), PCOB(hot). """ sections = [_section_pth(rb_path), _section_pvbr()] - if analysis: - if analysis.get("beats"): - sections.append(_section_pqtz(analysis["beats"])) - preview = analysis.get("preview") or b"" - tiny = analysis.get("tiny") or b"" - if preview: - sections.append(_section_pwav(preview)) - if tiny: - sections.append(_section_pwv2(tiny)) + if analysis is not None: + if analysis.beats: + sections.append(_section_pqtz(analysis.beats)) + if analysis.mono.preview: + sections.append(_section_pwav(analysis.mono.preview)) + if analysis.mono.tiny: + sections.append(_section_pwv2(analysis.mono.tiny)) sections.append(_section_pcob(1)) # hot cues (empty) — real writes hot-cues first sections.append(_section_pcob(0)) # memory cues (empty) body = b"".join(sections) return _pmai_header(ANLZ_HEADER_LEN + len(body)) + body -def build_ext(rb_path: str, analysis: dict | None = None) -> bytes: +def build_ext(rb_path: str, analysis: Analysis | None = None) -> bytes: """Build ANLZ0000.EXT — extended analysis data the loaded-deck reads. Section order matches the real Q1 reference: @@ -378,43 +376,42 @@ def build_ext(rb_path: str, analysis: dict | None = None) -> bytes: PWV5 (colour scroll), PWV4 (colour preview), PSSI (empty). """ sections = [_section_pth(rb_path)] - if analysis and analysis.get("scroll"): - sections.append(_section_pwv3(analysis["scroll"])) + if analysis is not None and analysis.mono.scroll: + sections.append(_section_pwv3(analysis.mono.scroll)) sections.append(_section_pcob(1)) sections.append(_section_pcob(0)) sections.append(_section_pco2(1)) sections.append(_section_pco2(0)) - if analysis and analysis.get("beats"): + if analysis is not None and analysis.beats: # PQT2 between PCO2 and PWV5 — matches real-export section order # 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"" + sections.append(_section_pqt2(analysis.beats)) + if analysis is not None: + scroll = analysis.mono.scroll + s_lo, s_md, s_hi = analysis.bands.scroll 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"" + p_lo, p_md, p_hi = analysis.bands.preview if p_lo and p_md and p_hi: sections.append(_section_pwv4(p_lo, p_md, p_hi)) - num_beats = len(analysis.get("beats") or []) - if num_beats: - sections.append(_section_pssi(num_beats)) + if analysis.beats: + sections.append(_section_pssi(len(analysis.beats))) body = b"".join(sections) return _pmai_header(ANLZ_HEADER_LEN + len(body)) + body -def build_2ex(rb_path: str, analysis: dict | None = None) -> bytes: +def build_2ex(rb_path: str, analysis: Analysis | None = None) -> bytes: """Build ANLZ0000.2EX — CDJ-3000 3-band waveforms. Section order matches the real Q1 reference: PPTH, PWV7 (3-band scroll), PWV6 (3-band preview), PWVC (skipped). """ sections = [_section_pth(rb_path)] - if analysis: - 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 analysis is not None: + s_lo, s_md, s_hi = analysis.bands.scroll if s_lo and s_md and s_hi: sections.append(_section_pwv7(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"" + p_lo, p_md, p_hi = analysis.bands.preview if p_lo and p_md and p_hi: sections.append(_section_pwv6(p_lo, p_md, p_hi)) sections.append(_section_pwvc()) # 20-byte trailer real exports always include @@ -423,13 +420,10 @@ def build_2ex(rb_path: str, analysis: dict | None = None) -> bytes: def build_anlz(rb_path: str, source_audio: Path | None = None) -> bytes: - """Compatibility wrapper for callers that want the .DAT bytes only.""" - analysis: dict | None = None - if source_audio is not None: - try: - analysis = analyze_audio(source_audio) - except Exception as e: - logger.warning("audio analysis failed for {}: {}", source_audio.name, e) + """Compatibility wrapper for callers that want the .DAT bytes only. + + Analysis failures propagate — a broken analysis is fatal, not swallowed.""" + analysis = Analysis.analyze(source_audio) if source_audio is not None else None return build_dat(rb_path, analysis) @@ -458,39 +452,37 @@ def write_anlz( dest: Path, source_audio: Path | None = None, analysis_cache_dir: Path | None = None, -) -> dict | None: +) -> Analysis | None: """Write ANLZ0000.DAT (and a minimal ANLZ0000.EXT alongside). `dest` is the .DAT path. The .EXT companion is written next to it automatically — every real CDJ export ships both files. - Returns the analysis dict (cache hit or fresh aubio run) or None when + Returns the `Analysis` (cache hit or fresh mixxx run) or None when no source audio was given. """ dest.parent.mkdir(parents=True, exist_ok=True) - analysis: dict | None = None + analysis: Analysis | None = None cache_hit = False if source_audio is not None: isrc = _isrc_for(source_audio) if analysis_cache_dir else None if isrc and analysis_cache_dir: - cached = load_analysis(analysis_cache_dir, isrc) + cached = Analysis.load(analysis_cache_dir, isrc) if cached is not None: analysis = cached cache_hit = True logger.info("anlz: cache hit for {} (isrc {})", source_audio.name, isrc) if analysis is None: - try: - analysis = analyze_audio(source_audio) - except Exception as e: - logger.warning("audio analysis failed for {}: {}", source_audio.name, e) - analysis = None + # Not a cache hit — analyse fresh. Failures propagate (a broken + # analysis is fatal, not swallowed into a grid-less ANLZ). + analysis = Analysis.analyze(source_audio, isrc=isrc or "") # Populate the cache on every successful fresh analysis so subsequent - # exports (and collection runs) can skip the ffmpeg+aubio pass. The + # exports (and collection runs) can skip the ffmpeg+mixxx pass. The # cache is the single source of truth — populated by whichever # pipeline (`just collection` or `just export`) sees a track first. if analysis is not None and not cache_hit and analysis_cache_dir is not None and isrc: try: - cp = save_analysis(analysis_cache_dir, isrc, analysis) + cp = analysis.save(analysis_cache_dir) if cp is not None: logger.info("anlz: cached analysis for {} → {}", source_audio.name, cp) except Exception as e: diff --git a/maple-py/src/maple/cli.py b/maple-py/src/maple/cli.py index 3428116..4a4d215 100644 --- a/maple-py/src/maple/cli.py +++ b/maple-py/src/maple/cli.py @@ -11,7 +11,7 @@ from pathlib import Path from loguru import logger -from maple.analyze import analyze_audio, cache_path_for, save_analysis +from maple.analyze import Analysis, cache_path_for from maple.anlz import _isrc_for, write_anlz from maple.artwork import _bucket as _artwork_bucket from maple.artwork import extract_artwork @@ -181,8 +181,8 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple # Parallel pre-analysis. The per-track build loop below is sequential # (the pdb / artwork-id assignments depend on iteration order), but - # `analyze_audio` is the slow step and trivially parallel: ffmpeg + - # aubio + scipy each release the GIL, so threads scale linearly with + # `Analysis.analyze` is the slow step and trivially parallel: ffmpeg + + # mixxx + scipy each release the GIL, so threads scale linearly with # cores. After this pass every track has its ISRC-keyed analysis # cached, so write_anlz in the build loop is a cache hit. def _dest_for(p: Path) -> Path: @@ -202,7 +202,8 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple def _ensure_analysis_cache(p: Path) -> bool: """Populate `/.json` if missing. Returns True iff a - fresh analyse_audio actually ran (i.e. cache was cold).""" + fresh analysis actually ran (i.e. cache was cold). Analysis failures + propagate — a broken analysis is fatal, not skipped.""" dest = _dest_for(p) if not dest.exists(): return False @@ -212,15 +213,9 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple cp = cache_path_for(cache_dir, isrc) if cp is None or cp.exists(): return False - try: - a = analyze_audio(dest) - except Exception as e: - logger.warning("analyze failed for {}: {}", dest.name, e) - return False - if a: - save_analysis(cache_dir, isrc, a) - return True - return False + a = Analysis.analyze(dest, isrc=isrc) + a.save(cache_dir) + return True pre_t0 = time.perf_counter() logger.info("pre-analysing {} tracks (workers={})...", len(audio_files), workers) @@ -230,11 +225,8 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple futures = {ex.submit(_ensure_analysis_cache, p): p for p in audio_files} for fut in as_completed(futures): done += 1 - try: - if fut.result(): - new_count += 1 - except Exception as e: - logger.warning("pre-analysis error: {}", e) + if fut.result(): # re-raises any worker analysis failure + new_count += 1 if done % 8 == 0 or done == len(audio_files): logger.info(" [{}/{}] cache misses analysed: {}", done, len(audio_files), new_count) logger.info( @@ -328,7 +320,7 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple 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 - # falling back to a fresh ffmpeg+aubio pass. + # falling back to a fresh ffmpeg+mixxx pass. analysis = write_anlz( file_path, anlz_abs, source_audio=dest_abs, analysis_cache_dir=input_dir / "_analysis", @@ -341,12 +333,12 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple # CDJ browse panel reads pdb.tempo; loaded-deck readout reads the PQTZ # we just wrote — keeping the two in sync avoids the "120.0 in browse, # blank on deck" UX bug. - if analysis and analysis.get("bpm"): - tempo_x100 = round(analysis["bpm"] * 100) + if analysis is not None and analysis.bpm: + tempo_x100 = analysis.grid.tempo_x100 # Also write the bpm TAG to the file (symmetric with `just # collection`'s _write_bpm_tag). overwrite=False so any # pre-existing label-set BPM is preserved. - write_bpm_tag(dest_abs, analysis["bpm"], overwrite=False) + write_bpm_tag(dest_abs, analysis.bpm, overwrite=False) else: tempo_x100 = meta.tempo db.tracks.append(Track( @@ -384,7 +376,7 @@ def build(input_dir: Path, usb_root: Path, with_playlists: bool = True) -> tuple validation_issues.extend(track_issues) track_dt = time.perf_counter() - track_t0 - bpm_str = f"{analysis['bpm']:.1f} bpm" if (analysis and analysis.get("bpm")) else f"tag {tempo_x100/100:.1f} bpm" + bpm_str = f"{analysis.bpm:.1f} bpm" if (analysis is not None and analysis.bpm) else f"tag {tempo_x100/100:.1f} bpm" issue_tag = f" [{len(track_issues)} issue(s)]" if track_issues else "" logger.info( "[{}/{}] {} - {} ({}, {:.1f}s){}", diff --git a/maple-py/src/maple/collection/core.py b/maple-py/src/maple/collection/core.py index 811a162..d3f8aa5 100644 --- a/maple-py/src/maple/collection/core.py +++ b/maple-py/src/maple/collection/core.py @@ -23,7 +23,7 @@ import aiohttp from loguru import logger from mutagen import File as MutagenFile -from maple.analyze import analyze_audio, detect_bpm, save_analysis +from maple.analyze import Analysis from maple.enrich import write_spotify_tags as _shared_write_spotify_tags from maple.validate import TrackIssue as _ValidateTrackIssue from maple.validate import validate_track as _validate_track @@ -479,25 +479,22 @@ class Collection: # Single audio analysis pass: BPM + beat grid + waveform envelopes. # Cached under collection/_analysis/.json so `just export` - # reads the bytes back instead of redoing ffmpeg+aubio. + # reads the bytes back instead of redoing ffmpeg+mixxx. Analysis + # failures propagate — a broken analysis is fatal, not skipped. isrc = (track.get("isrc") or "").strip() existing_bpm = self._existing_bpm(path) - try: - analysis = analyze_audio(path) - except Exception as e: - logger.warning("analyze_audio failed on {}: {}", path.name, e) - analysis = None - if analysis and analysis.get("bpm"): + analysis = Analysis.analyze(path, isrc=isrc) + if analysis.bpm: if existing_bpm is None: - self._write_bpm_tag(path, analysis["bpm"]) - logger.info("{}: bpm {:.2f} (aubio)", path.name, analysis["bpm"]) + self._write_bpm_tag(path, analysis.bpm) + logger.info("{}: bpm {:.2f} (mixxx)", path.name, analysis.bpm) else: logger.debug( - "{}: bpm {:.2f} (existing tag, kept; aubio measured {:.2f})", - path.name, existing_bpm, analysis["bpm"], + "{}: bpm {:.2f} (existing tag, kept; mixxx measured {:.2f})", + path.name, existing_bpm, analysis.bpm, ) if isrc: - cp = save_analysis(self.analysis_dir, isrc, analysis) + cp = analysis.save(self.analysis_dir) logger.info("cached analysis for {}: {} → {}", isrc, path.name, cp) else: logger.warning( diff --git a/maple-py/src/maple/enrich.py b/maple-py/src/maple/enrich.py index f1a17c9..abe4067 100644 --- a/maple-py/src/maple/enrich.py +++ b/maple-py/src/maple/enrich.py @@ -300,8 +300,8 @@ def enrich_from_manifest( # write_anlz will populate it and write the tag in a later pass. isrc = (track.get("isrc") or "").strip() if analysis_cache_dir is not None and isrc: - from maple.analyze import load_analysis - cached = load_analysis(analysis_cache_dir, isrc) - if cached and cached.get("bpm"): - write_bpm_tag(path, cached["bpm"], overwrite=False) + from maple.analyze import Analysis + cached = Analysis.load(analysis_cache_dir, isrc) + if cached is not None and cached.bpm: + write_bpm_tag(path, cached.bpm, overwrite=False) return track diff --git a/maple-py/src/maple/track.py b/maple-py/src/maple/track.py index d28a30e..c44f2ce 100644 --- a/maple-py/src/maple/track.py +++ b/maple-py/src/maple/track.py @@ -13,9 +13,13 @@ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path +from typing import TYPE_CHECKING from maple import model +if TYPE_CHECKING: + from maple.analyze import Analysis + @dataclass class TrackIssue: @@ -220,10 +224,10 @@ class Track: self.source = "spotify" return self - def merge_analysis(self, analysis: dict) -> "Track": - """Merge fields from `maple.analyze.analyze_audio`'s output.""" - if analysis.get("bpm"): - self.bpm = float(analysis["bpm"]) + def merge_analysis(self, analysis: "Analysis") -> "Track": + """Merge fields from a `maple.analyze.Analysis`.""" + if analysis.bpm: + self.bpm = float(analysis.bpm) return self # ---- validation ------------------------------------------------------ #