# -*- coding: utf-8 -*- """ InstallRunner — Fuehrt Installationsschritte aus. Steuert Progress, Logging und Abbruch. """ from __future__ import annotations import asyncio import os import shutil from typing import Callable from .config import InstallerConfig from .steps import InstallationStep, get_steps class InstallRunner: """ Fuehrt Installationsschritte sequentiell aus. Bietet gewichtete Progress-Updates und Echtzeit-Log. """ def __init__( self, config: InstallerConfig, log_callback: Callable[[str], None], progress_callback: Callable[[float, str], None], ) -> None: self._config = config self._log = log_callback self._progress = progress_callback self._cancelled = False self._current_process: asyncio.subprocess.Process | None = None # Wird von step_wifi_ap / step_wifi_connect gesetzt wenn # Netzwerk-Aenderungen geschrieben wurden die erst nach # Reboot aktiv werden. install_view liest das Flag am Ende # und zeigt eine prominente Warnung. self.reboot_required: bool = False self.reboot_reasons: list[str] = [] self._swap_file: str = "" # Pfad zur angelegten Swap-Datei def cancel(self) -> None: """Bricht die Installation ab.""" self._cancelled = True if self._current_process and self._current_process.returncode is None: self._current_process.terminate() async def run(self) -> bool: """ Fuehrt alle Schritte aus. Returns: True bei Erfolg, False bei Fehler/Abbruch. """ steps = get_steps(self._config.mode) total_weight = sum(s.weight for s in steps) completed_weight = 0 self._log(f"=== Trixy {self._config.mode.title()} Installation ===") self._log(f"Version: {self._config.version}") self._log(f"Ziel: {self._config.install_path}") self._log("") # Sudo-Rechte vorab sicherstellen needs_sudo = any( s.command_fn in ( "step_apt_packages", "step_hostname", "step_wifi_ap", "step_directories", "step_systemd", ) for s in steps ) if needs_sudo: self._log("Pruefe sudo-Rechte...") code, _ = await self._run_cmd("sudo", "-v", check=False) if code != 0: self._log(" FEHLER: sudo-Rechte benoetigt aber nicht verfuegbar.") self._log(" Bitte Installer mit 'sudo ./install-{mode}.sh' starten") self._log(" oder den Benutzer zur sudoers-Gruppe hinzufuegen.") return False self._log(" -> sudo verfuegbar\n") for i, step in enumerate(steps, 1): if self._cancelled: self._log("\n[ABBRUCH] Installation wurde abgebrochen.") return False self._log(f"[{i}/{len(steps)}] {step.name}...") self._progress(completed_weight / total_weight, step.name) method = getattr(self, step.command_fn, None) if method is None: self._log(f" WARNUNG: Methode {step.command_fn} nicht implementiert") completed_weight += step.weight continue try: success = await method() if not success: self._log(f" FEHLER bei Schritt: {step.name}") return False except Exception as e: self._log(f" FEHLER: {e}") return False completed_weight += step.weight self._progress(completed_weight / total_weight, step.name) self._log(f" -> {step.name} abgeschlossen\n") self._progress(1.0, "Fertig") self._log("=== Installation erfolgreich abgeschlossen ===") return True # --- Hilfsmethoden --- _ANSI_RE = None # Lazy-compiled Regex @staticmethod def _clean_ansi(text: str) -> str: """Entfernt ANSI Escape-Sequenzen aus Text.""" import re if InstallRunner._ANSI_RE is None: InstallRunner._ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]") return InstallRunner._ANSI_RE.sub("", text) async def _run_cmd(self, *args: str, check: bool = True) -> tuple[int, str]: """Fuehrt einen Befehl aus und loggt die Ausgabe.""" self._log(f" $ {' '.join(args)}") # TMPDIR auf Installationspfad umleiten (tmpfs auf Pi zu klein fuer pip) env = os.environ.copy() tmp_dir = f"{self._config.install_path}/tmp" os.makedirs(tmp_dir, exist_ok=True) env["TMPDIR"] = tmp_dir env["PIP_CACHE_DIR"] = f"{self._config.install_path}/pip-cache" proc = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env=env, ) self._current_process = proc output_lines: list[str] = [] assert proc.stdout is not None while True: line = await proc.stdout.readline() if not line: break decoded = line.decode("utf-8", errors="replace").rstrip() # \r-basierte Progress-Zeilen: nur letzten Teil nehmen if "\r" in decoded: decoded = decoded.rsplit("\r", 1)[-1].strip() # ANSI-Codes entfernen decoded = self._clean_ansi(decoded) if not decoded: continue output_lines.append(decoded) self._log(f" {decoded}") if self._cancelled: proc.terminate() await proc.wait() return -1, "Abgebrochen" await proc.wait() self._current_process = None output = "\n".join(output_lines) if check and proc.returncode != 0: self._log(f" Befehl fehlgeschlagen (Exit-Code: {proc.returncode})") return proc.returncode or 0, output async def _run_sudo(self, *args: str, check: bool = True) -> tuple[int, str]: """Fuehrt einen Befehl mit sudo aus.""" return await self._run_cmd("sudo", *args, check=check) # --- Installations-Schritte --- async def step_check_swap(self) -> bool: """Prueft RAM+Swap und legt bei Bedarf eine Swap-Datei an. Auf Raspberry Pi mit wenig RAM (1-2 GB) kann pip install oder Modell-Entpacken den OOM-Killer ausloesen. Eine temporaere Swap-Datei verhindert das. Logik: 1. Verfuegbaren RAM + bestehenden Swap ermitteln 2. Wenn Gesamt < MIN_TOTAL_MB → Swap-Datei anlegen 3. Swap-Datei wird in self._swap_file gespeichert 4. Am Ende der Installation: - Server/Standalone: Swap permanent behalten (fuer ML-Inferenz) - Client: Swap entfernen (leichtgewichtig) """ MIN_TOTAL_MB = 3072 # 3 GB Minimum fuer sichere Installation SWAP_SIZE_MB = 2048 # 2 GB Swap-Datei self._log("Pruefe verfuegbaren Speicher...") # RAM ermitteln try: code, output = await self._run_cmd("free", "-m", check=False) ram_total = 0 swap_total = 0 for line in output.splitlines(): parts = line.split() if not parts: continue if parts[0].lower().startswith("mem"): ram_total = int(parts[1]) elif parts[0].lower().startswith("swap"): swap_total = int(parts[1]) total = ram_total + swap_total self._log(f" RAM: {ram_total} MB, Swap: {swap_total} MB, Gesamt: {total} MB") if total >= MIN_TOTAL_MB: self._log(f" Speicher ausreichend (>= {MIN_TOTAL_MB} MB)") return True except Exception as e: self._log(f" WARNUNG: Speicher-Check fehlgeschlagen: {e}") self._log(f" Ueberspringe Swap-Setup") return True # Swap-Datei anlegen needed = MIN_TOTAL_MB - total swap_mb = max(SWAP_SIZE_MB, needed) swap_path = "/swapfile_trixy" self._log(f" Speicher zu gering ({total} MB < {MIN_TOTAL_MB} MB)") self._log(f" Lege {swap_mb} MB Swap-Datei an: {swap_path}") # Pruefen ob schon eine Trixy-Swap-Datei existiert code, _ = await self._run_cmd("test", "-f", swap_path, check=False) if code == 0: self._log(f" Swap-Datei existiert bereits — aktiviere sie") await self._run_sudo("swapon", swap_path, check=False) self._swap_file = swap_path return True # Pruefen ob genug Disk-Platz fuer Swap code, df_output = await self._run_cmd("df", "-m", "/", check=False) if code == 0: for line in df_output.splitlines()[1:]: parts = line.split() if len(parts) >= 4: avail_mb = int(parts[3]) if avail_mb < swap_mb + 1024: # +1 GB Reserve self._log( f" WARNUNG: Nicht genug Platz fuer Swap " f"({avail_mb} MB frei, {swap_mb + 1024} MB benoetigt)" ) self._log(f" Swap-Setup uebersprungen — Installation wird trotzdem versucht") return True break # Swap-Datei erstellen code, _ = await self._run_sudo( "fallocate", "-l", f"{swap_mb}M", swap_path, check=False, ) if code != 0: # Fallback: dd (fuer Dateisysteme ohne fallocate) code, _ = await self._run_sudo( "dd", "if=/dev/zero", f"of={swap_path}", "bs=1M", f"count={swap_mb}", "status=progress", check=False, ) if code != 0: self._log(f" WARNUNG: Swap-Datei konnte nicht erstellt werden") return True # Kein Abbruch — Installation trotzdem versuchen # Berechtigungen und Swap einrichten await self._run_sudo("chmod", "600", swap_path) code, _ = await self._run_sudo("mkswap", swap_path) if code != 0: self._log(f" WARNUNG: mkswap fehlgeschlagen") await self._run_sudo("rm", "-f", swap_path, check=False) return True code, _ = await self._run_sudo("swapon", swap_path) if code != 0: self._log(f" WARNUNG: swapon fehlgeschlagen") await self._run_sudo("rm", "-f", swap_path, check=False) return True self._swap_file = swap_path self._log(f" Swap-Datei aktiv: {swap_path} ({swap_mb} MB)") # Swappiness temporaer erhoehen damit Swap auch genutzt wird await self._run_sudo( "sysctl", "-w", "vm.swappiness=60", check=False, ) return True async def step_cleanup_swap(self) -> bool: """Entfernt oder behält die Swap-Datei nach der Installation. - Server/Standalone: Swap permanent behalten (fstab Eintrag) → ML-Inferenz und Plugin-Ausfuehrung brauchen Speicher - Client: Swap entfernen (leichtgewichtig) """ swap_path = getattr(self, "_swap_file", "") if not swap_path: self._log(" Keine Trixy-Swap-Datei vorhanden — nichts zu tun") return True if self._config.mode in ("server", "standalone"): # Swap permanent behalten self._log(f" Swap-Datei wird dauerhaft beibehalten: {swap_path}") # Pruefen ob schon in /etc/fstab code, fstab = await self._run_cmd("cat", "/etc/fstab", check=False) if swap_path not in fstab: # fstab-Eintrag hinzufuegen fstab_line = f"{swap_path} none swap sw 0 0" code, _ = await self._run_sudo( "bash", "-c", f"echo '{fstab_line}' >> /etc/fstab", ) if code == 0: self._log(f" fstab-Eintrag hinzugefuegt") else: self._log(f" WARNUNG: fstab-Eintrag konnte nicht hinzugefuegt werden") else: self._log(f" fstab-Eintrag existiert bereits") else: # Client: Swap entfernen self._log(f" Client-Modus: Swap-Datei wird entfernt") await self._run_sudo("swapoff", swap_path, check=False) await self._run_sudo("rm", "-f", swap_path, check=False) self._log(f" Swap-Datei entfernt: {swap_path}") return True async def step_stop_service(self) -> bool: """Stoppt den laufenden Service vor dem Update (falls vorhanden). Verhindert dass eine laufende Python-Instanz Dateien blockiert oder mit alten Modulen weiterlaeuft nachdem der Code aktualisiert wurde. Wird uebersprungen wenn der Service nicht existiert (Erstinstallation). """ service_name = f"trixy-{self._config.mode}.service" # Pruefen ob der Service ueberhaupt existiert code, _ = await self._run_cmd( "systemctl", "list-unit-files", service_name, check=False, ) if code != 0: self._log(f" Service {service_name} nicht gefunden — ueberspringe Stop") return True # Pruefen ob der Service laeuft code, output = await self._run_cmd( "systemctl", "is-active", service_name, check=False, ) if output.strip() != "active": self._log(f" Service {service_name} laeuft nicht — ueberspringe Stop") return True self._log(f" Stoppe {service_name}...") code, _ = await self._run_sudo( "systemctl", "stop", service_name, check=False, ) if code == 0: self._log(f" -> {service_name} gestoppt") else: self._log(f" WARNUNG: Konnte {service_name} nicht stoppen") # Kurz warten bis alle Prozesse wirklich weg sind await asyncio.sleep(2) return True async def step_start_service(self) -> bool: """Startet den Service nach der Installation (wenn install_service + autostart).""" cfg = self._config if not cfg.install_service: return True service_name = f"trixy-{cfg.mode}.service" if not cfg.autostart: self._log(f" Autostart deaktiviert — {service_name} nicht automatisch gestartet") self._log(f" Manuell starten: sudo systemctl start {service_name}") return True self._log(f" Starte {service_name}...") code, _ = await self._run_sudo( "systemctl", "start", service_name, check=False, ) if code == 0: self._log(f" -> {service_name} gestartet") # Kurz warten und Status pruefen await asyncio.sleep(2) _, status = await self._run_cmd( "systemctl", "is-active", service_name, check=False, ) self._log(f" Status: {status.strip()}") else: self._log(f" WARNUNG: {service_name} konnte nicht gestartet werden") self._log(f" Logs pruefen: journalctl -u {service_name} -n 50") return True async def step_extract(self) -> bool: """Entpackt das Quellarchiv nach source/.""" archive = self._config.source_archive source = self._config.source_path # install_path/source if not os.path.isfile(archive): self._log(f" Archiv nicht gefunden: {archive}") return False # source-Verzeichnis erstellen code, _ = await self._run_sudo("mkdir", "-p", source) if code != 0: return False # Besitzer setzen await self._run_sudo("chown", self._config.username, self._config.install_path) # Entpacken — strip-components=1 entfernt den Prefix (trixy-server-1.0.0/) self._log(f" Entpacke nach {source}...") code, _ = await self._run_cmd( "tar", "xzf", archive, "-C", source, "--strip-components=1", ) return code == 0 # Paketnamen pro Paketmanager # Schluessel = kanonischer Name, Werte = {pm: paketname} _PKG_MAP: dict[str, dict[str, str]] = { "python3": {"apt": "python3", "dnf": "python3", "pacman": "python", "zypper": "python3"}, "python3-venv": {"apt": "python3-venv", "dnf": "python3-venv", "pacman": "python", "zypper": "python3-venv"}, "python3-pip": {"apt": "python3-pip", "dnf": "python3-pip", "pacman": "python-pip", "zypper": "python3-pip"}, "python3-dev": {"apt": "python3-dev", "dnf": "python3-devel", "pacman": "python", "zypper": "python3-devel"}, "git": {"apt": "git", "dnf": "git", "pacman": "git", "zypper": "git"}, "curl": {"apt": "curl", "dnf": "curl", "pacman": "curl", "zypper": "curl"}, "wget": {"apt": "wget", "dnf": "wget", "pacman": "wget", "zypper": "wget"}, "alsa-utils": {"apt": "alsa-utils", "dnf": "alsa-utils", "pacman": "alsa-utils", "zypper": "alsa-utils"}, "ffmpeg": {"apt": "ffmpeg", "dnf": "ffmpeg-free", "pacman": "ffmpeg", "zypper": "ffmpeg"}, "portaudio-dev": {"apt": "portaudio19-dev", "dnf": "portaudio-devel", "pacman": "portaudio", "zypper": "portaudio-devel"}, "alsa-dev": {"apt": "libasound2-dev", "dnf": "alsa-lib-devel", "pacman": "alsa-lib", "zypper": "alsa-devel"}, "openblas-dev": {"apt": "libopenblas-dev", "dnf": "openblas-devel", "pacman": "openblas", "zypper": "openblas-devel"}, "lapack-dev": {"apt": "liblapack-dev", "dnf": "lapack-devel", "pacman": "lapack", "zypper": "lapack-devel"}, "gfortran": {"apt": "gfortran", "dnf": "gcc-gfortran", "pacman": "gcc-fortran", "zypper": "gcc-fortran"}, "ffi-dev": {"apt": "libffi-dev", "dnf": "libffi-devel", "pacman": "libffi", "zypper": "libffi-devel"}, "ssl-dev": {"apt": "libssl-dev", "dnf": "openssl-devel", "pacman": "openssl", "zypper": "libopenssl-devel"}, "hostapd": {"apt": "hostapd", "dnf": "hostapd", "pacman": "hostapd", "zypper": "hostapd"}, "dnsmasq": {"apt": "dnsmasq", "dnf": "dnsmasq", "pacman": "dnsmasq", "zypper": "dnsmasq"}, "iptables": {"apt": "iptables", "dnf": "iptables", "pacman": "iptables", "zypper": "iptables"}, "netfilter-persistent": {"apt": "netfilter-persistent", "dnf": "", "pacman": "", "zypper": ""}, "iptables-persistent": {"apt": "iptables-persistent", "dnf": "", "pacman": "", "zypper": ""}, "cmake": {"apt": "cmake", "dnf": "cmake", "pacman": "cmake", "zypper": "cmake"}, "g++": {"apt": "g++", "dnf": "gcc-c++", "pacman": "gcc", "zypper": "gcc-c++"}, "evtest": {"apt": "evtest", "dnf": "evtest", "pacman": "evtest", "zypper": "evtest"}, "input-utils": {"apt": "input-utils", "dnf": "input-utils", "pacman": "input-utils", "zypper": "input-utils"}, } def _detect_pkg_manager(self) -> tuple[str, list[str], list[str], list[str]]: """ Erkennt den Paketmanager. Returns: (name, update_cmd, install_cmd, check_cmd) install_cmd hat Platzhalter — Pakete werden angehaengt. """ managers = [ ("apt", ["apt-get", "update", "-qq"], ["apt-get", "install", "-y", "-qq"], ["dpkg", "-s"]), ("dnf", ["dnf", "check-update"], ["dnf", "install", "-y", "-q"], ["rpm", "-q"]), ("pacman", ["pacman", "-Sy"], ["pacman", "-S", "--noconfirm", "--needed"], ["pacman", "-Q"]), ("zypper", ["zypper", "refresh", "-q"], ["zypper", "install", "-y", "-q"], ["rpm", "-q"]), ] for name, update, install, check in managers: if shutil.which(install[0]): return name, update, install, check return "", [], [], [] def _resolve_pkg(self, canonical: str, pm: str) -> str: """Loest einen kanonischen Paketnamen fuer den erkannten Paketmanager auf.""" mapping = self._PKG_MAP.get(canonical, {}) return mapping.get(pm, canonical) async def _install_pkg_group( self, name: str, canonical_names: list[str], required: bool = True, ) -> bool: """Installiert eine Paketgruppe mit dem erkannten Paketmanager.""" pm_name, _, install_cmd, _ = self._detect_pkg_manager() if not pm_name: self._log(f" FEHLER: Kein unterstuetzter Paketmanager gefunden (apt/dnf/pacman/zypper)") return not required # Paketnamen aufloesen packages = [] for canonical in canonical_names: resolved = self._resolve_pkg(canonical, pm_name) if resolved not in packages: packages.append(resolved) self._log(f" {name} ({pm_name}): {', '.join(packages)}") code, _ = await self._run_sudo(*install_cmd, *packages) if code != 0: # Einzeln versuchen self._log(f" Gruppeninstallation fehlgeschlagen — installiere einzeln...") failed = [] for pkg in packages: c, _ = await self._run_sudo(*install_cmd, pkg) if c != 0: failed.append(pkg) if failed: self._log(f" WARNUNG: Nicht verfuegbar: {', '.join(failed)}") if required: return False return True async def step_apt_packages(self) -> bool: """Installiert System-Pakete in Gruppen.""" pm_name, update_cmd, _, _ = self._detect_pkg_manager() if not pm_name: self._log(" FEHLER: Kein unterstuetzter Paketmanager gefunden") self._log(" Unterstuetzt: apt (Debian/Ubuntu), dnf (Fedora/RHEL),") self._log(" pacman (Arch), zypper (openSUSE)") self._log(" Bitte Pakete manuell installieren.") return False self._log(f" Paketmanager: {pm_name}") self._log(f" Aktualisiere Paketlisten...") # dnf check-update gibt Exit 100 bei verfuegbaren Updates — das ist OK code, _ = await self._run_sudo(*update_cmd, check=False) if code != 0 and pm_name != "dnf": self._log(f" WARNUNG: Paketlisten-Update fehlgeschlagen") # Runtime-Pakete (immer benoetigt) runtime = ["python3", "python3-venv", "python3-pip", "git", "curl", "wget"] if not await self._install_pkg_group("Runtime", runtime): return False # Audio-Runtime audio = ["alsa-utils", "ffmpeg"] if not await self._install_pkg_group("Audio", audio): return False # Build-Abhaengigkeiten (optional auf x86_64 mit Wheels) build = [ "python3-dev", "portaudio-dev", "alsa-dev", "openblas-dev", "lapack-dev", "gfortran", "ffi-dev", "ssl-dev", ] await self._install_pkg_group("Build-Abhaengigkeiten", build, required=False) # HID / Input-Tools (Client/Standalone — Konferenzmikrofone, Headsets) if self._config.mode in ("client", "standalone"): hid = ["evtest", "input-utils"] await self._install_pkg_group("HID/Input", hid, required=False) # WiFi-Hotspot (nur Server) if self._config.mode == "server" and self._config.wifi_enabled: wifi = ["hostapd", "dnsmasq", "iptables"] if not await self._install_pkg_group("WiFi-Hotspot", wifi): return False # netfilter-persistent persistiert iptables-Regeln ueber # Reboots — optional, Fallback: Regeln werden beim Boot # erneut gesetzt oder per Firewall-Script angelegt. await self._install_pkg_group( "iptables-Persistenz", ["netfilter-persistent", "iptables-persistent"], required=False, ) return True async def step_hostname(self) -> bool: """Setzt den Hostnamen.""" code, _ = await self._run_sudo( "hostnamectl", "set-hostname", self._config.hostname, ) return code == 0 async def step_wifi_connect(self) -> bool: """Konfiguriert WLAN-Verbindung (Client/Standalone) — deferred. Legt NetworkManager-Profile an, verbindet aber NICHT sofort. Der Grund: wenn der User gerade ueber SSH auf dem aktuellen Netz installiert, wuerde ein `nmcli connection up` auf das neue Netz die Session killen. Stattdessen: - Primaeres Netz (Trixy-Hotspot) mit autoconnect-priority 100 - Fallback-Netz (z.B. Heim-WLAN) mit autoconnect-priority 10 - Auto-Switchback-Timer optional Beim naechsten Reboot entscheidet NetworkManager anhand der Prioritaeten: wenn das primaere Netz sichtbar ist, wird es bevorzugt — sonst bleibt der Pi auf dem Fallback. Setzt `self.reboot_required = True`. """ if not self._config.wifi_connect: self._log(" Keine WLAN-Aenderung — uebersprungen") return True cfg = self._config primary_ssid = cfg.wifi_connect_ssid primary_pw = cfg.wifi_connect_password if not shutil.which("nmcli"): self._log(" FEHLER: nmcli nicht gefunden — NetworkManager erforderlich") self._log(" Hinweis: Raspberry Pi OS Bookworm nutzt NetworkManager als Standard") return False self._log(" Verwende NetworkManager (nmcli)...") # Primaeres Netz konfigurieren (Prioritaet 100) if not await self._nmcli_add_connection( ssid=primary_ssid, password=primary_pw, priority=100, ): self._log(f" FEHLER: Primaere Verbindung '{primary_ssid}' konnte nicht angelegt werden") return False self._log(f" Primaeres Netz angelegt: {primary_ssid} (Prioritaet 100)") # Fallback-Netz konfigurieren wenn aktiviert fallback_ssid = "" if cfg.wifi_fallback_enabled: fallback_ssid = cfg.wifi_fallback_ssid if await self._nmcli_add_connection( ssid=fallback_ssid, password=cfg.wifi_fallback_password, priority=10, ): self._log(f" Fallback-Netz angelegt: {fallback_ssid} (Prioritaet 10)") else: self._log(f" WARNUNG: Fallback-Verbindung '{fallback_ssid}' konnte nicht angelegt werden") fallback_ssid = "" # Failover-Script + Timer installieren (wenn Fallback + Auto-Switchback) if cfg.wifi_fallback_enabled and cfg.wifi_auto_switchback: if await self._install_wifi_failover_service( primary_ssid=primary_ssid, fallback_ssid=fallback_ssid or cfg.wifi_fallback_ssid, interval=cfg.wifi_switchback_interval, ): self._log(f" -> Auto-Rueckwechsel aktiviert (alle {cfg.wifi_switchback_interval}s)") else: self._log(" WARNUNG: Auto-Rueckwechsel konnte nicht installiert werden") # --- Reboot-Flag setzen --- # Keine sofortige Aktivierung — NetworkManager waehlt beim # naechsten Boot das Netz mit der hoechsten erreichbaren Prioritaet. self.reboot_required = True reason = f"Client verbindet sich nach Reboot primaer zu '{primary_ssid}'" if fallback_ssid: reason += f" (Fallback: '{fallback_ssid}')" self.reboot_reasons.append(reason) self._log("") self._log(" HINWEIS: WLAN-Profile sind angelegt aber NICHT aktiviert.") self._log(" NetworkManager verbindet beim naechsten Reboot.") self._log(" Die aktuelle SSH-Verbindung bleibt bestehen.") return True # --- nmcli Helper --- async def _nmcli_add_connection( self, ssid: str, password: str, priority: int, ) -> bool: """Legt eine NetworkManager WLAN-Verbindung an (oder aktualisiert sie).""" con_name = f"trixy-{ssid}" # Bestehende Verbindung mit gleichem Namen entfernen (idempotent) await self._run_sudo( "nmcli", "connection", "delete", con_name, check=False, ) # Connection anlegen code, _ = await self._run_sudo( "nmcli", "connection", "add", "type", "wifi", "con-name", con_name, "ifname", "wlan0", "ssid", ssid, ) if code != 0: return False # Passwort, Prioritaet, Auto-Connect konfigurieren code, _ = await self._run_sudo( "nmcli", "connection", "modify", con_name, "wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", password, "connection.autoconnect", "yes", "connection.autoconnect-priority", str(priority), "connection.autoconnect-retries", "0", ) return code == 0 async def _nmcli_try_connect(self, ssid: str, max_wait: int = 30) -> bool: """Versucht eine Connection zu aktivieren, mit Rescan-Retry. Wartet bis das Netz sichtbar ist und aktiviert die Verbindung. """ con_name = f"trixy-{ssid}" waited = 0 # Rescan erzwingen await self._run_sudo("nmcli", "device", "wifi", "rescan", check=False) while waited < max_wait: # Ist das Netz im Scan sichtbar? code, output = await self._run_cmd( "nmcli", "-t", "-f", "SSID", "device", "wifi", "list", check=False, ) if code == 0 and ssid in output.splitlines(): # Verbinden code, _ = await self._run_sudo( "nmcli", "connection", "up", con_name, check=False, ) if code == 0: return True await asyncio.sleep(2) waited += 2 await self._run_sudo("nmcli", "device", "wifi", "rescan", check=False) return False async def _install_wifi_failover_service( self, primary_ssid: str, fallback_ssid: str, interval: int, ) -> bool: """Installiert ein Failover-Script + systemd-Timer fuer Auto-Switchback. Prueft periodisch ob das primaere Netz wieder erreichbar ist, und wechselt dann von Fallback zurueck zum primaeren Netz. """ primary_con = f"trixy-{primary_ssid}" fallback_con = f"trixy-{fallback_ssid}" # Shell-Escape fuer SSIDs (verhindert Injection) def sh_escape(s: str) -> str: return s.replace("\\", "\\\\").replace('"', '\\"').replace("$", "\\$") script_content = f"""#!/usr/bin/env bash # ============================================================================ # Trixy WLAN Auto-Switchback # Generiert vom Trixy-Installer — NICHT manuell editieren # # Prueft periodisch ob das primaere Netz erreichbar ist und wechselt zurueck # wenn der Client aktuell am Fallback-Netz haengt. # ============================================================================ set -u PRIMARY_CON="{sh_escape(primary_con)}" PRIMARY_SSID="{sh_escape(primary_ssid)}" FALLBACK_CON="{sh_escape(fallback_con)}" FALLBACK_SSID="{sh_escape(fallback_ssid)}" # Aktuell aktive Verbindung auf wlan0 ermitteln current=$(nmcli -t -f NAME,DEVICE connection show --active \\ | awk -F':' '$2 == "wlan0" {{ print $1; exit }}') # Bereits am primaeren Netz? -> fertig if [[ "${{current}}" == "${{PRIMARY_CON}}" ]]; then exit 0 fi # Nur handeln wenn wir am Fallback haengen if [[ "${{current}}" != "${{FALLBACK_CON}}" ]]; then exit 0 fi # Scan erzwingen nmcli device wifi rescan >/dev/null 2>&1 || true sleep 3 # Ist das primaere Netz wieder sichtbar? if ! nmcli -t -f SSID device wifi list | grep -qxF "${{PRIMARY_SSID}}"; then exit 0 fi # Wechsel versuchen logger -t trixy-wifi "Primaeres Netz ${{PRIMARY_SSID}} wieder sichtbar — wechsle zurueck" if nmcli connection up "${{PRIMARY_CON}}" >/dev/null 2>&1; then logger -t trixy-wifi "Wechsel erfolgreich" else logger -t trixy-wifi "Wechsel fehlgeschlagen — bleibe auf Fallback" fi """ service_content = """[Unit] Description=Trixy WLAN Auto-Switchback After=NetworkManager.service Wants=NetworkManager.service [Service] Type=oneshot ExecStart=/usr/local/bin/trixy-wifi-switchback.sh """ timer_content = f"""[Unit] Description=Trixy WLAN Auto-Switchback Timer [Timer] OnBootSec=2min OnUnitActiveSec={interval}s Unit=trixy-wifi-switchback.service [Install] WantedBy=timers.target """ # Script schreiben if not await self._write_file_sudo( "/usr/local/bin/trixy-wifi-switchback.sh", script_content, mode="755", ): return False # Service schreiben if not await self._write_file_sudo( "/etc/systemd/system/trixy-wifi-switchback.service", service_content, ): return False # Timer schreiben if not await self._write_file_sudo( "/etc/systemd/system/trixy-wifi-switchback.timer", timer_content, ): return False # Systemd reload await self._run_sudo("systemctl", "daemon-reload", check=False) # Timer aktivieren und starten code, _ = await self._run_sudo( "systemctl", "enable", "--now", "trixy-wifi-switchback.timer", check=False, ) return code == 0 async def _write_file_sudo( self, path: str, content: str, mode: str = "644", ) -> bool: """Schreibt eine Datei mit sudo tee und setzt die Berechtigungen.""" proc = await asyncio.create_subprocess_exec( "sudo", "tee", path, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) await proc.communicate(content.encode()) if proc.returncode != 0: return False code, _ = await self._run_sudo("chmod", mode, path, check=False) return code == 0 async def step_wifi_ap(self) -> bool: """Konfiguriert WiFi Access Point (Bookworm-kompatibel). Ablauf: 1. hostapd.conf schreiben (SSID, WPA2, Kanal) 2. dnsmasq.conf schreiben (DHCP-Range, DNS-Upstream) 3. systemd-networkd Profil fuer wlan0 statische IP (ersetzt den alten dhcpcd-Pfad — Bookworm hat kein dhcpcd) 4. NetworkManager Override: wlan0 wird "unmanaged", sonst klaut NM die Schnittstelle zurueck und ueberschreibt die IP 5. dnsmasq systemd-override mit ExecStartPre, der auf die statische wlan0-IP wartet (Race-Condition-Fix) 6. IP-Forwarding persistent + iptables MASQUERADE 7. Services nur `enable`, nicht `start` — die Umschaltung passiert bewusst erst beim naechsten Reboot, damit die aktuelle SSH-Session nicht stirbt. Setzt `self.reboot_required = True` wenn etwas geschrieben wurde. """ if not self._config.wifi_enabled: self._log(" WiFi-Hotspot deaktiviert — uebersprungen") return True cfg = self._config # --- 1. hostapd.conf --- hostapd_conf = f"""# Trixy Access Point — erzeugt vom Installer interface=wlan0 driver=nl80211 ssid={cfg.wifi_ssid} hw_mode=g channel={cfg.wifi_channel} wmm_enabled=0 macaddr_acl=0 auth_algs=1 ignore_broadcast_ssid=0 wpa=2 wpa_passphrase={cfg.wifi_password} wpa_key_mgmt=WPA-PSK wpa_pairwise=TKIP rsn_pairwise=CCMP country_code=DE ieee80211n=1 ieee80211d=1 """ if not await self._write_file_sudo( "/etc/hostapd/hostapd.conf", hostapd_conf, ): return False # Default-Config-Pfad fuer hostapd setzen default_hostapd = 'DAEMON_CONF="/etc/hostapd/hostapd.conf"\n' await self._write_file_sudo("/etc/default/hostapd", default_hostapd) self._log(" -> /etc/hostapd/hostapd.conf geschrieben") # --- 2. dnsmasq.conf --- dnsmasq_conf = f"""# Trixy DHCP-Server — erzeugt vom Installer interface=wlan0 bind-interfaces dhcp-range={cfg.dhcp_range_start},{cfg.dhcp_range_end},255.255.255.0,12h server=8.8.8.8 server=8.8.4.4 domain-needed bogus-priv address=/{cfg.wifi_ssid}/{cfg.ap_ip} address=/{cfg.hostname}/{cfg.ap_ip} """ if not await self._write_file_sudo( "/etc/dnsmasq.conf", dnsmasq_conf, ): return False self._log(" -> /etc/dnsmasq.conf geschrieben") # --- 3. systemd-networkd Profil fuer wlan0 --- networkd_conf = f"""# Trixy — statische IP fuer wlan0 im AP-Modus [Match] Name=wlan0 [Network] Address={cfg.ap_ip}/24 DHCP=no IPForward=yes """ await self._run_sudo("mkdir", "-p", "/etc/systemd/network") if not await self._write_file_sudo( "/etc/systemd/network/10-trixy-wlan0.network", networkd_conf, ): return False self._log(" -> /etc/systemd/network/10-trixy-wlan0.network geschrieben") # --- 4. NetworkManager: wlan0 unmanaged --- # Ohne diese Datei holt sich NetworkManager wlan0 nach jedem # Reboot zurueck und ueberschreibt die statische IP. `nmcli # device set wlan0 managed no` ist nur zur Laufzeit aktiv. nm_conf = """# Trixy — wlan0 aus NetworkManager-Kontrolle nehmen # Der Hotspot laeuft ueber systemd-networkd + hostapd. [keyfile] unmanaged-devices=interface-name:wlan0 """ await self._run_sudo("mkdir", "-p", "/etc/NetworkManager/conf.d") if not await self._write_file_sudo( "/etc/NetworkManager/conf.d/99-trixy-wlan0-unmanaged.conf", nm_conf, ): return False self._log(" -> /etc/NetworkManager/conf.d/99-trixy-wlan0-unmanaged.conf geschrieben") # --- 5. dnsmasq Race-Condition-Fix --- # dnsmasq darf erst starten wenn wlan0 tatsaechlich die statische # IP hat. Ohne ExecStartPre schlaegt dnsmasq "bind-interfaces" fehl. dnsmasq_override = f"""[Unit] After=systemd-networkd.service hostapd.service sys-subsystem-net-devices-wlan0.device Requires=sys-subsystem-net-devices-wlan0.device Wants=systemd-networkd.service hostapd.service [Service] ExecStartPre=/bin/sh -c 'for i in 1 2 3 4 5 6 7 8 9 10; do ip addr show wlan0 2>/dev/null | grep -q "inet {cfg.ap_ip}" && exit 0; sleep 1; done; exit 1' """ await self._run_sudo("mkdir", "-p", "/etc/systemd/system/dnsmasq.service.d") if not await self._write_file_sudo( "/etc/systemd/system/dnsmasq.service.d/trixy-wait-wlan0.conf", dnsmasq_override, ): return False self._log(" -> dnsmasq Race-Condition-Fix aktiv") # --- 6. IP-Forwarding persistent + iptables --- sysctl_conf = "# Trixy — IP-Forwarding fuer AP-Modus\nnet.ipv4.ip_forward=1\n" await self._write_file_sudo( "/etc/sysctl.d/99-trixy-ipforward.conf", sysctl_conf, ) # Zur Laufzeit sofort aktivieren — das stoert nix await self._run_sudo("sysctl", "-w", "net.ipv4.ip_forward=1", check=False) # NAT-Regel (wird nach Reboot von ifupdown-persistent / netfilter-persistent geladen) await self._run_sudo( "iptables", "-t", "nat", "-A", "POSTROUTING", "-o", "eth0", "-j", "MASQUERADE", check=False, ) # iptables-Regeln persistieren wenn netfilter-persistent verfuegbar if shutil.which("netfilter-persistent"): await self._run_sudo("netfilter-persistent", "save", check=False) # --- 7. Services enable (NICHT start!) --- # Bewusst nur enable — der Pi bleibt auf seiner aktuellen # Netzwerk-Verbindung bis zum Reboot. Das schuetzt die SSH-Session # des Users. await self._run_sudo("systemctl", "unmask", "hostapd") await self._run_sudo("systemctl", "enable", "hostapd") await self._run_sudo("systemctl", "enable", "dnsmasq") await self._run_sudo("systemctl", "enable", "systemd-networkd") await self._run_sudo("systemctl", "daemon-reload") self._log(" -> hostapd, dnsmasq, systemd-networkd: enable (deferred)") # --- Reboot-Flag setzen --- self.reboot_required = True self.reboot_reasons.append( f"Server wird nach Reboot zum WLAN-Hotspot '{cfg.wifi_ssid}' " f"auf IP {cfg.ap_ip}" ) self._log("") self._log(" HINWEIS: Netzwerk-Umschaltung erst nach Reboot aktiv.") self._log(" Die aktuelle SSH-Verbindung bleibt bestehen.") return True @staticmethod def _parse_requirements(path: str) -> list[str]: """ Liest eine requirements.txt und gibt aktive Paketnamen zurueck. Ignoriert leere Zeilen, Kommentare (#) und Options-Zeilen (--, -r). """ packages: list[str] = [] try: with open(path) as f: for line in f: line = line.strip() # Leer, Kommentar, Option if not line or line.startswith("#") or line.startswith("-"): continue # Inline-Kommentar entfernen: "numpy # Array-Verarbeitung" if " #" in line: line = line.split(" #")[0].strip() if line: packages.append(line) except OSError: pass return packages async def _pip_install_packages( self, pip: str, packages: list[str], label: str = "", ) -> tuple[int, int]: """ Installiert Pakete einzeln via pip. Jedes Paket wird separat installiert — so bleibt /tmp sauber und der Fortschritt ist sichtbar. Args: pip: Pfad zu pip im venv packages: Liste der zu installierenden Pakete label: Kontext-Label fuer Log (z.B. "Kern" oder Plugin-Name) Returns: (erfolgreich, fehlgeschlagen) """ success = 0 failed = 0 total = len(packages) for i, pkg in enumerate(packages, 1): if self._cancelled: return success, failed prefix = f" [{i}/{total}]" if total > 1 else " " self._log(f"{prefix} {pkg}...") code, _ = await self._run_cmd(pip, "install", pkg) if code == 0: success += 1 else: failed += 1 self._log(f" WARNUNG: {pkg} fehlgeschlagen") return success, failed async def step_venv(self) -> bool: """Erstellt Python venv und installiert Requirements einzeln.""" venv = self._config.venv_path source = self._config.source_path # venv erstellen self._log(" Erstelle Python venv...") code, _ = await self._run_cmd("python3", "-m", "venv", venv) if code != 0: return False # pip aktualisieren pip = f"{venv}/bin/pip" self._log(" Aktualisiere pip...") code, _ = await self._run_cmd(pip, "install", "--upgrade", "pip", "setuptools", "wheel") if code != 0: return False # Kern-Requirements einzeln installieren req_file = f"{source}/requirements.txt" packages = self._parse_requirements(req_file) if not packages: self._log(" Keine Requirements gefunden") return True self._log(f" Installiere {len(packages)} Kern-Pakete...\n") ok, fail = await self._pip_install_packages(pip, packages, "Kern") self._log(f"\n Ergebnis: {ok} installiert, {fail} fehlgeschlagen") if fail > 0: self._log(" WARNUNG: Einige Pakete konnten nicht installiert werden") # Nicht abbrechen — manche Pakete sind optional (z.B. kenlm) return True # Zusaetzliche pip-Pakete pro Plugin die nicht in requirements.txt stehen # (weil sie auskommentiert sind oder backend-abhaengig) _PLUGIN_EXTRA_DEPS: dict[str, list[str]] = { "stt_vosk": ["vosk"], "stt_whisper": ["openai-whisper"], "tts_coqui": ["TTS"], "tts_piper": ["piper-tts"], "tts_google": ["google-cloud-texttospeech"], "nlp_classifier": ["sentence-transformers", "onnxruntime", "transformers"], } # apt/system-Pakete die vor pip install noetig sind (fuer Kompilierung) _PLUGIN_SYSTEM_DEPS: dict[str, list[str]] = { "nlp_llm": ["cmake"], # llama-cpp-python braucht cmake zum Bauen } async def step_plugin_deps(self) -> bool: """Installiert Plugin-Abhaengigkeiten fuer ALLE vorhandenen Plugins. Durchsucht das plugins/ Verzeichnis rekursiv nach requirements.txt Dateien und installiert die darin enthaltenen Pakete. Zusaetzliche Backend-spezifische Pakete werden aus _PLUGIN_EXTRA_DEPS ergaenzt. Plugins ohne requirements.txt werden uebersprungen. """ import os pip = f"{self._config.venv_path}/bin/pip" source = self._config.source_path cfg = self._config plugins_dir = f"{source}/plugins" total_ok = 0 total_fail = 0 if not os.path.isdir(plugins_dir): self._log(f" Kein plugins/ Verzeichnis gefunden: {plugins_dir}") return True # Alle Plugin-Ordner ermitteln (direkte Unterordner) try: plugin_names = sorted( name for name in os.listdir(plugins_dir) if os.path.isdir(os.path.join(plugins_dir, name)) and not name.startswith((".", "_")) ) except OSError as e: self._log(f" FEHLER beim Lesen des Plugin-Verzeichnisses: {e}") return False self._log(f" Gefunden: {len(plugin_names)} Plugins in {plugins_dir}") for plugin in plugin_names: if self._cancelled: return False # 1. requirements.txt parsen req_file = f"{plugins_dir}/{plugin}/requirements.txt" packages = self._parse_requirements(req_file) # 2. Zusaetzliche Pakete (Backend-spezifisch) hinzufuegen extras = self._PLUGIN_EXTRA_DEPS.get(plugin, []) packages.extend(extras) if not packages: continue # Plugin hat keine Abhaengigkeiten self._log(f"\n Plugin: {plugin} ({len(packages)} Pakete)") ok, fail = await self._pip_install_packages(pip, packages, plugin) total_ok += ok total_fail += fail # 3. NLP Backend — spezielle Behandlung (nur wenn nlp_llm vorhanden) if "nlp_llm" in plugin_names: await self._install_nlp_backend() if total_ok > 0 or total_fail > 0: self._log(f"\n Plugin-Pakete: {total_ok} installiert, {total_fail} fehlgeschlagen") return True async def _install_nlp_backend(self) -> None: """Installiert das gewaehlte NLP-Backend.""" pip = f"{self._config.venv_path}/bin/pip" source = self._config.source_path backend = self._config.nlp_backend self._log(f" NLP Backend: {backend}") if backend == "llama_cpp": # cmake als System-Dep (fuer C++ Kompilierung) pm_name, _, install_cmd, _ = self._detect_pkg_manager() if pm_name: cmake_pkg = self._resolve_pkg("cmake", pm_name) if "cmake" in self._PKG_MAP else "cmake" self._log(f" Installiere cmake (Build-Abhaengigkeit)...") await self._run_sudo(*install_cmd, cmake_pkg, check=False) # llama-cpp-python installieren (kann auf ARM lange dauern) self._log(f" Installiere llama-cpp-python (kann 10-30 Min dauern auf ARM)...") code, _ = await self._run_cmd( pip, "install", "llama-cpp-python", ) if code != 0: self._log(f" FEHLER: llama-cpp-python Installation fehlgeschlagen") self._log(f" Alternativ: Backend auf 'ollama' umstellen") else: self._log(f" -> llama-cpp-python installiert") elif backend in ("ollama", "ollama_remote"): # aiohttp fuer Ollama REST-API self._log(f" Installiere aiohttp (Ollama Client)...") code, _ = await self._run_cmd(pip, "install", "aiohttp") if code != 0: self._log(f" WARNUNG: aiohttp fehlgeschlagen") if backend == "ollama": # Lokales Ollama — pruefen ob installiert if not shutil.which("ollama"): self._log(f" WARNUNG: Ollama nicht gefunden!") self._log(f" Installieren mit: curl -fsSL https://ollama.com/install.sh | sh") self._log(f" Danach: ollama pull qwen2.5:1.5b") else: self._log(f" -> Ollama gefunden: {shutil.which('ollama')}") else: # Externer Ollama-Server host = self._config.ollama_host self._log(f" Externer Ollama-Server: {host}") self._log(f" Stelle sicher dass der Server erreichbar ist und") self._log(f" das Modell geladen wurde (ollama pull qwen2.5:1.5b)") # 4. Config anpassen # ollama_remote wird intern als "ollama" gespeichert (selbes Backend, anderer Host) config_backend = "ollama" if backend == "ollama_remote" else backend config_file = f"{source}/plugins/nlp_llm/config.json" if os.path.isfile(config_file): try: import json with open(config_file) as f: nlp_cfg = json.load(f) nlp_cfg["backend"] = config_backend nlp_cfg["ollama_host"] = self._config.ollama_host if config_backend == "ollama": nlp_cfg["model_name"] = nlp_cfg.get("model_name", "qwen2.5:1.5b") with open(config_file, "w") as f: json.dump(nlp_cfg, f, indent=4, ensure_ascii=False) self._log(f" -> config.json: backend={config_backend}, host={self._config.ollama_host}") except Exception as e: self._log(f" WARNUNG: Config-Update fehlgeschlagen: {e}") async def step_directories(self) -> bool: """Erstellt Verzeichnisse und setzt Berechtigungen.""" source = self._config.source_path mode = self._config.mode # Basis-Verzeichnisse (alle Modi) dirs = [ "data", "logs", "cache", "certs", ] # Modell-Verzeichnisse — exakte Struktur, verhindert Tippfehler model_dirs = [ "models", "models/wakeword", "models/wakeword/trixy", "models/wakeword/system_command", "models/wakeword/alexa", "models/wakeword/hey_jarvis", "models/wakeword/hey_mycroft", "models/piper", "models/nlp", "models/symspell", ] # Server/Standalone-spezifisch if mode in ("server", "standalone"): dirs.extend([ "satellites", "sync_data", "assets", "assets/default", ]) # Plugin-Modell-Verzeichnisse for plugin in self._config.selected_plugins: dirs.append(f"plugins/{plugin}/models") # Client-spezifisch if mode == "client": dirs.extend([ "assets", ]) # Trainer-Verzeichnisse (Server/Standalone) if mode in ("server", "standalone"): dirs.extend([ "trainer/data/raw", "trainer/data/processed", "trainer/cache", "trainer/assets", ]) all_dirs = dirs + model_dirs self._log(f" Erstelle {len(all_dirs)} Verzeichnisse...") for d in all_dirs: path = f"{source}/{d}" await self._run_sudo("mkdir", "-p", path) # Shell-Scripte ausfuehrbar machen scripts = [ "run_linux.sh", "install_linux.sh", "install_requirements.sh", "bash/update_linux.sh", "bash/prepare_linux.sh", ] made_exec = 0 for script in scripts: path = f"{source}/{script}" if os.path.isfile(path): await self._run_cmd("chmod", "+x", path) made_exec += 1 # User zur 'input'-Gruppe hinzufuegen (fuer HID Media Keys / /dev/input/event*) if mode in ("client", "standalone"): code, _ = await self._run_sudo( "usermod", "-aG", "input", self._config.username, check=False, ) if code == 0: self._log(f" -> User '{self._config.username}' zur Gruppe 'input' hinzugefuegt (HID)") # Besitzer setzen await self._run_sudo( "chown", "-R", f"{self._config.username}:{self._config.username}", self._config.install_path, ) self._log(f" -> {len(all_dirs)} Verzeichnisse erstellt, {made_exec} Scripte ausfuehrbar") return True async def step_systemd(self) -> bool: """Erstellt den systemd-Service und aktiviert ihn bei Bedarf.""" cfg = self._config if not cfg.install_service: self._log(" Service-Installation deaktiviert — uebersprungen") return True service_name = f"trixy-{cfg.mode}" # ExecStart zusammenbauen # run_linux.sh fuehrt Update + Bereinigung + Start durch run_script = f"{cfg.source_path}/run_linux.sh" exec_args = ["/bin/bash", run_script] if cfg.mode == "client": exec_args.extend([ "client", "--host", cfg.server_host, "--port", str(cfg.server_port), "--room", cfg.room, "--alias", cfg.alias, ]) else: exec_args.append(cfg.mode) if cfg.auto_regist and cfg.mode in ("server", "standalone"): exec_args.append("--auto-regist") exec_args.append("--debug") exec_start = " ".join(exec_args) # After-Deps fuer Server mit WiFi-AP after_deps = "network-online.target" if cfg.mode == "server" and cfg.wifi_enabled: after_deps = "network-online.target hostapd.service dnsmasq.service" service_content = f"""[Unit] Description=Trixy Voice Assistant ({cfg.mode.title()}) After={after_deps} Wants=network-online.target [Service] Type=simple User={cfg.username} Group={cfg.username} WorkingDirectory={cfg.source_path} ExecStart={exec_start} Restart=on-failure RestartSec=5 StartLimitIntervalSec=60 StartLimitBurst=5 Environment=PYTHONUNBUFFERED=1 Environment=HOME=/home/{cfg.username} StandardOutput=journal StandardError=journal SyslogIdentifier=trixy [Install] WantedBy=multi-user.target """ # Service-Datei schreiben proc = await asyncio.create_subprocess_exec( "sudo", "tee", f"/etc/systemd/system/{service_name}.service", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, ) await proc.communicate(service_content.encode()) await self._run_sudo("systemctl", "daemon-reload") # Autostart je nach Einstellung if cfg.autostart: await self._run_sudo("systemctl", "enable", f"{service_name}.service") self._log(f" -> {service_name}.service aktiviert (Autostart)") else: await self._run_sudo("systemctl", "disable", f"{service_name}.service") self._log(f" -> {service_name}.service erstellt (kein Autostart)") self._log(f" Manuell starten: sudo systemctl start {service_name}") return True def _get_model_downloads(self) -> list[tuple[str, str, str, bool, str | None]]: """ Gibt die Download-Liste basierend auf Hardware-Empfehlung zurueck. Returns: Liste von (name, url, ziel_relativ_zu_source, entpacken?, plugin_name). plugin_name ist None fuer Core-Modelle. """ from .system_info import detect_system, get_model_recommendations info = detect_system() recs = get_model_recommendations(info, self._config.selected_plugins) downloads: list[tuple[str, str, str, bool, str | None]] = [] # Empfohlene Modelle ermitteln rec_names = {r.name for r in recs if r.recommended} self._log(f" Hardware: {info.ram_total_mb}MB RAM, {info.disk_free_mb}MB frei") if info.gpu.cuda_available: self._log(f" GPU: {info.gpu.name} ({info.gpu.vram_mb}MB VRAM)") else: self._log(" GPU: Keine CUDA-GPU erkannt") self._log(f" {len(rec_names)} empfohlene Modelle erkannt") self._log("") # --- STT Vosk --- if "stt_vosk" in self._config.selected_plugins: if "Vosk STT (Deutsch, gross)" in rec_names: downloads.append(( "Vosk STT (Deutsch, gross)", "https://alphacephei.com/vosk/models/vosk-model-de-0.21.zip", "plugins/stt_vosk/models", True, "stt_vosk", )) elif "Vosk STT (Deutsch, klein)" in rec_names: downloads.append(( "Vosk STT (Deutsch, klein)", "https://alphacephei.com/vosk/models/vosk-model-small-de-0.15.zip", "plugins/stt_vosk/models", True, "stt_vosk", )) # --- STT Whisper --- if "stt_whisper" in self._config.selected_plugins: # Bestes empfohlenes Whisper-Modell waehlen whisper_priority = ["medium", "small", "base", "tiny"] for model_name in whisper_priority: full_name = f"Whisper STT ({model_name})" if full_name in rec_names: self._log(f" Whisper: {model_name} empfohlen (wird beim Start geladen)") break # --- TTS Piper --- if "tts_piper" in self._config.selected_plugins: if "Piper TTS (thorsten-medium)" in rec_names: downloads.append(( "Piper TTS (thorsten-medium, onnx)", "https://huggingface.co/rhasspy/piper-voices/resolve/main/" "de/de_DE/thorsten/medium/de_DE-thorsten-medium.onnx", "models/piper/de_DE-thorsten-medium.onnx", False, None, )) downloads.append(( "Piper TTS (thorsten-medium, config)", "https://huggingface.co/rhasspy/piper-voices/resolve/main/" "de/de_DE/thorsten/medium/de_DE-thorsten-medium.onnx.json", "models/piper/de_DE-thorsten-medium.onnx.json", False, None, )) # --- NLP LLM --- if "nlp_llm" in self._config.selected_plugins: if "Qwen 2.5 1.5B (Q4_K_M)" in rec_names: downloads.append(( "Qwen 2.5 1.5B (Q4_K_M)", "https://huggingface.co/Qwen/Qwen2.5-1.5B-Instruct-GGUF/resolve/main/" "qwen2.5-1.5b-instruct-q4_k_m.gguf", "models/nlp/qwen2.5-1.5b-instruct-q4_k_m.gguf", False, None, )) return downloads async def step_download_models(self) -> bool: """Laedt ML-Modelle basierend auf Hardware-Empfehlung herunter.""" if not self._config.download_models: self._log(" Modell-Download deaktiviert — uebersprungen") return True source = self._config.source_path downloads = self._get_model_downloads() if not downloads: self._log(" Keine Modelle zum Herunterladen (keine passenden Empfehlungen)") return True downloaded = 0 total_size = 0 for name, url, target_rel, extract, plugin in downloads: if self._cancelled: return False target = f"{source}/{target_rel}" # Pruefen ob bereits vorhanden if not extract and os.path.isfile(target): self._log(f" {name} — bereits vorhanden") continue if extract and os.path.isdir(target) and os.listdir(target): self._log(f" {name} — bereits vorhanden") continue self._log(f" Lade herunter: {name}...") if extract: # ZIP herunterladen und entpacken zip_path = "/tmp/trixy_model_download.zip" code, _ = await self._run_cmd( "wget", "-q", "--show-progress", "-O", zip_path, url, ) if code != 0: self._log(f" WARNUNG: Download fehlgeschlagen — {name}") continue await self._run_sudo("mkdir", "-p", target) code, _ = await self._run_cmd( "unzip", "-q", "-o", zip_path, "-d", target, ) if os.path.exists(zip_path): os.unlink(zip_path) if code != 0: self._log(f" WARNUNG: Entpacken fehlgeschlagen — {name}") continue else: # Direkt herunterladen target_dir = os.path.dirname(target) await self._run_sudo("mkdir", "-p", target_dir) code, _ = await self._run_cmd( "wget", "-q", "--show-progress", "-O", target, url, ) if code != 0: self._log(f" WARNUNG: Download fehlgeschlagen — {name}") continue # Groesse ermitteln if os.path.isfile(target): total_size += os.path.getsize(target) // (1024 * 1024) elif os.path.isdir(target): for dirpath, _, filenames in os.walk(target): for f in filenames: total_size += os.path.getsize(os.path.join(dirpath, f)) // (1024 * 1024) downloaded += 1 self._log(f" -> {name} installiert") self._log(f" {downloaded} Modell(e) heruntergeladen ({total_size} MB)") return True async def step_verify(self) -> bool: """Verifiziert die Installation.""" checks = ["ffmpeg", "python3"] all_ok = True for cmd in checks: path = shutil.which(cmd) if path: self._log(f" OK: {cmd} ({path})") else: self._log(f" FEHLT: {cmd}") all_ok = False # venv pruefen venv_python = f"{self._config.venv_path}/bin/python" if os.path.isfile(venv_python): self._log(f" OK: venv ({venv_python})") else: self._log(f" FEHLT: venv ({venv_python})") all_ok = False # Temporaere Verzeichnisse aufraeumen (pip-Cache, TMPDIR) for tmp_name in ("tmp", "pip-cache"): tmp_path = f"{self._config.install_path}/{tmp_name}" if os.path.isdir(tmp_path): await self._run_sudo("rm", "-rf", tmp_path) self._log(f" Aufgeraeumt: {tmp_name}") return all_ok