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 _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
"""


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 (pipx reinstall) — sans désinstall/réinstall manuel."""
    import re
    import shutil
    import subprocess

    pipx = shutil.which("pipx")
    if not pipx:
        print("pipx introuvable — mise à jour automatique impossible.")
        print("Réinstalle manuellement depuis la source git.")
        return 1
    before = _installed_version()
    if before:
        print(f"Version installée : {before}")
    print("Récupération de la dernière version depuis git (pipx reinstall)…\n")
    res = subprocess.run([pipx, "reinstall", "stt"], capture_output=True, text=True)
    out = ((res.stdout or "") + (res.stderr or "")).strip()
    if out:
        print(out)
    m = re.search(r"installed package stt ([0-9][^\s,]*)", out)
    after = m.group(1) if m else None
    print()
    if after and before and after != before:
        print(f"✓ Mis à jour : {before} → {after}")
    elif after:
        print(f"✓ Déjà à jour ({after}).")
    else:
        print("✓ Réinstallation terminée — vérifie avec : stt --version")
    return res.returncode


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)
    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("--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.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())