feat: add voice command processing and input backend interface

Introduce InputBackend protocol with WtypeBackend and PrintBackend,
and a command processor that translates spoken commands (enter, new line,
question mark, comma, etc.) into key presses and punctuation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.gitignore

@@ -8,3 +8,7 @@ wheels/
 
 # Virtual environments
 .venv
+
+# Nix
+.direnv/
+result

flake.lock

@@ -0,0 +1,27 @@
+{
+  "nodes": {
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1779786838,
+        "narHash": "sha256-0geHoGiR5f8qiXg+gO4rSF6Up6Var+kKqiOv9AO/uUc=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "f44f7788c891fbe5542177df78374f8cdab10e8f",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixpkgs-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "nixpkgs": "nixpkgs"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -19,13 +19,13 @@
           python314
           portaudio
           cudaPackages.cudatoolkit
+          wtype
         ];
 
-        env = {
+        LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath [
-          LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath [
+          pkgs.portaudio
-            pkgs.cudaPackages.cudatoolkit
+          pkgs.cudaPackages.cudatoolkit
-          ];
+        ];
-        };
       };
     };
 }

main.py

@@ -1,6 +0,0 @@
-def main():
-    print("Hello from cohere!")
-
-
-if __name__ == "__main__":
-    main()

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
-name = "cohere"
+name = "cohere-transcribe"
 version = "0.1.0"
-description = "Add your description here"
+description = "Live speech transcription using Cohere ASR"
 readme = "README.md"
 requires-python = ">=3.14"
 dependencies = [
@@ -14,4 +14,16 @@ dependencies = [
     "soundfile>=0.13.1",
     "torch>=2.12.0",
     "transformers>=5.9.0",
+    "typer[all]>=0.15.0",
 ]
+
+[project.scripts]
+cohere = "cohere_transcribe.cli:main"
+cohere-transcribe = "cohere_transcribe.cli:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/cohere_transcribe"]

shell.nix

@@ -1,15 +0,0 @@
-{ pkgs ? import <nixpkgs> { config.allowUnfree = true; } }:
-
-pkgs.mkShell {
-  buildInputs = with pkgs; [
-    portaudio
-    cudaPackages.cudatoolkit
-    uv
-    python314
-  ];
-
-  shellHook = ''
-    export LD_LIBRARY_PATH="${pkgs.cudaPackages.cudatoolkit}/lib:$LD_LIBRARY_PATH"
-    echo "Dev shell ready - microphone input enabled"
-  '';
-}

src/cohere_transcribe/backend.py

@@ -0,0 +1,35 @@
+import subprocess
+import sys
+from typing import Protocol
+
+
+class InputBackend(Protocol):
+    def type_text(self, text: str) -> None: ...
+    def send_key(self, key: str) -> None: ...
+
+
+class WtypeBackend:
+    def type_text(self, text: str) -> None:
+        try:
+            subprocess.run(["wtype", "--", text], check=True, timeout=10)
+        except FileNotFoundError:
+            print("wtype not found — install it for keyboard injection", file=sys.stderr)
+        except subprocess.SubprocessError as e:
+            print(f"wtype error: {e}", file=sys.stderr)
+
+    def send_key(self, key: str) -> None:
+        try:
+            subprocess.run(["wtype", "-k", key], check=True, timeout=10)
+        except FileNotFoundError:
+            print("wtype not found — install it for keyboard injection", file=sys.stderr)
+        except subprocess.SubprocessError as e:
+            print(f"wtype error: {e}", file=sys.stderr)
+
+
+class PrintBackend:
+    def type_text(self, text: str) -> None:
+        print(text, end="", flush=True)
+
+    def send_key(self, key: str) -> None:
+        key_map = {"Return": "\n", "Tab": "\t", "BackSpace": "\b"}
+        print(key_map.get(key, f"[{key}]"), end="", flush=True)

src/cohere_transcribe/cli/__init__.py

@@ -0,0 +1,3 @@
+from .cli import main
+
+__all__ = ["main"]

