"""Offline fixed-audio STT benchmark runner. Run from backend: .\\.venv\\Scripts\\python.exe -m app.benchmarks.stt_offline_benchmark --vad none,silero """ from __future__ import annotations import argparse import hashlib import os from collections import defaultdict from difflib import SequenceMatcher import json import math import re import time import wave from array import array from pathlib import Path from typing import Any from app.services.transcript_diagnostics import repetition_diagnostics from app.services.pause_detection_service import PauseDetectionService from app.services.transcription_service import TranscriptionService from app.services.voice_activity_service import VoiceActivityService, cleanup_vad_result from app.services.voice_benchmark_logger import ( SCHEMA_VERSION, app_version, git_commit, LATEST_PAUSE_FAILURES_TOP20_PATH, LATEST_SUMMARY_JSON_PATH, LATEST_SUMMARY_PATH, LATEST_TERM_FAILURES_PATH, LATEST_TIMINGS_PATH, LATEST_TRANSCRIPT_FAILURES_TOP20_PATH, VOICE_RUNS_DIR, log_jsonl, new_request_id, now_iso, repo_root, write_json_atomic, write_jsonl_atomic, write_text_atomic, ) def main() -> int: args = _parse_args() manifest_path = _resolve_path(args.manifest) if args.manifest else None raw_dir = _resolve_path(args.raw_dir) vad_engines = [item.strip().lower() for item in args.vad.split(",") if item.strip()] pause_detectors = [item.strip().lower() for item in args.pause_detectors.split(",") if item.strip()] clip_index = _read_manifest_index(manifest_path) if manifest_path and manifest_path.exists() else {} clips = _discover_clips(raw_dir, clip_index) try: TranscriptionService.preflight_benchmark_model_cache() except Exception as exc: print(f"STT model preflight failed; benchmark aborted: {exc}") return 1 os.environ["STT_DISABLE_MODEL_RECOVERY"] = "1" print("Loading STT model...", flush=True) stt = TranscriptionService.get() _wait_for_stt(stt) if not stt.is_available: print("STT model failed to load; benchmark aborted.") return 1 print(f"STT model ready: {stt.model_config.model_dump()}", flush=True) run_config = _build_run_config( args=args, manifest_path=manifest_path, raw_dir=raw_dir, vad_engines=vad_engines, pause_detectors=pause_detectors, clip_count=len(clips), stt=stt, ) run_id = args.run_id or _build_run_id(run_config) run_dir = VOICE_RUNS_DIR / run_id run_dir.mkdir(parents=True, exist_ok=True) full_path = run_dir / "full.jsonl" full_path.write_text("", encoding="utf-8") total_jobs = len(clips) * (len(pause_detectors) + len(vad_engines)) completed_jobs = 0 print(f"Benchmark run_id: {run_id}", flush=True) print(f"Writing live rows to: {full_path}", flush=True) print(f"Clips: {len(clips)} | jobs: {total_jobs}", flush=True) vad = VoiceActivityService() pause_service = PauseDetectionService() run_rows: list[dict[str, Any]] = [] for clip_index, clip in enumerate(clips, start=1): clip_id = str(clip.get("clip_id") or "unknown") if pause_detectors: print(f"[{clip_index}/{len(clips)}] {clip_id} | pause detectors: {','.join(pause_detectors)}", flush=True) run_rows.extend(_run_pause_detector_clip(clip, pause_detectors, run_id, full_path, stt, pause_service)) completed_jobs += len(pause_detectors) print(f" completed {completed_jobs}/{total_jobs} jobs", flush=True) for engine in vad_engines: print(f"[{clip_index}/{len(clips)}] {clip_id} | vad={engine}", flush=True) row = _run_clip(clip, engine, run_id, full_path, stt, vad) run_rows.append(row) completed_jobs += 1 wer_value = row.get("wer") wer_text = f"{float(wer_value):.4f}" if wer_value is not None else "n/a" print(f" completed {completed_jobs}/{total_jobs} jobs | wer={wer_text}", flush=True) rows = _load_jsonl_rows(full_path) or run_rows print("Exporting benchmark summaries...", flush=True) _export_run_artifacts(run_dir, run_id, run_config, rows) _print_summary(rows, run_config, run_dir) return 0 def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run fixed WAV STT benchmarks.") parser.add_argument("--manifest", default=str(repo_root() / "benchmarks" / "stt_clips.json")) parser.add_argument("--raw-dir", default=str(repo_root() / "benchmarks" / "audio" / "raw")) parser.add_argument("--vad", default="", help="Optional comma-separated VAD trim engines: none,silero") parser.add_argument("--pause-detectors", default="rms_energy,silero", help="Comma-separated pause detectors: rms_energy,silero") parser.add_argument("--term-mapping", action="store_true", help="Record term mapping as enabled in the run config.") parser.add_argument("--run-id", default="") return parser.parse_args() def _resolve_path(path: str) -> Path: candidate = Path(path) if candidate.is_absolute(): return candidate cwd_candidate = Path.cwd() / candidate if cwd_candidate.exists(): return cwd_candidate return repo_root() / candidate def _read_manifest_index(path: Path) -> dict[str, dict[str, Any]]: clips = json.loads(path.read_text(encoding="utf-8")) index: dict[str, dict[str, Any]] = {} for clip in clips: clip_id = str(clip.get("clip_id") or "").strip() if clip_id: index[clip_id] = clip clip_path = str(clip.get("path") or "").strip() if clip_path: index[Path(clip_path).stem] = clip return index def _discover_clips(raw_dir: Path, clip_index: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: if clip_index: clips: list[dict[str, Any]] = [] for clip_id in sorted(clip_index): metadata = clip_index[clip_id] clip_path = _clip_path_from_manifest(raw_dir, clip_id, metadata) if clip_path is None: continue clips.append({ "clip_id": clip_id, "path": str(clip_path), "language": metadata.get("language") or _language_from_clip_id(clip_id), "recording_instruction": metadata.get("recording_instruction"), "expected_text": metadata.get("expected_text"), "expected_pause_text": metadata.get("expected_pause_text"), "category": metadata.get("clip_category") or metadata.get("category"), "notes": metadata.get("notes"), }) return clips if not raw_dir.exists(): return [] clips: list[dict[str, Any]] = [] for wav_path in sorted(raw_dir.rglob("*.wav")): clip_id = wav_path.stem clips.append({ "clip_id": clip_id, "path": str(wav_path), "language": _language_from_clip_id(clip_id), "recording_instruction": None, "expected_text": None, "expected_pause_text": None, "category": None, "notes": None, }) return clips def _clip_path_from_manifest(raw_dir: Path, clip_id: str, metadata: dict[str, Any]) -> Path | None: path_value = str(metadata.get("path") or "").strip() if path_value: candidate = _resolve_path(path_value) if candidate.exists(): return candidate fallback = raw_dir / f"{clip_id}.wav" if fallback.exists(): return fallback matches = list(raw_dir.rglob(f"{clip_id}.wav")) return matches[0] if matches else None def _language_from_clip_id(clip_id: str) -> str | None: if "_" not in clip_id: return None language = clip_id.split("_", 1)[0].strip() return language or None def _run_pause_detector_clip( clip: dict[str, Any], pause_detectors: list[str], run_id: str, output_path: Path, stt: TranscriptionService, pause_service: PauseDetectionService, ) -> list[dict[str, Any]]: request_id = new_request_id("stt_offline") clip_path = _resolve_path(str(clip.get("path", ""))) expected_text = str(clip.get("expected_text") or "") expected_pause_text = str(clip.get("expected_pause_text") or "") started = time.perf_counter() wav_meta: dict[str, Any] = {} original_wav_meta: dict[str, Any] = {} normalize_time_ms = 0.0 normalized_path: Path | None = None transcript = "" result = None error: str | None = None try: if not clip_path.exists(): raise FileNotFoundError(str(clip_path)) if not stt.is_available: raise RuntimeError("stt_unavailable") original_wav_meta = _wav_metadata(clip_path) normalized_path = _normalized_audio_path(clip) normalize_start = time.perf_counter() _normalize_wav_for_stt(clip_path, normalized_path) normalize_time_ms = round((time.perf_counter() - normalize_start) * 1000, 2) wav_meta = _wav_metadata(normalized_path) result = stt.transcribe(str(normalized_path)) transcript = result.transcript if result.success else "" error = result.error except Exception as exc: error = repr(exc) stt_total_ms = round((time.perf_counter() - started) * 1000, 2) model_config = ( result.model_config.model_dump() if result and result.model_config else stt.model_config.model_dump() ) success = bool(result.success) if result else False rows: list[dict[str, Any]] = [] for detector in pause_detectors: row_start = time.perf_counter() pause_result = None pause_text = transcript try: if normalized_path and normalized_path.exists(): pause_result = pause_service.detect(str(normalized_path), detector) pause_text = pause_service.build_pause_text( transcript, result.segments if result else None, pause_result.pauses, ).pause_text except Exception as exc: error = repr(exc) normalized_expected = normalize_text(expected_text) normalized_actual = normalize_text(transcript) normalized_expected_pause = normalize_pause_text(expected_pause_text) normalized_actual_pause = normalize_pause_text(pause_text) payload = { "schema_version": SCHEMA_VERSION, "event": "stt_transcription", "benchmark_type": "offline_model", "request_id": request_id, "run_id": run_id, "timestamp": now_iso(), "app_version": app_version(), "git_commit": git_commit(), **model_config, **wav_meta, **_normalization_log_fields(original_wav_meta, wav_meta, clip_path), "normalized_audio_path": str(normalized_path) if normalized_path else None, "clip_id": clip.get("clip_id"), "clip_category": clip.get("category"), "clip_language": clip.get("language"), "recording_instruction": clip.get("recording_instruction"), "expected_text": expected_text, "expected_pause_text": expected_pause_text or None, "actual_transcript": transcript, "transcript": transcript, "transcript_char_count": len(transcript), **repetition_diagnostics(transcript), "normalized_match": normalized_expected == normalized_actual if expected_text else None, "manual_score": None, "wer": word_error_rate(normalized_expected, normalized_actual) if expected_text else None, **_no_vad_log_fields(wav_meta), **_pause_log_fields(pause_result, pause_text), "pause_text_wer": word_error_rate(normalized_expected_pause, normalized_actual_pause) if expected_pause_text else None, "pause_text_normalized_match": normalized_expected_pause == normalized_actual_pause if expected_pause_text else None, "preprocess_time_ms": normalize_time_ms, "normalization_time_ms": normalize_time_ms, "metadata_time_ms": None, "stt_inference_time_ms": result.inference_time_ms if result else 0.0, "total_backend_time_ms": round(stt_total_ms + ((time.perf_counter() - row_start) * 1000), 2), "detected_language": result.detected_language if result else None, "detected_language_probability": result.detected_language_probability if result else None, "segments_count": result.segments_count if result else 0, "success": success, "error": None if success and not (pause_result and pause_result.error) else (pause_result.error if pause_result and pause_result.error else error), } log_jsonl(output_path, payload) rows.append(payload) return rows def _wait_for_stt(stt: TranscriptionService, timeout_s: float = 60.0) -> None: started = time.perf_counter() while not stt.is_ready and time.perf_counter() - started < timeout_s: time.sleep(0.1) def _run_clip( clip: dict[str, Any], vad_engine: str, run_id: str, output_path: Path, stt: TranscriptionService, vad: VoiceActivityService, ) -> dict[str, Any]: request_id = new_request_id("stt_offline") clip_path = _resolve_path(str(clip.get("path", ""))) expected_text = str(clip.get("expected_text") or "") expected_pause_text = str(clip.get("expected_pause_text") or "") started = time.perf_counter() vad_result = None wav_meta: dict[str, Any] = {} original_wav_meta: dict[str, Any] = {} normalize_time_ms = 0.0 normalized_path: Path | None = None transcript = "" result = None error: str | None = None try: if not clip_path.exists(): raise FileNotFoundError(str(clip_path)) if not stt.is_available: raise RuntimeError("stt_unavailable") original_wav_meta = _wav_metadata(clip_path) normalized_path = _normalized_audio_path(clip) normalize_start = time.perf_counter() _normalize_wav_for_stt(clip_path, normalized_path) normalize_time_ms = round((time.perf_counter() - normalize_start) * 1000, 2) wav_meta = _wav_metadata(normalized_path) processed_path = _processed_audio_path(clip, vad_engine) vad_result = vad.process(str(normalized_path), vad_engine, str(processed_path)) result = stt.transcribe(vad_result.audio_path) transcript = result.transcript if result.success else "" error = result.error except Exception as exc: error = repr(exc) finally: if vad_result: cleanup_vad_result(vad_result) total_ms = round((time.perf_counter() - started) * 1000, 2) normalized_expected = normalize_text(expected_text) normalized_actual = normalize_text(transcript) wer = word_error_rate(normalized_expected, normalized_actual) if expected_text else None model_config = ( result.model_config.model_dump() if result and result.model_config else stt.model_config.model_dump() ) success = bool(result.success) if result else False payload = { "schema_version": SCHEMA_VERSION, "event": "stt_transcription", "benchmark_type": "offline_model", "request_id": request_id, "run_id": run_id, "timestamp": now_iso(), "app_version": app_version(), "git_commit": git_commit(), **model_config, **wav_meta, **_normalization_log_fields(original_wav_meta, wav_meta, clip_path), "normalized_audio_path": str(normalized_path) if normalized_path else None, "clip_id": clip.get("clip_id"), "clip_category": clip.get("category"), "clip_language": clip.get("language"), "recording_instruction": clip.get("recording_instruction"), "expected_text": expected_text, "expected_pause_text": expected_pause_text or None, "actual_transcript": transcript, "transcript": transcript, "transcript_char_count": len(transcript), **repetition_diagnostics(transcript), "normalized_match": normalized_expected == normalized_actual if expected_text else None, "manual_score": None, "wer": wer, **_server_vad_log_fields(vad_result, wav_meta, vad_engine), "pause_detection_enabled": False, "pause_detector": None, "pause_detection_time_ms": 0.0, "pause_count": 0, "pauses": [], "pause_text": None, "pause_text_char_count": 0, "pause_text_wer": None, "pause_text_normalized_match": None, "pause_detector_error": "vad_trim_benchmark_row", "pause_detector_fallback_used": False, "preprocess_time_ms": normalize_time_ms, "normalization_time_ms": normalize_time_ms, "metadata_time_ms": None, "stt_inference_time_ms": result.inference_time_ms if result else 0.0, "total_backend_time_ms": total_ms, "detected_language": result.detected_language if result else None, "detected_language_probability": result.detected_language_probability if result else None, "segments_count": result.segments_count if result else 0, "success": success, "error": None if success else error, } log_jsonl(output_path, payload) return payload def _normalization_log_fields(raw_meta: dict[str, Any], normalized_meta: dict[str, Any], clip_path: Path) -> dict[str, Any]: if not raw_meta: return { "raw_audio_path": str(clip_path), "raw_audio_duration_ms": None, "raw_sample_rate": None, "raw_channels": None, "raw_sample_width_bytes": None, "raw_wav_size_bytes": None, "normalized_sample_rate": normalized_meta.get("sample_rate"), "normalized_channels": normalized_meta.get("channels"), "normalized_sample_width_bytes": normalized_meta.get("sample_width_bytes"), "normalization_applied": False, } normalization_applied = any([ raw_meta.get("sample_rate") != normalized_meta.get("sample_rate"), raw_meta.get("channels") != normalized_meta.get("channels"), raw_meta.get("sample_width_bytes") != normalized_meta.get("sample_width_bytes"), ]) return { "raw_audio_path": str(clip_path), "raw_audio_duration_ms": raw_meta.get("audio_duration_ms"), "raw_sample_rate": raw_meta.get("sample_rate"), "raw_channels": raw_meta.get("channels"), "raw_sample_width_bytes": raw_meta.get("sample_width_bytes"), "raw_wav_size_bytes": raw_meta.get("wav_size_bytes"), "normalized_sample_rate": normalized_meta.get("sample_rate"), "normalized_channels": normalized_meta.get("channels"), "normalized_sample_width_bytes": normalized_meta.get("sample_width_bytes"), "normalization_applied": normalization_applied, } def _server_vad_log_fields(vad_result: Any, wav_meta: dict[str, Any], requested_engine: str) -> dict[str, Any]: enabled = vad_result.vad_enabled if vad_result else requested_engine != "none" engine = vad_result.vad_engine if vad_result else (None if requested_engine == "none" else requested_engine) speech_duration_ms = vad_result.speech_duration_ms if vad_result else None segments_count = vad_result.vad_segments_count if vad_result else 0 speech_start_ms = vad_result.speech_start_ms if vad_result else None speech_end_ms = vad_result.speech_end_ms if vad_result else None silence_trimmed_ms = vad_result.silence_trimmed_ms if vad_result else 0.0 trim_ratio = vad_result.trim_ratio if vad_result else 0.0 fallback_used = vad_result.fallback_used if vad_result else False fallback_reason = vad_result.fallback_reason if vad_result else None processed_duration_ms = vad_result.processed_audio_duration_ms if vad_result else wav_meta.get("audio_duration_ms") processed_audio_path = vad_result.audio_path if vad_result else None vad_time_ms = vad_result.vad_time_ms if vad_result else 0.0 vad_error = vad_result.error if vad_result else None return { "server_vad_enabled": enabled, "server_vad_engine": engine, "server_speech_duration_ms": speech_duration_ms, "server_vad_segments_count": segments_count, "server_speech_start_ms": speech_start_ms, "server_speech_end_ms": speech_end_ms, "server_silence_trimmed_ms": silence_trimmed_ms, "server_trim_ratio": trim_ratio, "server_vad_fallback_used": fallback_used, "server_vad_fallback_reason": fallback_reason, "server_processed_audio_duration_ms": processed_duration_ms, "server_processed_audio_path": processed_audio_path, "server_vad_time_ms": vad_time_ms, "server_vad_error": vad_error, "vad_enabled": enabled, "vad_engine": engine, "speech_duration_ms": speech_duration_ms, "vad_segments_count": segments_count, "speech_start_ms": speech_start_ms, "speech_end_ms": speech_end_ms, "silence_trimmed_ms": silence_trimmed_ms, "trim_ratio": trim_ratio, "fallback_used": fallback_used, "vad_fallback_used": fallback_used, "vad_fallback_reason": fallback_reason, "processed_audio_duration_ms": processed_duration_ms, "processed_audio_path": processed_audio_path, "vad_time_ms": vad_time_ms, "vad_error": vad_error, } def _no_vad_log_fields(wav_meta: dict[str, Any]) -> dict[str, Any]: duration_ms = wav_meta.get("audio_duration_ms") return { "server_vad_enabled": False, "server_vad_engine": None, "server_speech_duration_ms": duration_ms, "server_vad_segments_count": 0, "server_speech_start_ms": None, "server_speech_end_ms": None, "server_silence_trimmed_ms": 0.0, "server_trim_ratio": 0.0, "server_vad_fallback_used": False, "server_vad_fallback_reason": None, "server_processed_audio_duration_ms": duration_ms, "server_processed_audio_path": None, "server_vad_time_ms": 0.0, "server_vad_error": None, "vad_enabled": False, "vad_engine": None, "speech_duration_ms": duration_ms, "vad_segments_count": 0, "speech_start_ms": None, "speech_end_ms": None, "silence_trimmed_ms": 0.0, "trim_ratio": 0.0, "fallback_used": False, "vad_fallback_used": False, "vad_fallback_reason": None, "processed_audio_duration_ms": duration_ms, "processed_audio_path": None, "vad_time_ms": 0.0, "vad_error": None, } def _pause_log_fields(pause_result: Any, pause_text: str) -> dict[str, Any]: if pause_result is None: return { "pause_detection_enabled": False, "pause_detector": None, "pause_detection_time_ms": 0.0, "pause_count": 0, "pauses": [], "pause_text": pause_text, "pause_text_char_count": len(pause_text or ""), "pause_detector_error": "pause_detection_not_run", "pause_detector_fallback_used": True, } return { "pause_detection_enabled": pause_result.enabled, "pause_detector": pause_result.detector, "pause_detection_time_ms": pause_result.detection_time_ms, "pause_count": pause_result.pause_count, "pauses": [pause.model_dump() for pause in pause_result.pauses], "pause_speech_regions_count": pause_result.speech_regions_count, "pause_speech_regions": [region.model_dump() for region in pause_result.speech_regions], "pause_rms_threshold": pause_result.threshold, "pause_frame_ms": pause_result.frame_ms, "pause_text": pause_text, "pause_text_char_count": len(pause_text or ""), "pause_detector_error": pause_result.error, "pause_detector_fallback_used": pause_result.fallback_used, } def _processed_audio_path(clip: dict[str, Any], vad_engine: str) -> Path: clip_id = str(clip.get("clip_id") or "clip") safe_clip_id = re.sub(r"[^a-zA-Z0-9_-]+", "_", clip_id).strip("_") or "clip" safe_engine = re.sub(r"[^a-zA-Z0-9_-]+", "_", vad_engine).strip("_") or "none" return repo_root() / "benchmarks" / "audio" / "processed" / f"{safe_clip_id}__vad_{safe_engine}.wav" def _normalized_audio_path(clip: dict[str, Any]) -> Path: clip_id = str(clip.get("clip_id") or "clip") safe_clip_id = re.sub(r"[^a-zA-Z0-9_-]+", "_", clip_id).strip("_") or "clip" return repo_root() / "benchmarks" / "audio" / "normalized" / f"{safe_clip_id}.wav" def _normalize_wav_for_stt(input_path: Path, output_path: Path, sample_rate: int = 16000) -> None: samples, source_rate = _read_wav_as_mono_float(input_path) normalized = _resample_linear(samples, source_rate, sample_rate) pcm = array("h") for sample in normalized: clipped = max(-1.0, min(1.0, sample)) pcm.append(int(clipped * 32767) if clipped >= 0 else int(clipped * 32768)) output_path.parent.mkdir(parents=True, exist_ok=True) with wave.open(str(output_path), "wb") as wav: wav.setnchannels(1) wav.setsampwidth(2) wav.setframerate(sample_rate) wav.writeframes(pcm.tobytes()) def _read_wav_as_mono_float(path: Path) -> tuple[list[float], int]: with wave.open(str(path), "rb") as wav: channels = wav.getnchannels() sample_width = wav.getsampwidth() sample_rate = wav.getframerate() frames = wav.readframes(wav.getnframes()) if sample_width == 1: values = [(byte - 128) / 128.0 for byte in frames] elif sample_width == 2: pcm = array("h") pcm.frombytes(frames) values = [sample / 32768.0 for sample in pcm] elif sample_width == 4: pcm = array("i") pcm.frombytes(frames) values = [sample / 2147483648.0 for sample in pcm] else: raise ValueError(f"unsupported_wav_sample_width:{sample_width}") if channels <= 1: return values, sample_rate mono: list[float] = [] for index in range(0, len(values), channels): frame = values[index:index + channels] if frame: mono.append(sum(frame) / len(frame)) return mono, sample_rate def _resample_linear(samples: list[float], source_rate: int, target_rate: int) -> list[float]: if source_rate == target_rate: return samples if not samples or source_rate <= 0 or target_rate <= 0: return [] ratio = source_rate / target_rate output_length = max(1, round(len(samples) / ratio)) output: list[float] = [] for index in range(output_length): source_index = index * ratio left_index = int(source_index) right_index = min(left_index + 1, len(samples) - 1) weight = source_index - left_index output.append(samples[left_index] * (1 - weight) + samples[right_index] * weight) return output def _wav_metadata(path: Path) -> dict[str, Any]: size_bytes = path.stat().st_size with wave.open(str(path), "rb") as wav: frames = wav.getnframes() sample_rate = wav.getframerate() channels = wav.getnchannels() sample_width = wav.getsampwidth() return { "audio_duration_ms": round((frames / sample_rate) * 1000, 2) if sample_rate else None, "sample_rate": sample_rate, "channels": channels, "sample_width_bytes": sample_width, "wav_size_bytes": size_bytes, } def normalize_text(text: str) -> str: text = text.lower().strip() text = re.sub(r"[^a-z0-9\s]", " ", text) return re.sub(r"\s+", " ", text).strip() def normalize_pause_text(text: str) -> str: text = text.lower().strip() text = re.sub(r"\.{9,}", " pause_long ", text) text = re.sub(r"\.{6,8}", " pause_medium ", text) text = re.sub(r"\.{3,5}", " pause_short ", text) text = re.sub(r"[^a-z0-9_\s]", " ", text) return re.sub(r"\s+", " ", text).strip() def word_error_rate(expected: str, actual: str) -> float | None: expected_words = expected.split() actual_words = actual.split() if not expected_words: return 0.0 if not actual_words else None distance = _levenshtein(expected_words, actual_words) return round(distance / len(expected_words), 4) def _levenshtein(left: list[str], right: list[str]) -> int: previous = list(range(len(right) + 1)) for i, left_word in enumerate(left, start=1): current = [i] for j, right_word in enumerate(right, start=1): insert = current[j - 1] + 1 delete = previous[j] + 1 replace = previous[j - 1] + (0 if left_word == right_word else 1) current.append(min(insert, delete, replace)) previous = current return previous[-1] def _build_run_config( args: argparse.Namespace, manifest_path: Path | None, raw_dir: Path, vad_engines: list[str], pause_detectors: list[str], clip_count: int, stt: TranscriptionService, ) -> dict[str, Any]: model_config = stt.model_config.model_dump() config: dict[str, Any] = { "backend": model_config.get("backend"), "wrapper": model_config.get("wrapper"), "model": model_config.get("model"), "model_file": model_config.get("model_file"), "model_path": model_config.get("model_path"), "language": model_config.get("language"), "task": model_config.get("task"), "translate": model_config.get("translate"), "device": model_config.get("device"), "quantization": model_config.get("quantization"), "compute_type": model_config.get("compute_type"), "vad_engines": vad_engines, "pause_detectors": pause_detectors, "term_mapping_enabled": bool(getattr(args, "term_mapping", False)), "dataset_manifest": _display_repo_path(manifest_path) if manifest_path else None, "dataset_manifest_hash": _sha256_file(manifest_path) if manifest_path and manifest_path.exists() else None, "raw_dir": _display_repo_path(raw_dir), "clip_count": clip_count, "created_at": now_iso(), "git_commit": git_commit(), } config["config_label"] = _build_config_label(config) config["config_hash"] = _sha256_text(_canonical_json(_config_hash_payload(config))) config["config_hash_short"] = config["config_hash"][:6] return config def _build_run_id(run_config: dict[str, Any]) -> str: timestamp = time.strftime("%Y%m%d_%H%M%S", time.localtime()) model = _slugify_token(str(run_config.get("model") or "model")) quantization = _slugify_token(str(run_config.get("quantization") or "unknown")) config_label = _slugify_token(str(run_config.get("config_label") or "cfg")) config_hash_short = str(run_config.get("config_hash_short") or "000000") return f"{timestamp}_{model}_{quantization}_{config_label}_cfg{config_hash_short}" def _export_run_artifacts(run_dir: Path, run_id: str, run_config: dict[str, Any], rows: list[dict[str, Any]]) -> None: config = dict(run_config) config["run_id"] = run_id config["run_dir"] = _display_repo_path(run_dir) summary = _build_summary_payload(run_id, config, rows) timings = _build_timings_payload(run_id, config, rows) transcript_failures, pause_failures, term_failures, failures_by_category = _build_failure_exports(rows, run_id) summary_text = _render_summary_text(summary, failures_by_category) write_json_atomic(run_dir / "config.json", config) write_json_atomic(run_dir / "summary.json", summary) write_text_atomic(run_dir / "summary.txt", summary_text) write_json_atomic(run_dir / "timings.json", timings) write_json_atomic(run_dir / "failures_by_category.json", failures_by_category) write_jsonl_atomic(run_dir / "transcript_failures_top20.jsonl", transcript_failures) write_jsonl_atomic(run_dir / "pause_failures_top20.jsonl", pause_failures) write_jsonl_atomic(run_dir / "term_failures.jsonl", term_failures) write_text_atomic(LATEST_SUMMARY_PATH, summary_text) write_json_atomic(LATEST_SUMMARY_JSON_PATH, summary) write_jsonl_atomic(LATEST_TRANSCRIPT_FAILURES_TOP20_PATH, transcript_failures) write_jsonl_atomic(LATEST_PAUSE_FAILURES_TOP20_PATH, pause_failures) write_jsonl_atomic(LATEST_TERM_FAILURES_PATH, term_failures) write_json_atomic(LATEST_TIMINGS_PATH, timings) def _build_summary_payload(run_id: str, run_config: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]: wer_values = [float(row["wer"]) for row in rows if row.get("wer") is not None] pause_wer_values = [float(row["pause_text_wer"]) for row in rows if row.get("pause_text_wer") is not None] inference_values = [float(row["stt_inference_time_ms"]) for row in rows if row.get("stt_inference_time_ms") is not None] summary = { "run_id": run_id, "rows": len(rows), "model": run_config.get("model"), "model_file": run_config.get("model_file"), "quantization": run_config.get("quantization"), "overall_wer": round(sum(wer_values) / len(wer_values), 4) if wer_values else None, "overall_pause_text_wer": round(sum(pause_wer_values) / len(pause_wer_values), 4) if pause_wer_values else None, "wer_by_category": _metric_average_by(rows, "clip_category", "wer"), "pause_text_wer_by_detector": _metric_average_by(rows, "pause_detector", "pause_text_wer", skip_none_key="none"), "median_inference_time_ms": round(_median(inference_values), 2) if inference_values else None, "p95_inference_time_ms": round(_percentile(inference_values, 95), 2) if inference_values else None, "config": dict(run_config), "generated_at": now_iso(), } return summary def _build_timings_payload(run_id: str, run_config: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]: payload: dict[str, Any] = { "run_id": run_id, "model": run_config.get("model"), "quantization": run_config.get("quantization"), "rows": len(rows), "stt_inference_time_ms": _metric_stats([float(row["stt_inference_time_ms"]) for row in rows if row.get("stt_inference_time_ms") is not None]), "total_backend_time_ms": _metric_stats([float(row["total_backend_time_ms"]) for row in rows if row.get("total_backend_time_ms") is not None]), "preprocess_time_ms": _metric_stats([float(row["preprocess_time_ms"]) for row in rows if row.get("preprocess_time_ms") is not None]), "normalization_time_ms": _metric_stats([float(row["normalization_time_ms"]) for row in rows if row.get("normalization_time_ms") is not None]), "pause_detection_time_ms_by_detector": {}, "stt_inference_time_ms_by_vad_engine": {}, } detector_groups: dict[str, list[float]] = defaultdict(list) vad_groups: dict[str, list[float]] = defaultdict(list) for row in rows: pause_detector = str(row.get("pause_detector") or "none") pause_time = row.get("pause_detection_time_ms") if pause_time is not None and pause_detector != "none": detector_groups[pause_detector].append(float(pause_time)) vad_engine = str(row.get("server_vad_engine") or "none") stt_time = row.get("stt_inference_time_ms") if stt_time is not None: vad_groups[vad_engine].append(float(stt_time)) payload["pause_detection_time_ms_by_detector"] = { detector: _metric_stats(values) for detector, values in sorted(detector_groups.items()) } payload["stt_inference_time_ms_by_vad_engine"] = { engine: _metric_stats(values) for engine, values in sorted(vad_groups.items()) } return payload def _build_failure_exports( rows: list[dict[str, Any]], run_id: str, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]: transcript_worst_rows = _best_rows_by_clip(rows, "wer") pause_worst_rows = _best_rows_by_clip(rows, "pause_text_wer") transcript_failures = [_transcript_failure_entry(row, run_id) for row in transcript_worst_rows] pause_failures = [_pause_failure_entry(row, run_id) for row in pause_worst_rows] term_candidates: list[dict[str, Any]] = [] for clip_rows in _rows_by_clip(rows).values(): best_term_entry = _best_term_failure_entry(clip_rows, run_id) if best_term_entry is not None: term_candidates.append(best_term_entry) term_failures = sorted( term_candidates, key=lambda row: ( -int(row.get("term_mismatch_count") or 0), -float(row.get("term_match_score") or 0.0), str(row.get("clip_id") or ""), ), ) failures_by_category = { "run_id": run_id, "categories": _category_failure_summary(transcript_worst_rows, pause_worst_rows, term_failures), } return transcript_failures[:20], pause_failures[:20], term_failures, failures_by_category def _rows_by_clip(rows: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]: grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in rows: clip_id = str(row.get("clip_id") or "").strip() if not clip_id: continue grouped[clip_id].append(row) return grouped def _load_jsonl_rows(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] rows: list[dict[str, Any]] = [] with path.open("r", encoding="utf-8") as handle: for raw_line in handle: line = raw_line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def _best_rows_by_clip(rows: list[dict[str, Any]], score_key: str) -> list[dict[str, Any]]: best_rows: list[dict[str, Any]] = [] for clip_rows in _rows_by_clip(rows).values(): eligible = [row for row in clip_rows if row.get(score_key) is not None] if not eligible: continue best_row = max(eligible, key=lambda row: ( float(row.get(score_key) or 0.0), float(row.get("stt_inference_time_ms") or 0.0), str(row.get("request_id") or ""), )) if float(best_row.get(score_key) or 0.0) > 0.0: best_rows.append(best_row) best_rows.sort(key=lambda row: ( -float(row.get(score_key) or 0.0), str(row.get("clip_id") or ""), str(row.get("request_id") or ""), )) return best_rows def _transcript_failure_entry(row: dict[str, Any], run_id: str) -> dict[str, Any]: return { "score_source": "transcript_wer", "run_id": run_id, "request_id": row.get("request_id"), "clip_id": row.get("clip_id"), "clip_category": row.get("clip_category") or "uncategorized", "wer": row.get("wer"), "expected_text": row.get("expected_text"), "transcript": row.get("transcript"), "likely_issue": _likely_issue(row), "vad_engine": row.get("server_vad_engine"), "pause_detector": row.get("pause_detector"), "segments_count": row.get("segments_count"), "error": row.get("error"), } def _pause_failure_entry(row: dict[str, Any], run_id: str) -> dict[str, Any]: return { "score_source": "pause_text_wer", "run_id": run_id, "request_id": row.get("request_id"), "clip_id": row.get("clip_id"), "clip_category": row.get("clip_category") or "uncategorized", "pause_detector": row.get("pause_detector"), "pause_text_wer": row.get("pause_text_wer"), "expected_pause_text": row.get("expected_pause_text"), "pause_text": row.get("pause_text"), "pauses": row.get("pauses") or [], "likely_issue": "pause_alignment", "vad_engine": row.get("server_vad_engine"), "pause_count": row.get("pause_count"), "pause_detector_error": row.get("pause_detector_error"), "pause_detector_fallback_used": row.get("pause_detector_fallback_used"), } def _best_term_failure_entry(rows: list[dict[str, Any]], run_id: str) -> dict[str, Any] | None: best_entry: dict[str, Any] | None = None for row in rows: entry = _term_failure_entry(row, run_id) if entry is None: continue if best_entry is None: best_entry = entry continue best_score = (int(best_entry.get("term_mismatch_count") or 0), float(best_entry.get("term_match_score") or 0.0)) candidate_score = (int(entry.get("term_mismatch_count") or 0), float(entry.get("term_match_score") or 0.0)) if candidate_score > best_score: best_entry = entry return best_entry def _term_failure_entry(row: dict[str, Any], run_id: str) -> dict[str, Any] | None: transcript = str(row.get("transcript") or "").strip() if not transcript: return None expected_terms = _extract_expected_terms(str(row.get("expected_text") or "")) if not expected_terms: return None transcript_candidates = _transcript_term_candidates(transcript) matches: list[dict[str, Any]] = [] for term in expected_terms: match = _best_term_match(term, transcript_candidates) if match is None: continue if match["normalized_expected"] == match["normalized_candidate"]: continue if float(match["similarity"]) < 0.75: continue matches.append(match) if not matches: return None observed_candidates: list[str] = [] for match in matches: candidate = str(match["observed_candidate"]) if candidate not in observed_candidates: observed_candidates.append(candidate) average_similarity = sum(float(match["similarity"]) for match in matches) / len(matches) confidence = "likely" if average_similarity >= 0.85 and len(matches) <= 3 else "uncertain" return { "score_source": "term_advisory", "run_id": run_id, "request_id": row.get("request_id"), "clip_id": row.get("clip_id"), "clip_category": row.get("clip_category") or "uncategorized", "expected_terms": [str(match["expected_term"]) for match in matches], "observed_candidates": observed_candidates, "term_matches": [ { "expected_term": match["expected_term"], "observed_candidate": match["observed_candidate"], "similarity": match["similarity"], } for match in matches ], "term_mismatch_count": len(matches), "term_match_score": round(average_similarity, 4), "confidence": confidence, "manual_review_needed": True, "transcript": row.get("transcript"), "expected_text": row.get("expected_text"), } def _extract_expected_terms(text: str) -> list[str]: tokens = list(re.finditer(r"[A-Za-z0-9][A-Za-z0-9\-]*", text or "")) terms: list[str] = [] for index, match in enumerate(tokens): token = match.group(0) if _looks_like_term(token, index): cleaned = token.strip(" ,.;:!?\"'()[]{}") if cleaned and cleaned not in terms: terms.append(cleaned) return terms def _looks_like_term(token: str, index: int) -> bool: if not token: return False if re.search(r"[0-9]", token): return True if "-" in token: return True if token.isupper() and len(token) >= 2: return True if any(char.isupper() for char in token[1:]): return True if index > 0 and token[0].isupper(): return True return False def _transcript_term_candidates(transcript: str) -> list[str]: words = [part for part in re.findall(r"[A-Za-z0-9][A-Za-z0-9\-]*", transcript or "") if part] candidates: list[str] = [] for size in (1, 2, 3): if len(words) < size: continue for start in range(0, len(words) - size + 1): phrase = " ".join(words[start:start + size]).strip() if phrase and phrase not in candidates: candidates.append(phrase) return candidates def _best_term_match(expected_term: str, candidates: list[str]) -> dict[str, Any] | None: normalized_expected = _normalize_term(expected_term) if not normalized_expected: return None best_candidate = "" best_similarity = 0.0 for candidate in candidates: normalized_candidate = _normalize_term(candidate) if not normalized_candidate: continue similarity = SequenceMatcher(None, normalized_expected, normalized_candidate).ratio() if similarity > best_similarity: best_similarity = similarity best_candidate = candidate if not best_candidate or best_similarity < 0.75: return None return { "expected_term": expected_term, "observed_candidate": best_candidate, "similarity": round(best_similarity, 4), "normalized_expected": normalized_expected, "normalized_candidate": _normalize_term(best_candidate), } def _normalize_term(term: str) -> str: return re.sub(r"[^a-z0-9]+", "", str(term or "").lower()) def _likely_issue(row: dict[str, Any]) -> str: category = str(row.get("clip_category") or "").strip().lower() if bool(row.get("possible_repetition_loop")): return "repetition_loop" if category == "model_names": return "model_name" if category in {"paper_results_metrics", "acronyms"}: return "metric_or_acronym" if category in {"dead_air_noise"}: return "silence_or_noise" if "pause" in category or category in {"abandoned_thought", "student_confusion", "long_realistic_dictation", "long_realistic"}: return "pause_alignment" if category == "emotion_sentiment": return "sentiment_label" if not str(row.get("transcript") or "").strip(): return "blank_transcript" return category or "general_transcript" def _category_failure_summary( transcript_failures: list[dict[str, Any]], pause_failures: list[dict[str, Any]], term_failures: list[dict[str, Any]], ) -> dict[str, Any]: categories: dict[str, dict[str, Any]] = {} for source, rows, score_key in ( ("transcript", transcript_failures, "wer"), ("pause", pause_failures, "pause_text_wer"), ("term", term_failures, "term_match_score"), ): for row in rows: category = str(row.get("clip_category") or "uncategorized") bucket = categories.setdefault(category, {}) source_bucket = bucket.setdefault(source, {"count": 0, "worst_row_ref": None}) source_bucket["count"] += 1 current_score = float(row.get(score_key) or 0.0) worst_ref = source_bucket["worst_row_ref"] if worst_ref is None or current_score > float(worst_ref.get("score") or 0.0): source_bucket["worst_row_ref"] = { "clip_id": row.get("clip_id"), "request_id": row.get("request_id"), "score": round(current_score, 4), "score_source": row.get("score_source"), "pause_detector": row.get("pause_detector"), "vad_engine": row.get("vad_engine"), } return categories def _metric_average_by( rows: list[dict[str, Any]], group_key: str, value_key: str, skip_none_key: str | None = None, ) -> dict[str, float]: grouped: dict[str, list[float]] = defaultdict(list) for row in rows: value = row.get(value_key) if value is None: continue group_value = str(row.get(group_key) or "uncategorized") if skip_none_key is not None and group_value == skip_none_key: continue grouped[group_value].append(float(value)) return { group: round(sum(values) / len(values), 4) for group, values in sorted(grouped.items()) if values } def _metric_stats(values: list[float]) -> dict[str, float | int | None]: if not values: return {"count": 0, "median": None, "p95": None, "min": None, "max": None} ordered = sorted(values) return { "count": len(values), "median": round(_median(ordered), 2), "p95": round(_percentile(ordered, 95), 2), "min": round(ordered[0], 2), "max": round(ordered[-1], 2), } def _render_summary_text(summary: dict[str, Any], failures_by_category: dict[str, Any]) -> str: lines = [ f"run_id: {summary.get('run_id')}", f"rows: {summary.get('rows')}", f"model: {summary.get('model')}", f"model_file: {summary.get('model_file')}", f"quantization: {summary.get('quantization')}", f"overall_wer: {summary.get('overall_wer')}", f"overall_pause_text_wer: {summary.get('overall_pause_text_wer')}", f"median_inference_time_ms: {summary.get('median_inference_time_ms')}", f"p95_inference_time_ms: {summary.get('p95_inference_time_ms')}", f"config_label: {summary.get('config', {}).get('config_label')}", f"config_hash: {summary.get('config', {}).get('config_hash_short')}", "", "wer_by_category:", ] for category, value in sorted(summary.get("wer_by_category", {}).items()): lines.append(f" {category}: {value}") lines.append("") lines.append("pause_text_wer_by_detector:") for detector, value in sorted(summary.get("pause_text_wer_by_detector", {}).items()): lines.append(f" {detector}: {value}") lines.append("") lines.append("failure_counts_by_category:") for category, payload in sorted(failures_by_category.get("categories", {}).items()): transcript_count = payload.get("transcript", {}).get("count", 0) pause_count = payload.get("pause", {}).get("count", 0) term_count = payload.get("term", {}).get("count", 0) lines.append(f" {category}: transcript={transcript_count} pause={pause_count} term={term_count}") return "\n".join(lines).rstrip() + "\n" def _display_repo_path(path: Path | None) -> str | None: if path is None: return None try: return str(path.resolve().relative_to(repo_root())) except Exception: return str(path.resolve()) def _sha256_file(path: Path) -> str | None: if path is None or not path.exists(): return None digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(65536), b""): digest.update(chunk) return digest.hexdigest() def _canonical_json(payload: dict[str, Any]) -> str: return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str) def _config_hash_payload(config: dict[str, Any]) -> dict[str, Any]: return { "backend": config.get("backend"), "wrapper": config.get("wrapper"), "model": config.get("model"), "model_file": config.get("model_file"), "model_path": config.get("model_path"), "language": config.get("language"), "task": config.get("task"), "translate": config.get("translate"), "quantization": config.get("quantization"), "compute_type": config.get("compute_type"), "vad_engines": config.get("vad_engines"), "pause_detectors": config.get("pause_detectors"), "term_mapping_enabled": config.get("term_mapping_enabled"), "dataset_manifest": config.get("dataset_manifest"), "dataset_manifest_hash": config.get("dataset_manifest_hash"), "raw_dir": config.get("raw_dir"), "clip_count": config.get("clip_count"), "git_commit": config.get("git_commit"), } def _build_config_label(config: dict[str, Any]) -> str: vad_engines = [str(item).strip().lower() for item in config.get("vad_engines") or [] if str(item).strip()] if not vad_engines or vad_engines == ["none"]: vad_label = "no_vad" else: vad_label = "vad_" + "_".join(_slugify_token(item) for item in vad_engines) pause_detectors = [str(item).strip().lower() for item in config.get("pause_detectors") or [] if str(item).strip()] if pause_detectors: pause_label = "pause_" + "_".join(_slugify_pause_detector(item) for item in pause_detectors) else: pause_label = "pause_none" term_label = "term_on" if config.get("term_mapping_enabled") else "term_off" return "_".join([vad_label, pause_label, term_label]) def _slugify_pause_detector(detector: str) -> str: if detector == "rms_energy": return "rms" return _slugify_token(detector) def _slugify_token(value: str) -> str: return re.sub(r"_+", "_", re.sub(r"[^a-zA-Z0-9]+", "_", value)).strip("_").lower() or "cfg" def _sha256_text(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest() def _print_summary(rows: list[dict[str, Any]], run_config: dict[str, Any], run_dir: Path) -> None: if not rows: print("No benchmark rows produced.") return print("\nBenchmark summary") print(f"Rows: {len(rows)}") print(f"Run dir: {run_dir}") print(f"Config: {run_config.get('config_label')} ({run_config.get('config_hash_short')})") _print_metric_block("Overall", rows) engines = sorted({str(row.get("server_vad_engine") or "none") for row in rows}) for engine in engines: engine_rows = [row for row in rows if str(row.get("server_vad_engine") or "none") == engine] _print_metric_block(f"VAD {engine}", engine_rows) detectors = sorted({str(row.get("pause_detector") or "none") for row in rows}) for detector in detectors: if detector == "none": continue detector_rows = [row for row in rows if str(row.get("pause_detector") or "none") == detector] _print_metric_block(f"Pause detector {detector}", detector_rows) def _print_metric_block(label: str, rows: list[dict[str, Any]]) -> None: if not rows: return wer_values = [float(row["wer"]) for row in rows if row.get("wer") is not None] inference_values = [float(row["stt_inference_time_ms"]) for row in rows if row.get("stt_inference_time_ms") is not None] blank_rows = [row for row in rows if not str(row.get("transcript") or "").strip()] fallback_rows = [row for row in rows if bool(row.get("server_vad_fallback_used"))] vad_error_rows = [row for row in rows if row.get("server_vad_error")] repetition_rows = [row for row in rows if bool(row.get("possible_repetition_loop"))] pause_wer_values = [float(row["pause_text_wer"]) for row in rows if row.get("pause_text_wer") is not None] pause_error_rows = [row for row in rows if row.get("pause_detector_error")] print(f"\n{label}") if wer_values: print(f" overall WER: {round(sum(wer_values) / len(wer_values), 4)}") else: print(" overall WER: n/a") if pause_wer_values: print(f" pause-text WER: {round(sum(pause_wer_values) / len(pause_wer_values), 4)}") else: print(" pause-text WER: n/a") category_map: dict[str, list[float]] = {} for row in rows: category = str(row.get("clip_category") or "uncategorized") wer_value = row.get("wer") if wer_value is None: continue category_map.setdefault(category, []).append(float(wer_value)) if category_map: print(" WER by category:") for category in sorted(category_map): values = category_map[category] print(f" {category}: {round(sum(values) / len(values), 4)}") else: print(" WER by category: n/a") if inference_values: print(f" median inference time: {round(_median(inference_values), 2)} ms") print(f" p95 inference time: {round(_percentile(inference_values, 95), 2)} ms") else: print(" median inference time: n/a") print(" p95 inference time: n/a") print(f" blank rate: {round(len(blank_rows) / len(rows) * 100, 2)}%") print(f" server_vad_fallback_rate: {round(len(fallback_rows) / len(rows) * 100, 2)}%") print(f" server_vad_error_rate: {round(len(vad_error_rows) / len(rows) * 100, 2)}%") print(f" pause_detector_error_rate: {round(len(pause_error_rows) / len(rows) * 100, 2)}%") print(f" hallucination/repetition rate: {round(len(repetition_rows) / len(rows) * 100, 2)}%") def _median(values: list[float]) -> float: ordered = sorted(values) midpoint = len(ordered) // 2 if len(ordered) % 2: return ordered[midpoint] return (ordered[midpoint - 1] + ordered[midpoint]) / 2 def _percentile(values: list[float], percentile: float) -> float: ordered = sorted(values) if not ordered: return math.nan if len(ordered) == 1: return ordered[0] rank = (percentile / 100) * (len(ordered) - 1) lower = math.floor(rank) upper = math.ceil(rank) if lower == upper: return ordered[int(rank)] weight = rank - lower return ordered[lower] * (1 - weight) + ordered[upper] * weight if __name__ == "__main__": raise SystemExit(main())