app/ = /opt/callrec (Node-Backend, Vite-Frontend, Python-Inferenz, Audio-Skripte, rkllm_server.cpp), system/ = systemd-Units, WirePlumber/PipeWire-Drop-ins, sudoers. Modelle, Binaries, Aufnahmen und node_modules bleiben draussen (.gitignore). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
# Listen to the Logitech H390 inline volume/mute buttons (they arrive as HID
|
|
# keyboard events on the BOARD, since the headset is plugged into the board) and
|
|
# apply them to the headset PipeWire sink. Headless => no desktop consumes these,
|
|
# so without this daemon the inline buttons do nothing.
|
|
import glob, os, re, struct, subprocess, time
|
|
|
|
RUNTIME = "/run/user/1000"
|
|
USER = "callrec"
|
|
HSSINK = "alsa_output.usb-Logi_USB_Headset_Logi_USB_Headset-00.analog-stereo"
|
|
DEV_NAME_RE = re.compile(r"Logi USB Headset", re.I)
|
|
|
|
# linux input_event (64-bit): struct timeval(2*long) + u16 type + u16 code + s32 value
|
|
EV_FMT = "llHHi"
|
|
EV_SIZE = struct.calcsize(EV_FMT)
|
|
EV_KEY = 0x01
|
|
KEY_MUTE, KEY_VOLUMEDOWN, KEY_VOLUMEUP = 113, 114, 115
|
|
|
|
def pactl(*args):
|
|
subprocess.run(["sudo", "-u", USER, "env", f"XDG_RUNTIME_DIR={RUNTIME}",
|
|
"pactl", *args], check=False,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
def find_event_device():
|
|
# match the event handler of the input device whose Name contains "Logi USB
|
|
# Headset" and that reports key events (EV=... has bit 1). Parse
|
|
# /proc/bus/input/devices so we survive event-number changes across reboots.
|
|
try:
|
|
blocks = open("/proc/bus/input/devices").read().split("\n\n")
|
|
except OSError:
|
|
return None
|
|
for b in blocks:
|
|
if not DEV_NAME_RE.search(b):
|
|
continue
|
|
m = re.search(r"Handlers=.*?(event\d+)", b)
|
|
if m:
|
|
return "/dev/input/" + m.group(1)
|
|
return None
|
|
|
|
def run():
|
|
dev = None
|
|
while True:
|
|
path = find_event_device()
|
|
if not path or not os.path.exists(path):
|
|
time.sleep(2); continue
|
|
try:
|
|
dev = open(path, "rb", buffering=0)
|
|
except OSError:
|
|
time.sleep(2); continue
|
|
try:
|
|
while True:
|
|
data = dev.read(EV_SIZE)
|
|
if not data or len(data) < EV_SIZE:
|
|
break
|
|
_, _, etype, code, value = struct.unpack(EV_FMT, data)
|
|
if etype != EV_KEY or value == 0: # key release => ignore
|
|
continue
|
|
if code == KEY_VOLUMEUP:
|
|
pactl("set-sink-volume", HSSINK, "+4%")
|
|
elif code == KEY_VOLUMEDOWN:
|
|
pactl("set-sink-volume", HSSINK, "-4%")
|
|
elif code == KEY_MUTE and value == 1:
|
|
pactl("set-sink-mute", HSSINK, "toggle")
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
try: dev.close()
|
|
except Exception: pass
|
|
time.sleep(1) # device vanished (unplug/reboot) -> re-find
|
|
|
|
if __name__ == "__main__":
|
|
run()
|