Funk-lab/stt/client/stt/ui/app.py

"""Serveur asyncio : sert le HUD (HTTP statique) et diffuse les états (websocket).

Le moteur vocal tourne dans un thread (audio bloquant) et pousse ses événements via
`emit()`, relayés sur la boucle asyncio puis diffusés à tous les clients HUD connectés.
"""

from __future__ import annotations

import asyncio
import json
import threading
import time
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

# Le HUD est embarqué DANS le package (stt/hud) pour être livré par pipx.
HUD_DIR = Path(__file__).resolve().parent.parent / "hud"


class _HudHandler(SimpleHTTPRequestHandler):
    """Sert le HUD en interdisant le cache navigateur.

    Sans ça, le navigateur garde l'ancien index.html après une mise à jour
    (symptôme classique : « je vois encore l'ancienne version »).
    """

    def end_headers(self):
        self.send_header("Cache-Control", "no-store, max-age=0")
        self.send_header("Pragma", "no-cache")
        super().end_headers()

    def log_message(self, *args):  # silence le log HTTP par requête
        pass


def _start_http(host: str, port: int) -> ThreadingHTTPServer:
    handler = partial(_HudHandler, directory=str(HUD_DIR))
    httpd = ThreadingHTTPServer((host, port), handler)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd


async def serve(cfg: dict[str, Any], make_engine, client) -> None:
    """Sert le HUD + diffuse les états ; applique les réglages renvoyés par le HUD.

    make_engine(emit) -> VoiceEngine. emit est thread-safe (depuis le thread voix).
    client : ServerClient — muté à chaud par les réglages (modèle, reset).
    """
    import websockets

    ui = cfg["ui"]
    loop = asyncio.get_running_loop()
    clients: set = set()
    last_state = {"type": "state", "state": "idle"}

    def broadcast(event: dict) -> None:
        nonlocal last_state
        if event.get("type") == "state":
            last_state = event
        data = json.dumps(event, ensure_ascii=False)
        for ws in list(clients):
            asyncio.create_task(_safe_send(ws, data))

    async def _safe_send(ws, data: str) -> None:
        try:
            await ws.send(data)
        except Exception:
            clients.discard(ws)

    # emit appelé depuis le thread voix → on repasse sur la boucle asyncio
    def emit(event: dict) -> None:
        loop.call_soon_threadsafe(broadcast, event)

    async def handler(ws):
        clients.add(ws)
        try:
            await ws.send(json.dumps(last_state, ensure_ascii=False))
            # Le HUD renvoie {"type":"settings","data":{...}} quand on change un réglage.
            async for raw in ws:
                try:
                    msg = json.loads(raw)
                except (ValueError, TypeError):
                    continue
                if isinstance(msg, dict) and msg.get("type") == "settings":
                    _apply_settings(client, engine, msg.get("data") or {}, loop)
        finally:
            clients.discard(ws)

    _start_http(ui["ws_host"], ui["http_port"])

    engine = make_engine(emit)
    # chargement Whisper + boucle audio dans un thread dédié
    threading.Thread(target=_run_engine, args=(engine, emit), daemon=True).start()

    async with websockets.serve(handler, ui["ws_host"], ui["ws_port"]):
        url = f"http://{ui['ws_host']}:{ui['http_port']}/"
        print(f"  HUD : {url}")
        mode = ui.get("window_mode") or ("kiosk" if ui.get("kiosk") else "none")
        if mode != "none":
            # ?v=<timestamp> : URL unique à chaque lancement → le navigateur ne peut
            # pas servir une version cachée du HUD (« ancien HUD persistant »).
            _open_browser(f"{url}?v={int(time.time())}", mode)
        await asyncio.Future()  # tourne indéfiniment


def _run_engine(engine, emit) -> None:
    try:
        engine.load()
        engine.run()
    except Exception as e:  # noqa: BLE001
        emit({"type": "error", "text": f"moteur vocal : {e}"})


def _open_browser(url: str, mode: str) -> None:
    """Ouvre le HUD comme une fenêtre d'application dédiée (pas la session Brave courante).

    On force un profil + une WM class propres (`--class=STT-Funk`, `--user-data-dir`) → la
    fenêtre est une instance séparée, avec sa propre icône / entrée de barre des tâches
    (associée au lanceur .desktop via `StartupWMClass`), au lieu de se fondre dans Brave.
    mode : 'app' (fenêtre chromeless type Discord) | 'kiosk' (plein écran).
    Cross-distro : binaire natif (rpm/deb) → Flatpak → repli navigateur par défaut.
    """
    import shutil
    import subprocess
    import webbrowser
    from pathlib import Path

    profile = Path.home() / ".local" / "share" / "stt" / "app-profile"
    profile.mkdir(parents=True, exist_ok=True)
    win = ["--kiosk", url] if mode == "kiosk" else [f"--app={url}"]
    # Instance autonome (profil + classe dédiés) — pour les navigateurs natifs.
    native = [
        "--class=STT-Funk", f"--user-data-dir={profile}",
        "--no-first-run", "--no-default-browser-check", *win,
    ]
    # Flatpak est déjà sandboxé : pas de --user-data-dir (chemin hôte inaccessible).
    flatpak = ["--class=STT-Funk", *win]

    # 1) Binaire natif (rpm/deb)
    for browser in (
        "brave-browser", "brave",
        "chromium", "chromium-browser",
        "google-chrome", "google-chrome-stable",
        "microsoft-edge", "vivaldi",
    ):
        if shutil.which(browser):
            subprocess.Popen([browser, *native])
            return

    # 2) Flatpak (universel — souvent la voie sur Fedora/AlmaLinux)
    if shutil.which("flatpak"):
        for appid in ("com.brave.Browser", "org.chromium.Chromium",
                      "com.google.Chrome", "com.microsoft.Edge"):
            installed = subprocess.run(
                ["flatpak", "info", appid], capture_output=True
            ).returncode == 0
            if installed:
                subprocess.Popen(["flatpak", "run", appid, *flatpak])
                return

    # 3) Repli : onglet dans le navigateur par défaut
    webbrowser.open(url)


def _apply_settings(client, engine, data: dict, loop) -> None:
    """Applique les réglages renvoyés par le HUD — best-effort, jamais bloquant ni fatal.

    Les réglages purement visuels (intensité d'accent, avatar) sont gérés côté
    navigateur (localStorage). Ici on n'applique que ce qui touche le backend :
      • action "reset" → vide la mémoire de session côté STT-server
      • brainMode      → bascule le modèle (local = Qwen gratuit / advanced = Claude)
      • wakeword       → mot de réveil, relu à chaud par le moteur vocal
    La voix Piper et le no-think ne sont pas applicables à chaud (ignorés ici).
    """
    if not isinstance(data, dict):
        return
    try:
        if data.get("action") == "reset":
            loop.run_in_executor(None, client.reset)  # appel HTTP → hors boucle
            return
        brain = data.get("brainMode")
        if brain in ("local", "advanced"):
            from stt.config import resolve_model

            client.model = resolve_model("hermes" if brain == "local" else "claude")
        wake = (data.get("wakeword") or "").strip()
        if wake and engine is not None:
            from stt.voice.engine import _deaccent

            parts = wake.split()
            engine.cfg["wake_word"] = _deaccent(parts[-1] if parts else wake)
    except Exception:  # noqa: BLE001 — un réglage invalide ne doit jamais tuer la connexion
        pass