Trixy7/trixy_core/tui/views/sat_conversation_view.py

# -*- coding: utf-8 -*-
"""
Satellite SubView: Conversation (F3).

Zeigt Informationen zu Wakeword-Erkennung, Voice Recognition
und aktiver Konversation.
"""

from __future__ import annotations

from textual.app import ComposeResult
from textual.containers import Vertical
from textual.widgets import Static

from trixy_core.tui.views.base import MainView
from trixy_core.tui.widgets.status_table import StatusTable


class SatConversationView(MainView):
    """F3 im Satellite-Kontext: Konversations-Informationen."""

    TITLE = "Konversation"
    VIEW_ID = "sat_conversation"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._sat_data: dict = {}
        self._detail_data: object | None = None
        self._refresh_counter: int = 0

    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]Wakeword[/bold]", classes="section-header")
            yield StatusTable(id="sat-conv-wakeword")

        with Vertical(classes="section-box"):
            yield Static("[bold]Spracherkennung[/bold]", classes="section-header")
            yield StatusTable(id="sat-conv-voicerec")

        with Vertical(classes="section-box"):
            yield Static("[bold]Konversation[/bold]", classes="section-header")
            yield StatusTable(id="sat-conv-conversation")

        with Vertical(classes="section-box"):
            yield Static("[bold]Verlauf[/bold]", classes="section-header")
            yield Static("", id="sat-conv-history")

    async def on_activate(self) -> None:
        """Laedt und zeigt die Konversations-Infos."""
        await self._load_detail()

    async def refresh_data(self) -> None:
        """Periodisches Update (alle 4 Sekunden fuer Live-Status)."""
        self._refresh_counter += 1
        if self._refresh_counter % 2 == 0:
            await self._load_detail()

    async def _load_detail(self) -> None:
        """Laedt detaillierte Satellite-Daten vom Server."""
        if not self.connection or not self._sat_data:
            return

        sat_id = self._sat_data.get("satellite_id", "")
        if not sat_id:
            return

        try:
            detail = await self.connection.request_satellite_detail(sat_id)
        except Exception:
            return

        if detail is None or detail.error:
            return

        self._detail_data = detail
        self._update_display()

    def _update_display(self) -> None:
        """Aktualisiert die Anzeige."""
        d = self._detail_data
        if d is None:
            return

        # Wakeword
        wakeword = self.query_one("#sat-conv-wakeword", StatusTable)
        wakeword.update_data([
            ("Modell", d.wakeword_model or "Unbekannt"),
            ("Letzte Erkennung", d.wakeword_last_detection or "Nie"),
        ])

        # Spracherkennung
        voicerec = self.query_one("#sat-conv-voicerec", StatusTable)
        voicerec.update_data([
            ("Modell", d.voice_recognition_model or "Unbekannt"),
            ("Letzter Sprecher", d.voice_recognition_last_speaker or "Unbekannt"),
        ])

        # Konversation
        conv = self.query_one("#sat-conv-conversation", StatusTable)
        if d.has_active_conversation:
            conv_status = "[yellow]Aktiv[/yellow]"
        else:
            conv_status = "Keine"

        rows = [
            ("Status", conv_status),
        ]

        if d.last_conversation_at:
            rows.append(("Letzte Konversation", d.last_conversation_at))
        if d.last_conversation_duration > 0:
            rows.append(("Dauer", f"{d.last_conversation_duration:.1f}s"))

        conv.update_data(rows)

        # Conversation-History
        history_widget = self.query_one("#sat-conv-history", Static)
        history = getattr(d, "conversation_history", [])
        if history:
            lines = []
            for entry in history:
                ts = entry.get("created_at", "")[:19].replace("T", " ")
                state = entry.get("state", "")
                duration = entry.get("duration_seconds", 0)
                turn_count = entry.get("turn_count", 0)
                wakeword = entry.get("wakeword_type", "")

                header = f"[bold]{ts}[/bold] ({state}, {duration:.1f}s, {turn_count} Turns)"
                if wakeword:
                    header += f" [dim]Wakeword: {wakeword}[/dim]"
                lines.append(header)

                # Turns anzeigen
                for turn in entry.get("turns", []):
                    turn_type = turn.get("turn_type", "")
                    content = turn.get("content", "")
                    if not content:
                        continue
                    if turn_type in ("user_speech", "user_text"):
                        lines.append(f"  [cyan]User:[/cyan] {content}")
                    elif turn_type == "follow_up":
                        lines.append(f"  [yellow]Rueckfrage:[/yellow] {content}")
                    elif turn_type == "system":
                        lines.append(f"  [dim]System: {content}[/dim]")
                    else:
                        lines.append(f"  [green]Assistent:[/green] {content}")

                lines.append("")  # Leerzeile zwischen Sessions

            history_widget.update("\n".join(lines))
        else:
            history_widget.update("[dim]Keine Konversationen vorhanden[/dim]")