Implement recording tool: lossless capture + frame-synced input log

Add ffmpeg/x11grab-based video capture (libx264rgb, qp=0, true lossless
RGB) and an evdev-based keyboard state logger, orchestrated by a single
frame-tick loop so each JSONL input row lines up 1:1 with its video
frame. Unit-tested with fake keyboard/video components (no real device
or ffmpeg needed); real hardware capture still needs validation on the
recording PC.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

recording/README.md

@@ -7,12 +7,62 @@ Output must be easy to slice into individual frames for labeling.
 
 See CLAUDE.md §3.1 for full requirements.
 
-**Status:** not yet implemented.
+**Status:** core capture + input logging implemented; not yet run against a live
+recording session on real hardware (developed off the recording PC — see Testing
+below).
+
+## How it works
+
+- **Video** (`capture.py`): shells out to `ffmpeg -f x11grab ... -c:v libx264rgb -qp 0`
+  to record the screen region losslessly (true RGB, no chroma subsampling) at a fixed
+  framerate. Requires an **Xorg session** — `x11grab` does not work under Wayland, so
+  the recording PC must log in via "Ubuntu on Xorg".
+- **Input** (`input_logger.py`): reads a raw `/dev/input/eventX` keyboard device via
+  `evdev` directly (bypasses the window system entirely), so key state is captured
+  reliably even while the game holds exclusive fullscreen focus.
+- **Sync** (`session.py`): a single frame-tick loop, paced off one monotonic clock,
+  starts both video and keyboard logging together and writes one JSONL row per video
+  frame (`{"frame": i, "t": seconds, "keys": [...]}`) plus a `_manifest.json` with fps/
+  resolution/frame count. Because both are ticked from the same start time at the same
+  fixed rate, frame `i` in the log lines up with frame `i` of the video without
+  post-hoc alignment.
 
 ## Setup
 
 ```
 python -m venv .venv
 source .venv/bin/activate
-pip install -e .
+pip install -e ".[dev]"
 ```
+
+Reading `/dev/input/eventX` requires the `input` group (or root):
+
+```
+sudo usermod -aG input $USER   # then log out/in
+```
+
+## Usage
+
+```
+# find the keyboard device path
+spelunkai-record list-devices
+
+# record until Ctrl+C
+spelunkai-record record --input-device /dev/input/event3 --output-dir recordings
+
+# fixed-length session
+spelunkai-record record --input-device /dev/input/event3 --duration 120 --name run01
+```
+
+## Testing
+
+Unit tests (`tests/`) cover the frame-tick pacing, JSONL/manifest output, and the
+ffmpeg command construction using fake keyboard/video components — no real device or
+ffmpeg binary required, so they run anywhere (including WSL):
+
+```
+pytest
+```
+
+Actual capture (`VideoCapture`/`KeyboardState` against real hardware) still needs to be
+validated end-to-end on the real recording PC with Spelunky Classic HD running.

recording/pyproject.toml

@@ -3,7 +3,15 @@ name = "spelunkai-recording"
 version = "0.0.0"
 description = "SpelunkAI Recording Tool: gameplay video + input log capture"
 requires-python = ">=3.10"
-dependencies = []
+dependencies = [
+    "evdev>=1.7",
+]
+
+[project.optional-dependencies]
+dev = ["pytest>=8"]
+
+[project.scripts]
+spelunkai-record = "spelunkai_recording.cli:main"
 
 [build-system]
 requires = ["setuptools>=68"]

recording/src/spelunkai_recording/__main__.py

@@ -0,0 +1,4 @@
+from .cli import main
+
+if __name__ == "__main__":
+    main()

recording/src/spelunkai_recording/capture.py

