scripts/note_alignment_report.py

#!/usr/bin/env python3
"""Build a compact, causal note alignment view from one NPTrace session.

The raw trace's event ``t`` is the canonical session clock.  AudioContext and
worklet clocks are retained as source fields; conversions below are estimates,
never replacement measurements.
"""
from __future__ import annotations
import argparse, csv, json, math, statistics
from pathlib import Path

def _num(x):
    return x if isinstance(x, (int, float)) and math.isfinite(x) else None

def _nearest(items, t, pred=lambda e: True, after=-1e9, before=1e9):
    xs = [e for e in items if pred(e) and after <= e.get("t", -1e9) <= before]
    return min(xs, key=lambda e: abs(e["t"]-t)) if xs else None

def _name(p):
    names = ['C','C#','D','D#','E','F','F#','G','G#','A','A#','B']
    return f"{names[p%12]}{p//12-1}" if isinstance(p, int) else "?"

def _cell(v):
    return "--" if v is None else (f"{v:.3f}" if isinstance(v, float) else str(v))

def build_alignment(trace):
    evs = trace.get("events", [])
    for n, e in enumerate(evs): e["_id"] = n
    meta = trace.get("meta", {})
    lat = meta.get("latency", {}) or {}
    sr = (meta.get("follower", {}) or {}).get("sr") or 48000
    hop = (meta.get("follower", {}) or {}).get("hop") or 512
    frames = [e for e in evs if e.get("type") == "follow.frame"]
    frame0 = frames[0] if frames else None
    min_midi = (meta.get("follower", {}) or {}).get("options", {}).get("minMidi", 33)
    def frame_ms(e):
        if not frame0: return None
        return round((e.get("at", 0)-frame0.get("at", 0))*1000/sr, 3)
    input_events = [e for e in evs if e.get("type") == "input.testEvent"]
    audio = [e for e in evs if e.get("type") == "audio.note" and e.get("role") == "solo"]
    advances = [e for e in evs if e.get("type") == "follow.advance" and e.get("accepted", True)]
    holds = [e for e in evs if e.get("type") == "follow.hold"]
    follower_frames = frames
    classic = [e for e in evs if e.get("type") == "mic.classic"]
    fires = [e for e in evs if e.get("type") == "accomp.fire"]
    cancelled = [e for e in evs if e.get("type") == "input.retryCancelled"]
    attempts = []
    for e in input_events:
        idx = e.get("scoreEventId", e.get("index"))
        pitches = e.get("pitches") or []
        aid = e.get("attemptId") or f"{meta.get('testRun', {}).get('runId', Path('trace').stem)}:e{idx}:a{e.get('attempt',0)}"
        a = {"scoreEventId": idx, "beat": e.get("beat"), "expected": pitches,
             "attempt": e.get("attempt", 0), "attemptId": aid,
             "playerIntentMs": _num(e.get("t")), "rawEventIds": [e["_id"]]}
        out = _nearest(audio, e["t"], lambda x: x.get("midi") in pitches, e["t"]-5, e["t"]+80)
        if out:
            a.update(audioScheduledMs=out["t"], physicalOutputMs=out["t"], audioTime=out.get("audioTime"), audioRawEventId=out["_id"])
            a["physicalMicArrivalMs"] = None
            a["physicalMicArrivalKnown"] = False
            a["detectionLagSnapshotMs"] = _num(lat.get("detectionMs"))
        else:
            a.update(audioScheduledMs=None, physicalOutputMs=None, audioTime=None,
                     physicalMicArrivalMs=None, physicalMicArrivalKnown=False,
                     detectionLagSnapshotMs=_num(lat.get("detectionMs")))
        # Frames logged before the synthetic note was emitted are residual
        # audio/queue state, not evidence for this attempt.
        lo = a.get("physicalOutputMs") if a.get("physicalOutputMs") is not None else e["t"]
        later_attempts = [x["t"] for x in input_events if x.get("t",0) > e["t"]]
        commit_probe = next((x["t"] for x in advances if x.get("toIndex") == idx and x.get("t",0) >= e["t"]), None)
        end_candidates = [x for x in (commit_probe, min(later_attempts, default=None), e["t"] + 3000) if x is not None]
        causal_end = min(end_candidates)
        a["causalWindow"] = {"startMs": lo, "endMs": causal_end,
                              "endReason": "commit" if commit_probe == causal_end else "nextAttempt" if later_attempts and min(later_attempts) == causal_end else "timeout"}
        fs = [f for f in follower_frames if lo <= f.get("t",0) <= causal_end]
        expected_idx = [p-min_midi for p in pitches if isinstance(p, int) and min_midi <= p <= min_midi+len((follower_frames[0].get("sal") if follower_frames else []))-1]
        onset = None
        for f in fs:
            sal = f.get("sal") or []
            evidence = min((sal[i] for i in expected_idx), default=999)
            if f.get("on", 0) > 0.3 and (f.get("inf") or evidence < 50):
                onset = f; break
        a["firstMicOnsetMs"] = onset.get("t") if onset else None
        a["onsetLatencyMs"] = round(onset["t"]-e["t"], 3) if onset else None
        if onset:
            sal = onset.get("sal") or []
            a["expectedEvidence"] = min((sal[i] for i in expected_idx), default=None)
            ranked = sorted(((v, min_midi+i) for i,v in enumerate(sal)), key=lambda x:x[0])[:5]
            a["topAcousticCandidates"] = [{"midi":m,"name":_name(m),"attenuationUnits":v,"dbBelowPeak":round(v*0.5,2)} for v,m in ranked]
        else:
            a["expectedEvidence"] = None; a["topAcousticCandidates"] = None
        causal_classic = [x for x in classic if lo <= x.get("t",0) <= causal_end]
        c_blind = next((x for x in causal_classic if x.get("yinMidi") is not None), None)
        c_shadow = causal_classic[0] if causal_classic else None
        post_shadow = next((x for x in classic if causal_end < x.get("t",0) <= causal_end+1000), None)
        a["blindPitch"] = ({"midi":c_blind.get("yinMidi"),"name":c_blind.get("yinName"),"clarity":c_blind.get("yinClarity"),"rawEventId":c_blind.get("_id")} if c_blind else None)
        a["classicShadow"] = ({k:c_shadow.get(k) for k in ("_id","t","midi","name","freq","clarity","yinMidi","yinName","rescue","reason","source")} if c_shadow else None)
        a["scoreBiasedShadow"] = ({k:c_shadow.get(k) for k in ("_id","t","midi","name","freq","clarity","rescue","reason","source")} if c_shadow and c_shadow.get("rescue") else None)
        a["postCommitShadow"] = ({k:post_shadow.get(k) for k in ("_id","t","midi","name","freq","clarity","yinMidi","yinName","rescue","reason","source")} if post_shadow else None)
        hs = [h for h in holds if h.get("index") == idx and lo <= h.get("t",0) <= causal_end]
        a["firstHoldReason"] = hs[0].get("reason") if hs else None
        # Preserve every decision snapshot in the causal window. These are
        # observations of the existing commit rule, not reconstructed fits.
        a["holdTimeline"] = [{k:v for k,v in h.items() if k != "_id"} for h in hs]
        hold_candidates = [h.get("candidate") for h in hs if isinstance(h.get("candidate"), dict)]
        peak_p = max(hold_candidates, key=lambda c: c.get("p", -math.inf), default=None)
        peak_ev = max(hold_candidates, key=lambda c: c.get("ev", -math.inf), default=None)
        a["peakPosteriorHold"] = peak_p
        a["peakEvidenceHold"] = peak_ev
        adv = _nearest(advances, e["t"], lambda x: x.get("toIndex") == idx, e["t"]-20, causal_end)
        if adv:
            a["commitMs"] = adv["t"]; a["playerToCommitMs"] = round(adv["t"]-e["t"],3); a["advanceRawEventId"] = adv["_id"]
            a["attribution"] = adv.get("explained")
            a["heardMidi"] = adv.get("heardMidi")
            hyps = [h for h in adv.get("hyps",[]) if h.get("i") != idx]
            a["strongestCompetitor"] = max(hyps, key=lambda h:h.get("w",-1), default=None)
        else:
            a.update(commitMs=None, playerToCommitMs=None, advanceRawEventId=None, attribution=None, heardMidi=None, strongestCompetitor=None)
        a["retryRequired"] = any(x.get("scoreEventId",x.get("index")) == idx and x.get("attempt",0) > e.get("attempt",0) for x in input_events)
        a["retryCancelled"] = any(x.get("scoreEventId",x.get("index")) == idx and x.get("attempt",0) == e.get("attempt",0) for x in cancelled)
        fire = _nearest(fires, a.get("commitMs") or e["t"], lambda x: x.get("t",0) >= (a.get("commitMs") or e["t"]), (a.get("commitMs") or e["t"]), (a.get("commitMs") or e["t"])+2500)
        a["accompanimentResponseMs"] = fire.get("t") if fire else None
        a["accompanimentRawEventId"] = fire.get("_id") if fire else None
        # Causal status, deliberately conservative.
        ex = a.get("attribution") or {}
        reasons = [h.get("reason") for h in hs]
        if a["retryCancelled"]: status = "STALE_RETRY_CANCELLED"
        elif adv and e.get("attempt",0) == 0 and not a["retryRequired"]: status = "PASS_FIRST"
        elif adv: status = "PASS_RETRY"
        elif ex.get("bleed",0) >= .5 and ex.get("new",1) < .1: status = "BLEED_DOMINATED"
        elif ex.get("prev",0) >= .4 and ex.get("new",1) <= .01: status = "PREVIOUS_DOMINATED"
        elif "low-posterior" in reasons: status = "LOW_POSTERIOR"
        elif onset: status = "ACOUSTIC_SEEN_FOLLOWER_HOLD"
        elif a.get("physicalOutputMs") is not None: status = "MIC_MISS"
        else: status = "UNRESOLVED"
        a["status"] = status
        a["stage"] = ("F" if adv and not fire else "E" if adv else "D" if hs and onset else "C" if onset else "A")
        a["heard"] = {"onsetPresent": onset is not None, "expectedEvidence": a.get("expectedEvidence"), "blindPitch": a.get("blindPitch"), "topFrontendRows": a.get("topAcousticCandidates")}
        c = peak_p or peak_ev or {}
        cx = c.get("explained") or {}
        a["recognized"] = {"new": ex.get("new") if ex else cx.get("new"), "prev": ex.get("prev") if ex else cx.get("prev"), "bleed": ex.get("bleed") if ex else cx.get("bleed"), "wrongOther": ex.get("other") if ex else cx.get("other"), "posterior": adv.get("p") if adv else (peak_p or {}).get("p"), "evidence": adv.get("ev") if adv else (peak_ev or {}).get("ev"), "distinct": c.get("distinct"), "pWrong": (hs[-1].get("wrong") or {}).get("p") if hs else None}
        a["accepted"] = {"commit": adv is not None, "commitMs": a.get("commitMs"), "firstHoldReason": a.get("firstHoldReason"), "guardRefusal": any(x.get("type")=="follow.guard" and lo <= x.get("t",0) <= causal_end for x in evs)}
        attempts.append(a)
    # Every accepted advance must have a prior attempt, otherwise it is an orphan.
    orphans=[]
    for adv in advances:
        idx=adv.get("toIndex"); prior=[a for a in attempts if a["scoreEventId"]==idx and a["playerIntentMs"] <= adv["t"]]
        if not prior:
            future=[a for a in attempts if a["scoreEventId"]==idx and a["playerIntentMs"] > adv["t"]]
            orphans.append({"advanceMs":adv["t"],"expectedEvent":idx,"mostRecentPriorAttempt":None,
                            "timeUntilActualAttemptMs":round(future[0]["playerIntentMs"]-adv["t"],3) if future else None,
                            "explained":adv.get("explained"),"heardMidi":adv.get("heardMidi"),"rawEventId":adv["_id"]})
    for a in attempts:
        if any(o["expectedEvent"]==a["scoreEventId"] and o["advanceMs"] < a["playerIntentMs"] for o in orphans): a["status"]="FALSE_ADVANCE"
    checks=[]
    for a in attempts:
        # scoreBiasedShadow is deliberately kept in its own field and never
        # copied into blindPitch; this check documents that invariant.
        if a.get("blindPitch") and a.get("blindPitch", {}).get("rawEventId") == (a.get("scoreBiasedShadow") or {}).get("_id"):
            checks.append({"code":"SCORE_BIASED_PITCH_AS_ACOUSTIC","scoreEventId":a["scoreEventId"],"severity":"error"})
    checks += [{"code":"ADVANCE_WITHOUT_PLAYER_ATTEMPT","rawEventId":o["rawEventId"],"severity":"error"} for o in orphans]
    checks += [{"code":"ATTEMPT_WITHOUT_OUTCOME","scoreEventId":a["scoreEventId"],"attempt":a["attempt"],"severity":"error"} for a in attempts if not a.get("status")]
    md = meta.get("micDistance") or {}
    refused = md.get("refused")
    run_quality = {"aboveRoomDb": _num(md.get("aboveRoomDb")), "band": md.get("band"), "refused": refused if isinstance(refused, bool) else None, "suitableForTuning": (not refused) if isinstance(refused, bool) else None}
    return {"schema":"notepilot.note-alignment.v3","source":"raw NPTrace events","clock":{"canonical":"events[].t (session-relative performance.now ms)","sessionMs":"observed monotonic trace time","audioTimeSec":"raw AudioContext.currentTime, observed","sampleIndex":"follow.frame.at, observed","frameIndex":"follow.frame.k, observed","frame":{"sampleRate":sr,"hop":hop,"cadenceMs":round(hop*1000/sr,3),"frameSampleMs":"(at-first.at)*1000/sr"}},"latencySemantics":{"baseMs":"classic speaker-probe detection/round-trip calibration value and slider fallback","micMs":"legacy alias for detectionMs; not physical mic propagation","detectionMs":"snapshot of detection-lag compensation term","outputMs":"configured output-delay slider","outputUsedMs":"browser output latency applied","effectiveMs":"detection plus output compensation; not an arrival timestamp","physicalMicArrivalKnown":False},"causalWindow":"starts at observed audio.note/player output and ends at earliest successful commit, next attempt, or +3000ms timeout","runQuality":run_quality,"attempts":attempts,"orphanAdvances":orphans,"sanityChecks":checks,"rawEventCount":len(evs)}