src/cohere_transcribe/cli/cli.py

@@ -0,0 +1,126 @@
+import os
+import subprocess
+import sys
+import time
+
+import typer
+from rich.console import Console
+
+from ..daemon import STATE_FILE, is_running, read_state, stop_daemon
+
+app = typer.Typer(help="Cohere live transcription — speaks into your keyboard.")
+console = Console()
+
+
+@app.command()
+def on(
+    language: str = typer.Option("en", "--lang", "-l", help="Language code"),
+    pause: float = typer.Option(0.3, "--pause", "-p", help="Seconds of silence before sending text"),
+    foreground: bool = typer.Option(False, "--fg", help="Run in foreground (don't daemonize)"),
+):
+    """Start transcribing and typing into your focused window."""
+    if is_running():
+        console.print("[yellow]Already running.[/yellow]")
+        raise typer.Exit(1)
+
+    if foreground:
+        from ..daemon import run_daemon
+        console.print("[green]Starting cohere (foreground)...[/green]")
+        run_daemon(language, pause=pause)
+        return
+
+    console.print("[green]Starting cohere daemon...[/green]")
+    os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
+    cmd = [sys.executable, "-m", "cohere_transcribe.daemon_main", "--lang", language]
+    if pause != 0.3:
+        cmd += ["--pause", str(pause)]
+    subprocess.Popen(
+        cmd,
+        start_new_session=True,
+        stdin=subprocess.DEVNULL,
+        stdout=open(os.path.join(os.path.dirname(STATE_FILE), "daemon.log"), "a"),
+        stderr=subprocess.STDOUT,
+    )
+
+    for _ in range(50):
+        time.sleep(0.1)
+        if is_running():
+            break
+
+    if is_running():
+        console.print("[green]Cohere is on — speak and it types.[/green]")
+    else:
+        console.print("[red]Failed to start daemon. Check ~/.local/state/cohere/daemon.log[/red]")
+        raise typer.Exit(1)
+
+
+@app.command()
+def off():
+    """Stop transcribing."""
+    if not is_running():
+        console.print("[yellow]Not running.[/yellow]")
+        raise typer.Exit(0)
+
+    if stop_daemon():
+        console.print("[red]Cohere is off.[/red]")
+    else:
+        console.print("[red]Failed to stop daemon.[/red]")
+        raise typer.Exit(1)
+
+
+@app.command()
+def status():
+    """Show whether cohere is running."""
+    state = read_state()
+    running = is_running()
+
+    if running:
+        started = state.get("started_at", 0)
+        elapsed = time.time() - started
+        minutes = int(elapsed) // 60
+        console.print(f"[green]ON[/green] — running for {minutes}m")
+    else:
+        console.print("[dim]OFF[/dim]")
+
+
+@app.command()
+def transcribe(
+    audio_file: str = typer.Argument(None, help="Audio file to transcribe"),
+    mic: int = typer.Option(None, "--mic", "-m", help="Record from mic for N seconds"),
+    stream: bool = typer.Option(False, "--stream", "-s", help="Live streaming mode (prints to terminal)"),
+    language: str = typer.Option("en", "--lang", "-l", help="Language code"),
+    pause: float = typer.Option(0.3, "--pause", "-p", help="Seconds of silence before sending text"),
+):
+    """One-shot transcription (file, mic, or stream to terminal)."""
+    from ..model import load_model, transcribe_audio
+    from ..vad import pause_seconds_to_frames
+
+    if stream:
+        from ..stream import stream_transcribe
+        processor, model = load_model()
+        stream_transcribe(processor, model, language, silence_frames=pause_seconds_to_frames(pause))
+    elif mic is not None:
+        from ..model import record_audio
+        processor, model = load_model()
+        try:
+            audio = record_audio(mic)
+            console.print("Transcribing...")
+            text = transcribe_audio(processor, model, audio, language)
+            console.print(f"\n{text}\n")
+        except OSError as e:
+            console.print(f"[red]Microphone error: {e}[/red]")
+            raise typer.Exit(1)
+    elif audio_file:
+        from transformers.audio_utils import load_audio as load_audio_file
+        from ..model import SAMPLE_RATE
+        processor, model = load_model()
+        audio = load_audio_file(audio_file, sampling_rate=SAMPLE_RATE)
+        text = transcribe_audio(processor, model, audio, language)
+        console.print(f"\n{text}\n")
+    else:
+        console.print("[yellow]Provide an audio file, --mic, or --stream[/yellow]")
+        raise typer.Exit(1)
+
+
+def main():
+    app()

