pipeline.py

# ============================================================================
#  pipeline.py — audio -> Demucs (vocal stem) -> YourMT3+ (transcription)
#                -> full post-processing -> notes
#
#  Transcription engine: YourMT3+ (YPTF.MoE+Multi, noPS checkpoint), loaded once
#  at startup on GPU. The model code is cloned from the official HF Space at
#  build time (see Dockerfile) into /app/yourmt3.
#
#  Post-processing (ported verbatim from the validated Colab pipeline):
#    raw notes -> tempo-driven merge (drop_short + register fold + conservative
#    merge) -> octave repair -> vibrato collapse -> key estimation + snapping
#    (from pmi_core) -> out-of-key fragment regularization -> final notes
#
#  YourMT3+ outputs no per-note confidence (velocity is constant), so the
#  snapping evidence comes from duration + isolation only; SNAP_THRESH is
#  recalibrated accordingly inside pmi_core.
# ============================================================================
import os, sys, glob, shutil
import numpy as np
import librosa
import soundfile as sf
import pretty_midi
def _patch_dependency_conflicts():

    import importlib.metadata as _md
    _orig_version = _md.version
    def _fake_version(name):
        n = (name or "").replace("_", "-").lower()
        if n == "huggingface-hub":
            return "0.999.0"   # satisfies transformers' <1.0 pin
        try:
            return _orig_version(name)
        except Exception:
            return "99.0.0"    # missing metadata: high ver so >= checks pass
    _md.version = _fake_version
    try:                       # transformers may read via the backport instead
        import importlib_metadata as _imd
        _imd.version = _fake_version
    except Exception:
        pass

    try:
        import lightning_utilities.core.imports as _lui
    except Exception:
        return                 # lightning absent: nothing to guard
    _BLOCK = {"torch_fidelity", "pesq", "pystoi", "fast_bss_eval", "lpips",
              "sacrebleu", "nltk", "rouge_score", "scienceplots"}
    _orig_bool = _lui.RequirementCache.__bool__
    def _patched_bool(self):
        req = getattr(self, "requirement", "") or ""
        mod = getattr(self, "module", "") or ""
        for _b in _BLOCK:
            if req == _b or mod == _b or req.startswith(_b):
                return False
        return _orig_bool(self)
    _lui.RequirementCache.__bool__ = _patched_bool

_patch_dependency_conflicts()   # before anything can import transformers/torchmetrics


def _ensure_transformers():

    try:
        import transformers  # already present?
        return
    except Exception:
        pass
    import subprocess, sys as _sys
    # transformers + tokenizers themselves, no deps (so hub isn't downgraded)
    subprocess.run([_sys.executable, "-m", "pip", "install", "--no-cache-dir",
                    "--no-deps", "transformers==4.45.1", "tokenizers==0.20.3"],
                   check=True)

    subprocess.run([_sys.executable, "-m", "pip", "install", "--no-cache-dir",
                    "--no-deps", "regex", "safetensors>=0.4.1"],
                   check=True)


# ---------------------------------------------------------------------------
# YourMT3+ model loading (once, at import time)
# ---------------------------------------------------------------------------
# On ZeroGPU (Gradio SDK) there is no Dockerfile, so the YourMT3+ code and
# checkpoint are fetched at import time into a writable directory.
_HOME = os.environ.get("HOME", "/home/user")
YOURMT3_DIR = os.environ.get("YOURMT3_DIR", os.path.join(_HOME, "yourmt3"))
_WORK = os.environ.get("MT3_WORK", "/tmp/mt3_work")
os.makedirs(os.path.join(_WORK, "model_output"), exist_ok=True)

_CKPT_NAME = "mc13_256_g4_all_v7_mt3f_sqr_rms_moe_wf4_n8k2_silu_rope_rp_b36_nops"