@@ -0,0 +1,69 @@
+"""ffmpeg-based lossless screen capture."""
+from __future__ import annotations
+
+import shutil
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+
+@dataclass(frozen=True)
+class CaptureConfig:
+    display: str = ":0.0"
+    width: int = 1280
+    height: int = 720
+    offset_x: int = 0
+    offset_y: int = 0
+    fps: int = 30
+
+
+class VideoCapture:
+    """Wraps ffmpeg to record an X11 screen region to a lossless MP4.
+
+    Requires an Xorg session (x11grab does not work under Wayland) and ffmpeg
+    on PATH. Uses libx264rgb at qp=0 for true lossless RGB output with no
+    chroma subsampling, so recorded frames match the game's exact pixels.
+    """
+
+    def __init__(self, config: CaptureConfig, output_path: Path):
+        self.config = config
+        self.output_path = output_path
+        self._process: Optional[subprocess.Popen] = None
+
+    def build_command(self) -> list[str]:
+        cfg = self.config
+        geometry = f"{cfg.display}+{cfg.offset_x},{cfg.offset_y}"
+        return [
+            "ffmpeg", "-y",
+            "-f", "x11grab",
+            "-video_size", f"{cfg.width}x{cfg.height}",
+            "-framerate", str(cfg.fps),
+            "-i", geometry,
+            "-an",
+            "-c:v", "libx264rgb",
+            "-qp", "0",
+            "-preset", "ultrafast",
+            str(self.output_path),
+        ]
+
+    def start(self) -> None:
+        if shutil.which("ffmpeg") is None:
+            raise RuntimeError("ffmpeg not found on PATH")
+        self.output_path.parent.mkdir(parents=True, exist_ok=True)
+        self._process = subprocess.Popen(
+            self.build_command(),
+            stdin=subprocess.PIPE,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+
+    def stop(self) -> None:
+        if self._process is None:
+            return
+        try:
+            self._process.communicate(input=b"q", timeout=15)
+        except subprocess.TimeoutExpired:
+            self._process.kill()
+            self._process.communicate()
+        self._process = None

recording/src/spelunkai_recording/cli.py

@@ -0,0 +1,76 @@
+"""Command-line entry point for the recording tool."""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime
+from pathlib import Path
+
+from .capture import CaptureConfig, VideoCapture
+from .devices import list_keyboard_devices
+from .input_logger import KeyboardState
+from .session import RecordingSession, session_paths
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        prog="spelunkai-record",
+        description="Record Spelunky Classic HD gameplay as a lossless MP4 with a frame-synced input log.",
+    )
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    record = subparsers.add_parser("record", help="Start a recording session")
+    record.add_argument("--output-dir", type=Path, default=Path("recordings"))
+    record.add_argument(
+        "--input-device", required=True,
+        help="Keyboard device path, e.g. /dev/input/event3 (see list-devices)",
+    )
+    record.add_argument("--display", default=":0.0")
+    record.add_argument("--width", type=int, default=1280)
+    record.add_argument("--height", type=int, default=720)
+    record.add_argument("--offset-x", type=int, default=0)
+    record.add_argument("--offset-y", type=int, default=0)
+    record.add_argument("--fps", type=int, default=30)
+    record.add_argument(
+        "--duration", type=float, default=None,
+        help="Stop after N seconds (default: run until Ctrl+C)",
+    )
+    record.add_argument(
+        "--name", default=None,
+        help="Session name, used as file prefix (default: timestamp)",
+    )
+
+    subparsers.add_parser("list-devices", help="List input devices that look like keyboards")
+
+    args = parser.parse_args()
+
+    if args.command == "list-devices":
+        for path, name in list_keyboard_devices():
+            print(f"{path}\t{name}")
+        return
+
+    session_name = args.name or datetime.now().strftime("%Y%m%d_%H%M%S")
+    paths = session_paths(args.output_dir, session_name)
+    config = CaptureConfig(
+        display=args.display, width=args.width, height=args.height,
+        offset_x=args.offset_x, offset_y=args.offset_y, fps=args.fps,
+    )
+    session = RecordingSession(
+        paths=paths,
+        keyboard=KeyboardState(args.input_device),
+        video=VideoCapture(config, paths.video),
+        fps=args.fps,
+        width=args.width,
+        height=args.height,
+    )
+
+    print(f"Recording to {paths.video} (Ctrl+C to stop)")
+    frame_count = None
+    try:
+        frame_count = session.run(duration_s=args.duration)
+    except KeyboardInterrupt:
+        pass
+    print(f"Stopped. Frames logged: {frame_count if frame_count is not None else 'interrupted'}")
+
+
+if __name__ == "__main__":
+    main()