src/cohere_transcribe/commands.py

@@ -0,0 +1,55 @@
+import re
+
+from .backend import InputBackend
+
+KEY_COMMANDS: dict[str, list[str]] = {
+    "new line": ["Return"],
+    "newline": ["Return"],
+    "enter": ["Return"],
+    "press enter": ["Return"],
+    "new paragraph": ["Return", "Return"],
+    "tab": ["Tab"],
+    "backspace": ["BackSpace"],
+}
+
+PUNCTUATION: dict[str, str] = {
+    "question mark": "?",
+    "exclamation mark": "!",
+    "exclamation point": "!",
+    "period": ".",
+    "full stop": ".",
+    "comma": ",",
+    "colon": ":",
+    "semicolon": ";",
+    "open quote": '"',
+    "close quote": '"',
+    "open paren": "(",
+    "close paren": ")",
+}
+
+
+def _build_pattern(commands: dict) -> re.Pattern:
+    sorted_keys = sorted(commands.keys(), key=len, reverse=True)
+    escaped = [re.escape(k) for k in sorted_keys]
+    return re.compile(r"\b(" + "|".join(escaped) + r")\b", re.IGNORECASE)
+
+
+_KEY_PATTERN = _build_pattern(KEY_COMMANDS)
+_PUNCT_PATTERN = _build_pattern(PUNCTUATION)
+
+
+def process_and_output(text: str, backend: InputBackend) -> None:
+    text = _PUNCT_PATTERN.sub(lambda m: PUNCTUATION[m.group(1).lower()], text)
+    text = re.sub(r"\s+([?.!,;:)\"])", r"\1", text)
+
+    parts = _KEY_PATTERN.split(text)
+
+    for part in parts:
+        cmd = part.strip().lower()
+        if cmd in KEY_COMMANDS:
+            for key in KEY_COMMANDS[cmd]:
+                backend.send_key(key)
+        else:
+            cleaned = part.strip()
+            if cleaned:
+                backend.type_text(cleaned + " ")

src/cohere_transcribe/daemon.py

