#!/usr/bin/env python3 """Record the documentation site's landing-page clips from a hidden window. zig build -Dplatform=gui -Doptimize=ReleaseSafe --prefix python3 -B test/record_demos.py /bin/pardes-gui [clip ...] Each clip drives a throwaway session (PARDES_TEST: an offscreen capture, the virtual clock) through its stdin's window-pixel mouse events, its keys and its 9P files, keeps the settled frame after every step, and lays the frames out on a timeline with a drawn pointer. ffmpeg encodes the timeline to docs/site/media/.webm and .mp4, with .png as its poster and captions.txt beside them. Needs a Wayland display for the hidden window, gcc and git for the demo project, and 9ns (cloud9) for the scripting clip. Frames and the intermediate video go under --work (default ../.scratch/rec/work, beside the checkout). The demo project lives at /tmp/hello, a short path for the tags, and the session's home at /tmp/pardes-rec-home; both are removed after. Nothing of the caller's environment that names a session, a mount or a bus reaches the recording. """ import argparse import contextlib import fcntl import json import os from pathlib import Path import shutil import signal import struct import subprocess import sys import termios import threading import time from PIL import Image, ImageDraw, ImageFont sys.path.insert(0, str(Path(__file__).resolve().parent)) from ninep import Client # noqa: E402 REPO = Path(__file__).resolve().parent.parent MEDIA = REPO / 'docs/site/media' BADGE_FONT = REPO / 'assets/AdwaitaMono-Regular.ttf' FPS = 30 COLS, ROWS = 60, 20 # A test window's text is 27 px (16 px cells); tags draw at 82% of it # (config.gui_tagline_font_percent), 13 px a character. TAG_W = 13 # acme's pale yellow and blue read at the size a landing page shows a clip, # and are the site's own colours. THEME = ['acme'] WIDTH = 960 # the clips' width: the window's own, never scaled PROJECT = Path('/tmp/hello') HOME = Path('/tmp/pardes-rec-home') # Nothing of the caller's session reaches the recording: no live pardes, # no mount, no bus, no editor. SCRUB = ('NINE_MOUNT', 'NAMESPACE', 'DBUS_SESSION_BUS_ADDRESS', 'GIT_EDITOR', 'EDITOR', 'VISUAL', 'PAGER', 'GIT_PAGER', 'MANPAGER', 'LESS', 'PROMPT_COMMAND', 'ENV', 'BASH_ENV', 'TMUX', 'ZMX_SESSION') def clean_env(): env = {k: v for k, v in os.environ.items() if not k.startswith('PARDES_') and k not in SCRUB} display = Path(os.environ.get('WAYLAND_DISPLAY', 'wayland-0')) if not display.is_absolute(): display = Path(os.environ.get('XDG_RUNTIME_DIR', '/run/user/%d' % os.getuid())) / display env.update(WAYLAND_DISPLAY=str(display), SDL_VIDEODRIVER='wayland', TERM='xterm-256color', LANG='C.UTF-8', LC_ALL='C.UTF-8', GIT_AUTHOR_DATE='2026-09-30T12:00:00', GIT_COMMITTER_DATE='2026-09-30T12:00:00') return env MAIN_C = '''#include static int square(int n) { return n * n; } int main(void) { int sum = 0; for (int i = 1; i <= 4; i++) sum += square(i); pritnf("sum: %d\\n", sum); return 0; } ''' README = '''hello: prints the sum of the first four squares. Build it with make, run it with ./hello. ''' MAKEFILE = '''CFLAGS = -Wall -fno-diagnostics-show-caret -fno-diagnostics-show-option hello: src/main.c \t@cc $(CFLAGS) -o hello src/main.c ''' def make_project(): shutil.rmtree(PROJECT, ignore_errors=True) (PROJECT / 'src').mkdir(parents=True) (PROJECT / 'src/main.c').write_text(MAIN_C) (PROJECT / 'Makefile').write_text(MAKEFILE) (PROJECT / 'README').write_text(README) class Session: """A hidden pardes-gui window, its capture directory, its 9P client.""" def __init__(self, binary, work, name, args, init='', cols=COLS, rows=ROWS): self.binary, self.work, self.name = Path(binary).resolve(), work, name self.cols, self.rows = cols, rows self.caps = work / 'cap' self.frames = work / 'frames' for d in (self.caps, self.frames): shutil.rmtree(d, ignore_errors=True) d.mkdir(parents=True) shutil.rmtree(HOME, ignore_errors=True) HOME.mkdir(mode=0o700) (HOME / 'config/pardes').mkdir(parents=True) theme = '' if THEME[0] == 'default' else f'Theme {THEME[0]}\n' (HOME / 'config/pardes/init').write_text('Shell /bin/sh\n' + theme + 'MessageLinger 0\n' + init) env = clean_env() env.update(HOME=str(HOME), XDG_RUNTIME_DIR=str(HOME), XDG_CONFIG_HOME=str(HOME / 'config'), PARDES_NOTIME='1', PARDES_DUMP=str(HOME / 'demo.dump.zon'), PARDES_TEST='1', PARDES_TEST_CLOCK='1', PARDES_TEST_COLS=str(cols), PARDES_TEST_ROWS=str(rows), PARDES_TEST_CAPTURE_DIR=str(self.caps)) for language in ['RS', 'C', 'GO', 'TS', 'PY', 'ZIG']: env['PARDES_LSP_' + language] = '' master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', rows, cols, 0, 0)) self.master = master self.child = subprocess.Popen( [str(self.binary), '--9p=' + name, *map(str, args)], cwd=PROJECT, env=env, stdin=slave, stdout=slave, stderr=open(work / 'stderr.log', 'wb'), start_new_session=True, preexec_fn=lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0)) os.close(slave) self.reader = threading.Thread(target=self._drain, daemon=True) self.reader.start() address = HOME / f'pardes-9p-{name}.sock' deadline = time.monotonic() + 30 while not address.exists(): if self.child.poll() is not None or time.monotonic() > deadline: raise RuntimeError('session did not start; see ' + str(work / 'stderr.log')) time.sleep(.01) self.client = Client(str(address)) self.client.__enter__() self.shots = [] # the timeline: see shot() self.pointer = None # the drawn pointer's (x, y), None for none self.button = None # the SDL button held, drawn as a ring and its name self.badge = None # a key or two, drawn in the corner self.settle() def _drain(self): try: while os.read(self.master, 65536): pass except OSError: pass def close(self): with contextlib.suppress(Exception): self.client.__exit__(None, None, None) if self.child.poll() is None: self.child.terminate() try: self.child.wait(timeout=10) except subprocess.TimeoutExpired: self.child.kill() self.child.wait() # The session's own process group: its shells and their children. with contextlib.suppress(ProcessLookupError, PermissionError): os.killpg(self.child.pid, signal.SIGKILL) with contextlib.suppress(OSError): os.close(self.master) self.reader.join(timeout=2) # 9P def read(self, path): return self.client.read(path).decode(errors='replace') def write(self, path, data): self.client.write(path, data.encode() if isinstance(data, str) else data) def panes(self): return {int(r.split()[0]): r.split()[1] for r in self.read('/index').splitlines() if r.strip()} def tag(self, serial): return self.read(f'/pane/{serial}/tag') def screen(self): s = json.loads(self.client.read('/screen')) cols = s['cols'] return [''.join((cell[0] or ' ') for cell in s['cells'][r * cols:(r + 1) * cols]) for r in range(s['rows'])] def find(self, text, after=0, nth=0): """The (col, row) of `text` on screen, searching rows from `after`.""" for r, line in enumerate(self.screen()): if r < after: continue at = line.find(text) while at >= 0: if nth == 0: return at, r nth -= 1 at = line.find(text, at + 1) raise LookupError(f'{text!r} not on screen:\n' + '\n'.join(self.screen())) def terminal_rows(self): """The screen's rows below the (one) terminal's tag.""" rows = self.screen() return rows[next(i for i, line in enumerate(rows) if 'Tty+sh' in line) + 1:] def wait(self, predicate, what, seconds=20): deadline = time.monotonic() + seconds while time.monotonic() < deadline: with contextlib.suppress(Exception): if predicate(): return time.sleep(.03) raise RuntimeError('timed out waiting for ' + what + '\n' + '\n'.join(self.screen())) def wait_text(self, text, seconds=20): self.wait(lambda: any(text in line for line in self.screen()), repr(text), seconds) # the window def cell_px(self, col, row): """The pixel at the middle of a cell (the window is an exact grid).""" w, h = self.size() cw, ch = w / self.cols, h / self.rows return int((col + .5) * cw), int((row + .55) * ch) def tag_px(self, col, row, start=3, offset=.5): """The pixel over a tag's character: a tag is drawn in the tagline font, narrower than the grid, from its first column `start` (3 in a pane's or column's tag, 0 in the workspace's).""" w, h = self.size() cw, ch = w / self.cols, h / self.rows return int(start * cw + (col - start + offset) * TAG_W), int((row + .55) * ch) def size(self): with open(self.caps / 'latest.ppm', 'rb') as f: f.readline() w, h = map(int, f.readline().split()) return w, h def mouse(self, kind, button, x, y): os.write(self.master, f'\x1b]777;mouse;{kind};{button};{x};{y}\x07'.encode()) time.sleep(.05) def keys(self, data): os.write(self.master, data.encode() if isinstance(data, str) else data) time.sleep(.03) def status(self): try: step, frames, idle = (self.caps / 'status').read_text().split()[:3] return int(step), int(frames), idle == '1' except (FileNotFoundError, ValueError): return None def settle(self): """The frame once the window says it is idle (test/gui_golden.py).""" deadline = time.monotonic() + 60 start = None while time.monotonic() < deadline: now = self.status() if now is not None: if start is None: start = now[0] elif now[0] >= start + 2 and now[2] and now[1] > 0: data = (self.caps / 'latest.ppm').read_bytes() if data.startswith(b'P6'): return data time.sleep(.02) raise RuntimeError('the window never went idle') # the timeline def shot(self, seconds, poster=False): """Hold the settled frame for `seconds`, the pointer where it is. `poster` marks the frame drawn as the clip's still.""" data = self.settle() path = self.frames / f'{len(self.shots):04d}.ppm' path.write_bytes(data) self.shots.append(dict(frame=path, secs=seconds, start=self.pointer, end=self.pointer, button=self.button, badge=self.badge, poster=poster)) def glide(self, x, y, seconds=.6): """Glide the drawn pointer to (x, y) over the frame on screen, then move the window's pointer there (dragging any button held).""" if self.shots and self.pointer is not None: last = self.shots[-1] self.shots.append(dict(last, secs=seconds, start=self.pointer, end=(x, y), poster=False)) self.pointer = (x, y) self.mouse('motion', self.button or 0, x, y) def sweep(self, button, start, end, hold=.5): """Press `button` (SDL: 1 left, 2 middle, 3 right) at `start`, drag it to `end` and release: the sweep acme's B2 and B3 take.""" if self.pointer != start: self.glide(*start) self.mouse('down', button, *start) self.button = button self.shot(.15) if end != start: self.glide(*end, seconds=.3) self.shot(hold) self.mouse('up', button, *self.pointer) self.button = None def press(self, keys, per_key=.4): """Send each (key bytes, badge) of `keys`, a frame a key with the badge in the corner.""" for key, badge in keys: self.badge = badge self.keys(key) self.shot(per_key) self.badge = None def type(self, text, expect, per_key=.07): """Type `text` a key at a time, a frame per key, once `expect(prefix)` says the window shows what is typed so far.""" for i, ch in enumerate(text): self.keys(ch) prefix = text[:i + 1] self.wait(lambda: expect(prefix), 'echo of ' + repr(prefix)) self.shot(per_key) # composing def pointer_sprite(scale): """An arrow pointer, white with a dark outline, its hotspot at (1, 1).""" pts = [(0, 0), (0, 17), (4, 13), (7, 20), (10, 19), (7, 12), (12, 12)] pts = [(x * scale + 1, y * scale + 1) for x, y in pts] im = Image.new('RGBA', (int(13 * scale) + 4, int(21 * scale) + 4), (0, 0, 0, 0)) ImageDraw.Draw(im).polygon(pts, fill=(255, 255, 255, 255), outline=(20, 20, 20, 255), width=2) return im # acme's colours for the buttons: B2 red, B3 green. BUTTONS = {1: ('click', (70, 70, 70)), 2: ('middle-click', (170, 0, 0)), 3: ('right-click', (0, 110, 40))} def draw_badge(img, text, xy, font, fill=(40, 40, 40), anchor='lt'): d = ImageDraw.Draw(img, 'RGBA') box = d.textbbox(xy, text, font=font, anchor=anchor) pad = 7 d.rounded_rectangle((box[0] - pad, box[1] - pad, box[2] + pad, box[3] + pad), radius=8, fill=fill + (235,)) d.text(xy, text, font=font, fill=(255, 255, 255), anchor=anchor) class Painter: def __init__(self, size): self.sprite = pointer_sprite(1.7) self.font = ImageFont.truetype(str(BADGE_FONT), 24) self.size = size def paint(self, img, pointer, button, badge): img = img.copy() w, h = self.size if badge: draw_badge(img, badge, (w - 16, h - 16), self.font, anchor='rb') if pointer is not None: x, y = pointer if button: label, colour = BUTTONS[button] d = ImageDraw.Draw(img, 'RGBA') r = 16 d.ellipse((x - r, y - r, x + r, y + r), outline=colour + (255,), width=4) # The name sits below the pointer, kept inside the frame. lx = min(x + 22, w - 16 - self.font.getlength(label)) draw_badge(img, label, (lx, y + 40), self.font, fill=colour) img.paste(self.sprite, (int(x) - 1, int(y) - 1), self.sprite) return img def ease(t): return t * t * (3 - 2 * t) def lerp(a, b, t): if a is None or b is None: return b if t >= .5 else a return (a[0] + (b[0] - a[0]) * t, a[1] + (b[1] - a[1]) * t) def compose(shots, out): """The shots at FPS with the pointer drawn, ending in a dissolve back to the first frame so the clip loops; written lossless (FFV1) to `out`. Answers the clip's length in seconds and its poster image.""" first = Image.open(shots[0]['frame']).convert('RGB') painter = Painter(first.size) w, h = first.size ff = subprocess.Popen(['ffmpeg', '-v', 'error', '-y', '-f', 'rawvideo', '-pix_fmt', 'rgb24', '-s', f'{w}x{h}', '-r', str(FPS), '-i', '-', '-c:v', 'ffv1', str(out)], stdin=subprocess.PIPE) poster = None clock, emitted = 0.0, 0 loaded = (None, None) for shot in shots: if loaded[0] != shot['frame']: loaded = (shot['frame'], Image.open(shot['frame']).convert('RGB')) img = loaded[1] clock += shot['secs'] n = round(clock * FPS) - emitted for i in range(n): t = ease((i + 1) / n) frame = painter.paint(img, lerp(shot['start'], shot['end'], t), shot['button'], shot['badge']) ff.stdin.write(frame.tobytes()) if shot['poster'] and poster is None: poster = painter.paint(img, shot['end'], shot['button'], None) emitted += n last = shots[-1] n = round(.7 * FPS) for i in range(n): t = ease((i + 1) / n) frame = Image.blend(loaded[1], first, t) ff.stdin.write(painter.paint(frame, lerp(last['end'], shots[0]['start'], t), None, None).tobytes()) emitted += n ff.stdin.close() if ff.wait() != 0: raise RuntimeError('ffmpeg (lossless) failed') return emitted / FPS, poster or painter.paint(first, None, None, None) def encode(lossless, name, poster, crf_webm=30, crf_mp4=20): """.webm (VP9) and .mp4 (H.264, yuv420p, faststart) and the poster PNG, in MEDIA.""" MEDIA.mkdir(parents=True, exist_ok=True) webm, mp4 = MEDIA / (name + '.webm'), MEDIA / (name + '.mp4') scale = [] if poster.width == WIDTH else ['-vf', f'scale={WIDTH}:-2:flags=lanczos'] subprocess.run(['ffmpeg', '-v', 'error', '-y', '-i', str(lossless), *scale, '-c:v', 'libvpx-vp9', '-b:v', '0', '-crf', str(crf_webm), '-row-mt', '1', '-deadline', 'good', '-cpu-used', '1', '-g', '240', '-pix_fmt', 'yuv420p', '-an', str(webm)], check=True) subprocess.run(['ffmpeg', '-v', 'error', '-y', '-i', str(lossless), *scale, '-c:v', 'libx264', '-preset', 'veryslow', '-crf', str(crf_mp4), '-tune', 'animation', '-pix_fmt', 'yuv420p', '-movflags', '+faststart', '-an', str(mp4)], check=True) if poster.width != WIDTH: poster = poster.resize((WIDTH, round(poster.height * WIDTH / poster.width)), Image.LANCZOS) poster.save(MEDIA / (name + '.png'), optimize=True) return [webm, mp4, MEDIA / (name + '.png')] # the clips def text_is_the_interface(s): """B2 on `make` in a tag runs it; B3 on the error opens the file there.""" s.write('/pane/1/tag', ' make') s.pointer = s.cell_px(36, 9) s.shot(1.0) col, row = s.find('make', 0) s.sweep(2, s.tag_px(col, row, offset=.15), s.tag_px(col + 3, row, offset=.85)) s.wait(lambda: any('exit' in s.tag(n) for n, kind in s.panes().items() if kind == 'cmd'), 'make to finish') s.shot(1.1) col, row = s.find('src/main.c:12:5', 0) s.sweep(3, s.cell_px(col, row), s.cell_px(col + 14, row)) s.wait(lambda: len(s.panes()) == 3, 'main.c to open') s.glide(*s.cell_px(40, 9), seconds=.5) s.shot(2.6, poster=True) def shell_setup(s): """A terminal under the file, at the project, with a short prompt.""" s.write('/pane/1/exec', 'Tty\n') s.wait(lambda: 2 in s.panes(), 'the terminal') s.write('/pane/2/pty/data', f"cd {PROJECT}; PS1='$ '; clear\r") s.wait(lambda: any(line.rstrip() == ' $' for line in s.screen()), 'the prompt') # The pointer rests off the tags, so no word's hint is up. s.mouse('motion', 0, *s.cell_px(COLS - 2, ROWS - 1)) def mounted_shell_setup(s): """A terminal under the file whose shell sees the session as files: cloud9's 9ns mounts the pane's $PARDES_9P in a private namespace.""" shell_setup(s) mount = HOME / 'n' mount.mkdir() s.write('/pane/2/pty/data', f'exec 9ns --unix "$PARDES_9P" --mount {mount} -- /bin/sh\r') time.sleep(1.5) # 9ns mounts and starts the shell that reads the next line s.write('/pane/2/pty/data', "cd \"$NINE_MOUNT\" && PS1='$ ' && clear\r") s.wait(lambda: [line.rstrip() for line in s.terminal_rows() if line.strip()] == [' $'], 'the mounted shell') def scripting_over_9p(s): """A shell reads the session's files and writes a pane's body.""" s.shot(1.0) for command, done in [('ls', lambda: any('index' in line for line in s.terminal_rows()[1:])), ('cat index', lambda: any(' term ' in line for line in s.terminal_rows())), ("echo 'hello from a script' > pane/1/body", lambda: 'hello from' in s.screen()[3])]: s.type(command, lambda typed: at_prompt(s, typed)(), per_key=.06) s.keys('\r') s.wait(done, command) s.wait(lambda: [line.rstrip() for line in s.terminal_rows() if line.strip()][-1] == ' $', 'the prompt') s.shot(1.3) s.shots[-1]['secs'] = 2.4 s.shots[-1]['poster'] = True def at_prompt(s, text): return lambda: any(line.startswith(' $ ' + text) for line in s.screen()) def terminals_are_panes(s): """A shell under the file; Ctrl-b and its output is text to move over.""" s.write('/pane/1/dot', '12\n') s.shot(1.0) s.type('make', lambda typed: at_prompt(s, typed)()) s.keys('\r') s.wait(lambda: 'Error 1' in '\n'.join(s.screen()), 'make') s.wait(lambda: any(line.rstrip() == ' $' for line in s.screen()), 'the next prompt') s.shot(1.4) s.press([('\x02', 'Ctrl-b')], per_key=.8) s.press([('k', 'k'), ('k', 'k k'), ('k', 'k k k')], per_key=.4) s.press([('x', 'x')], per_key=2.4) s.shots[-1]['poster'] = True def diff_setup(s): """The project committed, then its typo fixed; a lone terminal.""" env = dict(clean_env(), HOME=str(HOME), GIT_AUTHOR_NAME='demo', GIT_AUTHOR_EMAIL='demo@example.com', GIT_COMMITTER_NAME='demo', GIT_COMMITTER_EMAIL='demo@example.com') for args in (['init', '-q'], ['add', '.'], ['commit', '-qm', 'hello'], ['config', 'diff.context', '1']): subprocess.run(['git', '-C', str(PROJECT), *args], check=True, env=env, capture_output=True) main = PROJECT / 'src/main.c' main.write_text(main.read_text().replace('pritnf', 'printf')) shell_setup(s) s.write('/pane/1/exec', 'Del\n') s.wait(lambda: list(s.panes()) == [2], 'the README to close') def diff_review(s): """git diff in a shell pages into +Pager as a diff; B3 on its + line opens the file at that line.""" s.pointer = s.cell_px(40, 6) s.shot(.9) s.type('git diff', lambda typed: at_prompt(s, typed)()) s.keys('\r') s.wait(lambda: 'pager' in s.panes().values() or len(s.panes()) == 2, 'the pager') s.wait_text('+ printf') s.shot(1.4) col, row = s.find('+ printf', 0) s.sweep(3, s.cell_px(col, row), s.cell_px(col, row)) s.wait(lambda: len(s.panes()) == 3, 'main.c to open') s.glide(*s.cell_px(42, 9), seconds=.5) s.shot(2.6, poster=True) CLIPS = { 'text-is-the-interface': dict( run=text_is_the_interface, args=[PROJECT / 'README'], caption='Middle-click runs a word; right-click an error opens the file.'), 'terminals-are-panes': dict( setup=shell_setup, run=terminals_are_panes, args=[PROJECT / 'src/main.c'], caption='A shell is a pane: Ctrl-b, then move over its output.'), 'scripting-over-9p': dict( setup=mounted_shell_setup, run=scripting_over_9p, args=[PROJECT / 'README'], caption='The session is files: a shell script reads and writes its panes.'), 'diff-review': dict( setup=diff_setup, run=diff_review, args=[PROJECT / 'README'], caption='git diff pages into a pane; right-click a change to open it.'), } def record(binary, work, name): clip = CLIPS[name] make_project() s = Session(binary, work / name, 'rec-' + name, clip['args'], init=clip.get('init', '')) try: if 'setup' in clip: clip['setup'](s) clip['run'](s) finally: s.close() seconds, poster = compose(s.shots, work / name / 'lossless.mkv') files = encode(work / name / 'lossless.mkv', name, poster) for f in files: print(f'{f.name} {f.stat().st_size / 1e6:.2f} MB') print(f'{name}: {seconds:.1f} s') return seconds def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('binary', type=Path, help='a pardes-gui built with -Dplatform=gui') parser.add_argument('clips', nargs='*', help='which clips (default: all): ' + ', '.join(CLIPS)) parser.add_argument('--work', type=Path, default=REPO.parent / '.scratch/rec/work', help='frames and intermediates (large; default %(default)s)') parser.add_argument('--theme', default=THEME[0], help='a theme name, or default (default %(default)s)') args = parser.parse_args() THEME[0] = args.theme names = args.clips or list(CLIPS) for name in names: if name not in CLIPS: parser.error(f'no clip {name!r}: ' + ', '.join(CLIPS)) try: for name in names: record(args.binary, args.work.resolve(), name) finally: shutil.rmtree(PROJECT, ignore_errors=True) shutil.rmtree(HOME, ignore_errors=True) captions = MEDIA / 'captions.txt' lines = dict(line.split(': ', 1) for line in captions.read_text().splitlines() if ': ' in line) \ if captions.exists() else {} for name in names: lines[name] = CLIPS[name]['caption'] captions.write_text(''.join(f'{name}: {lines[name]}\n' for name in CLIPS if name in lines)) if __name__ == '__main__': main()