Trixy7/main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Trixy Voice Assistant - Haupteinstiegspunkt.

Unterstützt folgende Modi:
- server:     Zentraler Server für Satellite-Management und ML-Inferenz
- client:     Leichtgewichtiger Satellite mit Wakeword-Erkennung
- standalone: All-in-One-Modus ohne Server
- config:     Remote-Verwaltung per TUI (verbindet sich mit laufender Instanz)
- train:      ML-Trainer für Wakeword- und Sprechererkennungsmodelle

Verwendung:
    python3 main.py server [--debug] [--config <pfad>]
    python3 main.py client [--host <ip>] [--port <port>] [--room <raum>] [--alias <name>] [--debug]
    python3 main.py standalone [--debug] [--config <pfad>]
    python3 main.py config [--host <host>] [--port <port>] [--key <pfad>]
    python3 main.py train [wakeword|voice-recognition|all] [--debug] [--config <pfad>]
"""

import argparse
import asyncio
import os
import sys
from pathlib import Path

# Füge das aktuelle Verzeichnis zum Pfad hinzu
sys.path.insert(0, str(Path(__file__).parent))

# CUDA-Bibliotheken aus den nvidia Pip-Packages fuer onnxruntime-gpu verfuegbar machen
_venv_root = Path(__file__).resolve().parents[1]
_nvidia_base = _venv_root / "lib" / f"python{sys.version_info.major}.{sys.version_info.minor}" / "site-packages" / "nvidia"
if _nvidia_base.is_dir():
    _nvidia_lib_dirs = [str(p) for p in _nvidia_base.glob("*/lib") if p.is_dir()]
    if _nvidia_lib_dirs:
        # LD_LIBRARY_PATH fuer Subprocesses (spawn)
        _existing = os.environ.get("LD_LIBRARY_PATH", "")
        os.environ["LD_LIBRARY_PATH"] = ":".join(_nvidia_lib_dirs) + (":" + _existing if _existing else "")
        # Libs vorab laden fuer den aktuellen Prozess (onnxruntime-gpu)
        import ctypes
        for _lib_dir in _nvidia_lib_dirs:
            for _so in sorted(Path(_lib_dir).glob("*.so.*")):
                try:
                    ctypes.cdll.LoadLibrary(str(_so))
                except OSError:
                    pass

from trixy_core.utils.version import VERSION_STRING
from trixy_core.utils.debug import pinfo, perror


def create_parser() -> argparse.ArgumentParser:
    """Erstellt den Argument-Parser."""
    parser = argparse.ArgumentParser(
        prog="trixy",
        description="Trixy Voice Assistant System",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Beispiele:
  %(prog)s server --debug
  %(prog)s client --host 192.168.1.100 --port 2101 --room wohnzimmer --alias "Echo 1"
  %(prog)s standalone --debug
  %(prog)s train wakeword
        """
    )

    parser.add_argument(
        "--version",
        action="version",
        version=f"Trixy v{VERSION_STRING}"
    )

    # Subparser für Modi
    subparsers = parser.add_subparsers(
        dest="mode",
        title="Modi",
        description="Verfügbare Betriebsmodi",
        required=True
    )

    # Server-Modus
    server_parser = subparsers.add_parser(
        "server",
        help="Server-Modus starten"
    )
    server_parser.add_argument(
        "--debug",
        action="store_true",
        help="Debug-Modus aktivieren (stdout statt TUI)"
    )
    server_parser.add_argument(
        "--config",
        type=str,
        default="config/server_config.json",
        help="Pfad zur Konfigurationsdatei"
    )
    server_parser.add_argument(
        "--auto-regist",
        action="store_true",
        help="Dauerhafter Auto-Registrierungsmodus für neue Satellites"
    )

    # Client-Modus
    client_parser = subparsers.add_parser(
        "client",
        help="Client/Satellite-Modus starten"
    )
    client_parser.add_argument(
        "--host",
        type=str,
        default=None,
        help="Server-IP-Adresse (Standard: aus Konfigurationsdatei)"
    )
    client_parser.add_argument(
        "--port",
        type=int,
        default=None,
        help="Server-Port (Standard: aus Konfigurationsdatei, fallback 2101)"
    )
    client_parser.add_argument(
        "--room",
        type=str,
        default=None,
        help="Raum-Kennung (Standard: aus Konfigurationsdatei)"
    )
    client_parser.add_argument(
        "--alias",
        type=str,
        default=None,
        help="Satellite-Name (Standard: aus Konfigurationsdatei)"
    )
    client_parser.add_argument(
        "--debug",
        action="store_true",
        help="Debug-Modus aktivieren"
    )
    client_parser.add_argument(
        "--config",
        type=str,
        default="config/client_config.json",
        help="Pfad zur Konfigurationsdatei"
    )
    client_parser.add_argument(
        "-i", "--keyboard-input",
        action="store_true",
        help="Keyboard-Eingabe aktivieren (Text statt Sprache)"
    )

    # Standalone-Modus
    standalone_parser = subparsers.add_parser(
        "standalone",
        help="Standalone-Modus starten (Server + Client kombiniert)"
    )
    standalone_parser.add_argument(
        "--debug",
        action="store_true",
        help="Debug-Modus aktivieren"
    )
    standalone_parser.add_argument(
        "--config",
        type=str,
        default="config/standalone_config.json",
        help="Pfad zur Konfigurationsdatei"
    )
    standalone_parser.add_argument(
        "-i", "--keyboard-input",
        action="store_true",
        help="Keyboard-Eingabe aktivieren (Text statt Sprache)"
    )

    # Config-Tool-Modus
    config_parser = subparsers.add_parser(
        "config",
        help="Config-Tool starten (Remote-Verwaltung per TUI)"
    )
    config_parser.add_argument(
        "--host",
        type=str,
        default="localhost",
        help="Host der Trixy-Instanz (Standard: localhost)"
    )
    config_parser.add_argument(
        "--port",
        type=int,
        default=2105,
        help="Config-Port der Trixy-Instanz (Standard: 2105)"
    )
    config_parser.add_argument(
        "--key",
        type=str,
        default="certs/encryption.key",
        help="Pfad zum Verschluesselungsschluessel"
    )

    # Trainer-Modus
    train_parser = subparsers.add_parser(
        "train",
        help="ML-Trainer starten"
    )
    train_parser.add_argument(
        "target",
        nargs="?",
        choices=["wakeword", "voice-recognition", "all"],
        default="all",
        help="Zu trainierendes Modell (Standard: all)"
    )
    train_parser.add_argument(
        "--debug",
        action="store_true",
        help="Debug-Modus aktivieren"
    )
    train_parser.add_argument(
        "--config",
        type=str,
        default="config/trainer_config.json",
        help="Pfad zur Konfigurationsdatei"
    )

    return parser