recording/src/spelunkai_recording/devices.py

@@ -0,0 +1,13 @@
+"""Enumerate raw input devices via evdev to find the keyboard to record from."""
+from __future__ import annotations
+
+from evdev import InputDevice, ecodes, list_devices
+
+
+def list_keyboard_devices() -> list[tuple[str, str]]:
+    keyboards = []
+    for path in list_devices():
+        device = InputDevice(path)
+        if ecodes.KEY_A in device.capabilities().get(ecodes.EV_KEY, []):
+            keyboards.append((device.path, device.name))
+    return keyboards

recording/src/spelunkai_recording/input_logger.py

@@ -0,0 +1,60 @@
+"""Tracks currently-pressed keyboard keys by reading a raw evdev device.
+
+Reads the raw /dev/input device directly instead of hooking the X11/Wayland
+session, so key state is captured reliably even when the game holds
+exclusive fullscreen input focus.
+"""
+from __future__ import annotations
+
+import select
+import threading
+from typing import Optional
+
+from evdev import InputDevice, categorize, ecodes
+
+
+class KeyboardState:
+    def __init__(self, device_path: str):
+        self.device_path = device_path
+        self._device: Optional[InputDevice] = None
+        self._pressed: set[str] = set()
+        self._lock = threading.Lock()
+        self._thread: Optional[threading.Thread] = None
+        self._stop_event = threading.Event()
+
+    def start(self) -> None:
+        self._device = InputDevice(self.device_path)
+        self._stop_event.clear()
+        self._thread = threading.Thread(target=self._listen, daemon=True)
+        self._thread.start()
+
+    def stop(self) -> None:
+        self._stop_event.set()
+        if self._thread is not None:
+            self._thread.join(timeout=2)
+            self._thread = None
+        if self._device is not None:
+            self._device.close()
+            self._device = None
+
+    def snapshot(self) -> list[str]:
+        with self._lock:
+            return sorted(self._pressed)
+
+    def _listen(self) -> None:
+        assert self._device is not None
+        while not self._stop_event.is_set():
+            ready, _, _ = select.select([self._device.fd], [], [], 0.5)
+            if not ready:
+                continue
+            for event in self._device.read():
+                if event.type != ecodes.EV_KEY:
+                    continue
+                key_event = categorize(event)
+                keycode = key_event.keycode
+                name = keycode if isinstance(keycode, str) else keycode[0]
+                with self._lock:
+                    if key_event.keystate == key_event.key_down:
+                        self._pressed.add(name)
+                    elif key_event.keystate == key_event.key_up:
+                        self._pressed.discard(name)

recording/src/spelunkai_recording/session.py

