Funk-lab/tools/hermes-voice/hermes-voice.py

#!/usr/bin/env python3
"""
Hermes Voice Client
VAD → Whisper STT → keyword "hermes" → LiteLLM → piper TTS

Usage:
  python hermes-voice.py             # parlez naturellement, commencez par "Hermes, ..."
  python hermes-voice.py --ptt       # Entrée pour commencer/terminer (plus simple)
  python hermes-voice.py --no-tts    # réponses texte seulement (pour tester)
"""

import argparse
import queue
import subprocess
import tempfile
from collections import deque
from pathlib import Path

import numpy as np
import requests
import sounddevice as sd

# ── Configuration ──────────────────────────────────────────────────────────────
# Si port 4000 inaccessible : ouvrir un tunnel SSH avant de lancer :
#   ssh -L 4000:localhost:4000 user@192.168.1.200
# puis changer LITELLM_URL → http://localhost:4000/...
LITELLM_URL   = "http://192.168.1.200:4000/v1/chat/completions"
LITELLM_KEY   = "lm-studio"
HERMES_MODEL  = "hermes-default"
SYSTEM_PROMPT = (
    "Tu es Hermes, l'assistant vocal du homelab Funk. "
    "Réponds toujours en français, de façon concise (2-3 phrases maximum)."
)

SAMPLE_RATE   = 16000
BLOCK_SIZE    = 512    # 32ms par bloc
PRE_ROLL      = 10     # blocs à conserver avant la détection (début de phrase)
VOICE_THRESH  = 0.012  # seuil RMS normalisé pour détecter la parole
SILENCE_SEC   = 1.5    # silence pour terminer l'enregistrement
MIN_SPEECH    = 0.8    # durée minimale traitée (filtre les bruits courts)

WHISPER_SIZE  = "large-v3"  # large-v3 : meilleure reconnaissance FR (distil-large-v3 dérive vers l'anglais)
KEYWORD       = "hermes"
KEYWORD_POS   = 5        # le mot-clé doit être dans les N premiers mots

# Piper TTS — binaire : https://github.com/rhasspy/piper/releases
# Voix FR    : wget .../fr_FR-upmc-medium.onnx → ~/.local/share/piper/
PIPER_BIN     = Path.home() / ".local/share/piper-runtime/piper"
PIPER_VOICE   = Path.home() / ".local/share/piper/fr_FR-upmc-medium.onnx"

MAX_HISTORY   = 10  # nombre d'échanges (user+assistant) conservés en mémoire

# Mode "full Hermes" via SSH (soul + skills + RAG + mémoire long-terme)
# Passer HERMES_MODE = "ssh" pour router par `hermes -z` sur storage-01
# Passer HERMES_MODE = "litellm" pour appel direct LiteLLM (session history uniquement)
HERMES_MODE   = "ssh"       # "litellm" | "ssh"
HERMES_SSH    = "s01"       # alias SSH défini dans ~/.ssh/config

# ──────────────────────────────────────────────────────────────────────────────

_history: list[dict] = []  # historique session (mode litellm uniquement)


def beep(freq: int = 880, duration: float = 0.12):
    t    = np.linspace(0, duration, int(SAMPLE_RATE * duration), endpoint=False)
    wave = (0.25 * np.sin(2 * np.pi * freq * t)).astype(np.float32)
    sd.play(wave, samplerate=SAMPLE_RATE)
    sd.wait()


def rms(chunk: np.ndarray) -> float:
    return float(np.sqrt(np.mean(chunk.astype(np.float32) ** 2)) / 32768.0)


def detect_keyword(text: str) -> tuple[bool, str]:
    """Vérifie si KEYWORD est dans les premiers mots et retourne la commande sans le mot-clé."""
    words = text.strip().split()
    stripped = [w.lower().strip(",.!?«»") for w in words]
    found = any(KEYWORD in w for w in stripped[:KEYWORD_POS])
    if not found:
        return False, ""
    try:
        idx = next(i for i, w in enumerate(stripped) if KEYWORD in w)
        command = " ".join(words[idx + 1:]).strip()
    except StopIteration:
        command = text
    return True, command or text


def transcribe(whisper_model, audio: np.ndarray) -> str:
    audio_f32 = audio.astype(np.float32) / 32768.0
    segments, _ = whisper_model.transcribe(
        audio_f32,
        language="fr",
        task="transcribe",
        beam_size=5,
        vad_filter=True,
        initial_prompt="Transcription en français.",
    )
    return " ".join(seg.text for seg in segments).strip()


def ask_hermes(text: str) -> str:
    if HERMES_MODE == "ssh":
        return _ask_via_ssh(text)
    return _ask_via_litellm(text)