def print_report(out, event=None, window=1000, trace=None):
    rows=out["attempts"]
    print("evt note try heard maxP ev@maxP distinct new prev bleed wrongP maxEv p@maxEv result")
    for a in rows:
        if event is not None and a["scoreEventId"] != event: continue
        note="/".join(_name(p) for p in a["expected"])
        rec=a.get("recognized") or {}; pp=a.get("peakPosteriorHold") or {}; pe=a.get("peakEvidenceHold") or {}; ex=pp.get("explained") or pe.get("explained") or {}
        vals=[pp.get('p'),pp.get('ev'),pp.get('distinct'),ex.get('new'),ex.get('prev'),ex.get('bleed'),rec.get('pWrong'),pe.get('ev'),pe.get('p')]
        print(f"{a['scoreEventId']:>3} {note:<4} {a['attempt']:>3} {'YES' if a.get('heard',{}).get('onsetPresent') else 'NO':>5} " + " ".join(f"{_cell(v):>8}" for v in vals) + f" {a['status']}")
    vals=[a["playerToCommitMs"] for a in rows if a.get("playerToCommitMs") is not None]
    print(f"\nFALSE ADVANCES: {len(out['orphanAdvances'])}")
    print(f"STALE RETRIES: {sum(a.get('retryCancelled',False) for a in rows)}")
    print(f"FIRST-STRIKE PASS: {sum(a['status']=='PASS_FIRST' for a in rows)}/{len(rows)}")
    print(f"MEDIAN player→commit: {statistics.median(vals):.1f} ms" if vals else "MEDIAN player→commit: --")
    print(f"WORST player→commit: {max(vals):.1f} ms" if vals else "WORST player→commit: --")
    if event is not None and trace is not None:
        target = next((a for a in rows if a["scoreEventId"] == event), None)
        if target:
            lo = target["playerIntentMs"] - window; hi = target["playerIntentMs"] + window
            print(f"\nRAW TIMELINE event {event} ({lo:.1f}..{hi:.1f} ms)")
            for i, e in enumerate(trace.get("events", [])):
                if lo <= e.get("t", -1e9) <= hi and e.get("type") in {
                    "input.testEvent","input.key","audio.note","follow.frame","follow.hold",
                    "follow.advance","follow.guard","mic.classic","mic.display","accomp.fire",
                    "input.retryCancelled","tracker"
                }:
                    labels={"input.testEvent":"PLAYER","input.key":"PLAYER","audio.note":"AUDIO_OUT","follow.frame":"FOLLOWER","follow.hold":"FOLLOWER","follow.advance":"FOLLOWER","follow.guard":"GUARD","mic.classic":"CLASSIC_SHADOW","mic.display":"CLASSIC_SHADOW","accomp.fire":"ACCOMP","input.retryCancelled":"GUARD","tracker":"TRACKER"}
                    detail={k:e.get(k) for k in ("index","attempt","scoreEventId","midi","name","toIndex","reason","p","margin","ev","heardMidi","explained","best","pos","on","source","text") if k in e}
                    print(f"  [{i:04d}] {e.get('t'):8.1f} {labels.get(e.get('type'),'TRACE'):<14} {e.get('type'):<22} {detail}")

def main():
    ap=argparse.ArgumentParser(); ap.add_argument("trace"); ap.add_argument("--event",type=int); ap.add_argument("--window-ms",type=int,default=1000); ap.add_argument("--no-write",action="store_true")
    args=ap.parse_args(); p=Path(args.trace); trace=json.loads(p.read_text()); out=build_alignment(trace)
    if not args.no_write:
        p.with_name(p.name+".alignment.json").write_text(json.dumps(out,indent=2)+"\n")
        fields=sorted({k for a in out['attempts'] for k in a.keys()});
        with p.with_name(p.name+".alignment.csv").open('w',newline='') as f:
            w=csv.DictWriter(f,fieldnames=fields); w.writeheader();
            for a in out['attempts']: w.writerow({k:json.dumps(a.get(k)) if isinstance(a.get(k),(dict,list)) else a.get(k) for k in fields})
    print_report(out,args.event,args.window_ms,trace)
if __name__ == '__main__': main()