File size: 4,679 Bytes
5c2beba | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | #!/usr/bin/env python3
"""Tiny eval-comparison server: http://<box>:8080
Auto-discovers /opt/work/eval/<system>/ dirs (timing.jsonl present), shows each
eval sentence with side-by-side audio players + Soniox WER badges. Refresh to
pick up newly evaluated checkpoints.
"""
import json
import re
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
EVAL = Path("/opt/work/eval")
SENTS = Path("/opt/work/scripts/eval_sentences.jsonl")
def discover():
systems = []
for d in sorted(EVAL.iterdir()):
if d.is_dir() and any(d.glob("*.wav")) and d.name != "baseline":
systems.append(d.name)
# baseline first, then stepN numerically
def key(n):
if n.startswith("baseline"):
return (0, 0)
m = re.search(r"(\d+)", n)
return (1, int(m.group(1)) if m else 9e9)
return sorted(systems, key=key)
def reports(systems):
out = {}
for s in systems:
for fn in ("asr_report_soniox.json", "asr_report.json"):
p = EVAL / s / fn
if p.exists():
try:
rep = json.loads(p.read_text(encoding="utf-8"))
out[s] = {
"rows": {r["id"]: r for r in rep.get("rows", []) if "id" in r},
"summary": rep.get("summary", {}),
"engine": rep.get("summary", {}).get("engine", "whisper"),
}
except Exception:
pass
break
return out
def render():
systems = discover()
reps = reports(systems)
sents = [json.loads(l) for l in SENTS.read_text(encoding="utf-8").splitlines() if l.strip()]
# include ids that only exist in system timing files (e.g. mega paragraph)
known = {s["id"] for s in sents}
for s in systems:
tj = EVAL / s / "timing.jsonl"
if tj.exists():
for line in tj.read_text(encoding="utf-8").splitlines():
if line.strip():
r = json.loads(line)
if r.get("id") and r["id"] not in known and r.get("text"):
sents.insert(0, {"id": r["id"], "text": r["text"]})
known.add(r["id"])
head = "".join(
f"<th>{s}<br><small>{('WER %.0f%%' % (100*reps[s]['summary'].get('mean_wer',0))) if s in reps else 'scoring...'}</small></th>"
for s in systems)
rows_html = []
for sent in sents:
sid = sent["id"]
cells = []
for s in systems:
wav = EVAL / s / f"{sid}.wav"
if wav.exists():
badge = ""
r = reps.get(s, {}).get("rows", {}).get(sid)
if r and "wer" in r:
c = "#2a4" if r["wer"] < 0.2 else ("#a82" if r["wer"] < 0.5 else "#a33")
badge = f'<span class="b" style="background:{c}">WER {r["wer"]:.2f}</span>'
cells.append(
f'<td><audio controls preload="none" src="/{s}/{sid}.wav"></audio>{badge}</td>')
else:
cells.append("<td>—</td>")
rows_html.append(
f'<tr><td class="t"><b>{sid}</b><div dir="rtl">{sent["text"]}</div></td>'
+ "".join(cells) + "</tr>")
return f"""<!doctype html><html><head><meta charset="utf-8">
<title>s2-pro Egyptian eval</title>
<style>
body{{font-family:system-ui;background:#111;color:#eee;margin:20px}}
table{{border-collapse:collapse;width:100%}}
td,th{{border:1px solid #333;padding:8px;vertical-align:top}}
th{{background:#222;position:sticky;top:0}}
.t{{max-width:420px}} .t div{{font-size:15px;line-height:1.7}}
audio{{width:230px;display:block}}
.b{{font-size:11px;padding:2px 6px;border-radius:4px;display:inline-block;margin-top:4px}}
h1{{font-size:18px}}
</style></head><body>
<h1>s2-pro Egyptian fine-tune — A/B eval (refresh for new checkpoints)</h1>
<table><tr><th class="t">sentence</th>{head}</tr>
{''.join(rows_html)}
</table></body></html>"""
class H(SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=str(EVAL), **kw)
def do_GET(self):
if self.path in ("/", "/index.html"):
body = render().encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
else:
super().do_GET()
def log_message(self, *a):
pass
if __name__ == "__main__":
ThreadingHTTPServer(("0.0.0.0", 8080), H).serve_forever()
|