path: root/test/record_casts.py

#!/usr/bin/env python3
"""Record the documentation site's terminal casts (asciicast v2).

zig build -Dplatform=tty -Doptimize=ReleaseSafe --prefix <scratch dir>
python3 -B test/record_casts.py <scratch dir>/bin/pardes [cast ...] [--show]

Each cast runs the tty pardes in a pty this script owns, sized 100x30 with
TIOCSWINSZ, and plays the part of a plain xterm: it answers the queries an
xterm answers (device attributes, device status, cursor position,
XTVERSION, DECRQM as "not recognised", window sizes) and nothing else, so
kitty graphics and the kitty keyboard stay unanswered: images are drawn
as PETSCII glyph art, and a PDF as its text layer. Keys and SGR mouse
reports are scripted.

A cast is cut into steps: after each scripted input the recorder waits
until the screen shows what the step waits for and pardes has drawn and
gone idle, then stamps the step at its time on a virtual clock (starting
at 0, a scripted hold between steps, no gap over 2 s). pyte follows the
screen pardes draws, and an event is the cells the step changed, drawn
afresh: how many frames pardes took to get there (a shell's output racing
them) never reaches the file, so the same input makes the same cast.
Wall-clock time never reaches it either. The bytes pardes wrote, stamped
the same way, go to <work>/<name>.raw.cast beside it.

--show replays each finished cast through pyte, prints its last frame
(or the one --at seconds in) and draws it to <work>/<name>.png.

Nothing of the caller's environment reaches the recording: the session
gets a home, config, runtime and cache dirs under --work, a PATH with the
recorded pardes first, and no PARDES_*, 9P mount, display or bus. The demo
project and pictures live under /tmp/cast (short paths for the tags, a
few hundred KB) and are removed after. The PDF (--pdf) is only read,
through a symlink.

Needs pyte and Pillow (in a venv), cc, make and git.
"""
import argparse
import codecs
import contextlib
import fcntl
import json
import math
import os
from pathlib import Path
import re
import shutil
import signal
import struct
import subprocess
import termios
import threading
import time

import pyte

REPO = Path(__file__).resolve().parent.parent
OUT = REPO / 'docs/site/casts'
COLS, ROWS = 100, 30
CELL_W, CELL_H = 10, 20  # the pixels a cell claims to be, for TIOCSWINSZ and CSI 14/16 t
MAX_GAP = 2.0
PROJECT = Path('/tmp/cast/hello')

# ---------------------------------------------------------------------------
# the terminal: queries answered as a plain xterm answers them

DA1 = b'\x1b[?64;1;2;6;9;15;16;17;18;21;22;28c'
XTVERSION = b'\x1bP>|XTerm(390)\x1b\\'

# Everything pardes asks its terminal, answered in _answer as an xterm
# would; what xterm ignores (kitty's queries, OSC 66) goes unanswered.
# None of it reaches the cast.
QUERY = re.compile(rb'''
    \x1b\[\?(?P<decrqm>\d+)\$p            # DECRQM: answered "not recognised"
  | \x1b\[(?P<cpr>6)n                       # cursor position
  | \x1b\[(?P<dsr>5)n                       # device status: each frame ends asking it
  | \x1b\[(?P<da1>0?)c                      # primary device attributes
  | \x1b\[>(?P<xtv>0?)q                     # XTVERSION
  | \x1b\[(?P<winop>1[468])t                # window/cell size in pixels, size in cells
  | \x1b\[>\x20q                            # kitty multi-cursor query
  | \x1b\[\?u                               # kitty keyboard query
  | \x1b_G[^\x1b]*\x1b\\                    # kitty graphics
  | \x1b\]66;[^\x07\x1b]*(?:\x07|\x1b\\)    # explicit width / scaled text
  | \x1b\[\?2048h                           # in-band resize reports
''', re.X)

SGR = re.compile(rb'\x1b\[([0-9:;]*)m')


def sgr_semicolons(m):
    """Colon-form SGR (38:2:r:g:b, 4:3) in the semicolon form xterm, pyte
    and asciinema's player all read; an underline colour (58) is dropped."""
    params = m.group(1)
    if b':' not in params:
        return m.group(0)
    out = []
    for p in params.split(b';'):
        if b':' not in p:
            out.append(p)
            continue
        parts = p.split(b':')
        if parts[0] in (b'38', b'48'):
            if parts[1] == b'2' and len(parts) == 6:  # 38:2:<colorspace>:r:g:b
                parts = parts[:2] + parts[3:]
            out.append(b';'.join(parts))
        elif parts[0] == b'4':
            out.append(b'4' if parts[1] != b'0' else b'24')
        # 58:... (underline colour) has no semicolon form a player reads
    return b'\x1b[' + b';'.join(out) + b'm'


