Trixy7/make_installer/installer/preflight.py

# -*- coding: utf-8 -*-
"""
Preflight-Check — Ressourcen-Pruefung vor der Installation.

Berechnet den geschaetzten Gesamtbedarf (Festplatte, RAM)
und vergleicht mit der verfuegbaren Hardware.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field

from .config import InstallerConfig
from .system_info import SystemInfo, detect_system, get_model_recommendations


@dataclass
class ResourceEstimate:
    """Geschaetzter Ressourcenbedarf."""
    # Festplatte (MB)
    disk_source_mb: int = 0       # Entpackter Quellcode
    disk_venv_mb: int = 0         # Python venv + Requirements
    disk_plugins_mb: int = 0      # Plugin-Abhaengigkeiten
    disk_models_mb: int = 0       # Modell-Downloads
    disk_runtime_mb: int = 0      # Logs, Cache, Datenbanken

    # RAM (MB)
    ram_base_mb: int = 0          # Trixy Grundbedarf
    ram_models_mb: int = 0        # Modelle im RAM

    @property
    def disk_total_mb(self) -> int:
        return (self.disk_source_mb + self.disk_venv_mb +
                self.disk_plugins_mb + self.disk_models_mb +
                self.disk_runtime_mb)

    @property
    def ram_total_mb(self) -> int:
        return self.ram_base_mb + self.ram_models_mb


@dataclass
class PreflightResult:
    """Ergebnis der Preflight-Pruefung."""
    passed: bool = True
    warnings: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)
    estimate: ResourceEstimate = field(default_factory=ResourceEstimate)
    system: SystemInfo = field(default_factory=SystemInfo)


def run_preflight(config: InstallerConfig) -> PreflightResult:
    """
    Fuehrt die Preflight-Pruefung durch.

    Berechnet geschaetzten Bedarf und vergleicht mit Hardware.
    """
    result = PreflightResult()
    result.system = detect_system()
    est = result.estimate
    info = result.system

    # --- Festplatten-Schaetzung ---

    # Quellcode (entpackt)
    archive = config.source_archive
    if archive and os.path.isfile(archive):
        # Entpackt ~3-5x groesser als .tar.gz
        archive_mb = os.path.getsize(archive) // (1024 * 1024)
        est.disk_source_mb = archive_mb * 4
    else:
        # Fallback-Schaetzung nach Modus
        mode_sizes = {"server": 800, "standalone": 800, "client": 200, "config": 20}
        est.disk_source_mb = mode_sizes.get(config.mode, 400)

    # Python venv + Kern-Requirements
    est.disk_venv_mb = 500  # ~500MB fuer torch, numpy, scipy, textual, etc.

    # Plugin-Abhaengigkeiten (nur server/standalone haben Plugins)
    if config.mode in ("server", "standalone"):
        plugin_sizes = {
            "stt_whisper": 300,
            "nlp_llm": 200,
            "tts_coqui": 500,
            "stt_vosk": 50,
            "tts_piper": 50,
            "music": 100,
        }
        for plugin in config.selected_plugins:
            est.disk_plugins_mb += plugin_sizes.get(plugin, 20)

        # Modell-Downloads
        if config.download_models:
            recs = get_model_recommendations(info, config.selected_plugins)
            for rec in recs:
                if rec.recommended:
                    est.disk_models_mb += rec.size_mb

    # Runtime-Daten (Logs, Cache, Datenbanken)
    est.disk_runtime_mb = 200

    # --- RAM-Schaetzung ---

    # Basis: Python + Trixy Core + Event-Loop
    if config.mode == "client":
        est.ram_base_mb = 150   # Client: nur Wakeword + Audio + Netzwerk
    else:
        est.ram_base_mb = 300   # Server/Standalone: Event-Loop + Plugin-System

    # Wakeword-Modell (Client und Standalone)
    if config.mode in ("client", "standalone"):
        est.ram_models_mb += 50  # OpenWakeWord ONNX-Modelle (~50MB)

    # Plugin-Modelle im RAM (nur server/standalone)
    if config.mode in ("server", "standalone"):
        ram_per_plugin = {
            "stt_vosk": 200,       # Vosk-Modell im RAM
            "stt_whisper": 500,    # Whisper base
            "nlp_llm": 1200,      # Qwen 1.5B Q4 geladen
            "tts_piper": 100,     # ONNX Runtime
            "tts_coqui": 800,     # Neuronales TTS
        }
        for plugin in config.selected_plugins:
            est.ram_models_mb += ram_per_plugin.get(plugin, 0)

    # --- Pruefung ---

    # Festplatte
    disk_free = info.disk_free_mb
    disk_needed = est.disk_total_mb
    disk_buffer = 512 if config.mode == "client" else 1024

    if disk_free < disk_needed + disk_buffer:
        result.passed = False
        result.errors.append(
            f"Zu wenig Festplattenspeicher: "
            f"{_fmt_mb(disk_needed)} benoetigt + {_fmt_mb(disk_buffer)} Puffer, "
            f"aber nur {_fmt_mb(disk_free)} frei"
        )
    elif disk_free < disk_needed * 1.5:
        result.warnings.append(
            f"Festplattenspeicher knapp: "
            f"{_fmt_mb(disk_needed)} benoetigt, {_fmt_mb(disk_free)} frei"
        )

    # RAM
    ram_total = info.ram_total_mb
    ram_needed = est.ram_total_mb

    if ram_total < ram_needed:
        result.passed = False
        result.errors.append(
            f"Zu wenig RAM: "
            f"{_fmt_mb(ram_needed)} benoetigt (Laufzeit), "
            f"aber nur {_fmt_mb(ram_total)} verfuegbar"
        )
    elif ram_total < ram_needed * 1.3:
        result.warnings.append(
            f"RAM knapp: {_fmt_mb(ram_needed)} benoetigt, "
            f"{_fmt_mb(ram_total)} verfuegbar — System koennte langsam werden"
        )

    # Quellarchiv vorhanden
    if config.source_archive and not os.path.isfile(config.source_archive):
        result.warnings.append(f"Quellarchiv nicht gefunden: {config.source_archive}")

    return result


def format_preflight(result: PreflightResult) -> str:
    """Formatiert das Preflight-Ergebnis als lesbaren Text."""
    lines: list[str] = []
    est = result.estimate
    info = result.system

    lines.append("Geschaetzter Bedarf:")
    lines.append(f"  Quellcode:        {_fmt_mb(est.disk_source_mb)}")
    lines.append(f"  Python venv:      {_fmt_mb(est.disk_venv_mb)}")
    lines.append(f"  Plugin-Deps:      {_fmt_mb(est.disk_plugins_mb)}")
    if est.disk_models_mb:
        lines.append(f"  Modelle:          {_fmt_mb(est.disk_models_mb)}")
    lines.append(f"  Runtime-Daten:    {_fmt_mb(est.disk_runtime_mb)}")
    lines.append(f"  ─────────────────────────")
    lines.append(f"  Festplatte total: {_fmt_mb(est.disk_total_mb)}")
    lines.append(f"  RAM (Laufzeit):   {_fmt_mb(est.ram_total_mb)}")
    lines.append("")
    lines.append("Verfuegbar:")
    lines.append(f"  Festplatte frei:  {_fmt_mb(info.disk_free_mb)}")
    lines.append(f"  RAM gesamt:       {_fmt_mb(info.ram_total_mb)}")

    if result.errors:
        lines.append("")
        for err in result.errors:
            lines.append(f"  FEHLER: {err}")

    if result.warnings:
        lines.append("")
        for warn in result.warnings:
            lines.append(f"  WARNUNG: {warn}")

    if result.passed and not result.warnings:
        lines.append("")
        lines.append("  Ressourcen ausreichend.")

    return "\n".join(lines)


def _fmt_mb(mb: int) -> str:
    """Formatiert MB-Werte menschenlesbar."""
    if mb >= 1024:
        return f"{mb / 1024:.1f} GB"
    return f"{mb} MB"