diff --git a/packages/capture/openadapt_capture/recorder.py b/packages/capture/openadapt_capture/recorder.py index ca8e71c670..131ef71748 100644 --- a/packages/capture/openadapt_capture/recorder.py +++ b/packages/capture/openadapt_capture/recorder.py @@ -1792,6 +1792,14 @@ def write_video_event( raise ValueError( f"video frame {screenshot_image.size} differs from fixed stream {stream_size}" ) + if "last_frame" not in kwargs: + # The encoded stream starts at its first frame, and frame lookups map + # a wall-clock time to video time by subtracting video_start_time. A + # source that only delivers frames when the screen changes (macOS + # ScreenCaptureKit) can send its first frame long after the writer + # opened, so anchor the video start to that first frame instead. + video_start_timestamp = screenshot_timestamp + crud.update_video_start_time(db, recording_timestamp, video_start_timestamp) force_key_frame = last_pts == 0 # ensure that the first frame is available (otherwise occasionally it is not) # TODO: why isn't force_key_frame sufficient? diff --git a/packages/capture/tests/test_video_start_anchor.py b/packages/capture/tests/test_video_start_anchor.py new file mode 100644 index 0000000000..c35e36f25e --- /dev/null +++ b/packages/capture/tests/test_video_start_anchor.py @@ -0,0 +1,146 @@ +"""The video start time is anchored to the first encoded frame. + +The encoded stream begins at its first frame, and frame lookups turn a +wall-clock time into video time by subtracting ``video_start_time``. When the +first frame arrives long after the writer opened (macOS ScreenCaptureKit only +delivers a frame when the screen changes), the writer-open time put every +lookup that many seconds past the end of the video. +""" + +from __future__ import annotations + +import shutil +from types import SimpleNamespace + +import pytest +import sqlalchemy as sa +from PIL import Image + +from openadapt_capture import db, recorder, utils, video +from openadapt_capture.db import crud +from openadapt_capture.db.models import Recording +from openadapt_capture.extensions import synchronized_queue as sq + + +@pytest.fixture(autouse=True) +def _clock(): + utils.set_start_time() + + +def _screen_event(timestamp: float, color: str, ordinal: int) -> SimpleNamespace: + return SimpleNamespace( + type="screen/video", + data=Image.new("RGB", (64, 48), color), + timestamp=timestamp, + source_ordinal=ordinal, + ) + + +def test_first_frame_anchors_the_video_start(monkeypatch): + writes = [] + updates = [] + monkeypatch.setattr( + recorder.video, + "write_video_frame", + lambda _c, _s, _img, ts, start, last_pts, *_a, **_k: ( + writes.append((ts, start)) or last_pts + 1 + ), + ) + monkeypatch.setattr( + recorder.crud, + "update_video_start_time", + lambda _db, _rec, start: updates.append(start), + ) + perf_q = sq.SynchronizedQueue() + state = { + "video_container": None, + "video_stream": SimpleNamespace(width=64, height=48), + "video_start_timestamp": 100.0, + "last_pts": 0, + "video_file_path": "unused.mp4", + } + rec = SimpleNamespace(id=1, timestamp=99.0) + + state = recorder.write_video_event(None, rec, _screen_event(115.0, "red", 1), perf_q, **state) + assert updates == [115.0] + assert state["video_start_timestamp"] == 115.0 + assert {start for _, start in writes} == {115.0} + + state = recorder.write_video_event(None, rec, _screen_event(115.5, "blue", 2), perf_q, **state) + assert updates == [115.0] + assert writes[-1] == (115.5, 115.0) + assert state["video_start_timestamp"] == 115.0 + + +def _real_ffmpeg_mpeg4() -> str | None: + executable = shutil.which("ffmpeg") + if not executable: + return None + try: + provision = video.FFmpegProvision(executable, source="test PATH") + if "mpeg4" not in video.available_encoders(provision): + return None + except (video.FFmpegEncodingError, OSError): + return None + return executable + + +@pytest.mark.skipif( + _real_ffmpeg_mpeg4() is None, + reason="opt-in real FFmpeg/mpeg4 executable is not available", +) +def test_late_first_frame_is_found_at_its_wall_clock_time(tmp_path): + executable = _real_ffmpeg_mpeg4() + db_path = tmp_path / "recording.db" + engine, Session = db.create_db(str(db_path)) + session = Session() + recording = crud.insert_recording( + session, + { + "timestamp": 4.0, + "monitor_width": 64, + "monitor_height": 48, + "platform": "test", + "task_description": "late first frame", + }, + ) + detached = SimpleNamespace(id=recording.id, timestamp=recording.timestamp) + provision = video.FFmpegProvision( + executable, + codec="mpeg4", + pixel_format="yuv420p", + muxer="mp4", + source="test PATH", + ) + state = recorder.video_pre_callback( + session, + detached, + video_dir=str(tmp_path), + frame_size=(64, 48), + provision=provision, + timeout_seconds=60.0, + ) + opened = state["video_start_timestamp"] + perf_q = sq.SynchronizedQueue() + red_ts = opened + 15.0 + blue_ts = opened + 16.0 + state = recorder.write_video_event( + session, detached, _screen_event(red_ts, "red", 1), perf_q, **state + ) + state = recorder.write_video_event( + session, detached, _screen_event(blue_ts, "blue", 2), perf_q, **state + ) + recorder.video_post_callback(state) + + stored = session.execute( + sa.select(Recording.video_start_time).where(Recording.id == recording.id) + ).scalar_one() + assert stored == red_ts + for ts, channel in ((red_ts, 0), (blue_ts, 2)): + frame = video.extract_frame(state["video_file_path"], ts - stored, ffmpeg_path=executable) + pixel = frame.getpixel((10, 10)) + assert pixel[channel] > 200, (ts - opened, pixel) + while not perf_q.empty(): + perf_q.get() + session.close() + engine.dispose()