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