# -*- 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 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_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"}, "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 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).""" if not self._config.wifi_connect: self._log(" Keine WLAN-Aenderung — uebersprungen") return True ssid = self._config.wifi_connect_ssid password = self._config.wifi_connect_password self._log(f" Konfiguriere WLAN: {ssid}") # Methode 1: nmcli (NetworkManager — modernes Debian/Ubuntu/Raspberry Pi OS) if shutil.which("nmcli"): self._log(" Verwende NetworkManager (nmcli)...") # Bestehende Verbindung mit gleichem Namen entfernen await self._run_cmd( "nmcli", "connection", "delete", ssid, check=False, ) # Neue Verbindung erstellen code, _ = await self._run_sudo( "nmcli", "device", "wifi", "connect", ssid, "password", password, ) if code == 0: self._log(f" -> WLAN verbunden: {ssid}") return True else: self._log(f" WARNUNG: nmcli fehlgeschlagen — versuche wpa_supplicant") # Methode 2: wpa_supplicant (aeltere Systeme / Raspberry Pi OS Lite) self._log(" Verwende wpa_supplicant...") wpa_conf = "/etc/wpa_supplicant/wpa_supplicant.conf" # wpa_passphrase generiert verschluesselten PSK code, output = await self._run_cmd( "wpa_passphrase", ssid, password, check=False, ) if code != 0: # Fallback: Klartext-Config network_block = f""" network={{ ssid="{ssid}" psk="{password}" key_mgmt=WPA-PSK }} """ else: network_block = output # Pruefen ob Basis-Header vorhanden wpa_header = ( 'ctrl_interface=DIR=/var/run/wpa_supplicant GROUP=netdev\n' 'update_config=1\n' 'country=DE\n' ) # Config schreiben proc = await asyncio.create_subprocess_exec( "sudo", "tee", wpa_conf, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, ) full_config = wpa_header + "\n" + network_block await proc.communicate(full_config.encode()) # wlan0 neu verbinden await self._run_sudo("wpa_cli", "-i", "wlan0", "reconfigure", check=False) self._log(f" -> WLAN konfiguriert: {ssid}") return True async def step_wifi_ap(self) -> bool: """Konfiguriert WiFi Access Point.""" if not self._config.wifi_enabled: self._log(" WiFi-Hotspot deaktiviert — uebersprungen") return True cfg = self._config # hostapd konfigurieren hostapd_conf = f"""# Trixy Access Point 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 """ # Ueber tee schreiben (braucht sudo) proc = await asyncio.create_subprocess_exec( "sudo", "tee", "/etc/hostapd/hostapd.conf", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, ) await proc.communicate(hostapd_conf.encode()) # dnsmasq konfigurieren dnsmasq_conf = f"""# Trixy DHCP-Server 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 address=/{cfg.wifi_ssid}/{cfg.ap_ip} """ proc = await asyncio.create_subprocess_exec( "sudo", "tee", "/etc/dnsmasq.conf", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, ) await proc.communicate(dnsmasq_conf.encode()) # dhcpcd — statische IP dhcpcd_block = f""" # Trixy AP — wlan0 statische IP interface wlan0 static ip_address={cfg.ap_ip}/24 nohook wpa_supplicant """ proc = await asyncio.create_subprocess_exec( "sudo", "tee", "-a", "/etc/dhcpcd.conf", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, ) await proc.communicate(dhcpcd_block.encode()) # IP-Forwarding und NAT await self._run_sudo("sysctl", "-w", "net.ipv4.ip_forward=1") await self._run_sudo( "iptables", "-t", "nat", "-A", "POSTROUTING", "-o", "eth0", "-j", "MASQUERADE", ) # Services aktivieren await self._run_sudo("systemctl", "unmask", "hostapd") await self._run_sudo("systemctl", "enable", "hostapd") await self._run_sudo("systemctl", "enable", "dnsmasq") 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 einzeln inkl. Backend-spezifischer Pakete.""" pip = f"{self._config.venv_path}/bin/pip" source = self._config.source_path cfg = self._config total_ok = 0 total_fail = 0 for plugin in cfg.selected_plugins: if self._cancelled: return False # 1. requirements.txt parsen und einzeln installieren req_file = f"{source}/plugins/{plugin}/requirements.txt" packages = self._parse_requirements(req_file) # 2. Zusaetzliche Pakete hinzufuegen extras = self._PLUGIN_EXTRA_DEPS.get(plugin, []) packages.extend(extras) if not packages: continue 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 if "nlp_llm" in cfg.selected_plugins: 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