311 lines
13 KiB
Python
311 lines
13 KiB
Python
"""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])
|