def _ensure_yourmt3():
    """Clone the YourMT3+ code (no LFS) and download the checkpoint into the
    repo at amt/logs/2024/<name>/checkpoints/. Runs once at import."""
    import subprocess
    ckpt = os.path.join(YOURMT3_DIR, "amt", "logs", "2024", _CKPT_NAME,
                        "checkpoints", "last.ckpt")
    if os.path.exists(ckpt) and os.path.getsize(ckpt) > 100_000_000:
        return
    if not os.path.isdir(YOURMT3_DIR):
        env = dict(os.environ, GIT_LFS_SKIP_SMUDGE="1")
        subprocess.run(["git", "clone", "--depth", "1",
                        "https://huggingface.co/spaces/mimbres/YourMT3", YOURMT3_DIR],
                       check=True, env=env)
        # NB: YourMT3+'s Python deps are installed at BUILD time via our
        # requirements.txt (not here) so the runtime environment is never
        # mutated — mutating it at runtime corrupts numpy/hub/transformers.
    from huggingface_hub import snapshot_download
    dl = snapshot_download(repo_id="mimbres/YourMT3",
                           allow_patterns=[f"logs/2024/{_CKPT_NAME}/*"],
                           local_dir=os.path.join(_HOME, "yourmt3-ckpt"))
    dst = os.path.join(YOURMT3_DIR, "amt", "logs")
    shutil.rmtree(dst, ignore_errors=True)
    os.makedirs(dst, exist_ok=True)
    shutil.copytree(os.path.join(dl, "logs", "2024"),
                    os.path.join(dst, "2024"), dirs_exist_ok=True)
    if not (os.path.exists(ckpt) and os.path.getsize(ckpt) > 100_000_000):
        raise RuntimeError("YourMT3+ checkpoint download failed (git-lfs pointer?)")

_ensure_yourmt3()
_ensure_transformers()
sys.path.insert(0, os.path.join(YOURMT3_DIR, "amt", "src"))
sys.path.insert(0, YOURMT3_DIR)

_model = None


def _patch_torchaudio_load():
    """torchaudio 2.11 routes load()/save() through torchcodec, whose shared
    libraries need a matching system FFmpeg and can fail to load in the image.
    We only ever read standard WAV files (the demucs vocal stem we wrote with
    soundfile), so replace torchaudio.load with a soundfile-based reader.
    Idempotent; safe to call more than once."""
    import torch
    import torchaudio as ta
    if getattr(ta.load, "_sf_patched", False):
        return

    def _sf_load(uri, *args, **kwargs):
        data, sr = sf.read(str(uri), dtype="float32", always_2d=True)
        wav = torch.from_numpy(data.T.copy())   # (channels, samples)
        return wav, int(sr)

    _sf_load._sf_patched = True
    ta.load = _sf_load



def _load_model():
    """Load YourMT3+ once. Called lazily on first transcription request so the
    web server comes up fast; the first request pays the load cost."""
    global _model
    if _model is not None:
        return _model
    cwd = os.getcwd()
    try:
        os.chdir(YOURMT3_DIR)   # checkpoint paths are resolved relative to the repo
        from model_helper import load_model_checkpoint  # noqa: import from cloned repo
        checkpoint = ("mc13_256_g4_all_v7_mt3f_sqr_rms_moe_wf4_n8k2_silu_rope_rp_b36_nops"
                      "@last.ckpt")
        args = [checkpoint, "-p", "2024", "-tk", "mc13_full_plus_256",
                "-dec", "multi-t5", "-nl", "26", "-enc", "perceiver-tf",
                "-sqr", "1", "-ff", "moe", "-wf", "4", "-nmoe", "8",
                "-kmoe", "2", "-act", "silu", "-epe", "rope", "-rp", "1",
                "-ac", "spec", "-hop", "300", "-atc", "1", "-pr", "16"]
        m = load_model_checkpoint(args=args, device="cpu")
        # NB: stays on CPU here; moved to GPU inside the @spaces.GPU call.
        _model = m
        return _model
    finally:
        os.chdir(cwd)


