From c018a69acb3590034af152ee9b44f574638335db Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Sun, 9 Aug 2026 10:51:16 +0530 Subject: [PATCH] Switch to headless sway to record demos --- kitty/shaders/custom/demo.py | 303 +++++++++++++++++++++-------------- 1 file changed, 186 insertions(+), 117 deletions(-) diff --git a/kitty/shaders/custom/demo.py b/kitty/shaders/custom/demo.py index 3cc5eb8f4..1bbd713aa 100644 --- a/kitty/shaders/custom/demo.py +++ b/kitty/shaders/custom/demo.py @@ -1,74 +1,93 @@ #!/usr/bin/env python # License: GPLv3 Copyright: 2026, Kovid Goyal -import json +import glob import math import os import signal import subprocess import sys +import tempfile import time from contextlib import suppress from threading import Thread -from typing import Any, Literal +from typing import Any DEFAULT_DURATION = 3 DEFAULT_INITIAL_SLEEP = 3 +SCREEN_WIDTH = 600 +SCREEN_HEIGHT = 400 + +SWAY_CONFIG = f"""\ +output HEADLESS-1 resolution {SCREEN_WIDTH}x{SCREEN_HEIGHT} scale 1 +default_border none +default_floating_border none +hide_edge_borders both +gaps inner 0 +gaps outer 0 +bar {{ + mode invisible +}} +""" + +# Module-level state, set by do_one() before calling animate functions +_sway_socket: str = '' +_wayland_display: str = '' +_kitty_socket: str = '' -def move_mouse(geomerty: tuple[int, int, int, int], x: int, y: int) -> None: - x += geomerty[0] - y += geomerty[1] - subprocess.run(['hyprctl', 'dispatch', f'hl.dsp.cursor.move({{ x = {x}, y = {y} }})'], check=True, stdout=subprocess.DEVNULL) +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 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)) def click_mouse() -> None: - subprocess.run(['hyprctl', 'dispatch', 'hl.dsp.send_key_state({ key = "mouse:272", state = "down", mods = "" })'], check=True, stdout=subprocess.DEVNULL) + sway_msg('seat', 'seat0', 'cursor', 'press', 'button1') time.sleep(0.05) - subprocess.run(['hyprctl', 'dispatch', 'hl.dsp.send_key_state({ key = "mouse:272", state = "up", mods = "" })'], check=True, stdout=subprocess.DEVNULL) + sway_msg('seat', 'seat0', 'cursor', 'release', 'button1') -def key_event(key: str, mods: str = '', state: Literal['up', 'down'] = 'down') -> None: - subprocess.run( - ['hyprctl', 'dispatch', f'hl.dsp.send_key_state({{ key = "{key}", state = "{state}", mods = "{mods}" }})'], check=True, stdout=subprocess.DEVNULL - ) +def remote_control(*args: str) -> None: + subprocess.run(['kitten', '@', '--use-password=never', '--to', f'unix:{_kitty_socket}'] + list(args), check=True, stdout=subprocess.DEVNULL) -def press_and_release(key: str, mods: str = '') -> None: - key_event(key, mods) - key_event(key, mods, 'up') +def key_event(key: str) -> None: + remote_control('send-key', key.lower()) -def cursor_trail(window_id: str, geomerty: tuple[int, int, int, int]) -> None: - press_and_release('g') - press_and_release('g') - press_and_release('1') - press_and_release('0') - press_and_release('j') +def send_text(text: str) -> None: + remote_control('send-text', text) + + +def cursor_trail(window_id: str, geometry: tuple[int, int, int, int]) -> None: + send_text('gg10j') + time.sleep(0.3) + send_text('^') + time.sleep(0.4) + send_text('$') + time.sleep(0.3) + send_text(':hello world') + time.sleep(0.5) + key_event('Escape') time.sleep(0.2) - press_and_release('6', 'SHIFT') - time.sleep(0.2) - press_and_release('4', 'SHIFT') + send_text('$') + + +def pond_ripple(window_id: str, geometry: tuple[int, int, int, int]) -> None: time.sleep(0.1) - press_and_release('6', 'SHIFT') - time.sleep(0.1) - press_and_release('semicolon', 'SHIFT') - for k in 'hello world': - if k == ' ': - k = 'space' - press_and_release(k) - time.sleep(0.2) - press_and_release('Escape') - time.sleep(0.2) - press_and_release('4', 'SHIFT') - - -def pond_ripple(window_id: str, geomerty: tuple[int, int, int, int]) -> None: - time.sleep(0.1) - move_mouse(geomerty, 100, 100) + move_mouse(geometry, 100, 100) click_mouse() time.sleep(1) - move_mouse(geomerty, 200, 200) + move_mouse(geometry, 200, 200) click_mouse() @@ -99,73 +118,91 @@ metadata: dict[str, dict[str, Any]] = { } -def get_hyprland_window_id(pid: int) -> str: - try: - result = subprocess.run(['hyprctl', 'clients', '-j'], capture_output=True, text=True, check=True) - clients = json.loads(result.stdout) - for client in clients: - if client.get('pid') == pid: - return client.get('address') - except (subprocess.CalledProcessError, json.JSONDecodeError): - pass - return '' +def wait_for_file(path: str, timeout: float = 10.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if os.path.exists(path): + return True + time.sleep(0.1) + return False -def get_window_geometry(window_id: str) -> tuple[int, int, int, int]: - if not window_id.startswith('0x'): - window_id = f'0x{window_id}' - result = subprocess.run(['hyprctl', 'clients', '-j'], capture_output=True, text=True, check=True) - clients = json.loads(result.stdout) - for client in clients: - if client.get('address') == window_id: - x, y = client.get('at', [0, 0]) - width, height = client.get('size', [0, 0]) - return (x, y, width, height) - raise KeyError(f'No window with id: {window_id} found') +def start_sway(config_file: str) -> tuple['subprocess.Popen[bytes]', str, str]: + """Start a headless sway compositor. Returns (proc, wayland_display_name, ipc_socket_path).""" + runtime_dir = get_runtime_dir() + # Snapshot existing sockets before starting so we can identify the new one + existing = set(glob.glob(f'{runtime_dir}/wayland-[0-9]*')) + env: dict[str, str] = {k: v for k, v in os.environ.items()} + env['WLR_BACKENDS'] = 'headless' + # Remove parent display to avoid nested-compositor confusion + env.pop('WAYLAND_DISPLAY', None) + env.pop('DISPLAY', None) + proc = subprocess.Popen( + ['sway', '--config', config_file], + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + text=False, + ) + # Wait for sway to create its Wayland socket + deadline = time.monotonic() + 10 + wayland_display = '' + while time.monotonic() < deadline: + current = set(glob.glob(f'{runtime_dir}/wayland-[0-9]*')) + for path in sorted(current - existing): + if not path.endswith('.lock'): + wayland_display = os.path.basename(path) + break + if wayland_display: + break + time.sleep(0.1) + if not wayland_display: + proc.kill() + proc.wait() + raise SystemExit('Timed out waiting for headless sway Wayland socket') + # IPC socket path: $XDG_RUNTIME_DIR/sway-ipc.$UID.$PID.sock + ipc_socket = os.path.join(runtime_dir, f'sway-ipc.{os.getuid()}.{proc.pid}.sock') + if not wait_for_file(ipc_socket, timeout=10): + proc.kill() + proc.wait() + raise SystemExit(f'Timed out waiting for sway IPC socket: {ipc_socket}') + return proc, wayland_display, ipc_socket -def record_window_geometry(geometry: tuple[int, int, int, int], m: dict[str, Any], output_filename: str) -> str: - x, y, width, height = geometry - geometry_string = f'{x},{y} {width}x{height}' +def record_output(m: dict[str, Any], output_filename: str, wayland_env: dict[str, str]) -> str: duration_seconds = m.get('duration', DEFAULT_DURATION) output_filename += '.mkv' with suppress(FileNotFoundError): os.remove(output_filename) - cmd = ['wf-recorder', '-g', geometry_string, '-f', output_filename, '-y'] + cmd = ['wf-recorder', '-o', 'HEADLESS-1', '-f', output_filename, '-y'] # lossless recording cmd += '-c libx264rgb -p crf=0'.split() + env = {**os.environ, **wayland_env} try: - print(f'🎬 Starting recording of {geometry_string} for {duration_seconds} seconds...') - process = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + print(f'🎬 Starting recording for {duration_seconds} seconds...') + process = subprocess.Popen(cmd, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) time.sleep(duration_seconds) print('🛑 Stopping recording and saving file...') process.send_signal(signal.SIGINT) - process.wait(timeout=5) + process.wait(timeout=3) return output_filename - except subprocess.TimeoutExpired: - process.kill() - raise SystemExit('⚠️ wf-recorder hung and had to be forcefully killed.') except FileNotFoundError: raise SystemExit("❌ Error: 'wf-recorder' is not installed or not in your PATH.") except Exception as e: raise SystemExit(f'❌ An error occurred: {e}') -def drive_loop(kitty: subprocess.Popen, which: str, m: dict[str, Any], destdir: str) -> None: - winid = '' - st = time.monotonic() - while not winid and time.monotonic() - st < 2: - winid = get_hyprland_window_id(kitty.pid) - if not winid: - raise SystemExit('Could not get Hyprland window id for kitty process') - geom = get_window_geometry(winid) +def drive_loop(kitty: 'subprocess.Popen[bytes]', which: str, m: dict[str, Any], destdir: str, wayland_env: dict[str, str]) -> None: + st = m.get('initial_sleep', DEFAULT_INITIAL_SLEEP) + print(f'Waiting for {st} seconds before recording...') + time.sleep(st) + geometry = (0, 0, SCREEN_WIDTH, SCREEN_HEIGHT) output_file = os.path.join(destdir, which) animate = m.get('animate') - time.sleep(m.get('initial_sleep', DEFAULT_INITIAL_SLEEP)) if animate is not None: - t = Thread(target=animate, args=(winid, geom), daemon=True) + t = Thread(target=animate, args=('', geometry), daemon=True) t.start() - lossless_file = record_window_geometry(geom, m, output_file) + lossless_file = record_output(m, output_file, wayland_env) if animate is not None: t.join() kitty.terminate() @@ -173,18 +210,82 @@ def drive_loop(kitty: subprocess.Popen, which: str, m: dict[str, Any], destdir: print('Transcoding saved recording...') try: for fmt, cmd in { - 'webm': '-c:v libsvtav1 -preset 3 -crf 55 -pix_fmt yuv420p -svtav1-params tune=0:scm=1:enable-intrabc=1:scd=0:enable-hdr=0:aq-mode=0 -g 99999' + 'webm': '-r 60 -c:v libsvtav1 -preset 3 -crf 55 -pix_fmt yuv420p -svtav1-params tune=0:scm=1:enable-intrabc=1:scd=0:aq-mode=0 -g 99999' }.items(): of = f'{output_file}.{fmt}' with suppress(FileNotFoundError): os.remove(of) - subprocess.run(['ffmpeg', '-i', lossless_file] + cmd.split() + [of], check=True, capture_output=True) + # subprocess.run(['mpv', lossless_file]) + cp = subprocess.run(['ffmpeg', '-i', lossless_file] + cmd.split() + [of], capture_output=True) + if cp.returncode != 0: + sys.stderr.buffer.write(cp.stderr) + raise SystemExit(cp.returncode) print(f'Recording saved: {of}') subprocess.run(['ls', '-sh', of]) finally: os.remove(lossless_file) +def do_one(which: str) -> None: + global _sway_socket, _wayland_display, _kitty_socket + print('Generating demo video for custon shader:', which) + self = os.path.abspath(__file__) + m = metadata.get(which) or {} + cmd = m.get('cmd', ['nvim', '-R', '+1', self]) + d = os.path.dirname + destdir = os.path.join(d(d(d(d(self)))), 'docs', 'screenshots') + os.environ.pop('KITTY_PUBLIC_KEY', None) + + with tempfile.NamedTemporaryFile(mode='w', suffix='.conf', delete=False, prefix='sway-demo-') as f: + f.write(SWAY_CONFIG) + config_file = f.name + + kitty_socket_path = f'@kitty-custom-shader-demo-{os.getpid()}' + + try: + 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} + try: + kcmd = [ + 'kitty', + '--start-as=fullscreen', + '--config=NONE', + '-o', + 'font_size=8', + '-o', + 'cursor_trail=1', + '-o', + f'custom_shaders={which}', + '-o', + 'allow_remote_control=socket-only', + f'--listen-on=unix:{kitty_socket_path}', + ] + kitty_env: dict[str, str] = {k: v for k, v in os.environ.items()} + kitty_env.update(wayland_env) + kitty = subprocess.Popen(kcmd + cmd, env=kitty_env, text=False) + try: + drive_loop(kitty, which, m, destdir, wayland_env) + finally: + if kitty.returncode is None: + kitty.terminate() + kitty.wait() + finally: + sway_proc.terminate() + with suppress(subprocess.TimeoutExpired): + sway_proc.wait(timeout=5) + if sway_proc.returncode is None: + sway_proc.kill() + sway_proc.wait() + finally: + with suppress(FileNotFoundError): + os.remove(config_file) + with suppress(FileNotFoundError): + os.remove(kitty_socket_path) + + def main() -> None: if len(sys.argv) < 2: shaders = list(metadata) @@ -197,37 +298,5 @@ def main() -> None: raise SystemExit('Aborting on user interrupt') -def do_one(which: str) -> None: - self = os.path.abspath(__file__) - m = metadata.get(which) or {} - cmd = m.get('cmd', ['nvim', '-R', '+1', self]) - kcmd = [ - 'kitty', - '--class=float', - '--config=NONE', - '-o', - 'remember_window_size=no', - '-o', - 'font_size=8', - '-o', - 'initial_window_width=70c', - '-o', - 'initial_window_height=20c', - '-o', - 'cursor_trail=1', - '-o', - f'custom_shaders={which}', - ] - kitty = subprocess.Popen(kcmd + cmd) - d = os.path.dirname - destdir = os.path.join(d(d(d(d(self)))), 'docs', 'screenshots') - try: - drive_loop(kitty, which, m, destdir) - finally: - if kitty.returncode is None: - kitty.terminate() - kitty.wait() - - if __name__ == '__main__': main()