feat: komplette Codebase vom Board ins Repo gespiegelt
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>
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user