diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 65c4926c..a7a632a1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -225,3 +225,15 @@ jobs: # pinned, so `no-deps` audits exactly those resolved versions. inputs: audit-requirements.txt no-deps: true + # MAL-2026-4750 is a *malicious-package* advisory in the OSV + # malicious-packages dataset that matches the name `fastapi`. + # It does NOT describe a vulnerability in the legitimate PyPI + # fastapi we depend on (the dataset tracks typosquats / backdoored + # impostors — e.g. the `fastapi-toolkit` backdoor and fake + # `fastapi` copies); the advisory records no fixed version, and + # 0.136.3 is the latest release, so there is nothing to bump to. + # Auditing it produces a permanent false positive on the real + # dependency. Drop this ignore if/when OSV scopes the advisory to + # the actual malicious artifact instead of the name `fastapi`. + ignore-vulns: | + MAL-2026-4750 diff --git a/.gitignore b/.gitignore index bc3e1a6f..3b5dac98 100644 --- a/.gitignore +++ b/.gitignore @@ -23,6 +23,9 @@ recordings/ # next run so re-launching is one Enter keystroke. .tapscribe-install.json +# Live-benchmark results (tools/bench_live.py) — generated per run. +bench-results/ + # Frontend typecheck tooling (devDep only, never shipped) frontend/node_modules/ frontend/package-lock.json diff --git a/docs/live-tuning-research.md b/docs/live-tuning-research.md new file mode 100644 index 00000000..6d07f802 --- /dev/null +++ b/docs/live-tuning-research.md @@ -0,0 +1,175 @@ +# Live-transcription tuning: research notes + +Background for the live-quality investigation. The **live** path is not +the batch path — it runs WhisperLiveKit (WlK) as a subprocess, fed by +TapScribe's own Silero `SpeechGate`, with captions surfaced through +`WlKRelay`. This note records what the two upstreams recommend, how our +current defaults compare, and the concrete hypotheses the benchmark +(`tools/bench_live.py`) should confirm or refute. + +It is research, not conclusions: the actual numbers come from running the +sweep on a box that can load the models (see "Running the sweep"). + +## The knobs and where they live + +| Layer | Set by | Flag / field | +|---|---|---| +| Model size + backend | WlK CLI | `--model`, `--backend` | +| Streaming policy | WlK CLI | `--backend-policy` (simulstreaming / localagreement) | +| Decode cadence | WlK CLI | `--min-chunk-size` | +| Rolling-buffer reset | WlK CLI | `--buffer_trimming` (+ `_sec`) | +| Token confirmation | WlK CLI | `--confidence-validation` | +| Native VAD | WlK CLI | `--vac` / `--no-vac` | +| TapScribe speech gate | `LiveConfig` → `SpeechGate` | `gate_speech_threshold`, `gate_hangover_ms`, `gate_pre_roll_ms`, `gate_min_speech_ms` | + +`tapscribe.live.build_live_cmd` builds the WlK argv; `tapscribe.speech_gate` +runs the gate. With the default `gate_kind="tapscribe"`, build_live_cmd +passes `--no-vac` (our gate does the gating) and `--confidence-validation`. + +## Our current live defaults vs upstream defaults + +From `tapscribe/__main__.py` and `tapscribe/live.py:LiveConfig`: + +| Setting | TapScribe default | Upstream default | Note | +|---|---|---|---| +| Whisper model | **`tiny.en`** | `base`/`medium` (WlK), version-dependent | **Smallest model — prime accuracy suspect.** | +| `gate_kind` | `tapscribe` (→ `--no-vac`) | WlK VAC on | We replace WlK's VAC with our gate. | +| `confidence_validation` | **`True`** (`--confidence-validation`) | off | WlK warns this is *faster but punctuation less accurate*. We opt into the speed/accuracy trade by default. | +| `min_chunk_size` | unset → WlK default | ~1.0 s | More future context per decode = better accuracy + more lag. | +| `buffer_trimming` | unset → WlK default | `segment` | WlK docs: `segment` performs better in their tests; `sentence` needs a segmenter. | +| `backend-policy` (non-nb) | unset → WlK default `simulstreaming` (AlignAtt) | `simulstreaming` | nb-whisper path forces `localagreement`. Worth A/B-ing for the Whisper path. | +| gate speech threshold | `0.5` | Silero `0.5` | Matches Silero's recommended default. | +| gate hangover (→ `min_silence_duration_ms`) | `400 ms` | Silero `100 ms` | More conservative — avoids chopping mid-sentence pauses. | +| gate pre-roll | `300 ms` ring flushed on open | Silero `speech_pad_ms` `30 ms` | Much larger lead-in recovery than Silero's pad — recovers leading consonants. | +| gate `min_speech_ms` | `0` (open on first VAD "start") | — (no Silero equivalent) | `0` lets one-frame blips open the gate. | + +### Smells worth chasing + +1. **`tiny.en` is the live default.** The weakest Whisper model. The + single most likely cause of poor live quality. +2. **`confidence_validation=True` by default.** Upstream explicitly says + it trades punctuation accuracy for speed. We enable it unconditionally. +3. **The gate over-trims, badly, on noisy/continuous audio — confirmed + with real data** (see the gate-only benchmark below). On + `armstrong-en.wav` the default gate forwards only **~21 %** of the + audio and opens **once** (3.8–6.0 s), dropping the entire first phrase + *"That's one small step for man"* even though the clip is continuous + speech end-to-end. This is the strongest concrete lead for poor live + quality. + +## Gate-only benchmark (real data, model-free) + +The `SpeechGate` is the front half of the live path and runs entirely +locally (Silero VAD loads with no network), so we can measure exactly how +much audio each gate config forwards **without an ASR model**: + +```bash +python tools/bench_live.py --gate-only +``` + +Results on the two fixtures (`fwd%` = frames forwarded to the backend; +`segs` = silence→speech openings): + +| fixture | gate | thr | hang | fwd% | segs | kept_s / audio_s | +|---|---|---|---|---|---|---| +| armstrong-en | tapscribe | 0.50 | 400 | **20.8** | 1 | 2.5 / 12.0 | +| armstrong-en | tapscribe | 0.30 | 400 | 22.5 | 1 | 2.7 / 12.0 | +| armstrong-en | tapscribe | 0.70 | 400 | 20.8 | 1 | 2.5 / 12.0 | +| armstrong-en | tapscribe | 0.50 | 800 | 24.0 | 1 | 2.9 / 12.0 | +| armstrong-en | **backend** | — | — | **100.0** | 1 | 12.0 / 12.0 | +| marlene-nb | tapscribe | 0.50 | 400 | **91.6** | 4 | 13.7 / 15.0 | +| marlene-nb | backend | — | — | 100.0 | 1 | 15.0 / 15.0 | + +**Reading it:** + +- The Armstrong clip is continuous speech at −12 to −21 dBFS (no real + silence — verified by an RMS energy profile), yet the gate opens for a + single 2.2 s window at the **loudest** passage and discards the rest. + Lowering `gate_speech_threshold` to 0.3 barely changes it (22.5 %), so + this is **not** a simple threshold tweak — Silero classifies the quieter + (but still clearly-spoken) passages as non-speech on this noisy old + recording. In production a model gets only that 2.2 s, so most words are + never transcribed. +- The Marlene clip (a clean studio reading) forwards ~92 % across 4 + segments — the gate behaves well on clean speech. **So the failure is + audio-dependent: the gate collapses on noisy / low-level / continuous + speech, which is exactly what real meeting audio looks like.** +- `gate_kind=backend` (no TapScribe gate; WlK's own VAC) forwards 100 %. + A strong A/B candidate, and the quickest mitigation to validate. + +This points the live-quality investigation squarely at the gate. Open +questions for the code review / full sweep: is the single-open behaviour a +Silero limitation on noisy audio, or a bug in `SpeechGate.feed()` / +`make_silero_vad` (e.g. missing `speech_pad`, the 512-vs-320 sample +buffering, or the gate never re-opening)? The `gate_kind` A/B in the full +sweep quantifies the transcription cost directly (deletions). + +## WhisperLiveKit recommendations (upstream) + +- **`--buffer_trimming segment`** is the better-tested default and needs + no sentence segmenter; `sentence` trims on confirmed punctuation but + needs the segmenter installed. +- **`--min-chunk-size`** should align with the frontend's chunk cadence; + larger = more context per decode = better accuracy but more lag. Tune + against the per-tap lag reading. +- **`--backend auto`** picks MLX on macOS, then Faster-Whisper, then + Whisper. On Apple Silicon, MLX is the fast path (the harness sets + `--backend mlx-whisper` when `use_mlx`). +- **`--backend-policy`**: `simulstreaming` (AlignAtt, the default) vs + `localagreement`. AlignAtt has `--frame-threshold` (default 25) for the + speed/accuracy trade. +- General advice: **benchmark latency + accuracy for your model and VAD + settings on your actual device**, using `--warmup-file` and watching + CPU/latency. That is exactly what `tools/bench_live.py` automates. + +## Silero VAD recommendations (upstream) + +- **`threshold` 0.5** is a good lazy default; tune per dataset. (We use + 0.5.) +- **`min_silence_duration_ms` 100 ms** default — silence to wait before + closing a chunk. (We map `gate_hangover_ms=400 ms` here — more + conservative, fewer mid-sentence cuts.) +- **`speech_pad_ms` 30 ms** default — pads each side of a chunk. (We don't + use Silero's pad; our `gate_pre_roll_ms=300 ms` ring buffer does the + lead-in recovery, much larger.) +- Sampling rate must be 16 kHz for `VADIterator` (our wire format). + +## Hypotheses for the sweep to test + +1. Moving `tiny.en` → `base.en` → `small.en` drops WER substantially + (substitutions fall). **Most likely the main fix.** +2. `gate_kind=backend` vs `tapscribe`: if the tapscribe gate is clipping + speech, the `tapscribe` rows show more **deletions** than `backend`. +3. `confidence_validation` off improves accuracy (at some latency cost) — + add a sweep row toggling it once 1–2 are understood. +4. `min_chunk_size=1.0` and/or `buffer_trimming=segment` change the + latency (`lag_*`, `final_delay_s`) / accuracy balance. + +## Running the sweep + +```bash +pip install -e ".[whisper,bench]" # whisperlivekit + faster-whisper + jiwer +python tools/bench_live.py --sweep # matrix over every fixture → bench-results/*.json +# or a single config, detailed: +python tools/bench_live.py tests/fixtures/audio/armstrong-en.wav --model base.en +``` + +On Apple Silicon the harness auto-selects the MLX backend; elsewhere it +uses faster-whisper (CPU/CUDA). The metrics to read first: **WER** plus +the **sub / del / ins** split (substitutions ⇒ weak model, deletions ⇒ +dropped/clipped speech, insertions ⇒ hallucination), then **lag** and +**gate forward %**. + +> **Note on the managed dev box:** model weights download from +> HuggingFace, which the Claude-Code-on-the-web network policy blocks +> ("Host not in allowlist"). The harness reports this as a clean per-run +> error rather than hanging, but the sweep itself must run on a machine +> with model access (e.g. your Mac). The scoring, gate, and framing paths +> were validated here without a model. + +## Sources + +- [WhisperLiveKit (GitHub)](https://github.com/QuentinFuxa/WhisperLiveKit) +- [WhisperLiveKit default & custom models](https://github.com/quentinfuxa/whisperlivekit/blob/main/docs/default_and_custom_models.md) +- [Silero VAD (GitHub)](https://github.com/snakers4/silero-vad) +- [ufal/whisper_streaming (the streaming algorithm WlK builds on)](https://github.com/ufal/whisper_streaming) diff --git a/pyproject.toml b/pyproject.toml index 04ad5290..5341389c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -142,6 +142,11 @@ parakeet = [ "tapscribe[parakeet-cpu]; sys_platform != 'darwin' or platform_machine != 'arm64'", ] canary = ["tapscribe[canary-cpu]"] +# Scoring deps for the live-path benchmark harness (tools/bench_live.py): +# jiwer computes WER/CER + the hit/sub/del/ins breakdown the sweep +# reports. Kept out of `dev` so the unit-test CI install stays lean — +# only operators running the benchmark pull it in. +bench = ["jiwer>=3.0"] dev = [ "pytest>=8", "pytest-asyncio>=0.23", @@ -199,6 +204,7 @@ markers = [ "chaos: race-condition / error-recovery test exercising background tasks, abnormal close paths, or repeated rapid mutations. Opt out with -m 'not chaos'.", "browser_e2e: requires Playwright + Chromium (`pip install -e .[dev] && python -m playwright install chromium`). Opt out with -m 'not browser_e2e'.", "real_silero: opt out of the autouse silero stub in tests/conftest.py — exercises the real `detect_speech_silero` import path (only set on tests that need it).", + "real_live: drives the full live path (whisperlivekit-server child + SpeechGate + WlKRelay) on a real audio fixture and asserts a WER threshold. Skipped unless whisperlivekit + faster-whisper + jiwer are importable and the model can load. Heavy + downloads weights — not in the default CI matrix.", ] [tool.ruff] diff --git a/tapscribe/live.py b/tapscribe/live.py index faa241b0..dd0b6789 100644 --- a/tapscribe/live.py +++ b/tapscribe/live.py @@ -18,6 +18,7 @@ from __future__ import annotations +import contextlib import errno import os import shutil @@ -147,6 +148,13 @@ class LiveConfig: # equivalent knob — this filter lives entirely in SpeechGate. gate_min_speech_ms: int = 0 confidence_validation: bool = True + # WlK transcription policy. None = use WlK's default ("simulstreaming", + # AlignAtt — commits tokens as it decodes, keeps no unvalidated buffer, + # so `buffer_transcription` stays empty and the dashboard's in-flight + # preview never shows). "localagreement" holds tokens until they agree + # across chunks, which populates `buffer_transcription`. Only emitted to + # argv for non-nb models; nb-whisper always forces localagreement. + backend_policy: str | None = None # Forwarded to whisperlivekit-server when set — see build_live_cmd. min_chunk_size: float | None = None buffer_trimming: str | None = None # "sentence" | "segment" @@ -219,6 +227,8 @@ def build_live_cmd( cmd.extend(["--model", config.model]) if use_mlx: cmd.extend(["--backend", "mlx-whisper"]) + if config.backend_policy is not None: + cmd.extend(["--backend-policy", config.backend_policy]) if init_prompt: cmd.extend(["--init-prompt", init_prompt]) @@ -677,6 +687,14 @@ def _pump_logs(self, proc: subprocess.Popen) -> None: promoted = True finally: rc = proc.wait() + # Close the stdout pipe we've drained to EOF. Popen would + # eventually close it on GC, but that leaks an fd per child + # across the dashboard's stop→start "Apply model" restarts + # (and emits a ResourceWarning under -W error). The pump owns + # this end of the pipe, so closing it here is the clean spot. + if proc.stdout is not None: + with contextlib.suppress(Exception): + proc.stdout.close() # Only update INFO if this proc is still the active one; a fresh # start() may already have replaced it. if self._proc is proc: diff --git a/tapscribe/live_relay.py b/tapscribe/live_relay.py index 98cf6516..d27cdcd2 100644 --- a/tapscribe/live_relay.py +++ b/tapscribe/live_relay.py @@ -104,13 +104,6 @@ def __init__( # consecutive same-speaker segments — see # `tokens_alignment.py`) only emits the new suffix. self._emitted_by_key: dict[tuple, str] = {} - # Non-tail entries are immutable in WlK's wire format — the - # merger case only ever modifies the CURRENT tail in place; - # once a newer entry appears after a position, that position - # is frozen. Cache the lower bound of the scan so we don't - # re-walk hundreds of stable lines on every snapshot for long - # sessions. - self._last_emit_scan_upto: int = 0 # Tail-stability bookkeeping: counts how many consecutive # snapshots the tail's `(key, text)` has matched while the # buffer was empty. Reset whenever the tail changes or buffer @@ -245,17 +238,19 @@ async def _consume(self) -> None: if not isinstance(snapshot, list): continue self._last_snapshot = snapshot - # Emit every newly-non-tail entry. Non-tail positions - # are immutable in WlK's wire format, so we only need - # to scan from the last upto we processed. The dedup - # in `_consider_emit_line` covers a tail-becoming- - # non-tail whose text already settled (no re-emit) and - # the rare same-key text-change case (emits the diff). + # Emit every non-tail entry (the tail is held back for the + # stability flush below). We re-scan ALL non-tail positions + # each snapshot rather than caching a lower bound: although + # WlK normally only mutates the current tail, a late + # LocalAgreement correction can grow an already-committed + # line, and freezing non-tail positions would silently drop + # that suffix until close (then deliver it out of order). + # `_consider_emit_line` is keyed + idempotent, so re-walking + # settled lines is a no-op beyond a dict lookup per line — + # negligible at WlK's few-Hz snapshot rate. upto = max(0, len(snapshot) - 1) - for i in range(self._last_emit_scan_upto, upto): + for i in range(upto): self._consider_emit_line(snapshot[i]) - if upto > self._last_emit_scan_upto: - self._last_emit_scan_upto = upto # Tail-stability flush: once the tail text has held # steady (with the in-flight buffer empty) for enough # consecutive snapshots, WlK has nothing more to diff --git a/tapscribe/tap_fan_out.py b/tapscribe/tap_fan_out.py index c3760e2a..2b560524 100644 --- a/tapscribe/tap_fan_out.py +++ b/tapscribe/tap_fan_out.py @@ -149,6 +149,12 @@ async def write_frame(self, buf: bytes) -> None: if current_open != self._gate_open_last: self._gate_open_last = current_open await self._recorder.streams.update_gate_open(self._conn_id, current_open) + if not current_open: + # Tap just went idle — drop any stale lag immediately so + # the dashboard stops showing a backlog for a speaker who + # stopped talking, rather than waiting for _on_metrics to + # next fire (and it now suppresses while closed anyway). + await self._recorder.streams.update_lag(self._conn_id, None) else: frames_to_send = (buf,) @@ -277,7 +283,19 @@ async def _open(self) -> None: async def _on_metrics(self, lag_s: float) -> None: """Push the relay's latest reported lag to this tap's row so the - dashboard can render a per-tap backlog indicator.""" + dashboard can render a per-tap backlog indicator. + + Suppressed while the TapScribe gate is closed: WlK keeps emitting + `remaining_time_transcription` even after we stop feeding it, but + that value is `wall_clock - last_processed_audio` — it climbs purely + because time passes during the silence we're gating out, not because + there's a real decode backlog. Reporting it would show a phantom, + ever-growing lag for a tap whose speaker has gone quiet. When the + gate is open (actively forwarding, hangover included) the number is + genuine. Backend-gate mode (`self._gate is None`) feeds WlK + continuously, so its lag stays meaningful and is always forwarded.""" + if self._gate is not None and not self._gate.is_open: + return await self._recorder.streams.update_lag(self._conn_id, lag_s) def _on_buffer(self, text: str) -> None: diff --git a/tests/e2e/test_live_quality.py b/tests/e2e/test_live_quality.py new file mode 100644 index 00000000..946d986f --- /dev/null +++ b/tests/e2e/test_live_quality.py @@ -0,0 +1,76 @@ +"""Real live-path quality gate (the red→green target for live tuning). + +Unlike `test_pipeline_with_real_whisper` (which exercises the BATCH +transcribe-session route), this drives the *live* pipeline end to end — +the supervised `whisperlivekit-server` child, the TapScribe `SpeechGate`, +and the `WlKRelay` — on a known fixture and asserts a WER threshold on +the captions it captures. It reuses `tools/bench_live.run_one` so the +test and the operator's sweep tool share one code path. + +Heavy and slow (spawns a real ASR subprocess, downloads weights on first +run, paces audio in real time), so it's gated behind the `real_live` +marker and skips cleanly whenever the moving parts aren't present: +whisperlivekit + faster-whisper + jiwer importable, and +`whisperlivekit-server` resolvable. A model-download / startup failure +is treated as missing infrastructure (skip), not a quality regression +(fail) — only a *successful* run that misses the WER bar fails. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +from tapscribe.live import LiveConfig, WhisperLiveKitChannel + +FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures" / "audio" + +# base.en on the corrected ~9 s Armstrong clip (the iconic line) should +# land well under this — a clean batch decode is ~0 WER, and the live +# path's real-time streaming + gate add only modest error. The bar is +# deliberately loose so CPU faster-whisper doesn't flake; tighten it once +# a clean-fixture sweep confirms the real base.en number. +# +# History: this fixture previously held the "step off the LM now" lead-in +# rather than the reference line, so every model scored ~0.92 here and the +# bar read like a live-quality bug to chase. It was a mislabeled fixture, +# not a model deficiency — see tests/fixtures/audio/README.md. +_MAX_WER = 0.5 + + +@pytest.mark.real_live +async def test_live_path_meets_wer_threshold(): + for mod in ("whisperlivekit", "faster_whisper", "jiwer"): + if importlib.util.find_spec(mod) is None: + pytest.skip(f"{mod} not installed — install with `pip install -e .[whisper,bench]`") + if WhisperLiveKitChannel._find_exe() is None: + pytest.skip("whisperlivekit-server not on PATH — install with `pip install -e .[whisper]`") + + wav = FIXTURES_DIR / "armstrong-en.wav" + if not wav.is_file() or not (FIXTURES_DIR / "armstrong-en.reference.txt").is_file(): + pytest.skip("armstrong-en fixture missing — see tests/fixtures/audio/README.md") + + from tools.bench_live import run_one + + cfg = LiveConfig(model="base.en", language="en", host="127.0.0.1", port=0) + result = await run_one( + wav, + cfg, + use_mlx=False, + speed=1.0, + ready_timeout=300.0, + verbose=False, + ) + + if result.error: + pytest.skip(f"live channel could not run (infra, not quality): {result.error}") + + wer = result.metrics.get("wer") + assert result.hypothesis.strip(), f"live path produced no captions; lines={result.settled_lines}" + assert wer is not None, f"scoring failed: {result.metrics.get('score_error')}" + assert wer <= _MAX_WER, ( + f"live WER {wer:.2f} exceeds {_MAX_WER} for base.en on armstrong-en. " + f"hypothesis={result.hypothesis!r} metrics={result.metrics}" + ) diff --git a/tests/fixtures/audio/README.md b/tests/fixtures/audio/README.md index ed11e776..670ed66b 100644 --- a/tests/fixtures/audio/README.md +++ b/tests/fixtures/audio/README.md @@ -16,7 +16,7 @@ Skipped automatically when `faster-whisper` isn't installed ``` tests/fixtures/audio/ -├── armstrong-en.wav # 12 s, 16 kHz mono int16, NASA PD +├── armstrong-en.wav # ~9 s, 16 kHz mono int16, NASA PD ├── armstrong-en.reference.txt ├── marlene-nb.wav # 15 s, 16 kHz mono int16, CC-BY-SA 4.0 ├── marlene-nb.reference.txt @@ -33,9 +33,17 @@ rather than producing garbled transcripts later. ### `armstrong-en.wav` -First ~12 seconds of [`Armstrong_Small_Step.ogg`](https://commons.wikimedia.org/wiki/File:Armstrong_Small_Step.ogg) -from Wikimedia Commons, downsampled from 11 025 Hz mono OGG/Vorbis to -16 kHz mono int16 WAV with `soundfile` + `scipy.signal.resample_poly`. +The ~9 s segment containing Armstrong's iconic line (≈14.0–23.3 s of the +24 s [`Armstrong_Small_Step.ogg`](https://commons.wikimedia.org/wiki/File:Armstrong_Small_Step.ogg) +from Wikimedia Commons), resampled to 16 kHz mono int16 WAV with +`soundfile` + `scipy.signal.resample_poly`. + +The segment boundaries are located by word-timestamp transcription, not a +fixed offset: the source opens with a *different* utterance — "I'm going +to step off the LM now" — so the original "first ~12 s" trim captured the +wrong sentence and scored the reference below against audio that never +contained it. Regenerate with `python tools/recut_armstrong.py` (needs +outbound network + faster-whisper). - **Source**: https://upload.wikimedia.org/wikipedia/commons/d/dd/Armstrong_Small_Step.ogg - **Original work**: NASA recording of Neil Armstrong stepping onto @@ -44,8 +52,8 @@ from Wikimedia Commons, downsampled from 11 025 Hz mono OGG/Vorbis to federal government ("NASA material is not protected by copyright unless noted"). - **Reference transcript**: `"That's one small step for man, one giant - leap for mankind."` — the well-known phrase Armstrong utters in the - clip. + leap for mankind."` — the iconic phrase, now the actual content of the + trimmed clip. ### `marlene-nb.wav` diff --git a/tests/fixtures/audio/armstrong-en.wav b/tests/fixtures/audio/armstrong-en.wav index 59800c8c..83d9812c 100644 Binary files a/tests/fixtures/audio/armstrong-en.wav and b/tests/fixtures/audio/armstrong-en.wav differ diff --git a/tests/test_audio.py b/tests/test_audio.py index dcc927c5..44b95407 100644 --- a/tests/test_audio.py +++ b/tests/test_audio.py @@ -153,7 +153,7 @@ def _wav_to_frames(path: Path, frame_bytes: int = FRAME_BYTES) -> list[bytes]: def test_int16_peak_norm_armstrong_speech_wav(): - """Stream a real 12 s speech WAV through int16_peak_norm frame by + """Stream a real ~9 s speech WAV through int16_peak_norm frame by frame. The peak series must: - stay strictly within the 0.0..1.0 contract the renderer relies on @@ -175,7 +175,7 @@ def test_int16_peak_norm_armstrong_speech_wav(): synthesised silent WAV instead, not a real clip.""" fixture = FIXTURES_DIR / "armstrong-en.wav" frames = _wav_to_frames(fixture) - assert len(frames) > 500, "expected ~600 frames in a 12 s WAV" + assert len(frames) > 400, "expected ~460 frames in the ~9 s WAV" peaks = [audio.int16_peak_norm(f) for f in frames] assert all(0.0 <= p <= 1.0 for p in peaks), "peak outside [0,1] would break renderer" assert max(peaks) > 0.5, "real speech should peg the meter into the 'hot' zone somewhere" diff --git a/tests/test_live_cmd.py b/tests/test_live_cmd.py index 2509a097..745bcab4 100644 --- a/tests/test_live_cmd.py +++ b/tests/test_live_cmd.py @@ -247,6 +247,15 @@ def test_nb_whisper_uses_model_path_and_backend_policy_not_model_flag(tmp_path: assert "--backend" not in cmd +def test_backend_policy_emitted_only_when_set(): + base = LiveConfig(model="small.en", language="en", host="h", port=8000) + assert "--backend-policy" not in build_live_cmd(EXE, base, use_mlx=True) + + cfg = LiveConfig(model="small.en", language="en", host="h", port=8000, backend_policy="localagreement") + cmd = build_live_cmd(EXE, cfg, use_mlx=True) + assert cmd[cmd.index("--backend-policy") + 1] == "localagreement" + + def test_nb_whisper_ignores_use_mlx_true_in_argv(tmp_path: Path): """NB-Whisper routing forces faster-whisper regardless of operator MLX preference. The argv reflects that — no --backend mlx-whisper.""" diff --git a/tests/test_live_relay.py b/tests/test_live_relay.py index 51c99fca..a7ad7b63 100644 --- a/tests/test_live_relay.py +++ b/tests/test_live_relay.py @@ -228,6 +228,40 @@ async def test_relay_consumes_lines_into_callback_with_close_drain(fake_wlk: _Fa assert list(lines) == ["hello world", "second line"] +async def test_relay_emits_growth_of_a_committed_non_tail_line(fake_wlk: _FakeWlk): + """Regression (subagent Finding 3): when an already-committed, + non-tail line's text GROWS in a later snapshot, the new suffix must + still be emitted promptly — not silently held until close (and then + delivered out of order). + + WlK normally only grows the current tail, but a late LocalAgreement + correction / cross-position merge can extend a line that already has a + newer line after it. The relay must not freeze non-tail positions, or + those words are lost mid-session.""" + lines = _SignalList() + relay = WlKRelay( + host="localhost", + port=fake_wlk.port, + language="en", + on_settled_line=lines.append, + ) + await relay.connect() + l0 = {"text": "hello", "speaker": 1, "start": 0.0, "end": 1.0} + l1 = {"text": "second", "speaker": 1, "start": 1.0, "end": 2.0} + # Snapshot 1: single line, held as the in-flight tail. + await fake_wlk.push_lines_snapshot([l0]) + # Snapshot 2: a newer line appears, so l0 settles and is emitted. + await fake_wlk.push_lines_snapshot([dict(l0, text="hello world"), l1]) + await lines.wait_count(1) + assert list(lines) == ["hello world"] + # Snapshot 3: the now-non-tail l0 grows. Its suffix must be emitted + # immediately, in order — before the still-held tail (l1). + await fake_wlk.push_lines_snapshot([dict(l0, text="hello world again"), l1]) + await lines.wait_count(2) + assert list(lines) == ["hello world", "again"] + await relay.close() + + async def test_relay_dedupes_lines_across_repeated_snapshots(fake_wlk: _FakeWlk): """WlK re-sends the full lines list on every tick. The relay must NOT re-emit a line just because it appeared in three consecutive diff --git a/tests/test_tap_fan_out.py b/tests/test_tap_fan_out.py index fb1c31c4..0608e9e3 100644 --- a/tests/test_tap_fan_out.py +++ b/tests/test_tap_fan_out.py @@ -322,7 +322,7 @@ def _wav_to_frames(path: Path, frame_bytes: int = 640) -> list[bytes]: async def test_write_frame_level_tracks_real_speech_wav(recorder: Recorder): - """End-to-end: stream a real ~12 s speech recording through + """End-to-end: stream a real ~9 s speech recording through TapFanOut frame by frame and verify the per-tap volume meter behaves like a real-world meter would. @@ -337,7 +337,7 @@ async def test_write_frame_level_tracks_real_speech_wav(recorder: Recorder): """ fixture = FIXTURES_AUDIO / "armstrong-en.wav" frames = _wav_to_frames(fixture) - assert len(frames) > 500, "expected ~600 frames from a 12 s WAV" + assert len(frames) > 400, "expected ~460 frames from the ~9 s WAV" levels: list[float] = [] async with await TapFanOut.open( @@ -376,7 +376,10 @@ async def test_write_frame_level_decays_through_silence_after_real_audio(recorde fixture = FIXTURES_AUDIO / "armstrong-en.wav" frames = _wav_to_frames(fixture) silence_frame = b"\x00" * 640 - silent_tail = [silence_frame] * 30 # 30 * 20 ms = 600 ms — well past half-life + # 50 * 20 ms = 1 s. The recut clip ends on the loud "...for mankind" + # (held level ~0.65), so it needs ~1 s to drain below 0.05 at the ~165 ms + # half-life — vs the old clip's quiet tail that cleared in 600 ms. + silent_tail = [silence_frame] * 50 peak_during_speech = 0.0 @@ -398,7 +401,7 @@ async def test_write_frame_level_decays_through_silence_after_real_audio(recorde assert peak_during_speech > 0.3, f"meter never lit up during real speech (peak={peak_during_speech:.3f})" assert final_level_after_silence < 0.05, ( - f"meter failed to decay through 600 ms of silence (final={final_level_after_silence:.4f})" + f"meter failed to decay through 1 s of silence (final={final_level_after_silence:.4f})" ) @@ -843,6 +846,95 @@ def analyze(chunk): assert snap[0].gate_open is False +async def test_lag_only_reported_while_gate_open( + recorder_with_relay: Recorder, + monkeypatch: pytest.MonkeyPatch, +): + """A per-tap lag must only show while the gate is open. WlK's + `remaining_time_transcription` is wall_clock - last_processed_audio, so + it climbs during the silence we gate out even though no audio is in + flight — reporting that would paint a phantom, ever-growing backlog for + a speaker who has gone quiet. So: lag is surfaced while the gate is + open, cleared the instant it closes, and suppressed afterwards.""" + from dataclasses import replace as dc_replace + + from tapscribe.speech_gate import SpeechGate + + recorder_with_relay.live.config = dc_replace( + recorder_with_relay.live.config, gate_kind="tapscribe", gate_pre_roll_ms=0 + ) + + # VAD queue: open on call #1, hold, then close on call #3 (same shape as + # test_gate_open_state_propagates_to_active_stream). + def _open_then_close(*args, **kwargs): + events = [{"start": 0}, None, {"end": 0}] + + def analyze(chunk): + return events.pop(0) if events else None + + return SpeechGate(vad=analyze, pre_roll_ms=0) + + monkeypatch.setattr("tapscribe.tap_fan_out.build_gate_for_config", _open_then_close) + + async def _lag(): + snap = await recorder_with_relay.streams.snapshot() + return snap[0].lag_s + + async with await TapFanOut.open( + recorder_with_relay, + identity="alice", + name="Alice", + utterance_id="utt-lag-gate", + do_record=True, + do_live=True, + ) as fan_out: + # Frames 1-2 open the gate; a metric arriving while open is genuine. + await fan_out.write_frame(PCM_FRAME) + await fan_out.write_frame(PCM_FRAME) + await fan_out._on_metrics(2.0) + assert await _lag() is not None + + # Frames 3-5 drive the gate closed → stale lag cleared immediately. + await fan_out.write_frame(PCM_FRAME) + await fan_out.write_frame(PCM_FRAME) + await fan_out.write_frame(PCM_FRAME) + assert await _lag() is None + + # WlK keeps reporting a climbing remaining_time through the silence; + # it must stay suppressed, not resurface as phantom backlog. + await fan_out._on_metrics(99.0) + assert await _lag() is None + + +async def test_bench_drive_one_stream_uses_production_fan_out( + recorder_with_relay: Recorder, + fake_wlk: FakeWlkThread, +): + """tools/bench_live._drive_one_stream must run through the production + TapFanOut/Recorder path — not a parallel reimplementation. Drive it + against the fake WlK (no model needed) and confirm it opens a real tap, + feeds frames, samples per-tap state, reads settled lines back from + recorder.transcripts, and tears the tap down. Guards the contract that + the benchmark and production share one code path.""" + from tools.bench_live import _drive_one_stream + + frames = [PCM_FRAME] * 20 + metrics, hypothesis, lines = await _drive_one_stream( + recorder_with_relay, identity="alice", name="Alice", frames=frames, speed=8.0 + ) + + # Production-shaped results, read from the real recorder sinks. + assert isinstance(lines, list) + assert isinstance(hypothesis, str) + assert metrics["frames_in"] == len(frames) + for key in ("lag_mean_s", "lag_max_s", "gate_open_pct", "final_delay_s", "buffer_nonempty"): + assert key in metrics, f"missing production metric {key!r}" + # Frames reached the WlK relay (production path, not a side channel). + assert sum(len(c) for c in fake_wlk.received) >= len(PCM_FRAME) + # The tap was registered and then torn down via TapFanOut._close. + assert await recorder_with_relay.streams.snapshot() == [] + + async def test_level_meter_reads_zero_while_gate_is_closed_even_on_loud_input( recorder_with_relay: Recorder, monkeypatch: pytest.MonkeyPatch, diff --git a/tools/bench_live.py b/tools/bench_live.py new file mode 100644 index 00000000..71cf0e0b --- /dev/null +++ b/tools/bench_live.py @@ -0,0 +1,1155 @@ +#!/usr/bin/env python3 +"""Benchmark the *live* transcription path on known audio. + +Unlike `tools/bench_backends.py` (single-shot BATCH transcription), this +drives the exact production live pipeline a Bridge would hit: + + WAV → 20 ms PCM frames → TapFanOut (SpeechGate + WlKRelay) → Recorder + → whisperlivekit-server subprocess → ActiveStreams / LiveTranscripts + +It drives the production path verbatim — a real `Recorder` and +`TapFanOut.write_frame`, exactly as the /tap WebSocket endpoint does — so +the gate decisions, the per-tap lag, the in-flight buffer, and the settled +captions all come from the same code that runs live (read back from +`recorder.streams` / `recorder.transcripts`, the dashboard's own sources). +There is no parallel reimplementation to drift out of sync. It scores those captions +against a reference transcript with a BROAD metric set (WER/CER plus the +substitution/deletion/insertion breakdown, latency, and gate +pass-through) so the numbers themselves reveal what to tune rather than +us guessing up front. + +Usage: + # One WAV, one config (detailed report): + python tools/bench_live.py tests/fixtures/audio/armstrong-en.wav --model base.en + + # Override gate / streaming knobs: + python tools/bench_live.py tests/fixtures/audio/armstrong-en.wav \ + --model small.en --gate-kind backend --min-chunk-size 1.0 + + # Sweep a config matrix across every fixture, write results JSON: + python tools/bench_live.py --sweep + +Requirements (heavy, optional extras): + pip install -e ".[whisper,bench]" +i.e. whisperlivekit (provides whisperlivekit-server), faster-whisper +(CPU/CUDA) or mlx-whisper (Apple Silicon), plus jiwer for scoring. +silero-vad + torch (the gate) are already core dependencies. + +Caveats: + * Frames are paced in REAL TIME by default (--speed 1.0). Feeding + faster than 1x changes WhisperLiveKit's time-sensitive + LocalAgreement / buffer-trimming behaviour, so wall-clock RTF is ~1 + by construction — the meaningful latency signal is WlK's per-tick + `lag_s` (how far behind real time the decoder is) and the + finalization delay after the last frame. + * The first run for a model includes weight download + load; the + readiness wait is generous. If the model can't be fetched (offline + box / network policy), the run reports an error instead of hanging + forever. +""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import json +import re +import sys +import tempfile +import time +import uuid +import wave +from dataclasses import asdict, dataclass, field, replace +from datetime import UTC, datetime +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +# Make the in-tree package importable when run as `python tools/bench_live.py` +# without an editable install (the script's own dir, not the repo root, +# is sys.path[0] in that case). +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from tapscribe.live import LiveConfig, WhisperLiveKitChannel # noqa: E402 +from tapscribe.recorder import Recorder # noqa: E402 +from tapscribe.speech_gate import FRAME_BYTES, SAMPLE_RATE, build_gate_for_config # noqa: E402 +from tapscribe.tap_fan_out import TapFanOut # noqa: E402 + +DEFAULT_FIXTURE_DIR = REPO_ROOT / "tests" / "fixtures" / "audio" +RESULTS_DIR = REPO_ROOT / "bench-results" + +FRAME_MS = 20 +FRAME_INTERVAL_S = FRAME_MS / 1000.0 # 0.02 s — one 20 ms frame at 1x + +# How often (wall-clock seconds) a background task samples the recorder's +# per-tap state, matching the dashboard's poll-based view of lag / gate_open / +# buffer. Kept OFF the feed loop so sampling overhead can't slow the feed +# below real time (which would silently under-load WlK as N grows). +STATE_SAMPLE_INTERVAL_S = 0.1 + + +# --------------------------------------------------------------------------- +# WAV → frames (wire-format only, mirrors tests/e2e/harness.py) +# --------------------------------------------------------------------------- + + +def read_wav_as_pcm_bytes(path: Path) -> bytes: + """Raw 16 kHz mono int16 PCM body of a WAV. Raises if the file isn't + already in the recorder's wire format — same contract the live /tap + path enforces, so a misconverted fixture fails loudly here.""" + with wave.open(str(path), "rb") as w: + if w.getnchannels() != 1 or w.getsampwidth() != 2 or w.getframerate() != SAMPLE_RATE: + raise RuntimeError( + f"{path.name}: expected {SAMPLE_RATE} Hz mono int16, got " + f"{w.getframerate()} Hz / {w.getnchannels()}ch / {w.getsampwidth() * 8}-bit" + ) + return w.readframes(w.getnframes()) + + +def frame_pcm(pcm: bytes) -> list[bytes]: + """Slice raw PCM into 20 ms (640-byte) frames; drop the trailing + partial frame. Mirrors the Bridge wire pattern.""" + n = len(pcm) // FRAME_BYTES + return [pcm[i * FRAME_BYTES : (i + 1) * FRAME_BYTES] for i in range(n)] + + +# --------------------------------------------------------------------------- +# Scoring +# --------------------------------------------------------------------------- + +_PUNCT_RE = re.compile(r"[^\w\s]", flags=re.UNICODE) +_WS_RE = re.compile(r"\s+") + + +def normalize_text(s: str) -> str: + """Lowercase, strip punctuation, collapse whitespace. Language-neutral + so it works for both English and Norwegian fixtures (no Whisper + English normalizer assumptions).""" + s = s.lower().replace("’", "'") + s = _PUNCT_RE.sub(" ", s) + return _WS_RE.sub(" ", s).strip() + + +@dataclass +class Score: + wer: float + cer: float + hits: int + substitutions: int + deletions: int + insertions: int + ref_words: int + hyp_words: int + + +def score_text(reference: str, hypothesis: str) -> Score: + """WER/CER + the hit/sub/del/ins breakdown via jiwer. + + The breakdown is the point: deletions ⇒ words the live path dropped + (gate clipping speech, VAC too aggressive); insertions ⇒ + hallucination / repeats; substitutions ⇒ weak model / wrong + language. Lets the baseline self-diagnose instead of us pre-choosing + a single metric to chase.""" + import jiwer + + ref_n = normalize_text(reference) + hyp_n = normalize_text(hypothesis) + # jiwer raises on an empty reference; an empty hypothesis is fine + # (all deletions). Guard the degenerate empty-reference case. + if not ref_n: + raise ValueError("reference is empty after normalization") + out = jiwer.process_words(ref_n, hyp_n) + cer = jiwer.cer(ref_n, hyp_n) if hyp_n else 1.0 + return Score( + wer=out.wer, + cer=float(cer), + hits=out.hits, + substitutions=out.substitutions, + deletions=out.deletions, + insertions=out.insertions, + ref_words=len(ref_n.split()), + hyp_words=len(hyp_n.split()), + ) + + +# --------------------------------------------------------------------------- +# Run result +# --------------------------------------------------------------------------- + + +@dataclass +class RunResult: + fixture: str + config: dict + error: str | None = None + hypothesis: str = "" + reference: str = "" + settled_lines: list[str] = field(default_factory=list) + metrics: dict = field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# Live-path driver +# --------------------------------------------------------------------------- + + +async def _wait_until_ready(channel: WhisperLiveKitChannel, *, timeout: float) -> str | None: + """Poll the channel's INFO state until it reaches 'running'. Returns + None on success, else an error string. Detects the 'error' state and + an early child exit so a model that can't download fails fast instead + of hanging until the timeout.""" + deadline = time.time() + timeout + while time.time() < deadline: + state = channel.info.get("state") + if state == "running": + return None + if state == "error": + return channel.info.get("last_error") or "whisperlivekit-server reported error" + if not channel.running(): + tail = " | ".join(list(channel.log)[-5:]) + return f"whisperlivekit-server exited early: {tail or channel.info.get('last_error', '')}" + await asyncio.sleep(0.1) + return f"whisperlivekit-server not ready within {timeout:.0f}s (model load/download too slow?)" + + +def _build_bench_recorder(cfg: LiveConfig, use_mlx: bool) -> Recorder: + """A throwaway production Recorder (temp dirs) wired to a real + WhisperLiveKitChannel from `cfg`. The bench drives this exactly like + the /tap endpoint does, so it exercises the live path verbatim.""" + tmp = Path(tempfile.mkdtemp(prefix="bench-live-")) + (tmp / "recordings").mkdir() + (tmp / "config").mkdir() + return Recorder( + recordings_dir=tmp / "recordings", + config_dir=tmp / "config", + live_config=cfg, + use_mlx=use_mlx, + auth_password_file=tmp / ".auth-password", + ) + + +async def _drive_one_stream( + recorder: Recorder, + *, + identity: str, + name: str, + frames: list[bytes], + speed: float, + start_delay_s: float = 0.0, +) -> tuple[dict, str, list[str]]: + """Drive ONE gated stream through the PRODUCTION fan-out and return + (per-stream metrics, hypothesis, settled lines). + + Opens a real `TapFanOut` against `recorder` and feeds it 20 ms frames + exactly as the /tap WS would, so the SpeechGate, the WlKRelay, the + per-tap `lag_s` (with its gate-closed suppression), the in-flight + `buffer_transcription`, and the settled captions are all produced by + the production code — no parallel reimplementation to drift out of + sync. lag / gate_open / buffer are sampled from `recorder.streams` + (the same snapshot `/api/state` serves); settled lines are read back + from `recorder.transcripts`, filtered to this stream's identity. + + `start_delay_s` delays *opening* the tap, modelling a real bridge that + opens a fresh /tap WS when its speaker starts an utterance (the bridge + contract is one WebSocket per utterance). Staggering the open — rather + than padding a long-lived connection with leading silence — keeps WlK's + per-connection clock anchored at speech onset, so lag isn't inflated by + silence that, in production, simply wouldn't be on an open connection.""" + if start_delay_s > 0: + await asyncio.sleep(start_delay_s) + frame_interval = FRAME_INTERVAL_S / max(speed, 0.01) + + utterance_id = f"bench-{identity}-{uuid.uuid4().hex[:8]}" + fan = await TapFanOut.open( + recorder, + identity=identity, + name=name, + utterance_id=utterance_id, + do_record=False, + do_live=True, + ) + # Live was requested but the relay never attached (WlK down / connect + # failed): the stream can't produce captions, so flag it errored rather + # than letting it masquerade as a zero-caption (WER 1.0) result. + if not fan._relay_alive: + await fan._close() + return {"error": "WlK relay did not connect"}, "", [] + conn_id = fan._conn_id + + lag_samples: list[float] = [] + buffers: list[str] = [] + gate_open_hits = 0 + samples = 0 + + async def _sample() -> None: + nonlocal gate_open_hits, samples + for s in await recorder.streams.snapshot(): + if s.conn_id != conn_id: + continue + samples += 1 + if s.gate_open: + gate_open_hits += 1 + if s.lag_s is not None: + lag_samples.append(s.lag_s) + buf = (s.buffer_transcription or "").strip() + if buf: + buffers.append(buf) + break + + # Sample per-tap state from a background task, off the feed path. + sampling = True + + async def _sampler() -> None: + while sampling: + await _sample() + await asyncio.sleep(STATE_SAMPLE_INTERVAL_S) + + sampler_task = asyncio.create_task(_sampler()) + + # try/finally so a mid-feed exception can't leak the sampler task or the + # tap's open relay + ActiveStream — important under --concurrency, where a + # leaked stream would otherwise dangle for the rest of the sweep. + try: + # Pace against an ABSOLUTE schedule (target = start + i*interval) so + # per-frame production work (gate, relay send, ActiveStream lock) is + # absorbed instead of stacked on top of a fixed sleep — otherwise the + # feed drifts below real time as N grows and WlK looks falsely + # under-loaded. `max_slip` is how far behind schedule we fell: > ~0 + # means the host couldn't feed this many streams in real time, so the + # numbers are soft. + start = time.perf_counter() + max_slip = 0.0 + for i, frame in enumerate(frames): + await fan.write_frame(frame) + slip = time.perf_counter() - (start + (i + 1) * frame_interval) + max_slip = max(max_slip, slip) + if slip < 0: + await asyncio.sleep(-slip) + last_frame_wall = time.perf_counter() + sampling = False + await sampler_task + await _sample() + finally: + sampling = False + if not sampler_task.done(): + sampler_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await sampler_task + await fan._close() + final_delay = time.perf_counter() - last_frame_wall + + lines = [e["text"] for e in recorder.transcripts.snapshot() if e.get("identity") == identity] + hypothesis = " ".join(lines) + metrics: dict = { + "n_lines": len(lines), + "frames_in": len(frames), + "gate_open_pct": round(100.0 * gate_open_hits / max(1, samples), 1), + "lag_mean_s": round(sum(lag_samples) / len(lag_samples), 3) if lag_samples else None, + "lag_max_s": round(max(lag_samples), 3) if lag_samples else None, + "lag_samples": len(lag_samples), + "final_delay_s": round(final_delay, 2), + "pacing_slip_s": round(max(0.0, max_slip), 2), + "buffer_nonempty": len(buffers), + "buffer_sample": buffers[-1] if buffers else "", + } + return metrics, hypothesis, lines + + +async def run_one( + wav: Path, + cfg: LiveConfig, + *, + use_mlx: bool, + speed: float, + ready_timeout: float, + verbose: bool, +) -> RunResult: + """Drive one WAV through the production live path under one LiveConfig + and score the captions. Owns the whisperlivekit-server child for the + duration.""" + cfg_summary = _config_summary(cfg) + reference = _read_reference(wav) + + recorder = _build_bench_recorder(cfg, use_mlx) + ok, msg = recorder.live.start() + if not ok: + return RunResult( + fixture=wav.stem, config=cfg_summary, error=f"start failed: {msg}", reference=reference + ) + + try: + if verbose: + print( + f" [{wav.stem}] waiting for whisperlivekit-server (model={cfg.model}, mlx={use_mlx})...", + flush=True, + ) + err = await _wait_until_ready(recorder.live, timeout=ready_timeout) + if err is not None: + return RunResult(fixture=wav.stem, config=cfg_summary, error=err, reference=reference) + + recorder.transcripts.clear() + frames = frame_pcm(read_wav_as_pcm_bytes(wav)) + audio_s = len(frames) * FRAME_INTERVAL_S + feed_start = time.perf_counter() + smetrics, hypothesis, lines = await _drive_one_stream( + recorder, identity="bench", name="Bench Speaker", frames=frames, speed=speed + ) + wall_s = time.perf_counter() - feed_start + if "error" in smetrics: + return RunResult( + fixture=wav.stem, config=cfg_summary, error=smetrics["error"], reference=reference + ) + + metrics: dict = {"audio_s": round(audio_s, 2), "wall_s": round(wall_s, 2), **smetrics} + result = RunResult( + fixture=wav.stem, + config=cfg_summary, + hypothesis=hypothesis, + reference=reference, + settled_lines=lines, + metrics=metrics, + ) + try: + score = score_text(reference, hypothesis) + result.metrics.update(asdict(score)) + except Exception as e: # scoring is best-effort; keep raw output + result.metrics["score_error"] = str(e) + return result + finally: + recorder.live.stop() + + +def _aggregate_streams(outs: list[tuple[dict, str, list]], n: int, reference: str) -> dict: + errored = sum(1 for m, _, _ in outs if "error" in m) + lag_means = [m["lag_mean_s"] for m, _, _ in outs if m.get("lag_mean_s") is not None] + lag_maxes = [m["lag_max_s"] for m, _, _ in outs if m.get("lag_max_s") is not None] + fin = [m["final_delay_s"] for m, _, _ in outs if "final_delay_s" in m] + slips = [m["pacing_slip_s"] for m, _, _ in outs if "pacing_slip_s" in m] + wers: list[float] = [] + for m, hyp, _l in outs: + if "error" in m: + continue # didn't run — counted via `errored`, kept out of the WER mean + try: + wers.append(score_text(reference, hyp).wer) + except Exception: # empty reference etc. — skip; can't score + pass + return { + "streams": n, + "errored": errored, + "lag_mean_s": round(sum(lag_means) / len(lag_means), 2) if lag_means else None, + "lag_max_s": round(max(lag_maxes), 2) if lag_maxes else None, + "wer_mean": round(sum(wers) / len(wers), 2) if wers else None, + "fin_delay_max_s": round(max(fin), 2) if fin else None, + "pacing_slip_s": round(max(slips), 2) if slips else None, + } + + +def _print_concurrency_row(row: dict) -> None: + print( + f" streams={row['streams']}: lagμ={row['lag_mean_s']} lagX={row['lag_max_s']} " + f"WERμ={row['wer_mean']} finΔX={row['fin_delay_max_s']} slip={row.get('pacing_slip_s')} " + f"errored={row['errored']}", + flush=True, + ) + + +async def run_concurrency_sweep( + wav: Path, + cfg: LiveConfig, + *, + counts: list[int], + stagger_s: float, + use_mlx: bool, + speed: float, + ready_timeout: float, + verbose: bool, +) -> list[dict]: + """Stress the production live path with N concurrent gated /tap streams + of `wav` and report how lag / WER degrade as N grows. + + This is the multi-speaker reproduction, driven exactly as production + runs it: ONE shared `WhisperLiveKitChannel` (loaded once, reused across + all N) with N concurrent `TapFanOut` streams relaying into it — every + active tap contends for the one decoder, same as the real /tap fan-out. + Each stream gets its own identity so its settled captions can be read + back from `recorder.transcripts`. + + `stagger_s` offsets when each stream OPENS its tap (`stagger_s * index`), + modelling the bridge contract of one /tap WebSocket per utterance: a + later speaker's connection opens when they start talking, not at t0. So + stagger=0 is full overlap (all taps open at once) and stagger >= clip + length is pure turn-taking (taps open and close one after another, ~1 + connection live at a time). Staggering the OPEN — vs. padding a single + long-lived connection with leading silence — keeps WlK's per-connection + clock anchored at speech onset so lag reflects real backlog.""" + reference = _read_reference(wav) + frames = frame_pcm(read_wav_as_pcm_bytes(wav)) + + recorder = _build_bench_recorder(cfg, use_mlx) + ok, msg = recorder.live.start() + if not ok: + print(f"start failed: {msg}", file=sys.stderr) + return [] + + rows: list[dict] = [] + try: + err = await _wait_until_ready(recorder.live, timeout=ready_timeout) + if err is not None: + print(f"whisperlivekit-server not ready: {err}", file=sys.stderr) + return [] + + for n in counts: + recorder.transcripts.clear() + # return_exceptions=True so one stream blowing up is recorded as an + # errored row instead of aborting the sweep (and losing the rows + # already computed for smaller N). _drive_one_stream's try/finally + # has already closed that stream's tap by the time we see the exc. + raw = await asyncio.gather( + *( + _drive_one_stream( + recorder, + identity=f"spk{i}", + name=f"Speaker {i}", + frames=frames, + speed=speed, + start_delay_s=stagger_s * i, + ) + for i in range(n) + ), + return_exceptions=True, + ) + outs = [({"error": repr(o)}, "", []) if isinstance(o, BaseException) else o for o in raw] + row = _aggregate_streams(outs, n, reference) + rows.append(row) + if verbose: + _print_concurrency_row(row) + finally: + recorder.live.stop() + return rows + + +# --------------------------------------------------------------------------- +# Config / fixtures helpers +# --------------------------------------------------------------------------- + +# Knobs we vary; everything else stays at LiveConfig defaults. Kept as a +# tuple so the table printer and the JSON summary agree on the columns. +_SUMMARY_FIELDS = ( + "model", + "language", + "gate_kind", + "gate_min_speech_ms", + "min_chunk_size", + "buffer_trimming", + "buffer_trimming_sec", + "confidence_validation", +) + + +def _config_summary(cfg: LiveConfig) -> dict: + return {f: getattr(cfg, f) for f in _SUMMARY_FIELDS} + + +def _read_reference(wav: Path) -> str: + ref = wav.with_suffix("").with_suffix(".reference.txt") + if not ref.exists(): + # `armstrong-en.wav` → `armstrong-en.reference.txt` + ref = wav.parent / f"{wav.stem}.reference.txt" + return ref.read_text(encoding="utf-8").strip() if ref.exists() else "" + + +def discover_fixtures(fixture_dir: Path) -> list[Path]: + """Every *.wav in the dir that has a paired *.reference.txt.""" + out = [] + for wav in sorted(fixture_dir.glob("*.wav")): + ref = fixture_dir / f"{wav.stem}.reference.txt" + if ref.exists(): + out.append(wav) + return out + + +def detect_use_mlx() -> bool: + """MLX is the natural live backend on Apple Silicon when mlx-whisper + is importable. Everywhere else (incl. this Linux CI box) use the + faster-whisper CPU/CUDA path.""" + import platform + + if platform.system() != "Darwin" or platform.machine() != "arm64": + return False + try: + import mlx_whisper # noqa: F401 + + return True + except ImportError: + return False + + +# Language per fixture stem so the sweep picks the right WlK --lan. +_FIXTURE_LANG = {"marlene-nb": "no"} + + +def _lang_for_fixture(wav: Path) -> str: + return _FIXTURE_LANG.get(wav.stem, "en") + + +# Full-pipeline sweep matrix: each dict is a set of LiveConfig field +# overrides applied (via dataclasses.replace) onto a per-fixture baseline +# (the fixture's language). Row 0 of each matrix is the baseline every +# other row is compared against. Edit freely; any LiveConfig field name is +# a valid key. +# +# There are two matrices because the MODEL has to match the fixture's +# language: English-only Whisper (`*.en`) cannot transcribe Norwegian +# regardless of `--lan`, so a Norwegian fixture swept on `.en` models +# scores pure noise (this is exactly the bogus `marlene-nb` baseline that +# the first sweep produced). `_sweep_matrix_for` picks the right one. +# +# The non-model rows are chosen to isolate the hypotheses in +# docs/live-tuning-research.md: model size (1-3), the confidence/accuracy +# trade (4), whether our gate clips speech vs the backend VAD (5), blip +# suppression (6), and the WlK streaming knobs (7-8). + +# English fixtures — English-only Whisper. Row 0 is the PRODUCTION DEFAULT. +SWEEP_MATRIX_EN: list[dict] = [ + {"model": "tiny.en"}, # production default — baseline + {"model": "base.en"}, + {"model": "small.en"}, + {"model": "small.en", "confidence_validation": False}, + {"model": "small.en", "gate_kind": "backend"}, + {"model": "small.en", "gate_min_speech_ms": 200}, + {"model": "small.en", "min_chunk_size": 1.0}, + {"model": "small.en", "buffer_trimming": "segment"}, +] + +# Norwegian fixtures — NB-Whisper (NbAiLab, Norwegian-tuned). The channel +# auto-downloads the CT2 weights and build_live_cmd routes these via +# --model-path + --backend-policy localagreement; every other knob below +# still applies. Mirrors the EN matrix so the two are read side by side. +SWEEP_MATRIX_NB: list[dict] = [ + {"model": "nb-whisper-tiny"}, # Norwegian baseline + {"model": "nb-whisper-base"}, + {"model": "nb-whisper-small"}, + {"model": "nb-whisper-small", "confidence_validation": False}, + {"model": "nb-whisper-small", "gate_kind": "backend"}, + {"model": "nb-whisper-small", "gate_min_speech_ms": 200}, + {"model": "nb-whisper-small", "min_chunk_size": 1.0}, + {"model": "nb-whisper-small", "buffer_trimming": "segment"}, +] + + +def _sweep_matrix_for(wav: Path) -> list[dict]: + """Pick the model sweep that matches the fixture's language. Norwegian + fixtures need NB-Whisper; everything else uses the English `.en` + matrix.""" + return SWEEP_MATRIX_NB if _lang_for_fixture(wav) == "no" else SWEEP_MATRIX_EN + + +def _config_from_overrides(overrides: dict, *, language: str, host: str) -> LiveConfig: + """Build a LiveConfig from a per-fixture baseline plus a matrix row. + `replace` applies any LiveConfig field, so a row can tune gate knobs, + confidence_validation, streaming knobs — not just the hardcoded few.""" + base = LiveConfig(model="tiny.en", language=language, host=host, port=0) + return replace(base, **overrides) + + +# --------------------------------------------------------------------------- +# Output +# --------------------------------------------------------------------------- + + +def _fmt(v: object, width: int) -> str: + if v is None: + s = "-" + elif isinstance(v, float): + s = f"{v:.2f}" + else: + s = str(v) + return s[:width].rjust(width) + + +def print_table(results: list[RunResult]) -> None: + cols = [ + ("fixture", 12, lambda r: r.fixture), + ("model", 9, lambda r: r.config.get("model")), + ("gate", 9, lambda r: r.config.get("gate_kind")), + ("WER", 6, lambda r: r.metrics.get("wer")), + ("CER", 6, lambda r: r.metrics.get("cer")), + ("sub", 4, lambda r: r.metrics.get("substitutions")), + ("del", 4, lambda r: r.metrics.get("deletions")), + ("ins", 4, lambda r: r.metrics.get("insertions")), + ("lagμ", 6, lambda r: r.metrics.get("lag_mean_s")), + ("lagX", 6, lambda r: r.metrics.get("lag_max_s")), + ("finΔ", 6, lambda r: r.metrics.get("final_delay_s")), + ("gateOn%", 7, lambda r: r.metrics.get("gate_open_pct")), + ("lines", 5, lambda r: r.metrics.get("n_lines")), + ] + header = " ".join(name.rjust(w) if i else name.ljust(w) for i, (name, w, _) in enumerate(cols)) + print("=" * len(header)) + print(header) + print("-" * len(header)) + for r in results: + if r.error: + print(f"{r.fixture[:12].ljust(12)} {r.config.get('model', '')[:9]:>9} ERROR: {r.error}") + continue + row = [] + for i, (_name, w, get) in enumerate(cols): + v = get(r) + row.append(v[:w].ljust(w) if i == 0 and isinstance(v, str) else _fmt(v, w)) + print(" ".join(row)) + print("=" * len(header)) + print( + "WER/CER lower=better. sub/del/ins = word substitutions / deletions (dropped) / " + "insertions (hallucinated)." + ) + print( + "lagμ/lagX = mean/max WlK decode lag (s); finΔ = finalization delay after last frame (s); " + "gate% = frames forwarded." + ) + + +def write_results_json(results: list[RunResult], *, use_mlx: bool, speed: float) -> Path: + RESULTS_DIR.mkdir(exist_ok=True) + stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + path = RESULTS_DIR / f"live-{stamp}.json" + payload = { + "created": datetime.now(UTC).isoformat(), + "host_backend": "mlx-whisper" if use_mlx else "faster-whisper", + "speed": speed, + "runs": [ + { + "fixture": r.fixture, + "config": r.config, + "error": r.error, + "metrics": r.metrics, + "reference": r.reference, + "hypothesis": r.hypothesis, + "settled_lines": r.settled_lines, + } + for r in results + ], + } + path.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") + return path + + +def print_concurrency_table(rows: list[dict], *, wav: Path, cfg: LiveConfig, stagger_s: float) -> None: + overlap = "full overlap" if stagger_s == 0 else f"speech staggered {stagger_s:g}s/stream" + print( + f"concurrency stress — fixture={wav.stem} model={cfg.model} gate={cfg.gate_kind} " + f"({overlap}, 1 shared server, production fan-out)" + ) + cols = [ + ("streams", 8, lambda r: r.get("streams")), + ("lagμ", 7, lambda r: r.get("lag_mean_s")), + ("lagX", 7, lambda r: r.get("lag_max_s")), + ("finΔX", 7, lambda r: r.get("fin_delay_max_s")), + ("slipX", 7, lambda r: r.get("pacing_slip_s")), + ("WERμ", 7, lambda r: r.get("wer_mean")), + ("errored", 8, lambda r: r.get("errored")), + ] + header = " ".join(name.rjust(w) for name, w, _ in cols) + print("=" * len(header)) + print(header) + print("-" * len(header)) + for r in rows: + print(" ".join(_fmt(get(r), w) for _name, w, get in cols)) + print("=" * len(header)) + print( + "lag/WER are read from the production ActiveStreams + LiveTranscripts (same source as the " + "dashboard). Each stream opens its tap at stagger*i (one /tap per utterance, as the bridge" + ) + print( + "does). lagμ/lagX climbing with N = decode contention; WERμ rising / errored>0 = dropped" + " speech. slipX = worst real-time pacing slip; if it's > ~0.1s the host couldn't feed N" + ) + print( + "streams in real time, so lag/WER for that row are soft (the feed under-loaded WlK)." + " Compare stagger=0 (all overlap) vs large stagger (turn-taking, ~1 tap live)." + ) + + +def print_detailed(result: RunResult) -> None: + print() + print(f"fixture: {result.fixture}") + print(f"config: {json.dumps(result.config)}") + if result.error: + print(f"ERROR: {result.error}") + return + print(f"reference: {result.reference!r}") + print(f"hypothesis: {result.hypothesis!r}") + print() + print("settled lines:") + for ln in result.settled_lines: + print(f" · {ln}") + print() + print("metrics:") + for k, v in result.metrics.items(): + print(f" {k:18} {v}") + + +# --------------------------------------------------------------------------- +# Gate-only analysis (no ASR model — Silero loads locally) +# --------------------------------------------------------------------------- +# +# The SpeechGate is the front half of the live path and a prime quality +# suspect: if it clips speech, words are dropped before the backend ever +# sees them. Silero VAD loads locally with no network, so we can measure +# exactly how much audio each gate config forwards — useful on boxes that +# can't download an ASR model (and fast: no subprocess, no real-time +# pacing). This is NOT a transcription benchmark; it's a gate-aggression +# benchmark. + +_GATE_SUMMARY_FIELDS = ( + "gate_kind", + "gate_speech_threshold", + "gate_hangover_ms", + "gate_pre_roll_ms", + "gate_min_speech_ms", +) + +# Gate-config matrix for --gate-only. Row 0 is the production default. +GATE_MATRIX: list[dict] = [ + {}, # production default: thr 0.5, hang 400, pre-roll 300, min-speech 0 + {"gate_speech_threshold": 0.3}, + {"gate_speech_threshold": 0.7}, + {"gate_hangover_ms": 800}, + {"gate_pre_roll_ms": 500}, + {"gate_min_speech_ms": 200}, + {"gate_kind": "backend"}, # no TapScribe gate → forwards everything +] + + +@dataclass +class GateStats: + fixture: str + config: dict + frames_in: int + frames_forwarded: int + forward_pct: float + segments: int + retained_s: float + audio_s: float + + +def analyze_gate(wav: Path, cfg: LiveConfig) -> GateStats: + """Feed a WAV through the SpeechGate and report how much it forwards. + + `segments` counts silence→speech openings (gate.is_open False→True), + a proxy for how finely the gate chops the audio. `gate_kind=backend` + means no TapScribe gate, so everything is forwarded as one segment.""" + gate = build_gate_for_config(cfg) + frames = frame_pcm(read_wav_as_pcm_bytes(wav)) + forwarded = 0 + segments = 0 + prev_open = False + for fr in frames: + out = gate.feed(fr) if gate is not None else [fr] + forwarded += len(out) + now_open = gate.is_open if gate is not None else True + if now_open and not prev_open: + segments += 1 + prev_open = now_open + n = len(frames) + return GateStats( + fixture=wav.stem, + config={f: getattr(cfg, f) for f in _GATE_SUMMARY_FIELDS}, + frames_in=n, + frames_forwarded=forwarded, + forward_pct=round(100.0 * forwarded / max(1, n), 1), + segments=segments, + retained_s=round(forwarded * FRAME_INTERVAL_S, 2), + audio_s=round(n * FRAME_INTERVAL_S, 2), + ) + + +def print_gate_table(rows: list[GateStats]) -> None: + cols = [ + ("fixture", 12, lambda r: r.fixture), + ("kind", 9, lambda r: r.config.get("gate_kind")), + ("thr", 5, lambda r: r.config.get("gate_speech_threshold")), + ("hang", 5, lambda r: r.config.get("gate_hangover_ms")), + ("proll", 5, lambda r: r.config.get("gate_pre_roll_ms")), + ("minsp", 5, lambda r: r.config.get("gate_min_speech_ms")), + ("fwd%", 6, lambda r: r.forward_pct), + ("segs", 5, lambda r: r.segments), + ("kept_s", 7, lambda r: r.retained_s), + ("audio_s", 7, lambda r: r.audio_s), + ] + header = " ".join(name.rjust(w) if i else name.ljust(w) for i, (name, w, _) in enumerate(cols)) + print("=" * len(header)) + print(header) + print("-" * len(header)) + for r in rows: + cells = [] + for i, (_n, w, get) in enumerate(cols): + v = get(r) + cells.append(v[:w].ljust(w) if i == 0 and isinstance(v, str) else _fmt(v, w)) + print(" ".join(cells)) + print("=" * len(header)) + print( + "fwd% = frames forwarded to backend; segs = silence→speech openings; " + "kept_s = forwarded audio seconds." + ) + print("A low fwd% on a mostly-speech clip suggests the gate is clipping speech (→ dropped words).") + + +def run_gate_sweep(fixture_dir: Path) -> list[GateStats]: + fixtures = discover_fixtures(fixture_dir) + if not fixtures: + print(f"No fixtures under {fixture_dir}", file=sys.stderr) + return [] + rows: list[GateStats] = [] + for wav in fixtures: + for overrides in GATE_MATRIX: + cfg = _config_from_overrides(overrides, language=_lang_for_fixture(wav), host="127.0.0.1") + rows.append(analyze_gate(wav, cfg)) + return rows + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_base_config(args) -> LiveConfig: + kwargs: dict = dict( + model=args.model, + language=args.language, + host=args.live_host, + port=0, # ephemeral — channel picks a free port at spawn + gate_kind=args.gate_kind, + ) + if args.min_chunk_size is not None: + kwargs["min_chunk_size"] = args.min_chunk_size + if args.buffer_trimming is not None: + kwargs["buffer_trimming"] = args.buffer_trimming + if args.buffer_trimming_sec is not None: + kwargs["buffer_trimming_sec"] = args.buffer_trimming_sec + if args.gate_min_speech_ms is not None: + kwargs["gate_min_speech_ms"] = args.gate_min_speech_ms + if args.confidence_validation is not None: + kwargs["confidence_validation"] = args.confidence_validation + if args.backend_policy is not None: + kwargs["backend_policy"] = args.backend_policy + return LiveConfig(**kwargs) + + +async def run_sweep(args, *, use_mlx: bool) -> list[RunResult]: + fixtures = discover_fixtures(Path(args.fixture_dir)) + if not fixtures: + print(f"No fixtures (paired *.wav + *.reference.txt) under {args.fixture_dir}", file=sys.stderr) + return [] + results: list[RunResult] = [] + total = sum(len(_sweep_matrix_for(w)) for w in fixtures) + i = 0 + for wav in fixtures: + lang = _lang_for_fixture(wav) + for overrides in _sweep_matrix_for(wav): + i += 1 + cfg = _config_from_overrides(overrides, language=lang, host=args.live_host) + print(f"[{i}/{total}] {wav.stem} {_config_summary(cfg)}", flush=True) + res = await run_one( + wav, + cfg, + use_mlx=use_mlx, + speed=args.speed, + ready_timeout=args.ready_timeout, + verbose=args.verbose, + ) + if res.error: + print(f" -> ERROR: {res.error}", flush=True) + else: + m = res.metrics + print( + f" -> WER={m.get('wer')} del={m.get('deletions')} " + f"ins={m.get('insertions')} lagX={m.get('lag_max_s')}", + flush=True, + ) + results.append(res) + return results + + +def main() -> None: + p = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + p.add_argument("wav", nargs="?", type=Path, help="WAV to stream (omit with --sweep)") + p.add_argument("--sweep", action="store_true", help="Run the config matrix over every fixture") + p.add_argument( + "--gate-only", + action="store_true", + help="Model-free: sweep GATE configs over every fixture and report how much audio each " + "forwards (no ASR model needed — Silero loads locally).", + ) + p.add_argument( + "--concurrency", + default=None, + metavar="N1,N2,...", + help="Multi-speaker stress: feed this many concurrent copies of the fixture into one WlK " + "server (e.g. '1,2,3,4') and report how lag/WER degrade with load. Uses --wav (default: " + "armstrong-en).", + ) + p.add_argument( + "--stagger", + type=float, + default=0.0, + metavar="SECONDS", + help="With --concurrency: offset when each stream OPENS its tap by this many seconds (one " + "/tap per utterance, as the bridge does). 0 = all taps open at once (full overlap); >= clip " + "length = pure turn-taking (taps open/close in sequence, ~1 live at a time).", + ) + p.add_argument("--fixture-dir", default=str(DEFAULT_FIXTURE_DIR), help="Fixture directory for --sweep") + p.add_argument( + "--model", default="tiny.en", help="WhisperLiveKit model (default: tiny.en, the prod default)" + ) + p.add_argument("--language", default="en", help="Language hint (en, no, auto). Default: en") + p.add_argument("--gate-kind", choices=("tapscribe", "backend"), default="tapscribe") + p.add_argument("--gate-min-speech-ms", type=int, default=None) + p.add_argument( + "--confidence-validation", + dest="confidence_validation", + action="store_true", + default=None, + help="Force WlK confidence-validation on (commits tokens fast; no in-flight buffer).", + ) + p.add_argument( + "--no-confidence-validation", + dest="confidence_validation", + action="store_false", + help="Turn confidence-validation off (LocalAgreement; populates the in-flight buffer preview).", + ) + p.add_argument( + "--backend-policy", + dest="backend_policy", + choices=("simulstreaming", "localagreement"), + default=None, + help="WlK transcription policy. Default (None) = WlK's own default (simulstreaming: commits " + "as it decodes, empty in-flight buffer). 'localagreement' holds tokens until they agree, " + "populating buffer_transcription (the dashboard's in-flight preview).", + ) + p.add_argument("--min-chunk-size", type=float, default=None) + p.add_argument("--buffer-trimming", choices=("sentence", "segment"), default=None) + p.add_argument("--buffer-trimming-sec", type=float, default=None) + p.add_argument("--live-host", default="127.0.0.1") + p.add_argument( + "--speed", + type=float, + default=1.0, + help="Frame pacing multiplier. 1.0 = real time (faithful). >1 is faster but " + "changes WlK's time-based commit behaviour — use with caution.", + ) + p.add_argument( + "--ready-timeout", + type=float, + default=240.0, + help="Max seconds to wait for whisperlivekit-server to come up (first run downloads weights)", + ) + p.add_argument("--mlx", dest="mlx", action="store_true", default=None, help="Force MLX backend") + p.add_argument("--no-mlx", dest="mlx", action="store_false", help="Force faster-whisper (CPU/CUDA)") + p.add_argument("--json", action="store_true", help="Also write a results JSON (always on for --sweep)") + p.add_argument("--verbose", action="store_true") + args = p.parse_args() + + use_mlx = detect_use_mlx() if args.mlx is None else args.mlx + + # Gate-only mode needs neither jiwer nor an ASR model — handle it + # before the scoring-dep check so it runs on a model-less box. + if args.gate_only: + rows = run_gate_sweep(Path(args.fixture_dir)) + if not rows: + sys.exit(1) + print() + print_gate_table(rows) + return + + # Fail fast with an actionable message if jiwer isn't importable — + # scoring is the whole point and a cryptic ImportError mid-run wastes + # a model load. + try: + import jiwer # noqa: F401 + except ImportError: + print("ERROR: jiwer is required for scoring. Install with: pip install jiwer", file=sys.stderr) + sys.exit(1) + + if args.concurrency: + counts: list[int] = [] + try: + counts = [int(x) for x in args.concurrency.split(",") if x.strip()] + except ValueError: + p.error("--concurrency must be a comma list of integers, e.g. 1,2,4") + if not counts or any(c < 1 for c in counts): + p.error("--concurrency needs positive stream counts, e.g. 1,2,4") + wav = args.wav or (DEFAULT_FIXTURE_DIR / "armstrong-en.wav") + if not wav.exists(): + print(f"ERROR: {wav} not found", file=sys.stderr) + sys.exit(1) + cfg = build_base_config(args) + print( + f"backend: {'mlx-whisper' if use_mlx else 'faster-whisper'} " + f"concurrency {counts} stagger={args.stagger:g}s on {wav.stem} " + f"config: {_config_summary(cfg)}", + flush=True, + ) + rows = asyncio.run( + run_concurrency_sweep( + wav, + cfg, + counts=counts, + stagger_s=args.stagger, + use_mlx=use_mlx, + speed=args.speed, + ready_timeout=args.ready_timeout, + verbose=True, + ) + ) + if not rows: + sys.exit(1) + print() + print_concurrency_table(rows, wav=wav, cfg=cfg, stagger_s=args.stagger) + return + + if args.sweep: + results = asyncio.run(run_sweep(args, use_mlx=use_mlx)) + if not results: + sys.exit(1) + print() + print_table(results) + out = write_results_json(results, use_mlx=use_mlx, speed=args.speed) + print(f"\nresults written to {out}") + return + + if args.wav is None: + p.error("provide a WAV path or use --sweep") + if not args.wav.exists(): + print(f"ERROR: {args.wav} not found", file=sys.stderr) + sys.exit(1) + + cfg = build_base_config(args) + print( + f"backend: {'mlx-whisper' if use_mlx else 'faster-whisper'} config: {_config_summary(cfg)}", + flush=True, + ) + result = asyncio.run( + run_one( + args.wav, + cfg, + use_mlx=use_mlx, + speed=args.speed, + ready_timeout=args.ready_timeout, + verbose=True, + ) + ) + print_detailed(result) + print() + print_table([result]) + if args.json: + out = write_results_json([result], use_mlx=use_mlx, speed=args.speed) + print(f"\nresults written to {out}") + if result.error: + sys.exit(2) + + +if __name__ == "__main__": + main() diff --git a/tools/recut_armstrong.py b/tools/recut_armstrong.py new file mode 100644 index 00000000..01da2de4 --- /dev/null +++ b/tools/recut_armstrong.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Regenerate `tests/fixtures/audio/armstrong-en.wav` from its source. + +The fixture must contain Armstrong's iconic line — the exact text in +`armstrong-en.reference.txt`. The source recording, however, opens with a +*different* utterance ("I'm going to step off the LM now"), so a fixed +"first N seconds" trim grabs the wrong sentence (and the live benchmark +then scores good transcripts against a reference the audio never spoke). + +This script avoids that by locating the line via word-timestamp +transcription rather than a hardcoded offset: download the source, find +the "small … mankind" span, trim to it (+ padding), write 16 kHz mono +int16 PCM, then re-transcribe the cut so the result is self-verifying. + +Run on a box WITH outbound network + faster-whisper (e.g. a dev laptop): + + python tools/recut_armstrong.py + +Needs `soundfile` + `scipy` for OGG decode/resample (pip install if +missing) and `faster-whisper` (the `[whisper]` / `[bench]` extra). +""" + +from __future__ import annotations + +import io +import sys +import urllib.request +import wave +from math import gcd +from pathlib import Path + +import numpy as np +import soundfile as sf +from scipy.signal import resample_poly + +SRC = "https://upload.wikimedia.org/wikipedia/commons/d/dd/Armstrong_Small_Step.ogg" +OUT = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "audio" / "armstrong-en.wav" +SAMPLE_RATE = 16000 +PAD_S = 0.35 # margin each side of the located span so word edges aren't clipped + + +def load_16k_mono(url: str) -> np.ndarray: + """Download `url` and return it as a 16 kHz mono float32 waveform.""" + req = urllib.request.Request(url, headers={"User-Agent": "TapScribe-recut/1.0"}) + raw = urllib.request.urlopen(req, timeout=60).read() # noqa: S310 — fixed https Wikimedia URL + data, sr = sf.read(io.BytesIO(raw), dtype="float32", always_2d=True) + mono = data.mean(axis=1) + if sr != SAMPLE_RATE: + g = gcd(int(sr), SAMPLE_RATE) + mono = resample_poly(mono, SAMPLE_RATE // g, int(sr) // g) + return mono.astype(np.float32) + + +def words_with_ts(audio: np.ndarray) -> list[tuple[str, float, float]]: + """Transcribe `audio` (16 kHz mono float32) into (word, start, end).""" + from faster_whisper import WhisperModel + + model = WhisperModel("base.en", device="cpu", compute_type="int8") + segments, _ = model.transcribe(audio, language="en", word_timestamps=True) + out: list[tuple[str, float, float]] = [] + for seg in segments: + for w in seg.words or (): + out.append((w.word.strip().lower().strip(".,!?;:'\""), w.start, w.end)) + return out + + +def main() -> None: + print(f"downloading {SRC}") + audio = load_16k_mono(SRC) + print(f"source: {len(audio) / SAMPLE_RATE:.1f}s @ {SAMPLE_RATE} Hz mono") + + words = words_with_ts(audio) + norm = [w for w, _, _ in words] + print("full transcript:\n " + " ".join(norm)) + if "small" not in norm or "mankind" not in norm: + sys.exit("!! 'small'/'mankind' anchors not found — inspect the transcript above and trim by hand") + + s_idx = norm.index("small") + m_idx = len(norm) - 1 - norm[::-1].index("mankind") + # Back up two words from "small" to include the leading "that's one". + start = max(0.0, words[max(0, s_idx - 2)][1] - PAD_S) + end = words[m_idx][2] + PAD_S + print(f"trim window: {start:.2f}..{end:.2f}s ({end - start:.2f}s)") + + clip = audio[int(start * SAMPLE_RATE) : int(end * SAMPLE_RATE)] + peak = max(float(np.abs(clip).max()), 1e-9) + int16 = (clip / peak * 0.9 * 32767).astype(np.int16) + OUT.parent.mkdir(parents=True, exist_ok=True) + with wave.open(str(OUT), "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(SAMPLE_RATE) + w.writeframes(int16.tobytes()) + print(f"wrote {OUT} ({len(int16) / SAMPLE_RATE:.2f}s)") + + check = words_with_ts(int16.astype(np.float32) / 32767.0) + print("re-transcribed cut (should match the reference):\n " + " ".join(w for w, _, _ in check)) + + +if __name__ == "__main__": + main()