konan-ai-assistant/konan_voice.py

#!/usr/bin/env python3
"""
Konan Voice — interface vocale directe sur le serveur.
Parlez naturellement, Konan répond à voix haute.
Ctrl+C pour quitter.
"""
import warnings
warnings.filterwarnings("ignore")
import logging
logging.getLogger("whisper").setLevel(logging.ERROR)

import sounddevice as sd
import numpy as np
import whisper
import requests
import subprocess
import tempfile
import os
import scipy.io.wavfile as wav

SAMPLE_RATE    = 16000
CHUNK_SIZE     = int(SAMPLE_RATE * 0.3)   # 300 ms par chunk
SILENCE_THRESH = 1200                      # seuil énergie RMS (calibré au bruit ambiant)
SILENCE_DUR    = 1.5                       # secondes de silence pour arrêter
MIN_SPEECH_SEC = 0.5                       # durée min pour traiter

KONAN_API  = "http://localhost:8765/api/chat"
USER_ID    = "voice_local"
PIPER_BIN  = os.path.expanduser("~/agent/venv/bin/piper")
VOICE_MODEL = os.path.expanduser("~/agent/voices/fr_FR-upmc-medium.onnx")

def rms(chunk):
    return np.sqrt(np.mean(chunk.astype(np.float32) ** 2))

def speak(text):
    print(f"\nKonan : {text}\n")
    try:
        proc = subprocess.Popen(
            [PIPER_BIN, "--model", VOICE_MODEL, "--output-raw"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
        )
        raw, _ = proc.communicate(input=text.encode())
        audio = np.frombuffer(raw, dtype=np.int16)
        sd.play(audio, samplerate=22050)
        sd.wait()
    except Exception as e:
        print(f"[TTS erreur] {e}")

def record_utterance():
    silence_chunks_needed = int(SILENCE_DUR / 0.3)
    recorded = []
    silence_count = 0
    speech_started = False

    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype="int16") as stream:
        while True:
            chunk, _ = stream.read(CHUNK_SIZE)
            energy = rms(chunk)
            if energy > SILENCE_THRESH:
                if not speech_started:
                    print("🎙️  Écoute...")
                    speech_started = True
                recorded.append(chunk.copy())
                silence_count = 0
            elif speech_started:
                recorded.append(chunk.copy())
                silence_count += 1
                if silence_count >= silence_chunks_needed:
                    break

    if not speech_started or len(recorded) < int(MIN_SPEECH_SEC / 0.3):
        return None
    return np.concatenate(recorded, axis=0).flatten()

def transcribe(audio):
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        wav.write(f.name, SAMPLE_RATE, audio)
        result = model.transcribe(f.name, language="fr")
        os.unlink(f.name)
    return result["text"].strip()

def ask_konan(text):
    try:
        r = requests.post(KONAN_API, json={"text": text, "user_id": USER_ID}, timeout=120)
        return r.json().get("response", "Pas de réponse.")
    except Exception as e:
        return f"Erreur connexion au bot : {e}"

print("Chargement de Whisper tiny...")
model = whisper.load_model("tiny")
print("Whisper prêt.\n")

speak("Konan est prêt. Je vous écoute.")
print("▶  Parlez naturellement. Ctrl+C pour quitter.\n")

try:
    while True:
        audio = record_utterance()
        if audio is None:
            continue
        print("Transcription...")
        text = transcribe(audio)
        if not text:
            continue
        print(f"Vous : {text}")
        print("Konan réfléchit...")
        response = ask_konan(text)
        speak(response)
except KeyboardInterrupt:
    speak("À bientôt.")
    print("\nAu revoir.")