def incomplete_tail(buf):
    """Where an escape sequence the buffer ends inside of starts, or None."""
    i = buf.rfind(b'\x1b')
    if i < 0:
        return None
    tail = buf[i:]
    if len(tail) == 1:
        return i
    kind = tail[1:2]
    if kind == b'[':
        return None if re.match(rb'\x1b\[[\x30-\x3f]*[\x20-\x2f]*[\x40-\x7e]', tail) else i
    if kind in (b']', b'_', b'P', b'^'):
        # Ended by BEL, else open: its ST (ESC \\) would be a later ESC.
        return None if b'\x07' in tail[2:] else i
    return None


def utf8_tail(buf):
    """How many bytes at the end of buf start a UTF-8 sequence not yet whole."""
    for back in range(1, min(4, len(buf)) + 1):
        b = buf[-back]
        if b & 0xC0 == 0x80:
            continue
        need = 2 if b & 0xE0 == 0xC0 else 3 if b & 0xF0 == 0xE0 else 4 if b & 0xF8 == 0xF0 else 1
        return back if need > back else 0
    return 0


def clean_env(home, binary):
    """The environment from nothing: no session, mount, display or bus of
    the caller's, and the recorded pardes first on PATH."""
    runtime = home / 'run'
    env = dict(
        PATH=f'{binary.parent}:/usr/local/bin:/usr/bin:/bin',
        HOME=str(home), USER='demo', LOGNAME='demo',
        SHELL='/bin/sh', TERM='xterm-256color', LANG='C.UTF-8', LC_ALL='C.UTF-8', TZ='UTC',
        XDG_CONFIG_HOME=str(home / 'config'), XDG_DATA_HOME=str(home / 'data'),
        XDG_STATE_HOME=str(home / 'state'), XDG_CACHE_HOME=str(home / 'cache'), XDG_RUNTIME_DIR=str(runtime),
        PS1='$ ', GIT_CONFIG_NOSYSTEM='1',
        GIT_AUTHOR_NAME='Demo', GIT_AUTHOR_EMAIL='[email protected]',
        GIT_COMMITTER_NAME='Demo', GIT_COMMITTER_EMAIL='[email protected]',
        # The session's own, set for the recording: no clock in messages, no
        # language server racing the keys.
        PARDES_NOTIME='1',
    )
    for language in ['RS', 'C', 'GO', 'TS', 'PY', 'ZIG']:
        env['PARDES_LSP_' + language] = ''
    return env


def descendants(pid):
    """pid's descendants, by /proc's parent links."""
    children = {}
    for d in Path('/proc').iterdir():
        if not d.name.isdigit():
            continue
        try:
            stat = (d / 'stat').read_text()
        except OSError:
            continue
        ppid = int(stat[stat.rfind(')') + 2:].split()[1])
        children.setdefault(ppid, []).append(int(d.name))
    out, todo = [], [pid]
    while todo:
        for c in children.get(todo.pop(), []):
            out.append(c)
            todo.append(c)
    return out


