feat: add --normalize compressor + limiter for input audio

Adds a feedforward dynamic range compressor with a brick-wall limiter
applied in the audio callback. Quiet speech gets +12 dB makeup gain,
loud bursts are attenuated 4:1 above -20 dBFS, and the output is
hard-limited at -1 dBFS so nothing clips. Enabled via --normalize/-n
on `cohere on` and `cohere transcribe`.

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

src/cohere_transcribe/cli/cli.py

@@ -12,10 +12,21 @@ app = typer.Typer(help="Cohere live transcription — speaks into your keyboard.
 console = Console()
 
 
+def _parse_device(value: str | None):
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except ValueError:
+        return value
+
+
 @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"),
+    device: str = typer.Option(None, "--device", "-d", help="Input device index or name substring (see `cohere devices`)"),
+    normalize: bool = typer.Option(False, "--normalize", "-n", help="Enable compressor + limiter to even out loudness"),
     foreground: bool = typer.Option(False, "--fg", help="Run in foreground (don't daemonize)"),
 ):
     """Start transcribing and typing into your focused window."""
@@ -26,7 +37,7 @@ def on(
     if foreground:
         from ..daemon import run_daemon
         console.print("[green]Starting cohere (foreground)...[/green]")
-        run_daemon(language, pause=pause)
+        run_daemon(language, pause=pause, device=_parse_device(device), normalize=normalize)
         return
 
     console.print("[green]Starting cohere daemon...[/green]")
@@ -34,6 +45,10 @@ def on(
     cmd = [sys.executable, "-m", "cohere_transcribe.daemon_main", "--lang", language]
     if pause != 0.3:
         cmd += ["--pause", str(pause)]
+    if device is not None:
+        cmd += ["--device", device]
+    if normalize:
+        cmd += ["--normalize"]
     subprocess.Popen(
         cmd,
         start_new_session=True,
@@ -90,20 +105,24 @@ def transcribe(
     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"),
+    device: str = typer.Option(None, "--device", "-d", help="Input device index or name substring (see `cohere devices`)"),
+    normalize: bool = typer.Option(False, "--normalize", "-n", help="Enable compressor + limiter to even out loudness"),
 ):
     """One-shot transcription (file, mic, or stream to terminal)."""
     from ..model import load_model, transcribe_audio
     from ..vad import pause_seconds_to_frames
 
+    dev = _parse_device(device)
+
     if stream:
         from ..stream import stream_transcribe
         processor, model = load_model()
-        stream_transcribe(processor, model, language, silence_frames=pause_seconds_to_frames(pause))
+        stream_transcribe(processor, model, language, silence_frames=pause_seconds_to_frames(pause), device=dev, normalize=normalize)
     elif mic is not None:
         from ..model import record_audio
         processor, model = load_model()
         try:
-            audio = record_audio(mic)
+            audio = record_audio(mic, device=dev, normalize=normalize)
             console.print("Transcribing...")
             text = transcribe_audio(processor, model, audio, language)
             console.print(f"\n{text}\n")
@@ -122,5 +141,26 @@ def transcribe(
         raise typer.Exit(1)
 
 
+@app.command()
+def devices():
+    """List available audio input devices."""
+    import sounddevice as sd
+
+    default_in = sd.default.device[0]
+    for idx, dev in enumerate(sd.query_devices()):
+        if dev["max_input_channels"] <= 0:
+            continue
+        marker = "[green]*[/green]" if idx == default_in else " "
+        hostapi = sd.query_hostapis(dev["hostapi"])["name"]
+        console.print(
+            f"{marker} [bold]{idx:>2}[/bold]  {dev['name']}  "
+            f"[dim]({dev['max_input_channels']}ch, {int(dev['default_samplerate'])}Hz, {hostapi})[/dim]"
+        )
+    console.print(
+        "\n[dim]Tip: indices can shift between runs on PipeWire. "
+        "Prefer [bold]-d pipewire[/bold] (uses PipeWire's default source) or pass a name substring like [bold]-d Sipeed[/bold].[/dim]"
+    )
+
+
 def main():
     app()

src/cohere_transcribe/compressor.py

@@ -0,0 +1,52 @@
+import math
+
+import numpy as np
+
+from .model import SAMPLE_RATE
+
+
+class Compressor:
+    """Feedforward dynamic range compressor + brick-wall limiter for speech.
+
+    Per sample: track an attack/release-smoothed envelope of |x|, compute gain
+    reduction above the threshold, apply makeup gain, then hard-limit to the ceiling.
+    """
+
+    def __init__(
+        self,
+        threshold_db: float = -20.0,
+        ratio: float = 4.0,
+        attack_ms: float = 5.0,
+        release_ms: float = 80.0,
+        makeup_db: float = 12.0,
+        ceiling: float = 10 ** (-1.0 / 20),  # -1 dBFS
+        sample_rate: int = SAMPLE_RATE,
+    ):
+        self.threshold_db = threshold_db
+        self.ratio = ratio
+        self.makeup_gain = 10 ** (makeup_db / 20)
+        self.ceiling = ceiling
+        self.knee = 1.0 - 1.0 / ratio
+        self.a_att = math.exp(-1.0 / (attack_ms * 0.001 * sample_rate))
+        self.a_rel = math.exp(-1.0 / (release_ms * 0.001 * sample_rate))
+        self.envelope = 0.0
+
+    def process(self, x: np.ndarray) -> np.ndarray:
+        abs_x = np.abs(x)
+        env_out = np.empty_like(x)
+        e = self.envelope
+        a_att = self.a_att
+        a_rel = self.a_rel
+        for i in range(len(x)):
+            target = abs_x[i]
+            coef = a_att if target > e else a_rel
+            e = coef * e + (1.0 - coef) * target
+            env_out[i] = e
+        self.envelope = e
+
+        env_db = 20.0 * np.log10(np.maximum(env_out, 1e-10))
+        over = env_db - self.threshold_db
+        gr_db = np.where(over > 0, over * self.knee, 0.0)
+        gain = 10 ** (-gr_db / 20.0) * self.makeup_gain
+        y = x * gain
+        return np.clip(y, -self.ceiling, self.ceiling).astype(np.float32)

src/cohere_transcribe/daemon.py

@@ -10,8 +10,18 @@ import sounddevice as sd
 
 from .backend import WtypeBackend
 from .commands import process_and_output
+from .compressor import Compressor
 from .model import SAMPLE_RATE, load_model, transcribe_audio
-from .vad import DEFAULT_SILENCE_FRAMES, FRAME_SIZE, VADStateMachine, calibrate_silence, pause_seconds_to_frames
+from .vad import (
+    DEFAULT_SILENCE_FRAMES,
+    FRAME_SIZE,
+    VADStateMachine,
+    calibrate_silence,
+    describe_input_device,
+    pause_seconds_to_frames,
+    resample_to_target,
+    resolve_input_rate,
+)
 
 STATE_DIR = os.path.expanduser("~/.local/state/cohere")
 STATE_FILE = os.path.join(STATE_DIR, "state.json")
@@ -71,7 +81,7 @@ def stop_daemon() -> bool:
         return False
 
 
-def run_daemon(language: str = "en", pause: float | None = None):
+def run_daemon(language: str = "en", pause: float | None = None, device=None, normalize: bool = False):
     pid = os.getpid()
     _write_state(pid, "starting")
 
@@ -82,7 +92,13 @@ def run_daemon(language: str = "en", pause: float | None = None):
 
     silence_frames = pause_seconds_to_frames(pause) if pause else DEFAULT_SILENCE_FRAMES
     processor, model = load_model()
-    threshold = calibrate_silence()
+    print(f"Using input device: {describe_input_device(device)}")
+    comp = Compressor() if normalize else None
+    if comp:
+        print("  Normalization: compressor+limiter enabled")
+    threshold = calibrate_silence(device=device, compressor=comp)
+    capture_rate = resolve_input_rate(device)
+    capture_blocksize = FRAME_SIZE * capture_rate // SAMPLE_RATE
     vad = VADStateMachine(threshold, silence_frames=silence_frames)
     seg_queue: queue.Queue = queue.Queue()
     stop_event = threading.Event()
@@ -108,13 +124,16 @@ def run_daemon(language: str = "en", pause: float | None = None):
         if stop_event.is_set():
             return
         elapsed = time.monotonic() - start_time
-        result = vad.process_frame(indata[:, 0].copy(), elapsed)
+        frame = resample_to_target(indata[:, 0].copy(), capture_rate)
+        if comp is not None:
+            frame = comp.process(frame)
+        result = vad.process_frame(frame, 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,
+        samplerate=capture_rate, channels=1, dtype="float32",
+        callback=audio_callback, blocksize=capture_blocksize, device=device,
     )
 
     try:

src/cohere_transcribe/daemon_main.py

@@ -2,8 +2,20 @@ import argparse
 
 from .daemon import run_daemon
 
+
+def _parse_device(value):
+    if value is None:
+        return None
+    try:
+        return int(value)
+    except ValueError:
+        return value
+
+
 parser = argparse.ArgumentParser()
 parser.add_argument("--lang", default="en")
 parser.add_argument("--pause", type=float, default=None)
+parser.add_argument("--device", default=None)
+parser.add_argument("--normalize", action="store_true")
 args = parser.parse_args()
-run_daemon(args.lang, pause=args.pause)
+run_daemon(args.lang, pause=args.pause, device=_parse_device(args.device), normalize=args.normalize)

src/cohere_transcribe/model.py

@@ -23,10 +23,16 @@ def transcribe_audio(processor, model, audio, language="en"):
     return " ".join(texts) if isinstance(texts, list) else texts
 
 
-def record_audio(duration):
+def record_audio(duration, device=None, normalize=False):
     import sounddevice as sd
+    from .vad import resolve_input_rate, resample_to_target
 
     print(f"Recording for {duration} seconds...")
-    audio = sd.rec(int(duration * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+    rate = resolve_input_rate(device)
+    audio = sd.rec(int(duration * rate), samplerate=rate, channels=1, dtype="float32", device=device)
     sd.wait()
-    return audio.flatten()
+    audio = resample_to_target(audio.flatten(), rate)
+    if normalize:
+        from .compressor import Compressor
+        audio = Compressor().process(audio)
+    return audio

src/cohere_transcribe/stream.py

@@ -6,12 +6,19 @@ import time
 import numpy as np
 import sounddevice as sd
 
+from .compressor import Compressor
 from .model import SAMPLE_RATE, transcribe_audio
-from .vad import DEFAULT_SILENCE_FRAMES, FRAME_SIZE, VADStateMachine, calibrate_silence
+from .vad import DEFAULT_SILENCE_FRAMES, FRAME_SIZE, VADStateMachine, calibrate_silence, describe_input_device, resample_to_target, resolve_input_rate
 
 
-def stream_transcribe(processor, model, language, silence_frames=DEFAULT_SILENCE_FRAMES):
-    threshold = calibrate_silence()
+def stream_transcribe(processor, model, language, silence_frames=DEFAULT_SILENCE_FRAMES, device=None, normalize=False):
+    print(f"Using input device: {describe_input_device(device)}")
+    comp = Compressor() if normalize else None
+    if comp:
+        print("  Normalization: compressor+limiter enabled")
+    threshold = calibrate_silence(device=device, compressor=comp)
+    capture_rate = resolve_input_rate(device)
+    capture_blocksize = FRAME_SIZE * capture_rate // SAMPLE_RATE
     vad = VADStateMachine(threshold, silence_frames=silence_frames)
     seg_queue = queue.Queue()
     stop_event = threading.Event()
@@ -36,14 +43,17 @@ def stream_transcribe(processor, model, language, silence_frames=DEFAULT_SILENCE
         if stop_event.is_set():
             return
         elapsed = time.monotonic() - start_time
-        result = vad.process_frame(indata[:, 0].copy(), elapsed)
+        frame = resample_to_target(indata[:, 0].copy(), capture_rate)
+        if comp is not None:
+            frame = comp.process(frame)
+        result = vad.process_frame(frame, 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,
+        samplerate=capture_rate, channels=1, dtype="float32",
+        callback=audio_callback, blocksize=capture_blocksize, device=device,
     )
 
     try:

src/cohere_transcribe/vad.py

@@ -1,4 +1,5 @@
 import collections
+from math import gcd
 
 import numpy as np
 import sounddevice as sd
@@ -10,17 +11,58 @@ 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
-MIN_SPEECH_SECONDS = 0.3  # discard segments shorter than this (mic bumps, clicks)
+MIN_LOUD_FRAMES = 8       # need at least ~400ms of loud frames to count as speech
 
 
 def pause_seconds_to_frames(seconds: float) -> int:
     return max(1, round(seconds / (FRAME_SIZE / SAMPLE_RATE)))
 
 
-def calibrate_silence(duration=0.5):
+def _query_input(device):
+    resolved = device if device is not None else sd.default.device[0]
+    info = sd.query_devices(resolved)
+    if info["max_input_channels"] < 1:
+        raise ValueError(
+            f"Device {device!r} ({info['name']}) is not an input device. "
+            f"Run `cohere devices` to see current input indices — they can shift between runs on PipeWire."
+        )
+    return info
+
+
+def describe_input_device(device) -> str:
+    info = _query_input(device)
+    return info["name"]
+
+
+def resolve_input_rate(device) -> int:
+    """Pick a samplerate the device will accept. Prefer SAMPLE_RATE; if the device
+    refuses (e.g. raw ALSA hw: that doesn't resample), fall back to its native rate."""
+    info = _query_input(device)
+    try:
+        sd.check_input_settings(device=device, samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+        return SAMPLE_RATE
+    except sd.PortAudioError:
+        rate = int(info["default_samplerate"])
+        print(f"  Device doesn't support {SAMPLE_RATE}Hz; capturing at {rate}Hz and resampling.")
+        return rate
+
+
+def resample_to_target(audio: np.ndarray, src_rate: int) -> np.ndarray:
+    if src_rate == SAMPLE_RATE:
+        return audio
+    from scipy.signal import resample_poly
+    g = gcd(SAMPLE_RATE, src_rate)
+    return resample_poly(audio, SAMPLE_RATE // g, src_rate // g).astype(np.float32)
+
+
+def calibrate_silence(duration=0.5, device=None, compressor=None):
     print("Calibrating silence threshold...")
-    audio = sd.rec(int(duration * SAMPLE_RATE), samplerate=SAMPLE_RATE, channels=1, dtype="float32")
+    rate = resolve_input_rate(device)
+    audio = sd.rec(int(duration * rate), samplerate=rate, channels=1, dtype="float32", device=device)
     sd.wait()
+    audio = resample_to_target(audio.flatten(), rate)
+    if compressor is not None:
+        audio = compressor.process(audio)
     rms = np.sqrt(np.mean(audio ** 2))
     threshold = max(rms * 3, 0.01)
     print(f"  Ambient RMS: {rms:.4f}, threshold: {threshold:.4f}")
@@ -34,6 +76,7 @@ class VADStateMachine:
         self.speaking = False
         self.speech_frames = 0
         self.silence_frames = 0
+        self.loud_frames = 0
         self.pre_roll = collections.deque(maxlen=PRE_ROLL_FRAMES)
         self.segment = []
         self.segment_start_time = 0.0
@@ -63,18 +106,19 @@ class VADStateMachine:
 
         if is_loud:
             self.silence_frames = 0
+            self.loud_frames += 1
         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:
-            speech_duration = segment_duration - self.silence_frames * FRAME_SIZE / SAMPLE_RATE
             result = None
-            if speech_duration >= MIN_SPEECH_SECONDS:
+            if self.loud_frames >= MIN_LOUD_FRAMES:
                 result = (self.segment_start_time, np.concatenate(self.segment))
             self.speaking = False
             self.speech_frames = 0
             self.silence_frames = 0
+            self.loud_frames = 0
             self.segment = []
             self.pre_roll = collections.deque(maxlen=PRE_ROLL_FRAMES)
             return result