Funk-lab/stt/client/stt/cli.py

"""Entrypoint de la commande `stt` (client)."""

from __future__ import annotations

import argparse
import asyncio
import os
import sys


def _cmd_setup() -> int:
    from stt.config import MODEL_PRESETS, write_default_config

    path = write_default_config()
    print(f"✓ Config : {path}")
    print("\nUtilisation :")
    print("  stt --text                 # chat texte simple (sans micro ni HUD)")
    print("  stt                        # voix + HUD")
    print(f"  stt --model claude         # modèles : {', '.join(MODEL_PRESETS)}")
    print("\nÀ configurer si besoin :")
    print(f"  • URL du STT-server dans {path} ([server] url)")
    print("  • Pour la voix : Piper + voix → ~/.local/share/piper/<voix>.onnx")
    return 0


SERVICE_NAME = "stt.service"


def _service_unit(stt_bin: str) -> str:
    """Contenu du unit systemd --user : lance `stt` (HUD + voix) avec la session graphique."""
    return f"""[Unit]
Description=STT — Assistant vocal Jarvis (HUD + voix)
After=graphical-session.target
PartOf=graphical-session.target

[Service]
Type=simple
ExecStart={stt_bin}
Restart=on-failure
RestartSec=5
Environment=PYTHONUNBUFFERED=1
StandardOutput=journal
StandardError=journal
SyslogIdentifier=stt

[Install]
WantedBy=graphical-session.target
"""


def _cmd_install_service() -> int:
    """Installe et démarre le service systemd --user (auto-start au login + kiosk)."""
    import shutil
    import subprocess
    from pathlib import Path

    stt_bin = shutil.which("stt") or os.path.realpath(sys.argv[0])
    unit_dir = Path.home() / ".config" / "systemd" / "user"
    unit_dir.mkdir(parents=True, exist_ok=True)
    unit_path = unit_dir / SERVICE_NAME
    unit_path.write_text(_service_unit(stt_bin))
    print(f"✓ Unit écrit : {unit_path}")
    print(f"  ExecStart = {stt_bin}")

    def sc(*args):
        subprocess.run(["systemctl", "--user", *args], check=False)

    # Rendre DISPLAY/WAYLAND dispo au gestionnaire systemd --user (pour le kiosk)
    subprocess.run(
        ["systemctl", "--user", "import-environment",
         "DISPLAY", "WAYLAND_DISPLAY", "XAUTHORITY"],
        check=False,
    )
    sc("daemon-reload")
    sc("enable", SERVICE_NAME)
    sc("start", SERVICE_NAME)

    print("\n✓ Service activé et démarré (systemd --user).")
    print("\nCommandes utiles :")
    print("  systemctl --user status stt")
    print("  journalctl --user -u stt -f")
    print("  systemctl --user restart stt")
    print("  stt --uninstall-service")
    print("\nDémarrer dès le boot (kiosk dédié, sans login manuel) :")
    print('  sudo loginctl enable-linger "$USER"   # + auto-login du bureau')
    print("\nLe HUD s'ouvre en kiosk si [ui].kiosk = true (défaut).")
    print("Si l'écran ne s'ouvre pas : voir le caveat DISPLAY dans admin/ia/stt.md.")
    return 0


def _cmd_uninstall_service() -> int:
    """Arrête, désactive et supprime le service systemd --user."""
    import subprocess
    from pathlib import Path

    def sc(*args):
        subprocess.run(["systemctl", "--user", *args], check=False)

    sc("stop", SERVICE_NAME)
    sc("disable", SERVICE_NAME)
    unit_path = Path.home() / ".config" / "systemd" / "user" / SERVICE_NAME
    if unit_path.exists():
        unit_path.unlink()
        print(f"✓ Unit supprimé : {unit_path}")
    else:
        print("(unit déjà absent)")
    sc("daemon-reload")
    print("✓ Service STT désinstallé.")
    return 0


def _service_installed() -> bool:
    """Le unit systemd --user stt.service est-il installé ?"""
    import subprocess

    return subprocess.run(
        ["systemctl", "--user", "cat", SERVICE_NAME],
        capture_output=True, text=True,
    ).returncode == 0