class Recording:
    """pardes in a pty of ours, the screen it draws, and the cast so far."""

    def __init__(self, binary, home, args, cwd, log=None):
        self.binary = Path(binary).resolve()
        self.log = log
        self.screen = pyte.Screen(COLS, ROWS)
        self.stream = pyte.ByteStream(self.screen)
        self.lock = threading.Lock()
        self.carry = b''        # an escape sequence or UTF-8 character not yet whole
        self.pending = bytearray()  # cleaned output not yet in an event
        self.events = []        # (seconds, the screen's changes as text)
        self.raw_events = []    # (seconds, bytes pardes wrote)
        self.shown = None       # the screen as the cast has drawn it
        self.clock = 0.0
        self.last_out = time.monotonic()
        env = clean_env(home, self.binary)
        master, slave = os.openpty()
        fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack('HHHH', ROWS, COLS, COLS * CELL_W, ROWS * CELL_H))
        self.master = master
        self.child = subprocess.Popen(
            [str(self.binary), *map(str, args)], cwd=cwd, env=env,
            stdin=slave, stdout=slave, stderr=open(home / 'stderr.log', 'ab'), start_new_session=True,
            preexec_fn=lambda: fcntl.ioctl(0, termios.TIOCSCTTY, 0))
        os.close(slave)
        self.reader = threading.Thread(target=self._read, daemon=True)
        self.reader.start()

    # output

    def _read(self):
        while True:
            try:
                data = os.read(self.master, 65536)
            except OSError:
                return
            if not data:
                return
            with self.lock:
                self._take(data)
                self.last_out = time.monotonic()

    def _take(self, data):
        buf = self.carry + data
        cut = incomplete_tail(buf)
        if cut is None:
            cut = len(buf) - utf8_tail(buf)
        buf, self.carry = buf[:cut], buf[cut:]
        at = 0
        for m in QUERY.finditer(buf):
            self._show(buf[at:m.start()])
            at = m.end()
            self._answer(m)
        self._show(buf[at:])

    def _show(self, data):
        if not data:
            return
        data = SGR.sub(sgr_semicolons, data)
        self.pending += data
        self.stream.feed(data)

    def _answer(self, m):
        reply = None
        if m.group('decrqm') is not None:
            reply = b'\x1b[?%s;0$y' % m.group('decrqm')
        elif m.group('cpr') is not None:
            reply = b'\x1b[%d;%dR' % (self.screen.cursor.y + 1, self.screen.cursor.x + 1)
        elif m.group('dsr') is not None:
            reply = b'\x1b[0n'
        elif m.group('da1') is not None:
            reply = DA1
        elif m.group('xtv') is not None:
            reply = XTVERSION
        elif m.group('winop') == b'14':
            reply = b'\x1b[4;%d;%dt' % (ROWS * CELL_H, COLS * CELL_W)
        elif m.group('winop') == b'16':
            reply = b'\x1b[6;%d;%dt' % (CELL_H, CELL_W)
        elif m.group('winop') == b'18':
            reply = b'\x1b[8;%d;%dt' % (ROWS, COLS)
        if reply:
            os.write(self.master, reply)

    # the screen

    def lines(self):
        with self.lock:
            return list(self.screen.display)

    def text(self):
        return '\n'.join(self.lines())

    def find(self, text, after=0, nth=0, before=None):
        """(col, row) of `text` on screen, searching rows from `after`."""
        for r, line in enumerate(self.lines()):
            if r < after or (before is not None and r >= before):
                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' + self.text())

    def cursor(self):
        with self.lock:
            return self.screen.cursor.x, self.screen.cursor.y

    # input

    def send(self, data):
        os.write(self.master, data.encode() if isinstance(data, str) else data)

    def mouse(self, button, col, row, release=False, motion=False):
        """An SGR mouse report: button 0 left, 1 middle, 2 right, 3 none;
        col and row from 0."""
        code = button + (32 if motion else 0)
        self.send(f'\x1b[<{code};{col + 1};{row + 1}{"m" if release else "M"}')

    # steps

    def busy(self):
        """Whether a thread of pardes is running (or in disk wait)."""
        tasks = Path(f'/proc/{self.child.pid}/task')
        with contextlib.suppress(OSError):
            for t in tasks.iterdir():
                with contextlib.suppress(OSError):
                    stat = (t / 'stat').read_text()
                    if stat[stat.rfind(')') + 2] in 'RD':
                        return True
        return False

    def settle(self, until=None, quiet=.15, timeout=20, what='', sent=None):
        """Wait for `until` and for pardes to go quiet and idle, after
        drawing at least once since the input sent at `sent` (or for .8 s
        to pass without its drawing: an input that changes nothing)."""
        deadline = time.monotonic() + timeout
        while True:
            now = time.monotonic()
            with self.lock:
                drew = sent is None or self.last_out > sent or now - sent > .8
                idle = drew and now - self.last_out >= quiet
            idle = idle and not self.busy()
            ok = until is None or until(self)
            if ok and idle:
                return
            if now > deadline:
                raise RuntimeError(f'timed out waiting for {what or until}\n' + self.text())
            if self.child.poll() is not None:
                raise RuntimeError('pardes exited\n' + self.text())
            time.sleep(.01)

    def step(self, keys=None, hold=.5, until=None, quiet=.15, timeout=20, what=''):
        """Send `keys` (bytes, str or a callable), wait for `until` and quiet,
        stamp the output at the clock and move the clock on by `hold`."""
        sent = time.monotonic() if keys else None
        if callable(keys):
            keys(self)
        elif keys:
            self.send(keys)
        self.settle(until, quiet, timeout, what, sent)
        self.flush()
        self.clock += min(hold, MAX_GAP)
        if self.log:
            self.log.write(f'--- t={self.clock:.2f} {keys!r} {what}\n' + self.text() + '\n')

    def flush(self):
        """Stamp what pardes wrote since the last step at the clock: as the
        screen cells it changed (the cast), and as its raw bytes (--raw)."""
        with self.lock:
            data, self.pending = bytes(self.pending), bytearray()
            now = snapshot(self.screen)
        t = round(self.clock, 3)
        if data:
            self.raw_events.append((t, data))
        delta = repaint(self.shown, now)
        if delta:
            self.events.append((t, delta))
            self.shown = now

    def type(self, text, per_key=.09, last_hold=.4, until=None):
        for i, ch in enumerate(text):
            last = i == len(text) - 1
            self.step(ch, hold=last_hold if last else per_key, until=until if last else None, quiet=.06)

    def click(self, button, col, row, hold=.4, until=None, what='', quiet=.15):
        """Press and release `button` over one cell."""
        def act(s):
            s.mouse(3, col, row, motion=True)
            time.sleep(.03)
            s.mouse(button, col, row)
            time.sleep(.03)
            s.mouse(button, col, row, release=True)
        self.step(act, hold=hold, until=until, what=what, quiet=quiet)

    def sweep(self, button, start, end, show=.6, hold=.6, until=None, what='', quiet=.15):
        """Press `button` at `start`, drag to `end`, hold it there for a
        frame (the sweep's colour shows), then release."""
        def press(s):
            s.mouse(3, *start, motion=True)
            time.sleep(.03)
            s.mouse(button, *start)
            c0, r0 = start
            c1, r1 = end
            n = max(abs(c1 - c0), abs(r1 - r0))
            for i in range(1, n + 1):
                time.sleep(.01)
                s.mouse(button, c0 + (c1 - c0) * i // n, r0 + (r1 - r0) * i // n, motion=True)
        self.step(press, hold=show)
        self.step(lambda s: s.mouse(button, *end, release=True), hold=hold, until=until, what=what, quiet=quiet)

    def hover(self, col, row, hold=.3):
        self.step(lambda s: s.mouse(3, col, row, motion=True), hold=hold)

    def close(self):
        pids = descendants(self.child.pid)
        if self.child.poll() is None:
            self.child.terminate()
            try:
                self.child.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.child.kill()
                self.child.wait()
        for pid in pids + descendants(self.child.pid):
            with contextlib.suppress(ProcessLookupError, PermissionError):
                os.kill(pid, signal.SIGKILL)
        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)

    def cast(self, title, raw=False):
        """The asciicast v2 text: the events, then the last hold as an
        empty event so the player keeps the last frame up."""
        head = dict(version=2, width=COLS, height=ROWS, title=title,
                    env=dict(TERM='xterm-256color', SHELL='/bin/sh'))
        out = [json.dumps(head)]
        if raw:
            decoder = codecs.getincrementaldecoder('utf-8')('replace')
            events = [(t, decoder.decode(data)) for t, data in self.raw_events]
        else:
            events = self.events
        # Steps that changed nothing leave their holds end to end: no gap
        # longer than MAX_GAP survives.
        last = shift = 0.0
        for t, text in events + [(self.clock, '')]:
            shift += max(0.0, t - last - MAX_GAP)
            last = t
            out.append(json.dumps([round(t - shift, 3), 'o', text], ensure_ascii=False))
        return '\n'.join(out) + '\n'


