gogo-telefon/scripts/poc/stt_test.py

"""M0 PoC #1 — STT accuracy on local speech samples (§14 Phase 0).

Kill-the-risk test: can faster-whisper (language hint 'sr') reliably transcribe
service names, personal names and PHONE NUMBERS SPOKEN ALOUD over phone-quality
audio?

Prepare a directory of WAV/MP3/OGG samples recorded by the founder + a
manifest.csv with two columns:  filename,expected_text

Run:
    python scripts/poc/stt_test.py <samples_dir> [--model large-v3] [--device cuda]

Outputs per-file transcript vs expected, character error rate (CER), word error
rate (WER), digit accuracy (for phone numbers) and timing.
GATE (§14): if phone numbers/names are unusable → stop and rethink providers.
"""

from __future__ import annotations

import csv
import re
import sys
import time
from pathlib import Path


def levenshtein(a: list, b: list) -> int:
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        cur = [i]
        for j, cb in enumerate(b, 1):
            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
        prev = cur
    return prev[-1]


def normalize(s: str) -> str:
    return re.sub(r"[^\wšđčćž ]", "", s.lower()).strip()


def digits(s: str) -> str:
    words = {
        "nula": "0", "jedan": "1", "dva": "2", "tri": "3", "četiri": "4",
        "pet": "5", "šest": "6", "sedam": "7", "osam": "8", "devet": "9",
    }
    out = []
    for token in normalize(s).split():
        if token.isdigit():
            out.append(token)
        elif token in words:
            out.append(words[token])
    return "".join(out)


def main() -> None:
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(1)
    samples_dir = Path(sys.argv[1])
    model_name = sys.argv[sys.argv.index("--model") + 1] if "--model" in sys.argv else "large-v3"
    device = sys.argv[sys.argv.index("--device") + 1] if "--device" in sys.argv else "auto"

    manifest = samples_dir / "manifest.csv"
    if not manifest.exists():
        print(f"missing {manifest} (columns: filename,expected_text)")
        sys.exit(1)

    from faster_whisper import WhisperModel

    print(f"loading {model_name} on {device}…")
    t0 = time.monotonic()
    model = WhisperModel(model_name, device=device,
                         compute_type="float16" if device == "cuda" else "int8")
    print(f"model loaded in {time.monotonic() - t0:.1f}s\n")

    rows = list(csv.DictReader(manifest.open()))
    total_cer = total_wer = 0.0
    digit_ok = digit_total = 0
    for row in rows:
        path = samples_dir / row["filename"]
        expected = row["expected_text"]
        t0 = time.monotonic()
        segments, info = model.transcribe(str(path), language="sr", beam_size=5, vad_filter=True)
        got = " ".join(s.text.strip() for s in segments).strip()
        elapsed = time.monotonic() - t0

        exp_n, got_n = normalize(expected), normalize(got)
        cer = levenshtein(list(exp_n), list(got_n)) / max(1, len(exp_n))
        wer = levenshtein(exp_n.split(), got_n.split()) / max(1, len(exp_n.split()))
        total_cer += cer
        total_wer += wer

        exp_digits = digits(expected)
        if exp_digits:
            digit_total += 1
            digit_ok += exp_digits == digits(got)

        flag = "✓" if wer < 0.2 else ("~" if wer < 0.5 else "✗")
        print(f"{flag} {row['filename']}  ({elapsed:.1f}s, audio {info.duration:.1f}s)")
        print(f"   očekivano: {expected}")
        print(f"   dobijeno : {got}\n")

    n = max(1, len(rows))
    print("=" * 60)
    print(f"samples: {len(rows)}  avg CER: {total_cer / n:.1%}  avg WER: {total_wer / n:.1%}")
    if digit_total:
        print(f"phone-number samples exactly right: {digit_ok}/{digit_total}")
    print("GATE: WER > 50% on names or wrong digits on most numbers → rethink STT (§14)")


if __name__ == "__main__":
    main()