#!/usr/bin/env python3 """Run each speech-to-text provider over the dataset and score it. run.py DATA_DIR OUT_DIR PROVIDER [PROVIDER ...] PROVIDER is one of: transcribe, whisper, deepgram, assemblyai, speechmatics, gladia. Credentials come from the environment: transcribe AWS credentials (boto3 default chain) + STT_BENCH_BUCKET (S3 bucket for input) deepgram DEEPGRAM_API_KEY assemblyai ASSEMBLYAI_API_KEY (ASSEMBLYAI_MODELS defaults to universal-3-pro,universal-2) speechmatics SPEECHMATICS_API_KEY gladia GLADIA_API_KEY whisper none (faster-whisper, runs locally; WHISPER_MODEL defaults to large-v3) Each run writes OUT_DIR//.json with the raw hypothesis, wall-clock seconds from request to final transcript, and the version/model string reported or requested. `score.py` computes WER from those files so scoring is reproducible without re-running. pip install boto3 requests faster-whisper """ import json import os import sys import time import uuid import wave import requests def audio_seconds(path): with wave.open(path) as w: return w.getnframes() / w.getframerate() # ---------------------------------------------------------------- providers def run_transcribe(path): import boto3 bucket = os.environ["STT_BENCH_BUCKET"] region = os.environ.get("AWS_REGION", "us-east-1") s3 = boto3.client("s3", region_name=region) tr = boto3.client("transcribe", region_name=region) key = f"stt-bench/{uuid.uuid4()}/{os.path.basename(path)}" s3.upload_file(path, bucket, key) job = f"stt-bench-{uuid.uuid4()}" t0 = time.time() tr.start_transcription_job( TranscriptionJobName=job, LanguageCode="en-US", MediaFormat="wav", Media={"MediaFileUri": f"s3://{bucket}/{key}"}, ) while True: status = tr.get_transcription_job(TranscriptionJobName=job)["TranscriptionJob"] if status["TranscriptionJobStatus"] in ("COMPLETED", "FAILED"): break time.sleep(5) elapsed = time.time() - t0 if status["TranscriptionJobStatus"] == "FAILED": raise RuntimeError(status.get("FailureReason")) body = requests.get(status["Transcript"]["TranscriptFileUri"], timeout=60).json() text = body["results"]["transcripts"][0]["transcript"] for cleanup in ( lambda: s3.delete_object(Bucket=bucket, Key=key), lambda: tr.delete_transcription_job(TranscriptionJobName=job), ): try: cleanup() except Exception as exc: # scratch cleanup is best-effort print(f" cleanup skipped: {exc}", file=sys.stderr) return text, elapsed, "Amazon Transcribe batch, en-US, default model" def run_whisper(path): from faster_whisper import WhisperModel model_name = os.environ.get("WHISPER_MODEL", "large-v3") model = WhisperModel(model_name, device="cpu", compute_type="int8") t0 = time.time() segments, _ = model.transcribe(path, language="en", beam_size=5) text = " ".join(s.text.strip() for s in segments) return text, time.time() - t0, f"faster-whisper {model_name} int8 CPU ({os.cpu_count()} vCPU)" def run_deepgram(path): key = os.environ["DEEPGRAM_API_KEY"] model = os.environ.get("DEEPGRAM_MODEL", "nova-3") t0 = time.time() with open(path, "rb") as f: r = requests.post( "https://api.deepgram.com/v1/listen", params={"model": model, "language": "en", "smart_format": "true", "punctuate": "true"}, headers={"Authorization": f"Token {key}", "Content-Type": "audio/wav"}, data=f, timeout=600, ) elapsed = time.time() - t0 r.raise_for_status() body = r.json() text = body["results"]["channels"][0]["alternatives"][0]["transcript"] version = body.get("metadata", {}).get("model_info", {}) version = next(iter(version.values()), {}) if isinstance(version, dict) else {} return text, elapsed, f"Deepgram {model} ({version.get('name', '?')} {version.get('version', '')})".strip() def run_assemblyai(path): key = os.environ["ASSEMBLYAI_API_KEY"] base = "https://api.assemblyai.com/v2" h = {"authorization": key} t0 = time.time() with open(path, "rb") as f: up = requests.post(f"{base}/upload", headers=h, data=f, timeout=600) up.raise_for_status() models = os.environ.get("ASSEMBLYAI_MODELS", "universal-3-pro,universal-2").split(",") payload = {"audio_url": up.json()["upload_url"], "language_code": "en_us", "speech_models": models} r = requests.post(f"{base}/transcript", headers=h, json=payload, timeout=60) r.raise_for_status() tid = r.json()["id"] while True: s = requests.get(f"{base}/transcript/{tid}", headers=h, timeout=60).json() if s["status"] in ("completed", "error"): break time.sleep(3) elapsed = time.time() - t0 if s["status"] == "error": raise RuntimeError(s.get("error")) return s["text"], elapsed, f"AssemblyAI speech_models={','.join(s.get('speech_models') or models)}" def run_speechmatics(path): key = os.environ["SPEECHMATICS_API_KEY"] base = os.environ.get("SPEECHMATICS_URL", "https://asr.api.speechmatics.com/v2") h = {"Authorization": f"Bearer {key}"} op = os.environ.get("SPEECHMATICS_OPERATING_POINT", "enhanced") config = {"type": "transcription", "transcription_config": {"language": "en", "operating_point": op}} t0 = time.time() with open(path, "rb") as f: r = requests.post( f"{base}/jobs", headers=h, files={"data_file": f, "config": (None, json.dumps(config), "application/json")}, timeout=600, ) r.raise_for_status() jid = r.json()["id"] while True: s = requests.get(f"{base}/jobs/{jid}", headers=h, timeout=60).json()["job"] if s["status"] in ("done", "rejected", "deleted"): break time.sleep(3) elapsed = time.time() - t0 if s["status"] != "done": raise RuntimeError(s) t = requests.get(f"{base}/jobs/{jid}/transcript", headers=h, params={"format": "txt"}, timeout=60) t.raise_for_status() return t.text, elapsed, f"Speechmatics batch, operating_point={op}" def run_gladia(path): key = os.environ["GLADIA_API_KEY"] base = "https://api.gladia.io/v2" h = {"x-gladia-key": key} t0 = time.time() with open(path, "rb") as f: up = requests.post(f"{base}/upload", headers=h, files={"audio": (os.path.basename(path), f, "audio/wav")}, timeout=600) up.raise_for_status() r = requests.post(f"{base}/pre-recorded", headers=h, json={"audio_url": up.json()["audio_url"], "language": "en"}, timeout=60) r.raise_for_status() result_url = r.json()["result_url"] while True: s = requests.get(result_url, headers=h, timeout=60).json() if s["status"] in ("done", "error"): break time.sleep(3) elapsed = time.time() - t0 if s["status"] == "error": raise RuntimeError(s) return s["result"]["transcription"]["full_transcript"], elapsed, "Gladia pre-recorded v2" PROVIDERS = { "transcribe": run_transcribe, "whisper": run_whisper, "deepgram": run_deepgram, "assemblyai": run_assemblyai, "speechmatics": run_speechmatics, "gladia": run_gladia, } def main(): data_dir, out_dir, names = sys.argv[1], sys.argv[2], sys.argv[3:] wavs = sorted(p for p in os.listdir(data_dir) if p.endswith(".wav")) for name in names: fn = PROVIDERS[name] os.makedirs(os.path.join(out_dir, name), exist_ok=True) for wav in wavs: path = os.path.join(data_dir, wav) out = os.path.join(out_dir, name, wav.replace(".wav", ".json")) if os.path.exists(out): print(f"skip {name} {wav}") continue print(f"run {name} {wav}", flush=True) text, elapsed, version = fn(path) with open(out, "w") as f: json.dump( { "provider": name, "file": wav, "audio_seconds": audio_seconds(path), "wall_seconds": round(elapsed, 2), "version": version, "run_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "hypothesis": text, }, f, indent=1, ) print(f"done {name} {wav} {elapsed:.1f}s", flush=True) if __name__ == "__main__": main()