def _cmd_start() -> int:
    """Démarre le service STT (systemd --user)."""
    import subprocess

    if not _service_installed():
        print("Service non installé — lance d'abord : stt --install-service")
        print("(ou simplement `stt` pour un lancement direct, sans service)")
        return 1
    subprocess.run(["systemctl", "--user", "start", SERVICE_NAME], check=False)
    state = subprocess.run(
        ["systemctl", "--user", "is-active", SERVICE_NAME],
        capture_output=True, text=True,
    ).stdout.strip()
    print(f"✓ Service STT démarré (état : {state}).")
    return 0


def _cmd_restart() -> int:
    """Redémarre le service STT — recharge la config (modèle ASR, wake word…)."""
    import subprocess

    if not _service_installed():
        print("Service non installé — lance d'abord : stt --install-service")
        return 1
    # systemd possède le job : même si ce process est tué (cas restart depuis le
    # HUD, qui tourne DANS le service), le redémarrage va à son terme.
    subprocess.run(["systemctl", "--user", "restart", SERVICE_NAME], check=False)
    print("✓ Service STT redémarré.")
    return 0


def _running_stt_pids() -> list[int]:
    """PID des instances `stt` (service voix/HUD) en cours, hors la commande courante.

    Lit /proc (Linux) : on cible les process dont argv[0] se nomme `stt` et qui ne
    portent pas un flag utilitaire (--stop, --update, …) → uniquement le service vocal.
    """
    me = os.getpid()
    utilitaires = {
        "--stop", "--start", "--restart", "--update", "--version", "--setup", "--text",
        "--install-service", "--uninstall-service",
        "--install-desktop", "--uninstall-desktop",
    }
    pids: list[int] = []
    try:
        entries = os.listdir("/proc")
    except OSError:
        return pids
    for entry in entries:
        if not entry.isdigit():
            continue
        pid = int(entry)
        if pid == me:
            continue
        try:
            with open(f"/proc/{pid}/cmdline", "rb") as f:
                args = [a.decode("utf-8", "ignore") for a in f.read().split(b"\0") if a]
        except OSError:
            continue
        if not args:
            continue
        # Le chemin du bin `stt` peut être argv[0] (exec direct via shebang) ou un
        # token plus loin (lancement pipx : `python -E /…/bin/stt --window app` →
        # le `-E` décale le chemin). On cherche un token qui est un *chemin* vers
        # `stt` (slash + basename `stt`) — ça écarte le `stt` nu d'un
        # `journalctl -u stt` — et on ignore les sous-commandes utilitaires.
        is_stt = any("/" in a and os.path.basename(a) == "stt" for a in args)
        if is_stt and not (utilitaires & set(args)):
            pids.append(pid)
    return pids


def _cmd_stop() -> int:
    """Éteint le service STT : unit systemd --user en priorité, sinon le process vocal."""
    import signal
    import subprocess

    # 1) Service systemd --user (cas auto-start) — chemin propre
    active = subprocess.run(
        ["systemctl", "--user", "is-active", SERVICE_NAME],
        capture_output=True, text=True,
    ).stdout.strip()
    if active == "active":
        subprocess.run(["systemctl", "--user", "stop", SERVICE_NAME], check=False)
        print("✓ Service STT arrêté (systemctl --user stop stt).")
        return 0

    # 2) Lancé à la main (terminal / .desktop) → SIGTERM au(x) process vocal(aux)
    pids = _running_stt_pids()
    if not pids:
        print("Aucun service ni instance STT en cours.")
        return 0
    for pid in pids:
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            pass
    mot = "instance" if len(pids) == 1 else "instances"
    print(f"✓ {len(pids)} {mot} STT arrêtée(s) (PID {', '.join(map(str, pids))}).")
    return 0


def _desktop_entry(stt_bin: str, icon_path: str) -> str:
    """Contenu du lanceur .desktop : STT comme application de bureau (fenêtre app)."""
    return f"""[Desktop Entry]
Type=Application
Name=STT — Assistant vocal
GenericName=Assistant vocal
Comment=Assistant vocal Jarvis du homelab Funk
Exec={stt_bin} --window app
Icon={icon_path}
Terminal=false
Categories=Utility;AudioVideo;
Keywords=jarvis;assistant;voix;hermes;
StartupNotify=true
StartupWMClass=STT-Funk
"""


