steam-frame/DeeJanuz--frametop
1
0 Fork 0
mirror of https://github.com/DeeJanuz/frametop.git synced 2026-10-06 06:00:09 +02:00
Code Issues Packages Projects Releases 4 Wiki Activity
Files
8f051db08380ae573a1ea4afb4be5311a66f35fe
DeeJanuz--frametop/input/input-relay.py
T
DeeJanuzandClaude Opus 5.5 8f051db083 Send typing to the panel clicked last 
An ungrabbed keyboard reached both sides at once: gamescope, which in VR
reads every input device itself (the SteamOS build's InputStealer), typed
into its focused app, and ft-screens typed into the desktop. Space in the
desktop paused Spotify on the dashboard, including every space frame-voice
dictated. And while the SteamVR dashboard was open, the desktop got no
keys at all.

Typing now follows the last click. ft-screens sees clicks on its own
screens; the pointer helper reports the panel under the dot on each mouse
press, so a click on any other panel sends typing to Steam. While typing
goes to the desktop and the screens are showing, ft-screens tells the
relay every second, and the relay grabs pass-through keyboards. A grab
waits until no key is down, and the relay lets go if ft-screens stops
reporting. Controller clicks on other panels aren't visible to overlay
apps, so they don't move typing (noted in the README).

A program that reads every keyboard for a hotkey loses a grabbed one.
Repeating the keys on another input device doesn't work, since gamescope
reads that too, so with SHARE_KEYS=1 the relay sends them to
@frametop_keys as datagrams with the device name. It's off by default:
any local process that binds that name first would get every key typed
into the desktop.

The docs now say gamescope reads keyboards and the headset's buttons
itself; they said SteamVR passed keys on to it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-27 21:52:21 -06:00

882 lines
37 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Input relay for the Steam Frame: stable virtual devices, the 3D pointer, device rules.
SteamVR opens /dev/input/event* only when it starts and never hotplugs, so a
Bluetooth mouse that sleeps and reconnects (new event nodes) stops working
until SteamVR restarts. This relay creates a virtual mouse and a virtual
keyboard through /dev/uinput once, before SteamVR starts, and feeds them from
the physical devices as they come and go.
Every USB or Bluetooth mouse and keyboard is a candidate. A physical device is
identified by its Bluetooth address (EVIOCGUNIQ) or USB bus:vendor:product:name,
so all its event nodes share one role (the Swiftpoint Z3 has a mouse node and a
keyboard node for its extra buttons). Roles, from ~/.config/frametop-input.json
(written by the Frametop Input Settings app):
pointer grabbed; drives the universal 3D mouse (default for devices with a mouse node)
passthrough keys go to the desktop; grabbed only while typing goes there, otherwise only observed,
e.g. for the Meta dashboard shortcut (default for keyboards; the shortcut is off
unless META_DASHBOARD=1 is in ~/.config/frametop.conf)
ignore not grabbed, only observed for identification in the settings app
Buttons and keys of pointer devices go through a per-device map to actions
(left, right, middle, back, scroll_up, scroll_down, dashboard, recenter,
pointer_toggle, sens_up, sens_down, layout_reset = put the desktop screens back in
their saved layout, screens_toggle = hide or show the desktop screens, key = pass
through as a key, none).
Keys also go to ft-screens (@ft_screens, the Frametop desktop's compositor), which
types them into the desktop screen that has focus: from pass-through keyboards, and
keys a pointer device passes through. Typing goes to the panel clicked last, and
ft-screens says which ("keyboard desktop|steam" on the control socket, every second).
While it's the desktop, pass-through keyboards are grabbed, so gamescope, which reads
every keyboard itself, doesn't type them into its focused app too. Without word from
ft-screens for 3 seconds they're released. With SHARE_KEYS=1 in ~/.config/frametop.conf,
a grabbed keyboard's keys also go out as "key <code> <value> <device name>" datagrams on
@frametop_keys, for programs that watch every keyboard for a hotkey and lose it to the grab.
It's off by default: any local process that binds that name first gets every key typed
into the desktop.
Volume keys, from every device that has them (the headset's own buttons included),
are handled here: wpctl steps the default output. Nothing else may see a volume key,
because gamescope aborts on one when no window has keyboard focus, which ends the
whole VR session. Devices with a keymap (the headset's gpio-keys, USB and Bluetooth
keyboards) get their volume entries remapped to unused stand-in codes, so their
other keys keep working for SteamVR; a device without a keymap that has only volume
keys (the headset's pmic_resin) is grabbed. The keymaps go back when the relay exits.
Pointer mode (POINTER=1 in ~/.config/frametop.conf) sends pointer devices to
the ft-pointer helper (pointer/helper), which drives the ft_pointer
SteamVR driver. With POINTER=0, pointer devices go to the virtual mouse and
keyboard instead.
Control socket (abstract datagram @frametop_relay, JSON replies to the sender):
devices list event nodes with id, name, kinds, role, grabbed
watch <seconds> stream input events from every candidate node (identification)
reload re-read both config files, re-apply roles, tell the helper
Runs on the Frame host as a user service (frametop-input-relay.service). The
virtual devices are parked in systemd's file descriptor store, so a relay
restart gets the same devices back and SteamVR never loses them. The service is
Type=notify: READY=1 goes out only after the devices exist, so SteamVR (ordered
after it) always finds them. Dependency-free: Python standard library plus the
kernel's evdev and uinput interfaces.
input-relay.py the service
input-relay.py --no-grab never grab, for testing next to a running SteamVR
"""
import array
import atexit
import errno
import fcntl
import json
import os
import select
import signal
import socket
import struct
import subprocess
import sys
import time
# Linux input constants (include/uapi/linux/input-event-codes.h, input.h, uinput.h).
EV_SYN, EV_KEY, EV_REL, EV_MSC = 0x00, 0x01, 0x02, 0x04
SYN_REPORT = 0
BTN_MISC, KEY_MAX = 0x100, 0x2FF
KEY_A = 30
REL_X, REL_Y, REL_WHEEL, REL_MAX = 0x00, 0x01, 0x08, 0x0F
BTN_LEFT, BTN_RIGHT, BTN_MIDDLE, BTN_SIDE, BTN_EXTRA = 0x110, 0x111, 0x112, 0x113, 0x114
KEY_LEFTMETA, KEY_RIGHTMETA = 125, 126
KEY_VOLUMEDOWN, KEY_VOLUMEUP = 114, 115
# Volume keys are remapped to KEY_MACRO29 and KEY_MACRO30: above 255, so X11 can't
# carry them, and bound to nothing in the default keymap.
VOLUME_STANDIN = {KEY_VOLUMEUP: 0x2AC, KEY_VOLUMEDOWN: 0x2AD}
VOLUME_ORIGINAL = {v: k for k, v in VOLUME_STANDIN.items()}
VOLUME_CODES = set(VOLUME_STANDIN) | set(VOLUME_ORIGINAL)
BUS_USB, BUS_BLUETOOTH, BUS_VIRTUAL = 0x03, 0x05, 0x06
# struct input_event on 64-bit: struct timeval (2 x long), u16 type, u16 code, s32 value.
EVENT = struct.Struct("llHHi")
def _ioc(direction, nr, size, kind):
return (direction << 30) | (size << 16) | (ord(kind) << 8) | nr
def _iow(kind, nr, size):
return _ioc(1, nr, size, kind)
def _ior(kind, nr, size):
return _ioc(2, nr, size, kind)
UI_DEV_CREATE = _ioc(0, 1, 0, "U")
UI_DEV_DESTROY = _ioc(0, 2, 0, "U")
UI_DEV_SETUP = _iow("U", 3, 92) # struct uinput_setup: input_id (4 x u16), name[80], u32
UI_SET_EVBIT = _iow("U", 100, 4)
UI_SET_KEYBIT = _iow("U", 101, 4)
UI_SET_RELBIT = _iow("U", 102, 4)
EVIOCGRAB = _iow("E", 0x90, 4)
EVIOCGKEY = _ior("E", 0x18, (KEY_MAX + 8) // 8)
EVIOCGID = _ior("E", 0x02, 8)
KEYMAP_ENTRY = struct.Struct("BBHI32s") # struct input_keymap_entry: flags, len, index, keycode, scancode
INPUT_KEYMAP_BY_INDEX = 1
EVIOCGKEYCODE_V2 = _ior("E", 0x04, KEYMAP_ENTRY.size)
EVIOCSKEYCODE_V2 = _iow("E", 0x04, KEYMAP_ENTRY.size)
EV_NAMES = {EV_KEY: "key", EV_REL: "rel"}
def eviocgbit(ev, length):
return _ior("E", 0x20 + ev, length)
def eviocgname(length):
return _ior("E", 0x06, length)
def eviocguniq(length):
return _ior("E", 0x08, length)
VIRTUAL_PREFIX = "frametop virtual"
RULES_PATH = os.path.expanduser("~/.config/frametop-input.json")
ACTIONS = ("left", "right", "middle", "back", "scroll_up", "scroll_down", "dashboard", "recenter",
"pointer_toggle", "sens_up", "sens_down", "layout_reset", "screens_toggle", "key", "none")
SCREENS = "\0ft_screens"
KEYS = "\0frametop_keys" # keys of keyboards grabbed for the desktop, for other readers
FT_LAYOUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "layout", "ft-layout")
DEFAULT_BUTTONS = {BTN_LEFT: "left", BTN_RIGHT: "right", BTN_MIDDLE: "middle",
BTN_SIDE: "back", BTN_EXTRA: "back"}
def log(*args):
print(*args, flush=True)
def notify(state, fds=()):
"""sd_notify, with optional file descriptors for the fd store. No-op outside systemd."""
addr = os.environ.get("NOTIFY_SOCKET")
if not addr:
return
if addr.startswith("@"):
addr = "\0" + addr[1:]
# socket.send_fds() ignores its address argument (Python 3.12), so use sendmsg.
ancillary = [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds))] if fds else []
try:
with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
sock.sendmsg([state.encode()], ancillary, 0, addr)
except OSError as e:
log(f"sd_notify failed ({state.splitlines()[0]}): {e}")
def stored_fds():
"""File descriptors handed back by systemd's fd store, by name."""
if os.environ.get("LISTEN_PID") != str(os.getpid()):
return {}
names = os.environ.get("LISTEN_FDNAMES", "").split(":")
count = int(os.environ.get("LISTEN_FDS", "0"))
return {names[i]: 3 + i for i in range(count) if i < len(names)}
class Virtual:
"""One uinput device, reused from systemd's fd store when possible."""
def __init__(self, name, product, keys, rels, stored):
self.dirty = False
store_name = f"vdev{product}"
if store_name in stored:
self.fd = stored[store_name]
os.set_blocking(self.fd, False)
log(f"reusing {name} from the fd store")
return
self.fd = os.open("/dev/uinput", os.O_WRONLY | os.O_NONBLOCK)
fcntl.ioctl(self.fd, UI_SET_EVBIT, EV_KEY)
for code in keys:
fcntl.ioctl(self.fd, UI_SET_KEYBIT, code)
if rels:
fcntl.ioctl(self.fd, UI_SET_EVBIT, EV_REL)
for code in rels:
fcntl.ioctl(self.fd, UI_SET_RELBIT, code)
setup = struct.pack("HHHH80sI", BUS_VIRTUAL, 0x4D44, product, 1, name.encode(), 0)
fcntl.ioctl(self.fd, UI_DEV_SETUP, setup)
fcntl.ioctl(self.fd, UI_DEV_CREATE)
notify(f"FDSTORE=1\nFDNAME={store_name}", [self.fd])
log(f"created {name}")
def emit(self, etype, code, value):
os.write(self.fd, EVENT.pack(0, 0, etype, code, value))
self.dirty = True
def sync(self):
if self.dirty:
os.write(self.fd, EVENT.pack(0, 0, EV_SYN, SYN_REPORT, 0))
self.dirty = False
def bits(fd, ev, count):
buf = bytearray((count + 7) // 8)
try:
fcntl.ioctl(fd, eviocgbit(ev, len(buf)), buf)
except OSError:
return set()
return {i for i in range(count) if buf[i // 8] >> (i % 8) & 1}
def read_config(path=os.path.expanduser("~/.config/frametop.conf")):
"""KEY=VALUE lines, # comments allowed. Missing file means defaults."""
conf = {}
try:
with open(path) as f:
for line in f:
line = line.split("#", 1)[0].strip()
if "=" in line:
key, value = line.split("=", 1)
conf[key.strip()] = value.strip()
except OSError:
pass
return conf
def read_rules(path=RULES_PATH):
"""{"devices": {id: {"role", "name"}}, "buttons": {id: {"<code>": action}}}."""
try:
with open(path) as f:
rules = json.load(f)
except (OSError, ValueError):
rules = {}
rules.setdefault("devices", {})
rules.setdefault("buttons", {})
return rules
def remap_volume(fd, restore=False):
"""Point a device's volume keys at their stand-ins in its keymap, or back with restore.
Returns how many keymap entries are volume keys or stand-ins, or None when the
device has no keymap to change (uinput devices, some platform buttons).
"""
swap = VOLUME_ORIGINAL if restore else VOLUME_STANDIN
found = 0
for index in range(8192):
entry = bytearray(KEYMAP_ENTRY.pack(INPUT_KEYMAP_BY_INDEX, 0, index, 0, b""))
try:
fcntl.ioctl(fd, EVIOCGKEYCODE_V2, entry)
except OSError:
return found if index else None # past the last entry
_, length, _, code, scancode = KEYMAP_ENTRY.unpack(entry)
if code in VOLUME_CODES:
found += 1
if code in swap:
fcntl.ioctl(fd, EVIOCSKEYCODE_V2,
KEYMAP_ENTRY.pack(INPUT_KEYMAP_BY_INDEX, length, index, swap[code], scancode))
return found
class Volume:
"""Volume keys: wpctl steps the default output, repeating while a key is held.
The repeat is our own, since the headset's buttons have none; kernel autorepeat
from keyboards is ignored so every device repeats the same way.
"""
STEP = 5 # percent
DELAY, RATE = 0.4, 0.1 # seconds before repeating, and between repeats
def __init__(self):
self.held = None # (fd, code) of the key being held
self.next_at = None
def key(self, fd, code, value, now):
if value == 1:
self.held = (fd, code)
self.step(code)
self.next_at = now + self.DELAY
elif value == 0 and self.held == (fd, code):
self.release()
def release(self):
self.held = self.next_at = None
def step(self, code):
sign = "+" if VOLUME_ORIGINAL.get(code, code) == KEY_VOLUMEUP else "-"
subprocess.Popen(["wpctl", "set-volume", "--limit", "1.0", "@DEFAULT_AUDIO_SINK@",
f"{self.STEP}%{sign}"],
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
def tick(self, now):
if self.next_at is not None and now >= self.next_at:
self.step(self.held[1])
self.next_at = now + self.RATE
def timeout(self, now, default):
return default if self.next_at is None else max(0.0, min(default, self.next_at - now))
class Pointer:
"""Drives the ft_pointer SteamVR driver from a mouse (pointer mode).
The virtual controller connects when the mouse is used (taking the right
hand role and recentering on the gaze) and disconnects after `idle` seconds
without mouse activity, so the real controllers get their role back: the
last used device wins.
"""
DRIVER_BUTTONS = {"left": "trigger", "right": "b", "middle": "x", "back": "joystick"}
SCROLL_PULSE = 0.08 # seconds of joystick deflection per wheel notch
CLAIM_PULSE = 0.06 # seconds the claim button (switchlaserhand, no click) is held
RESUME_PAUSE = 1.5 # mouse idle this long, then moving again, re-claims the laser
WAKE_WINDOW = 1.0 # seconds in which WAKE_COUNTS of motion must add up
def __init__(self, sensitivity, idle, wake_counts=40):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
self.sensitivity = sensitivity # degrees per mouse count
self.idle = idle
self.active = False
self.last_used = 0.0
self.dx = self.dy = 0
self.scroll_until = None
self.claim_at = None # when to press the claim button
self.claim_release = None
self.system_at = None # dashboard toggle: when to press the virtual system button
self.system_release = None
# Waking (or re-claiming after a pause) needs deliberate movement, so sensor
# jitter from a mouse lying on a desk can't steal the laser from a controller.
self.wake_counts = wake_counts
self.pending = 0
self.pending_since = 0.0
def send(self, command):
try:
self.sock.sendto(command.encode(), "\0ft_pointer_helper")
except OSError:
pass # helper not running (SteamVR not running)
def wake(self, now):
if not self.active:
self.send("show")
self.send("recenter")
self.active = True
self.claim_at = now + 0.3 # let SteamVR bind the freshly connected device first
log("pointer on")
elif now - self.last_used > self.RESUME_PAUSE and self.claim_at is None:
self.claim_at = now # another device may have taken the laser meanwhile
self.last_used = now
def motion(self, code, value, now):
dormant = not self.active or now - self.last_used > self.RESUME_PAUSE
if dormant and code in (REL_X, REL_Y):
if now - self.pending_since > self.WAKE_WINDOW:
self.pending, self.pending_since = 0, now
self.pending += abs(value)
if self.pending < self.wake_counts:
return # not yet deliberate movement
self.pending = 0
self.wake(now)
if code == REL_X:
self.dx += value
elif code == REL_Y:
self.dy += value
elif code == REL_WHEEL and value:
self.send(f"scroll 0 {1 if value > 0 else -1}")
self.scroll_until = now + self.SCROLL_PULSE
def action(self, name, value, now):
"""A mapped button: value 1 press, 0 release, 2 autorepeat (ignored)."""
if value == 2:
return
driver = self.DRIVER_BUTTONS.get(name)
if driver:
self.wake(now)
self.flush()
self.send(f"btn {driver} {value}")
elif value != 1:
return # the rest act on press
elif name in ("scroll_up", "scroll_down"):
self.wake(now)
self.send(f"scroll 0 {1 if name == 'scroll_up' else -1}")
self.scroll_until = now + self.SCROLL_PULSE
elif name == "dashboard":
self.dashboard(now)
elif name == "recenter":
self.wake(now)
self.send("recenter")
elif name == "pointer_toggle":
if self.active:
self.send("hide")
self.active = False
log("pointer off (toggle)")
else:
self.wake(now)
elif name == "screens_toggle":
try:
self.sock.sendto(b"toggle", SCREENS)
except OSError:
pass # ft-screens not running
elif name == "layout_reset":
# Runs a few seconds and borrows the pointer; ft-layout refuses a second copy.
subprocess.Popen([FT_LAYOUT, "apply"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, start_new_session=True)
log("layout reset")
elif name in ("sens_up", "sens_down"):
self.sensitivity *= 1.25 if name == "sens_up" else 0.8
log(f"sensitivity {self.sensitivity:.4f} deg/count")
def flush(self):
if self.dx or self.dy:
# Mouse right turns the ray right (negative yaw); mouse down tilts it down.
self.send(f"move {-self.dx * self.sensitivity:.4f} {-self.dy * self.sensitivity:.4f}")
self.dx = self.dy = 0
def dashboard(self, now=None):
"""Toggle the SteamVR dashboard with the virtual controller's system button.
SteamVR needs the button held for a frame or two (press and release in the
same instant is ignored), and the virtual controller must be connected and
bound, so wake it first when needed.
"""
now = time.monotonic() if now is None else now
woke = not self.active
self.wake(now)
self.system_at = now + (0.4 if woke else 0.0)
def tick(self, now):
if self.system_at is not None and now >= self.system_at:
self.send("btn system 1")
self.system_at = None
self.system_release = now + 0.12
elif self.system_release is not None and now >= self.system_release:
self.send("btn system 0")
self.system_release = None
if self.claim_at is not None and now >= self.claim_at:
self.send("btn a 1")
self.claim_at = None
self.claim_release = now + self.CLAIM_PULSE
elif self.claim_release is not None and now >= self.claim_release:
self.send("btn a 0")
self.claim_release = None
if self.scroll_until is not None and now >= self.scroll_until:
self.send("scroll 0 0")
self.scroll_until = None
if self.active and now - self.last_used > self.idle:
self.send("hide")
self.active = False
log("pointer off (idle)")
def timeout(self):
pending = (self.scroll_until, self.claim_at, self.claim_release, self.system_at, self.system_release)
return 0.02 if any(t is not None for t in pending) else 0.5
class Node:
"""One input event node: a candidate device (mouse or keyboard, USB or Bluetooth),
or any other device with volume keys (candidate False, role "volume")."""
def __init__(self, path, fd, name, bus, vendor, product, uniq, is_mouse, is_keyboard,
candidate=True, volume_keys=False, only_volume=False):
self.path, self.fd, self.name = path, fd, name
self.bus, self.vendor, self.product, self.uniq = bus, vendor, product, uniq
self.is_mouse, self.is_keyboard = is_mouse, is_keyboard
self.candidate, self.volume_keys, self.only_volume = candidate, volume_keys, only_volume
# One physical device, whatever its node: Bluetooth address, else USB ids plus name.
base = self.name.split(" Mouse")[0].split(" Keyboard")[0]
self.id = uniq.lower() if uniq else f"usb:{vendor:04x}:{product:04x}:{base}"
self.role = None if candidate else "volume"
self.grabbed = False
self.remapped = False # volume keys remapped to their stand-ins
self.held = set() # keys and buttons currently down, released if the device vanishes
self.last_watch = 0.0
def describe(self):
kinds = [k for k, on in (("mouse", self.is_mouse), ("keyboard", self.is_keyboard)) if on]
return {"path": self.path, "name": self.name, "id": self.id, "uniq": self.uniq,
"bus": {BUS_USB: "usb", BUS_BLUETOOTH: "bluetooth"}.get(self.bus, str(self.bus)),
"kinds": kinds, "role": self.role, "grabbed": self.grabbed}
def probe(path):
"""Open a node if it is a USB or Bluetooth mouse or keyboard, or has volume keys,
else return None."""
try:
fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
except OSError:
return None
try:
buf = bytearray(256)
fcntl.ioctl(fd, eviocgname(len(buf)), buf)
name = buf.split(b"\0", 1)[0].decode(errors="replace")
ident = bytearray(8)
fcntl.ioctl(fd, EVIOCGID, ident)
bus, vendor, product, _ = struct.unpack("HHHH", ident)
if name.startswith(VIRTUAL_PREFIX):
raise ValueError
uniq_buf = bytearray(64)
try:
fcntl.ioctl(fd, eviocguniq(len(uniq_buf)), uniq_buf)
uniq = uniq_buf.split(b"\0", 1)[0].decode(errors="replace")
except OSError:
uniq = ""
keys = bits(fd, EV_KEY, KEY_MAX + 1)
is_mouse = REL_X in bits(fd, EV_REL, REL_MAX + 1)
is_keyboard = KEY_A in keys
candidate = bus in (BUS_USB, BUS_BLUETOOTH) and (is_mouse or is_keyboard)
volume_keys = bool(keys & VOLUME_CODES) # stand-ins too: kept from before a relay restart
if not (candidate or volume_keys):
raise ValueError
return Node(path, fd, name, bus, vendor, product, uniq, is_mouse, is_keyboard,
candidate, volume_keys, keys <= VOLUME_CODES)
except (OSError, ValueError):
os.close(fd)
return None
def main():
can_grab = "--no-grab" not in sys.argv
stored = stored_fds()
mouse = Virtual(f"{VIRTUAL_PREFIX} mouse", 1,
keys=range(BTN_MISC, 0x118), rels=range(REL_MAX + 1), stored=stored)
keyboard = Virtual(f"{VIRTUAL_PREFIX} keyboard", 2,
keys=range(1, BTN_MISC), rels=(), stored=stored)
notify("READY=1")
control = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
control.bind("\0frametop_relay")
control.setblocking(False)
watchers = {} # address -> watch end time
# desktop_until: typing goes to the Frametop desktop until then (ft-screens says so
# every second); typing_applied: the grabs match that as of the last apply_roles().
state = {"pointer": None, "rules": {}, "meta_dashboard": False, "share_keys": False,
"desktop_until": 0.0, "typing_applied": None}
def load_config():
conf = read_config()
state["rules"] = read_rules()
state["meta_dashboard"] = conf.get("META_DASHBOARD", "0") == "1"
state["share_keys"] = conf.get("SHARE_KEYS", "0") == "1"
if conf.get("POINTER", "0") == "1":
p = state["pointer"] or Pointer(0.03, 30)
p.sensitivity = float(conf.get("POINTER_SENSITIVITY", "0.03"))
p.idle = float(conf.get("POINTER_IDLE", "30"))
p.wake_counts = int(conf.get("POINTER_WAKE_COUNTS", "40"))
state["pointer"] = p
log(f"pointer mode: {p.sensitivity} deg/count, idle {p.idle} s, wake {p.wake_counts} counts")
else:
state["pointer"] = None
log("pointer mode off: pointer devices feed the virtual mouse and keyboard")
load_config()
meta_down = False # Meta pressed with no other key yet: a tap toggles the dashboard
screens_sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_NONBLOCK)
def to_screens(code, value):
"""A key for the desktop screens (ft-screens decides whether it types)."""
if value in (0, 1) and code < BTN_MISC:
try:
screens_sock.sendto(f"key {code} {value}".encode(), SCREENS)
except OSError:
pass # ft-screens not running
nodes = {} # fd -> Node
# Nodes already probed (rejected or open): path -> inode. A device that disconnects and
# reconnects between two scans often gets the same event numbers back, so the path alone
# would hide it; the re-created node has a new inode.
seen = {}
next_scan = 0.0
volume = Volume()
keys_sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_NONBLOCK)
def share_key(node, code, value):
"""With SHARE_KEYS=1, a key from a keyboard grabbed for the desktop, for programs
that watch every keyboard for a hotkey and lose it to the grab. It can't go on
another input device: gamescope reads every keyboard itself and would type it
into its focused app."""
if not state["share_keys"]:
return
try:
keys_sock.sendto(f"key {code} {value} {node.name}".encode(), KEYS)
except OSError:
pass # nobody listening
def restore_keymaps():
"""Give remapped devices their volume keys back, so they work without the relay."""
for node in nodes.values():
if node.remapped:
try:
remap_volume(node.fd, restore=True)
except OSError:
pass # device already gone
atexit.register(restore_keymaps)
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0)) # so atexit runs on systemctl stop
def take_volume(node):
"""Keep the node's volume keys from gamescope and SteamVR, which read every device.
Returns False for a non-candidate node there's nothing to do with.
"""
if not can_grab:
return node.candidate
try:
found = remap_volume(node.fd)
except OSError as e:
log(f"remapping volume keys failed for {node.name}: {e}")
# Some entries may have changed already: handle their stand-ins and restore them.
node.remapped = True
found = None
if found:
node.remapped = True
log(f"{node.name} ({node.path}): volume keys taken over (remapped)")
return True
if found is None and node.only_volume:
try:
fcntl.ioctl(node.fd, EVIOCGRAB, 1)
node.grabbed = True
log(f"{node.name} ({node.path}): volume keys taken over (grabbed)")
return True
except OSError as e:
log(f"grab failed for {node.name}: {e}")
# Grabbed pointer devices still have their volume keys handled here.
log(f"{node.name} ({node.path}): can't take over its volume keys")
return node.candidate
def role_of(node):
rule = state["rules"]["devices"].get(node.id, {})
if rule.get("role") in ("pointer", "passthrough", "ignore"):
return rule["role"]
has_mouse = any(n.is_mouse for n in nodes.values() if n.id == node.id) or node.is_mouse
return "pointer" if has_mouse else "passthrough"
def release_held(node):
for code in node.held:
if node.role == "passthrough":
share_key(node, code, 0)
else:
(mouse if code >= BTN_MISC else keyboard).emit(EV_KEY, code, 0)
node.held.clear()
mouse.sync()
keyboard.sync()
def keys_down(node):
buf = bytearray((KEY_MAX + 8) // 8)
try:
fcntl.ioctl(node.fd, EVIOCGKEY, buf)
except OSError:
return False
return any(buf)
def apply_roles():
"""Grab pointer devices, and pass-through keyboards while typing goes to the desktop.
A keyboard with a key down keeps its grab state until it's released, or the key
would stay held on one side. Returns True if one is still waiting.
"""
desktop = time.monotonic() < state["desktop_until"]
waiting = False
for node in nodes.values():
if not node.candidate:
continue # volume keys only, taken over when found
role = role_of(node)
want_grab = can_grab and (role == "pointer"
or (role == "passthrough" and node.is_keyboard and desktop))
if want_grab != node.grabbed and role == "passthrough" and keys_down(node):
waiting = True
elif want_grab != node.grabbed:
try:
fcntl.ioctl(node.fd, EVIOCGRAB, 1 if want_grab else 0)
node.grabbed = want_grab
except OSError as e:
log(f"{'grab' if want_grab else 'ungrab'} failed for {node.name}: {e}")
if not node.grabbed:
release_held(node)
if role != node.role:
log(f"{node.name} ({node.path}, {node.id}): {role}{', grabbed' if node.grabbed else ''}")
node.role = role
if not waiting and desktop != state["typing_applied"]:
state["typing_applied"] = desktop
log(f"typing goes to {'the desktop (keyboards grabbed)' if desktop else 'Steam'}")
return waiting
def drop(node, reason):
release_held(node)
if volume.held and volume.held[0] == node.fd:
volume.release()
os.close(node.fd)
del nodes[node.fd]
seen.pop(node.path, None)
log(f"released {node.name} ({node.path}): {reason}")
def reply(addr, obj):
try:
control.sendto(json.dumps(obj).encode(), addr)
except OSError:
watchers.pop(addr, None)
def handle_control(now):
while True:
try:
data, addr = control.recvfrom(4096)
except BlockingIOError:
return
words = data.decode(errors="replace").split()
cmd = words[0] if words else ""
if cmd == "keyboard":
# From ft-screens (unbound, no reply): where typing goes, repeated every second.
desktop = len(words) > 1 and words[1] == "desktop"
state["desktop_until"] = now + 3.0 if desktop else 0.0
continue
if not addr:
continue # unbound sender, nowhere to reply
if cmd == "devices":
reply(addr, {"t": "devices", "pointer_mode": state["pointer"] is not None,
"actions": ACTIONS,
"nodes": [n.describe() for n in nodes.values() if n.candidate]})
elif cmd == "watch":
seconds = float(words[1]) if len(words) > 1 else 30
watchers[addr] = now + min(seconds, 600)
reply(addr, {"t": "watching", "seconds": seconds})
elif cmd == "reload":
load_config()
apply_roles()
if state["pointer"]:
state["pointer"].send("reload")
reply(addr, {"t": "reloaded"})
else:
reply(addr, {"t": "error", "error": f"unknown command {cmd!r}"})
def broadcast(node, etype, code, value, now):
if not watchers or not node.candidate:
return
if etype == EV_REL and now - node.last_watch < 0.05:
return # motion: enough for an activity light
node.last_watch = now
msg = {"t": "event", "id": node.id, "path": node.path, "name": node.name,
"type": EV_NAMES.get(etype, str(etype)), "code": code, "value": value}
for addr, until in list(watchers.items()):
if now > until:
del watchers[addr]
else:
reply(addr, msg)
waiting = False # a keyboard's grab waits for its keys to come up
while True:
now = time.monotonic()
pointer = state["pointer"]
if now >= next_scan:
next_scan = now + 1.0
current = {}
for name in os.listdir("/dev/input"):
if name.startswith("event"):
try:
current[f"/dev/input/{name}"] = os.stat(f"/dev/input/{name}").st_ino
except OSError:
pass
for path in [p for p in seen if p not in current]:
del seen[path]
added = False
for path, ino in sorted(current.items()):
if seen.get(path) == ino:
continue
# New here: a new device, or one that came back in the same place.
for old in [n for n in nodes.values() if n.path == path]:
drop(old, "replaced by a new device node")
seen[path] = ino
node = probe(path)
if node and node.volume_keys and not take_volume(node):
os.close(node.fd)
node = None
if node:
nodes[node.fd] = node
added = True
if added:
apply_roles()
ready, _, _ = select.select(list(nodes) + [control], [], [],
volume.timeout(now, pointer.timeout() if pointer else 0.5))
now = time.monotonic()
if pointer:
pointer.tick(now)
volume.tick(now)
if (now < state["desktop_until"]) != state["typing_applied"] or waiting:
waiting = apply_roles()
for fd in ready:
if fd is control:
handle_control(now)
continue
node = nodes[fd]
try:
data = os.read(fd, EVENT.size * 64)
except OSError as e:
if e.errno == errno.EAGAIN:
continue
drop(node, os.strerror(e.errno))
continue
if not data:
drop(node, "closed")
continue
buttons = state["rules"]["buttons"].get(node.id, {})
for off in range(0, len(data) - EVENT.size + 1, EVENT.size):
_, _, etype, code, value = EVENT.unpack_from(data, off)
if etype in (EV_KEY, EV_REL):
broadcast(node, etype, VOLUME_ORIGINAL.get(code, code) if node.remapped else code,
value, now)
if etype == EV_KEY and ((node.remapped and code in VOLUME_ORIGINAL)
or (node.grabbed and code in VOLUME_STANDIN)):
volume.key(fd, code, value, now)
if value == 1:
meta_down = False # Meta used as a modifier, not a tap
continue
if node.role == "volume":
continue
if node.role != "pointer":
# Observed only, unless typing goes to the desktop. With META_DASHBOARD=1,
# a Meta tap on any keyboard toggles the dashboard.
if node.role == "passthrough" and etype == EV_KEY:
to_screens(code, value)
if node.grabbed and code < BTN_MISC and value in (0, 1):
share_key(node, code, value)
if value:
node.held.add(code)
else:
node.held.discard(code)
if (pointer and state["meta_dashboard"] and node.role == "passthrough"
and etype == EV_KEY):
if code in (KEY_LEFTMETA, KEY_RIGHTMETA):
if value == 1:
meta_down = True
elif value == 0 and meta_down:
meta_down = False
pointer.dashboard()
elif value == 1:
meta_down = False # Meta used as a modifier, not a tap
continue
if etype == EV_KEY:
action = buttons.get(str(code), DEFAULT_BUTTONS.get(code, "key"))
if pointer and action not in ("key", "none"):
pointer.action(action, value, now)
continue
if action == "none":
continue
target = mouse if code >= BTN_MISC else keyboard
target.emit(etype, code, value)
to_screens(code, value)
if value:
node.held.add(code)
else:
node.held.discard(code)
elif etype == EV_REL:
if pointer:
pointer.motion(code, value, now)
else:
mouse.emit(etype, code, value)
elif etype == EV_SYN and code == SYN_REPORT:
if pointer:
pointer.flush()
mouse.sync()
keyboard.sync()
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
pass
Reference in new issue View Git Blame Copy Permalink