@@ -0,0 +1,133 @@
+import json
+import os
+import signal
+import queue
+import threading
+import time
+
+import numpy as np
+import sounddevice as sd
+
+from .backend import WtypeBackend
+from .commands import process_and_output
+from .model import SAMPLE_RATE, load_model, transcribe_audio
+from .vad import DEFAULT_SILENCE_FRAMES, FRAME_SIZE, VADStateMachine, calibrate_silence, pause_seconds_to_frames
+
+STATE_DIR = os.path.expanduser("~/.local/state/cohere")
+STATE_FILE = os.path.join(STATE_DIR, "state.json")
+LOG_FILE = os.path.join(STATE_DIR, "daemon.log")
+
+
+def _write_state(pid: int, status: str):
+    os.makedirs(STATE_DIR, exist_ok=True)
+    with open(STATE_FILE, "w") as f:
+        json.dump({"pid": pid, "status": status, "started_at": time.time()}, f)
+
+
+_backend = WtypeBackend()
+
+
+def read_state() -> dict | None:
+    try:
+        with open(STATE_FILE) as f:
+            return json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return None
+
+
+def is_running() -> bool:
+    state = read_state()
+    if state is None:
+        return False
+    pid = state.get("pid")
+    if pid is None:
+        return False
+    try:
+        os.kill(pid, 0)
+        return True
+    except OSError:
+        return False
+
+
+def stop_daemon() -> bool:
+    state = read_state()
+    if state is None:
+        return False
+    pid = state.get("pid")
+    if pid is None:
+        return False
+    try:
+        os.kill(pid, signal.SIGTERM)
+        for _ in range(20):
+            time.sleep(0.1)
+            try:
+                os.kill(pid, 0)
+            except OSError:
+                break
+        _write_state(pid, "stopped")
+        return True
+    except OSError:
+        _write_state(pid, "stopped")
+        return False
+
+
+def run_daemon(language: str = "en", pause: float | None = None):
+    pid = os.getpid()
+    _write_state(pid, "starting")
+
+    def handle_sigterm(signum, frame):
+        raise KeyboardInterrupt
+
+    signal.signal(signal.SIGTERM, handle_sigterm)
+
+    silence_frames = pause_seconds_to_frames(pause) if pause else DEFAULT_SILENCE_FRAMES
+    processor, model = load_model()
+    threshold = calibrate_silence()
+    vad = VADStateMachine(threshold, silence_frames=silence_frames)
+    seg_queue: queue.Queue = queue.Queue()
+    stop_event = threading.Event()
+    start_time = time.monotonic()
+
+    _write_state(pid, "running")
+
+    def transcription_worker():
+        while not stop_event.is_set() or not seg_queue.empty():
+            try:
+                _seg_start, audio = seg_queue.get(timeout=0.5)
+            except queue.Empty:
+                continue
+            text = transcribe_audio(processor, model, audio, language)
+            text = text.strip()
+            if text:
+                process_and_output(text, _backend)
+
+    worker = threading.Thread(target=transcription_worker, daemon=True)
+    worker.start()
+
+    def audio_callback(indata, frames, time_info, status):
+        if stop_event.is_set():
+            return
+        elapsed = time.monotonic() - start_time
+        result = vad.process_frame(indata[:, 0].copy(), elapsed)
+        if result is not None:
+            seg_queue.put(result)
+
+    stream = sd.InputStream(
+        samplerate=SAMPLE_RATE, channels=1, dtype="float32",
+        callback=audio_callback, blocksize=FRAME_SIZE,
+    )
+
+    try:
+        with stream:
+            while True:
+                time.sleep(0.1)
+    except KeyboardInterrupt:
+        pass
+
+    stop_event.set()
+
+    if vad.speaking and vad.segment:
+        seg_queue.put((vad.segment_start_time, np.concatenate(vad.segment)))
+
+    worker.join(timeout=30)
+    _write_state(pid, "stopped")

src/cohere_transcribe/daemon_main.py

@@ -0,0 +1,9 @@
+import argparse
+
+from .daemon import run_daemon
+
+parser = argparse.ArgumentParser()
+parser.add_argument("--lang", default="en")
+parser.add_argument("--pause", type=float, default=None)
+args = parser.parse_args()
+run_daemon(args.lang, pause=args.pause)

src/cohere_transcribe/model.py

