# -*- 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}%"