"""Audio backend: talks to PipeWire through pipewire-pulse's `pactl` compat layer — pipewire-pulse IS the PipeWire project's own pulse-protocol server (see hyprdrive.sh's package list: pipewire + pipewire-pulse + wireplumber), not the legacy PulseAudio daemon. `pactl -f json` gives structured per-channel volume, card profiles and stream routing with no text-scraping, and `pactl subscribe` gives event-driven refresh instead of polling (same philosophy as astro-menu's hypr_ipc-style event feeds). wpctl (WirePlumber's own CLI) was considered and dropped: it has no per-channel volume or card-profile verbs, both of which the Applications/Output/Input tabs need. Every read is async (Gio.Subprocess.communicate_utf8_async, never a blocking `subprocess.run`) — a hung pactl must not freeze the panel, the same stance transmitter-panel's history_client.py takes toward beacon's D-Bus calls. Mutations are fire-and-forget Gio.Subprocess spawns; the resulting `pactl subscribe` event is what actually refreshes the UI; no mutation waits on its own result. """ from __future__ import annotations import json from typing import Callable, Optional import gi gi.require_version("Gtk", "4.0") from gi.repository import Gio, GLib # noqa: E402 # Coalesce bursts of subscribe events (e.g. a drag-driven volume change fires # many 'change' events in a row) into one refresh instead of one per event. REFRESH_DEBOUNCE_MS = 120 _MUTATE_CMD = { "sink": ("set-sink-volume", "set-sink-mute"), "source": ("set-source-volume", "set-source-mute"), "sink-input": ("set-sink-input-volume", "set-sink-input-mute"), "source-output": ("set-source-output-volume", "set-source-output-mute"), } def _percent(vol_entry: dict) -> int: raw = (vol_entry or {}).get("value_percent") or "0%" try: return int(str(raw).rstrip("%")) except ValueError: return 0 def _clamp(pct: int) -> int: return max(0, min(150, pct)) def _normalize_device(raw: dict) -> dict: """Sinks (output devices) and sources (input devices) share this shape.""" props = raw.get("properties") or {} volume = raw.get("volume") or {} # dict insertion order mirrors pactl's own channel_map order (front-left # before front-right for stereo) — Python's json.loads preserves it. channels = list(volume.keys()) return { "id": raw.get("index"), "name": raw.get("name") or "", "description": raw.get("description") or raw.get("name") or "", "mute": bool(raw.get("mute")), "channels": channels, "volume_percent": {ch: _percent(v) for ch, v in volume.items()}, "card": raw.get("card"), "icon": props.get("device.icon_name") or "audio-card", "form_factor": props.get("device.form_factor") or "", "state": raw.get("state") or "", } def _normalize_stream(raw: dict) -> dict: """Sink-inputs (apps playing audio) and source-outputs (apps recording).""" props = raw.get("properties") or {} volume = raw.get("volume") or {} channels = list(volume.keys()) name = (props.get("application.name") or props.get("media.name") or props.get("node.name") or f"Stream #{raw.get('index')}") return { "id": raw.get("index"), "name": name, "icon": props.get("application.icon_name") or props.get("window.icon_name") or "audio-x-generic", "mute": bool(raw.get("mute")), "channels": channels, "volume_percent": {ch: _percent(v) for ch, v in volume.items()}, # whichever sink (sink-input) or source (source-output) it's routed to "device": raw.get("sink") if "sink" in raw else raw.get("source"), "corked": bool(raw.get("corked", False)), "binary": props.get("application.process.binary") or "", } def _normalize_card(raw: dict) -> dict: props = raw.get("properties") or {} profiles_raw = raw.get("profiles") or {} profiles = [] for name, info in profiles_raw.items(): if isinstance(info, dict): profiles.append({ "name": name, "description": info.get("description", name), "available": bool(info.get("available", True)), }) else: # defensive: older/other pactl builds may just give a string profiles.append({"name": name, "description": str(info), "available": True}) return { "id": raw.get("index"), "name": raw.get("name") or "", "description": props.get("device.description") or raw.get("name") or "", "profiles": profiles, "active_profile": raw.get("active_profile") or "", } class AudioBackend: def __init__(self, on_changed: Callable[[], None]) -> None: self._on_changed = on_changed self._debounce_id: Optional[int] = None self._sub_proc: Optional[Gio.Subprocess] = None self._sub_stream: Optional[Gio.DataInputStream] = None self._start_subscribe() # -- live event stream ---------------------------------------------------- def _start_subscribe(self) -> None: try: self._sub_proc = Gio.Subprocess.new( ["pactl", "subscribe"], Gio.SubprocessFlags.STDOUT_PIPE) except GLib.Error: self._sub_proc = None return self._sub_stream = Gio.DataInputStream.new(self._sub_proc.get_stdout_pipe()) self._read_next_event() def _read_next_event(self) -> None: if self._sub_stream is None: return self._sub_stream.read_line_async(GLib.PRIORITY_DEFAULT, None, self._on_event_line) def _on_event_line(self, stream: Gio.DataInputStream, result: Gio.AsyncResult) -> None: try: line, _length = stream.read_line_finish_utf8(result) except GLib.Error: line = None if line is None: # `pactl subscribe` died (pipewire-pulse restarted, etc.) — respawn # after a short delay instead of going silent for the panel's life. self._sub_proc = None self._sub_stream = None GLib.timeout_add_seconds(2, self._restart_subscribe) return self._schedule_refresh() self._read_next_event() def _restart_subscribe(self) -> bool: self._start_subscribe() return False def _schedule_refresh(self) -> None: if self._debounce_id is not None: GLib.source_remove(self._debounce_id) self._debounce_id = GLib.timeout_add(REFRESH_DEBOUNCE_MS, self._fire_refresh) def _fire_refresh(self) -> bool: self._debounce_id = None self._on_changed() return False def stop(self) -> None: if self._debounce_id is not None: GLib.source_remove(self._debounce_id) self._debounce_id = None if self._sub_proc is not None: self._sub_proc.force_exit() self._sub_proc = None # -- reads ------------------------------------------------------------------ def _list_async(self, kind: str, normalize, callback: Callable[[list[dict]], None]) -> None: def done(proc, result) -> None: try: ok, stdout, _stderr = proc.communicate_utf8_finish(result) except GLib.Error: ok, stdout = False, "" if not ok or not stdout: callback([]) return try: raw = json.loads(stdout) except json.JSONDecodeError: raw = [] callback([normalize(r) for r in raw]) try: proc = Gio.Subprocess.new( ["pactl", "-f", "json", "list", kind], Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE) except GLib.Error: callback([]) return proc.communicate_utf8_async(None, None, done) def list_sinks(self, callback: Callable[[list[dict]], None]) -> None: self._list_async("sinks", _normalize_device, callback) def list_sources(self, callback: Callable[[list[dict]], None]) -> None: self._list_async("sources", _normalize_device, callback) def list_sink_inputs(self, callback: Callable[[list[dict]], None]) -> None: self._list_async("sink-inputs", _normalize_stream, callback) def list_source_outputs(self, callback: Callable[[list[dict]], None]) -> None: self._list_async("source-outputs", _normalize_stream, callback) def list_cards(self, callback: Callable[[list[dict]], None]) -> None: self._list_async("cards", _normalize_card, callback) def get_defaults(self, callback: Callable[[str, str], None]) -> None: """callback(default_sink_name, default_source_name), read from `pactl info`.""" def done(proc, result) -> None: try: ok, stdout, _stderr = proc.communicate_utf8_finish(result) except GLib.Error: ok, stdout = False, "" sink = source = "" if ok: for line in stdout.splitlines(): if line.startswith("Default Sink:"): sink = line.split(":", 1)[1].strip() elif line.startswith("Default Source:"): source = line.split(":", 1)[1].strip() callback(sink, source) try: proc = Gio.Subprocess.new( ["pactl", "info"], Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE) except GLib.Error: callback("", "") return proc.communicate_utf8_async(None, None, done) def get_server_info(self, callback: Callable[[dict], None]) -> None: """callback({"server_name", "server_version", "default_sink", "default_source"}), read from `pactl info`. `server_name` is what confirms this is really talking to pipewire-pulse ("PulseAudio (on PipeWire ...)") rather than a legacy PulseAudio daemon — surfaced in the General tab.""" empty = {"server_name": "", "server_version": "", "default_sink": "", "default_source": ""} def done(proc, result) -> None: try: ok, stdout, _stderr = proc.communicate_utf8_finish(result) except GLib.Error: ok, stdout = False, "" info = dict(empty) if ok: for line in stdout.splitlines(): for key, prefix in (("server_name", "Server Name:"), ("server_version", "Server Version:"), ("default_sink", "Default Sink:"), ("default_source", "Default Source:")): if line.startswith(prefix): info[key] = line.split(":", 1)[1].strip() callback(info) try: proc = Gio.Subprocess.new( ["pactl", "info"], Gio.SubprocessFlags.STDOUT_PIPE | Gio.SubprocessFlags.STDERR_SILENCE) except GLib.Error: callback(dict(empty)) return proc.communicate_utf8_async(None, None, done) # -- mutations (fire-and-forget) -------------------------------------------- @staticmethod def _spawn(argv: list[str]) -> None: try: Gio.Subprocess.new( argv, Gio.SubprocessFlags.STDOUT_SILENCE | Gio.SubprocessFlags.STDERR_SILENCE) except GLib.Error: pass def set_volume(self, kind: str, ident, percents: list[int]) -> None: """kind: 'sink' | 'source' | 'sink-input' | 'source-output'. `percents` is one value per channel in that node's channel order (see the `channels` list on its normalized dict) — pass a single-item list to set every channel uniformly, or one value per channel for independent left/right gain.""" cmd, _mute_cmd = _MUTATE_CMD[kind] values = [f"{_clamp(p)}%" for p in percents] self._spawn(["pactl", cmd, str(ident), *values]) def set_mute(self, kind: str, ident, mute: bool) -> None: _vol_cmd, cmd = _MUTATE_CMD[kind] self._spawn(["pactl", cmd, str(ident), "1" if mute else "0"]) def set_default_sink(self, name: str) -> None: self._spawn(["pactl", "set-default-sink", name]) def set_default_source(self, name: str) -> None: self._spawn(["pactl", "set-default-source", name]) def move_sink_input(self, ident, sink: str) -> None: self._spawn(["pactl", "move-sink-input", str(ident), sink]) def move_source_output(self, ident, source: str) -> None: self._spawn(["pactl", "move-source-output", str(ident), source]) def set_card_profile(self, card_ident, profile: str) -> None: self._spawn(["pactl", "set-card-profile", str(card_ident), profile])