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>
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
import json
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from spelunkai_recording.frames import FrameExtractionResult, extract_frames
|
|
from spelunkai_recording.session import session_paths
|
|
|
|
FFMPEG_AVAILABLE = shutil.which("ffmpeg") is not None
|
|
|
|
|
|
def test_is_in_sync_true_when_counts_match():
|
|
result = FrameExtractionResult(
|
|
frames_dir=Path("."), extracted_frame_count=3, manifest_frame_count=3, input_log_line_count=3,
|
|
)
|
|
assert result.is_in_sync
|
|
|
|
|
|
def test_is_in_sync_false_when_counts_differ():
|
|
result = FrameExtractionResult(
|
|
frames_dir=Path("."), extracted_frame_count=3, manifest_frame_count=3, input_log_line_count=2,
|
|
)
|
|
assert not result.is_in_sync
|
|
|
|
|
|
@pytest.mark.skipif(not FFMPEG_AVAILABLE, reason="ffmpeg not installed")
|
|
def test_extract_frames_matches_synthetic_video(tmp_path: Path):
|
|
paths = session_paths(tmp_path, "synthetic")
|
|
frame_count = 5
|
|
|
|
subprocess.run(
|
|
[
|
|
"ffmpeg", "-y",
|
|
"-f", "lavfi", "-i", "testsrc=size=64x64:rate=10",
|
|
"-frames:v", str(frame_count),
|
|
"-c:v", "libx264rgb", "-qp", "0",
|
|
str(paths.video),
|
|
],
|
|
check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
paths.manifest.write_text(
|
|
json.dumps({
|
|
"started_at": "2026-01-01T00:00:00+00:00",
|
|
"video_file": paths.video.name,
|
|
"input_log_file": paths.input_log.name,
|
|
"fps": 10,
|
|
"width": 64,
|
|
"height": 64,
|
|
"frame_count": frame_count,
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with paths.input_log.open("w", encoding="utf-8") as f:
|
|
for i in range(frame_count):
|
|
f.write(json.dumps({"frame": i, "t": i / 10, "keys": []}) + "\n")
|
|
|
|
frames_dir = tmp_path / "synthetic_frames"
|
|
result = extract_frames(paths, frames_dir)
|
|
|
|
assert result.is_in_sync
|
|
assert result.extracted_frame_count == frame_count
|
|
assert (frames_dir / "frame_000000.png").exists()
|
|
assert (frames_dir / f"frame_{frame_count - 1:06d}.png").exists()
|