|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Drive an lnav session through a scripted keystroke sequence and |
| 3 | +emit an animated SVG that cycles through the captured frames.""" |
| 4 | + |
| 5 | +from __future__ import annotations |
| 6 | + |
| 7 | +import argparse |
| 8 | +import copy |
| 9 | +import os |
| 10 | +import select |
| 11 | +import signal |
| 12 | +import sys |
| 13 | +import time |
| 14 | + |
| 15 | +import pyte |
| 16 | + |
| 17 | +# Reuse the PTY + pyte + SVG-render infrastructure. |
| 18 | +import render as R |
| 19 | + |
| 20 | + |
| 21 | +def drain(stream: pyte.ByteStream, fd: int, duration: float, quiet: float, |
| 22 | + raw_sink: bytearray | None = None) -> None: |
| 23 | + """Read from `fd` for at most `duration` seconds, but bail early |
| 24 | + once the child has been idle for `quiet` seconds.""" |
| 25 | + start = time.time() |
| 26 | + deadline = start + duration |
| 27 | + last_read = start |
| 28 | + min_capture = start + 0.15 |
| 29 | + while True: |
| 30 | + now = time.time() |
| 31 | + if now >= deadline: |
| 32 | + return |
| 33 | + if now > min_capture and (now - last_read) > quiet: |
| 34 | + return |
| 35 | + r, _, _ = select.select([fd], [], [], 0.05) |
| 36 | + if not r: |
| 37 | + continue |
| 38 | + try: |
| 39 | + data = os.read(fd, 65536) |
| 40 | + except OSError: |
| 41 | + return |
| 42 | + if not data: |
| 43 | + return |
| 44 | + if raw_sink is not None: |
| 45 | + raw_sink.extend(data) |
| 46 | + stream.feed(data) |
| 47 | + last_read = time.time() |
| 48 | + |
| 49 | + |
| 50 | +def snapshot(screen: pyte.Screen) -> pyte.Screen: |
| 51 | + """Deep-copy just the buffer we need for rendering. A full |
| 52 | + deepcopy of the Screen is slow; we only need the cell grid.""" |
| 53 | + snap = pyte.Screen(screen.columns, screen.lines) |
| 54 | + for y in range(screen.lines): |
| 55 | + for x in range(screen.columns): |
| 56 | + snap.buffer[y][x] = screen.buffer[y][x] |
| 57 | + return snap |
| 58 | + |
| 59 | + |
| 60 | +def run(argv: list[str], cols: int, rows: int, |
| 61 | + steps: list[tuple[float, bytes, str]], |
| 62 | + initial_settle: float, initial_quiet: float, |
| 63 | + step_settle: float, step_quiet: float, |
| 64 | + debug: bool = False) -> list[tuple[pyte.Screen, str]]: |
| 65 | + """Spawn lnav, capture an initial frame, then replay each |
| 66 | + (wait-after-input, bytes, label) step. Returns a list of |
| 67 | + (screen, label) tuples.""" |
| 68 | + pid, fd = R.spawn(argv, cols, rows) |
| 69 | + screen = R._AnsweringScreen(cols, rows, fd) |
| 70 | + stream = R._AnsweringByteStream(screen) |
| 71 | + |
| 72 | + frames: list[tuple[pyte.Screen, str]] = [] |
| 73 | + drain(stream, fd, initial_settle, initial_quiet) |
| 74 | + frames.append((snapshot(screen), "initial")) |
| 75 | + if debug: |
| 76 | + print(f" frame 0: initial", file=sys.stderr) |
| 77 | + |
| 78 | + for i, (wait_after, data, label) in enumerate(steps, 1): |
| 79 | + if data: |
| 80 | + try: |
| 81 | + os.write(fd, data) |
| 82 | + except OSError as exc: |
| 83 | + print(f" write failed: {exc}", file=sys.stderr) |
| 84 | + break |
| 85 | + # Pure-wait steps (empty data) hold for the full duration so |
| 86 | + # late-arriving renders (e.g. async syntax-highlight parsers |
| 87 | + # that update the screen after a debounce) get captured. |
| 88 | + q = step_quiet if data else wait_after + 1.0 |
| 89 | + drain(stream, fd, wait_after, q) |
| 90 | + frames.append((snapshot(screen), label)) |
| 91 | + if debug: |
| 92 | + print(f" frame {i}: {label}", file=sys.stderr) |
| 93 | + |
| 94 | + # Clean up |
| 95 | + try: |
| 96 | + os.close(fd) |
| 97 | + except OSError: |
| 98 | + pass |
| 99 | + for _ in range(30): |
| 100 | + try: |
| 101 | + wpid, _ = os.waitpid(pid, os.WNOHANG) |
| 102 | + except ChildProcessError: |
| 103 | + break |
| 104 | + if wpid: |
| 105 | + break |
| 106 | + time.sleep(0.05) |
| 107 | + else: |
| 108 | + try: |
| 109 | + os.kill(pid, signal.SIGKILL) |
| 110 | + os.waitpid(pid, 0) |
| 111 | + except (ProcessLookupError, ChildProcessError): |
| 112 | + pass |
| 113 | + |
| 114 | + return frames |
| 115 | + |
| 116 | + |
| 117 | +_KEY_NAMES = { |
| 118 | + b"\t": "Tab", |
| 119 | + b"\r": "Enter", |
| 120 | + b"\n": "Enter", |
| 121 | + b"\x1b": "Esc", |
| 122 | + b" ": "Space", |
| 123 | + b"\x7f": "⌫", |
| 124 | +} |
| 125 | + |
| 126 | + |
| 127 | +def human_key(data: bytes) -> str: |
| 128 | + if data in _KEY_NAMES: |
| 129 | + return _KEY_NAMES[data] |
| 130 | + try: |
| 131 | + s = data.decode("utf-8") |
| 132 | + except UnicodeDecodeError: |
| 133 | + return data.hex() |
| 134 | + if len(s) == 1 and s.isprintable(): |
| 135 | + return s |
| 136 | + return data.hex() |
| 137 | + |
| 138 | + |
| 139 | +def render_animated_svg(frames: list[tuple[pyte.Screen, str]], |
| 140 | + steps: list[tuple[float, bytes, str]], |
| 141 | + cols: int, rows: int, |
| 142 | + per_frame: float, |
| 143 | + hold_last: float) -> str: |
| 144 | + """Emit an SVG where each frame is a <g> with CSS keyframe |
| 145 | + animation driving its visibility in sequence, looping forever.""" |
| 146 | + lay = R.layout(cols, rows) |
| 147 | + n = len(frames) |
| 148 | + durations = [per_frame] * (n - 1) + [hold_last] |
| 149 | + starts: list[float] = [] |
| 150 | + t = 0.0 |
| 151 | + for d in durations: |
| 152 | + starts.append(t) |
| 153 | + t += d |
| 154 | + total = t |
| 155 | + |
| 156 | + # Build per-frame CSS keyframes that hold each frame visible over |
| 157 | + # its [start, start+dur) window and hidden everywhere else. Using |
| 158 | + # `visibility` (not `display`) so the elements stay laid out — this |
| 159 | + # avoids reflow artifacts in some renderers. |
| 160 | + styles: list[str] = [] |
| 161 | + styles.append(".lnav-frame { visibility: hidden; }") |
| 162 | + for i, d in enumerate(durations): |
| 163 | + start_pct = (starts[i] / total) * 100 |
| 164 | + end_pct = ((starts[i] + d) / total) * 100 |
| 165 | + # Three-stop keyframe: hidden until start, visible through end, hidden after. |
| 166 | + # Use a tiny epsilon (0.0001) so transitions are instant. |
| 167 | + parts = [] |
| 168 | + if start_pct > 0: |
| 169 | + parts.append(f"0% {{ visibility: hidden; }}") |
| 170 | + parts.append(f"{start_pct - 0.0001:.4f}% {{ visibility: hidden; }}") |
| 171 | + parts.append(f"{start_pct:.4f}% {{ visibility: visible; }}") |
| 172 | + if end_pct < 100: |
| 173 | + parts.append(f"{end_pct - 0.0001:.4f}% {{ visibility: visible; }}") |
| 174 | + parts.append(f"{end_pct:.4f}% {{ visibility: hidden; }}") |
| 175 | + parts.append(f"100% {{ visibility: hidden; }}") |
| 176 | + else: |
| 177 | + parts.append(f"100% {{ visibility: visible; }}") |
| 178 | + styles.append(f"@keyframes lnav-f{i} {{ " + " ".join(parts) + " }") |
| 179 | + styles.append( |
| 180 | + f".lnav-f{i} {{ " |
| 181 | + f"animation: lnav-f{i} {total:.3f}s infinite; " |
| 182 | + f"animation-timing-function: step-end; " |
| 183 | + f"}}" |
| 184 | + ) |
| 185 | + |
| 186 | + # Build the cumulative keypress badge sequence for each frame. |
| 187 | + # Frame 0 has no keys; frame i>=1 has the keys from steps[0..i-1]. |
| 188 | + key_displays: list[str] = [""] * len(frames) |
| 189 | + acc: list[str] = [] |
| 190 | + for i, (_, data, _) in enumerate(steps, start=1): |
| 191 | + acc.append(human_key(data)) |
| 192 | + if i < len(frames): |
| 193 | + key_displays[i] = " ".join(acc) |
| 194 | + |
| 195 | + # Extend the SVG viewport to add a strip below the window for the |
| 196 | + # keypress badge. Leave room for the window's drop shadow too. |
| 197 | + badge_strip = 56.0 # total vertical space reserved below chrome |
| 198 | + badge_gap = 28.0 # shadow clearance between chrome bottom and text |
| 199 | + badge_font_size = 20.0 |
| 200 | + full_width = lay["width"] |
| 201 | + full_height = lay["height"] + badge_strip |
| 202 | + badge_y = lay["height"] + badge_gap + badge_font_size * 0.7 |
| 203 | + |
| 204 | + # svg_header emits a fixed width/height from lay; rebuild the header |
| 205 | + # to use the expanded height instead. |
| 206 | + out = [ |
| 207 | + '<?xml version="1.0" encoding="UTF-8"?>', |
| 208 | + (f'<svg xmlns="http://www.w3.org/2000/svg" ' |
| 209 | + f'width="{full_width:.2f}" height="{full_height:.2f}" ' |
| 210 | + f'viewBox="0 0 {full_width:.2f} {full_height:.2f}">'), |
| 211 | + '<defs>', |
| 212 | + ' <filter id="shadow" x="-10%" y="-10%" width="120%" height="130%">', |
| 213 | + ' <feDropShadow dx="0" dy="12" stdDeviation="16" flood-opacity="0.5"/>', |
| 214 | + ' </filter>', |
| 215 | + (f' <clipPath id="titleClip"><rect x="0" y="0" ' |
| 216 | + f'width="{lay["width"]:.2f}" height="{lay["title_bar"]:.2f}"/></clipPath>'), |
| 217 | + '</defs>', |
| 218 | + ] |
| 219 | + out.append("<style>\n" + "\n".join(styles) + "\n</style>") |
| 220 | + out += R.render_chrome(lay) |
| 221 | + # Keypress label strip below the window. |
| 222 | + badge_font = ('font-family="-apple-system, BlinkMacSystemFont, \'SF Pro Text\', ' |
| 223 | + '\'Helvetica Neue\', Helvetica, Arial, sans-serif" ' |
| 224 | + f'font-size="{badge_font_size}px" font-weight="500"') |
| 225 | + for i, kd in enumerate(key_displays): |
| 226 | + if not kd: |
| 227 | + continue |
| 228 | + out.append( |
| 229 | + f'<text class="lnav-frame lnav-f{i}" ' |
| 230 | + f'x="{full_width / 2:.2f}" y="{badge_y:.2f}" ' |
| 231 | + f'fill="#c8c8d0" {badge_font} ' |
| 232 | + f'text-anchor="middle">{kd}</text>' |
| 233 | + ) |
| 234 | + out.append(R.text_group_open(lay)) |
| 235 | + for i, (screen, label) in enumerate(frames): |
| 236 | + out.append(f'<g class="lnav-frame lnav-f{i}"><!-- frame {i}: {label} -->') |
| 237 | + out += [f' {s}' for s in R.render_frame(screen, lay)] |
| 238 | + out.append('</g>') |
| 239 | + out.append('</g>') |
| 240 | + out.append('</svg>') |
| 241 | + return "\n".join(out) + "\n" |
| 242 | + |
| 243 | + |
| 244 | +def parse_steps(spec: str) -> list[tuple[float, bytes, str]]: |
| 245 | + """Parse a simple step spec. Each line is: |
| 246 | +
|
| 247 | + wait_seconds \\t bytes (python-escape-decoded) \\t label |
| 248 | +
|
| 249 | + Example: |
| 250 | + 0.4\\t/\\tsearch prompt |
| 251 | + 0.3\\tv\\ttype v |
| 252 | + 0.6\\t\\x09\\tTAB |
| 253 | + """ |
| 254 | + out: list[tuple[float, bytes, str]] = [] |
| 255 | + for raw_line in spec.splitlines(): |
| 256 | + line = raw_line.rstrip("\n") |
| 257 | + if not line or line.lstrip().startswith("#"): |
| 258 | + continue |
| 259 | + parts = line.split("\t") |
| 260 | + if len(parts) < 2: |
| 261 | + continue |
| 262 | + wait = float(parts[0]) |
| 263 | + data = parts[1].encode("utf-8").decode("unicode_escape").encode("latin-1") |
| 264 | + label = parts[2] if len(parts) > 2 else f"step ({parts[1]!r})" |
| 265 | + out.append((wait, data, label)) |
| 266 | + return out |
| 267 | + |
| 268 | + |
| 269 | +def main() -> int: |
| 270 | + ap = argparse.ArgumentParser(description=__doc__) |
| 271 | + ap.add_argument("--cols", type=int, default=120) |
| 272 | + ap.add_argument("--rows", type=int, default=40) |
| 273 | + ap.add_argument("--initial-settle", type=float, default=6.0) |
| 274 | + ap.add_argument("--initial-quiet", type=float, default=1.2) |
| 275 | + ap.add_argument("--step-settle", type=float, default=2.0, |
| 276 | + help="Max seconds to wait after each keystroke") |
| 277 | + ap.add_argument("--step-quiet", type=float, default=0.8, |
| 278 | + help="Stop waiting after each keystroke once output " |
| 279 | + "has been quiet this long") |
| 280 | + ap.add_argument("--per-frame", type=float, default=0.7) |
| 281 | + ap.add_argument("--hold-last", type=float, default=2.5) |
| 282 | + ap.add_argument("--out", required=True) |
| 283 | + ap.add_argument("--steps", required=True, |
| 284 | + help="Path to a step file (see parse_steps)") |
| 285 | + ap.add_argument("--debug", action="store_true") |
| 286 | + ap.add_argument("cmd", nargs="+") |
| 287 | + args = ap.parse_args() |
| 288 | + |
| 289 | + with open(args.steps) as f: |
| 290 | + steps = parse_steps(f.read()) |
| 291 | + if not steps: |
| 292 | + print("error: no steps parsed", file=sys.stderr) |
| 293 | + return 2 |
| 294 | + frames = run(args.cmd, args.cols, args.rows, steps, |
| 295 | + args.initial_settle, args.initial_quiet, |
| 296 | + args.step_settle, args.step_quiet, |
| 297 | + debug=args.debug) |
| 298 | + |
| 299 | + svg = render_animated_svg(frames, steps, args.cols, args.rows, |
| 300 | + per_frame=args.per_frame, |
| 301 | + hold_last=args.hold_last) |
| 302 | + with open(args.out, "w") as f: |
| 303 | + f.write(svg) |
| 304 | + print(f"wrote {args.out} ({len(svg)} bytes, {len(frames)} frames)", |
| 305 | + file=sys.stderr) |
| 306 | + return 0 |
| 307 | + |
| 308 | + |
| 309 | +if __name__ == "__main__": |
| 310 | + sys.exit(main()) |
0 commit comments