mirror of
https://github.com/kovidgoyal/kitty.git
synced 2026-10-22 13:12:23 +00:00
Switch to headless sway to record demos
This commit is contained in:
Kovid Goyal
parent
f1cbe7dad5
commit
c018a69acb
1 changed files with 186 additions and 117 deletions
|
|
@ -1,74 +1,93 @@
|
|||
#!/usr/bin/env python
|
||||
# License: GPLv3 Copyright: 2026, Kovid Goyal <kovid at kovidgoyal.net>
|
||||
|
||||
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()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue