linux.sipsoftphone/utils/audio_manager.py

"""Audio-Manager - Audiogeräte auflisten, wechseln und Lautstärke steuern.

Nutzt PipeWire/PulseAudio-Namen (wie KDE) statt kryptischer ALSA-Bezeichnungen.
PJSUA2 wird auf das 'pulse'-Device gesetzt, damit PipeWire das Routing übernimmt.
"""

import subprocess
import pjsua2 as pj


def _pipewire_geraete_holen():
    """PipeWire/PulseAudio-Geräte mit KDE-freundlichen Namen holen.

    Returns:
        (sinks, sources): Dicts mit {pw_name: description}
    """
    sinks = {}   # Wiedergabe
    sources = {}  # Aufnahme

    try:
        # Sinks (Wiedergabegeräte)
        result = subprocess.run(
            ["pactl", "list", "sinks"],
            capture_output=True, text=True, timeout=3,
        )
        aktueller_name = None
        for zeile in result.stdout.splitlines():
            zeile = zeile.strip()
            if zeile.startswith("Name:"):
                aktueller_name = zeile.split(":", 1)[1].strip()
            elif zeile.startswith("Description:") and aktueller_name:
                sinks[aktueller_name] = zeile.split(":", 1)[1].strip()
                aktueller_name = None

        # Sources (Aufnahmegeräte)
        result = subprocess.run(
            ["pactl", "list", "sources"],
            capture_output=True, text=True, timeout=3,
        )
        aktueller_name = None
        for zeile in result.stdout.splitlines():
            zeile = zeile.strip()
            if zeile.startswith("Name:"):
                aktueller_name = zeile.split(":", 1)[1].strip()
            elif zeile.startswith("Description:") and aktueller_name:
                # Monitor-Quellen ausfiltern (sind keine echten Mikrofone)
                if ".monitor" not in aktueller_name:
                    sources[aktueller_name] = zeile.split(":", 1)[1].strip()
                aktueller_name = None
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        pass

    return sinks, sources


def _alsa_name_zu_pw_name(alsa_name):
    """Versucht einen ALSA-Gerätenamen auf PipeWire-Namen abzubilden.

    PJSUA2 gibt z.B. 'Razer Nari: USB Audio (hw:0,0)' zurück.
    PipeWire nutzt z.B. 'alsa_output.usb-Razer_Razer_Nari-00.mono-fallback'.

    Matching über Schlüsselwörter im ALSA-Namen.
    """
    # Extrahiere Schlüsselwörter aus dem ALSA-Namen (ohne hw:X,Y)
    bereinigt = alsa_name.split("(")[0].strip().lower()
    # Wörter extrahieren
    woerter = [w for w in bereinigt.replace(":", " ").replace("-", " ").split()
               if len(w) > 2]
    return woerter


