From b06884551ec15b7a9fc9e71ef564b95828a4cc65 Mon Sep 17 00:00:00 2001
From: dgallagher33 <dgallagher33@users.noreply.github.com>
Date: Mon, 10 Nov 2025 22:19:49 -0500
Subject: [PATCH] Add local Whisper voice capture and transcription

---
 impulse_offloader/capture.py                  | 128 ++++++++--
 .../utils/whisper_integration.py              | 228 +++++++++++++++++-
 requirements.txt                              |   3 +
 tests/test_whisper_integration.py             |  91 +++++++
 4 files changed, 422 insertions(+), 28 deletions(-)
 create mode 100644 tests/test_whisper_integration.py

diff --git a/impulse_offloader/capture.py b/impulse_offloader/capture.py
index 6a7b0b1..b9bd275 100644
--- a/impulse_offloader/capture.py
+++ b/impulse_offloader/capture.py
@@ -14,6 +14,7 @@
 
 def _prompt_for_text() -> str:
     """Prompt the user for free-form input until a non-empty value is provided."""
+
     prompt = "What is on your mind? "
     text = input(prompt).strip()
     while not text:
@@ -22,28 +23,56 @@ def _prompt_for_text() -> str:
     return text
 
 
-def capture_impulse(text: Optional[str] = None, use_voice: bool = False, vault_path: Optional[Path] = None) -> Path:
-    """Capture an impulse as a Markdown file inside the vault inbox.
+def _format_relative_path(path: Path) -> str:
+    """Return a user-friendly string for ``path`` relative to the CWD when possible."""
+
+    try:
+        return str(path.relative_to(Path.cwd()))
+    except ValueError:
+        return str(path)
 
-    Args:
-        text: Optional pre-supplied text input.
-        use_voice: Whether to capture audio using Whisper.
-        vault_path: Optional override for the vault location. Defaults to the
-            path resolved by :func:`file_ops.get_vault_path`.
 
-    Returns:
-        Path: The path to the created Markdown file.
-    """
+def capture_impulse(
+    text: Optional[str] = None,
+    use_voice: bool = False,
+    vault_path: Optional[Path] = None,
+    model: str = "base",
+    device: Optional[str] = None,
+    max_seconds: int = 90,
+    language: str = "en",
+    keep_audio: bool = True,
+) -> Path:
+    """Capture an impulse as a Markdown file inside the vault inbox."""
+
     env.load_environment()
     resolved_vault = vault_path or file_ops.get_vault_path()
     inbox_dir = resolved_vault / "Inbox"
     file_ops.ensure_directory(inbox_dir)
 
+    audio_path: Optional[Path] = None
+
     if use_voice:
+        click.echo(f"Recording (max {max_seconds}s)…")
+        click.echo(f"Transcribing locally with Whisper (model={model})…")
         try:
-            text = whisper_integration.capture_voice_input()
-        except NotImplementedError as exc:  # pragma: no cover - placeholder path
-            raise click.ClickException("Voice capture is not implemented yet.") from exc
+            audio_path, text = whisper_integration.capture_and_transcribe_local(
+                vault_dir=resolved_vault,
+                model=model,
+                device=device,
+                max_seconds=max_seconds,
+                language=language,
+                keep_audio=keep_audio,
+            )
+        except whisper_integration.AudioCaptureError as exc:
+            raise click.ClickException(str(exc)) from exc
+        except whisper_integration.TranscriptionError as exc:
+            path_hint = exc.audio_path
+            message = str(exc)
+            if path_hint is not None:
+                message = f"{message}\nRaw audio saved at: {_format_relative_path(path_hint)}"
+            raise click.ClickException(message) from exc
+
+        click.echo(f"Saved audio: {_format_relative_path(audio_path)}")
 
     if text is None:
         text = _prompt_for_text()
