Spaces:
Running on Zero
Running on Zero
| """ | |
| A/B comparison on the frozen probe set. Compares two result dirs without | |
| re-transcribing. Default: per-channel (A) vs per-channel+preprocess (B). | |
| The 'base' column is the static mono-mix baseline_acc recorded in probe_set.json. | |
| Also reports: | |
| - no_speech_prob diagnostics per dir (mean, #high-ns segments, total words) to | |
| show whether preprocessing reduced Whisper's silence-hallucination signal | |
| - a confidence-filter experiment on B (drop words with segment ns > threshold) | |
| Usage: | |
| python compare_probe.py # chan vs chan+pp | |
| python compare_probe.py --a-dir results --a-label mix --b-dir results_channels --b-label chan | |
| """ | |
| import os | |
| import json | |
| import argparse | |
| import jiwer | |
| from eval_common import normalise | |
| DATA_DIR = r"d:\Desktop\ai-ml-capstone\data\na_testset" | |
| MANIFEST = os.path.join(DATA_DIR, "manifest.json") | |
| PROBE_SET = os.path.join(DATA_DIR, "probe_set.json") | |
| def load_result(root, accent, cid): | |
| p = os.path.join(DATA_DIR, root, accent, cid + ".json") | |
| if not os.path.exists(p): | |
| return None | |
| with open(p, encoding="utf-8") as f: | |
| return json.load(f) | |
| def text_of(words, ns_max=None): | |
| if ns_max is None: | |
| return " ".join(w["word"] for w in words) | |
| return " ".join(w["word"] for w in words if w.get("ns", 0.0) <= ns_max) | |
| def score_call(ref_a, ref_c, hyp_a, hyp_c): | |
| ra, rc = normalise(ref_a), normalise(ref_c) | |
| ha, hc = normalise(hyp_a), normalise(hyp_c) | |
| na, nc = len(ra.split()), len(rc.split()) | |
| if na + nc == 0: | |
| return None, 0 | |
| wa = jiwer.wer(ra, ha) if na else 0.0 | |
| wc = jiwer.wer(rc, hc) if nc else 0.0 | |
| return (wa * na + wc * nc) / (na + nc), na + nc | |
| def acc(wer): | |
| return None if wer is None else (1 - wer) * 100 | |
| def ns_diag(result): | |
| """(#high-ns segments, total segments, total hyp words) from captured segments.""" | |
| if not result or "segments" not in result: | |
| return None | |
| segs = result["segments"].get("agent", []) + result["segments"].get("customer", []) | |
| hi = sum(1 for s in segs if s.get("no_speech_prob", 0) > 0.6) | |
| words = len(result.get("agent", [])) + len(result.get("customer", [])) | |
| return hi, len(segs), words | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--a-dir", default="results_channels"); ap.add_argument("--a-label", default="chan") | |
| ap.add_argument("--b-dir", default="results_channels_pp"); ap.add_argument("--b-label", default="chan+pp") | |
| ap.add_argument("--ns-thresholds", nargs="*", type=float, default=[0.6, 0.8]) | |
| args = ap.parse_args() | |
| with open(MANIFEST, encoding="utf-8") as f: | |
| manifest = {m["call_id"]: m for m in json.load(f)} | |
| with open(PROBE_SET, encoding="utf-8") as f: | |
| probe = json.load(f)["calls"] | |
| print(f"{'tier':<13} {'call_id':<32} {'base':>6} {args.a_label:>7} {args.b_label:>8} {'delta':>7}") | |
| print("-" * 80) | |
| agg = {"a": [0.0, 0], "b": [0.0, 0]} | |
| agg_ns = {t: [0.0, 0] for t in args.ns_thresholds} | |
| nsA = [0, 0, 0]; nsB = [0, 0, 0] # hi, segs, words | |
| missing = [] | |
| for p in probe: | |
| cid, accent = p["call_id"], p["accent"] | |
| m = manifest[cid] | |
| ra_, rc_ = m["agent_transcript"], m["customer_transcript"] | |
| A = load_result(args.a_dir, accent, cid) | |
| B = load_result(args.b_dir, accent, cid) | |
| if A is None or B is None: | |
| missing.append((cid, args.a_dir if A is None else args.b_dir)) | |
| continue | |
| a_wer, n = score_call(ra_, rc_, text_of(A["agent"]), text_of(A["customer"])) | |
| b_wer, _ = score_call(ra_, rc_, text_of(B["agent"]), text_of(B["customer"])) | |
| agg["a"][0] += a_wer * n; agg["a"][1] += n | |
| agg["b"][0] += b_wer * n; agg["b"][1] += n | |
| for t in args.ns_thresholds: | |
| w, _ = score_call(ra_, rc_, text_of(B["agent"], t), text_of(B["customer"], t)) | |
| agg_ns[t][0] += w * n; agg_ns[t][1] += n | |
| for store, res in [(nsA, A), (nsB, B)]: | |
| d = ns_diag(res) | |
| if d: | |
| store[0] += d[0]; store[1] += d[1]; store[2] += d[2] | |
| delta = acc(b_wer) - acc(a_wer) | |
| flag = " <<" if delta >= 3 else (" !!" if delta <= -3 else "") | |
| print(f"{p['tier']:<13} {cid:<32} {p['baseline_acc']*100:>5.1f}% " | |
| f"{acc(a_wer):>6.1f}% {acc(b_wer):>7.1f}% {delta:>+6.1f}{flag}") | |
| print("-" * 80) | |
| a_acc = acc(agg["a"][0] / agg["a"][1]) | |
| b_acc = acc(agg["b"][0] / agg["b"][1]) | |
| print(f"{'OVERALL (word-weighted)':<46} {a_acc:>6.1f}% {b_acc:>7.1f}% {b_acc - a_acc:>+6.1f}") | |
| print(f"\nno_speech_prob diagnostics (hallucination signal):") | |
| print(f" {args.a_label:<10} high-ns segs: {nsA[0]:>4} / {nsA[1]:<5} segments | total hyp words: {nsA[2]}") | |
| print(f" {args.b_label:<10} high-ns segs: {nsB[0]:>4} / {nsB[1]:<5} segments | total hyp words: {nsB[2]}") | |
| print(f"\nConfidence filter on {args.b_label} (drop words where segment no_speech_prob > t):") | |
| print(f" no filter : {b_acc:5.1f}%") | |
| for t in args.ns_thresholds: | |
| a = acc(agg_ns[t][0] / agg_ns[t][1]) | |
| print(f" ns <= {t:<4} : {a:5.1f}% ({a - b_acc:+.1f})") | |
| if missing: | |
| print("\nMissing results:") | |
| for cid, which in missing: | |
| print(f" {cid} ({which})") | |
| if __name__ == "__main__": | |
| main() | |