diff dictation/main.py @ 275:78699f810817

Add Qwen3-VL and WebRTC dictation services Add Bazel targets for the CUDA-backed Qwen3-VL server and a local WebRTC faster-whisper dictation service. Co-authored-by: Copilot <[email protected]> Copilot-Session: e3d8cb06-6c95-4ae0-9757-651d3796ab00
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 10:58:47 -0700
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/dictation/main.py	Mon Aug 17 10:58:47 2026 -0700
@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+from pathlib import Path
+import sys
+
+from dictation.config import (
+    DictationConfig,
+    MODEL_REPOSITORY,
+    MODEL_REVISION,
+)
+
+
+def download_model(config: DictationConfig) -> None:
+    from huggingface_hub import snapshot_download
+
+    config.model_dir.mkdir(parents=True, exist_ok=True)
+    snapshot_download(
+        repo_id=MODEL_REPOSITORY,
+        revision=MODEL_REVISION,
+        local_dir=config.model_dir,
+        allow_patterns=[
+            "config.json",
+            "model.bin",
+            "tokenizer.json",
+            "vocabulary.txt",
+        ],
+    )
+    marker = config.model_dir / ".model-revision"
+    marker.write_text(f"{MODEL_REPOSITORY}@{MODEL_REVISION}\n", encoding="ascii")
+    print(f"Model ready: {config.model_dir}")
+
+
+def preflight(config: DictationConfig) -> None:
+    import ctranslate2
+
+    compute_types = sorted(ctranslate2.get_supported_compute_types("cuda"))
+    if config.compute_type not in compute_types:
+        raise RuntimeError(
+            f"{config.compute_type} is unsupported on CUDA; "
+            f"available: {', '.join(compute_types)}"
+        )
+    result = {
+        "cudaDeviceCount": ctranslate2.get_cuda_device_count(),
+        "computeTypes": compute_types,
+        "selectedComputeType": config.compute_type,
+        "modelDir": str(config.model_dir),
+        "modelPresent": (config.model_dir / "model.bin").is_file(),
+    }
+    if result["cudaDeviceCount"] < 1:
+        raise RuntimeError("CTranslate2 did not detect a CUDA device")
+    print(json.dumps(result, indent=2))
+
+
+async def transcribe_file(config: DictationConfig, path: Path) -> None:
+    from dictation.transcriber import FasterWhisperTranscriber
+
+    transcriber = FasterWhisperTranscriber(
+        config.model_dir,
+        config.compute_type,
+    )
+    try:
+        from av import open as av_open
+        from av.audio.resampler import AudioResampler
+        import numpy as np
+
+        samples = []
+        resampler = AudioResampler(format="s16", layout="mono", rate=16000)
+        with av_open(str(path)) as container:
+            for frame in container.decode(audio=0):
+                for converted in resampler.resample(frame):
+                    samples.append(converted.to_ndarray().reshape(-1))
+        if not samples:
+            raise RuntimeError("Audio file contains no decodable samples")
+        audio = np.concatenate(samples).astype(np.float32) / 32768.0
+        result = await transcriber.transcribe(audio, final=True)
+        print(
+            json.dumps(
+                {
+                    "text": result.text,
+                    "language": result.language,
+                    "probability": result.probability,
+                },
+                ensure_ascii=False,
+            )
+        )
+    finally:
+        transcriber.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "command",
+        choices=["server", "preflight", "download_model", "transcribe"],
+    )
+    parser.add_argument("path", nargs="?")
+    args = parser.parse_args()
+    try:
+        config = DictationConfig.from_environment()
+        if args.command == "download_model":
+            download_model(config)
+        elif args.command == "preflight":
+            preflight(config)
+        elif args.command == "transcribe":
+            if not args.path:
+                parser.error("transcribe requires an audio file path")
+            asyncio.run(transcribe_file(config, Path(args.path)))
+        else:
+            import uvicorn
+            from dictation.server import create_app
+
+            uvicorn.run(
+                create_app(config),
+                host=config.host,
+                port=config.port,
+                access_log=False,
+            )
+    except Exception as error:
+        print(f"dictation: {error}", file=sys.stderr)
+        raise SystemExit(1) from error
+
+
+if __name__ == "__main__":
+    main()