def _ask_via_ssh(text: str) -> str:
    """Route par le vrai Hermes sur storage-01 (soul + skills + RAG + mémoire)."""
    import subprocess
    # Forcer le français + échapper les guillemets simples pour bash
    query = f"Réponds en français. {text}"
    safe  = query.replace("'", "'\\''")
    cmd = [
        "ssh", HERMES_SSH,
        f"sudo -i -u hermes bash -c 'HERMES_HOME=/srv/data/hermes hermes --profile funk-ai -z \"{safe}\"'"
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    output = result.stdout.strip()
    if not output and result.returncode != 0:
        raise RuntimeError(result.stderr.strip() or "hermes -z a échoué")
    return output


def _ask_via_litellm(text: str) -> str:
    """Appel direct LiteLLM avec historique de session."""
    global _history

    lower = text.lower().strip()
    if any(w in lower for w in ("reset", "nouveau contexte", "oublie tout", "efface")):
        _history.clear()
        return "Contexte effacé, on repart de zéro."

    _history.append({"role": "user", "content": text})
    window = _history[-(MAX_HISTORY * 2):]

    resp = requests.post(
        LITELLM_URL,
        json={
            "model": HERMES_MODEL,
            "messages": [{"role": "system", "content": SYSTEM_PROMPT}] + window,
            "max_tokens": 300,
            "temperature": 0.7,
        },
        headers={"Authorization": f"Bearer {LITELLM_KEY}"},
        timeout=30,
    )
    resp.raise_for_status()
    response = resp.json()["choices"][0]["message"]["content"].strip()
    _history.append({"role": "assistant", "content": response})
    return response


def speak(text: str):
    voice   = Path(PIPER_VOICE).expanduser()
    bin_path = Path(PIPER_BIN).expanduser()
    if not voice.exists():
        print(f"  [TTS] voix introuvable : {voice}")
        return
    if not bin_path.exists():
        print(f"  [TTS] piper introuvable : {bin_path}")
        return
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        out = Path(f.name)
    # Les .so de piper doivent être dans le même répertoire que le binaire
    lib_dir = str(bin_path.parent)
    env = {"LD_LIBRARY_PATH": lib_dir}
    try:
        subprocess.run(
            [str(bin_path), "--model", str(voice), "--output_file", str(out)],
            input=text.encode(),
            capture_output=True,
            check=True,
            env=env,
        )
        subprocess.run(["aplay", "-q", str(out)], check=False)
    except FileNotFoundError:
        print("  [TTS] piper non trouvé — voir https://github.com/rhasspy/piper/releases")
    finally:
        out.unlink(missing_ok=True)


def handle(whisper_model, audio: np.ndarray, no_tts: bool, require_keyword: bool = True):
    if len(audio) / SAMPLE_RATE < MIN_SPEECH:
        return

    print("  ⏳ Transcription...", end=" ", flush=True)
    text = transcribe(whisper_model, audio)
    if not text:
        print("(rien compris)")
        return

    if require_keyword:
        found, command = detect_keyword(text)
        if not found:
            print(f"(ignoré : \"{text[:60]}\")")
            return
        query = command
    else:
        query = text

    print(f"\n  🗣️  Vous   : {text}")
    print("  ⏳ Hermes...", end=" ", flush=True)
    try:
        response = ask_hermes(query)
    except requests.RequestException as e:
        print(f"\n  ❌ Réseau : {e}")
        return
    print(f"\n  🤖 Hermes : {response}\n")

    if not no_tts:
        speak(response)


def run_vad(whisper_model, no_tts: bool):
    """Écoute continue par VAD — commencer par 'Hermes, ...' pour déclencher."""
    print(f"🎙️  En écoute — commencez par \"{KEYWORD.capitalize()}, ...\" (Ctrl+C pour quitter)\n")

    audio_q: queue.Queue = queue.Queue()
    pre_roll: deque = deque(maxlen=PRE_ROLL)

    def callback(indata, frames, time_info, status):
        chunk = (indata[:, 0] if indata.ndim > 1 else indata.flatten()).copy()
        audio_q.put(chunk)

    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16",
                        blocksize=BLOCK_SIZE, callback=callback):
        while True:
            chunk = audio_q.get()
            pre_roll.append(chunk)

            if rms(chunk) < VOICE_THRESH:
                continue

            # Voix détectée — enregistrer jusqu'au silence
            print("  🔴 ", end="", flush=True)
            captured = list(pre_roll)
            silent_blocks  = 0
            required_silent = int(SILENCE_SEC * SAMPLE_RATE / BLOCK_SIZE)

            while True:
                try:
                    chunk = audio_q.get(timeout=3.0)
                except queue.Empty:
                    break
                captured.append(chunk)
                if rms(chunk) < VOICE_THRESH:
                    silent_blocks += 1
                else:
                    silent_blocks = 0
                    print(".", end="", flush=True)
                if silent_blocks >= required_silent:
                    break

            print()
            beep(440)
            handle(whisper_model, np.concatenate(captured), no_tts, require_keyword=True)


def run_ptt(whisper_model, no_tts: bool):
    """Push-to-talk : Entrée pour commencer, Entrée pour terminer."""
    print("🎙️  Mode push-to-talk — Entrée pour parler, Entrée pour terminer (Ctrl+C pour quitter)\n")

    while True:
        input("[ ↵ Entrée pour parler ]")
        audio_q: queue.Queue = queue.Queue()

        def callback(indata, frames, time_info, status):
            chunk = (indata[:, 0] if indata.ndim > 1 else indata.flatten()).copy()
            audio_q.put(chunk)

        print("  🔴 Enregistrement... (Entrée pour terminer)")
        with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16",
                            blocksize=BLOCK_SIZE, callback=callback):
            input()
            chunks = []
            while not audio_q.empty():
                chunks.append(audio_q.get_nowait())

        if not chunks:
            continue
        beep(440)
        handle(whisper_model, np.concatenate(chunks), no_tts, require_keyword=False)


def main():
    parser = argparse.ArgumentParser(description="Hermes Voice Client")
    parser.add_argument("--ptt",    action="store_true", help="Mode push-to-talk")
    parser.add_argument("--no-tts", action="store_true", help="Réponses texte seulement")
    args = parser.parse_args()

    print("Hermes Voice Client")
    print("=" * 40)
    print(f"  ⏳ Chargement Whisper ({WHISPER_SIZE})...", end=" ", flush=True)
    from faster_whisper import WhisperModel
    whisper = WhisperModel(WHISPER_SIZE, device="cpu", compute_type="int8")
    print("✓\n")

    try:
        if args.ptt:
            run_ptt(whisper, args.no_tts)
        else:
            run_vad(whisper, args.no_tts)
    except KeyboardInterrupt:
        print("\n👋 Au revoir")


if __name__ == "__main__":
    main()