@@ -0,0 +1,32 @@
+import numpy as np
+from transformers import AutoProcessor, CohereAsrForConditionalGeneration
+from transformers.audio_utils import load_audio
+
+MODEL_ID = "CohereLabs/cohere-transcribe-03-2026"
+SAMPLE_RATE = 16000
+
+
+def load_model():
+    print("Loading model...")
+    processor = AutoProcessor.from_pretrained(MODEL_ID)
+    model = CohereAsrForConditionalGeneration.from_pretrained(
+        MODEL_ID, device_map="auto"
+    )
+    return processor, model
+
+
+def transcribe_audio(processor, model, audio, language="en"):
+    inputs = processor(audio, sampling_rate=SAMPLE_RATE, return_tensors="pt", language=language)
+    inputs.to(model.device, dtype=model.dtype)
+    outputs = model.generate(**inputs, max_new_tokens=256)
+    texts = processor.decode(outputs, skip_special_tokens=True)
+    return " ".join(texts) if isinstance(texts, list) else texts
+
+
+def record_audio(duration):
+    import sounddevice as sd
+
+    print(f"Recording for {duration} seconds...")
+    audio = sd.rec(int(duration * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+    sd.wait()
+    return audio.flatten()

src/cohere_transcribe/stream.py

@@ -0,0 +1,64 @@
+import sys
+import queue
+import threading
+import time
+
+import numpy as np
+import sounddevice as sd
+
+from .model import SAMPLE_RATE, transcribe_audio
+from .vad import DEFAULT_SILENCE_FRAMES, FRAME_SIZE, VADStateMachine, calibrate_silence
+
+
+def stream_transcribe(processor, model, language, silence_frames=DEFAULT_SILENCE_FRAMES):
+    threshold = calibrate_silence()
+    vad = VADStateMachine(threshold, silence_frames=silence_frames)
+    seg_queue = queue.Queue()
+    stop_event = threading.Event()
+    start_time = time.monotonic()
+
+    def transcription_worker():
+        while not stop_event.is_set() or not seg_queue.empty():
+            try:
+                seg_start, audio = seg_queue.get(timeout=0.5)
+            except queue.Empty:
+                continue
+            minutes = int(seg_start) // 60
+            seconds = int(seg_start) % 60
+            text = transcribe_audio(processor, model, audio, language)
+            if text.strip():
+                print(f"[{minutes:02d}:{seconds:02d}] {text.strip()}")
+
+    worker = threading.Thread(target=transcription_worker, daemon=True)
+    worker.start()
+
+    def audio_callback(indata, frames, time_info, status):
+        if stop_event.is_set():
+            return
+        elapsed = time.monotonic() - start_time
+        result = vad.process_frame(indata[:, 0].copy(), elapsed)
+        if result is not None:
+            seg_queue.put(result)
+
+    print("Listening... (Ctrl+C to stop)")
+    stream = sd.InputStream(
+        samplerate=SAMPLE_RATE, channels=1, dtype="float32",
+        callback=audio_callback, blocksize=FRAME_SIZE,
+    )
+
+    try:
+        with stream:
+            while True:
+                time.sleep(0.1)
+    except KeyboardInterrupt:
+        pass
+
+    stop_event.set()
+
+    if vad.speaking and vad.segment:
+        seg_queue.put((vad.segment_start_time, np.concatenate(vad.segment)))
+
+    worker.join(timeout=30)
+    if worker.is_alive():
+        print("Warning: transcription worker did not finish in time.", file=sys.stderr)
+    print("\nDone.")

src/cohere_transcribe/vad.py

@@ -0,0 +1,78 @@
+import collections
+
+import numpy as np
+import sounddevice as sd
+
+from .model import SAMPLE_RATE
+
+FRAME_SIZE = 800          # 50ms at 16kHz
+PRE_ROLL_FRAMES = 6       # ~0.3s of audio before speech onset
+DEFAULT_SILENCE_FRAMES = 16  # ~0.8s of silence to end a segment
+SPEECH_ONSET_FRAMES = 3   # ~150ms of speech to trigger
+MAX_SPEECH_SECONDS = 30   # force chunk boundary
+
+
+def pause_seconds_to_frames(seconds: float) -> int:
+    return max(1, round(seconds / (FRAME_SIZE / SAMPLE_RATE)))
+
+
+def calibrate_silence(duration=0.5):
+    print("Calibrating silence threshold...")
+    audio = sd.rec(int(duration * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+    sd.wait()
+    rms = np.sqrt(np.mean(audio ** 2))
+    threshold = max(rms * 3, 0.01)
+    print(f"  Ambient RMS: {rms:.4f}, threshold: {threshold:.4f}")
+    return threshold
+
+
+class VADStateMachine:
+    def __init__(self, threshold, silence_frames=DEFAULT_SILENCE_FRAMES):
+        self.threshold = threshold
+        self.silence_limit = silence_frames
+        self.speaking = False
+        self.speech_frames = 0
+        self.silence_frames = 0
+        self.pre_roll = collections.deque(maxlen=PRE_ROLL_FRAMES)
+        self.segment = []
+        self.segment_start_time = 0.0
+
+    def process_frame(self, frame, elapsed_time):
+        """Process one 50ms frame. Returns a (start_time, audio_array) tuple when a
+        complete speech segment is detected, otherwise None."""
+        rms = np.sqrt(np.mean(frame ** 2))
+        is_loud = rms > self.threshold
+
+        if not self.speaking:
+            self.pre_roll.append(frame)
+
+            if is_loud:
+                self.speech_frames += 1
+                if self.speech_frames >= SPEECH_ONSET_FRAMES:
+                    self.speaking = True
+                    self.silence_frames = 0
+                    self.segment = list(self.pre_roll)
+                    self.segment_start_time = max(0.0, elapsed_time - len(self.pre_roll) * FRAME_SIZE / SAMPLE_RATE)
+                    self.pre_roll = collections.deque(maxlen=PRE_ROLL_FRAMES)
+            else:
+                self.speech_frames = 0
+            return None
+
+        self.segment.append(frame)
+
+        if is_loud:
+            self.silence_frames = 0
+        else:
+            self.silence_frames += 1
+
+        segment_duration = len(self.segment) * FRAME_SIZE / SAMPLE_RATE
+        if self.silence_frames >= self.silence_limit or segment_duration >= MAX_SPEECH_SECONDS:
+            result = (self.segment_start_time, np.concatenate(self.segment))
+            self.speaking = False
+            self.speech_frames = 0
+            self.silence_frames = 0
+            self.segment = []
+            self.pre_roll = collections.deque(maxlen=PRE_ROLL_FRAMES)
+            return result
+
+        return None

tests/test_mic.py

@@ -0,0 +1,88 @@
+"""Quick microphone tests. Run: uv run python test_mic.py"""
+
+import numpy as np
+import sounddevice as sd
+import sys
+import time
+
+SAMPLE_RATE = 16000
+
+
+def test_device_info():
+    """Show which device will be used for recording."""
+    default_input = sd.default.device[0]
+    info = sd.query_devices(default_input)
+    print(f"Default input device [{default_input}]: {info['name']}")
+    print(f"  Max input channels: {info['max_input_channels']}")
+    print(f"  Default sample rate: {info['default_samplerate']}")
+    assert info["max_input_channels"] > 0, "Default device has no input channels!"
+    print("  PASS\n")
+
+
+def test_record_1s():
+    """Record 1 second and check we got non-silent audio."""
+    print("Recording 1 second... (speak or make noise!)")
+    audio = sd.rec(SAMPLE_RATE, samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+    sd.wait()
+    audio = audio.flatten()
+
+    peak = np.max(np.abs(audio))
+    rms = np.sqrt(np.mean(audio ** 2))
+    print(f"  Samples: {len(audio)}")
+    print(f"  Peak amplitude: {peak:.4f}")
+    print(f"  RMS: {rms:.6f}")
+
+    assert len(audio) == SAMPLE_RATE, f"Expected {SAMPLE_RATE} samples, got {len(audio)}"
+    assert peak > 0, "All zeros — mic not capturing anything"
+    if peak < 0.001:
+        print("  WARNING: Very low signal — mic might be muted or too far away")
+    else:
+        print("  Signal level looks good")
+    print("  PASS\n")
+
+
+def test_record_levels():
+    """Record 3 seconds in 1-second chunks, show live levels."""
+    print("Recording 3 seconds — speak during seconds 2-3 for comparison...")
+    for i in range(3):
+        audio = sd.rec(SAMPLE_RATE, samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+        sd.wait()
+        audio = audio.flatten()
+        rms = np.sqrt(np.mean(audio ** 2))
+        peak = np.max(np.abs(audio))
+        bar = "#" * int(min(peak * 200, 50))
+        print(f"  Second {i+1}: peak={peak:.4f} rms={rms:.6f} |{bar}")
+    print("  PASS\n")
+
+
+def test_stream_callback():
+    """Test that InputStream callback fires correctly."""
+    frames_received = []
+
+    def callback(indata, frames, time_info, status):
+        if status:
+            print(f"  Status: {status}")
+        frames_received.append(len(indata))
+
+    print("Testing InputStream callback for 1 second...")
+    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="float32",
+                        callback=callback, blocksize=800):
+        time.sleep(1)
+
+    total_frames = sum(frames_received)
+    expected = SAMPLE_RATE
+    print(f"  Callbacks fired: {len(frames_received)}")
+    print(f"  Total frames: {total_frames} (expected ~{expected})")
+    print(f"  Blocksize per callback: {frames_received[0] if frames_received else 'N/A'}")
+    assert len(frames_received) > 0, "No callbacks received!"
+    assert abs(total_frames - expected) < expected * 0.2, f"Frame count off by >20%"
+    print("  PASS\n")
+
+
+if __name__ == "__main__":
+    print("=== Microphone Tests ===\n")
+    test_device_info()
+    test_record_1s()
+    test_record_levels()
+    test_stream_callback()
+    print("All tests passed!")

transcribe.py

@@ -1,51 +0,0 @@
-import sys
-import numpy as np
-import sounddevice as sd
-from transformers import AutoProcessor, CohereAsrForConditionalGeneration
-from transformers.audio_utils import load_audio
-from huggingface_hub import hf_hub_download
-
-# Load model
-print("Loading model...")
-processor = AutoProcessor.from_pretrained("CohereLabs/cohere-transcribe-03-2026")
-model = CohereAsrForConditionalGeneration.from_pretrained(
-    "CohereLabs/cohere-transcribe-03-2026",
-    device_map="auto"
-)
-
-def transcribe_audio(audio, language="en"):
-    inputs = processor(audio, sampling_rate=16000, return_tensors="pt", language=language)
-    inputs.to(model.device, dtype=model.dtype)
-    
-    outputs = model.generate(**inputs, max_new_tokens=256)
-    text = processor.decode(outputs, skip_special_tokens=True)
-    return text
-
-def record_audio(duration, samplerate=16000):
-    print(f"Recording for {duration} seconds...")
-    audio = sd.rec(int(duration * samplerate), samplerate=samplerate, channels=1, dtype='float32')
-    sd.wait()
-    return audio.flatten()
-
-# Parse arguments
-if len(sys.argv) > 1 and sys.argv[1] == "--mic":
-    duration = int(sys.argv[2]) if len(sys.argv) > 2 else 5
-    try:
-        mic_audio = record_audio(duration)
-        print("Transcribing...")
-        text = transcribe_audio(mic_audio)
-        print(f"\nTranscription:\n{text}\n")
-    except OSError as e:
-        print(f"Microphone error: {e}")
-        print("Hint: Run with nix-shell for PortAudio support")
-else:
-    print("Loading demo audio...")
-    audio_file = hf_hub_download(
-        repo_id="CohereLabs/cohere-transcribe-03-2026",
-        filename="demo/voxpopuli_test_en_demo.wav",
-    )
-    audio = load_audio(audio_file, sampling_rate=16000)
-
-    print("Transcribing...")
-    text = transcribe_audio(audio)
-    print(f"\nTranscription:\n{text}\n")

uv.lock

@@ -190,9 +190,9 @@ wheels = [
 ]
 
 [[package]]
-name = "cohere"
+name = "cohere-transcribe"
 version = "0.1.0"
-source = { virtual = "." }
+source = { editable = "." }
 dependencies = [
     { name = "accelerate" },
     { name = "huggingface-hub" },
@@ -203,6 +203,7 @@ dependencies = [
     { name = "soundfile" },
     { name = "torch" },
     { name = "transformers" },
+    { name = "typer" },
 ]
 
 [package.metadata]
@@ -216,6 +217,7 @@ requires-dist = [
     { name = "soundfile", specifier = ">=0.13.1" },
     { name = "torch", specifier = ">=2.12.0" },
     { name = "transformers", specifier = ">=5.9.0" },
+    { name = "typer", extras = ["all"], specifier = ">=0.15.0" },
 ]
 
 [[package]]