dot

vizie's NixOS desktop: Hyprland, CachyOS kernel and gaming tweaks

SearchHelp
80 lines2079 bytesPythonexecutable
1#!/usr/bin/env python3
2"""Streams PipeWire/PulseAudio state as one JSON line per change (Eww `deflisten`)."""
3import json
4import re
5import select
6import subprocess
7import time
8
9
10def run(*args):
11 try:
12 return subprocess.run(args, capture_output=True, text=True, timeout=3).stdout.strip()
13 except Exception:
14 return ""
15
16
17def pactl_json(kind):
18 try:
19 return json.loads(run("pactl", "--format=json", "list", kind) or "[]")
20 except ValueError:
21 return []
22
23
24def percent(dev):
25 for ch in (dev.get("volume") or {}).values():
26 m = re.match(r"\d+", str(ch.get("value_percent", "")))
27 if m:
28 return int(m.group())
29 return 0
30
31
32def snapshot():
33 default_sink = run("pactl", "get-default-sink")
34 default_source = run("pactl", "get-default-source")
35 sinks = pactl_json("sinks")
36 sources = pactl_json("sources")
37 sink = next((s for s in sinks if s.get("name") == default_sink), None)
38 src = next((s for s in sources if s.get("name") == default_source), None)
39 return {
40 "vol": percent(sink) if sink else 0,
41 "muted": bool(sink and sink.get("mute")),
42 "mic": percent(src) if src else 0,
43 "mic_muted": bool(src and src.get("mute")),
44 }
45
46
47last = None
48
49
50def emit():
51 global last
52 line = json.dumps(snapshot(), separators=(",", ":"))
53 if line != last:
54 print(line, flush=True)
55 last = line
56
57
58while True:
59 emit()
60 try:
61 proc = subprocess.Popen(
62 ["pactl", "subscribe"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0
63 )
64 except OSError:
65 time.sleep(5)
66 continue
67 out = proc.stdout
68 while True:
69 line = out.readline()
70 if not line:
71 break
72 if not re.search(rb"on (sink|source|server)", line):
73 continue
74 # Collapse bursts of events (e.g. while dragging a slider) into one update.
75 while select.select([out], [], [], 0.08)[0]:
76 if not out.readline():
77 break
78 emit()
79 proc.wait()
80 time.sleep(2)