From 2aa1f211ad21d1f1fcec98ada5b4f53da481f7ea Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 23 Jul 2026 18:00:11 -0700 Subject: [PATCH] Stream capture frames directly to FFmpeg --- README.md | 18 +- docs/DESIGN.md | 27 +- openadapt_capture/recorder.py | 2 +- openadapt_capture/video.py | 468 +++++++++++++++++++++++++++------- tests/test_video.py | 442 +++++++++++++++++++++++++++----- 5 files changed, 770 insertions(+), 187 deletions(-) diff --git a/README.md b/README.md index a8e5340..bd05684 100644 --- a/README.md +++ b/README.md @@ -120,17 +120,21 @@ my-capture/ └── profiling.json ``` -Video remains the default evidence format. Capture stages exact-timestamp, -lossless frames during the session, asks a separately provisioned FFmpeg -executable to encode them, verifies the resulting MP4, and promotes it -atomically. Capture never downloads, bundles, or links FFmpeg/PyAV. Set +Video remains the default evidence format. Capture streams in-memory RGB frames +directly to a separately provisioned FFmpeg executable while recording. Missing +integer PTS slots reuse the preceding frame, so encoding is deterministic and +independent of scheduler or queue latency. A compact MP4 metadata box retains +the logical capture-frame timestamps used by nearest-frame extraction. Capture +then verifies and atomically promotes the MP4; no intermediate screenshot +sequence is written. Capture never downloads, bundles, or links FFmpeg/PyAV. Set `OPENADAPT_FFMPEG_PATH`, pass `Recorder(ffmpeg_path=...)`, use Desktop's user-data `ffmpeg.json` provision manifest, or place `ffmpeg` and `ffprobe` on `PATH`. Recording performs a real encode-and-decode probe and refuses before input listeners start if the selected executable, codec, or PNG verification -path is unavailable. A minimal managed runtime must provide the selected video -encoder, MP4 demuxing/muxing, PNG decoding/encoding, the `image2pipe` muxer, and -the `select` video filter; Desktop provisions and probes that exact closure. +path is unavailable. A minimal managed runtime must provide raw-video input +through a pipe, the selected video encoder, MP4 demuxing/muxing, PNG +decoding/encoding, the `image2pipe` muxer, and the `select` video filter; +Desktop provisions and probes that exact closure. ## Window-scoped recording diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 3d20be5..3cb762c 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -167,16 +167,20 @@ class Capture: ### Video Encoding Capture keeps its video-first behavior without importing PyAV. During a -recording it stages numeric-name lossless PNG frames plus exact integer PTS. -Finalization writes a safe `ffconcat` manifest, invokes an externally -provisioned FFmpeg argument vector with bounded execution, verifies the output -by decoding a PNG frame, and atomically promotes it. Successful finalization -deletes the transient stage; failure retains it for recovery and never reports -an MP4 as complete. A sibling or explicitly configured `ffprobe` preserves -nearest-frame extraction and metadata inspection without linking codec -libraries into Capture. The real preflight exercises the selected encoder, -MP4, PNG through `image2pipe`, and the `select` filter before any input listener -starts. +recording it streams in-memory RGB frames directly into an externally +provisioned FFmpeg process. Missing integer PTS slots reuse the preceding RGB +frame, producing a deterministic constant-rate stream without depending on +wall-clock arrival time. A compact ignored MP4 UUID box maps logical capture +timestamps to their decoded-frame indexes, preserving the existing +nearest-frame contract without an image sidecar. Finalization closes the +bounded process, appends that metadata, verifies the output by decoding a PNG +frame, and atomically promotes it. No intermediate screenshot sequence or +`ffconcat` manifest is written. Failure retains only an explicitly incomplete +partial MP4 and never reports it as complete. A sibling or explicitly +configured `ffprobe` supports legacy nearest-frame extraction and metadata +inspection without linking codec libraries into Capture. The real preflight +exercises raw-video input through a pipe, the selected encoder, MP4, PNG +through `image2pipe`, and the `select` filter before any input listener starts. ```python codec = None # Probe platform encoders, then portable mpeg4 fallback @@ -209,7 +213,8 @@ capture_abc123/ **Decision:** Default to video mode. - More storage-efficient than permanent individual screenshots -- Exact timestamp alignment is preserved by staged-frame PTS and VFR durations +- Exact timestamp alignment is preserved by deterministic PTS-gap filling and + logical timestamps embedded in the MP4 - Screenshots can be extracted from video when needed - Individual screenshots remain optional for debugging frame alignment diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 9d59187..8843529 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -770,7 +770,7 @@ def write_video_event( recording_timestamp: The timestamp of the recording. event: A screen event to be written. perf_q: A queue for collecting performance data. - video_container: The external-encoder staging container. + video_container: The direct external-encoder stream. video_stream: The configured external-encoder stream. video_start_timestamp (float): The base timestamp from which the video recording started. diff --git a/openadapt_capture/video.py b/openadapt_capture/video.py index 9e7acbe..dd41f49 100644 --- a/openadapt_capture/video.py +++ b/openadapt_capture/video.py @@ -1,19 +1,21 @@ """Video capture through a separately provisioned FFmpeg executable. -Capture itself never downloads or bundles FFmpeg. Frames are staged losslessly -with exact presentation timestamps, encoded transactionally at finalization, -verified, and only then promoted to the public MP4 path. A failed encode retains -the staging directory for diagnosis/recovery and never leaves a false-success -output file. +Capture itself never downloads or bundles FFmpeg. In-memory RGB frames stream +directly into the encoder process and become compact video while recording. +The finished output is verified and atomically promoted; an encoder failure is +reported and never leaves a false-success public MP4. """ from __future__ import annotations import io import json +import math import os +import queue import re import shutil +import struct import subprocess import sys import tempfile @@ -22,7 +24,7 @@ from dataclasses import dataclass from fractions import Fraction from pathlib import Path -from typing import TYPE_CHECKING, Sequence +from typing import TYPE_CHECKING, BinaryIO, Sequence from loguru import logger from PIL import Image @@ -41,8 +43,12 @@ DEFAULT_PROCESS_TIMEOUT_SECONDS = 900.0 PROBE_TIMEOUT_SECONDS = 10.0 EXTRACT_TIMEOUT_SECONDS = 120.0 +FRAME_WRITE_TIMEOUT_SECONDS = 30.0 _ENCODER_LINE = re.compile(r"^\s*V\S*\s+(?P\S+)\s", re.MULTILINE) _OPTION_TOKEN = re.compile(r"^[A-Za-z0-9_.-]+$") +_TIMING_BOX_UUID = uuid.UUID("d8e90f06-20b4-4e0c-b449-4f70656e4164").bytes +_TIMING_SCHEMA = "openadapt.capture-video-timing/v1" +_MAX_TIMING_PAYLOAD_BYTES = 16 * 1024 * 1024 class FFmpegUnavailableError(RuntimeError): @@ -50,7 +56,7 @@ class FFmpegUnavailableError(RuntimeError): class FFmpegEncodingError(RuntimeError): - """FFmpeg failed to encode or verify a staged recording.""" + """FFmpeg failed to encode or verify a recording.""" @dataclass(frozen=True) @@ -77,10 +83,107 @@ class FFmpegVideoStream: muxer: str -@dataclass(frozen=True) -class _StagedFrame: - path: Path - pts: int +def _append_timing_box( + path: Path, + *, + fps: Fraction, + frames: list[tuple[int, float]], +) -> 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: + 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.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.""" + file_size = path.stat().st_size + with path.open("rb") as source: + offset = 0 + while offset + 8 <= file_size: + source.seek(offset) + header = source.read(8) + size32, box_type = struct.unpack(">I4s", header) + header_size = 8 + if size32 == 1: + extended = source.read(8) + if len(extended) != 8: + return None + box_size = struct.unpack(">Q", extended)[0] + header_size = 16 + elif size32 == 0: + box_size = file_size - offset + else: + box_size = size32 + if box_size < header_size or offset + box_size > file_size: + return None + if box_type == b"uuid" and box_size >= header_size + 16: + user_type = source.read(16) + if user_type == _TIMING_BOX_UUID: + payload_size = box_size - header_size - 16 + if payload_size > _MAX_TIMING_PAYLOAD_BYTES: + raise FFmpegEncodingError("Video timing metadata exceeds its read bound") + try: + payload = json.loads(source.read(payload_size).decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise FFmpegEncodingError("Video timing metadata is invalid") from exc + if not isinstance(payload, dict) or payload.get("schema") != _TIMING_SCHEMA: + raise FFmpegEncodingError("Video timing metadata has an unknown schema") + try: + fps = Fraction(payload["fps"]) + except (KeyError, TypeError, ValueError, ZeroDivisionError) as exc: + raise FFmpegEncodingError( + "Video timing metadata has an invalid frame rate" + ) from exc + if fps <= 0: + raise FFmpegEncodingError( + "Video timing metadata has a non-positive frame rate" + ) + result: list[tuple[int, float]] = [] + previous_index = -1 + previous_timestamp = -1.0 + for entry in payload.get("frames", []): + if ( + not isinstance(entry, list) + or len(entry) != 2 + or not isinstance(entry[0], int) + ): + raise FFmpegEncodingError("Video timing metadata has an invalid frame") + try: + timestamp = float(entry[1]) + except (TypeError, ValueError) as exc: + raise FFmpegEncodingError( + "Video timing metadata has an invalid timestamp" + ) from exc + if ( + entry[0] <= previous_index + or not math.isfinite(timestamp) + or timestamp < previous_timestamp + ): + raise FFmpegEncodingError( + "Video timing metadata is not strictly ordered" + ) + result.append((entry[0], timestamp)) + previous_index = entry[0] + previous_timestamp = timestamp + return fps, result + offset += box_size + return None def _validate_option_token(label: str, value: str) -> str: @@ -425,18 +528,24 @@ def _probe_encoder( """Perform a real one-frame encode and decode, not a name-only probe.""" with tempfile.TemporaryDirectory(prefix="openadapt-ffmpeg-probe-") as temp: root = Path(temp) - source = root / "probe.png" output = root / f"probe.{muxer}" - Image.new("RGB", (64, 64), "black").save(source, format="PNG") + frame = Image.new("RGB", (64, 64), "black").tobytes() command = [ provision.executable, "-hide_banner", "-loglevel", "error", - "-nostdin", "-y", + "-f", + "rawvideo", + "-pixel_format", + "rgb24", + "-video_size", + "64x64", + "-framerate", + "1", "-i", - str(source), + "pipe:0", "-frames:v", "1", "-an", @@ -449,7 +558,7 @@ def _probe_encoder( muxer, str(output), ] - _run_checked(command, timeout=timeout) + _run_checked(command, timeout=timeout, input_bytes=frame) if not output.is_file() or output.stat().st_size == 0: raise FFmpegEncodingError(f"Encoder {codec!r} returned no non-empty {muxer} output") _decode_first_frame_png( @@ -540,7 +649,7 @@ def require_video_encoder( class FFmpegFrameStage: - """Lossless timestamped frames awaiting transactional FFmpeg encoding.""" + """Compatibility writer that streams timestamped frames directly to FFmpeg.""" def __init__( self, @@ -559,68 +668,45 @@ def __init__( self.crf = crf self.preset = preset self.timeout_seconds = timeout_seconds - self.stage_dir = Path( - tempfile.mkdtemp( - prefix=f".{self.output_path.stem}-frames-", - dir=self.output_path.parent, - ) + self.partial_path = self.output_path.with_name( + f".{self.output_path.name}.{uuid.uuid4().hex}.partial.{self.stream.muxer}" ) - self.frames: list[_StagedFrame] = [] + self._process: subprocess.Popen[bytes] | None = None + self._stderr_file: BinaryIO | None = None + self._input_queue: queue.Queue[tuple[bytes, int] | None] = queue.Queue(maxsize=4) + self._input_thread: threading.Thread | None = None + self._input_error: BaseException | None = None + self._last_frame: bytes | None = None + self._first_pts: int | None = None + self._last_pts = -1 + self._emitted_frames = 0 + self._logical_frames: list[tuple[int, float]] = [] self._closed = False self._lock = threading.Lock() - def stage_frame(self, image: "PILImage", pts: int) -> None: - """Persist one numeric-name lossless frame and its exact PTS.""" - with self._lock: - if self._closed: - raise FFmpegEncodingError("Video stage is already closed") - frame_path = self.stage_dir / f"frame_{len(self.frames):08d}.png" - image.convert("RGB").save(frame_path, format="PNG") - self.frames.append(_StagedFrame(frame_path, pts)) - - def _write_manifest(self) -> Path: - if not self.frames: - raise FFmpegEncodingError("Cannot encode a video with no frames") - manifest = self.stage_dir / "frames.ffconcat" - fps = float(self.stream.average_rate) - lines = ["ffconcat version 1.0"] - for index, frame in enumerate(self.frames): - if index + 1 < len(self.frames): - next_pts = self.frames[index + 1].pts - duration = max((next_pts - frame.pts) / fps, 1.0 / fps) - else: - duration = 1.0 / fps - # Staging names are generated numeric tokens, never user input. - lines.append(f"file {frame.path.name}") - lines.append(f"duration {duration:.9f}") - # ffconcat applies the final duration only when the last file is - # repeated. This is an intentional duplicate manifest entry, not - # duplicate raw-frame traffic. - lines.append(f"file {self.frames[-1].path.name}") - manifest.write_text("\n".join(lines) + "\n", encoding="utf-8") - return manifest - - def _encode_command(self, manifest: Path, output: Path, *, fps_mode: bool) -> list[str]: + def _encode_command(self) -> list[str]: + rate = self.stream.average_rate command = [ self.provision.executable, "-hide_banner", "-loglevel", "error", - "-nostdin", "-y", "-f", - "concat", - "-safe", - "1", + "rawvideo", + "-pixel_format", + "rgb24", + "-video_size", + f"{self.stream.width}x{self.stream.height}", + "-framerate", + f"{rate.numerator}/{rate.denominator}", "-i", - manifest.name, + "pipe:0", "-an", "-c:v", self.stream.codec, "-pix_fmt", self.stream.pix_fmt, - "-movflags", - "+faststart", ] command.extend( _codec_arguments( @@ -629,58 +715,238 @@ def _encode_command(self, manifest: Path, output: Path, *, fps_mode: bool) -> li preset=self.preset, ) ) - command.extend(["-fps_mode", "vfr"] if fps_mode else ["-vsync", "vfr"]) command.extend(["-f", self.stream.muxer]) - command.append(str(output)) + command.append(str(self.partial_path)) return command + def _start(self) -> subprocess.Popen[bytes]: + stderr_file = tempfile.TemporaryFile(mode="w+b") + try: + process = subprocess.Popen( + self._encode_command(), + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=stderr_file, + bufsize=0, + shell=False, + ) + except OSError as exc: + stderr_file.close() + raise FFmpegEncodingError(f"Could not execute FFmpeg: {exc}") from exc + if process.stdin is None: + process.kill() + process.wait() + stderr_file.close() + raise FFmpegEncodingError("FFmpeg did not expose its raw-video input pipe") + self._stderr_file = stderr_file + self._process = process + self._input_thread = threading.Thread( + target=self._write_input, + name="openadapt-ffmpeg-input", + daemon=True, + ) + try: + self._input_thread.start() + except BaseException: + process.kill() + process.wait() + stderr_file.close() + self._process = None + self._stderr_file = None + self._input_thread = None + raise + return process + + def _stderr_detail(self) -> str: + stderr_file = self._stderr_file + if stderr_file is None: + return "" + stderr_file.flush() + stderr_file.seek(0, os.SEEK_END) + size = stderr_file.tell() + stderr_file.seek(max(0, size - 64 * 1024)) + return stderr_file.read().decode("utf-8", errors="replace").strip() + + def _write_input(self) -> None: + process = self._process + assert process is not None and process.stdin is not None + try: + while True: + item = self._input_queue.get() + try: + if item is None: + return + frame, repetitions = item + for _ in range(repetitions): + remaining = memoryview(frame) + while remaining: + written = process.stdin.write(remaining) + if written is None or written <= 0: + raise OSError("FFmpeg input pipe made no write progress") + remaining = remaining[written:] + process.stdin.flush() + finally: + self._input_queue.task_done() + except BaseException as exc: + self._input_error = exc + finally: + try: + process.stdin.close() + except BaseException as exc: + if self._input_error is None: + self._input_error = exc + + def _abort_and_reap(self) -> None: + process = self._process + if process is None: + return + if process.poll() is None: + process.kill() + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + pass + input_thread = self._input_thread + if input_thread is not None and input_thread is not threading.current_thread(): + input_thread.join(timeout=5) + + def _enqueue_frames(self, frame: bytes, repetitions: int) -> None: + if repetitions <= 0: + return + process = self._process or self._start() + if process.poll() is not None: + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg exited with code {process.returncode}: {detail or '(no stderr)'}" + ) + if self._input_error is not None: + self._abort_and_reap() + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg input pipe failed: {detail or self._input_error}" + ) from self._input_error + try: + self._input_queue.put( + (frame, repetitions), + timeout=min(self.timeout_seconds, FRAME_WRITE_TIMEOUT_SECONDS), + ) + except queue.Full as exc: + self._abort_and_reap() + detail = self._stderr_detail() + raise FFmpegEncodingError( + "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.""" + with self._lock: + if self._closed: + raise FFmpegEncodingError("Video stream is already closed") + if pts <= self._last_pts: + raise FFmpegEncodingError(f"Video PTS must increase ({pts} <= {self._last_pts})") + if image.size != (self.stream.width, self.stream.height): + raise FFmpegEncodingError( + f"Video frame size {image.size} does not match " + f"{self.stream.width}x{self.stream.height}" + ) + frame = image.convert("RGB").tobytes() + fps = float(self.stream.average_rate) + if self._first_pts is None: + self._enqueue_frames(frame, 1) + self._first_pts = pts + encoded_index = 0 + emitted = 1 + else: + assert self._last_frame is not None + gap = pts - self._last_pts + self._enqueue_frames(self._last_frame, gap - 1) + self._enqueue_frames(frame, 1) + encoded_index = self._emitted_frames + gap - 1 + emitted = gap + assert self._first_pts is not None + self._logical_frames.append((encoded_index, (pts - self._first_pts) / fps)) + self._emitted_frames += emitted + self._last_frame = frame + self._last_pts = pts + + def _finish_input(self) -> None: + process = self._process + input_thread = self._input_thread + assert process is not None and input_thread is not None + try: + self._input_queue.put( + None, + timeout=min(self.timeout_seconds, FRAME_WRITE_TIMEOUT_SECONDS), + ) + except queue.Full as exc: + self._abort_and_reap() + raise FFmpegEncodingError("FFmpeg input queue did not finish") from exc + input_thread.join(timeout=min(self.timeout_seconds, FRAME_WRITE_TIMEOUT_SECONDS)) + if input_thread.is_alive(): + self._abort_and_reap() + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg input pipe timed out after " + f"{min(self.timeout_seconds, FRAME_WRITE_TIMEOUT_SECONDS):g}s" + + (f": {detail}" if detail else "") + ) + if self._input_error is not None: + self._abort_and_reap() + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg input pipe failed: {detail or self._input_error}" + ) from self._input_error + + def _wait_for_process(self, process: subprocess.Popen[bytes]) -> None: + try: + process.wait(timeout=self.timeout_seconds) + except subprocess.TimeoutExpired as exc: + self._abort_and_reap() + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg timed out after {self.timeout_seconds:g}s" + + (f": {detail}" if detail else "") + ) from exc + def close(self) -> None: - """Encode, verify, atomically promote, then remove staging data.""" + """Finalize, verify, and atomically promote the directly encoded video.""" with self._lock: if self._closed: return - if not self.frames: + if self._process is None: self._closed = True - shutil.rmtree(self.stage_dir) return - - manifest = self._write_manifest() - temp_output = self.output_path.with_name( - f".{self.output_path.name}.{uuid.uuid4().hex}.tmp.{self.stream.muxer}" - ) + process = self._process try: - try: - _run_checked( - self._encode_command(manifest, temp_output, fps_mode=True), - timeout=self.timeout_seconds, - cwd=self.stage_dir, - ) - except FFmpegEncodingError as exc: - # FFmpeg <5 does not support -fps_mode. Retry only when the - # failure identifies that exact compatibility condition. - if "fps_mode" not in str(exc).lower() or "option" not in str(exc).lower(): - raise - _run_checked( - self._encode_command(manifest, temp_output, fps_mode=False), - timeout=self.timeout_seconds, - cwd=self.stage_dir, + self._finish_input() + self._wait_for_process(process) + if process.returncode != 0: + detail = self._stderr_detail() + raise FFmpegEncodingError( + f"FFmpeg exited with code {process.returncode}: {detail or '(no stderr)'}" ) - - if not temp_output.is_file() or temp_output.stat().st_size == 0: + if not self.partial_path.is_file() or self.partial_path.stat().st_size == 0: raise FFmpegEncodingError("FFmpeg returned success without a non-empty output") + _append_timing_box( + self.partial_path, + fps=self.stream.average_rate, + frames=self._logical_frames, + ) _decode_first_frame_png( self.provision, - temp_output, + self.partial_path, timeout=min(self.timeout_seconds, EXTRACT_TIMEOUT_SECONDS), ) - os.replace(temp_output, self.output_path) + os.replace(self.partial_path, self.output_path) except BaseException: - if temp_output.exists(): - temp_output.unlink() - logger.error(f"Video encoding failed; retained lossless stage at {self.stage_dir}") + self._abort_and_reap() + logger.error( + f"Video encoding failed; retained incomplete output at {self.partial_path}" + ) raise - else: - shutil.rmtree(self.stage_dir) + finally: + if self._stderr_file is not None: + self._stderr_file.close() self._closed = True @@ -867,7 +1133,6 @@ def finalize_video_writer( video_file_path: str, fix_moov: bool = False, ) -> None: - del video_file_path write_video_frame( video_container, video_stream, @@ -879,7 +1144,7 @@ def finalize_video_writer( ) video_container.close() if fix_moov: - logger.info("FFmpeg encoding already applied +faststart") + move_moov_atom(video_file_path) def move_moov_atom( @@ -890,6 +1155,7 @@ def move_moov_atom( ) -> None: provision = resolve_ffmpeg(ffmpeg_path or config.VIDEO_FFMPEG_PATH) input_path = Path(input_file) + timing = _read_timing_box(input_path) temp_file: Path | None = None if output_file is None: temp_file = input_path.with_name(f".{input_path.name}.{uuid.uuid4().hex}.mp4") @@ -914,6 +1180,9 @@ 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) if temp_file is not None: os.replace(temp_file, input_path) @@ -960,6 +1229,9 @@ def _frame_catalog( provision: FFmpegProvision, ) -> list[tuple[int, float]]: """Return decoded-frame indexes and timestamps in presentation order.""" + timing = _read_timing_box(video_path) + if timing is not None: + return timing[1] payload = _probe_json( provision, [ diff --git a/tests/test_video.py b/tests/test_video.py index 798dcc1..1dd56ee 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -38,6 +38,101 @@ def _stream(codec: str = "libx264") -> video.FFmpegVideoStream: ) +def _small_stream(codec: str = "mpeg4") -> video.FFmpegVideoStream: + return video.FFmpegVideoStream( + width=2, + height=1, + average_rate=Fraction(24), + pix_fmt="yuv420p", + codec=codec, + muxer="mp4", + ) + + +class _FakeInput: + def __init__( + self, + *, + broken: bool = False, + close_broken: bool = False, + max_write: int | None = None, + ) -> None: + self.data = bytearray() + self.broken = broken + self.close_broken = close_broken + self.max_write = max_write + self.closed = False + + def write(self, payload: bytes) -> int: + if self.broken: + raise BrokenPipeError("encoder exited") + written = len(payload) + if self.max_write is not None: + written = min(written, self.max_write) + self.data.extend(payload[:written]) + return written + + def flush(self) -> None: + if self.broken: + raise BrokenPipeError("encoder exited") + + def close(self) -> None: + self.closed = True + if self.close_broken: + raise OSError("stdin close failed") + + +class _FakeProcess: + def __init__( + self, + command: list[str], + *, + returncode: int = 0, + stderr: bytes = b"", + broken: bool = False, + close_broken: bool = False, + max_write: int | None = None, + timeout: bool = False, + ) -> None: + self.command = command + self.pipe = _FakeInput( + broken=broken, + close_broken=close_broken, + max_write=max_write, + ) + self.stdin = self.pipe + self.stderr_payload = stderr + self.returncode: int | None = None + self._final_returncode = returncode + self._timeout = timeout + self._killed = False + + def poll(self) -> int | None: + return self.returncode + + def wait(self, timeout: float | None = None) -> int: + if self._timeout and not self._killed: + raise subprocess.TimeoutExpired(self.command, timeout) + self.returncode = -9 if self._killed else self._final_returncode + if self.returncode == 0: + Path(self.command[-1]).write_bytes(b"\x00\x00\x00\x08ftyp") + return self.returncode + + def kill(self) -> None: + self._killed = True + + +def _install_fake_popen(monkeypatch, process: _FakeProcess): + def popen(command, **kwargs): + process.command = list(command) + stderr_file = kwargs["stderr"] + stderr_file.write(process.stderr_payload) + stderr_file.flush() + return process + + monkeypatch.setattr(video.subprocess, "Popen", popen) + + def _png_bytes(color: str = "black") -> bytes: output = io.BytesIO() Image.new("RGB", (2, 2), color).save(output, format="PNG") @@ -255,120 +350,299 @@ def timeout(*args, **kwargs): video._run_checked(["/tmp/ffmpeg"], timeout=3) -def test_staged_pts_become_exact_ffconcat_durations(tmp_path): +def test_encoder_probe_uses_one_raw_frame_over_stdin(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + captured: dict[str, object] = {} + + def fake_run(command, **kwargs): + captured["command"] = list(command) + captured["kwargs"] = kwargs + Path(command[-1]).write_bytes(b"probe-video") + return subprocess.CompletedProcess(command, 0, stdout=b"", stderr=b"") + + monkeypatch.setattr(video, "_run_checked", fake_run) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_args, **_kwargs: _png_bytes()) + + video._probe_encoder(_provision(executable), "mpeg4", "yuv420p", "mp4") + + command = captured["command"] + kwargs = captured["kwargs"] + assert isinstance(command, list) + assert command[command.index("-f") + 1] == "rawvideo" + assert "pipe:0" in command + assert "concat" not in command + assert isinstance(kwargs, dict) + assert len(kwargs["input_bytes"]) == 64 * 64 * 3 + + +def test_direct_stream_preserves_pts_timing_without_png_staging(tmp_path, monkeypatch): executable = tmp_path / "ffmpeg" executable.write_bytes(b"fake") output = tmp_path / "capture.mp4" stage = video.FFmpegFrameStage( output, - _stream(), + _small_stream(), _provision(executable), ) - image = Image.new("RGB", (100, 80), color="red") - stage.stage_frame(image, 0) - stage.stage_frame(image, 24) - stage.stage_frame(image, 60) - - manifest = stage._write_manifest() - lines = manifest.read_text(encoding="utf-8").splitlines() - - assert lines == [ - "ffconcat version 1.0", - "file frame_00000000.png", - "duration 1.000000000", - "file frame_00000001.png", - "duration 1.500000000", - "file frame_00000002.png", - "duration 0.041666667", - "file frame_00000002.png", + processes: list[_FakeProcess] = [] + + def popen(command, **kwargs): + assert kwargs["shell"] is False + process = _FakeProcess(list(command)) + kwargs["stderr"].flush() + processes.append(process) + return process + + monkeypatch.setattr(video.subprocess, "Popen", popen) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_args, **_kwargs: _png_bytes()) + + red = Image.new("RGB", (2, 1), color="red") + blue = Image.new("RGB", (2, 1), color="blue") + stage.stage_frame(red, 0) + stage.stage_frame(blue, 3) + stage.close() + + assert len(processes) == 1 + process = processes[0] + red_bytes = red.tobytes() + blue_bytes = blue.tobytes() + assert bytes(process.pipe.data) == red_bytes * 3 + blue_bytes + assert process.pipe.closed is True + assert "-f" in process.command + assert "rawvideo" in process.command + assert "pipe:0" in process.command + assert "-use_wallclock_as_timestamps" not in process.command + assert "-fps_mode" not in process.command + assert "concat" not in process.command + assert "-nostdin" not in process.command + assert "+faststart" not in process.command + assert output.read_bytes().startswith(b"\x00\x00\x00\x08ftyp") + timing = video._read_timing_box(output) + assert timing is not None + assert timing[0] == Fraction(24) + assert timing[1] == [ + (0, pytest.approx(0.0)), + (3, pytest.approx(3 / 24)), ] + assert not list(tmp_path.glob("*.png")) + assert not list(tmp_path.glob("*.ffconcat")) -def test_successful_encode_is_verified_promoted_and_cleans_stage(tmp_path, monkeypatch): +def test_successful_direct_encode_is_verified_and_atomically_promoted(tmp_path, monkeypatch): executable = tmp_path / "ffmpeg" executable.write_bytes(b"fake") output = tmp_path / "capture.mp4" stage = video.FFmpegFrameStage( output, - _stream(), + _small_stream(), _provision(executable), ) - stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) - commands: list[list[str]] = [] + process = _FakeProcess(stage._encode_command()) + _install_fake_popen(monkeypatch, process) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_args, **_kwargs: _png_bytes()) - def fake_run(command, **_kwargs): - commands.append(list(command)) - if "-f" in command and "concat" in command: - Path(command[-1]).write_bytes(b"verified-video") - stdout = b"" - elif "image2pipe" in command: - stdout = _png_bytes() - else: - stdout = b"" - return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=b"") - - monkeypatch.setattr(video, "_run_checked", fake_run) - stage_dir = stage.stage_dir + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) stage.close() - assert output.read_bytes() == b"verified-video" - assert not stage_dir.exists() - assert any("-fps_mode" in command for command in commands) - assert any("0:v:0" in command for command in commands) + assert output.read_bytes().startswith(b"\x00\x00\x00\x08ftyp") + assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)]) + assert not stage.partial_path.exists() -def test_failed_encode_retains_stage_and_never_promotes_output(tmp_path, monkeypatch): +def test_direct_stream_normalizes_nonzero_initial_pts(tmp_path, monkeypatch): executable = tmp_path / "ffmpeg" executable.write_bytes(b"fake") output = tmp_path / "capture.mp4" stage = video.FFmpegFrameStage( output, - _stream(), + _small_stream(), _provision(executable), ) - stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) + process = _FakeProcess(stage._encode_command()) + _install_fake_popen(monkeypatch, process) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_args, **_kwargs: _png_bytes()) + red = Image.new("RGB", (2, 1), "red") + blue = Image.new("RGB", (2, 1), "blue") - def fail(*_args, **_kwargs): - raise video.FFmpegEncodingError("encoder failed") + stage.stage_frame(red, 100) + stage.stage_frame(blue, 103) + stage.close() + + assert bytes(process.pipe.data) == red.tobytes() * 3 + blue.tobytes() + assert video._read_timing_box(output) == ( + Fraction(24), + [(0, 0.0), (3, 3 / 24)], + ) - monkeypatch.setattr(video, "_run_checked", fail) + +def test_failed_direct_encode_retains_partial_and_never_promotes_output(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + returncode=7, + stderr=b"encoder failed", + ) + stage.partial_path.write_bytes(b"incomplete") + _install_fake_popen(monkeypatch, process) + + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) with pytest.raises(video.FFmpegEncodingError, match="encoder failed"): stage.close() - assert stage.stage_dir.is_dir() - assert (stage.stage_dir / "frame_00000000.png").is_file() + assert stage.partial_path.read_bytes() == b"incomplete" assert not output.exists() -def test_fps_mode_compatibility_fallback_is_narrow_and_transactional(tmp_path, monkeypatch): +def test_direct_encode_timeout_is_bounded_and_fail_loud(tmp_path, monkeypatch): executable = tmp_path / "ffmpeg" executable.write_bytes(b"fake") output = tmp_path / "capture.mp4" stage = video.FFmpegFrameStage( output, - _stream(), + _small_stream(), _provision(executable), + timeout_seconds=0.01, ) - stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) - commands: list[list[str]] = [] + process = _FakeProcess(stage._encode_command(), timeout=True) + _install_fake_popen(monkeypatch, process) - def fake_run(command, **_kwargs): - commands.append(list(command)) - if "-fps_mode" in command: - raise video.FFmpegEncodingError( - "FFmpeg exited with code 1: Unrecognized option 'fps_mode'" - ) - if "-vsync" in command: - Path(command[-1]).write_bytes(b"legacy-compatible") - stdout = _png_bytes() if "image2pipe" in command else b"" - return subprocess.CompletedProcess(command, 0, stdout=stdout, stderr=b"") + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) + with pytest.raises(video.FFmpegEncodingError, match="timed out"): + stage.close() + assert process._killed is True + assert not output.exists() - monkeypatch.setattr(video, "_run_checked", fake_run) + +def test_direct_encode_broken_pipe_is_fail_loud(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + stage = video.FFmpegFrameStage( + tmp_path / "capture.mp4", + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + returncode=9, + stderr=b"codec crashed", + broken=True, + ) + _install_fake_popen(monkeypatch, process) + + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) + with pytest.raises(video.FFmpegEncodingError, match="codec crashed"): + stage.close() + assert process._killed is True + + +def test_direct_encode_retries_partial_pipe_writes(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + max_write=2, + ) + _install_fake_popen(monkeypatch, process) + monkeypatch.setattr(video, "_decode_first_frame_png", lambda *_args, **_kwargs: _png_bytes()) + frame = Image.new("RGB", (2, 1), "red") + + stage.stage_frame(frame, 0) stage.close() - assert output.read_bytes() == b"legacy-compatible" - assert any("-fps_mode" in command for command in commands) - assert any("-vsync" in command for command in commands) + assert bytes(process.pipe.data) == frame.tobytes() + assert video._read_timing_box(output) == (Fraction(24), [(0, 0.0)]) + + +def test_direct_encode_zero_length_pipe_write_fails_loudly(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + stage = video.FFmpegFrameStage( + tmp_path / "capture.mp4", + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + max_write=0, + ) + _install_fake_popen(monkeypatch, process) + + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) + with pytest.raises(video.FFmpegEncodingError, match="no write progress"): + stage.close() + assert process._killed is True + + +def test_direct_encode_stdin_close_failure_reaps_process(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + stage = video.FFmpegFrameStage( + tmp_path / "capture.mp4", + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + close_broken=True, + ) + _install_fake_popen(monkeypatch, process) + + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) + with pytest.raises(video.FFmpegEncodingError, match="stdin close failed"): + stage.close() + assert process._killed is True + + +def test_direct_encode_bounded_stderr_tail_does_not_block(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + stage = video.FFmpegFrameStage( + tmp_path / "capture.mp4", + _small_stream(), + _provision(executable), + ) + process = _FakeProcess( + stage._encode_command(), + returncode=2, + stderr=b"x" * (128 * 1024) + b" final encoder error", + ) + _install_fake_popen(monkeypatch, process) + + stage.stage_frame(Image.new("RGB", (2, 1), "red"), 0) + with pytest.raises(video.FFmpegEncodingError, match="final encoder error"): + stage.close() + + +def test_empty_direct_stream_closes_without_starting_ffmpeg(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + stage = video.FFmpegFrameStage( + tmp_path / "capture.mp4", + _small_stream(), + _provision(executable), + ) + monkeypatch.setattr( + video.subprocess, + "Popen", + lambda *_args, **_kwargs: pytest.fail("empty stream must not start FFmpeg"), + ) + + stage.close() + assert stage._closed is True def test_record_refuses_missing_encoder_before_display_or_listeners(tmp_path, monkeypatch): @@ -390,9 +664,7 @@ def touch_display(): assert display_touched is False -def test_preflight_provision_survives_spawn_without_child_config( - tmp_path, monkeypatch -): +def test_preflight_provision_survives_spawn_without_child_config(tmp_path, monkeypatch): """Spawned writers use the exact parent-selected provision, not fresh defaults.""" executable = tmp_path / "managed-ffmpeg" executable.write_bytes(b"preflighted in parent") @@ -614,6 +886,9 @@ def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): start, -1, ) + # Simulate a writer queue that falls behind capture time. Encoded PTS must + # follow the supplied capture timestamp, never this processing delay. + time.sleep(1.05) last_pts = video.write_video_frame( container, stream, @@ -636,7 +911,28 @@ def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): assert info["codec"] == "mpeg4" assert info["width"] == 100 assert info["height"] == 80 - assert info["frames"] >= 2 + assert info["frames"] == 26 + provision = video.resolve_ffmpeg(executable) + payload = video._probe_json( + provision, + [ + "-select_streams", + "v:0", + "-show_frames", + "-show_entries", + "frame=best_effort_timestamp_time", + str(output), + ], + ) + decoded_timestamps = [float(item["best_effort_timestamp_time"]) for item in payload["frames"]] + assert decoded_timestamps == pytest.approx( + [index / 24 for index in range(26)], + abs=1e-6, + ) + assert video._read_timing_box(output) == ( + Fraction(24), + [(0, 0.0), (24, 1.0), (25, 25 / 24)], + ) frame = video.extract_frame( output, 0.8, @@ -644,3 +940,9 @@ def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): ffmpeg_path=executable, ) 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)], + )