def _audio_info(path):
    """soundfile-based audio metadata (torchaudio.info was removed upstream)."""
    with sf.SoundFile(path) as f:
        sr, n_frames, n_ch = f.samplerate, len(f), f.channels
    return dict(filepath=path,
                track_name=os.path.basename(path).rsplit(".", 1)[0],
                sample_rate=int(sr), bits_per_sample=16,
                num_channels=int(n_ch), num_frames=int(n_frames),
                duration=int(n_frames / sr), encoding="pcm_s")


def yourmt3_transcribe(audio_path):
    """Vocal-stem audio -> raw notes [(start, end, pitch)] via YourMT3+.
    transcribe() writes './model_output/<track>.mid' relative to cwd."""
    model = _load_model()
    # ZeroGPU: a GPU is only attached inside the @spaces.GPU call, so the model
    # is moved here (a no-op if it is already resident).
    #
    # This FAILS LOUDLY if there is no GPU, and must keep doing so. The
    # checkpoint is loaded at fp16 (-pr 16), and fp16 inference on CPU is not
    # properly supported in PyTorch -- it returns numerical garbage, which
    # surfaces as a dense mess of nonsense notes rather than as an error. The
    # previous version skipped the move when cuda.is_available() was False and
    # swallowed exceptions, so a worker whose GPU was not yet attached silently
    # transcribed on CPU and produced exactly that garbage. Plausible-looking
    # but wrong output is the worst failure mode for a forensic tool: better to
    # raise and let the caller retry than to hand back numbers nobody can trust.
    import torch
    if not torch.cuda.is_available():
        raise RuntimeError(
            "No GPU attached inside the transcription call. Refusing to run "
            "fp16 inference on CPU, which yields invalid output. Please retry.")
    model.to("cuda")
    _patch_torchaudio_load()   # model_helper.transcribe reads audio via torchaudio.load
    cwd = os.getcwd()
    try:
        os.chdir(_WORK)
        from model_helper import transcribe  # same cloned module
        out_rel = transcribe(model, _audio_info(audio_path))
        src = out_rel if os.path.isabs(out_rel) else os.path.join(_WORK, out_rel)
        if not os.path.exists(src):
            raise RuntimeError(f"YourMT3+ returned {out_rel!r} but the file is missing")
        pm = pretty_midi.PrettyMIDI(src)
        cands = [i for i in pm.instruments if not i.is_drum and len(i.notes) > 0]
        if not cands:
            return []
        inst = max(cands, key=lambda i: len(i.notes))
        raw = [(float(n.start), float(n.end), int(n.pitch)) for n in inst.notes]
        raw.sort(key=lambda t: (t[0], t[1], t[2]))
        try:
            os.unlink(src)
        except OSError:
            pass
        return raw
    finally:
        os.chdir(cwd)


# ---------------------------------------------------------------------------
# Demucs vocal separation
# ---------------------------------------------------------------------------
DEMUCS_OUT = os.environ.get("DEMUCS_OUT", "/tmp/demucs_out")


def _patch_demucs_save():
    """torchaudio 2.11 routes ta.save() through torchcodec, whose GPU wheel can
    fail to load if its CUDA build does not match the image. Replace demucs'
    save_audio with a soundfile-based writer so the separation result is saved
    regardless of the torchcodec state.

    demucs.separate does `from .audio import save_audio`, which copies the
    reference into its own namespace — so BOTH modules must be patched."""
    try:
        import demucs.audio as _da
        import demucs.separate as _ds
    except ImportError:
        return

    def _sf_save_audio(wav, path, samplerate, **kwargs):
        arr = wav.detach().cpu().numpy() if hasattr(wav, "detach") else np.asarray(wav)
        if arr.ndim == 2:            # (channels, samples) -> (samples, channels)
            arr = arr.T
        sf.write(str(path), arr, int(samplerate))

    _da.save_audio = _sf_save_audio
    _ds.save_audio = _sf_save_audio