def _cmd_install_desktop() -> int:
    """Crée l'entrée de menu (.desktop) + icône → STT lançable comme une appli (type Discord)."""
    import shutil
    import subprocess
    from pathlib import Path

    stt_bin = shutil.which("stt") or os.path.realpath(sys.argv[0])
    icon_path = Path(__file__).resolve().parent / "hud" / "icon.svg"
    apps_dir = Path.home() / ".local" / "share" / "applications"
    apps_dir.mkdir(parents=True, exist_ok=True)
    entry = apps_dir / "stt.desktop"
    entry.write_text(_desktop_entry(stt_bin, str(icon_path)))
    entry.chmod(0o755)
    subprocess.run(["update-desktop-database", str(apps_dir)], check=False)

    print(f"✓ Lanceur : {entry}")
    print(f"  Icône    : {icon_path}")
    print(f"  Exec     : {stt_bin} --window app")
    print("\nSTT apparaît dans le menu des applications (épinglable au dock).")
    print("Il s'ouvre en fenêtre « app » (sans onglets), comme Discord.")
    print("Retirer : stt --uninstall-desktop")
    return 0


def _cmd_uninstall_desktop() -> int:
    """Supprime le lanceur de bureau."""
    import subprocess
    from pathlib import Path

    entry = Path.home() / ".local" / "share" / "applications" / "stt.desktop"
    if entry.exists():
        entry.unlink()
        print(f"✓ Lanceur supprimé : {entry}")
    else:
        print("(lanceur déjà absent)")
    subprocess.run(["update-desktop-database", str(entry.parent)], check=False)
    return 0


def _installed_version() -> str | None:
    from importlib.metadata import PackageNotFoundError, version

    try:
        return version("stt")
    except PackageNotFoundError:
        return None


def _cmd_version() -> int:
    print(_installed_version() or "version inconnue (pas installé via pip/pipx)")
    return 0


def _cmd_update() -> int:
    """Met à jour STT depuis sa source git, sans désinstall/réinstall manuel.

    Utilise `pipx upgrade` (non destructif : conserve l'install actuelle si l'upgrade échoue),
    contrairement à `pipx reinstall` qui désinstalle d'abord et peut laisser sans rien.
    """
    import json
    import shutil
    import subprocess

    pipx = shutil.which("pipx")
    if not pipx:
        print("pipx introuvable — mise à jour automatique impossible.")
        return 1

    # Source d'installation (pour la commande de secours en cas d'échec)
    spec = None
    try:
        listing = subprocess.run(
            [pipx, "list", "--json"], capture_output=True, text=True
        ).stdout or "{}"
        spec = json.loads(listing)["venvs"]["stt"]["metadata"]["main_package"]["package_or_url"]
    except Exception:  # noqa: BLE001
        pass

    before = _installed_version()
    if before:
        print(f"Version installée : {before}")
    print("Mise à jour depuis git (pipx upgrade)…\n")
    res = subprocess.run([pipx, "upgrade", "stt"], capture_output=True, text=True)
    out = ((res.stdout or "") + (res.stderr or "")).strip()
    if out:
        print(out)

    print()
    if res.returncode != 0 or shutil.which("stt") is None:
        print("❌ Échec de la mise à jour — ton install précédente est conservée.")
        print("   Réessaie (souvent réseau/SSH), ou en dernier recours :")
        print("     uv cache clean")
        if spec:
            print(f'     pipx install --force "{spec}"')
        return res.returncode or 1
    print("✓ Mise à jour terminée — vérifie avec : stt --version")
    return 0


def _make_client(args):
    from stt.api import ServerClient
    from stt.config import load_config, resolve_model

    cfg = load_config()
    if args.server:
        cfg["server"]["url"] = args.server
    client = ServerClient(cfg["server"])
    if args.model:
        client.model = resolve_model(args.model)
    # Jeton Ghostfolio (source unique côté client) → transmis au serveur par requête
    gf_token = (cfg.get("ghostfolio") or {}).get("access_token")
    if gf_token:
        client.secrets["ghostfolio_token"] = gf_token
    return cfg, client