class AudioManager:
    """Verwaltet Audiogeräte über PipeWire/PulseAudio mit PJSUA2-Backend.

    Zeigt PipeWire-Geräte mit KDE-freundlichen Namen an.
    PJSUA2 nutzt das 'pulse'-ALSA-Plugin für PipeWire-Routing.
    """

    def __init__(self, engine):
        self._engine = engine
        self._pulse_dev_id = None  # PJSUA2-ID des 'pulse'-Geräts

    @property
    def _aud_mgr(self):
        """Zugriff auf den PJSUA2 AudDevManager."""
        if self._engine.ep:
            return self._engine.ep.audDevManager()
        return None

    def _pjsua2_geraete(self):
        """Rohe PJSUA2-Geräteliste holen."""
        if not self._aud_mgr:
            return []
        geraete = []
        try:
            dev_list = self._aud_mgr.enumDev2()
            for i, dev in enumerate(dev_list):
                info = dev.info if hasattr(dev, "info") else dev
                geraete.append({
                    "id": i,
                    "name": info.name,
                    "eingaenge": info.inputCount,
                    "ausgaenge": info.outputCount,
                    "driver": info.driver,
                })
        except (pj.Error, AttributeError):
            pass
        return geraete

    def _pulse_device_finden(self):
        """Finde die PJSUA2-Device-ID für 'pulse' (PipeWire-Routing)."""
        if self._pulse_dev_id is not None:
            return self._pulse_dev_id

        geraete = self._pjsua2_geraete()
        for dev in geraete:
            if dev["name"] == "pulse":
                self._pulse_dev_id = dev["id"]
                return self._pulse_dev_id
        return None

    def pulse_als_standard_setzen(self):
        """Setzt 'pulse' als Standard-Audiogerät (PipeWire-Routing).

        Damit übernimmt PipeWire/KDE die Gerätezuordnung.
        """
        pulse_id = self._pulse_device_finden()
        if pulse_id is not None and self._aud_mgr:
            try:
                self._aud_mgr.setCaptureDev(pulse_id)
                self._aud_mgr.setPlaybackDev(pulse_id)
                return True
            except pj.Error:
                pass
        return False

    def aufnahme_geraete(self):
        """Aufnahmegeräte mit PipeWire-Beschreibung (wie KDE).

        Returns:
            Liste von Dicts: {pw_name, beschreibung, pjsua2_id}
        """
        _, sources = _pipewire_geraete_holen()
        pj_geraete = self._pjsua2_geraete()

        geraete = []
        for pw_name, beschreibung in sources.items():
            # Versuche passendes PJSUA2-Device zu finden
            pj_id = self._pw_zu_pjsua2(pw_name, pj_geraete, aufnahme=True)
            geraete.append({
                "id": pj_id,
                "pw_name": pw_name,
                "name": beschreibung,
                "eingaenge": 1,
                "ausgaenge": 0,
            })

        return geraete

    def wiedergabe_geraete(self):
        """Wiedergabegeräte mit PipeWire-Beschreibung (wie KDE).

        Returns:
            Liste von Dicts: {pw_name, beschreibung, pjsua2_id}
        """
        sinks, _ = _pipewire_geraete_holen()
        pj_geraete = self._pjsua2_geraete()

        geraete = []
        for pw_name, beschreibung in sinks.items():
            pj_id = self._pw_zu_pjsua2(pw_name, pj_geraete, aufnahme=False)
            geraete.append({
                "id": pj_id,
                "pw_name": pw_name,
                "name": beschreibung,
                "eingaenge": 0,
                "ausgaenge": 1,
            })

        return geraete

    def _pw_zu_pjsua2(self, pw_name, pj_geraete, aufnahme=False):
        """Ordnet einen PipeWire-Namen einem PJSUA2-Device-Index zu.

        Da PJSUA2 über das 'pulse'-Device läuft, wird die
        PipeWire-Device-Auswahl über `pactl set-default-*` gesteuert.
        Rückgabe: pw_name als Identifikator (kein PJSUA2-Index nötig).
        """
        # Wir nutzen den PipeWire-Namen direkt als ID-String
        # Die tatsächliche Umschaltung passiert über pactl
        return pw_name

    def aufnahme_geraet_setzen(self, pw_name):
        """Aufnahmegerät über PipeWire setzen.

        Args:
            pw_name: PipeWire-Device-Name (z.B. 'alsa_input.usb-...')
        """
        if isinstance(pw_name, int):
            # Fallback: Direkte PJSUA2-ID (Kompatibilität)
            if self._aud_mgr:
                try:
                    self._aud_mgr.setCaptureDev(pw_name)
                    return True
                except pj.Error:
                    return False
            return False

        # PipeWire-Default-Source setzen
        try:
            subprocess.run(
                ["pactl", "set-default-source", pw_name],
                capture_output=True, timeout=3,
            )
            return True
        except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
            return False

    def wiedergabe_geraet_setzen(self, pw_name):
        """Wiedergabegerät über PipeWire setzen.

        Args:
            pw_name: PipeWire-Device-Name (z.B. 'alsa_output.usb-...')
        """
        if isinstance(pw_name, int):
            # Fallback: Direkte PJSUA2-ID (Kompatibilität)
            if self._aud_mgr:
                try:
                    self._aud_mgr.setPlaybackDev(pw_name)
                    return True
                except pj.Error:
                    return False
            return False

        # PipeWire-Default-Sink setzen
        try:
            subprocess.run(
                ["pactl", "set-default-sink", pw_name],
                capture_output=True, timeout=3,
            )
            return True
        except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
            return False

    def mikrofon_lautstaerke(self, level):
        """Mikrofon-Lautstärke setzen (0.0 = stumm, 1.0 = normal, 2.0 = laut)."""
        if self._aud_mgr:
            try:
                self._aud_mgr.getCaptureDevMedia().adjustTxLevel(level)
            except pj.Error:
                pass

    def lautsprecher_lautstaerke(self, level):
        """Lautsprecher-Lautstärke setzen (0.0 = stumm, 1.0 = normal, 2.0 = laut)."""
        if self._aud_mgr:
            try:
                self._aud_mgr.getPlaybackDevMedia().adjustRxLevel(level)
            except pj.Error:
                pass