# Which demucs model carries which stem. The vocal path stays on htdemucs on
# purpose: that is the model the PMI benchmark was validated against, so it must
# not change silently. guitar/piano only exist in htdemucs_6s, which is
# experimental -- guitar is usable, piano bleeds badly (upstream's own caveat).
DEMUCS_MODEL = {"vocals": "htdemucs", "guitar": "htdemucs_6s", "piano": "htdemucs_6s"}

def demucs_stem(audio_path, stem="vocals"):
    """Separate one stem. Defaults to vocals -- the validated path.

    A UNIQUE output dir per call (never a filename-based cache): a shared cache
    keyed on the basename caused cross-request bleed, where a second upload with
    the same basename matched the first request's leftover stem and transcribed
    the WRONG audio. A fresh dir per call guarantees we separate the audio we
    were actually given.
    """
    if stem not in DEMUCS_MODEL:
        stem = "vocals"
    model = DEMUCS_MODEL[stem]
    import uuid
    out_dir = os.path.join(DEMUCS_OUT, uuid.uuid4().hex)
    os.makedirs(out_dir, exist_ok=True)
    tag = os.path.basename(audio_path).rsplit(".", 1)[0]
    _patch_demucs_save()
    import demucs.separate as _ds
    # Do NOT pass --segment: htdemucs is a Transformer and refuses any segment
    # longer than the 7.8 s it was trained on ("FATAL: Cannot use a Transformer
    # model with a longer segment than it was trained for"), so there is no
    # window-size win to be had here. --shifts is likewise left at its default:
    # it multiplies inference time by the shift count for ~0.2 dB SDR.
    _ds.main([f"--two-stems={stem}", "-n", model, "-o", out_dir, audio_path])
    out = glob.glob(f"{out_dir}/{model}/{tag}/{stem}.wav")
    if not out:                       # fall back to any match under this call's dir
        out = glob.glob(f"{out_dir}/*/*/{stem}.wav")
    if not out:
        raise RuntimeError(f"demucs produced no {stem}.wav for " + audio_path)
    return out[0]

# ---------------------------------------------------------------------------
# Tempo estimation (from the vocal stem)
# ---------------------------------------------------------------------------
TEMPO_PRIOR = 120.0
TEMPO_CLAMP = (50.0, 200.0)

def estimate_bpm(audio_path, prior=TEMPO_PRIOR, clamp=TEMPO_CLAMP):
    y, sr = librosa.load(audio_path, sr=16000, mono=True)
    try:
        tempo, _ = librosa.beat.beat_track(y=y, sr=sr, start_bpm=prior)
        bpm = float(np.atleast_1d(tempo).ravel()[0])
    except Exception:
        bpm = prior
    if not np.isfinite(bpm) or bpm <= 0:
        bpm = prior
    return float(np.clip(bpm, *clamp))


# ---------------------------------------------------------------------------
# Tempo -> merge thresholds (octave-folded; verbatim from the batch pipeline)
# ---------------------------------------------------------------------------
FRAG_BEATS   = 1/8
GAP_BEATS    = 1/8
MINDUR_BEATS = 1/16
FRAG_CLAMP   = (0.05, 0.16)
GAP_CLAMP    = (0.025, 0.10)
MINDUR_CLAMP = (0.03, 0.09)
FOLD_BAND    = (70.0, 140.0)

def merge_thresholds_from_tempo(tempo):
    bpm = np.atleast_1d(np.asarray(tempo, dtype=float)).ravel()
    bpm = float(bpm[0]) if bpm.size else 0.0
    while bpm < FOLD_BAND[0]:
        bpm *= 2.0
    while bpm > FOLD_BAND[1]:
        bpm /= 2.0
    beat = 60.0 / bpm
    return (float(np.clip(beat * MINDUR_BEATS, *MINDUR_CLAMP)),
            float(np.clip(beat * GAP_BEATS,    *GAP_CLAMP)),
            float(np.clip(beat * FRAG_BEATS,   *FRAG_CLAMP)))


