1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
#!/usr/bin/env python3
import asyncio
import json
import os
import subprocess
import sys
ICONS = {
"mute": "\U000f075f", #
"low": "\U000f057f", #
"medium": "\U000f0580", #
"high": "\U000f057e", #
}
def is_muted():
return (
subprocess.run(
["pactl", "get-sink-mute", "@DEFAULT_SINK@"],
capture_output=True,
encoding="UTF-8",
)
.stdout.removeprefix("Mute: ")
.strip()
== "yes"
)
def get_volume():
stdout = subprocess.run(
["pactl", "get-sink-volume", "@DEFAULT_SINK@"],
capture_output=True,
encoding="UTF-8",
).stdout.strip()
for s in stdout.removeprefix("Volume: ").split():
if s.endswith("%"):
return int(s.rstrip("%"))
def print_status():
match get_volume():
case None:
output = {}
case v if is_muted():
output = {
"full_text": f" {ICONS["mute"]} {v}% ",
"color": f"#{os.environ.get("BASE16_COLOR_00_HEX")}",
"background": f"#{os.environ.get("BASE16_COLOR_0A_HEX")}",
}
case v if v > 66:
output = {"full_text": f" {ICONS["high"]} {v}% "}
case v if v > 33:
output = {"full_text": f" {ICONS["medium"]} {v}% "}
case v:
output = {"full_text": f" {ICONS["low"]} {v}% "}
print(json.dumps(output, ensure_ascii=False), flush=True)
async def listener():
process = await asyncio.create_subprocess_exec(
"pactl",
"--format=json",
"subscribe",
stdout=asyncio.subprocess.PIPE,
)
while True:
line = await process.stdout.readline()
if not line:
await asyncio.sleep(1)
continue
match json.loads(line.decode("UTF-8")):
case {"on": "sink"} | {"on": "source-output"}:
print_status()
async def button_handler():
loop = asyncio.get_event_loop()
reader = asyncio.StreamReader()
protocol = asyncio.StreamReaderProtocol(reader)
await loop.connect_read_pipe(lambda: protocol, sys.stdin)
proc: subprocess.Popen = None
while True:
line = await reader.readline()
if not line:
await asyncio.sleep(1)
continue
match json.loads(line):
case {"button": 1} if proc:
proc.terminate()
proc = None
case {"button": 1, "gui": cmd}:
proc = subprocess.Popen(cmd, shell=True)
case {"button": 2}:
pass
case {"button": 3}:
subprocess.run(["wpctl", "set-mute", "@DEFAULT_AUDIO_SINK@", "toggle"])
case {"button": 4}:
subprocess.run(
["wpctl", "set-volume", "-l", "1.5", "@DEFAULT_AUDIO_SINK@", "5%+"]
)
case {"button": 5}:
subprocess.run(
["wpctl", "set-volume", "-l", "1.5", "@DEFAULT_AUDIO_SINK@", "5%-"]
)
async def main():
print_status()
try:
async with asyncio.TaskGroup() as task_group:
task_group.create_task(listener())
task_group.create_task(button_handler())
except asyncio.CancelledError:
return
if __name__ == "__main__":
asyncio.run(main())
|