async def run_server(args: argparse.Namespace) -> int:
    """Startet den Server-Modus."""
    from trixy_core.server import ServerApplication

    app = ServerApplication(
        config_path=args.config,
        debug=args.debug,
        auto_regist=getattr(args, "auto_regist", False)
    )

    try:
        await app.run()
        return 0
    except KeyboardInterrupt:
        pinfo("Beende auf Benutzeranfrage...")
        return 0
    except Exception as e:
        perror(f"Server-Fehler: {e}")
        return 1


async def run_client(args: argparse.Namespace) -> int:
    """Startet den Client-Modus."""
    from trixy_core.client import ClientApplication

    app = ClientApplication(
        host=args.host,
        port=args.port,
        room=args.room,
        alias=args.alias,
        config_path=args.config,
        debug=args.debug,
        keyboard_input=getattr(args, "keyboard_input", False),
    )

    try:
        await app.run()
        return 0
    except KeyboardInterrupt:
        pinfo("Beende auf Benutzeranfrage...")
        return 0
    except Exception as e:
        perror(f"Client-Fehler: {e}")
        return 1


async def run_standalone(args: argparse.Namespace) -> int:
    """Startet den Standalone-Modus."""
    from trixy_core.standalone import StandaloneApplication

    app = StandaloneApplication(
        config_path=args.config,
        debug=args.debug,
        keyboard_input=getattr(args, "keyboard_input", False),
    )

    try:
        await app.run()
        return 0
    except KeyboardInterrupt:
        pinfo("Beende auf Benutzeranfrage...")
        return 0
    except Exception as e:
        perror(f"Standalone-Fehler: {e}")
        return 1


async def run_config(args: argparse.Namespace) -> int:
    """Startet das Config-Tool."""
    from trixy_core.config_app import ConfigApplication

    app = ConfigApplication(
        host=args.host,
        port=args.port,
        encryption_key_path=args.key,
    )

    try:
        await app.run()
        return 0
    except ConnectionError as e:
        perror(f"Verbindungsfehler: {e}")
        return 1
    except KeyboardInterrupt:
        pinfo("Config-Tool beendet")
        return 0
    except Exception as e:
        perror(f"Config-Tool-Fehler: {e}")
        return 1


async def run_trainer(args: argparse.Namespace) -> int:
    """Startet den Trainer-Modus."""
    from trixy_core.trainer import TrainerApplication, TrainingTarget

    # Ziel-Mapping
    target_map = {
        "wakeword": TrainingTarget.WAKEWORD,
        "voice-recognition": TrainingTarget.VOICE_RECOGNITION,
        "all": TrainingTarget.ALL,
    }

    target = target_map.get(args.target, TrainingTarget.ALL)

    app = TrainerApplication(
        target=target,
        config_path=args.config,
        debug=args.debug
    )

    try:
        await app.run()
        return 0
    except KeyboardInterrupt:
        pinfo("Training abgebrochen...")
        return 0
    except Exception as e:
        perror(f"Trainer-Fehler: {e}")
        return 1


def main() -> int:
    """Haupteinstiegspunkt."""
    parser = create_parser()
    args = parser.parse_args()

    # Modus-Dispatcher
    mode_runners = {
        "server": run_server,
        "client": run_client,
        "standalone": run_standalone,
        "config": run_config,
        "train": run_trainer,
    }

    runner = mode_runners.get(args.mode)

    if runner is None:
        parser.print_help()
        return 1

    # Asyncio-Loop ausführen
    try:
        return asyncio.run(runner(args))
    except KeyboardInterrupt:
        return 0


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