# ---------------------------------------------------------------------------
# Post-processing tunables (verbatim from the batch pipeline)
# ---------------------------------------------------------------------------
REGISTER_FOLD = 11
OCT_CTX_WIN   = 4
OCT_GOOD_JUMP = 7
OCT_NEAR      = 2
OCT_CONF_GATE = None
VIB_BAND      = 2
VIB_MIN_RUN   = 3
VIB_MAX_PIECE = 0.30
VIB_MAX_SPAN  = 2
FRAG_BAND       = 2
FRAG_SHORT_FRAC = 0.60
CLUSTER_MAX_DUR = 1.2


def _merge_same_pitch(notes, max_gap, frag_dur=None):
    """Join consecutive SAME-PITCH notes separated by less than max_gap.

    frag_dur guards against destroying real repeats: when given, a pair is only
    merged if at least one side is shorter than it, so two full-length notes on
    the same pitch (a genuine re-articulation) are left alone. Pass None to merge
    on gap alone, which is what the out-of-key pass wants -- there the notes have
    just been retuned to a common pitch and the runs are fragments by definition.

    Takes and returns lists of [start, end, pitch, ...]; extra fields on the
    first note of a run are preserved, since merging extends its end time.
    """
    if not notes:
        return []
    merged = [list(notes[0])]
    for n in notes[1:]:
        s, e, p = n[0], n[1], n[2]
        ls, le, lp = merged[-1][0], merged[-1][1], merged[-1][2]
        same = (p == lp) and (s - le < max_gap)
        if frag_dur is not None:
            same = same and ((le - ls) < frag_dur or (e - s) < frag_dur)
        if same:
            merged[-1][1] = e
        else:
            merged.append(list(n))
    return merged


def _grow_narrow_band(notes, i, band, max_piece=None):
    n = len(notes); j = i + 1
    while j < n:
        seg = [notes[k][2] for k in range(i, j + 1)]
        center = np.median(seg)
        within = max(abs(pp - center) for pp in seg) <= band
        short_ok = (max_piece is None or
                    all((notes[k][1] - notes[k][0]) <= max_piece for k in range(i, j + 1)))
        if within and short_ok:
            j += 1
        else:
            break
    return j


def octave_repair(notes, win=OCT_CTX_WIN, good_jump=OCT_GOOD_JUMP,
                  near_oct=OCT_NEAR, conf_gate=OCT_CONF_GATE):
    if len(notes) < 3:
        return notes
    out = [list(n) for n in notes]
    P = [n[2] for n in out]
    for i in range(len(out)):
        lo = max(0, i - win); hi = min(len(out), i + win + 1)
        ctx = [P[j] for j in range(lo, hi) if j != i]
        if not ctx:
            continue
        center = float(np.median(ctx))
        cur = out[i][2]
        if abs(cur - center) <= good_jump:
            continue
        if conf_gate is not None and len(out[i]) > 3 and out[i][3] >= conf_gate:
            continue
        best, best_d = cur, abs(cur - center)
        for shift in (-12, 12, -24, 24):
            c2 = cur + shift
            if not (40 <= c2 <= 95):
                continue
            if abs(abs(cur - center) - abs(shift)) > near_oct:
                continue
            d = abs(c2 - center)
            if d < best_d:
                best, best_d = c2, d
        if best != cur:
            out[i][2] = best; P[i] = best
    return [tuple(n) for n in out]


def _vibrato_runs(notes, band=VIB_BAND, min_run=VIB_MIN_RUN, max_piece=VIB_MAX_PIECE):
    n = len(notes)
    if n < min_run:
        return []
    p = np.array([nn[2] for nn in notes])
    runs, i = [], 0
    while i < n - min_run + 1:
        j = _grow_narrow_band(notes, i, band, max_piece)
        if j - i >= min_run:
            seg = p[i:j]; d = np.diff(seg); nz = d[d != 0]
            reversals = int(np.sum(nz[:-1] * nz[1:] < 0)) if len(nz) >= 2 else 0
            if reversals >= 1 and len(set(seg.tolist())) >= 2:
                runs.append((i, j))
            i = j
        else:
            i += 1
    return runs