# ---------------------------------------------------------------------------
# the cast's own drawing: pyte's screen, cell by cell
#
# pardes draws diffs against its last frame, and how many frames a step
# took (a shell's output racing the frames) varies from run to run while
# the screen it ends on does not. So an event is the cells the step left
# changed, drawn afresh: the same steps make the same file.

NAMED = {name: i for i, name in enumerate(['black', 'red', 'green', 'brown', 'blue', 'magenta', 'cyan', 'white'])}


def snapshot(screen):
    """The screen's cells as tuples, and its cursor."""
    cells = [[tuple(c) for c in (screen.buffer[y][x] for x in range(screen.columns))]
             for y in range(screen.lines)]
    return cells, (screen.cursor.x, screen.cursor.y, screen.cursor.hidden)


def colour(name, base):
    """SGR parameters for a pyte colour: base 30 for fg, 40 for bg."""
    if name == 'default':
        return [str(base + 9)]
    if name in NAMED:
        return [str(base + NAMED[name])]
    if name.startswith('bright') and name[6:] in NAMED:
        return [str(base + 60 + NAMED[name[6:]])]
    r, g, b = (int(name[i:i + 2], 16) for i in (0, 2, 4))
    return [str(base + 8), '2', str(r), str(g), str(b)]


def pen(cell):
    _, fg, bg, bold, italics, underscore, strikethrough, reverse, blink = cell
    params = ['0'] + colour(fg, 30) + colour(bg, 40)
    for on, code in ((bold, '1'), (italics, '3'), (underscore, '4'), (blink, '5'), (reverse, '7'), (strikethrough, '9')):
        if on:
            params.append(code)
    return '\x1b[' + ';'.join(params) + 'm'


def repaint(old, new):
    """The text that turns screen `old` (None: a fresh terminal) into `new`."""
    cells, (cx, cy, hidden) = new
    out = []
    if old is None:
        out.append('\x1b[?25l\x1b[0m\x1b[2J')
        blank = (' ', 'default', 'default', False, False, False, False, False, False)
        old_cells, old_cursor = [[blank] * len(row) for row in cells], None
    else:
        old_cells, old_cursor = old
    at, drawn_pen = None, None
    for y, (row, old_row) in enumerate(zip(cells, old_cells)):
        for x, cell in enumerate(row):
            if cell == old_row[x] or cell[0] == '':  # unchanged, or the right half of a wide character
                continue
            if at != (x, y):
                out.append(f'\x1b[{y + 1};{x + 1}H')
            p = pen(cell)
            if p != drawn_pen:
                out.append(p)
                drawn_pen = p
            out.append(cell[0])
            at = (x + (2 if x + 1 < len(row) and row[x + 1][0] == '' else 1), y)
    if not out and old_cursor == (cx, cy, hidden):
        return ''
    out.append(f'\x1b[0m\x1b[{cy + 1};{cx + 1}H' + ('\x1b[?25l' if hidden else '\x1b[?25h'))
    return ''.join(out)


# ---------------------------------------------------------------------------
# the fixtures

MAIN_C = '''#include <stdio.h>

static int square(int n)
{
    return n * n;
}

int main(void)
{
    int sum = 0; /* 1 + 4 + 9 + 16 */
    for (int i = 1; i <= 4; i++)
        sum += square(i);
    printf("squares: %d\\n", sum);
    return 0;
}
'''

