From 6a00e14abe5386f2d3f48d04fc59b8e0f53359e0 Mon Sep 17 00:00:00 2001 From: abrichr Date: Fri, 21 Aug 2026 15:40:40 -0400 Subject: [PATCH] feat: bind each action to its exact retained screen frame The recorder now retains an exact wall-clock capture timestamp for every encoded frame in the MP4 timing box (captures timeline), validates it strictly on read, and preserves it across moov-atom rewrites. A new fail-closed extract_exact_frame decodes THE bound frame or raises: no nearest-frame substitute, no silent tolerance guess. Actions carry their bound screenshot_timestamp through conversion and every client-side merge (click, double-click, drag, type, shortcut, scroll, move), and Action.screenshot resolves that exact frame through CaptureSession.get_exact_frame. Image-only captures resolve the retained PNG row with the exact timestamp. Actions recorded before binding existed keep the legacy nearest-frame lookup. This closes the frame-binding gap for release evidence: a consumer can prove which pixels an action pointed at instead of trusting a 0.5s nearest-frame window. --- openadapt_capture/capture.py | 51 +++- openadapt_capture/events.py | 8 + openadapt_capture/processing.py | 28 +++ openadapt_capture/video.py | 152 ++++++++++-- tests/test_frame_binding.py | 399 ++++++++++++++++++++++++++++++++ tests/test_video.py | 19 +- 6 files changed, 628 insertions(+), 29 deletions(-) create mode 100644 tests/test_frame_binding.py diff --git a/openadapt_capture/capture.py b/openadapt_capture/capture.py index 75574b3..4c05142 100644 --- a/openadapt_capture/capture.py +++ b/openadapt_capture/capture.py @@ -5,6 +5,7 @@ from __future__ import annotations +import io from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Iterator @@ -112,6 +113,7 @@ def _convert_action_event(db_event) -> PydanticActionEvent: "structural_observation": _parse_structural_observation( getattr(db_event, "structural_observation", None) ), + "screenshot_timestamp": getattr(db_event, "screenshot_timestamp", None), } if db_event.name == "move": @@ -449,11 +451,22 @@ def structural_observation(self) -> StructuralObservation | None: @property def screenshot(self) -> "Image" | None: - """Get the screenshot at the time of this action. + """Get the exact retained screen frame this action is bound to. + + The action's ``screenshot_timestamp`` names the one frame the recorder + retained as this action's evidence; it is decoded exactly or this + raises. Actions recorded before frame binding existed fall back to a + nearest-frame lookup at the action timestamp. Returns: - PIL Image of the screen at action time, or None if not available. + PIL Image of the bound frame. + + Raises: + LookupError: If the bound frame cannot be resolved exactly. """ + bound = getattr(self.event, "screenshot_timestamp", None) + if bound is not None: + return self._capture.get_exact_frame(bound) return self._capture.get_frame_at(self.timestamp) @@ -739,6 +752,10 @@ def browser_event_count(self) -> int: def get_frame_at(self, timestamp: float, tolerance: float = 0.5) -> "Image" | None: """Get the screen frame closest to a timestamp. + This is the lenient legacy lookup. Evidence consumers should prefer + :meth:`get_exact_frame`, which resolves one exact retained frame and + fails closed when it cannot. + Args: timestamp: Unix timestamp. tolerance: Maximum time difference in seconds. @@ -764,6 +781,36 @@ def get_frame_at(self, timestamp: float, tolerance: float = 0.5) -> "Image" | No except Exception: return None + def get_exact_frame(self, capture_timestamp: float) -> "Image": + """Decode THE retained frame bound to this exact capture timestamp. + + Prefers the video's capture-timeline binding; for image-only captures + (no video stream) the retained PNG row with that exact timestamp is + the bound frame. + + Raises: + LookupError: If no frame was retained at exactly this timestamp, + or the binding metadata is unavailable (fail-closed). + """ + from PIL import Image + + video_path = self.video_path + if video_path is not None: + from openadapt_capture.video import extract_exact_frame + + return extract_exact_frame(video_path, capture_timestamp) + for screenshot in self._recording.screenshots: + if screenshot.timestamp == capture_timestamp: + if not screenshot.png_data: + raise LookupError( + f"the screenshot retained at {capture_timestamp!r} has no image data" + ) + return Image.open(io.BytesIO(screenshot.png_data)).convert("RGB") + raise LookupError( + f"no frame was retained at exactly {capture_timestamp!r} " + "(fail-closed; refusing a nearest-frame substitute)" + ) + def close(self) -> None: """Close the capture and release resources. diff --git a/openadapt_capture/events.py b/openadapt_capture/events.py index 57a7d3c..9c1362a 100644 --- a/openadapt_capture/events.py +++ b/openadapt_capture/events.py @@ -71,6 +71,14 @@ class ActionBaseEvent(BaseEvent): default=None, description="Versioned accessibility evidence observed at action time", ) + screenshot_timestamp: float | None = Field( + default=None, + description=( + "Exact capture timestamp of the retained screen frame this action " + "is bound to. Extract that exact frame instead of guessing a " + "nearest one." + ), + ) # ============================================================================= diff --git a/openadapt_capture/processing.py b/openadapt_capture/processing.py index 49daba5..557aaa4 100644 --- a/openadapt_capture/processing.py +++ b/openadapt_capture/processing.py @@ -90,6 +90,21 @@ def _first_structural_observation( return None +def _bound_screenshot_timestamp(events: list[ActionEvent]) -> float | None: + """Keep the frame bound where the merged action completed. + + The recorder binds an action to the screen frame retained when the action + was emitted (a click at button-up, a typed run at its last key), so the + merged event carries the LAST child's binding, not the first's. + """ + + bound: float | None = None + for event in events: + if event.screenshot_timestamp is not None: + bound = event.screenshot_timestamp + return bound + + # ============================================================================= # Event Processing Functions # ============================================================================= @@ -250,6 +265,9 @@ def flush_buffer() -> None: structural_observation=_first_structural_observation( list(keyboard_buffer) ), + screenshot_timestamp=_bound_screenshot_timestamp( + list(keyboard_buffer) + ), ) ) else: @@ -266,6 +284,7 @@ def flush_buffer() -> None: structural_observation=_first_structural_observation( list(keyboard_buffer) ), + screenshot_timestamp=_bound_screenshot_timestamp(list(keyboard_buffer)), ) result.append(type_event) @@ -328,6 +347,7 @@ def flush_buffer() -> None: structural_observation=_first_structural_observation( list(move_buffer) ), + screenshot_timestamp=_bound_screenshot_timestamp(list(move_buffer)), ) result.append(merged) @@ -380,6 +400,7 @@ def flush_buffer() -> None: structural_observation=_first_structural_observation( list(scroll_buffer) ), + screenshot_timestamp=_bound_screenshot_timestamp(list(scroll_buffer)), ) result.append(merged) @@ -487,6 +508,9 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: structural_observation=_first_structural_observation( [down, up, next_down, next_up] ), + screenshot_timestamp=_bound_screenshot_timestamp( + [down, up, next_down, next_up] + ), ) result.append(double_click) skip_timestamps.add(up.timestamp) @@ -504,6 +528,7 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: structural_observation=_first_structural_observation( [down, up] ), + screenshot_timestamp=_bound_screenshot_timestamp([down, up]), ) result.append(single_click) skip_timestamps.add(up.timestamp) @@ -571,6 +596,9 @@ def calculate_distance(x1: float, y1: float, x2: float, y2: float) -> float: structural_observation=_first_structural_observation( [down_event] + moves + [event] ), + screenshot_timestamp=_bound_screenshot_timestamp( + [down_event] + moves + [event] + ), ) result.append(drag) else: diff --git a/openadapt_capture/video.py b/openadapt_capture/video.py index f2f1e6d..e0418aa 100644 --- a/openadapt_capture/video.py +++ b/openadapt_capture/video.py @@ -88,30 +88,43 @@ def _append_timing_box( *, fps: Fraction, frames: list[tuple[int, float]], + captures: list[tuple[int, float]] | None = None, ) -> None: - """Append logical capture-frame timestamps in an ignored MP4 UUID box.""" - payload = json.dumps( - { - "schema": _TIMING_SCHEMA, - "fps": f"{fps.numerator}/{fps.denominator}", - "frames": [[index, timestamp] for index, timestamp in frames], - }, - separators=(",", ":"), - ).encode("utf-8") - box_size = 24 + len(payload) - if len(payload) > _MAX_TIMING_PAYLOAD_BYTES or box_size >= 2**32: + """Append logical capture-frame timestamps in an ignored MP4 UUID box. + + ``frames`` maps encoded frame index to stream-relative seconds. + ``captures`` optionally binds encoded frame indexes to the exact + wall-clock capture timestamps of the retained source frames, so a + consumer can decode THE frame an action was bound to instead of a + nearest-frame guess. + """ + payload: dict = { + "schema": _TIMING_SCHEMA, + "fps": f"{fps.numerator}/{fps.denominator}", + "frames": [[index, timestamp] for index, timestamp in frames], + } + if captures: + payload["captures"] = [[index, timestamp] for index, timestamp in captures] + serialized = json.dumps(payload, separators=(",", ":")).encode("utf-8") + box_size = 24 + len(serialized) + if len(serialized) > _MAX_TIMING_PAYLOAD_BYTES or box_size >= 2**32: raise FFmpegEncodingError("Video timing metadata exceeds its bounded MP4 box") with path.open("ab") as output: output.write(struct.pack(">I4s16s", box_size, b"uuid", _TIMING_BOX_UUID)) - output.write(payload) + output.write(serialized) output.flush() os.fsync(output.fileno()) def _read_timing_box( path: Path, -) -> tuple[Fraction, list[tuple[int, float]]] | None: - """Read OpenAdapt logical timestamps from top-level MP4 boxes, if present.""" +) -> tuple[Fraction, list[tuple[int, float]], list[tuple[int, float]] | None]: + """Read OpenAdapt logical timestamps from top-level MP4 boxes, if present. + + Returns ``(fps, frames, captures)``. ``captures`` is ``None`` for media + recorded before exact frame binding existed; consumers that require the + exact retained frame must treat that as unavailable (fail closed). + """ file_size = path.stat().st_size with path.open("rb") as source: offset = 0 @@ -181,7 +194,42 @@ def _read_timing_box( result.append((entry[0], timestamp)) previous_index = entry[0] previous_timestamp = timestamp - return fps, result + captures: list[tuple[int, float]] | None = None + raw_captures = payload.get("captures") + if raw_captures is not None: + captures = [] + capture_index = -1 + capture_timestamp = -1.0 + for entry in raw_captures: + if ( + not isinstance(entry, list) + or len(entry) != 2 + or not isinstance(entry[0], int) + ): + raise FFmpegEncodingError( + "Video timing metadata has an invalid capture binding" + ) + try: + bound_timestamp = float(entry[1]) + except (TypeError, ValueError) as exc: + raise FFmpegEncodingError( + "Video timing metadata has an invalid capture timestamp" + ) from exc + if entry[0] <= capture_index or not math.isfinite(bound_timestamp): + raise FFmpegEncodingError( + "Video timing metadata capture bindings are not ordered" + ) + # A duplicated first frame may legitimately repeat one + # capture timestamp at a new encoded index, so the + # timestamps are non-decreasing rather than strict. + if bound_timestamp < capture_timestamp: + raise FFmpegEncodingError( + "Video timing metadata capture bindings are not ordered" + ) + captures.append((entry[0], bound_timestamp)) + capture_index = entry[0] + capture_timestamp = bound_timestamp + return fps, result, captures offset += box_size return None @@ -681,6 +729,7 @@ def __init__( self._last_pts = -1 self._emitted_frames = 0 self._logical_frames: list[tuple[int, float]] = [] + self._capture_frames: list[tuple[int, float]] = [] self._closed = False self._lock = threading.Lock() @@ -837,8 +886,20 @@ def _enqueue_frames(self, frame: bytes, repetitions: int) -> None: "FFmpeg input pipe stopped accepting frames" + (f": {detail}" if detail else "") ) from exc - def stage_frame(self, image: "PILImage", pts: int) -> None: - """Stream one frame, filling PTS gaps deterministically without disk.""" + def stage_frame( + self, + image: "PILImage", + pts: int, + capture_timestamp: float | None = None, + ) -> None: + """Stream one frame, filling PTS gaps deterministically without disk. + + ``capture_timestamp`` is the exact wall-clock time the source frame was + captured. It is retained only for newly captured frames (gap fillers + repeat the previous frame and bind to no new evidence), so extraction + can later decode THE retained frame for an action instead of a + nearest-frame guess. + """ with self._lock: if self._closed: raise FFmpegEncodingError("Video stream is already closed") @@ -865,6 +926,13 @@ def stage_frame(self, image: "PILImage", pts: int) -> None: emitted = gap assert self._first_pts is not None self._logical_frames.append((encoded_index, (pts - self._first_pts) / fps)) + if capture_timestamp is not None: + bound_timestamp = float(capture_timestamp) + if not math.isfinite(bound_timestamp): + raise FFmpegEncodingError( + "Capture timestamp must be a finite wall-clock value" + ) + self._capture_frames.append((encoded_index, bound_timestamp)) self._emitted_frames += emitted self._last_frame = frame self._last_pts = pts @@ -931,6 +999,7 @@ def close(self) -> None: self.partial_path, fps=self.stream.average_rate, frames=self._logical_frames, + captures=self._capture_frames, ) _decode_first_frame_png( self.provision, @@ -1119,7 +1188,7 @@ def write_video_frame( pts = int(time_diff * float(video_stream.average_rate)) if pts <= last_pts: pts = last_pts + 1 - video_container.stage_frame(screenshot, pts) + video_container.stage_frame(screenshot, pts, capture_timestamp=timestamp) return pts @@ -1181,8 +1250,8 @@ def move_moov_atom( timeout=DEFAULT_PROCESS_TIMEOUT_SECONDS, ) if timing is not None: - fps, logical_frames = timing - _append_timing_box(output_path, fps=fps, frames=logical_frames) + fps, logical_frames, captures = timing + _append_timing_box(output_path, fps=fps, frames=logical_frames, captures=captures) if temp_file is not None: os.replace(temp_file, input_path) @@ -1341,6 +1410,49 @@ def extract_frame( )[0] +def extract_exact_frame( + video_path: str | Path, + capture_timestamp: float, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, +) -> "PILImage": + """Decode THE retained frame bound to this exact capture wall-clock timestamp. + + The binding comes from the timing box's ``captures`` timeline, which the + recorder populates with the source-frame timestamp at encode time. There + is deliberately no nearest-frame fallback: when the binding is absent + (older media, a stripped timing box) or the exact timestamp was never + bound, this raises instead of returning a frame an action never pointed + at. + """ + path = Path(video_path) + timing = _read_timing_box(path) + if timing is None: + raise LookupError( + f"{path}: no OpenAdapt timing metadata; " + "the exact retained frame cannot be resolved (fail-closed)" + ) + _, _, captures = timing + if not captures: + raise LookupError( + f"{path}: timing metadata predates exact frame binding; " + "the exact retained frame cannot be resolved (fail-closed)" + ) + matches = [index for index, bound in captures if bound == capture_timestamp] + if not matches: + nearest = min(captures, key=lambda item: abs(item[1] - capture_timestamp)) + raise LookupError( + f"{path}: no retained frame bound to {capture_timestamp!r} " + f"(nearest binding: index {nearest[0]} at {nearest[1]!r})" + ) + provision = resolve_ffmpeg( + ffmpeg_path or config.VIDEO_FFMPEG_PATH, + ffprobe_path or config.VIDEO_FFPROBE_PATH, + ) + return _extract_frame_index_png(path, matches[0], provision) + + def get_video_info( video_path: str | Path, *, diff --git a/tests/test_frame_binding.py b/tests/test_frame_binding.py new file mode 100644 index 0000000..5e30948 --- /dev/null +++ b/tests/test_frame_binding.py @@ -0,0 +1,399 @@ +"""Contracts for binding each action to its exact retained screen frame. + +An action's evidence frame must be THE frame the recorder retained when the +action completed, not a nearest-frame guess. These tests pin: + +- the video timing box retaining an exact capture-timestamp binding per + encoded frame (and rejecting malformed bindings), +- fail-closed exact extraction (no nearest-frame substitute), +- propagation of the bound timestamp through event merging, +- Action.screenshot resolving the bound frame or raising, never substituting. +""" + +from __future__ import annotations + +import json +import struct +from fractions import Fraction + +import pytest +from PIL import Image + +from openadapt_capture import video +from openadapt_capture.capture import Action +from openadapt_capture.events import ( + KeyDownEvent, + KeyTypeEvent, + MouseClickEvent, + MouseDownEvent, + MouseDragEvent, + MouseMoveEvent, + MouseUpEvent, +) +from openadapt_capture.processing import ( + detect_drag_events, + merge_consecutive_keyboard_events, + merge_consecutive_mouse_click_events, +) + +_TIMING_BOX_UUID = video._TIMING_BOX_UUID + + +# --------------------------------------------------------------------------- +# Timing box: exact capture bindings +# --------------------------------------------------------------------------- + + +def test_timing_box_round_trips_exact_capture_bindings(tmp_path): + path = tmp_path / "capture.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + frames = [(0, 0.0), (3, 3 / 24), (4, 4 / 24)] + captures = [(0, 1000.5), (3, 1001.25)] + video._append_timing_box( + path, + fps=Fraction(24), + frames=frames, + captures=captures, + ) + + fps, read_frames, read_captures = video._read_timing_box(path) + assert fps == Fraction(24) + assert read_frames == frames + # JSON float round-trip preserves the double exactly. + assert read_captures == captures + assert read_captures[1][1] == 1001.25 + + +def test_timing_box_without_bindings_reads_as_none(tmp_path): + """Media recorded before exact binding stays readable (captures=None).""" + path = tmp_path / "legacy.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box(path, fps=Fraction(24), frames=[(0, 0.0)]) + + _, frames, captures = video._read_timing_box(path) + assert frames == [(0, 0.0)] + assert captures is None + + +@pytest.mark.parametrize( + "captures", + [ + [(1, 1.0), (0, 2.0)], # indexes not increasing + [(0, 2.0), (1, 1.0)], # timestamps decreasing + [(0, "later")], # non-numeric timestamp + [("zero", 1.0)], # non-integer index + [(0, None)], # null timestamp + ], +) +def test_timing_box_rejects_malformed_capture_bindings(tmp_path, captures): + path = tmp_path / "capture.mp4" + payload = json.dumps( + { + "schema": video._TIMING_SCHEMA, + "fps": "24/1", + "frames": [[0, 0.0]], + "captures": captures, + }, + separators=(",", ":"), + ).encode("utf-8") + box_size = 24 + len(payload) + path.write_bytes(b"\x00\x00\x00\x08ftyp") + with path.open("ab") as output: + output.write(struct.pack(">I4s16s", box_size, b"uuid", _TIMING_BOX_UUID)) + output.write(payload) + + with pytest.raises(video.FFmpegEncodingError, match="capture"): + video._read_timing_box(path) + + +def test_timing_box_accepts_duplicated_first_frame_binding(tmp_path): + """The first frame is written twice; one capture ts may hold two indexes.""" + path = tmp_path / "capture.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box( + path, + fps=Fraction(24), + frames=[(0, 0.0), (1, 1 / 24)], + captures=[(0, 500.0), (1, 500.0)], + ) + _, _, captures = video._read_timing_box(path) + assert captures == [(0, 500.0), (1, 500.0)] + + +# --------------------------------------------------------------------------- +# extract_exact_frame: fail closed, never substitute +# --------------------------------------------------------------------------- + + +def test_extract_exact_frame_fails_closed_without_timing_metadata(tmp_path): + path = tmp_path / "bare.mp4" + path.write_bytes(b"not an openadapt recording") + with pytest.raises(LookupError, match="no OpenAdapt timing metadata"): + video.extract_exact_frame(path, 123.0) + + +def test_extract_exact_frame_fails_closed_on_legacy_bindings(tmp_path): + path = tmp_path / "legacy.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box(path, fps=Fraction(24), frames=[(0, 0.0)]) + with pytest.raises(LookupError, match="predates exact frame binding"): + video.extract_exact_frame(path, 123.0) + + +def test_extract_exact_frame_fails_closed_when_timestamp_unbound(tmp_path): + path = tmp_path / "capture.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box( + path, + fps=Fraction(24), + frames=[(0, 0.0)], + captures=[(0, 900.0)], + ) + with pytest.raises(LookupError, match="no retained frame bound to") as exc_info: + video.extract_exact_frame(path, 901.0) + assert "nearest binding" in str(exc_info.value) + + +def test_extract_exact_frame_decodes_the_bound_index(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + path = tmp_path / "capture.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box( + path, + fps=Fraction(24), + frames=[(0, 0.0), (1, 1 / 24)], + captures=[(0, 777.0), (1, 777.25)], + ) + + decoded = {} + + def fake_extract(video_path, frame_index, provision): + decoded["index"] = frame_index + return Image.new("RGB", (2, 1), "blue") + + monkeypatch.setattr(video, "_extract_frame_index_png", fake_extract) + frame = video.extract_exact_frame(path, 777.25, ffmpeg_path=executable) + assert decoded["index"] == 1 + assert frame.size == (2, 1) + + +def test_extract_exact_frame_prefers_the_first_duplicate_binding( + tmp_path, monkeypatch +): + """The duplicated first frame binds twice; decode its first index.""" + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + path = tmp_path / "capture.mp4" + path.write_bytes(b"\x00\x00\x00\x08ftyp") + video._append_timing_box( + path, + fps=Fraction(24), + frames=[(0, 0.0), (1, 1 / 24)], + captures=[(0, 500.0), (1, 500.0)], + ) + decoded = {} + monkeypatch.setattr( + video, + "_extract_frame_index_png", + lambda _path, index, _provision: decoded.setdefault("index", index), + ) + video.extract_exact_frame(path, 500.0, ffmpeg_path=executable) + assert decoded["index"] == 0 + + +# --------------------------------------------------------------------------- +# Encoder-side retention of capture bindings +# --------------------------------------------------------------------------- + + +class _FakeInput: + def __init__(self) -> None: + self.data = bytearray() + + def write(self, payload: bytes) -> int: + self.data.extend(payload) + return len(payload) + + def flush(self) -> None: + pass + + def close(self) -> None: + pass + + +class _FakeProcess: + def __init__(self, command: list[str]) -> None: + self.command = command + self.pipe = _FakeInput() + self.stdin = self.pipe + self.stderr_payload = b"" + self.returncode: int | None = None + self._killed = False + + def poll(self) -> int | None: + return self.returncode + + def wait(self, timeout: float | None = None) -> int: + self.returncode = -9 if self._killed else 0 + if self.returncode == 0: + from pathlib import Path + + Path(self.command[-1]).write_bytes(b"\x00\x00\x00\x08ftyp") + return self.returncode + + def kill(self) -> None: + self._killed = True + + +def _small_stream() -> video.FFmpegVideoStream: + return video.FFmpegVideoStream( + width=2, + height=1, + average_rate=Fraction(24), + pix_fmt="yuv420p", + codec="mpeg4", + muxer="mp4", + ) + + +def _install_fake_popen(monkeypatch, process: _FakeProcess) -> None: + def popen(command, **kwargs): + process.command = list(command) + kwargs["stderr"].write(process.stderr_payload) + return process + + monkeypatch.setattr(video.subprocess, "Popen", popen) + + +def _png_bytes(color: str = "black") -> bytes: + import io + + output = io.BytesIO() + Image.new("RGB", (2, 2), color).save(output, format="PNG") + return output.getvalue() + + +def _stage_with_fake_encoder(tmp_path, monkeypatch): + from pathlib import Path + + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _small_stream(), + video.FFmpegProvision(str(executable)), + ) + process = _FakeProcess(stage._encode_command()) + _install_fake_popen(monkeypatch, process) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_a, **_k: _png_bytes()) + return stage, Path(output) + + +def test_stage_frame_binds_only_newly_captured_frames(tmp_path, monkeypatch): + stage, output = _stage_with_fake_encoder(tmp_path, monkeypatch) + red = Image.new("RGB", (2, 1), "red") + blue = Image.new("RGB", (2, 1), "blue") + + # Frame at pts 27 fills three PTS slots (two gap fillers + itself). + stage.stage_frame(red, 24, capture_timestamp=111.0) + stage.stage_frame(blue, 27, capture_timestamp=113.5) + stage.close() + + _, logical, captures = video._read_timing_box(output) + assert [index for index, _ in logical] == [0, 3] + assert captures == [(0, 111.0), (3, 113.5)] + + +def test_stage_frame_refuses_a_non_finite_capture_timestamp(tmp_path, monkeypatch): + stage, _ = _stage_with_fake_encoder(tmp_path, monkeypatch) + red = Image.new("RGB", (2, 1), "red") + with pytest.raises(video.FFmpegEncodingError, match="finite wall-clock"): + stage.stage_frame(red, 24, capture_timestamp=float("nan")) + + +# --------------------------------------------------------------------------- +# Merge propagation of the bound frame +# --------------------------------------------------------------------------- + + +def _click_pair(down_ts=10.0, up_ts=10.05, down_bound=9.9, up_bound=10.0): + down = MouseDownEvent( + timestamp=down_ts, x=1.0, y=2.0, button="left", screenshot_timestamp=down_bound + ) + up = MouseUpEvent( + timestamp=up_ts, x=1.0, y=2.0, button="left", screenshot_timestamp=up_bound + ) + return down, up + + +def test_click_merge_keeps_the_up_childs_binding(): + down, up = _click_pair() + (merged,) = merge_consecutive_mouse_click_events([down, up]) + assert isinstance(merged, MouseClickEvent) + assert merged.screenshot_timestamp == 10.0 + + +def test_drag_merge_keeps_the_final_bindings(): + down = MouseDownEvent(timestamp=1.0, x=1.0, y=2.0, button="left", screenshot_timestamp=0.9) + move = MouseMoveEvent(timestamp=1.5, x=30.0, y=30.0, screenshot_timestamp=1.6) + up = MouseUpEvent(timestamp=2.0, x=30.0, y=30.0, button="left", screenshot_timestamp=2.0) + (drag,) = detect_drag_events([down, move, up]) + assert isinstance(drag, MouseDragEvent) + assert drag.screenshot_timestamp == 2.0 + + +def test_type_merge_keeps_the_last_key_binding(): + first = KeyDownEvent(timestamp=20.0, key_char="a", screenshot_timestamp=19.5) + second = KeyDownEvent(timestamp=21.0, key_char="b", screenshot_timestamp=20.75) + (merged,) = merge_consecutive_keyboard_events([first, second]) + assert isinstance(merged, KeyTypeEvent) + assert merged.text == "ab" + assert merged.screenshot_timestamp == 20.75 + + +def test_merge_leaves_legacy_children_unbound(): + down = MouseDownEvent(timestamp=1.0, x=0.0, y=0.0, button="left") + up = MouseUpEvent(timestamp=1.1, x=0.0, y=0.0, button="left") + (merged,) = merge_consecutive_mouse_click_events([down, up]) + assert merged.screenshot_timestamp is None + + +# --------------------------------------------------------------------------- +# Action.screenshot resolves the bound frame exactly +# --------------------------------------------------------------------------- + + +class _StubCapture: + def __init__(self): + self.exact_calls: list[float] = [] + self.lenient_calls: list[float] = [] + + def get_exact_frame(self, capture_timestamp: float) -> Image.Image: + self.exact_calls.append(capture_timestamp) + return Image.new("RGB", (1, 1), "red") + + def get_frame_at(self, timestamp: float) -> Image.Image: + self.lenient_calls.append(timestamp) + return Image.new("RGB", (1, 1), "blue") + + +def test_action_screenshot_uses_exact_binding(): + stub = _StubCapture() + _, up = _click_pair() + action = Action(event=up, _capture=stub) + image = action.screenshot + assert stub.exact_calls == [10.0] + assert stub.lenient_calls == [] + assert image.getpixel((0, 0)) == (255, 0, 0) + + +def test_action_screenshot_falls_back_for_legacy_events(): + stub = _StubCapture() + legacy = MouseUpEvent(timestamp=42.0, x=0.0, y=0.0, button="left") + action = Action(event=legacy, _capture=stub) + image = action.screenshot + assert stub.exact_calls == [] + assert stub.lenient_calls == [42.0] + assert image is not None diff --git a/tests/test_video.py b/tests/test_video.py index 4e6b010..91e7b0c 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -446,7 +446,7 @@ def test_successful_direct_encode_is_verified_and_atomically_promoted(tmp_path, stage.close() assert output.read_bytes().startswith(b"\x00\x00\x00\x08ftyp") - assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)]) + assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)], None) assert not stage.partial_path.exists() @@ -473,6 +473,7 @@ def test_direct_stream_normalizes_nonzero_initial_pts(tmp_path, monkeypatch): assert video._read_timing_box(output) == ( Fraction(24), [(0, 0.0), (3, 3 / 24)], + None, ) @@ -564,7 +565,7 @@ def test_direct_encode_retries_partial_pipe_writes(tmp_path, monkeypatch): stage.close() assert bytes(process.pipe.data) == frame.tobytes() - assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)]) + assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)], None) def test_direct_encode_worker_start_failure_reaps_process(tmp_path, monkeypatch): @@ -956,10 +957,16 @@ def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): [index / 24 for index in range(26)], abs=1e-6, ) - assert video._read_timing_box(output) == ( + fps, logical_frames, captures = video._read_timing_box(output) + assert (fps, logical_frames) == ( Fraction(24), [(0, 0.0), (24, 1.0), (25, 25 / 24)], ) + # The duplicated first frame and the finalized last frame each bind to + # their exact capture wall-clock timestamps. + assert captures == [(0, start), (24, start + 1), (25, start + 1)] + bound_frame = video.extract_exact_frame(output, start + 1, ffmpeg_path=executable) + assert bound_frame.getpixel((10, 10))[2] > bound_frame.getpixel((10, 10))[0] frame = video.extract_frame( output, 0.8, @@ -969,7 +976,5 @@ def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): assert frame.getpixel((10, 10))[2] > frame.getpixel((10, 10))[0] video.move_moov_atom(output, ffmpeg_path=executable) - assert video._read_timing_box(output) == ( - Fraction(24), - [(0, 0.0), (24, 1.0), (25, 25 / 24)], - ) + _, _, moved_captures = video._read_timing_box(output) + assert moved_captures == [(0, start), (24, start + 1), (25, start + 1)]