def collapse_vibrato(notes, band=VIB_BAND, min_run=VIB_MIN_RUN,
                     max_piece=VIB_MAX_PIECE, max_span=VIB_MAX_SPAN):
    runs = _vibrato_runs(notes, band, min_run, max_piece)
    actions = {}
    for (i, j) in runs:
        seg = notes[i:j]
        span = max(nn[2] for nn in seg) - min(nn[2] for nn in seg)
        if span > max_span:
            continue
        wt = {}
        for nn in seg:
            wt[nn[2]] = wt.get(nn[2], 0.0) + (nn[1] - nn[0])
        actions[i] = (j, int(max(wt, key=wt.get)))
    if not actions:
        return [tuple(nn) for nn in notes]
    out, k, n = [], 0, len(notes)
    while k < n:
        if k in actions:
            j, center = actions[k]; seg = notes[k:j]
            s, e = seg[0][0], seg[-1][1]
            out.append((s, e, center)); k = j
        else:
            out.append(tuple(notes[k])); k += 1
    return out


def postprocess_notes(notes, tempo=None, min_dur=0.05, merge_gap=0.04,
                      frag_dur=0.08, fix_octaves=True, collapse_vib=True):
    """drop_short + register fold + conservative merge + octave repair +
    vibrato collapse. Thresholds are tempo-driven when tempo is given."""
    if tempo is not None:
        min_dur, merge_gap, frag_dur = merge_thresholds_from_tempo(tempo)
    notes = sorted([list(n) for n in notes])
    if not notes:
        return []
    med = int(np.median([p for _, _, p, *_ in notes]))
    for n in notes:
        while n[2] - med > REGISTER_FOLD:
            n[2] -= 12
        while med - n[2] > REGISTER_FOLD:
            n[2] += 12
    merged = _merge_same_pitch(notes, merge_gap, frag_dur)
    result = [tuple(n[:3]) for n in merged]
    if fix_octaves:
        result = octave_repair(result)
    if collapse_vib:
        result = collapse_vibrato(result)
    # Drop what is STILL too short, after every reassembly step. Filtering first
    # -- as this did previously -- deleted the fragments of a broken-up note
    # before merge/vibrato-collapse could put them back, so a sustained note
    # could vanish outright. Vibrato was the worst case: its fragments differ in
    # pitch, so the merge loop (which requires p == lp) can never rescue them --
    # only collapse_vibrato can, and it runs last. Anything still under min_dur
    # here is a fragment nothing could reassemble, i.e. genuine noise.
    result = [n for n in result if (n[1] - n[0]) >= min_dur]
    return result


# ---------------------------------------------------------------------------
# Key-aware final stage: snapping (via pmi_core) + OOK fragment regularization
# ---------------------------------------------------------------------------
import pmi_core   # single source of truth for K-S estimation + snapping


def regularize_oot_fragments(notes, info, tempo=None, band=FRAG_BAND,
                             short_frac=FRAG_SHORT_FRAC, max_dur=CLUSTER_MAX_DUR):
    """Inside TIGHT narrow-band clusters (most notes short AND cluster brief),
    retune SHORT + OUT-OF-KEY notes to the cluster's longest in-key note, then
    merge same-pitch neighbours. In-key / normal-length notes are never moved."""
    n = len(notes)
    if n < 2 or info is None:
        return [tuple(x) for x in notes]
    scale = pmi_core.scale_for(info["tonic"], info["mode"])
    _, _, frag_dur = merge_thresholds_from_tempo(tempo) if tempo else (0, 0, 0.08)
    p   = [int(x[2]) for x in notes]
    dur = [x[1] - x[0] for x in notes]
    short = [d < frag_dur for d in dur]
    oot   = [(pp % 12) not in scale for pp in p]
    out = [list(x) for x in notes]
    i = 0
    while i < n:
        j = _grow_narrow_band(notes, i, band)
        seg = range(i, j); size = j - i
        tot = notes[j - 1][1] - notes[i][0]
        n_short = sum(short[k] for k in seg)
        in_idx  = [k for k in seg if not oot[k]]
        targets = [k for k in seg if short[k] and oot[k]]
        tight = (size >= 2 and n_short / size >= short_frac and tot <= max_dur)
        if tight and in_idx and targets:
            tgt = p[max(in_idx, key=lambda k: dur[k])]
            for k in targets:
                out[k][2] = tgt
            i = j
        else:
            i += 1
    merged = _merge_same_pitch(out, frag_dur)
    return [tuple(x[:3]) for x in merged]


