| """Long-audio check of the chunked CoreML encoder on real speech: 120 s segments of a Japanese livestream. |
| Reference = ElevenLabs word transcript of the same span. Compares CoreML chunked / fp32 ONNX chunked / |
| fp32 ONNX whole segment (no chunking). Usage: python long_eval.py <audio> <elevenlabs transcript.json> <start_s,...> [seg_s]""" |
| import json, os, subprocess, sys, time, unicodedata |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) |
| import numpy as np |
| from rapidfuzz.distance import Levenshtein |
| from haas_coreml import SR, Haas, chunks |
| from validate import onnx_encode, decode, norm, h |
|
|
| audio, tj, starts = sys.argv[1], sys.argv[2], [float(s) for s in sys.argv[3].split(",")] |
| seg = float(sys.argv[4]) if len(sys.argv) > 4 else 120.0 |
| words = json.load(open(tj))["words"] |
| tot = {k: [0, 0] for k in ["coreml_chunked", "onnx_chunked", "onnx_full"]} |
| agree = [0, 0] |
| for s in starts: |
| pcm = subprocess.run(["ffmpeg", "-v", "error", "-ss", str(s), "-t", str(seg), "-i", audio, "-ac", "1", "-ar", str(SR), |
| "-f", "f32le", "-"], capture_output=True, check=True).stdout |
| a = np.frombuffer(pcm, np.float32).copy() |
| ref = "".join(w["text"] for w in words if w["type"] == "word" and s <= w["start"] and w["end"] <= s + seg) |
| mel, L = h.front(a) |
| t0 = time.time(); c_ids, c_text = h.transcribe(a); dt = time.time() - t0 |
| o_ids = decode(onnx_encode(mel, L, True)) |
| texts = {"coreml_chunked": c_text, "onnx_chunked": h.tok.decode(o_ids), "onnx_full": h.tok.decode(decode(onnx_encode(mel, L, False)))} |
| row = {"start_s": s, "n_chunks": len(chunks(L)), "rtf": round(dt / seg, 4), |
| "coreml_vs_onnx_chunked_token_err": Levenshtein.distance(c_ids, o_ids), "n_tokens": len(o_ids)} |
| agree[0] += row["coreml_vs_onnx_chunked_token_err"]; agree[1] += len(o_ids) |
| for k, t in texts.items(): |
| e = Levenshtein.distance(norm(t), norm(ref)); row[f"{k}_cer"] = round(100 * e / len(norm(ref)), 2) |
| tot[k][0] += e; tot[k][1] += len(norm(ref)) |
| print(json.dumps(row), flush=True) |
| print(json.dumps({"cer_vs_elevenlabs": {k: f"{100 * v[0] / v[1]:.2f}%" for k, v in tot.items()}, |
| "coreml_vs_onnx_chunked_token_diff": f"{100 * agree[0] / agree[1]:.2f}%"})) |
|
|