@@ -0,0 +1,99 @@
+"""Orchestrates a recording session: video capture + a frame-synced input log.
+
+Both the video (fixed framerate via ffmpeg) and the input log (one row per
+frame, ticked against the same wall clock from the same start time) are
+paced off the same monotonic clock, so `frame N` in the log lines up with
+frame N of the video without needing post-hoc alignment.
+"""
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional, Protocol
+
+
+class VideoCaptureProtocol(Protocol):
+    def start(self) -> None: ...
+    def stop(self) -> None: ...
+
+
+class KeyboardStateProtocol(Protocol):
+    def start(self) -> None: ...
+    def stop(self) -> None: ...
+    def snapshot(self) -> list[str]: ...
+
+
+@dataclass
+class RecordingPaths:
+    video: Path
+    input_log: Path
+    manifest: Path
+
+
+def session_paths(output_dir: Path, session_name: str) -> RecordingPaths:
+    return RecordingPaths(
+        video=output_dir / f"{session_name}.mp4",
+        input_log=output_dir / f"{session_name}_input.jsonl",
+        manifest=output_dir / f"{session_name}_manifest.json",
+    )
+
+
+class RecordingSession:
+    def __init__(
+        self,
+        paths: RecordingPaths,
+        keyboard: KeyboardStateProtocol,
+        video: VideoCaptureProtocol,
+        fps: int,
+        width: int,
+        height: int,
+    ):
+        self.paths = paths
+        self.keyboard = keyboard
+        self.video = video
+        self.fps = fps
+        self.width = width
+        self.height = height
+
+    def run(self, duration_s: Optional[float] = None) -> int:
+        self.paths.video.parent.mkdir(parents=True, exist_ok=True)
+        start_wall = datetime.now(timezone.utc)
+        self.keyboard.start()
+        self.video.start()
+        start = time.monotonic()
+        frame_period = 1.0 / self.fps
+        frame_index = 0
+        try:
+            with self.paths.input_log.open("w", encoding="utf-8") as log_file:
+                while duration_s is None or (time.monotonic() - start) < duration_s:
+                    target = start + frame_index * frame_period
+                    now = time.monotonic()
+                    if now < target:
+                        time.sleep(target - now)
+                    row = {
+                        "frame": frame_index,
+                        "t": round(time.monotonic() - start, 6),
+                        "keys": self.keyboard.snapshot(),
+                    }
+                    log_file.write(json.dumps(row) + "\n")
+                    frame_index += 1
+        finally:
+            self.video.stop()
+            self.keyboard.stop()
+            self._write_manifest(start_wall, frame_index)
+        return frame_index
+
+    def _write_manifest(self, start_wall: datetime, frame_count: int) -> None:
+        manifest = {
+            "started_at": start_wall.isoformat(),
+            "video_file": self.paths.video.name,
+            "input_log_file": self.paths.input_log.name,
+            "fps": self.fps,
+            "width": self.width,
+            "height": self.height,
+            "frame_count": frame_count,
+        }
+        self.paths.manifest.write_text(json.dumps(manifest, indent=2), encoding="utf-8")

recording/tests/test_capture.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+from spelunkai_recording.capture import CaptureConfig, VideoCapture
+
+
+def test_build_command_includes_resolution_offset_and_output_path():
+    config = CaptureConfig(display=":1", width=800, height=600, offset_x=10, offset_y=20, fps=25)
+    capture = VideoCapture(config, Path("/tmp/out.mp4"))
+
+    cmd = capture.build_command()
+
+    assert "800x600" in cmd
+    assert ":1+10,20" in cmd
+    assert "25" in cmd
+    assert cmd[-1] == "/tmp/out.mp4"
+    assert "libx264rgb" in cmd
+    assert "0" in cmd  # -qp 0 (lossless)

recording/tests/test_session.py

@@ -0,0 +1,60 @@
+import json
+from pathlib import Path
+
+from spelunkai_recording.session import RecordingSession, session_paths
+
+
+class FakeKeyboard:
+    def __init__(self, keys_sequence):
+        self._keys_sequence = keys_sequence
+        self._i = 0
+        self.started = False
+        self.stopped = False
+
+    def start(self):
+        self.started = True
+
+    def stop(self):
+        self.stopped = True
+
+    def snapshot(self):
+        keys = self._keys_sequence[min(self._i, len(self._keys_sequence) - 1)]
+        self._i += 1
+        return keys
+
+
+class FakeVideo:
+    def __init__(self):
+        self.started = False
+        self.stopped = False
+
+    def start(self):
+        self.started = True
+
+    def stop(self):
+        self.stopped = True
+
+
+def test_session_logs_one_row_per_frame_and_writes_manifest(tmp_path: Path):
+    paths = session_paths(tmp_path, "test-session")
+    keyboard = FakeKeyboard([["LEFT"], [], ["RIGHT", "KEY_Z"]])
+    video = FakeVideo()
+    session = RecordingSession(paths=paths, keyboard=keyboard, video=video, fps=30, width=1280, height=720)
+
+    frame_count = session.run(duration_s=0.1)
+
+    assert frame_count >= 1
+    assert keyboard.started and keyboard.stopped
+    assert video.started and video.stopped
+
+    lines = paths.input_log.read_text(encoding="utf-8").splitlines()
+    assert len(lines) == frame_count
+
+    first_row = json.loads(lines[0])
+    assert first_row["frame"] == 0
+    assert first_row["keys"] == ["LEFT"]
+
+    manifest = json.loads(paths.manifest.read_text(encoding="utf-8"))
+    assert manifest["fps"] == 30
+    assert manifest["frame_count"] == frame_count
+    assert manifest["video_file"] == paths.video.name