VoiceAgent/selftest.py

#!/usr/bin/env python3
"""Check each piece of the voice pipeline independently.

Run this first when something isn't working — it tells you which stage is at
fault instead of making you read pipeline logs.
"""

import asyncio
import sys
import time
import wave
from pathlib import Path

PHRASE = "The quick brown fox jumps over the lazy dog."
KOKORO_CACHE = Path.home() / ".cache/pipecat/kokoro-onnx"

results: list[tuple[str, bool, str]] = []


def record(name: str, ok: bool, detail: str):
    results.append((name, ok, detail))
    print(f"  {'PASS' if ok else 'FAIL'}  {name}: {detail}")


def check_devices():
    import sounddevice as sd

    try:
        default_in, default_out = sd.default.device
        names = sd.query_devices()
        record(
            "audio devices",
            True,
            f"in={names[default_in]['name']!r}, out={names[default_out]['name']!r}",
        )
    except Exception as e:
        record("audio devices", False, str(e))


def check_microphone():
    import numpy as np
    import sounddevice as sd

    try:
        print("  ... recording 3 seconds, please say something")
        rec = sd.rec(3 * 16000, samplerate=16000, channels=1, dtype="int16")
        sd.wait()
        peak = int(np.abs(rec).max())
    except Exception as e:
        record("microphone", False, str(e))
        return None

    if peak == 0:
        record(
            "microphone",
            False,
            "captured pure silence — grant your terminal microphone access in "
            "System Settings > Privacy & Security > Microphone",
        )
        return None
    record("microphone", True, f"peak amplitude {peak}")
    return rec


def check_hold_key():
    """Hold-to-talk needs Input Monitoring; without it we fall back to a toggle."""
    from global_hotkey import permission_granted

    if permission_granted():
        record("hold-to-talk permission", True, "Input Monitoring granted")
    else:
        record(
            "hold-to-talk (optional)",
            False,
            "Input Monitoring not granted — add your terminal under System Settings > "
            "Privacy & Security > Input Monitoring and restart it. SPACE toggle still works.",
        )


def check_macos_voices():
    from apple_tts import available_voices, find_voice

    voices = available_voices()
    irish = [name for name, language in voices if language == "en-IE"]
    if not irish:
        record(
            "macos voices (optional)",
            False,
            "no Irish (en-IE) voice installed; add one under System Settings > "
            "Accessibility > Spoken Content > System Voice",
        )
        return
    record(
        "macos voices",
        bool(find_voice("Moira")),
        f"{len(voices)} installed, Irish: {', '.join(irish)}",
    )


def check_kokoro():
    from kokoro_onnx import Kokoro

    try:
        kokoro = Kokoro(str(KOKORO_CACHE / "kokoro-v1.0.onnx"), str(KOKORO_CACHE / "voices-v1.0.bin"))
        samples, rate = kokoro.create(PHRASE, voice="af_heart", speed=1.0, lang="en-us")
    except Exception as e:
        record("kokoro tts", False, str(e))
        return None
    record("kokoro tts", True, f"{len(samples) / rate:.2f}s of audio at {rate} Hz")
    return samples, rate


def write_wav(synthesized, path="/tmp/voice-agent-probe.wav"):
    import numpy as np

    samples, rate = synthesized
    pcm = (np.clip(samples, -1, 1) * 32767).astype(np.int16)
    with wave.open(path, "wb") as f:
        f.setnchannels(1)
        f.setsampwidth(2)
        f.setframerate(rate)
        f.writeframes(pcm.tobytes())
    return path


def matches(heard: str) -> bool:
    return heard.lower().strip(" .") == PHRASE.lower().strip(" .")


def check_apple_stt(synthesized):
    """Transcribe Kokoro's own output — a full loop through the audio stack."""
    from apple_stt import _recognize_file, probe

    available, reason = probe()
    if not available:
        record("apple speech to text", False, reason)
        return
    if synthesized is None:
        record("apple speech to text", False, "skipped, Kokoro produced no audio to transcribe")
        return

    started = time.time()
    try:
        heard = _recognize_file(write_wav(synthesized), "en-US", 20.0).strip()
    except Exception as e:
        record("apple speech to text", False, str(e))
        return
    record("apple speech to text", matches(heard), f"heard {heard!r} in {time.time() - started:.2f}s")


def check_mlx_whisper(synthesized):
    """Optional — only matters if you want --stt-engine mlx."""
    if synthesized is None:
        record("mlx whisper (optional)", False, "skipped, no audio to transcribe")
        return

    import mlx_whisper
    import numpy as np
    import soxr

    samples, rate = synthesized
    audio = soxr.resample(samples.astype(np.float32), rate, 16000)
    try:
        heard = mlx_whisper.transcribe(
            audio, path_or_hf_repo="mlx-community/whisper-large-v3-turbo-q4", language="en"
        )["text"].strip()
    except Exception as e:
        detail = str(e).splitlines()[0]
        record("mlx whisper (optional)", False, f"{detail} — use --stt-engine apple or cpu")
        return
    record("mlx whisper (optional)", matches(heard), f"heard {heard!r}")


def check_opencode_llm():
    from opencode_llm import probe_opencode

    available, reason = probe_opencode(model="ollama-cloud/gemma4:31b")
    record("opencode llm", available, reason)


def check_macos_llm():
    from apple_llm import probe_apple_llm

    available, reason = probe_apple_llm()
    record("macos llm (optional)", available, reason)


async def check_claude():
    from claude_agent_sdk import ClaudeSDKClient, ResultMessage, StreamEvent

    from bot import build_claude_options
    from claude_llm import _text_delta

    import argparse

    from bot import DEFAULT_CLAUDE_MODEL

    options = build_claude_options(
        argparse.Namespace(
            allow_writes=False,
            cwd=None,
            claude_model=DEFAULT_CLAUDE_MODEL,
            load_settings=False,
        )
    )
    try:
        async with ClaudeSDKClient(options=options) as client:
            await client.query("Reply with exactly one word: ready")
            spoken = []
            async for message in client.receive_response():
                if isinstance(message, StreamEvent):
                    text = _text_delta(message)
                    if text:
                        spoken.append(text)
                elif isinstance(message, ResultMessage) and message.is_error:
                    record("claude session (optional)", False, f"error: {message.result}")
                    return
    except Exception as e:
        hint = ""
        if "-9" in str(e):
            hint = " — the CLI was killed applying its own sandbox; run this from a normal terminal"
        record("claude session (optional)", False, f"{e}{hint}")
        return
    record(
        "claude session (optional)",
        True,
        f"replied {''.join(spoken).strip()!r} using {DEFAULT_CLAUDE_MODEL}",
    )


async def main():
    print("Checking the voice pipeline...\n")
    check_devices()
    check_microphone()
    check_hold_key()
    check_macos_voices()
    synthesized = check_kokoro()
    check_apple_stt(synthesized)
    check_mlx_whisper(synthesized)
    check_macos_llm()
    check_opencode_llm()
    await check_claude()

    required_failures = [
        name for name, ok, _ in results if not ok and not name.endswith("(optional)")
    ]
    print()
    if required_failures:
        print(f"{len(required_failures)} check(s) failed: {', '.join(required_failures)}")
        return 1
    print("Everything works. Run ./talk to start a conversation.")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))