MAKEFILE = '''CFLAGS = -Wall -fno-diagnostics-show-caret -fno-diagnostics-show-option -fno-diagnostics-color

hello: main.c
\t@cc $(CFLAGS) -o hello main.c
'''

README = '''hello: prints the sum of the first four squares.

Build it with make, run it with ./hello.
'''

NOTES = '''hello: notes

The sum of the squares of 1 to 4 is 30.
What ./hello prints:

'''


def git(*args, cwd=PROJECT, env=None, date=None):
    e = dict(env or {})
    if date:
        e.update(GIT_AUTHOR_DATE=date, GIT_COMMITTER_DATE=date)
    subprocess.run(['git', *args], cwd=cwd, env=e, check=True, capture_output=True)


def make_project(home):
    shutil.rmtree(PROJECT, ignore_errors=True)
    PROJECT.mkdir(parents=True)
    env = clean_env(home, Path('/nonexistent/pardes'))
    env.pop('PS1')
    git('init', '-q', '-b', 'main', env=env)
    git('config', 'color.ui', 'always', env=env)
    history = [
        ('2026-09-21T10:00:00Z', 'Start hello', {'README': README}),
        ('2026-09-22T11:30:00Z', 'Add a Makefile', {'Makefile': MAKEFILE}),
        ('2026-09-24T09:15:00Z', 'Print the sum of squares', {'main.c': MAIN_C}),
        ('2026-09-27T16:40:00Z', 'Keep notes', {'NOTES': NOTES}),
    ]
    for date, message, files in history:
        for name, body in files.items():
            (PROJECT / name).write_text(body)
        git('add', '.', env=env)
        git('commit', '-q', '-m', message, env=env, date=date)
    git('tag', 'v0.1', env=env)
    (PROJECT / '.gitignore').write_text('hello\n')


ART = PROJECT.parent / 'art'
SDM = Path('/home/goblin/01-resources/intel-devref-ii.pdf')

# The escape-time bands, outside in: deep blue to white to orange, chosen
# to land on distinct C64 colours when the matcher picks its 16.
CYCLE = [(20, 40, 160), (40, 120, 230), (120, 210, 250), (250, 250, 250),
         (250, 220, 60), (230, 90, 20), (120, 30, 40), (20, 40, 160)]


def band(t):
    t = t % 1.0 * (len(CYCLE) - 1)
    i, u = int(t), t % 1.0
    return tuple(round(a + (b - a) * u) for a, b in zip(CYCLE[i], CYCLE[i + 1]))


def mandelbrot(path, cx, cy, width, iterations, period, size=(480, 320)):
    """The Mandelbrot set around cx+cy·i, `width` across: the set black,
    the outside in smooth escape-time bands that repeat every `period`
    (in √iterations). Pure Python, a few seconds."""
    from PIL import Image
    w, h = size
    height = width * h / w
    im = Image.new('RGB', size)
    px = im.load()
    for j in range(h):
        y0 = cy + height / 2 - height * (j + .5) / h
        for i in range(w):
            x0 = cx - width / 2 + width * (i + .5) / w
            x = y = 0.0
            n = 0
            while x * x + y * y <= 256 and n < iterations:
                x, y = x * x - y * y + x0, 2 * x * y + y0
                n += 1
            if n == iterations:
                px[i, j] = (0, 0, 0)
            else:
                nu = n + 1 - math.log2(math.log2(math.sqrt(x * x + y * y)))
                px[i, j] = band(math.sqrt(max(nu, 0)) / period)
    im.save(path)


ART_README = '''pictures, right-click a name:

  mandelbrot.png   the whole set
  seahorse.png     a seahorse valley, zoomed in
  sdm.pdf          Intel SDM vol. 2, 2,515 pages
'''


def make_art(pdf):
    """The classic view, a seahorse valley's spiral, a short name for the
    manual (a symlink: the PDF is only read), and a README naming them."""
    shutil.rmtree(ART, ignore_errors=True)
    ART.mkdir(parents=True)
    mandelbrot(ART / 'mandelbrot.png', -0.65, 0.0, 3.2, 96, 10.5)
    mandelbrot(ART / 'seahorse.png', -0.7436, 0.1318, 0.0035, 400, 6.0)
    (ART / 'sdm.pdf').symlink_to(pdf)
    (ART / 'README').write_text(ART_README)


# ---------------------------------------------------------------------------
# checking: replay a cast through pyte, print and draw its last frame

def replay(path, until=None):
    """The pyte screen after the cast's events (up to `until` seconds)."""
    screen = pyte.Screen(COLS, ROWS)
    stream = pyte.Stream(screen)
    with open(path) as f:
        head = json.loads(f.readline())
        assert head['width'] == COLS and head['height'] == ROWS
        for line in f:
            t, kind, data = json.loads(line)
            if until is not None and t > until:
                break
            stream.feed(data)
    return screen


