dot

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

SearchHelp
89 lines2393 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 "sink": (sink or {}).get("description", "No output"),
45 "sinks": [
46 {
47 "name": s.get("name", ""),
48 "desc": s.get("description") or s.get("name", ""),
49 "active": s.get("name") == default_sink,
50 }
51 for s in sinks
52 ],
53 }
54
55
56last = None
57
58
59def emit():
60 global last
61 line = json.dumps(snapshot(), separators=(",", ":"))
62 if line != last:
63 print(line, flush=True)
64 last = line
65
66
67while True:
68 emit()
69 try:
70 proc = subprocess.Popen(
71 ["pactl", "subscribe"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0
72 )
73 except OSError:
74 time.sleep(5)
75 continue
76 out = proc.stdout
77 while True:
78 line = out.readline()
79 if not line:
80 break
81 if not re.search(rb"on (sink|source|server)", line):
82 continue
83 # Collapse bursts of events (e.g. while dragging a slider) into one update.
84 while select.select([out], [], [], 0.08)[0]:
85 if not out.readline():
86 break
87 emit()
88 proc.wait()
89 time.sleep(2)