99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from typing import Callable
|
|
|
|
from evdev import InputDevice, categorize, ecodes, list_devices
|
|
|
|
from .input_devices import inaccessible_keyboard_names as _inaccessible_keyboard_names
|
|
from .input_devices import is_keyboard as _is_keyboard
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class HotkeyListener:
|
|
"""Reads raw key-down events from /dev/input via evdev and reports, on
|
|
every key-down, the full set of currently-held keys plus which one was
|
|
just pressed - so titles can be bound to multi-key combos (e.g.
|
|
Ctrl+F1) that still fire while an unrelated key (a movement key, most
|
|
likely) happens to also be held; see TitleTimerEngine.handle_hotkey for
|
|
the actual matching rules.
|
|
|
|
This works regardless of which Wayland compositor is running (unlike
|
|
compositor-specific global-shortcut APIs), because it reads the input
|
|
device directly rather than going through the window system. The
|
|
user's account needs read access to /dev/input/event*, i.e. membership
|
|
in the `input` group.
|
|
"""
|
|
|
|
def __init__(self, on_hotkey: Callable[[frozenset[str], str], None]):
|
|
self._on_hotkey = on_hotkey
|
|
self._threads: list[threading.Thread] = []
|
|
self._stop = threading.Event()
|
|
# Shared across every keyboard-reading thread (in case someone has
|
|
# more than one keyboard plugged in) - a combo could span devices.
|
|
self._held: set[str] = set()
|
|
self._lock = threading.Lock()
|
|
|
|
def start(self) -> None:
|
|
devices = [InputDevice(path) for path in list_devices()]
|
|
logger.debug(
|
|
"Input devices seen: %s",
|
|
[(d.path, d.name) for d in devices],
|
|
)
|
|
missing = _inaccessible_keyboard_names()
|
|
if missing:
|
|
logger.warning(
|
|
"Kernel reports keyboard device(s) that this process can't "
|
|
"actually open (permission denied): %s. These will NOT "
|
|
"receive hotkeys. If you just added your user to the "
|
|
"'input' group, a full logout/login is required for that "
|
|
"to take effect - a new terminal in the same session isn't "
|
|
"enough.",
|
|
missing,
|
|
)
|
|
|
|
keyboards = [d for d in devices if _is_keyboard(d)]
|
|
if not keyboards:
|
|
raise RuntimeError(
|
|
"No keyboard-like input device found under /dev/input. "
|
|
"Make sure your user is in the 'input' group (and re-login)."
|
|
)
|
|
logger.info(
|
|
"Listening on %d keyboard device(s): %s",
|
|
len(keyboards), [(d.path, d.name) for d in keyboards],
|
|
)
|
|
for dev in keyboards:
|
|
t = threading.Thread(target=self._read_loop, args=(dev,), daemon=True)
|
|
t.start()
|
|
self._threads.append(t)
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
|
|
def _read_loop(self, dev: InputDevice) -> None:
|
|
try:
|
|
for event in dev.read_loop():
|
|
if self._stop.is_set():
|
|
return
|
|
if event.type != ecodes.EV_KEY:
|
|
continue
|
|
key_event = categorize(event)
|
|
keycode = key_event.keycode
|
|
# evdev sometimes reports a list of aliases for one scancode
|
|
name = keycode[0] if isinstance(keycode, list) else keycode
|
|
with self._lock:
|
|
if key_event.keystate == key_event.key_down:
|
|
self._held.add(name)
|
|
held = frozenset(self._held)
|
|
elif key_event.keystate == key_event.key_up:
|
|
self._held.discard(name)
|
|
continue
|
|
else:
|
|
continue
|
|
logger.debug("Key down: %s (device %s), held=%s", name, dev.path, held)
|
|
self._on_hotkey(held, name)
|
|
except OSError:
|
|
logger.warning("Input device %s disconnected", dev.path)
|