#!/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()