#!/usr/bin/env python3 import io import json import re import time import unicodedata from pathlib import Path import jiwer import numpy as np import pyarrow.parquet as pq import soundfile as sf PERSIAN_DIGITS = "۰۱۲۳۴۵۶۷۸۹٠١٢٣٤٥٦٧٨٩" ASCII_DIGITS = "01234567890123456789" DIGIT_MAP = str.maketrans(PERSIAN_DIGITS, ASCII_DIGITS) PUNCT_RE = re.compile(r"[،؛؟,.\!?\(\)\[\]\{\}\"'`«»“”‘’\-_=+/\\:;]") def normalize(text): text = unicodedata.normalize("NFKC", text or "") text = text.replace("‌", " ").replace("​", "") text = text.replace("ي", "ی").replace("ك", "ک").translate(DIGIT_MAP) text = PUNCT_RE.sub(" ", text) return re.sub(r"\s+", " ", text).strip() def read_audio(path_or_bytes, base_dir=None): if isinstance(path_or_bytes, (bytes, bytearray)): wav, sr = sf.read(io.BytesIO(path_or_bytes), dtype="float32") else: path = Path(path_or_bytes) if base_dir is not None and not path.is_absolute(): path = Path(base_dir) / path wav, sr = sf.read(path, dtype="float32") if wav.ndim > 1: wav = wav.mean(axis=-1) return wav, sr def load_gold(path="/workspace/golha/gold.jsonl"): base = Path(path).parent rows = [] for line in open(path): item = json.loads(line) ref = (item.get("gold_transcript") or "").strip() if not ref: continue wav, sr = read_audio(item["audio_path"], base) rows.append({"id": str(item.get("programmeitem_id")), "wav": wav, "sr": sr, "ref": ref}) return rows def load_fleurs(fleurs_dir="/workspace/golha/persian-eval"): files = sorted(Path(fleurs_dir).glob("data/*.parquet")) or sorted(Path(fleurs_dir).glob("*.parquet")) records = pq.read_table([str(p) for p in files]).to_pylist() rows = [] for idx, item in enumerate(records): audio = item["audio"] audio_bytes = audio.get("bytes") if isinstance(audio, dict) else None if audio_bytes is None: continue wav, sr = read_audio(audio_bytes) rows.append( { "id": str(item.get("id", idx)), "wav": wav, "sr": sr, "ref": item.get("transcription") or item.get("raw_transcription") or "", } ) return rows def ensure_wavs(set_name, clips): wav_dir = Path("/workspace/golha/nemo_eval_wavs") / set_name wav_dir.mkdir(parents=True, exist_ok=True) rows = [] for i, clip in enumerate(clips): safe_id = re.sub(r"[^A-Za-z0-9_.-]+", "_", clip["id"] or str(i)) path = wav_dir / f"{i:04d}_{safe_id}.wav" if not path.exists(): sf.write(path, clip["wav"], clip["sr"]) rows.append({**clip, "path": str(path)}) return rows def hyp_text(item): if isinstance(item, str): return item if hasattr(item, "text"): return item.text or "" if isinstance(item, dict): return item.get("text") or item.get("pred_text") or str(item) return str(item or "") def eval_set(model, set_name, clips, pred_dir, batch_size=16): clips = ensure_wavs(set_name, clips) refs = [] preds = [] pred_path = pred_dir / f"nvidia__stt_fa_fastconformer_hybrid_large__{set_name}_predictions.jsonl" start = time.time() batch_times = [] with pred_path.open("w") as f: for start_i in range(0, len(clips), batch_size): batch = clips[start_i : start_i + batch_size] t0 = time.time() hyps = model.transcribe([c["path"] for c in batch], batch_size=len(batch)) elapsed = time.time() - t0 batch_times.extend([elapsed / max(1, len(batch))] * len(batch)) for clip, hyp_obj in zip(batch, hyps): hyp = hyp_text(hyp_obj) ref_norm = normalize(clip["ref"]) hyp_norm = normalize(hyp) refs.append(ref_norm) preds.append(hyp_norm) f.write( json.dumps( {"id": clip["id"], "ref": clip["ref"], "hyp": hyp, "ref_norm": ref_norm, "hyp_norm": hyp_norm}, ensure_ascii=False, ) + "\n" ) done = min(start_i + batch_size, len(clips)) if done % 64 == 0 or done == len(clips): print(f"[progress] nvidia/stt_fa_fastconformer_hybrid_large {set_name} {done}/{len(clips)} mean_ms={np.mean(batch_times)*1000:.1f}", flush=True) return { "model": "nvidia/stt_fa_fastconformer_hybrid_large", "repo": "nvidia/stt_fa_fastconformer_hybrid_large", "family": "NVIDIA FastConformer Hybrid", "precision": "fp32", "params_b": 0.115, "set": set_name, "n": len(clips), "wer": jiwer.wer(refs, preds), "cer": jiwer.cer(refs, preds), "mean_decode_ms": float(np.mean(batch_times) * 1000.0), "wall_sec": time.time() - start, "predictions": str(pred_path), "status": "complete", } def main(): import torch import nemo.collections.asr as nemo_asr pred_dir = Path("/workspace/golha/public_asr_double_benchmark_predictions") pred_dir.mkdir(exist_ok=True) out = Path("/workspace/golha/public_asr_nvidia_benchmark.jsonl") done = set() if out.exists(): for line in out.read_text().splitlines(): try: row = json.loads(line) done.add((row.get("model"), row.get("set"))) except Exception: pass print("[load] nvidia/stt_fa_fastconformer_hybrid_large", flush=True) model = nemo_asr.models.ASRModel.from_pretrained("nvidia/stt_fa_fastconformer_hybrid_large") model = model.to("cuda" if torch.cuda.is_available() else "cpu") model.eval() evalsets = {"gold69": load_gold(), "fleurs": load_fleurs()} with out.open("a", buffering=1) as f: for set_name, clips in evalsets.items(): key = ("nvidia/stt_fa_fastconformer_hybrid_large", set_name) if key in done: print("[skip] nvidia/stt_fa_fastconformer_hybrid_large", set_name, flush=True) continue print("[eval] nvidia/stt_fa_fastconformer_hybrid_large", set_name, "n=" + str(len(clips)), flush=True) rec = eval_set(model, set_name, clips, pred_dir) print(f"[result] nvidia/stt_fa_fastconformer_hybrid_large {set_name} WER={rec['wer']*100:.2f}% CER={rec['cer']*100:.2f}% wall={rec['wall_sec']:.1f}s", flush=True) f.write(json.dumps(rec, ensure_ascii=False) + "\n") if __name__ == "__main__": main()