459 lines
17 KiB
Python
459 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
# hyprland-tablet-daemon.py — Hyprland tablet-mode rotation + on-screen keyboard.
|
|
#
|
|
# Wired to the ThinkPad X1 Yoga Gen 6 (Intel HID switches, /dev/input/event17):
|
|
# * SW_TABLET_MODE tells us the LCD is folded into tablet mode.
|
|
# * "Accelerometer orientation changed: <orient>" lines are parsed from
|
|
# `monitor-sensor` (a thin client of iio-sensor-proxy, which runs as a
|
|
# system service on this host). iio-sensor-proxy merges the accel axes +
|
|
# mount matrix into a single compass-style orientation string.
|
|
#
|
|
# While in tablet mode the eDP-1 monitor transform (and matching per-device
|
|
# transforms for touch inputs) follow the accelerometer. The on-screen
|
|
# keyboard (wvkbd) follows text focus instead of Hyprland's input-method v2
|
|
# protocol (which Hyprland advertises but never fires):
|
|
# * We watch `.socket2.sock` "activewindow>>class,title" events. If the
|
|
# focused window class is in the text-capable allowlist the keyboard is
|
|
# shown (SIGUSR2), otherwise hidden (SIGUSR1). This gives reliable
|
|
# auto-hide when there is nothing to type into.
|
|
# * Manual control (super+crtl+k / the bar chip) flips wvkbd directly via
|
|
# SIGRTMIN. The daemon re-applies focus state on the next focus change,
|
|
# so a manual dismissal lasts until the user switches windows.
|
|
# Leaving tablet mode snaps transform back to 0 and kills wvkbd.
|
|
#
|
|
# State is mirrored to a small JSON file ($XDG_RUNTIME_DIR/hyprland-tablet)
|
|
# that the QuickShell bar chip reads via `hyprland-tablet status`:
|
|
# {"tablet": bool, "osk": visible-or-not}
|
|
#
|
|
# Control commands arrive over a Unix datagram socket
|
|
# ($XDG_RUNTIME_DIR/hyprland-tablet.ctl): "toggle" | "show" | "hide".
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
# Orientation string (from iio-sensor-proxy) -> Hyprland monitor transform.
|
|
# Hyprland transform: 0 normal, 1 90° CCW, 2 180°, 3 270° CCW.
|
|
# If the screen rotates the wrong way on hardware, swap the 1 and 3 below.
|
|
ORIENTATION_TRANSFORM = {
|
|
"normal": 0,
|
|
"left-up": 1,
|
|
"bottom-up": 2,
|
|
"right-up": 3,
|
|
}
|
|
|
|
SW_TABLET_MODE = 0x01 # input event code for SW_TABLET_MODE
|
|
|
|
# Paths resolved at build time by the Nix module (wrapped with pkgs.python3).
|
|
# The placeholder strings below are in at-sign-delimited form because
|
|
# pkgs.replaceVarsWith substitutes exactly that syntax.
|
|
EVDEV_DEVICE = "@EVDEV_DEVICE@"
|
|
WVKBD_PATH = "@WVKBD_PATH@"
|
|
WVKBD_ARGS = @WVKBD_ARGS@ # replaced with a JSON array (valid Python list)
|
|
MONITOR_SENSOR = "@MONITOR_SENSOR@"
|
|
HYPRCTL = "@HYPRCTL@"
|
|
TEXT_APPS = @TEXT_APPS_JSON@ # replaced with a JSON array (valid Python list)
|
|
# State file lives in $XDG_RUNTIME_DIR (set by the user systemd service); the
|
|
# QuickShell bar chip + CLI read the same path.
|
|
STATE_FILE = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "hyprland-tablet")
|
|
CONTROL_SOCKET = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "hyprland-tablet.ctl")
|
|
|
|
DEBUG = os.environ.get("HYPRLAND_TABLET_DEBUG") == "1"
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
if DEBUG:
|
|
print(f"[hyprland-tablet] {msg}", file=sys.stderr, flush=True)
|
|
|
|
|
|
_state_lock = threading.Lock()
|
|
|
|
|
|
def write_state(tablet: bool, osk: bool) -> None:
|
|
try:
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump({"tablet": tablet, "osk": osk}, f)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def hyprctl_transform(monitor: str, transform: int) -> None:
|
|
"""Apply a transform to the monitor and every touch/tablet device."""
|
|
# hyprland.nix uses the Lua config parser (0.55+), so the legacy
|
|
# `monitor NAME,transform,N` / input keyword lines don't apply at runtime
|
|
# ("keyword can't work with non-legacy parsers"). The equivalent Lua API:
|
|
# hl.monitor({ output = "NAME", mode = ..., scale = ..., transform = N })
|
|
# hl.config({ input = { touchdevice/tablet = { transform = N } } })
|
|
# scale MUST be passed explicitly: omitting it makes Hyprland re-derive
|
|
# HiDPI zoom (1.5 here), blowing up the bar on rotation.
|
|
tx = [HYPRCTL, "eval", f'hl.monitor({{ output = "{monitor}", mode = "preferred", position = "auto", scale = 1, transform = {transform} }})']
|
|
subprocess.run(tx, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
|
|
# touchdevice transform follows the display so touch coordinates track the
|
|
# rotated framebuffer. input:touchdevice:transform is a global input option
|
|
# (not per-device device:touchdevice:transform which doesn't exist).
|
|
subprocess.run(
|
|
[HYPRCTL, "eval", f'hl.config({{ input = {{ touchdevice = {{ transform = {transform} }} }} }})'],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
# same for the tablet (pen) — this Yoga presents the Wacom digitizer as a
|
|
# tablet device, and without this the pen coordinates stay in screen space.
|
|
subprocess.run(
|
|
[HYPRCTL, "eval", f'hl.config({{ input = {{ tablet = {{ transform = {transform} }} }} }})'],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def transform_for(orientation: str) -> int:
|
|
return ORIENTATION_TRANSFORM.get(orientation, 0)
|
|
|
|
|
|
def _hypr_socket(name: str) -> str:
|
|
"""Locate a Hyprland IPC socket (".socket.sock" / ".socket2.sock")."""
|
|
runtime = os.environ.get("XDG_RUNTIME_DIR", "/tmp")
|
|
inst = os.environ.get("HYPRLAND_INSTANCE_SIGNATURE")
|
|
if inst:
|
|
candidate = os.path.join(runtime, "hypr", inst, name)
|
|
if os.path.exists(candidate):
|
|
return candidate
|
|
base = os.path.join(runtime, "hypr")
|
|
try:
|
|
for entry in os.listdir(base):
|
|
candidate = os.path.join(base, entry, name)
|
|
if os.path.exists(candidate):
|
|
return candidate
|
|
except OSError:
|
|
pass
|
|
return ""
|
|
|
|
|
|
class Daemon:
|
|
def __init__(self, monitor: str) -> None:
|
|
self.monitor = monitor
|
|
self.tablet = False
|
|
self.orientation = "normal"
|
|
self.current_transform = 0
|
|
self.osk_proc = None
|
|
self.osk_visible = False
|
|
self.dev = None
|
|
self.sensor = None
|
|
self.last_focus_class = ""
|
|
|
|
# -------------------------------------------------------------
|
|
# evdev switch watcher (thread)
|
|
# -------------------------------------------------------------
|
|
def open_evdev(self) -> bool:
|
|
try:
|
|
import evdev
|
|
import evdev.ecodes as ecodes
|
|
|
|
self.dev = evdev.InputDevice(EVDEV_DEVICE)
|
|
except Exception as e:
|
|
log(f"cannot open {EVDEV_DEVICE}: {e}")
|
|
return False
|
|
# Seed the current switch state via EVIOCGSW so the daemon is correct
|
|
# if it starts mid-tablet.
|
|
EVIOCGSW = (2 << 30) | (ord("E") << 8) | 0x1B | (64 << 16)
|
|
buf = bytearray(64)
|
|
try:
|
|
import fcntl
|
|
|
|
fcntl.ioctl(self.dev.fd, EVIOCGSW, buf)
|
|
vals = struct.unpack("16i", buf)
|
|
self.tablet = bool(vals[0] >> SW_TABLET_MODE & 1)
|
|
except OSError:
|
|
self.tablet = False
|
|
log(f"opened {self.dev.name}, starting tablet={self.tablet}")
|
|
return True
|
|
|
|
def evdev_thread(self) -> None:
|
|
import evdev.ecodes as ecodes
|
|
|
|
for event in self.dev.read_loop():
|
|
if event.type != ecodes.EV_SW or event.code != SW_TABLET_MODE:
|
|
continue
|
|
new = event.value == 1
|
|
if new != self.tablet:
|
|
self.tablet = new
|
|
log(f"SW_TABLET_MODE -> {new}")
|
|
self.on_tablet_changed(new)
|
|
|
|
# -------------------------------------------------------------
|
|
# orientation watcher (thread)
|
|
# -------------------------------------------------------------
|
|
def _start_sensor(self) -> bool:
|
|
try:
|
|
self.sensor = subprocess.Popen(
|
|
[MONITOR_SENSOR],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
text=True,
|
|
bufsize=1,
|
|
)
|
|
except FileNotFoundError:
|
|
log("monitor-sensor not found")
|
|
return False
|
|
return self.sensor.stdout is not None
|
|
|
|
def sensor_thread(self) -> None:
|
|
# monitor-sensor (a thin client of iio-sensor-proxy) only emits
|
|
# "Accelerometer orientation changed:" on *changes*; the initial
|
|
# orientation arrives as "Has accelerometer (orientation: X, ...)".
|
|
# Match both, and restart the client if it ever exits so orientation
|
|
# tracking survives.
|
|
first = True
|
|
while True:
|
|
if not self._start_sensor():
|
|
return
|
|
if self.sensor is None or self.sensor.stdout is None:
|
|
return
|
|
if first:
|
|
log("monitor-sensor started")
|
|
first = False
|
|
for line in self.sensor.stdout:
|
|
m = re.search(r"Accelerometer orientation changed:\s*(\S+)", line.strip())
|
|
if m:
|
|
self.on_orientation(m.group(1))
|
|
continue
|
|
m = re.search(r"Has accelerometer \(orientation:\s*(\S+)", line.strip())
|
|
if m:
|
|
# value is "normal," / "left-up," — strip the separator
|
|
self.on_orientation(m.group(1).rstrip(","))
|
|
# monitor-sensor exited: reopen it (iio-sensor-proxy may have
|
|
# dropped our client if a second one connected).
|
|
log("monitor-sensor exited, restarting")
|
|
time.sleep(1)
|
|
|
|
# -------------------------------------------------------------
|
|
# Hyprland focus watcher (thread) — drives auto show/hide
|
|
# -------------------------------------------------------------
|
|
def focus_thread(self) -> None:
|
|
path = _hypr_socket(".socket2.sock")
|
|
if not path:
|
|
log("hyprland .socket2.sock not found; focus auto-hide disabled")
|
|
return
|
|
log(f"watching focus events on {path}")
|
|
while True:
|
|
try:
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.connect(path)
|
|
f = s.makefile("r")
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line.startswith("activewindow>>"):
|
|
continue
|
|
payload = line.split(">>", 1)[1].split(",", 1)
|
|
self.on_focus_changed(payload[0].strip())
|
|
except OSError as e:
|
|
log(f"focus socket error: {e}")
|
|
time.sleep(2)
|
|
|
|
def _is_text_app(self, window_class: str) -> bool:
|
|
if not window_class:
|
|
return False
|
|
lowered = window_class.lower()
|
|
for spec in TEXT_APPS:
|
|
if lowered == spec or lowered.startswith(spec):
|
|
return True
|
|
return False
|
|
|
|
def on_focus_changed(self, window_class: str) -> None:
|
|
self.last_focus_class = window_class
|
|
if not self.tablet:
|
|
return # in laptop mode the OSK duty is entirely manual
|
|
want = self._is_text_app(window_class)
|
|
log(f"focus={window_class!r} text_app={want}")
|
|
self.set_osk_visible(want)
|
|
|
|
# -------------------------------------------------------------
|
|
# control socket (thread) — manual show/hide/toggle
|
|
# -------------------------------------------------------------
|
|
def control_thread(self) -> None:
|
|
try:
|
|
os.unlink(CONTROL_SOCKET)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
|
|
s.bind(CONTROL_SOCKET)
|
|
s.settimeout(1.0)
|
|
except OSError as e:
|
|
log(f"cannot bind control socket: {e}")
|
|
return
|
|
log(f"control socket ready at {CONTROL_SOCKET}")
|
|
while True:
|
|
try:
|
|
data, _ = s.recvfrom(128)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
break
|
|
cmd = data.decode("ascii", "replace").strip()
|
|
log(f"control command: {cmd}")
|
|
if cmd == "toggle":
|
|
self.toggle_osk()
|
|
elif cmd == "show":
|
|
self.set_osk_visible(True)
|
|
elif cmd == "hide":
|
|
self.set_osk_visible(False)
|
|
|
|
# -------------------------------------------------------------
|
|
# actions
|
|
# -------------------------------------------------------------
|
|
def ensure_osk(self) -> None:
|
|
with _state_lock:
|
|
if self.osk_proc is not None and self.osk_proc.poll() is None:
|
|
return
|
|
try:
|
|
# Start hidden: focus state decides whether to SIGUSR2 it.
|
|
self.osk_proc = subprocess.Popen([WVKBD_PATH, *WVKBD_ARGS, "--hidden"])
|
|
self.osk_visible = False
|
|
log(f"wvkbd started (hidden, args={WVKBD_ARGS!r})")
|
|
except FileNotFoundError:
|
|
log("wvkbd not found")
|
|
write_state(self.tablet, self.osk_visible)
|
|
|
|
def kill_osk(self) -> None:
|
|
with _state_lock:
|
|
if self.osk_proc is not None and self.osk_proc.poll() is None:
|
|
self.osk_proc.terminate()
|
|
try:
|
|
self.osk_proc.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
self.osk_proc.kill()
|
|
self.osk_proc.wait()
|
|
log("wvkbd stopped")
|
|
self.osk_proc = None
|
|
self.osk_visible = False
|
|
write_state(self.tablet, False)
|
|
|
|
def _signal_osk(self, sig: int) -> None:
|
|
if self.osk_proc is not None and self.osk_proc.poll() is None:
|
|
try:
|
|
os.kill(self.osk_proc.pid, sig)
|
|
except OSError:
|
|
pass
|
|
|
|
def set_osk_visible(self, visible: bool) -> None:
|
|
if visible == self.osk_visible:
|
|
return
|
|
if visible:
|
|
if self.osk_proc is None or self.osk_proc.poll() is not None:
|
|
self.ensure_osk()
|
|
if self.osk_proc is None:
|
|
return
|
|
self._signal_osk(signal.SIGUSR2)
|
|
self.osk_visible = True
|
|
log("wvkbd show")
|
|
else:
|
|
self._signal_osk(signal.SIGUSR1)
|
|
self.osk_visible = False
|
|
log("wvkbd hide")
|
|
write_state(self.tablet, self.osk_visible)
|
|
|
|
def toggle_osk(self) -> None:
|
|
with _state_lock:
|
|
if self.osk_proc is None or self.osk_proc.poll() is not None:
|
|
self.osk_proc = subprocess.Popen([WVKBD_PATH, *WVKBD_ARGS, "--hidden"])
|
|
self.osk_visible = False
|
|
log("wvkbd started (hidden, manual toggle)")
|
|
write_state(self.tablet, self.osk_visible)
|
|
return
|
|
# wvkbd SIGRTMIN toggles visibility; we shadow its state here.
|
|
self.osk_visible = not self.osk_visible
|
|
self._signal_osk(signal.SIGRTMIN)
|
|
log(f"wvkbd toggled -> visible={self.osk_visible}")
|
|
write_state(self.tablet, self.osk_visible)
|
|
|
|
def apply_transform(self, t: int) -> None:
|
|
if t != self.current_transform:
|
|
hyprctl_transform(self.monitor, t)
|
|
self.current_transform = t
|
|
log(f"transform -> {t}")
|
|
|
|
# -------------------------------------------------------------
|
|
# event handlers
|
|
# -------------------------------------------------------------
|
|
def on_tablet_changed(self, new: bool) -> None:
|
|
if new:
|
|
self.apply_transform(transform_for(self.orientation))
|
|
self.ensure_osk()
|
|
else:
|
|
self.apply_transform(0)
|
|
self.kill_osk()
|
|
write_state(self.tablet, self.osk_visible)
|
|
|
|
def on_orientation(self, orient: str) -> None:
|
|
self.orientation = orient
|
|
if self.tablet:
|
|
self.apply_transform(transform_for(orient))
|
|
# laptop mode: never rotate (gated by the switch above)
|
|
|
|
# -------------------------------------------------------------
|
|
def run(self) -> int:
|
|
have_evdev = self.open_evdev()
|
|
|
|
# If we started mid-tablet, bring Hyprland to the current state right
|
|
# away instead of waiting for the next SW_TABLET_MODE edge. The OSK
|
|
# duty for the currently-focused window is applied below.
|
|
if self.tablet:
|
|
self.apply_transform(transform_for(self.orientation))
|
|
self.ensure_osk()
|
|
cls = self.cur_focus_class()
|
|
if cls:
|
|
self.on_focus_changed(cls)
|
|
write_state(self.tablet, self.osk_visible)
|
|
|
|
threads = []
|
|
if have_evdev:
|
|
t = threading.Thread(target=self.evdev_thread, daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
t = threading.Thread(target=self.sensor_thread, daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
t = threading.Thread(target=self.focus_thread, daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
t = threading.Thread(target=self.control_thread, daemon=True)
|
|
t.start()
|
|
threads.append(t)
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(3600)
|
|
except KeyboardInterrupt:
|
|
return 0
|
|
|
|
def cur_focus_class(self) -> str:
|
|
try:
|
|
out = subprocess.run(
|
|
[HYPRCTL, "activewindow", "-j"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3,
|
|
)
|
|
info = json.loads(out.stdout)
|
|
return info.get("class") or ""
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def main() -> int:
|
|
monitor = os.environ.get("HYPRLAND_TABLET_MONITOR", "eDP-1")
|
|
d = Daemon(monitor)
|
|
return d.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main()) |