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

#!/usr/bin/env python3
"""
Hermes Voice Client
Deux modes : VEILLE (wake word "Ok Hermes") → CONVERSATION (parole libre 60s)

Usage:
  python hermes-voice.py             # écoute en permanence
  python hermes-voice.py --ptt       # Entrée pour parler
  python hermes-voice.py --no-tts    # réponses texte seulement (debug)
"""

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

import numpy as np
import requests
import sounddevice as sd

# ── Configuration ──────────────────────────────────────────────────────────────
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
SILENCE_SEC    = 1.2    # secondes de silence pour terminer l'enregistrement
MIN_SPEECH     = 0.6    # durée minimale (filtre les bruits courts)
PRE_ROLL       = 15     # frames à conserver avant la détection

WHISPER_SIZE   = "large-v3"
KEYWORD        = "hermes"
CHAT_TIMEOUT   = 60     # secondes avant retour en veille (sans interaction)
CHAT_EXIT      = {"au revoir", "bonne nuit", "stop hermes", "fin"}

PIPER_BIN      = Path.home() / ".local/share/piper-runtime/piper"
PIPER_VOICE    = Path.home() / ".local/share/piper/fr_FR-upmc-medium.onnx"

MAX_HISTORY    = 10
HERMES_MODE    = "ssh"
HERMES_SSH     = "s01"

VOICE_INJECT   = "[Réponse vocale : courte, 1-2 phrases max, sans markdown ni listes] "

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

_history: list[dict] = []


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 _deaccent(s: str) -> str:
    return unicodedata.normalize("NFD", s).encode("ascii", "ignore").decode("ascii").lower()


def detect_keyword(text: str) -> tuple[bool, str]:
    words  = text.strip().split()
    low    = [_deaccent(w).strip(",.!?«»") for w in words]
    try:
        idx = next(i for i, w in enumerate(low) if KEYWORD in w)
        return True, " ".join(words[idx + 1:]).strip()
    except StopIteration:
        return False, ""


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=3,
        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:
    query = VOICE_INJECT + 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:
    global _history
    lower = _deaccent(text)
    if any(w in lower for w in ("reset", "nouveau contexte", "oublie tout", "efface")):
        _history.clear()
        return "Contexte effacé."
    _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": 200,
            "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() or not bin_path.exists():
        return
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        out = Path(f.name)
    try:
        subprocess.run(
            [str(bin_path), "--model", str(voice), "--output_file", str(out)],
            input=text.encode(),
            capture_output=True,
            check=True,
            env={"LD_LIBRARY_PATH": str(bin_path.parent)},
        )
        subprocess.run(["aplay", "-q", str(out)], check=False)
    except FileNotFoundError:
        pass
    finally:
        out.unlink(missing_ok=True)


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  ❌ {e}")
        return
    print(f"\n  🤖 Hermes : {response}\n")
    if not no_tts:
        speak(response)


def record_until_silence(audio_q: queue.Queue, vad, frame_samples: int, frame_ms: int) -> np.ndarray | None:
    """Enregistre depuis la queue jusqu'à SILENCE_SEC de silence webrtcvad."""
    required_silent = int(SILENCE_SEC * 1000 / frame_ms)
    captured        = []
    silent_frames   = 0

    while True:
        try:
            frame = audio_q.get(timeout=3.0)
        except queue.Empty:
            break
        captured.append(frame)
        try:
            is_speech = vad.is_speech(frame.tobytes(), SAMPLE_RATE)
        except Exception:
            is_speech = True
        if not is_speech:
            silent_frames += 1
        else:
            silent_frames = 0
        if silent_frames >= required_silent:
            break

    if not captured:
        return None
    audio = np.concatenate(captured)
    return audio if len(audio) / SAMPLE_RATE >= MIN_SPEECH else None


def run_vad(whisper_model, no_tts: bool):
    import webrtcvad

    vad           = webrtcvad.Vad(mode=3)
    FRAME_MS      = 20
    FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_MS / 1000)  # 320 samples

    chat_mode     = False
    last_activity = 0.0

    print("💤  Veille — dites \"Ok Hermès\" pour commencer (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=FRAME_SAMPLES, callback=callback):
        while True:
            # Timeout conversation
            if chat_mode and time.time() - last_activity > CHAT_TIMEOUT:
                chat_mode = False
                print("💤  Retour en veille — dites \"Ok Hermès\" pour reprendre\n")

            frame = audio_q.get()
            pre_roll.append(frame)

            try:
                is_speech = vad.is_speech(frame.tobytes(), SAMPLE_RATE)
            except Exception:
                continue

            if not is_speech:
                continue

            # Parole détectée — vider le pre-roll dans captured et continuer
            captured = list(pre_roll)
            # Drainer la queue le temps de silence
            audio = record_until_silence(audio_q, vad, FRAME_SAMPLES, FRAME_MS)
            if audio is None:
                continue
            captured_audio = np.concatenate([np.concatenate(captured), audio])

            text = transcribe(whisper_model, captured_audio)
            if not text:
                continue

            if chat_mode:
                # En conversation — mots de fin ?
                if any(w in _deaccent(text) for w in CHAT_EXIT):
                    chat_mode = False
                    print(f"\n  🗣️  Vous   : {text}")
                    beep(440)
                    speak("À bientôt !")
                    print("💤  Retour en veille\n")
                    continue
                # Envoyer directement
                print(f"\n  🗣️  Vous   : {text}")
                last_activity = time.time()
                _respond(text, no_tts)

            else:
                # En veille — chercher le wake word
                found, command = detect_keyword(text)
                if not found:
                    continue
                # Wake word confirmé
                chat_mode     = True
                last_activity = time.time()
                beep(880)
                print(f"\n  🗣️  Vous   : {text}")
                print(f"💬  Mode conversation ({CHAT_TIMEOUT}s) — parlez librement\n")
                if command:
                    _respond(command, no_tts)


def run_ptt(whisper_model, no_tts: bool):
    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=512, callback=callback):
            input()
            chunks = []
            while not audio_q.empty():
                chunks.append(audio_q.get_nowait())

        if not chunks:
            continue
        beep(440)
        audio = np.concatenate(chunks)
        if len(audio) / SAMPLE_RATE < MIN_SPEECH:
            continue
        print("  ⏳ Transcription...", end=" ", flush=True)
        text = transcribe(whisper_model, audio)
        if not text:
            print("(rien compris)")
            continue
        print(f"\n  🗣️  Vous   : {text}")
        _respond(text, 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)
    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()