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

#!/usr/bin/env python3
"""
Hermes Voice Client
VAD → Whisper STT (2 niveaux) → keyword "hermes" → hermes -z SSH → piper TTS

Usage:
  python hermes-voice.py             # dites "Ok Hermes, ..." — écoute en permanence
  python hermes-voice.py --ptt       # Entrée pour commencer/terminer
  python hermes-voice.py --no-tts    # réponses texte seulement (debug)
"""

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_vad(whisper_fast, whisper_full, audio: np.ndarray, no_tts: bool):
    """VAD : détection rapide (small) puis transcription précise (large-v3)."""
    if len(audio) / SAMPLE_RATE < MIN_SPEECH:
        return

    # Étape 1 — vérification rapide du mot-clé avec le petit modèle
    quick = transcribe(whisper_fast, audio)
    found, _ = detect_keyword(quick)
    if not found:
        print(f"(ignoré : \"{quick[:60]}\")")
        return

    # Étape 2 — transcription précise uniquement si mot-clé trouvé
    print("  ⏳ Transcription...", end=" ", flush=True)
    text = transcribe(whisper_full, audio)
    if not text:
        print("(rien compris)")
        return
    _, command = detect_keyword(text)
    query = command or text

    print(f"\n  🗣️  Vous   : {text}")
    _respond(query, no_tts)


def handle_ptt(whisper_full, audio: np.ndarray, no_tts: bool):
    """PTT : transcription directe sans filtre mot-clé."""
    if len(audio) / SAMPLE_RATE < MIN_SPEECH:
        return

    print("  ⏳ Transcription...", end=" ", flush=True)
    text = transcribe(whisper_full, audio)
    if not text:
        print("(rien compris)")
        return
    print(f"\n  🗣️  Vous   : {text}")
    _respond(text, no_tts)


def _respond(query: str, no_tts: bool):
    print("  ⏳ Hermes...", end=" ", flush=True)
    try:
        response = ask_hermes(query)
    except (requests.RequestException, RuntimeError) as e:
        print(f"\n  ❌ Erreur : {e}")
        return
    print(f"\n  🤖 Hermes : {response}\n")
    if not no_tts:
        speak(response)


def run_vad(whisper_fast, whisper_full, no_tts: bool):
    """Écoute continue — dites 'Ok Hermes, ...' pour déclencher."""
    print(f"🎙️  En écoute — dites \"Ok {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_vad(whisper_fast, whisper_full, np.concatenate(captured), no_tts)


def run_ptt(whisper_full, 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_ptt(whisper_full, np.concatenate(chunks), no_tts)


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)

    from faster_whisper import WhisperModel

    if args.ptt:
        print(f"  ⏳ Chargement Whisper ({WHISPER_SIZE})...", end=" ", flush=True)
        whisper_full = WhisperModel(WHISPER_SIZE, device="cpu", compute_type="int8")
        print("✓\n")
        try:
            run_ptt(whisper_full, args.no_tts)
        except KeyboardInterrupt:
            print("\n👋 Au revoir")
    else:
        # Mode VAD : deux modèles — small pour détection rapide, large pour transcription
        print("  ⏳ Chargement Whisper rapide (small)...", end=" ", flush=True)
        whisper_fast = WhisperModel("small", device="cpu", compute_type="int8")
        print("✓")
        print(f"  ⏳ Chargement Whisper précis ({WHISPER_SIZE})...", end=" ", flush=True)
        whisper_full = WhisperModel(WHISPER_SIZE, device="cpu", compute_type="int8")
        print("✓\n")
        try:
            run_vad(whisper_fast, whisper_full, args.no_tts)
        except KeyboardInterrupt:
            print("\n👋 Au revoir")


if __name__ == "__main__":
    main()