@@ -51,14 +80,34 @@ def capture_impulse(text: Optional[str] = None, use_voice: bool = False, vault_p
     if not text:
         raise click.ClickException("No impulse provided.")
 
-    timestamp = datetime.now()
-    filename = file_ops.timestamped_filename(timestamp=timestamp)
-    file_path = inbox_dir / filename
-
-    front_matter = file_ops.render_front_matter({
-        "captured_at": timestamp.isoformat(),
-    })
-    content = f"{front_matter}\n{text.strip()}\n"
+    if audio_path is not None:
+        timestamp_str = audio_path.stem
+        timestamp = datetime.strptime(timestamp_str, "%Y-%m-%dT%H-%M-%S")
+        note_filename = f"{timestamp_str}.md"
+        try:
+            relative_to_vault = audio_path.relative_to(resolved_vault)
+            audio_relative = str(Path(resolved_vault.name) / relative_to_vault)
+        except ValueError:
+            audio_relative = _format_relative_path(audio_path)
+        metadata = {
+            "date": timestamp.isoformat(),
+            "type": "voice-dump",
+            "backend": "local",
+            "audio_path": audio_relative,
+            "lang": language,
+            "summary": "",
+            "tags": ["voice", "offload"],
+        }
+    else:
+        timestamp = datetime.now()
+        note_filename = file_ops.timestamped_filename(timestamp=timestamp)
+        metadata = {
+            "captured_at": timestamp.isoformat(),
+        }
+
+    file_path = inbox_dir / note_filename
+    front_matter = file_ops.render_front_matter(metadata)
+    content = f"{front_matter}\n\n{text.strip()}\n"
     file_ops.safe_write(file_path, content)
 
     return file_path
@@ -67,10 +116,41 @@ def capture_impulse(text: Optional[str] = None, use_voice: bool = False, vault_p
 @click.command()
 @click.argument("text", required=False)
 @click.option("--voice", "use_voice", is_flag=True, help="Capture voice input via Whisper.")
-def capture(text: Optional[str], use_voice: bool) -> None:
+@click.option("--model", default="base", show_default=True, help="Whisper model to use for transcription.")
+@click.option("--device", default=None, help="Input device index or name for recording.")
+@click.option("--max-seconds", default=90, show_default=True, type=int, help="Maximum recording duration in seconds.")
+@click.option("--lang", "language", default="en", show_default=True, help="Language hint for Whisper transcription.")
+@click.option(
+    "--keep-audio/--discard-audio",
+    default=True,
+    show_default=True,
+    help="Keep or discard the raw audio file after transcription.",
+)
+def capture(
+    text: Optional[str],
+    use_voice: bool,
+    model: str,
+    device: Optional[str],
+    max_seconds: int,
+    language: str,
+    keep_audio: bool,
+) -> None:
     """Capture an impulse as Markdown in the Inbox."""
-    file_path = capture_impulse(text=text, use_voice=use_voice)
-    click.echo(f"Impulse captured: {file_path}")
+
+    if use_voice:
+        text = None
+
+    file_path = capture_impulse(
+        text=text,
+        use_voice=use_voice,
+        model=model,
+        device=device,
+        max_seconds=max_seconds,
+        language=language,
+        keep_audio=keep_audio,
+    )
+
+    click.echo(f"Impulse captured: {_format_relative_path(file_path)}")
 
 
 if __name__ == "__main__":  # pragma: no cover - CLI entry point
diff --git a/impulse_offloader/utils/whisper_integration.py b/impulse_offloader/utils/whisper_integration.py
index 274a99e..f1d1cd5 100644
--- a/impulse_offloader/utils/whisper_integration.py
+++ b/impulse_offloader/utils/whisper_integration.py
@@ -1,7 +1,227 @@
-"""Wrappers for capturing audio and transcribing with Whisper."""
+"""Prompt for Codex - Implement Local Whisper Voice Capture (Linux, 90s, keep audio).
+
+Utilities for recording microphone input and transcribing it locally with the
+Whisper model suite. Tailored for Linux environments with ``ffmpeg`` and ALSA/
+PulseAudio support.
+"""
 from __future__ import annotations
 
+import shutil
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, Tuple
+
+try:  # pragma: no cover - import guard for optional dependency
+    import numpy as np
+except ImportError:  # pragma: no cover - handled at runtime
+    np = None  # type: ignore[assignment]
+
+try:  # pragma: no cover - import guard for optional dependency
+    import sounddevice as sd
+except ImportError:  # pragma: no cover - handled at runtime
+    sd = None  # type: ignore[assignment]
+
+try:  # pragma: no cover - import guard for optional dependency
+    from scipy.io import wavfile
+except ImportError:  # pragma: no cover - handled at runtime
+    wavfile = None  # type: ignore[assignment]
+
+
+class AudioCaptureError(RuntimeError):
+    """Raised when audio capture fails for a user-facing reason."""
+
+
+class TranscriptionError(RuntimeError):
+    """Raised when transcription fails for a user-facing reason."""
+
+    def __init__(self, message: str, *, audio_path: Optional[Path] = None) -> None:
+        super().__init__(message)
+        self.audio_path = audio_path
+
+
+def _resolve_device(device: Optional[str]) -> Optional[int]:
+    """Resolve a user-provided device string into a sounddevice identifier."""
+
+    if device is None:
+        return None
+
+    devices = sd.query_devices()
+    if device.isdigit():
+        idx = int(device)
+        if 0 <= idx < len(devices):
+            return idx
+    for idx, entry in enumerate(devices):
+        if device.lower() in (str(entry.get("name", "")) or "").lower():
+            return idx
+
+    device_list = "\n".join(
+        f"[{idx}] {entry.get('name', 'Unknown')}" for idx, entry in enumerate(devices)
+    )
+    message = (
+        "Unable to use audio input device "
+        f"{device!r}. Available devices:\n{device_list}\n"
+        "Specify a device index or name using --device."
+    )
+    raise AudioCaptureError(message)
+
+
+def record_audio(
+    out_wav: Path,
+    device: Optional[str] = None,
+    max_seconds: int = 90,
+    sample_rate: int = 16_000,
+    channels: int = 1,
+) -> Path:
+    """Record microphone audio to ``out_wav`` for up to ``max_seconds`` seconds.
+
+    Parameters
+    ----------
+    out_wav:
+        Destination path for the recorded WAV file.
+    device:
+        Optional sounddevice identifier. Can be a device index or partial name.
+    max_seconds:
+        Maximum number of seconds to record. Recording stops automatically once
+        this limit is reached.
+    sample_rate:
+        Sampling rate for recording in Hertz.
+    channels:
+        Number of audio channels to record.
+
+    Returns
+    -------
+    Path
+        The path to the recorded WAV file.
+
+    Raises
+    ------
+    AudioCaptureError
+        If no audio could be captured or the device configuration fails.
+    """
+
+    if sd is None:
+        raise AudioCaptureError(
+            "The `sounddevice` package is required for audio recording."
+        )
+    if np is None:
+        raise AudioCaptureError("The `numpy` package is required for audio recording.")
+    if wavfile is None:
+        raise AudioCaptureError("The `scipy` package is required for audio recording.")
+
+    resolved_device = _resolve_device(device)
+    duration_frames = max_seconds * sample_rate
+
+    try:
+        sd.check_input_settings(
+            device=resolved_device, samplerate=sample_rate, channels=channels
+        )
+    except Exception as exc:  # pragma: no cover - relies on system configuration
+        raise AudioCaptureError(str(exc)) from exc
+
+    try:
+        recording = sd.rec(
+            frames=duration_frames,
+            samplerate=sample_rate,
+            channels=channels,
+            dtype="float32",
+            device=resolved_device,
+        )
+        sd.wait()
+    except Exception as exc:  # pragma: no cover - depends on audio backend
+        devices = sd.query_devices()
+        device_list = "\n".join(
+            f"[{idx}] {entry.get('name', 'Unknown')}" for idx, entry in enumerate(devices)
+        )
+        raise AudioCaptureError(
+            f"Failed to record audio: {exc}\nAvailable devices:\n{device_list}"
+        ) from exc
+
+    if recording.size == 0:
+        raise AudioCaptureError("No audio was captured from the microphone.")
+
+    peak = float(np.max(np.abs(recording)))
+    if peak > 1.0:
+        recording = recording / peak
+
+    pcm_audio = np.clip(recording, -1.0, 1.0)
+    pcm_audio = (pcm_audio * np.iinfo(np.int16).max).astype(np.int16)
+
+    out_wav.parent.mkdir(parents=True, exist_ok=True)
+    wavfile.write(out_wav, sample_rate, pcm_audio)
+    return out_wav
+
+
+def _ensure_ffmpeg_available() -> None:
+    """Ensure ``ffmpeg`` is available on the system path."""
+
+    if shutil.which("ffmpeg"):
+        return
+    hint = (
+        "ffmpeg is required for Whisper transcription. "
+        "Install it via `sudo apt-get install ffmpeg` or your distro's package manager."
+    )
+    raise TranscriptionError(hint)
+
+
+def transcribe_local(audio_path: Path, model: str = "base", language: str = "en") -> str:
+    """Transcribe ``audio_path`` using a local Whisper model."""
+
+    _ensure_ffmpeg_available()
+
+    try:
+        import whisper
+    except ImportError as exc:  # pragma: no cover - depends on environment
+        raise TranscriptionError(
+            "The `whisper` package is not installed.", audio_path=audio_path
+        ) from exc
+
+    try:
+        whisper_model = whisper.load_model(model)
+        result = whisper_model.transcribe(str(audio_path), language=language)
+    except Exception as exc:  # pragma: no cover - heavy dependency path
+        raise TranscriptionError(
+            f"Failed to transcribe audio: {exc}", audio_path=audio_path
+        ) from exc
+
+    text = result.get("text", "").strip()
+    if not text:
+        raise TranscriptionError(
+            "Whisper produced an empty transcription.", audio_path=audio_path
+        )
+    return text
+
+
+def capture_and_transcribe_local(
+    vault_dir: Path,
+    model: str = "base",
+    device: Optional[str] = None,
+    max_seconds: int = 90,
+    language: str = "en",
+    keep_audio: bool = True,
+) -> Tuple[Path, str]:
+    """Capture audio and transcribe it locally with Whisper."""
+
+    timestamp = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
+    inbox_dir = vault_dir / "Inbox"
+    audio_dir = inbox_dir / "audio"
+    audio_dir.mkdir(parents=True, exist_ok=True)
+    inbox_dir.mkdir(parents=True, exist_ok=True)
+
+    audio_path = audio_dir / f"{timestamp}.wav"
+    record_audio(audio_path, device=device, max_seconds=max_seconds)
+
+    try:
+        transcription = transcribe_local(audio_path, model=model, language=language)
+    except TranscriptionError as exc:
+        if not keep_audio:
+            audio_path.unlink(missing_ok=True)
+        raise TranscriptionError(str(exc), audio_path=audio_path) from exc
+    except Exception as exc:
+        if not keep_audio:
+            audio_path.unlink(missing_ok=True)
+        raise TranscriptionError(str(exc), audio_path=audio_path) from exc
+
+    if not keep_audio:
+        audio_path.unlink(missing_ok=True)
 
-def capture_voice_input() -> str:
-    """Record a short audio clip and return a Whisper transcription."""
-    raise NotImplementedError("Voice capture is not implemented yet.")
+    return audio_path, transcription
diff --git a/requirements.txt b/requirements.txt
index 658bd51..3ce4265 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -5,3 +5,6 @@ pydub
 markdown
 pyyaml
 click
+sounddevice
+scipy
+numpy
diff --git a/tests/test_whisper_integration.py b/tests/test_whisper_integration.py
new file mode 100644
index 0000000..ccd4a81
--- /dev/null
+++ b/tests/test_whisper_integration.py
@@ -0,0 +1,91 @@
+"""Tests for the Whisper integration utilities."""
+from __future__ import annotations
+
+import types
+import wave
+from pathlib import Path
+
+import pytest
+
+np = pytest.importorskip("numpy")
+
+from impulse_offloader.utils import whisper_integration as wi
+
+
+def test_record_audio_creates_wav_linux(monkeypatch, tmp_path: Path) -> None:
+    frames = 16_000
+    buffer = np.zeros((frames, 1), dtype="float32")
+
+    stub_sd = types.SimpleNamespace()
+    stub_sd.query_devices = lambda: [{"name": "Default"}]
+    stub_sd.check_input_settings = lambda **kwargs: None
+
+    def fake_rec(*, frames: int, samplerate: int, channels: int, dtype: str, device: object):
+        assert frames == 16_000
+        assert samplerate == 16_000
+        assert channels == 1
+        assert dtype == "float32"
+        return buffer
+
+    stub_sd.rec = lambda *args, **kwargs: fake_rec(*args, **kwargs)
+    stub_sd.wait = lambda: None
+
+    monkeypatch.setattr(wi, "sd", stub_sd)
+
+    class FakeWavfile:
+        @staticmethod
+        def write(path: Path, rate: int, data: np.ndarray) -> None:
+            channels = data.shape[1] if data.ndim > 1 else 1
+            with wave.open(str(path), "wb") as wav_file:
+                wav_file.setnchannels(channels)
+                wav_file.setsampwidth(2)
+                wav_file.setframerate(rate)
+                wav_file.writeframes(data.tobytes())
+
+    monkeypatch.setattr(wi, "wavfile", FakeWavfile)
+
+    output = tmp_path / "audio.wav"
+    path = wi.record_audio(output, max_seconds=1)
+    assert path == output
+    assert path.exists()
+
+    with wave.open(str(path), "rb") as wav_file:
+        assert wav_file.getframerate() == 16_000
+        assert wav_file.getnchannels() == 1
+
+
+def test_transcribe_local_missing_ffmpeg(monkeypatch, tmp_path: Path) -> None:
+    dummy_audio = tmp_path / "dummy.wav"
+    dummy_audio.write_bytes(b"")
+
+    monkeypatch.setattr(wi.shutil, "which", lambda _: None)
+
+    with pytest.raises(wi.TranscriptionError) as exc:
+        wi.transcribe_local(dummy_audio)
+
+    assert "ffmpeg is required" in str(exc.value)
+
+
+def test_capture_and_transcribe_local_paths(monkeypatch, tmp_path: Path) -> None:
+    created_files: dict[str, Path] = {}
+
+    def fake_record(out: Path, **kwargs) -> Path:
+        out.parent.mkdir(parents=True, exist_ok=True)
+        out.write_bytes(b"RIFF")
+        created_files["audio"] = out
+        return out
+
+    def fake_transcribe(path: Path, **kwargs) -> str:
+        created_files["transcribed"] = path
+        return "hello world"
+
+    monkeypatch.setattr(wi, "record_audio", fake_record)
+    monkeypatch.setattr(wi, "transcribe_local", fake_transcribe)
+
+    audio_path, text = wi.capture_and_transcribe_local(tmp_path, model="base", keep_audio=True)
+
+    assert text == "hello world"
+    assert audio_path.parent == tmp_path / "Inbox" / "audio"
+    assert audio_path.name.endswith(".wav")
+    assert audio_path == created_files["audio"]
+    assert created_files["transcribed"] == audio_path