dataclass refactor anlz

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`).

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."""
+    positions in seconds). BPM/beat detection is integral to an export, so a
-    try:
+    failure here propagates rather than degrading to an analysis with no grid."""
-        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]]:
+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(),
+    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())
+        _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(band: Band) -> np.ndarray:
-    def env(lo: int, hi: int) -> np.ndarray:
+        sos = butter(4, [band.lo_hz, band.hi_hz], btype="band", fs=sr, output="sos")
-        sos = butter(4, [lo, hi], 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)
+# --------------------------------------------------------------------------- #
-
+# ISRC-keyed cache paths                                                      #
-# 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)
 
 
 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."""
+# Composed analysis dataclasses                                               #
-    target = cache_path_for(cache_dir, isrc)
+# --------------------------------------------------------------------------- #
+
+
+@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 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)
-    payload = {
+        target.write_text(json.dumps(self.to_cache(), separators=(",", ":")))
-        "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
 
-
+    @classmethod
-def load_analysis(cache_dir: Path, isrc: str) -> dict | None:
+    def load(cls, cache_dir: Path, isrc: str) -> "Analysis | None":
-    """Return cached analysis for `isrc`, or None if absent/stale/unreadable."""
+        """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:
-        d = json.loads(target.read_text())
+            return cls.from_cache(json.loads(target.read_text()))
-        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
-
-
-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

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 is not None:
-        if analysis.get("beats"):
+        if analysis.beats:
-            sections.append(_section_pqtz(analysis["beats"]))
+            sections.append(_section_pqtz(analysis.beats))
-        preview = analysis.get("preview") or b""
+        if analysis.mono.preview:
-        tiny    = analysis.get("tiny") or b""
+            sections.append(_section_pwav(analysis.mono.preview))
-        if preview:
+        if analysis.mono.tiny:
-            sections.append(_section_pwav(preview))
+            sections.append(_section_pwv2(analysis.mono.tiny))
-        if tiny:
-            sections.append(_section_pwv2(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"):
+    if analysis is not None and analysis.mono.scroll:
-        sections.append(_section_pwv3(analysis["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"]))
+        sections.append(_section_pqt2(analysis.beats))
-    if analysis:
+    if analysis is not None:
-        scroll = analysis.get("scroll") or b""
+        scroll = analysis.mono.scroll
-        s_lo, s_md, s_hi = analysis.get("scroll_low") or b"", analysis.get("scroll_mid") or b"", analysis.get("scroll_high") or b""
+        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 analysis.beats:
-        if num_beats:
+            sections.append(_section_pssi(len(analysis.beats)))
-            sections.append(_section_pssi(num_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:
+    if analysis is not None:
-        s_lo, s_md, s_hi = analysis.get("scroll_low") or b"", analysis.get("scroll_mid") or b"", analysis.get("scroll_high") or b""
+        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."""
+    """Compatibility wrapper for callers that want the .DAT bytes only.
-    analysis: dict | None = None
+
-    if source_audio is not None:
+    Analysis failures propagate — a broken analysis is fatal, not swallowed."""
-        try:
+    analysis = Analysis.analyze(source_audio) if source_audio is not None else None
-            analysis = analyze_audio(source_audio)
-        except Exception as e:
-            logger.warning("audio analysis failed for {}: {}", source_audio.name, e)
     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:
+            # Not a cache hit — analyse fresh. Failures propagate (a broken
-                analysis = analyze_audio(source_audio)
+            # analysis is fatal, not swallowed into a grid-less ANLZ).
-            except Exception as e:
+            analysis = Analysis.analyze(source_audio, isrc=isrc or "")
-                logger.warning("audio analysis failed for {}: {}", source_audio.name, e)
-                analysis = None
         # 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:

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 +
+    # `Analysis.analyze` is the slow step and trivially parallel: ffmpeg +
-    # aubio + scipy each release the GIL, so threads scale linearly with
+    # 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 `<cache>/<isrc>.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 = Analysis.analyze(dest, isrc=isrc)
-            a = analyze_audio(dest)
+        a.save(cache_dir)
-        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
 
     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():          # re-raises any worker analysis failure
-                if fut.result():
                 new_count += 1
-            except Exception as e:
-                logger.warning("pre-analysis error: {}", e)
             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"):
+        if analysis is not None and analysis.bpm:
-            tempo_x100 = round(analysis["bpm"] * 100)
+            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){}",

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/<isrc>.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 = Analysis.analyze(path, isrc=isrc)
-            analysis = analyze_audio(path)
+        if analysis.bpm:
-        except Exception as e:
-            logger.warning("analyze_audio failed on {}: {}", path.name, e)
-            analysis = None
-        if analysis and analysis.get("bpm"):
             if existing_bpm is None:
-                self._write_bpm_tag(path, analysis["bpm"])
+                self._write_bpm_tag(path, analysis.bpm)
-                logger.info("{}: bpm {:.2f} (aubio)", path.name, analysis["bpm"])
+                logger.info("{}: bpm {:.2f} (mixxx)", path.name, analysis.bpm)
             else:
                 logger.debug(
-                    "{}: bpm {:.2f} (existing tag, kept; aubio measured {:.2f})",
+                    "{}: bpm {:.2f} (existing tag, kept; mixxx measured {:.2f})",
-                    path.name, existing_bpm, analysis["bpm"],
+                    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(

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
+        from maple.analyze import Analysis
-        cached = load_analysis(analysis_cache_dir, isrc)
+        cached = Analysis.load(analysis_cache_dir, isrc)
-        if cached and cached.get("bpm"):
+        if cached is not None and cached.bpm:
-            write_bpm_tag(path, cached["bpm"], overwrite=False)
+            write_bpm_tag(path, cached.bpm, overwrite=False)
     return track

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":
+    def merge_analysis(self, analysis: "Analysis") -> "Track":
-        """Merge fields from `maple.analyze.analyze_audio`'s output."""
+        """Merge fields from a `maple.analyze.Analysis`."""
-        if analysis.get("bpm"):
+        if analysis.bpm:
-            self.bpm = float(analysis["bpm"])
+            self.bpm = float(analysis.bpm)
         return self
 
     # ---- validation ------------------------------------------------------ #