PYTE_COLOURS = dict(black=(0, 0, 0), red=(205, 0, 0), green=(0, 205, 0), brown=(205, 205, 0),
                    blue=(0, 0, 238), magenta=(205, 0, 205), cyan=(0, 205, 205), white=(229, 229, 229))


def rgb(colour, default):
    if colour == 'default':
        return default
    if colour in PYTE_COLOURS:
        return PYTE_COLOURS[colour]
    if colour.startswith('bright') and colour[6:] in PYTE_COLOURS:
        return tuple(min(255, v + 50) for v in PYTE_COLOURS[colour[6:]])
    with contextlib.suppress(ValueError):
        return tuple(int(colour[i:i + 2], 16) for i in (0, 2, 4))
    return default


def block_rects(cp, w, h):
    """The filled rectangles of a block element or sextant, else None."""
    if 0x1FB00 <= cp <= 0x1FB3B:  # sextants, skipping the four that are blocks
        n = cp - 0x1FB00 + 1
        n += n >= 21
        n += n >= 42
        bits = n
    elif cp in (0x258C, 0x2590, 0x2588):
        bits = {0x258C: 21, 0x2590: 42, 0x2588: 63}[cp]
    else:
        halves = {0x2580: [(0, 0, 1, .5)], 0x2584: [(0, .5, 1, 1)], 0x2596: [(0, .5, .5, 1)],
                  0x2597: [(.5, .5, 1, 1)], 0x2598: [(0, 0, .5, .5)], 0x259D: [(.5, 0, 1, .5)],
                  0x2599: [(0, 0, .5, 1), (.5, .5, 1, 1)], 0x259B: [(0, 0, 1, .5), (0, .5, .5, 1)],
                  0x259C: [(0, 0, 1, .5), (.5, .5, 1, 1)], 0x259F: [(.5, 0, 1, 1), (0, .5, .5, 1)],
                  0x259A: [(0, 0, .5, .5), (.5, .5, 1, 1)], 0x259E: [(.5, 0, 1, .5), (0, .5, .5, 1)]}
        if cp not in halves:
            return None
        return [(x0 * w, y0 * h, x1 * w, y1 * h) for x0, y0, x1, y1 in halves[cp]]
    rects = []
    for k in range(6):
        if bits >> k & 1:
            col, row = k % 2, k // 2
            rects.append((col * w / 2, row * h / 3, (col + 1) * w / 2, (row + 1) * h / 3))
    return rects


def render_png(screen, path, cw=10, ch=20):
    """A picture of a pyte screen: blocks and sextants drawn, the rest in
    the repo's Adwaita Mono."""
    from PIL import Image, ImageDraw, ImageFont
    font = ImageFont.truetype(str(REPO / 'assets/AdwaitaMono-Regular.ttf'), 16)
    im = Image.new('RGB', (screen.columns * cw, screen.lines * ch), (0, 0, 0))
    d = ImageDraw.Draw(im)
    for y in range(screen.lines):
        row = screen.buffer[y]
        for x in range(screen.columns):
            c = row[x]
            fg, bg = rgb(c.fg, (229, 229, 229)), rgb(c.bg, (0, 0, 0))
            if c.reverse != (not screen.cursor.hidden and (x, y) == (screen.cursor.x, screen.cursor.y)):
                fg, bg = bg, fg
            x0, y0 = x * cw, y * ch
            d.rectangle((x0, y0, x0 + cw - 1, y0 + ch - 1), fill=bg)
            if not c.data or c.data == ' ':
                continue
            rects = block_rects(ord(c.data[0]), cw, ch) if len(c.data) == 1 else None
            if rects is not None:
                for a, b, e, f in rects:
                    d.rectangle((x0 + a, y0 + b, x0 + e - 1, y0 + f - 1), fill=fg)
            else:
                d.text((x0, y0 + 1), c.data, font=font, fill=fg)
    im.save(path)


# ---------------------------------------------------------------------------
# the casts

def tag_end(r, row):
    """Two cells past the last word of the tag on `row`: a B1 click there
    types at the tag's end."""
    return len(r.lines()[row].rstrip()) + 1, row


def type_in_tag(r, row, text):
    r.click(0, *tag_end(r, row), hold=.3)
    r.type(text, until=lambda s: text.strip() in s.lines()[row], last_hold=.3)
    r.step('\x1b', hold=.4)


def word_at(r, word, row):
    """(start, end) cells of `word` on `row`, end exclusive, as a sweep takes it."""
    c, w = r.find(word, after=row, before=row + 1)
    return (c, w), (c + len(word), w)


def keys(r, seq, hold, per_key=.22):
    for i, k in enumerate(seq):
        r.step(k, hold=hold if i == len(seq) - 1 else per_key)