def notes_from_audio(vocal_path, tempo=None):
    """Full transcription line for the /transcribe endpoint:
    vocal stem -> YourMT3+ -> tempo-driven postprocess -> key estimation +
    snapping (pmi_core.estimate_key_locked) -> OOK fragment regularization."""
    raw = yourmt3_transcribe(vocal_path)
    if not raw:
        return []
    if tempo is None:
        tempo = estimate_bpm(vocal_path)
    pp = postprocess_notes(raw, tempo=tempo)
    if not pp:
        return []
    key, cls = pmi_core.estimate_key_locked(pp, confs=None)
    if key is None:
        return pp
    snapped_pcs = pmi_core.corrected_pitches(pp, cls)
    snapped = []
    for (s, e, p), new_pc in zip(pp, snapped_pcs):
        octave = (int(p) // 12) * 12
        best = octave + new_pc
        if abs(best - p) > 6:
            best += 12 if best < p else -12
        snapped.append((s, e, int(best)))
    return regularize_oot_fragments(snapped, key, tempo=tempo)


# ---------------------------------------------------------------------------
# Chroma for the note-completion hints (assistive only, never auto-fills).
# ---------------------------------------------------------------------------
def chroma_from_audio(vocal_path, sr=22050, hop=512):
    """Per-frame chroma (pitch-class salience) + per-frame energy of the vocal
    stem, for the editor's note-completion hints.

    When the user completes a note the model missed, we show what the AUDIO
    carries there. Evidence, not a decision -- the user picks.

    Two arrays are returned because they answer different questions:
      chroma : WHICH pitch class dominates. Normalised per frame (strongest
               pc = 255). Division by the frame max preserves the *shape*
               within a frame, so how peaked vs flat it is survives -- that is
               what tells us whether the audio is decisive or ambiguous.
      energy : WHETHER there is any signal at all. Per-frame RMS, scaled to the
               track max. The per-frame chroma normalisation deliberately
               throws loudness away, so silence would otherwise amplify into a
               confident-looking peak. This array is what stops the hint from
               claiming "100%" over a breath or a gap.

    chroma is frame-major (T x 12): frontend reads data[frame*12 + pc].
    """
    y, _ = librosa.load(vocal_path, sr=sr, mono=True)
    C = librosa.feature.chroma_cqt(y=y, sr=sr, hop_length=hop)      # 12 x T
    C = C / (C.max(axis=0, keepdims=True) + 1e-9)                   # per-frame shape
    q = np.clip(C * 255.0, 0, 255).astype(np.uint8)               # 12 x T

    rms = librosa.feature.rms(y=y, frame_length=2048, hop_length=hop)[0]  # T
    rms = rms[: q.shape[1]]
    if rms.shape[0] < q.shape[1]:                                   # pad if short
        rms = np.pad(rms, (0, q.shape[1] - rms.shape[0]))
    e = rms / (rms.max() + 1e-9)
    e = np.clip(e * 255.0, 0, 255).astype(np.uint8)               # T

    return {"data": q.T.tobytes(), "energy": e.tobytes(),
            "n_frames": int(q.shape[1]), "sr": int(sr), "hop": int(hop)}