Trixy7/trixy_core/tui/views/sat_audio_view.py

# -*- coding: utf-8 -*-
"""
Satellite SubView: Audio (F6).

ALSA/PulseAudio Lautstaerke- und Mute-Steuerung pro Satellite.
Werte werden in der jeweiligen client_config.json persistiert.

Bedient sich des SatelliteAudioRequest-Hops:
    Config-Tool ── ConfigSatelliteAudioRequest ──> Server
    Server      ── SatelliteAudioRequest       ──> Satellite
    Satellite   ── SatelliteAudioResponse      ──> Server
    Server      ── ConfigSatelliteAudioResponse ──> Config-Tool

Aenderungen die hier ausgeloest werden, laufen mit source="user"
durch den VolumeHandler und werden daher in client_config.json
gespeichert.
"""

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.tui.widgets.status_table import StatusTable


VOLUME_STEP = 5


class SatAudioView(MainView):
    """F6 im Satellite-Kontext: ALSA Lautstaerke + Mute."""

    TITLE = "Audio"
    VIEW_ID = "sat_audio"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._sat_data: dict = {}
        self._status: object | None = None
        self._refresh_counter: int = 0
        # Lokale Speaker-Bearbeitung — wird beim Laden befuellt und
        # bei "Anwenden" an den Satellite gesendet.
        self._editable_speakers: list[dict] = []
        self._cursor: int = 0
        self._dirty: bool = False
        # Analog fuer Mikrofone
        self._editable_mics: list[dict] = []
        self._mic_cursor: int = 0
        self._mics_dirty: bool = False

    def set_satellite(self, sat_data: dict) -> None:
        self._sat_data = sat_data

    # === Layout ===

    def compose(self) -> ComposeResult:
        with Vertical(classes="section-box"):
            yield Static("[bold]Lautsprecher (Output)[/bold]", classes="section-header")
            yield StatusTable(id="sat-audio-output-info")
            with Horizontal(classes="button-row"):
                yield Button("− 5%", id="audio-out-down")
                yield Button("+ 5%", id="audio-out-up")
                yield Button("Mute", id="audio-out-mute")

        with Vertical(classes="section-box"):
            yield Static("[bold]Mikrofon (Input)[/bold]", classes="section-header")
            yield StatusTable(id="sat-audio-input-info")
            with Horizontal(classes="button-row"):
                yield Button("− 5%", id="audio-in-down")
                yield Button("+ 5%", id="audio-in-up")
                yield Button("Mute", id="audio-in-mute")

        with Vertical(classes="section-box"):
            yield Static("[bold]Lautsprecher-Verbund (Multi-Speaker)[/bold]", classes="section-header")
            yield Static("", id="sat-audio-speakers-header")
            yield StatusTable(id="sat-audio-speakers-list")
            with Horizontal(classes="button-row"):
                yield Button("Aktualisieren", id="audio-speakers-refresh")
                yield Button("Naechster", id="audio-speakers-cursor")
                yield Button("An/Aus", id="audio-speakers-toggle")
                yield Button("Vol −", id="audio-speakers-vol-down")
                yield Button("Vol +", id="audio-speakers-vol-up")
                yield Button("Anwenden", id="audio-speakers-apply")

        with Vertical(classes="section-box"):
            yield Static("[bold]Mikrofone (Multi-Mic Mix)[/bold]", classes="section-header")
            yield Static("", id="sat-audio-mics-header")
            yield StatusTable(id="sat-audio-mics-list")
            with Horizontal(classes="button-row"):
                yield Button("Aktualisieren", id="audio-mics-refresh")
                yield Button("Naechster", id="audio-mics-cursor")
                yield Button("An/Aus", id="audio-mics-toggle")
                yield Button("Anwenden", id="audio-mics-apply")

        with Vertical(classes="section-box"):
            yield Static("[bold]Hinweis[/bold]", classes="section-header")
            yield Static(
                "Aenderungen werden in client_config.json des Satellites "
                "gespeichert. Audio-Ducking veraendert nur die Live-"
                "Lautstaerke und stellt sie aus diesen Werten wieder her. "
                "Multi-Speaker erfordert PulseAudio oder PipeWire.",
                id="sat-audio-hint",
            )

    # === Lifecycle ===

    async def on_activate(self) -> None:
        await self._reload()

    async def refresh_data(self) -> None:
        self._refresh_counter += 1
        if self._refresh_counter % 5 == 0:
            await self._reload()

    # === Daten-Operationen ===

    def _satellite_id(self) -> str:
        return self._sat_data.get("satellite_id", "") if self._sat_data else ""

    async def _reload(self) -> None:
        sat_id = self._satellite_id()
        if not self.connection or not sat_id:
            return
        try:
            # Sinks laden — fuellt auch den Basis-Audio-Status
            sinks_response = await self.connection.request_satellite_audio(
                sat_id, action="list_pulse_sinks"
            )
            # Sources in einem zweiten Request — kommt aus einer anderen
            # Pulse-Operation und wird nicht beim list_pulse_sinks mitgeliefert
            sources_response = await self.connection.request_satellite_audio(
                sat_id, action="list_pulse_sources"
            )
        except Exception:
            return
        if sinks_response is None:
            return

        # Mergen: Basis aus Sinks-Response + Sources aus der zweiten
        self._status = sinks_response
        if sources_response is not None:
            self._status.pulse_sources = list(
                getattr(sources_response, "pulse_sources", []) or []
            )
        if not self._dirty:
            self._sync_editable_from_status()
        if not self._mics_dirty:
            self._sync_mics_from_status()
        self._update_display()

    def _sync_mics_from_status(self) -> None:
        """Baut die editierbare Mic-Liste aus dem Status auf.

        Analog zu _sync_editable_from_status: alle Pulse-Sources +
        konfigurierte aber offline Mics.
        """
        st = self._status
        if st is None:
            return
        configured = {
            m.get("device", ""): m
            for m in (getattr(st, "input_microphones", []) or [])
        }
        sources = list(getattr(st, "pulse_sources", []) or [])

        merged: list[dict] = []
        seen: set[str] = set()
        for src in sources:
            name = src.get("name", "")
            if not name:
                continue
            cfg = configured.get(name, {})
            merged.append({
                "device": name,
                "description": src.get("description", name),
                "enabled": bool(cfg.get("enabled", False)),
                "label": cfg.get("label", ""),
            })
            seen.add(name)

        # Konfigurierte Mics die aktuell offline sind
        for name, cfg in configured.items():
            if name in seen:
                continue
            merged.append({
                "device": name,
                "description": f"{name} [dim](offline)[/dim]",
                "enabled": bool(cfg.get("enabled", False)),
                "label": cfg.get("label", ""),
            })

        self._editable_mics = merged
        if self._mic_cursor >= len(merged):
            self._mic_cursor = 0

    def _sync_editable_from_status(self) -> None:
        """Baut die editierbare Speaker-Liste aus dem Status auf.

        Strategie: Alle Pulse-Sinks aufnehmen, mit den in der Config
        gespeicherten enabled/volume-Werten anreichern. So sieht der
        User immer das volle Set an verfuegbaren Sinks und kann sie
        an-/abwaehlen.
        """
        st = self._status
        if st is None:
            return
        configured = {
            sp.get("device", ""): sp
            for sp in (getattr(st, "output_speakers", []) or [])
        }
        sinks = list(getattr(st, "pulse_sinks", []) or [])

        merged: list[dict] = []
        seen: set[str] = set()
        for sink in sinks:
            name = sink.get("name", "")
            if not name:
                continue
            cfg = configured.get(name, {})
            merged.append({
                "device": name,
                "description": sink.get("description", name),
                "enabled": bool(cfg.get("enabled", False)),
                "volume": int(cfg.get("volume", 80)),
            })
            seen.add(name)

        # Konfigurierte Sinks die aktuell nicht im System sichtbar sind
        # trotzdem behalten (z.B. abgesteckter USB-Speaker)
        for name, cfg in configured.items():
            if name in seen:
                continue
            merged.append({
                "device": name,
                "description": f"{name} [dim](offline)[/dim]",
                "enabled": bool(cfg.get("enabled", False)),
                "volume": int(cfg.get("volume", 80)),
            })

        self._editable_speakers = merged
        if self._cursor >= len(merged):
            self._cursor = 0

    async def _send(
        self,
        action: str,
        target: str,
        volume: int = 0,
        muted: bool = False,
    ) -> None:
        sat_id = self._satellite_id()
        if not self.connection or not sat_id:
            return
        try:
            response = await self.connection.request_satellite_audio(
                sat_id,
                action=action,
                target=target,
                volume=volume,
                muted=muted,
            )
        except Exception:
            return
        if response is not None:
            self._status = response
            self._update_display()

    # === Anzeige ===

    def _update_display(self) -> None:
        st = self._status
        if st is None:
            return

        out_table = self.query_one("#sat-audio-output-info", StatusTable)
        if getattr(st, "output_available", False):
            volume = int(getattr(st, "output_volume", 0))
            muted = bool(getattr(st, "output_muted", False))
            mute_str = "[red]MUTE[/red]" if muted else "[green]aktiv[/green]"
            out_table.update_data([
                ("Mixer-Control", getattr(st, "output_control", "") or "?"),
                ("Lautstaerke", _bar(volume)),
                ("Status", mute_str),
            ])
        else:
            out_table.update_data([("Status", "[dim]Nicht verfuegbar[/dim]")])

        in_table = self.query_one("#sat-audio-input-info", StatusTable)
        if getattr(st, "input_available", False):
            volume = int(getattr(st, "input_volume", 0))
            muted = bool(getattr(st, "input_muted", False))
            mute_str = "[red]MUTE[/red]" if muted else "[green]aktiv[/green]"
            in_table.update_data([
                ("Mixer-Control", getattr(st, "input_control", "") or "?"),
                ("Lautstaerke", _bar(volume)),
                ("Status", mute_str),
            ])
        else:
            in_table.update_data([("Status", "[dim]Nicht verfuegbar[/dim]")])

        # Multi-Speaker
        header = self.query_one("#sat-audio-speakers-header", Static)
        speakers_table = self.query_one("#sat-audio-speakers-list", StatusTable)
        pulse_avail = bool(getattr(st, "pulse_available", False))
        active = bool(getattr(st, "combined_sink_active", False))

        if not pulse_avail:
            header.update(
                "[dim]PulseAudio/PipeWire nicht verfuegbar — Multi-Speaker "
                "wird auf diesem System nicht unterstuetzt.[/dim]"
            )
            speakers_table.update_data([])
            return

        enabled_count = sum(1 for sp in self._editable_speakers if sp["enabled"])
        status_parts = [
            f"Verfuegbar: [bold]{len(self._editable_speakers)}[/bold]",
            f"Aktiviert: [bold]{enabled_count}[/bold]",
        ]
        if active:
            status_parts.append("Combined-Sink: [green]aktiv[/green]")
        else:
            status_parts.append("Combined-Sink: [dim]inaktiv[/dim]")
        if self._dirty:
            status_parts.append("[yellow]ungespeicherte Aenderungen[/yellow]")
        header.update("  ·  ".join(status_parts))

        rows: list[tuple[str, str]] = []
        for idx, sp in enumerate(self._editable_speakers):
            cursor = "▶" if idx == self._cursor else " "
            check = "[green][x][/green]" if sp["enabled"] else "[ ]"
            label = f"{cursor} {check} {sp['description']}"
            value = _bar(int(sp["volume"]))
            rows.append((label, value))
        if not rows:
            rows = [("", "[dim]Keine Sinks gefunden[/dim]")]
        speakers_table.update_data(rows)

        # Multi-Mic
        mics_header = self.query_one("#sat-audio-mics-header", Static)
        mics_table = self.query_one("#sat-audio-mics-list", StatusTable)

        if not pulse_avail:
            mics_header.update(
                "[dim]PulseAudio/PipeWire nicht verfuegbar — Multi-Mic "
                "wird auf diesem System nicht unterstuetzt.[/dim]"
            )
            mics_table.update_data([])
            return

        mic_active = bool(getattr(st, "input_mix_active", False))
        mic_enabled_count = sum(1 for m in self._editable_mics if m["enabled"])
        mic_status_parts = [
            f"Verfuegbar: [bold]{len(self._editable_mics)}[/bold]",
            f"Aktiviert: [bold]{mic_enabled_count}[/bold]",
        ]
        if mic_active:
            mic_status_parts.append("Mix: [green]aktiv[/green]")
        elif mic_enabled_count >= 2:
            mic_status_parts.append("Mix: [yellow]inaktiv (noch nicht angewendet)[/yellow]")
        else:
            mic_status_parts.append("Mix: [dim]inaktiv (Single-Mic)[/dim]")
        if self._mics_dirty:
            mic_status_parts.append("[yellow]ungespeicherte Aenderungen[/yellow]")
        mics_header.update("  ·  ".join(mic_status_parts))

        mic_rows: list[tuple[str, str]] = []
        for idx, m in enumerate(self._editable_mics):
            cursor = "▶" if idx == self._mic_cursor else " "
            check = "[green][x][/green]" if m["enabled"] else "[ ]"
            label = f"{cursor} {check} {m['description']}"
            value = m.get("label", "") or "[dim]—[/dim]"
            mic_rows.append((label, value))
        if not mic_rows:
            mic_rows = [("", "[dim]Keine Mikrofone gefunden[/dim]")]
        mics_table.update_data(mic_rows)

    # === Button-Events ===

    async def on_button_pressed(self, event: Button.Pressed) -> None:
        bid = event.button.id or ""

        # === Multi-Speaker Buttons ===
        if bid.startswith("audio-speakers-"):
            await self._handle_speakers_button(bid)
            return

        # === Multi-Mic Buttons ===
        if bid.startswith("audio-mics-"):
            await self._handle_mics_button(bid)
            return

        st = self._status
        if st is None:
            return

        if bid.startswith("audio-out-"):
            target = "output"
            current = int(getattr(st, "output_volume", 0))
            currently_muted = bool(getattr(st, "output_muted", False))
        elif bid.startswith("audio-in-"):
            target = "input"
            current = int(getattr(st, "input_volume", 0))
            currently_muted = bool(getattr(st, "input_muted", False))
        else:
            return

        if bid.endswith("-up"):
            await self._send("set_volume", target, volume=min(100, current + VOLUME_STEP))
        elif bid.endswith("-down"):
            await self._send("set_volume", target, volume=max(0, current - VOLUME_STEP))
        elif bid.endswith("-mute"):
            await self._send("set_mute", target, muted=not currently_muted)

    async def _handle_speakers_button(self, bid: str) -> None:
        if bid == "audio-speakers-refresh":
            self._dirty = False
            await self._reload()
            return

        if not self._editable_speakers:
            return

        if bid == "audio-speakers-cursor":
            self._cursor = (self._cursor + 1) % len(self._editable_speakers)
            self._update_display()
            return

        if bid == "audio-speakers-toggle":
            sp = self._editable_speakers[self._cursor]
            sp["enabled"] = not sp["enabled"]
            self._dirty = True
            self._update_display()
            return

        if bid in ("audio-speakers-vol-up", "audio-speakers-vol-down"):
            sp = self._editable_speakers[self._cursor]
            step = VOLUME_STEP if bid.endswith("-up") else -VOLUME_STEP
            sp["volume"] = max(0, min(100, int(sp["volume"]) + step))
            self._dirty = True
            self._update_display()
            return

        if bid == "audio-speakers-apply":
            payload = [
                {
                    "device": sp["device"],
                    "enabled": bool(sp["enabled"]),
                    "volume": int(sp["volume"]),
                }
                for sp in self._editable_speakers
            ]
            try:
                response = await self.connection.request_satellite_audio(
                    self._satellite_id(),
                    action="set_speakers",
                    speakers=payload,
                )
            except Exception:
                return
            if response is not None:
                self._status = response
                self._dirty = False
                self._sync_editable_from_status()
                self._update_display()

    async def _handle_mics_button(self, bid: str) -> None:
        if bid == "audio-mics-refresh":
            self._mics_dirty = False
            await self._reload()
            return

        if not self._editable_mics:
            return

        if bid == "audio-mics-cursor":
            self._mic_cursor = (self._mic_cursor + 1) % len(self._editable_mics)
            self._update_display()
            return

        if bid == "audio-mics-toggle":
            m = self._editable_mics[self._mic_cursor]
            m["enabled"] = not m["enabled"]
            self._mics_dirty = True
            self._update_display()
            return

        if bid == "audio-mics-apply":
            payload = [
                {
                    "device": m["device"],
                    "enabled": bool(m["enabled"]),
                    "label": m.get("label", ""),
                }
                for m in self._editable_mics
            ]
            try:
                response = await self.connection.request_satellite_audio(
                    self._satellite_id(),
                    action="set_microphones",
                    microphones=payload,
                )
            except Exception:
                return
            if response is not None:
                self._status = response
                self._mics_dirty = False
                # Nach Apply frische Sources holen damit input_mix_active
                # korrekt angezeigt wird — das Apply alleine liefert nur den
                # Mix-Status, nicht die Quellen-Liste.
                await self._reload()


def _bar(percent: int) -> str:
    """Rendert eine 20-Zeichen Lautstaerke-Bar mit Prozent-Angabe."""
    percent = max(0, min(100, percent))
    filled = int(round(percent / 5))
    bar = "█" * filled + "░" * (20 - filled)
    return f"[cyan]{bar}[/cyan] {percent:3d}%"