File size: 2,564 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 | #!/usr/bin/env python3
"""ASR-judge a folder of synthesized wavs: faster-whisper large-v3 -> WER/CER.
Compares ASR transcript vs reference text (both normalized: tashkeel stripped,
punctuation removed, alef/yaa variants unified) using jiwer.
Usage: python asr_eval.py --wavdir /opt/work/eval/baseline
Writes <wavdir>/asr_report.json and prints a summary table.
"""
import argparse
import json
import re
from pathlib import Path
import jiwer
TASHKEEL_RE = re.compile("[\u0610-\u061a\u064b-\u065f\u0670\u06d6-\u06dc\u06df-\u06e8\u06ea-\u06ed\u0640]")
PUNCT_RE = re.compile(r"[^\w\s]|[_]", re.UNICODE)
WS_RE = re.compile(r"\s+")
def norm(t: str) -> str:
t = TASHKEEL_RE.sub("", t)
t = t.replace("أ", "ا").replace("إ", "ا").replace("آ", "ا")
t = t.replace("ى", "ي").replace("ة", "ه")
t = PUNCT_RE.sub(" ", t)
return WS_RE.sub(" ", t).strip()
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--wavdir", required=True)
ap.add_argument("--model", default="large-v3")
args = ap.parse_args()
wavdir = Path(args.wavdir)
refs = {}
for line in (wavdir / "timing.jsonl").read_text(encoding="utf-8").splitlines():
r = json.loads(line)
if "text" in r:
refs[r["id"]] = r["text"]
from faster_whisper import WhisperModel
m = WhisperModel(args.model, device="cuda", compute_type="float16")
rows = []
for sid, ref in refs.items():
wav = wavdir / f"{sid}.wav"
if not wav.exists():
rows.append({"id": sid, "error": "missing_wav"})
continue
segs, _ = m.transcribe(str(wav), language="ar", beam_size=5, vad_filter=False)
hyp = " ".join(s.text for s in segs).strip()
r_n, h_n = norm(ref), norm(hyp)
wer = jiwer.wer(r_n, h_n) if r_n else 1.0
cer = jiwer.cer(r_n, h_n) if r_n else 1.0
rows.append(
{"id": sid, "ref": ref, "hyp": hyp, "wer": round(wer, 3), "cer": round(cer, 3)}
)
print(f"{sid:12s} WER={wer:.2f} CER={cer:.2f} | {hyp[:70]}")
ok = [r for r in rows if "wer" in r]
summary = {
"n": len(rows),
"n_ok": len(ok),
"mean_wer": round(sum(r["wer"] for r in ok) / max(len(ok), 1), 4),
"mean_cer": round(sum(r["cer"] for r in ok) / max(len(ok), 1), 4),
}
print("SUMMARY", json.dumps(summary))
(wavdir / "asr_report.json").write_text(
json.dumps({"summary": summary, "rows": rows}, ensure_ascii=False, indent=1),
encoding="utf-8",
)
if __name__ == "__main__":
main()
|