def _cmd_text(args) -> int:
    """Chat texte simple : tape, le serveur répond. Pas de micro ni HUD."""
    _, client = _make_client(args)
    print(f"STT (texte) — serveur {client.url} — modèle '{client.model}'")
    if not client.healthy():
        print("  ⚠️  serveur injoignable — vérifie [server] url et le réseau")
    print("  commandes : /model <nom>, /models, /reset, /quit\n")

    from stt.config import resolve_model

    while True:
        try:
            line = input("vous> ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n👋")
            return 0
        if not line:
            continue
        if line in ("/quit", "/exit", "/q"):
            return 0
        if line == "/reset":
            client.reset()
            print("  → mémoire de session effacée\n")
            continue
        if line == "/models":
            try:
                m = client.models()
                print(f"  défaut: {m['default']} — dispo: {', '.join(m['available'])}\n")
            except Exception as e:  # noqa: BLE001
                print(f"  ❌ {e}\n")
            continue
        if line.startswith("/model "):
            client.model = resolve_model(line.split(maxsplit=1)[1].strip())
            print(f"  → modèle : {client.model}\n")
            continue
        try:
            reply = client.ask(line)
        except Exception as e:  # noqa: BLE001
            print(f"  ❌ {e}\n")
            continue
        print(f"hermes [{client.model}]> {reply}\n")


def _cmd_voice(args) -> int:
    from stt.config import load_config  # noqa: F401  (cfg via _make_client)
    from stt.memory import LocalMemory
    from stt.ui import serve
    from stt.voice import VoiceEngine

    cfg, client = _make_client(args)
    # mode d'ouverture du HUD : --window > config window_mode > kiosk (compat) > none
    ui = cfg["ui"]
    ui["window_mode"] = (
        args.window or ui.get("window_mode") or ("kiosk" if ui.get("kiosk") else "none")
    )
    print(f"STT-server : {client.url} — modèle '{client.model}'")
    if not client.healthy():
        print("  ⚠️  serveur injoignable — vérifie [server] url et le réseau")

    memory = None
    if cfg["memory"].get("enabled"):
        memory = LocalMemory(cfg["memory"]["local_db"])

    def make_engine(emit):
        return VoiceEngine(
            cfg["voice"], client.ask, emit, memory=memory, no_tts=args.no_tts,
            model_label=lambda: client.model,
        )

    try:
        asyncio.run(serve(cfg, make_engine, client))
    except KeyboardInterrupt:
        print("\n👋 Au revoir")
    finally:
        if memory:
            memory.close()
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="stt", description="Client vocal Jarvis de Funk")
    p.add_argument("--setup", action="store_true", help="Écrit la config par défaut")
    p.add_argument("--text", action="store_true", help="Mode chat texte (sans micro ni HUD)")
    p.add_argument("--model", help="Modèle : hermes, claude, qwen, opus (ou alias LiteLLM)")
    p.add_argument("--server", help="Force l'URL du STT-server (sinon config)")
    p.add_argument("--no-tts", action="store_true", help="Réponses texte seulement (voix)")
    p.add_argument("--window", choices=["app", "kiosk", "none"],
                   help="Ouverture du HUD : app (fenêtre, type Discord) | kiosk (plein écran) | none")
    p.add_argument("--start", action="store_true",
                   help="Démarre le service STT (systemd --user)")
    p.add_argument("--restart", action="store_true",
                   help="Redémarre le service STT (recharge la config : ASR, wake word…)")
    p.add_argument("--stop", action="store_true",
                   help="Éteint le service STT en cours (systemd --user ou process vocal)")
    p.add_argument("--install-service", action="store_true",
                   help="Installe le service systemd --user (auto-start au login + kiosk)")
    p.add_argument("--uninstall-service", action="store_true",
                   help="Désinstalle le service systemd --user")
    p.add_argument("--install-desktop", action="store_true",
                   help="Crée le lanceur de bureau (.desktop + icône) — appli type Discord")
    p.add_argument("--uninstall-desktop", action="store_true",
                   help="Supprime le lanceur de bureau")
    p.add_argument("--update", action="store_true",
                   help="Met à jour STT depuis git (pipx reinstall)")
    p.add_argument("--version", action="store_true", help="Affiche la version installée")
    args = p.parse_args(argv)

    if args.version:
        return _cmd_version()
    if args.setup:
        return _cmd_setup()
    if args.update:
        return _cmd_update()
    if args.start:
        return _cmd_start()
    if args.restart:
        return _cmd_restart()
    if args.stop:
        return _cmd_stop()
    if args.install_service:
        return _cmd_install_service()
    if args.uninstall_service:
        return _cmd_uninstall_service()
    if args.install_desktop:
        return _cmd_install_desktop()
    if args.uninstall_desktop:
        return _cmd_uninstall_desktop()
    if args.text:
        return _cmd_text(args)
    return _cmd_voice(args)


if __name__ == "__main__":
    sys.exit(main())