diff --git a/kitty/shaders/custom/demo.py b/kitty/shaders/custom/demo.py
index 39d8787c5..41498cc53 100644
--- a/kitty/shaders/custom/demo.py
+++ b/kitty/shaders/custom/demo.py
@@ -4,6 +4,7 @@
import glob
import math
import os
+import shutil
import signal
import subprocess
import sys
@@ -34,26 +35,200 @@ bar {{
_sway_socket: str = ''
_wayland_display: str = ''
_kitty_socket: str = ''
+_virt_mouse: 'subprocess.Popen[bytes] | None' = None
+
+# Protocol XML for zwlr_virtual_pointer_manager_v1 (wlr-protocols)
+_VIRTUAL_POINTER_XML = """\
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+# Daemon that holds a virtual pointer Wayland connection and processes commands on stdin.
+# Commands: "move X Y EX EY\\n", "press BTN\\n", "release BTN\\n"
+_VIRTUAL_POINTER_C = r"""
+#include
+#include
+#include
+#include
+#include
+#include "virt-pointer-client.h"
+
+static struct zwlr_virtual_pointer_manager_v1 *g_manager;
+static struct zwlr_virtual_pointer_v1 *g_pointer;
+
+static void on_global(void *data, struct wl_registry *registry, uint32_t name,
+ const char *interface, uint32_t version)
+{
+ if (!strcmp(interface, "zwlr_virtual_pointer_manager_v1"))
+ g_manager = wl_registry_bind(registry, name,
+ &zwlr_virtual_pointer_manager_v1_interface,
+ version > 1 ? 1 : version);
+}
+static void on_global_remove(void *d, struct wl_registry *r, uint32_t n) { (void)d; (void)r; (void)n; }
+static const struct wl_registry_listener reg_listener = { on_global, on_global_remove };
+
+static uint32_t ms_now(void) {
+ struct timespec ts;
+ clock_gettime(CLOCK_MONOTONIC, &ts);
+ return (uint32_t)(ts.tv_sec * 1000 + ts.tv_nsec / 1000000);
+}
+
+int main(void) {
+ struct wl_display *dpy = wl_display_connect(NULL);
+ if (!dpy) { fputs("Cannot connect to Wayland display\n", stderr); return 1; }
+ struct wl_registry *reg = wl_display_get_registry(dpy);
+ wl_registry_add_listener(reg, ®_listener, NULL);
+ wl_display_roundtrip(dpy);
+ if (!g_manager) {
+ fputs("zwlr_virtual_pointer_manager_v1 not supported by this compositor\n", stderr);
+ return 1;
+ }
+ g_pointer = zwlr_virtual_pointer_manager_v1_create_virtual_pointer(g_manager, NULL);
+ wl_display_roundtrip(dpy);
+
+ char line[128];
+ while (fgets(line, sizeof(line), stdin)) {
+ uint32_t t = ms_now();
+ uint32_t a, b, c, d;
+ if (sscanf(line, "move %u %u %u %u", &a, &b, &c, &d) == 4) {
+ zwlr_virtual_pointer_v1_motion_absolute(g_pointer, t, a, b, c, d);
+ zwlr_virtual_pointer_v1_frame(g_pointer);
+ } else if (sscanf(line, "press %u", &a) == 1) {
+ zwlr_virtual_pointer_v1_button(g_pointer, t, a, 1);
+ zwlr_virtual_pointer_v1_frame(g_pointer);
+ } else if (sscanf(line, "release %u", &a) == 1) {
+ zwlr_virtual_pointer_v1_button(g_pointer, t, a, 0);
+ zwlr_virtual_pointer_v1_frame(g_pointer);
+ }
+ wl_display_flush(dpy);
+ }
+ zwlr_virtual_pointer_v1_destroy(g_pointer);
+ zwlr_virtual_pointer_manager_v1_destroy(g_manager);
+ wl_display_disconnect(dpy);
+ return 0;
+}
+"""
def get_runtime_dir() -> str:
return os.environ.get('XDG_RUNTIME_DIR', f'/run/user/{os.getuid()}')
-def sway_msg(*args: str) -> None:
- subprocess.run(['swaymsg', '-s', _sway_socket] + list(args), check=True, stdout=subprocess.DEVNULL)
+def build_virt_mouse(build_dir: str) -> str:
+ xml_path = os.path.join(build_dir, 'virt-pointer.xml')
+ header_path = os.path.join(build_dir, 'virt-pointer-client.h')
+ proto_c_path = os.path.join(build_dir, 'virt-pointer-client.c')
+ main_c_path = os.path.join(build_dir, 'virt-pointer-main.c')
+ binary_path = os.path.join(build_dir, 'virt-pointer')
+ with open(xml_path, 'w') as f:
+ f.write(_VIRTUAL_POINTER_XML)
+ with open(main_c_path, 'w') as f:
+ f.write(_VIRTUAL_POINTER_C)
+ subprocess.run(['wayland-scanner', 'client-header', xml_path, header_path], check=True, capture_output=True)
+ subprocess.run(['wayland-scanner', 'private-code', xml_path, proto_c_path], check=True, capture_output=True)
+ result = subprocess.run(
+ ['gcc', '-O2', '-I', build_dir, proto_c_path, main_c_path, '-lwayland-client', '-o', binary_path],
+ capture_output=True,
+ text=True,
+ )
+ if result.returncode != 0:
+ raise SystemExit(f'Failed to compile virtual pointer tool:\n{result.stderr}')
+ return binary_path
+
+
+def start_virt_mouse(binary: str, wayland_display: str) -> None:
+ global _virt_mouse
+ env = {**os.environ, 'WAYLAND_DISPLAY': wayland_display}
+ _virt_mouse = subprocess.Popen([binary], stdin=subprocess.PIPE, env=env, text=False)
+ time.sleep(0.2) # allow the virtual pointer to register with the compositor
+
+
+def stop_virt_mouse() -> None:
+ global _virt_mouse
+ if _virt_mouse is not None:
+ with suppress(Exception):
+ assert _virt_mouse.stdin is not None
+ _virt_mouse.stdin.close()
+ with suppress(subprocess.TimeoutExpired):
+ _virt_mouse.wait(timeout=2)
+ if _virt_mouse.returncode is None:
+ _virt_mouse.kill()
+ _virt_mouse.wait()
+ _virt_mouse = None
+
+
+def send_virt_mouse(cmd: str) -> None:
+ if _virt_mouse is not None and _virt_mouse.stdin is not None:
+ _virt_mouse.stdin.write(cmd.encode())
+ _virt_mouse.stdin.flush()
def move_mouse(geometry: tuple[int, int, int, int], x: int, y: int) -> None:
abs_x = geometry[0] + x
abs_y = geometry[1] + y
- sway_msg('seat', 'seat0', 'cursor', 'set', str(abs_x), str(abs_y))
+ send_virt_mouse(f'move {abs_x} {abs_y} {SCREEN_WIDTH} {SCREEN_HEIGHT}\n')
def click_mouse() -> None:
- sway_msg('seat', 'seat0', 'cursor', 'press', 'button1')
+ send_virt_mouse('press 272\n') # BTN_LEFT = 0x110 = 272
time.sleep(0.05)
- sway_msg('seat', 'seat0', 'cursor', 'release', 'button1')
+ send_virt_mouse('release 272\n')
def remote_control(*args: str) -> None:
@@ -240,14 +415,19 @@ def do_one(which: str) -> None:
f.write(SWAY_CONFIG)
config_file = f.name
+ build_dir = tempfile.mkdtemp(prefix='kitty-demo-build-')
kitty_socket_path = f'@kitty-custom-shader-demo-{os.getpid()}'
try:
+ virt_mouse_binary = build_virt_mouse(build_dir)
sway_proc, wayland_display, ipc_socket = start_sway(config_file)
_sway_socket = ipc_socket
_wayland_display = wayland_display
_kitty_socket = kitty_socket_path
wayland_env = {'WAYLAND_DISPLAY': wayland_display}
+ # Start virtual pointer daemon before kitty so the seat has pointer
+ # capability when kitty connects and creates its wl_pointer object.
+ start_virt_mouse(virt_mouse_binary, wayland_display)
try:
kcmd = [
'kitty',
@@ -273,6 +453,7 @@ def do_one(which: str) -> None:
kitty.terminate()
kitty.wait()
finally:
+ stop_virt_mouse()
sway_proc.terminate()
with suppress(subprocess.TimeoutExpired):
sway_proc.wait(timeout=5)
@@ -284,6 +465,7 @@ def do_one(which: str) -> None:
os.remove(config_file)
with suppress(FileNotFoundError):
os.remove(kitty_socket_path)
+ shutil.rmtree(build_dir, ignore_errors=True)
def main() -> None: