#!/usr/bin/env python3 # Published copy, nothing is changed except this line. The sha256 of the original is in sha256.txt. """WER of a transcript against the LibriSpeech reference of the test file. Text normalization (documented in README-WER.md): openasr PRIMARY. whisper_normalizer.english.EnglishTextNormalizer (the normalizer family used by the Hugging Face Open ASR Leaderboard) applied to reference AND hypothesis: lower case, removes bracketed text, fillers (hmm, uh ...), expands contractions (isn't -> is not), spelled-out numbers <-> digits ("twenty" -> 20), British -> American spelling, strips punctuation and symbols. basic SECONDARY. lower case, every character outside a-z 0-9 apostrophe becomes a space, whitespace collapsed. No number, spelling or contraction handling. Stricter, so usually a higher WER. WER = (S + D + I) / N, N = reference words after normalization, word-level Levenshtein (jiwer / rapidfuzz). Inputs: --ref reference-librispeech.txt --manifest manifest.json --hyp FILE (nemo-speech --format json output, or plain text) or --hyp-chunks DIR (chunk-00.json|txt ... chunk-NN.json|txt, concatenated in order) Outputs: --out result.json --errors errors.tsv (one line per error with ref/hyp context) """ import argparse import collections import json import re import sys import warnings from pathlib import Path warnings.filterwarnings("ignore") import jiwer # noqa: E402 from whisper_normalizer.english import EnglishTextNormalizer # noqa: E402 _EN = EnglishTextNormalizer() def norm_openasr(s: str) -> str: return _EN(s) def norm_basic(s: str) -> str: s = s.lower().replace("’", "'") s = re.sub(r"[^a-z0-9']+", " ", s) return re.sub(r"\s+", " ", s).strip() NORMS = {"openasr": norm_openasr, "basic": norm_basic} def read_hyp(path: Path) -> str: raw = path.read_text(encoding="utf-8") if path.suffix == ".json": j = json.loads(raw) if isinstance(j, dict) and "text" in j: return j["text"] if isinstance(j, dict) and "words" in j: return " ".join(w["word"] for w in j["words"]) raise SystemExit(f"{path}: JSON without text/words") return raw def evaluate(ref_chapters, hyp_text, norm): """ref_chapters: list of (chapter_id, raw_ref_text). Returns result dict and error list.""" ref_words, bounds = [], [] for cid, txt in ref_chapters: w = norm(txt).split() bounds.append((cid, len(ref_words), len(ref_words) + len(w))) ref_words.extend(w) hyp_words = norm(hyp_text).split() ref_s, hyp_s = " ".join(ref_words), " ".join(hyp_words) out = jiwer.process_words(ref_s, hyp_s) S, D, I, H = out.substitutions, out.deletions, out.insertions, out.hits N = len(ref_words) per = {cid: {"ref_words": e - s, "S": 0, "D": 0, "I": 0} for cid, s, e in bounds} def chap_of(i): i = min(max(i, 0), N - 1) for cid, s, e in bounds: if s <= i < e: return cid return bounds[-1][0] errors, subs, dels, inss = [], collections.Counter(), collections.Counter(), collections.Counter() for ch in out.alignments[0]: t = ch.type if t == "equal": continue r0, r1, h0, h1 = ch.ref_start_idx, ch.ref_end_idx, ch.hyp_start_idx, ch.hyp_end_idx ref_seg, hyp_seg = ref_words[r0:r1], hyp_words[h0:h1] cid = chap_of(r0) if t == "substitute": per[cid]["S"] += r1 - r0 for a, b in zip(ref_seg, hyp_seg): subs[(a, b)] += 1 elif t == "delete": per[cid]["D"] += r1 - r0 for a in ref_seg: dels[a] += 1 elif t == "insert": per[cid]["I"] += h1 - h0 for b in hyp_seg: inss[b] += 1 errors.append((t, cid, r0, " ".join(ref_words[max(0, r0 - 4):r0]), " ".join(ref_seg), " ".join(hyp_seg), " ".join(ref_words[r1:r1 + 4]))) for cid, v in per.items(): v["wer_pct"] = round(100.0 * (v["S"] + v["D"] + v["I"]) / max(1, v["ref_words"]), 3) tail_ref = ref_words[-30:] tail_hyp = hyp_words[-30:] tail_match = sum(a == b for a, b in zip(tail_ref, tail_hyp)) / max(1, len(tail_ref)) res = { "wer_pct": round(100.0 * (S + D + I) / N, 4), "S": S, "D": D, "I": I, "hits": H, "N_ref_words": N, "N_hyp_words": len(hyp_words), "hyp_over_ref_words": round(len(hyp_words) / N, 5), "tail_30_words_exact_match": round(tail_match, 3), "top_substitutions": [[a, b, c] for (a, b), c in subs.most_common(25)], "top_deletions": dels.most_common(15), "top_insertions": inss.most_common(15), "per_chapter": per, } return res, errors def main(): ap = argparse.ArgumentParser() ap.add_argument("--ref", required=True) ap.add_argument("--manifest", required=True) g = ap.add_mutually_exclusive_group(required=True) g.add_argument("--hyp") g.add_argument("--hyp-chunks") ap.add_argument("--chunk-list", help="JSON list of {file: chunk-NN.wav} in order (default: the chunks of the manifest)") ap.add_argument("--out", required=True) ap.add_argument("--errors") a = ap.parse_args() man = json.loads(Path(a.manifest).read_text(encoding="utf-8")) by_chap = collections.OrderedDict() for u in man["utterances"]: by_chap.setdefault(u["chapter"], []).append(u["text"]) ref_lines = Path(a.ref).read_text(encoding="utf-8").splitlines() assert [t for ts in by_chap.values() for t in ts] == ref_lines, "reference file does not match manifest" ref_chapters = [(cid, " ".join(ts)) for cid, ts in by_chap.items()] if a.hyp: hyp_text = read_hyp(Path(a.hyp)) src = a.hyp else: d = Path(a.hyp_chunks) parts = [] for ch in (json.load(open(a.chunk_list)) if a.chunk_list else man["chunks"]): stem = ch["file"].rsplit(".", 1)[0] f = d / f"{stem}.json" if not f.exists(): f = d / f"{stem}.txt" if not f.exists(): raise SystemExit(f"missing chunk transcript {stem}.json|txt in {d}") parts.append(read_hyp(f)) hyp_text = " ".join(parts) src = a.hyp_chunks result = {"hyp_source": src, "ref_source": a.ref, "manifest_wav_sha256": man["wav_sha256"], "audio_duration_s": man["duration_s"], "normalizers": {}} all_errors = {} for name, fn in NORMS.items(): res, errs = evaluate(ref_chapters, hyp_text, fn) result["normalizers"][name] = res all_errors[name] = errs Path(a.out).write_text(json.dumps(result, indent=1, ensure_ascii=False), encoding="utf-8") if a.errors: with open(a.errors, "w", encoding="utf-8") as f: f.write("normalizer\ttype\tchapter\tref_idx\tleft_context\tref\thyp\tright_context\n") for name, errs in all_errors.items(): for e in errs: f.write(name + "\t" + "\t".join(str(x) for x in e) + "\n") for name, res in result["normalizers"].items(): print(f"{name}: WER {res['wer_pct']:.3f} % (S {res['S']} D {res['D']} I {res['I']}, N {res['N_ref_words']}, " f"hyp/ref words {res['hyp_over_ref_words']}, tail match {res['tail_30_words_exact_match']})") if __name__ == "__main__": sys.exit(main())