def live_text(r):
    """Helix keys rename a variable and save; B2 on make runs it, B3 on the
    error it prints lands on the line, and the rename goes on there."""
    r.step(None, until=lambda s: '#include' in s.text(), quiet=.4, hold=1.6)
    for _ in range(9):
        r.step('j', hold=.13, quiet=.05)
    r.clock += .3
    keys(r, 'wwe', .35)
    keys(r, 'miw', .7)
    r.step('c', hold=.3)
    r.type('total', until=lambda s: 'int total' in s.text(), last_hold=.3)
    r.step('\x1b', hold=.6)
    keys(r, ' fs', 1.1)
    type_in_tag(r, 2, ' make')
    start, end = word_at(r, 'make', 2)
    r.sweep(1, start, end, show=.8, hold=2.0, until=lambda s: 'exit 2' in s.text(), what='make')
    c, w = r.find('main.c:12:9')
    r.sweep(2, (c, w), (c + len('main.c:12:9'), w), show=.8, hold=1.6,
            until=lambda s: s.cursor()[1] < 15, what='the look')
    keys(r, 'miw', .6)
    r.step('c', hold=.3)
    r.type('total', until=lambda s: 'total += square' in s.text(), last_hold=.3)
    r.step('\x1b', hold=2.0)


def terminals(r):
    """A shell beside the notes: run, Ctrl-b, copy a line of output into
    the file; then git log paged through pardes - into a +Pager."""
    r.step(None, until=lambda s: 'What ./hello prints' in s.text(), quiet=.4, hold=1.2)
    start, end = word_at(r, 'Tty', 2)
    r.sweep(1, start, end, show=.5, hold=.6, until=lambda s: 'Tty+sh' in s.text(), what='the shell')
    r.step(None, until=lambda s: s.text().count('$') >= 2, hold=.4)
    # Its own column, beside the file: Ctrl-b to normal, Alt-c, Ctrl-b back.
    r.step('\x02', hold=.3)
    r.step('\x1bc', hold=.4, until=lambda s: s.lines()[1].count('Delcol') == 2, what='the new column')
    r.step('\x02', hold=.6)
    r.type('make && ./hello', until=lambda s: 'make && ./hello' in s.text(), per_key=.07)
    r.step('\r', hold=1.2, until=lambda s: 'squares: 30' in s.text(), what='./hello')
    r.step('\x02', hold=.6)
    r.step('k', hold=.4)
    r.step('x', hold=.6)
    r.step('y', hold=.5)
    r.step('\x17', hold=.2)
    r.step('h', hold=.6)
    keys(r, 'ge', .5)
    r.step('p', hold=1.0, until=lambda s: 'squares: 30' in s.lines()[8][:48], what='the paste')
    r.step('\x17', hold=.2)
    r.step('l', hold=.4)
    r.step('\x02', hold=.6)
    command = 'git log --stat | pardes -'
    r.type(command, until=lambda s: command in s.text(), per_key=.07)
    r.step('\r', hold=2.0, until=lambda s: '+Pager' in s.text() and 'Keep notes' in s.text(), quiet=.3,
           what='the pager')


def petscii(r):
    """Right-click opens the Mandelbrot views, drawn as PETSCII glyph art;
    then the Intel manual: its sections, a jump to CPUID and a search."""
    def look(name, until, hold):
        c, w = r.find(name)
        r.sweep(2, (c, w), (c + len(name), w), show=.5, hold=hold, until=until, quiet=.4, what=name)

    def close(name, hold):
        c, w = r.find('art/' + name, after=2)
        start, end = word_at(r, 'Del', w)
        r.sweep(1, start, end, show=.4, hold=hold, until=lambda s: 'art/' + name not in s.text(), what='Del ' + name)

    r.step(None, until=lambda s: 'sdm.pdf' in s.text(), quiet=.4, hold=1.6)
    look('mandelbrot.png', lambda s: 'art/mandelbrot.png' in s.text(), 2.0)
    close('mandelbrot.png', .5)
    look('seahorse.png', lambda s: 'art/seahorse.png' in s.text(), 2.0)
    close('seahorse.png', .5)
    look('sdm.pdf', lambda s: '[1/2515]' in s.text(), 1.5)
    c, w = r.find('PdfSections')
    r.sweep(1, (c, w), (c + len('PdfSections'), w), show=.5, hold=1.3,
            until=lambda s: 'Chapter 1' in s.text(), what='the sections')
    c, w = r.find('+PdfSections')
    type_in_tag(r, w, ' CPUID')
    for _ in range(2):  # B3 on a word in a tag finds it in the body
        start, end = word_at(r, 'CPUID', w)
        r.sweep(2, start, end, show=.4, hold=.9)
    c, row = r.find('intel-devref-ii.pdf:317:200')
    r.sweep(2, (c, row), (c + len('intel-devref-ii.pdf:317'), row), show=.5, hold=1.8,
            until=lambda s: '[317/2515]' in s.text(), what='the jump')
    start, end = word_at(r, 'Del', w)
    r.sweep(1, start, end, show=.4, hold=.8, until=lambda s: '+PdfSections' not in s.text(), what='Del sections')
    c, w = r.find('CPUID—CPU Identification')
    r.click(0, c, w + 3, hold=.3)
    r.step('/', hold=.3)
    r.type('bit 21', until=lambda s: '/bit 21' in s.text(), last_hold=.3)
    r.step('\r', hold=2.0, until=lambda s: '+Search' in s.text(), quiet=.4, timeout=60, what='the search')


