Jonas
2989a0814c
Slice a recorded session's lossless video into individual frame PNGs (ffmpeg -vsync 0, no drop/dup) via a new `extract-frames` subcommand, with frame count cross-checked against the manifest and input log so any capture-rate drift surfaces immediately instead of silently misaligning frames and logged input later. Includes a real end-to-end test against an ffmpeg-generated synthetic video (skipped when ffmpeg isn't installed, e.g. bare WSL). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""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 .frames import extract_frames
|
|
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")
|
|
|
|
extract = subparsers.add_parser(
|
|
"extract-frames",
|
|
help="Slice a recorded session's video into individual frame images",
|
|
)
|
|
extract.add_argument(
|
|
"--output-dir", type=Path, default=Path("recordings"),
|
|
help="Directory containing the session's video/log/manifest",
|
|
)
|
|
extract.add_argument("--name", required=True, help="Session name (file prefix used during recording)")
|
|
extract.add_argument(
|
|
"--frames-dir", type=Path, default=None,
|
|
help="Where to write frame images (default: <output-dir>/<name>_frames)",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "list-devices":
|
|
_run_list_devices()
|
|
elif args.command == "record":
|
|
_run_record(args)
|
|
elif args.command == "extract-frames":
|
|
_run_extract_frames(args)
|
|
|
|
|
|
def _run_list_devices() -> None:
|
|
for path, name in list_keyboard_devices():
|
|
print(f"{path}\t{name}")
|
|
|
|
|
|
def _run_record(args: argparse.Namespace) -> None:
|
|
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'}")
|
|
|
|
|
|
def _run_extract_frames(args: argparse.Namespace) -> None:
|
|
paths = session_paths(args.output_dir, args.name)
|
|
frames_dir = args.frames_dir or (args.output_dir / f"{args.name}_frames")
|
|
result = extract_frames(paths, frames_dir)
|
|
print(f"Extracted {result.extracted_frame_count} frames to {result.frames_dir}")
|
|
if not result.is_in_sync:
|
|
print(
|
|
f"WARNING: frame count mismatch (video={result.extracted_frame_count}, "
|
|
f"manifest={result.manifest_frame_count}, input_log={result.input_log_line_count}). "
|
|
"Video capture may have dropped or duplicated frames relative to the input log."
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|