# -*- coding: utf-8 -*- """ PulseAudio / PipeWire Combined-Sink Wrapper. Erlaubt es, mehrere Lautsprecher gleichzeitig zu bespielen indem ein virtueller Sink (`module-combine-sink`) erstellt wird, der alle ausgewaehlten echten Sinks bündelt. Funktioniert auf: - Linux mit PulseAudio - Linux mit PipeWire (pactl-API-kompatibel) - WSL2 mit WSLg (bringt Pulse mit) Funktioniert NICHT auf: - Linux ohne Pulse/PipeWire (z.B. Pi-OS-Lite headless) - Windows nativ - macOS In diesen Faellen meldet `available()` False und der Client faellt auf den klassischen Single-Output-Pfad zurueck. """ from __future__ import annotations import asyncio import os import shutil from typing import NamedTuple from trixy_core.utils.debug import pinfo, pdebug, perror class PulseSink(NamedTuple): """Ein einzelner PulseAudio/PipeWire Sink.""" index: int name: str # interner Name z.B. "alsa_output.usb-Generic-00.analog-stereo" description: str # Lesbarer Name z.B. "USB Audio CODEC Analog Stereo" sample_spec: str # z.B. "s16le 2ch 44100Hz" state: str # "RUNNING" | "IDLE" | "SUSPENDED" class PulseCombinedSink: """Verwaltet einen virtuellen Combined-Sink ueber pactl.""" DEFAULT_NAME = "trixy_combined" def __init__(self, sink_name: str = DEFAULT_NAME) -> None: self._sink_name = sink_name self._module_index: int | None = None self._slaves: list[str] = [] self._available: bool | None = None # Lazy-detect self._previous_default: str | None = None # Fuer Restore beim Teardown @property def sink_name(self) -> str: return self._sink_name @property def module_index(self) -> int | None: return self._module_index @property def slaves(self) -> list[str]: return list(self._slaves) async def available(self) -> bool: """True wenn pactl im PATH ist UND ein Pulse/PipeWire-Server antwortet.""" if self._available is not None: return self._available if not shutil.which("pactl"): pdebug("pactl nicht gefunden — Combined-Sink-Support deaktiviert") self._available = False return False # Echten Server-Kontakt pruefen — pactl info schlaegt fehl wenn # kein Pulse/PipeWire-Daemon laeuft (z.B. ALSA-only). rc, _ = await self._run("info") self._available = rc == 0 if not self._available: pdebug("pactl info fehlgeschlagen — kein Pulse/PipeWire-Server") return self._available # === Sink-Listing === async def list_sinks(self) -> list[PulseSink]: """Listet alle realen Sinks (ohne den eigenen Combined-Sink).""" if not await self.available(): return [] rc, out = await self._run("list", "sinks") if rc != 0: return [] return [s for s in self._parse_sinks(out) if s.name != self._sink_name] @staticmethod def _parse_sinks(text: str) -> list[PulseSink]: """Parst die Ausgabe von `pactl list sinks`.""" sinks: list[PulseSink] = [] current: dict = {} def flush() -> None: if not current: return try: sinks.append(PulseSink( index=int(current.get("index", -1)), name=current.get("name", ""), description=current.get("description", current.get("name", "")), sample_spec=current.get("sample_spec", ""), state=current.get("state", ""), )) except Exception: pass current.clear() for raw in text.splitlines(): line = raw.rstrip() if line.startswith("Sink #"): flush() try: current["index"] = int(line.split("#", 1)[1]) except Exception: pass continue stripped = line.lstrip() if stripped.startswith("Name:"): current["name"] = stripped.split(":", 1)[1].strip() elif stripped.startswith("Description:"): current["description"] = stripped.split(":", 1)[1].strip() elif stripped.startswith("Sample Specification:"): current["sample_spec"] = stripped.split(":", 1)[1].strip() elif stripped.startswith("State:"): current["state"] = stripped.split(":", 1)[1].strip() flush() return sinks # === Combined-Sink-Lifecycle === async def find_existing_module(self) -> int | None: """Sucht ein bereits geladenes module-combine-sink mit unserem Namen.""" rc, out = await self._run("list", "modules", "short") if rc != 0: return None for line in out.splitlines(): parts = line.split("\t") if len(parts) < 2: continue try: idx = int(parts[0]) except ValueError: continue module = parts[1] args = parts[2] if len(parts) >= 3 else "" if module == "module-combine-sink" and f"sink_name={self._sink_name}" in args: return idx return None async def setup(self, slaves: list[str]) -> bool: """Erstellt (oder ersetzt) den Combined-Sink mit den gewaehlten Slaves. Args: slaves: Liste der Sink-Namen die gebuendelt werden sollen. Returns: True bei Erfolg, False sonst. """ if not await self.available(): return False if not slaves: await self.teardown() return False if len(slaves) == 1: # Bei nur einem Sink macht ein Combined-Sink keinen Sinn — # Caller soll den Sink direkt nutzen. await self.teardown() return False # Bestehendes Modul vorher entladen await self.teardown() slaves_arg = ",".join(slaves) args = ( "load-module", "module-combine-sink", f"sink_name={self._sink_name}", f"slaves={slaves_arg}", "sink_properties=device.description=Trixy_Multi_Speaker", ) rc, out = await self._run(*args) if rc != 0: perror(f"Combined-Sink konnte nicht erstellt werden: {out.strip()}") return False # pactl gibt den Modul-Index als einzige Zahl zurueck try: self._module_index = int(out.strip().splitlines()[-1]) except Exception: self._module_index = await self.find_existing_module() self._slaves = list(slaves) pinfo( f"Combined-Sink '{self._sink_name}' erstellt " f"(Modul #{self._module_index}, {len(slaves)} Slaves)" ) return True async def teardown(self) -> None: """Entlaedt das Combined-Sink-Modul und stellt den alten Default wieder her.""" if not await self.available(): return # Vorherigen Default-Sink wiederherstellen BEVOR der Combined-Sink entladen wird await self.restore_default() idx = self._module_index if idx is None: idx = await self.find_existing_module() if idx is None: self._slaves = [] return rc, _ = await self._run("unload-module", str(idx)) if rc == 0: pdebug(f"Combined-Sink Modul #{idx} entladen") self._module_index = None self._slaves = [] # === Default-Sink-Verwaltung === # # Wir aendern den System-Default-Sink auf den Combined-Sink, damit # der Audio-Player (sounddevice ohne explizites device=) automatisch # in den Verbund streamt. Ohne diesen Schritt wuerde der Player # weiterhin auf einem einzelnen Sink landen. # # Beim Teardown wird der vorherige Default wiederhergestellt, damit # das System nach `trixy stop` nicht auf einem entladenen Sink haengt. async def get_default_sink(self) -> str: """Liest den aktuellen System-Default-Sink-Namen.""" if not await self.available(): return "" rc, out = await self._run("get-default-sink") if rc != 0: return "" return out.strip().splitlines()[0] if out.strip() else "" async def set_default_sink(self, sink_name: str) -> bool: """Setzt den System-Default-Sink.""" if not await self.available(): return False rc, _ = await self._run("set-default-sink", sink_name) return rc == 0 async def make_default(self) -> bool: """Macht den Combined-Sink zum System-Default und merkt sich den vorherigen.""" if self._module_index is None: return False current = await self.get_default_sink() # Nur merken wenn es nicht schon der Combined-Sink ist (z.B. nach Reload) if current and current != self._sink_name: self._previous_default = current ok = await self.set_default_sink(self._sink_name) if ok: pinfo( f"System-Default-Sink → '{self._sink_name}' " f"(vorher: '{self._previous_default or '?'}')" ) return ok async def restore_default(self) -> None: """Stellt den vorherigen Default-Sink wieder her (wenn vorhanden).""" if not self._previous_default: return if not await self.available(): return # Nur restoren wenn der aktuelle Default noch unser Combined-Sink ist current = await self.get_default_sink() if current == self._sink_name: ok = await self.set_default_sink(self._previous_default) if ok: pdebug(f"System-Default-Sink wiederhergestellt → '{self._previous_default}'") self._previous_default = None # === Per-Slave-Lautstaerke === async def set_sink_volume(self, sink_name: str, percent: int) -> bool: """Setzt die Lautstaerke eines beliebigen Sinks (0-100).""" if not await self.available(): return False percent = max(0, min(100, percent)) rc, _ = await self._run("set-sink-volume", sink_name, f"{percent}%") return rc == 0 async def get_sink_volume(self, sink_name: str) -> int: """Liest die aktuelle Lautstaerke eines Sinks (0-100, Mittelwert).""" if not await self.available(): return 0 rc, out = await self._run("get-sink-volume", sink_name) if rc != 0: return 0 # Ausgabe enthaelt z.B. "Volume: front-left: 52428 / 80% / -5.81 dB, ..." for line in out.splitlines(): if "%" not in line: continue try: # Alle Prozentwerte sammeln, Mittelwert nehmen values: list[int] = [] idx = 0 while True: pct_idx = line.find("%", idx) if pct_idx < 0: break # Zahl rueckwaerts lesen j = pct_idx - 1 while j >= 0 and (line[j].isdigit() or line[j] == " "): j -= 1 num = line[j + 1:pct_idx].strip() if num: values.append(int(num)) idx = pct_idx + 1 if values: return sum(values) // len(values) except Exception: pass return 0 # === Hilfsmethoden === async def _run(self, *args: str) -> tuple[int, str]: """Fuehrt einen pactl-Befehl aus, gibt (returncode, stdout+stderr) zurueck. Erzwingt LC_ALL=C damit die Ausgabe sprachunabhaengig ist — sonst kommt z.B. auf deutschen Systemen "Ziel #65 / Beschreibung: ..." statt "Sink #65 / Description: ...". """ env = os.environ.copy() env["LC_ALL"] = "C" env["LANG"] = "C" try: proc = await asyncio.create_subprocess_exec( "pactl", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, env=env, ) stdout, _ = await proc.communicate() return proc.returncode or 0, stdout.decode(errors="replace") except FileNotFoundError: return 127, "" except Exception as e: perror(f"pactl Fehler: {e}") return 1, str(e)