CASTS = {
    'live-text': dict(
        run=live_text, cwd=PROJECT, args=['main.c'],
        title='Live text',
        caption='Helix keys edit; middle-click runs make; right-click the error opens its line.'),
    'terminals': dict(
        run=terminals, cwd=PROJECT, args=['NOTES'],
        title='Terminals as panes',
        caption='A shell beside a file: Ctrl-b, copy a line, page git log.'),
    'petscii': dict(
        run=petscii, cwd=ART, args=['README'],
        title='Images and PDFs in a plain terminal',
        caption='No kitty graphics: a Mandelbrot in PETSCII, then the Intel manual searched.'),
}


def make_home(work):
    home = work / 'home'
    shutil.rmtree(home, ignore_errors=True)
    (home / 'config/pardes').mkdir(parents=True)
    (home / 'run').mkdir(mode=0o700)
    # No animation or blink: a frame drawn on a timer would land in the cast
    # wherever the wall clock put it.
    (home / 'config/pardes/init').write_text(
        'Theme orchard\nShell /bin/sh\nMessageLinger 0\nMessageAnimation off\nCursorBlink off\n'
        'SelectionGlow off\nHoverGlow off\nJumpTrail off\nThumbFlash off\n')
    (home / '.gitconfig').write_text('[log]\n\tabbrevCommit = true\n[color]\n\tui = always\n')
    return home


def record(binary, work, name, pdf):
    cast = CASTS[name]
    home = make_home(work)
    make_project(home)
    make_art(pdf)
    log = open(work / f'{name}.steps.log', 'w')
    r = Recording(binary, home, cast['args'], cast['cwd'], log=log)
    try:
        cast['run'](r)
    finally:
        r.close()
        log.close()
    OUT.mkdir(parents=True, exist_ok=True)
    out = OUT / f'{name}.cast'
    out.write_text(r.cast(cast['title']))
    (work / f'{name}.raw.cast').write_text(r.cast(cast['title'], raw=True))
    length = json.loads(out.read_text().splitlines()[-1])[0]
    print(f'{name}: {length:.1f} s, {out.stat().st_size / 1024:.0f} KB')


def write_toml():
    lines = ['# The landing page\'s terminal recordings (asciicast v2), in the order',
             '# shown: casts/<name>.cast for each. The first plays by itself; all loop',
             '# and fit the page\'s width. While `ready` is false the cast shows',
             '# placeholder.svg, so the site builds before a recording exists.',
             '# test/record_casts.py writes the casts and this file.', '']
    for name, cast in CASTS.items():
        lines += ['[[cast]]', f'name = "{name}"', f'title = "{cast["title"]}"',
                  f'caption = "{cast["caption"]}"', f'cols = {COLS}', f'rows = {ROWS}',
                  f'ready = {"true" if (OUT / f"{name}.cast").exists() else "false"}', '']
    (OUT / 'casts.toml').write_text('\n'.join(lines))


def show(name, work, at=None):
    """Print a cast's frame (the last, or the one at `at` seconds) as text
    and draw it to <work>/<name>.png."""
    screen = replay(OUT / f'{name}.cast', at)
    print(f'--- {name}' + (f' at {at} s' if at is not None else ''))
    for line in screen.display:
        print(line.rstrip())
    render_png(screen, work / f'{name}.png')


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('binary', type=Path, nargs='?', help='a pardes built with -Dplatform=tty')
    parser.add_argument('casts', nargs='*', help='which casts (default: all): ' + ', '.join(CASTS))
    parser.add_argument('--work', type=Path, default=REPO.parent / '.scratch/casts/work',
                        help='the session home, step logs and check pictures (default %(default)s)')
    parser.add_argument('--pdf', type=Path, default=SDM, help='the manual the petscii cast reads (default %(default)s)')
    parser.add_argument('--show', action='store_true', help='print and draw each cast\'s last frame after')
    parser.add_argument('--show-only', action='store_true', help='record nothing; print and draw the casts as they are')
    parser.add_argument('--at', type=float, help='with --show: the frame at this many seconds instead')
    args = parser.parse_args()
    if args.binary is not None and args.binary.name in CASTS and not args.binary.exists():
        args.casts.insert(0, args.binary.name)
        args.binary = None
    names = args.casts or list(CASTS)
    for name in names:
        if name not in CASTS:
            parser.error(f'no cast {name!r}: ' + ', '.join(CASTS))
    work = args.work.resolve()
    work.mkdir(parents=True, exist_ok=True)
    if not args.show_only:
        if args.binary is None:
            parser.error('the pardes binary to record is required')
        if not args.pdf.is_file():
            parser.error(f'no PDF at {args.pdf}')
        try:
            for name in names:
                record(args.binary, work, name, args.pdf.resolve())
        finally:
            shutil.rmtree(PROJECT.parent, ignore_errors=True)
        write_toml()
    if args.show or args.show_only:
        for name in names:
            show(name, work, args.at)


if __name__ == '__main__':
    main()