Trixy7/trixy_core/audio/pulse_combined.py

# -*- 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)