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

    from stt.portal import StatusPoller, load_services

    ui = cfg["ui"]
    loop = asyncio.get_running_loop()
    clients: set = set()
    last_state = {"type": "state", "state": "idle"}
    last_mic = {"type": "mic", "muted": False}
    last_status: dict | None = None  # dernier portal-status (santé des services)

    # Portail : registre des services (HUD + voix). Construit une fois ; le HUD le
    # reçoit à la connexion et affiche une tuile par service.
    services = load_services(cfg)
    services_by_id = {s.id: s for s in services}
    services_msg = {
        "type": "services",
        "services": [
            {"id": s.id, "name": s.name, "url": s.url, "icon": s.icon,
             "description": s.description}
            for s in services
        ],
    }

    # Contextes présélectionnables (récupérés du serveur ; None = serveur antérieur).
    # Appel HTTP bloquant → hors boucle. Le HUD reçoit la liste à la connexion.
    contexts_info = await loop.run_in_executor(None, client.contexts)
    contexts_msg = {
        "type": "contexts",
        "contexts": (contexts_info or {}).get("contexts", []),
        "default": (contexts_info or {}).get("default"),
        "current": client.context or (contexts_info or {}).get("default"),
    }

    # Version installée (affichée près du bouton « Mettre à jour » du HUD)
    from importlib.metadata import PackageNotFoundError
    from importlib.metadata import version as _pkg_version

    try:
        _stt_version = _pkg_version("stt")
    except PackageNotFoundError:
        _stt_version = "?"
    meta_msg = {"type": "meta", "version": _stt_version}

    def broadcast(event: dict) -> None:
        nonlocal last_state, last_mic, last_status
        if event.get("type") == "state":
            last_state = event
        elif event.get("type") == "mic":
            last_mic = event
        elif event.get("type") == "portal-status":
            last_status = 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)

    def open_service(sid: str) -> None:
        """Ouvre l'URL d'un service dans le navigateur NORMAL + confirme via toast.

        Best-effort : un service inconnu ou un navigateur absent ne casse rien.
        """
        svc = services_by_id.get(sid)
        if svc is None:
            return
        # ouverture bloquante (xdg-open / Popen) → hors boucle asyncio
        loop.run_in_executor(None, _open_url_external, svc.url)
        emit({"type": "toast", "text": f"{svc.icon} Ouverture de {svc.name}"})

    # Binaire `stt` résolu une fois (PATH du service --user ≠ shell interactif)
    import shutil as _shutil
    import sys as _sys

    stt_bin = _shutil.which("stt") or _sys.argv[0]

    def run_service_action(action: str) -> None:
        """Pilote le service depuis le HUD : restart / stop / update.

        Lance `stt --<action>` en processus DÉTACHÉ : pour restart/stop, le process
        courant (le HUD = le service) meurt, mais systemd possède le job et le mène à
        terme. Pour update : `stt --update` puis `stt --restart` (la maj précède le
        redémarrage, sinon on se couperait l'herbe sous le pied).
        """
        labels = {
            "service-restart": "↻ Redémarrage du service…",
            "service-stop": "⏻ Arrêt du service…",
            "service-update": "⬆ Mise à jour en cours (peut prendre un moment)…",
        }
        if action not in labels:
            return
        emit({"type": "toast", "text": labels[action]})
        if action == "service-update":
            argv = ["sh", "-c", f'"{stt_bin}" --update && "{stt_bin}" --restart']
        elif action == "service-stop":
            argv = [stt_bin, "--stop"]
        else:
            argv = [stt_bin, "--restart"]
        loop.run_in_executor(None, _spawn_detached, argv)

    async def handler(ws):
        clients.add(ws)
        try:
            await ws.send(json.dumps(last_state, ensure_ascii=False))
            await ws.send(json.dumps(last_mic, ensure_ascii=False))  # état micro courant
            await ws.send(json.dumps(services_msg, ensure_ascii=False))  # tuiles du portail
            await ws.send(json.dumps(contexts_msg, ensure_ascii=False))  # contextes Asa
            await ws.send(json.dumps(meta_msg, ensure_ascii=False))  # version installée
            if last_status is not None:  # santé connue → l'envoyer tout de suite
                await ws.send(json.dumps(last_status, ensure_ascii=False))
            # Le HUD envoie {"type":"settings","data":{...}} (réglage),
            # {"type":"text","text":"…"} (message tapé) ou
            # {"type":"control","action":"stop|mute|unmute"} (boutons HUD).
            async for raw in ws:
                try:
                    msg = json.loads(raw)
                except (ValueError, TypeError):
                    continue
                if not isinstance(msg, dict):
                    continue
                mtype = msg.get("type")
                if mtype == "settings":
                    _apply_settings(client, engine, msg.get("data") or {}, loop)
                elif mtype == "text":
                    txt = (msg.get("text") or "").strip()
                    if txt and engine is not None:
                        # responder = appel HTTP bloquant → hors boucle asyncio
                        loop.run_in_executor(None, engine.respond_text, txt)
                elif mtype == "control" and msg.get("action") == "open-service":
                    # indépendant du moteur vocal → cliquable même sans micro
                    open_service(str(msg.get("service") or ""))
                elif mtype == "control" and msg.get("action") in (
                    "service-restart", "service-stop", "service-update"
                ):
                    run_service_action(str(msg.get("action")))
                elif mtype == "control" and engine is not None:
                    action = msg.get("action")
                    if action == "stop":
                        # peut tuer un sous-process (aplay) → hors boucle asyncio
                        loop.run_in_executor(None, engine.stop_response)
                    elif action == "mute":
                        engine.set_muted(True)
                    elif action == "unmute":
                        engine.set_muted(False)
                    elif action == "toggle-mute":
                        engine.set_muted(not engine.muted)
        finally:
            clients.discard(ws)

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

    engine = make_engine(emit)
    # le moteur expose au HUD le dernier contexte assemblé renvoyé par le serveur
    engine.context_provider = lambda: client.last_context

    # Intentions vocales locales (lecture seule) : « ouvre X », « état du cluster »,
    # « combien sur ghostfolio ». Court-circuitent le LLM. Ghostfolio passe par le
    # serveur (jeton client transmis) → une seule source de jeton, un seul code API.
    from stt.portal import route_intent, summarize_status

    engine.intent_router = lambda text: route_intent(
        text,
        services=services,
        open_fn=open_service,
        status_fn=lambda: summarize_status(last_status, services),
        ghostfolio_fn=client.portfolio,
    )

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

    # Santé live du portail : thread de sondage (HTTP + Prometheus + Alertmanager)
    portal_cfg = cfg.get("portal") or {}
    if portal_cfg.get("status_enabled", True) and services:
        StatusPoller(services, portal_cfg, emit).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}"]
    # --log-level=3 : tait le bruit Chromium (viz/vaapi/RADV) à la source ;
    # les flux sont en plus redirigés vers /dev/null (voir _spawn) pour ne pas
    # polluer le terminal qui lance `stt`.
    quiet = ["--log-level=3"]
    # 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", *quiet, *win,
    ]
    # Flatpak est déjà sandboxé : pas de --user-data-dir (chemin hôte inaccessible).
    flatpak = ["--class=STT-Funk", *quiet, *win]

    def _spawn(argv: list[str]) -> None:
        # Détaché et muet : la sortie du navigateur ne remonte pas dans `stt`.
        subprocess.Popen(
            argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
        )

    # 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):
            _spawn([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:
                _spawn(["flatpak", "run", appid, *flatpak])
                return

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


def _spawn_detached(argv: list[str]) -> None:
    """Lance une commande détachée (nouvelle session) et muette — best-effort.

    `start_new_session=True` : la commande survit à la mort du HUD (cas restart/stop
    déclenché depuis l'interface, où le process courant est lui-même arrêté).
    """
    import subprocess

    try:
        subprocess.Popen(
            argv,
            start_new_session=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except Exception:  # noqa: BLE001 — action best-effort, jamais fatale
        pass


def _open_url_external(url: str) -> None:
    """Ouvre `url` dans le navigateur HABITUEL de l'utilisateur (onglet normal).

    Contrairement à `_open_browser` (fenêtre app dédiée du HUD, profil isolé), ici on
    veut la session courante — favoris, connexions, onglets. `xdg-open` respecte le
    navigateur par défaut du bureau ; repli sur `webbrowser` si absent. Best-effort.
    """
    import shutil
    import subprocess
    import webbrowser

    try:
        if shutil.which("xdg-open"):
            subprocess.Popen(
                ["xdg-open", url],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
            return
    except Exception:  # noqa: BLE001 — on retombe sur webbrowser
        pass
    try:
        webbrowser.open(url, new=2)  # new=2 : nouvel onglet
    except Exception:  # noqa: BLE001 — ouverture best-effort, jamais fatale
        pass


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")
        ctx = (data.get("context") or "").strip()
        if ctx:
            client.context = ctx  # contexte présélectionné, envoyé au prochain /v1/ask
        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)
        vol = data.get("volume")
        if vol is not None and engine is not None:
            # gain TTS appliqué à chaud (0.0 = muet … 2.0 = amplifié). Relu par engine._speak.
            engine.cfg["tts_volume"] = max(0.0, min(2.0, float(vol)))
    except Exception:  # noqa: BLE001 — un réglage invalide ne doit jamais tuer la connexion
        pass