Trixy7/trixy_core/tui/views/sat_updates_view.py

# -*- coding: utf-8 -*-
"""
Satellite SubView: Updates (F5).

Bietet Buttons fuer verschiedene Update- und Steuerungs-Aktionen:
Wakeword-Modell, Voice-Recognition-Modell, Script-Update,
System-Reboot, Asset-Updates, Plugin-Updates, Test-Funktionen.
"""

from __future__ import annotations

from textual.app import ComposeResult
from textual.containers import Vertical, Horizontal
from textual.widgets import Static, Button

from trixy_core.tui.views.base import MainView
from trixy_core.audio.player import BlockingStreamPlayer, PlayerConfig


class SatUpdatesView(MainView):
    """F5 im Satellite-Kontext: Updates und Steuerung."""

    TITLE = "Updates"
    VIEW_ID = "sat_updates"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._sat_data: dict = {}
        self._mic_test_active: bool = False
        self._mic_test_chunk_count: int = 0
        self._music_playing: bool = False
        # Audio-Player fuer Mikrofon-Test (16KHz, mono, 16-bit)
        self._mic_player: BlockingStreamPlayer | None = None

    def set_satellite(self, sat_data: dict) -> None:
        """Setzt die Satellite-Basisdaten."""
        self._sat_data = sat_data

    def compose(self) -> ComposeResult:
        with Vertical(classes="section-box"):
            yield Static("[bold]Modell-Updates[/bold]", classes="section-header")
            with Horizontal(classes="sat-button-row"):
                yield Button("Wakeword-Modell", id="sat-upd-wakeword", variant="primary")
                yield Button("Voice-Recognition", id="sat-upd-voicerec", variant="primary")

        with Vertical(classes="section-box"):
            yield Static("[bold]System[/bold]", classes="section-header")
            with Horizontal(classes="sat-button-row"):
                yield Button("Script-Update", id="sat-upd-scripts", variant="primary")
                yield Button("System Neustart", id="sat-upd-reboot", variant="error")

        with Vertical(classes="section-box"):
            yield Static("[bold]Assets & Plugins[/bold]", classes="section-header")
            with Horizontal(classes="sat-button-row"):
                yield Button("Assets aktualisieren", id="sat-upd-assets", variant="primary")
                yield Button("Plugins aktualisieren", id="sat-upd-plugins", variant="primary")

        with Vertical(classes="section-box"):
            yield Static("[bold]Test & Diagnose[/bold]", classes="section-header")
            with Horizontal(classes="sat-button-row"):
                yield Button("Wakeword simulieren", id="sat-upd-fake-wakeword", variant="warning")
                yield Button("Test-Audio abspielen", id="sat-upd-test-audio", variant="warning")
                yield Button("Test Mikrofon", id="sat-upd-mic-test", variant="warning")
            with Horizontal(classes="sat-button-row"):
                yield Button("Musik abspielen", id="sat-upd-music-toggle", variant="warning")

        yield Static("", id="sat-upd-status")

    async def on_activate(self) -> None:
        """Aktiviert den View."""
        self._update_status("")

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        """Verarbeitet Button-Klicks."""
        btn_id = event.button.id
        if not self.connection or not self._sat_data:
            self._update_status("Keine Verbindung oder kein Satellite", error=True)
            return

        sat_id = self._sat_data.get("satellite_id", "")

        if btn_id == "sat-upd-wakeword":
            await self._execute_action(sat_id, "update_wakeword_model")
        elif btn_id == "sat-upd-voicerec":
            await self._execute_action(sat_id, "update_voice_recognition_model")
        elif btn_id == "sat-upd-scripts":
            await self._execute_action(sat_id, "update_scripts")
        elif btn_id == "sat-upd-reboot":
            await self._execute_action(sat_id, "reboot")
        elif btn_id == "sat-upd-assets":
            await self._execute_action(sat_id, "update_assets")
        elif btn_id == "sat-upd-plugins":
            await self._execute_action(sat_id, "update_plugins")
        elif btn_id == "sat-upd-fake-wakeword":
            await self._execute_action(sat_id, "fake_wakeword")
        elif btn_id == "sat-upd-test-audio":
            await self._execute_action(sat_id, "test_audio")
        elif btn_id == "sat-upd-music-toggle":
            await self._toggle_music(sat_id)
        elif btn_id == "sat-upd-mic-test":
            await self._toggle_mic_test(sat_id)

    async def _execute_action(self, satellite_id: str, action: str) -> None:
        """Fuehrt eine Satellite-Aktion aus."""
        try:
            result = await self.connection.satellite_action(
                action=action,
                satellite_id=satellite_id,
            )
            if result and result.success:
                self._update_status(result.message or f"Aktion '{action}' erfolgreich")
            else:
                error = result.error if result else "Unbekannter Fehler"
                self._update_status(f"Fehler: {error}", error=True)
        except Exception as e:
            self._update_status(f"Fehler: {e}", error=True)

    async def _toggle_music(self, satellite_id: str) -> None:
        """Startet oder stoppt die Musik-Wiedergabe."""
        btn = self.query_one("#sat-upd-music-toggle", Button)
        if self._music_playing:
            try:
                result = await self.connection.satellite_action(
                    action="stop_music", satellite_id=satellite_id,
                )
                if result and result.success:
                    self._music_playing = False
                    btn.label = "Musik abspielen"
                    btn.variant = "warning"
                    self._update_status(result.message or "Musik gestoppt")
                else:
                    self._update_status(f"Fehler: {result.error if result else 'Unbekannt'}", error=True)
            except Exception as e:
                self._update_status(f"Fehler: {e}", error=True)
        else:
            try:
                result = await self.connection.satellite_action(
                    action="play_music", satellite_id=satellite_id,
                )
                if result and result.success:
                    self._music_playing = True
                    btn.label = "Musik stoppen"
                    btn.variant = "error"
                    self._update_status(result.message or "Musik wird abgespielt")
                else:
                    self._update_status(f"Fehler: {result.error if result else 'Unbekannt'}", error=True)
            except Exception as e:
                self._update_status(f"Fehler: {e}", error=True)

    async def _toggle_mic_test(self, satellite_id: str) -> None:
        """Startet oder stoppt den Mikrofon-Test."""
        if self._mic_test_active:
            await self._stop_mic_test(satellite_id)
        else:
            await self._start_mic_test(satellite_id)

    async def _start_mic_test(self, satellite_id: str) -> None:
        """Startet den Mikrofon-Test."""
        self._update_status("Mikrofon-Test wird gestartet (TTS wird gesendet)...")
        self._mic_test_chunk_count = 0

        # Audio-Player erstellen (16KHz, mono, 16-bit — wie Satellite-Audio)
        self._mic_player = BlockingStreamPlayer(PlayerConfig(
            sample_rate=16000,
            channels=1,
            sample_width=2,
            prebuffer_ms=200,
        ))

        try:
            result = await self.connection.start_mic_test(
                satellite_id=satellite_id,
                audio_callback=self._on_mic_test_audio,
            )
            if result and result.success:
                self._mic_test_active = True
                btn = self.query_one("#sat-upd-mic-test", Button)
                btn.label = "Stop Test"
                btn.variant = "error"
                self._update_status("Mikrofon-Test laeuft — Audio wird empfangen")
            else:
                error = result.error if result else "Keine Antwort"
                self._update_status(f"Fehler: {error}", error=True)
                if self._mic_player:
                    self._mic_player.stop()
                    self._mic_player = None
        except Exception as e:
            self._update_status(f"Fehler: {e}", error=True)
            if self._mic_player:
                self._mic_player.stop()
                self._mic_player = None

    async def _stop_mic_test(self, satellite_id: str) -> None:
        """Stoppt den Mikrofon-Test."""
        try:
            await self.connection.stop_mic_test(satellite_id=satellite_id)
        except Exception:
            pass

        self._mic_test_active = False

        # Audio-Player stoppen
        if self._mic_player:
            self._mic_player.finish()
            self._mic_player = None

        btn = self.query_one("#sat-upd-mic-test", Button)
        btn.label = "Test Mikrofon"
        btn.variant = "warning"
        chunks = self._mic_test_chunk_count
        self._update_status(f"Mikrofon-Test beendet ({chunks} Audio-Pakete empfangen)")

    def _on_mic_test_audio(self, audio_data: bytes) -> None:
        """Callback fuer empfangene Mikrofon-Test Audio-Daten."""
        self._mic_test_chunk_count += 1

        # Audio abspielen
        if self._mic_player and self._mic_test_active:
            self._mic_player.add_chunk(audio_data)

    async def on_deactivate(self) -> None:
        """Stoppt den Mikrofon-Test beim Verlassen des Views."""
        if self._mic_test_active and self.connection and self._sat_data:
            sat_id = self._sat_data.get("satellite_id", "")
            if sat_id:
                await self._stop_mic_test(sat_id)
        elif self._mic_player:
            self._mic_player.stop()
            self._mic_player = None

    def _update_status(self, text: str, error: bool = False) -> None:
        """Aktualisiert die Statuszeile."""
        status = self.query_one("#sat-upd-status", Static)
        if error:
            status.update(f"[red]{text}[/red]")
        elif text:
            status.update(f"[green]{text}[/green]")
        else:
            status.update("")