improvise/docs/record-casts.py

#!/usr/bin/env python3
"""Record the four asciicasts under docs/casts/ from a driven improvise session.

Not `asciinema rec`: that records a SHELL, and the shell it records is the
recorder's own — every previous cast opens and closes on a real hostname and
home-directory path on what is now the site root (improvise-2oxt). Here there is
no shell at all. `improvise` is the pty's only child, so there is nothing for a
prompt to leak into, and the geometry, the pacing and the keys are inputs rather
than whatever the recording machine happened to have.

Usage:  python3 docs/record-casts.py [--out docs/casts] [--bin target/release/improvise]

The output is asciicast v2, which is what `asciinema-player` on docs/index.html
reads: one JSON header line, then `[time, "o", data]` per chunk.
"""

import argparse
import fcntl
import json
import os
import pty
import select
import signal
import struct
import subprocess
import sys
import termios
import time

COLS, ROWS = 120, 37

# Long enough that a reader can follow, short enough that a loop is bearable.
BEAT = 0.9
SETTLE = 0.35


def keys(*items):
    """A step list from (text, pause) pairs, with the pause defaulted."""
    return [(k, BEAT) if isinstance(k, str) else k for k in items]


ESC = "\x1b"


def scenarios(root):
    """The four the landing page embeds, each named for the section over it."""
    demo = os.path.join(root, "examples/demo.improv")
    csv = os.path.join(root, "examples/demo.csv")
    return {
        # "Reassign an axis": the tile bar, a category moved, the grid redrawn.
        "pivot": (
            [demo],
            keys(
                ("", 1.6),
                ("T", 1.2),
                *[("\x1b[C", 0.35)] * 5,
                ("c", 2.2),
                ("r", 2.2),
                (ESC, 1.6),
            ),
        ),
        # "Drill into an aggregate": the records grid behind one cell, which is
        # the cast 2oxt says is materially wrong — it predates the +cat and +
        # columns entirely.
        "drill": (
            [demo],
            keys(
                ("", 1.6),
                # Customer off the page axis first, so a cell aggregates over
                # all nine of them and the drill has more than one record to
                # show. Drilling the shipped layout shows exactly one row,
                # which is true and demonstrates nothing.
                ("T", 1.0),
                *[("\x1b[C", 0.35)] * 3,
                (" ", 1.8),
                (ESC, 1.2),
                # Gadgets/East at Cost 2025-01, which has records behind it.
                # The origin cell is blank, and a drill into nothing draws an
                # empty records grid — a true frame about the wrong thing.
                ("j", 1.0),
                (">", 2.2),
                *[("\x1b[C", 0.45)] * 6,
                ("", 1.4),
                (ESC, 1.6),
            ),
        ),
        # "Formulas over dimensions": the panel, then the formula that is
        # already in the document, read as text rather than as a cell address.
        "formulas": (
            [demo],
            keys(
                ("", 1.6),
                # The panel first: the formula is TEXT over categories and
                # items, which is the claim the section makes.
                ("F", 2.4),
                (ESC, 1.0),
                # Then move a category and let the Profit column follow. A
                # formula naming `B3` could not survive this, which is the
                # whole point and is only visible if the grid really moves.
                ("T", 1.0),
                *[("\x1b[C", 0.35)] * 7,
                ("r", 2.6),
                (ESC, 2.0),
            ),
        ),
        # "Import a CSV": the wizard's own steps, stepped through.
        "import": (
            ["import", csv, "-o", "/dev/null"],
            keys(
                ("", 1.4),
                ("\r", 1.2),
                ("\r", 1.2),
                ("\r", 1.2),
                ("\r", 1.2),
                ("\r", 1.4),
                (ESC, 1.0),
            ),
        ),
    }


def record(binary, argv, steps, cwd):
    """Drive one session on a pty and return its asciicast event list."""
    master, slave = pty.openpty()
    fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", ROWS, COLS, 0, 0))
    env = {
        "TERM": "xterm-256color",
        "PATH": "/usr/bin:/bin",
        "LANG": "en_US.UTF-8",
        "HOME": "/tmp",
        # `improvise` is the only child, so nothing else can print a path.
        "COLUMNS": str(COLS),
        "LINES": str(ROWS),
    }
    child = subprocess.Popen(
        [binary, *argv],
        stdin=slave,
        stdout=slave,
        stderr=slave,
        env=env,
        cwd=cwd,
        close_fds=True,
    )
    os.close(slave)

    started = time.time()
    events = []

    def drain(until):
        """Everything the child says before `until`, timestamped as it arrives."""
        while True:
            budget = until - time.time()
            if budget <= 0:
                return
            ready, _, _ = select.select([master], [], [], min(budget, 0.05))
            if not ready:
                continue
            try:
                chunk = os.read(master, 65536)
            except OSError:
                return
            if not chunk:
                return
            events.append(
                [round(time.time() - started, 6), "o", chunk.decode("utf-8", "replace")]
            )

    for text, pause in steps:
        if text:
            os.write(master, text.encode())
        drain(time.time() + pause)

    # No `:q` typed at the end: the player loops, so the last frame is also the
    # frame after the last one, and a minibuffer reading `:q` is what it would
    # show. The process is signalled instead and the recording ends on the UI.
    drain(time.time() + SETTLE)
    if child.poll() is None:
        child.send_signal(signal.SIGTERM)
        try:
            child.wait(timeout=3)
        except subprocess.TimeoutExpired:
            child.kill()
    os.close(master)
    return events


def main():
    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default=os.path.join(root, "docs/casts"))
    ap.add_argument("--bin", default=os.path.join(root, "target/release/improvise"))
    ap.add_argument("--only", default=None)
    args = ap.parse_args()

    if not os.path.exists(args.bin):
        sys.exit(f"no binary at {args.bin} — cargo build --release --bin improvise")

    for name, (argv, steps) in scenarios(root).items():
        if args.only and name != args.only:
            continue
        events = record(args.bin, argv, steps, root)
        header = {
            "version": 2,
            "width": COLS,
            "height": ROWS,
            # No `timestamp` and no `env`: the first carries the recording
            # machine's clock and the second carried $SHELL (improvise-2oxt).
            "idle_time_limit": 2.0,
            "title": f"improvise — {name}",
        }
        path = os.path.join(args.out, f"{name}.cast")
        with open(path, "w") as out:
            out.write(json.dumps(header) + "\n")
            for event in events:
                out.write(json.dumps(event) + "\n")
        print(f"{path}: {len(events)} chunks, {events[-1][0] if events else 0:.1f}s")


if __name__ == "__main__":
    main()