diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 1317196..e067a3f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -59,6 +59,9 @@ jobs: - name: Set up Python run: uv python install 3.12 + - name: Install external video tools + run: choco install ffmpeg -y --no-progress + - name: Install dependencies run: uv sync --extra dev @@ -76,8 +79,8 @@ jobs: run: uv run pytest tests/test_performance.py -v -m slow --timeout=300 timeout-minutes: 20 - # macOS coverage: runs the unit suite (event processing, storage, video - # encode/decode via PyAV, headless-import invariant, Recorder API surface — + # macOS coverage: runs the unit suite (event processing, storage, external + # video boundary, headless-import invariant, Recorder API surface — # CGWindowList-backed screen capture works on GitHub macOS runners). The # 'slow' live-recording tests stay skipped here: synthetic input injection # requires Accessibility permissions a hosted runner cannot grant, and the @@ -96,6 +99,9 @@ jobs: - name: Set up Python run: uv python install 3.12 + - name: Install external video tools + run: brew install ffmpeg + - name: Install dependencies run: uv sync --extra dev diff --git a/README.md b/README.md index bb4cf24..a8e5340 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,7 @@ Documentation for the whole stack lives at | Recording path | Current implementation | | --- | --- | -| Windows and RDP demonstrations | `openadapt-capture` records native input and screen video; `openadapt-flow` converts the session into compiler input. | +| Windows and RDP demonstrations | `openadapt-capture` records native input and action-gated screen video; `openadapt-flow` converts the session into compiler input. | | Browser demonstrations | `openadapt-flow` records its Playwright browser directly. It does not require this package. | | Chrome extension in this repository | Experimental DOM-capture code for development; it is not the supported web recorder or governed replay path. | @@ -120,7 +120,17 @@ my-capture/ └── profiling.json ``` -Audio and individual images are optional. +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 +`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. ## Window-scoped recording diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 502168b..3d20be5 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -17,7 +17,9 @@ We need a platform-agnostic representation of GUI interactions that: ### Key Features 1. **Production-ready** - Designed for continuous operation without degradation -2. **No external dependencies** - Native video capture via PyAV (no OBS required) +2. **Process-isolated video** - The MIT package stages frames and invokes a + separately provisioned FFmpeg executable; it does not bundle or link codec + libraries 3. **Low resource footprint** - Non-blocking capture that doesn't slow down user's work 4. **Chunked media** - Video/audio split into manageable segments for long captures 5. **Audio capture** - Built-in audio recording with Whisper transcription @@ -164,11 +166,21 @@ class Capture: ### Video Encoding -Based on OpenAdapt's working implementation: +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. ```python -codec = "libx264" # H.264 for compatibility -pix_fmt = "yuv444p" # Full color (vs yuv420p for smaller files) +codec = None # Probe platform encoders, then portable mpeg4 fallback +pix_fmt = None # Selected with the verified codec crf = 0 # Lossless (adjustable for size vs quality) preset = "veryslow" # Maximum compression fps = 24 # Configurable @@ -194,12 +206,12 @@ capture_abc123/ ### Screenshots vs Video -**Recommendation:** Default to video mode. +**Decision:** Default to video mode. -- More storage-efficient (H.264 compression) -- Easier timestamp alignment +- More storage-efficient than permanent individual screenshots +- Exact timestamp alignment is preserved by staged-frame PTS and VFR durations - Screenshots can be extracted from video when needed -- Individual screenshots useful for debugging frame alignment +- Individual screenshots remain optional for debugging frame alignment ## Audio Handling @@ -269,7 +281,7 @@ def scrub_capture(capture: Capture, scrubber: ScrubbingProvider) -> Capture: 1. Define exact event schemas (Pydantic models) 2. Implement SQLite storage for events -3. Implement video capture with PyAV +3. Maintain external-process video capture without bundling codec binaries 4. Port event processing from OpenAdapt's `events.py` 5. Add drag detection 6. Integrate with `openadapt-privacy` diff --git a/openadapt_capture/cli.py b/openadapt_capture/cli.py index 59f2158..e8bfc44 100644 --- a/openadapt_capture/cli.py +++ b/openadapt_capture/cli.py @@ -27,9 +27,10 @@ def record( Args: output_dir: Directory to save capture. description: Optional task description. - video: Capture video (default: True). + video: Capture MP4 video (default: True). Requires an externally + provisioned FFmpeg executable. audio: Capture audio (default: False). - images: Save screenshots as PNGs (default: False). + images: Also save screenshots as PNGs (default: False). browser_events: Capture browser DOM events via Chrome extension (default: False). Requires the openadapt-capture Chrome extension to be installed and connects via WebSocket on localhost:8765. diff --git a/openadapt_capture/config.py b/openadapt_capture/config.py index 9a4220e..ba0e7c8 100644 --- a/openadapt_capture/config.py +++ b/openadapt_capture/config.py @@ -48,8 +48,16 @@ class Settings(BaseSettings): RECORD_WINDOW_TITLE: str | None = None # useful for debugging but expensive computationally LOG_MEMORY: bool = False - VIDEO_ENCODING: str = "libx264" - VIDEO_PIXEL_FORMAT: str = "yuv444p" + # None means: use the Desktop provision manifest when present, otherwise + # probe a usable platform encoder with a portable mpeg4 fallback. + VIDEO_ENCODING: str | None = None + VIDEO_PIXEL_FORMAT: str | None = None + VIDEO_MUXER: str | None = None + # Optional exact path to an externally provisioned FFmpeg executable. + # Capture never downloads or bundles FFmpeg. + VIDEO_FFMPEG_PATH: str | None = None + VIDEO_FFPROBE_PATH: str | None = None + VIDEO_FFMPEG_TIMEOUT_SECONDS: float = 900.0 # sequences that when typed, will stop the recording of ActionEvents STOP_SEQUENCES: list[list[str]] = [ list(stop_str) for stop_str in STOP_STRS @@ -97,6 +105,10 @@ class Settings(BaseSettings): "capture_full_video": "RECORD_FULL_VIDEO", "video_encoding": "VIDEO_ENCODING", "video_pixel_format": "VIDEO_PIXEL_FORMAT", + "video_muxer": "VIDEO_MUXER", + "ffmpeg_path": "VIDEO_FFMPEG_PATH", + "ffprobe_path": "VIDEO_FFPROBE_PATH", + "ffmpeg_timeout_seconds": "VIDEO_FFMPEG_TIMEOUT_SECONDS", "stop_sequences": "STOP_SEQUENCES", "log_memory": "LOG_MEMORY", "plot_performance": "PLOT_PERFORMANCE", @@ -119,6 +131,10 @@ class RecordingConfig: capture_full_video: bool | None = None video_encoding: str | None = None video_pixel_format: str | None = None + video_muxer: str | None = None + ffmpeg_path: str | None = None + ffprobe_path: str | None = None + ffmpeg_timeout_seconds: float | None = None stop_sequences: list[list[str]] | None = None log_memory: bool | None = None plot_performance: bool | None = None diff --git a/openadapt_capture/recorder.py b/openadapt_capture/recorder.py index 277b7c7..9d59187 100644 --- a/openadapt_capture/recorder.py +++ b/openadapt_capture/recorder.py @@ -33,7 +33,6 @@ from functools import partial from typing import Any, Callable -import av import fire import numpy as np import psutil @@ -272,6 +271,22 @@ def _join_tasks( return lingering +def _raise_for_failed_processes(task_by_name: dict[str, Any]) -> None: + """Surface required child-process failures through the recording boundary.""" + failures = { + name: task.exitcode + for name, task in task_by_name.items() + if isinstance(task, multiprocessing.process.BaseProcess) + and task.exitcode not in (None, 0) + } + if failures: + detail = ", ".join( + f"{name} (exit code {exitcode})" + for name, exitcode in sorted(failures.items()) + ) + raise RuntimeError(f"Recording child process failed: {detail}") + + def collect_stats(performance_snapshots: list[tracemalloc.Snapshot]) -> None: """Collects and appends performance snapshots using tracemalloc. @@ -670,6 +685,8 @@ def video_pre_callback( recording: Recording, video_dir: str = None, frame_size: tuple[int, int] | None = None, + provision: video.FFmpegProvision | None = None, + timeout_seconds: float | None = None, ) -> dict[str, Any]: """Function to call before main loop. @@ -680,6 +697,9 @@ def video_pre_callback( frame_size: Explicit (width, height) for the video stream. Used by window-scoped capture, whose frames are the target window's pixels rather than the monitor's. Defaults to the full-screen size. + provision: Parent-preflighted encoder contract. Passing this immutable + value keeps spawn-based writers on the exact executable and codec. + timeout_seconds: Bound for final encoding and verification. Returns: dict[str, Any]: The updated state. @@ -691,7 +711,13 @@ def video_pre_callback( # TODO XXX replace with utils.get_monitor_dims() once fixed monitor_width, monitor_height = utils.take_screenshot().size video_container, video_stream, video_start_timestamp = ( - video.initialize_video_writer(video_file_path, monitor_width, monitor_height) + video.initialize_video_writer( + video_file_path, + monitor_width, + monitor_height, + timeout_seconds=timeout_seconds, + preflight_provision=provision, + ) ) crud.update_video_start_time(db, recording, video_start_timestamp) return { @@ -730,8 +756,8 @@ def write_video_event( recording_timestamp: float, event: Event, perf_q: sq.SynchronizedQueue, - video_container: av.container.OutputContainer, - video_stream: av.stream.Stream, + video_container: Any, + video_stream: Any, video_start_timestamp: float, last_pts: int = 0, num_copies: int = 2, @@ -744,9 +770,8 @@ 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 (av.container.OutputContainer): The output container to which - the frame is written. - video_stream (av.stream.Stream): The video stream within the container. + video_container: The external-encoder staging container. + video_stream: The configured external-encoder stream. video_start_timestamp (float): The base timestamp from which the video recording started. last_pts: The last presentation timestamp. @@ -1620,6 +1645,17 @@ def record( config.RECORD_VIDEO, config.RECORD_IMAGES, ) + # Refuse before touching the display or starting any worker. Capture never + # downloads or bundles FFmpeg; Desktop/standalone provisioning owns it. + video_provision = None + if config.RECORD_VIDEO: + video_provision = video.require_video_encoder( + ffmpeg_path=config.VIDEO_FFMPEG_PATH, + ffprobe_path=config.VIDEO_FFPROBE_PATH, + codec=config.VIDEO_ENCODING, + pixel_format=config.VIDEO_PIXEL_FORMAT, + muxer=config.VIDEO_MUXER, + ) # Configure loguru level for recording (without destroying global config) logger.configure(handlers=[{"sink": sys.stderr, "level": LOG_LEVEL}]) @@ -1884,6 +1920,8 @@ def record( if initial_window_frame is not None else None ), + provision=video_provision, + timeout_seconds=config.VIDEO_FFMPEG_TIMEOUT_SECONDS, ), video_post_callback, ), @@ -2013,6 +2051,7 @@ def record( task_name, task_error = task_errors.get_nowait() add_exception_note(task_error, f"recording task {task_name!r} failed") raise task_error + _raise_for_failed_processes(task_by_name) if config.PLOT_PERFORMANCE and startup_ready: from openadapt_capture import plotting @@ -2141,6 +2180,10 @@ def __init__( capture_full_video: bool | None = None, video_encoding: str | None = None, video_pixel_format: str | None = None, + video_muxer: str | None = None, + ffmpeg_path: str | None = None, + ffprobe_path: str | None = None, + ffmpeg_timeout_seconds: float | None = None, stop_sequences: list[list[str]] | None = None, log_memory: bool | None = None, plot_performance: bool | None = None, @@ -2170,6 +2213,10 @@ def __init__( capture_full_video=capture_full_video, video_encoding=video_encoding, video_pixel_format=video_pixel_format, + video_muxer=video_muxer, + ffmpeg_path=ffmpeg_path, + ffprobe_path=ffprobe_path, + ffmpeg_timeout_seconds=ffmpeg_timeout_seconds, stop_sequences=stop_sequences, log_memory=log_memory, plot_performance=plot_performance, diff --git a/openadapt_capture/video.py b/openadapt_capture/video.py index 51fd5d1..9e7acbe 100644 --- a/openadapt_capture/video.py +++ b/openadapt_capture/video.py @@ -1,373 +1,874 @@ -"""Video capture and frame extraction using PyAV. +"""Video capture through a separately provisioned FFmpeg executable. -This module provides video recording capabilities using libx264 encoding, -following OpenAdapt's proven implementation. Includes both a VideoWriter class -and legacy functional API (initialize/write/finalize) copied from legacy OpenAdapt. +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. """ from __future__ import annotations +import io +import json import os +import re +import shutil import subprocess +import sys import tempfile import threading +import uuid +from dataclasses import dataclass from fractions import Fraction from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Sequence -import av from loguru import logger +from PIL import Image from openadapt_capture import utils from openadapt_capture.config import config if TYPE_CHECKING: - from PIL import Image + from PIL.Image import Image as PILImage +DEFAULT_FPS = 24 +DEFAULT_CODEC = "libx264" +DEFAULT_PIXEL_FORMAT = "yuv444p" +DEFAULT_CRF = 0 +DEFAULT_PRESET = "veryslow" +DEFAULT_PROCESS_TIMEOUT_SECONDS = 900.0 +PROBE_TIMEOUT_SECONDS = 10.0 +EXTRACT_TIMEOUT_SECONDS = 120.0 +_ENCODER_LINE = re.compile(r"^\s*V\S*\s+(?P\S+)\s", re.MULTILINE) +_OPTION_TOKEN = re.compile(r"^[A-Za-z0-9_.-]+$") -# ============================================================================= -# Video Writer -# ============================================================================= +class FFmpegUnavailableError(RuntimeError): + """No separately provisioned, usable FFmpeg executable was found.""" -class VideoWriter: - """H.264 video writer using PyAV. - Writes frames to an MP4 file with H.264 encoding for maximum compatibility - and efficient compression. +class FFmpegEncodingError(RuntimeError): + """FFmpeg failed to encode or verify a staged recording.""" + + +@dataclass(frozen=True) +class FFmpegProvision: + """Resolved executable and optional Desktop-declared codec.""" + + executable: str + ffprobe: str | None = None + codec: str | None = None + pixel_format: str | None = None + muxer: str | None = None + source: str = "unknown" + + +@dataclass +class FFmpegVideoStream: + """Small compatibility surface used by the recorder's writer pipeline.""" + + width: int + height: int + average_rate: Fraction + pix_fmt: str + codec: str + muxer: str - Usage: - writer = VideoWriter("output.mp4", width=1920, height=1080) - writer.write_frame(image, timestamp) - writer.close() - Or as context manager: - with VideoWriter("output.mp4", width=1920, height=1080) as writer: - writer.write_frame(image, timestamp) +@dataclass(frozen=True) +class _StagedFrame: + path: Path + pts: int + + +def _validate_option_token(label: str, value: str) -> str: + if not _OPTION_TOKEN.fullmatch(value): + raise FFmpegUnavailableError( + f"Invalid FFmpeg {label} token {value!r}; expected a codec-style name" + ) + return value + + +def _desktop_data_dirs() -> list[Path]: + """Return platform user-data roots where Desktop may provision FFmpeg.""" + home = Path.home() + candidates: list[Path] = [] + if sys_platform := os.environ.get("OPENADAPT_PLATFORM_OVERRIDE"): + platform_name = sys_platform + else: + import sys + + platform_name = sys.platform + + if platform_name == "darwin": + candidates.append(home / "Library" / "Application Support" / "ai.openadapt.desktop") + elif platform_name == "win32": + local_app_data = os.environ.get("LOCALAPPDATA") + if local_app_data: + candidates.append(Path(local_app_data) / "ai.openadapt.desktop") + else: + xdg_data_home = os.environ.get("XDG_DATA_HOME") + candidates.append(Path(xdg_data_home) if xdg_data_home else home / ".local" / "share") + candidates[-1] = candidates[-1] / "ai.openadapt.desktop" + return candidates + + +def _load_desktop_manifest(root: Path) -> FFmpegProvision | None: + """Load a narrow Desktop-owned encoder manifest, if present. + + Schema:: + + { + "version": 1, + "executable": "bin/ffmpeg", + "ffprobe": "bin/ffprobe", + "codec": "h264_videotoolbox", + "pixel_format": "yuv420p", + "muxer": "mp4" + } + + The executable must remain within the manifest's user-data root. Capture + does not create, download, or update this file. """ + manifest_path = root / "ffmpeg.json" + if not manifest_path.is_file(): + return None + try: + payload = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise FFmpegUnavailableError( + f"Invalid Desktop FFmpeg manifest at {manifest_path}: {exc}" + ) from exc + if payload.get("version") != 1: + raise FFmpegUnavailableError( + f"Unsupported Desktop FFmpeg manifest version at {manifest_path}" + ) + executable_value = payload.get("executable") + ffprobe_value = payload.get("ffprobe") + codec = payload.get("codec") + pixel_format = payload.get("pixel_format") + muxer = payload.get("muxer") + if not isinstance(executable_value, str) or not executable_value: + raise FFmpegUnavailableError(f"Desktop FFmpeg manifest has no executable: {manifest_path}") + if ffprobe_value is not None and (not isinstance(ffprobe_value, str) or not ffprobe_value): + raise FFmpegUnavailableError( + f"Desktop FFmpeg manifest has an invalid ffprobe: {manifest_path}" + ) + if codec is not None and (not isinstance(codec, str) or not codec): + raise FFmpegUnavailableError( + f"Desktop FFmpeg manifest has an invalid codec: {manifest_path}" + ) + for key, value in (("pixel_format", pixel_format), ("muxer", muxer)): + if value is not None and (not isinstance(value, str) or not value): + raise FFmpegUnavailableError( + f"Desktop FFmpeg manifest has an invalid {key}: {manifest_path}" + ) + for key, value in ( + ("codec", codec), + ("pixel_format", pixel_format), + ("muxer", muxer), + ): + if value is not None: + _validate_option_token(key, value) + root_resolved = root.resolve() + executable = (root / executable_value).resolve() + try: + executable.relative_to(root_resolved) + except ValueError as exc: + raise FFmpegUnavailableError( + f"Desktop FFmpeg manifest executable escapes {root_resolved}" + ) from exc + ffprobe: Path | None = None + if ffprobe_value is not None: + ffprobe = (root / ffprobe_value).resolve() + try: + ffprobe.relative_to(root_resolved) + except ValueError as exc: + raise FFmpegUnavailableError( + f"Desktop FFmpeg manifest ffprobe escapes {root_resolved}" + ) from exc + return FFmpegProvision( + executable=str(executable), + ffprobe=str(ffprobe) if ffprobe is not None else None, + codec=codec, + pixel_format=pixel_format, + muxer=muxer, + source=f"desktop manifest {manifest_path}", + ) + + +def _candidate_is_file(value: str | os.PathLike[str]) -> str | None: + path = Path(value).expanduser() + if path.is_file(): + return str(path.resolve()) + return None + + +def _sibling_ffprobe(ffmpeg: str) -> str | None: + """Find an ffprobe installed beside FFmpeg without searching or downloading.""" + executable = Path(ffmpeg) + suffix = ".exe" if executable.suffix.lower() == ".exe" else "" + return _candidate_is_file(executable.with_name(f"ffprobe{suffix}")) + + +def _resolve_ffprobe( + ffmpeg: str, + explicit: str | os.PathLike[str] | None = None, +) -> str | None: + requested = explicit or os.environ.get("OPENADAPT_FFPROBE_PATH") + if requested: + resolved = _candidate_is_file(requested) + if resolved is None: + raise FFmpegUnavailableError( + f"Configured ffprobe executable does not exist: {requested}" + ) + return resolved + return _sibling_ffprobe(ffmpeg) or shutil.which("ffprobe") + + +def resolve_ffmpeg( + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, +) -> FFmpegProvision: + """Resolve FFmpeg without downloading or modifying external state. + + Order: + 1. Explicit API/config path. + 2. ``OPENADAPT_FFMPEG_PATH`` / ``OPENADAPT_DESKTOP_FFMPEG_PATH``. + 3. Desktop user-data ``ffmpeg.json`` manifest. + 4. System ``PATH``. + + ``ffprobe`` is resolved from an explicit path, the Desktop manifest, beside + FFmpeg, or from ``PATH``. Encoding does not require it; exact frame lookup + and metadata inspection do. + """ + explicit = ffmpeg_path or os.environ.get("OPENADAPT_FFMPEG_PATH") + if explicit: + resolved = _candidate_is_file(explicit) + if resolved is None: + raise FFmpegUnavailableError(f"Configured FFmpeg executable does not exist: {explicit}") + return FFmpegProvision( + resolved, + ffprobe=_resolve_ffprobe(resolved, ffprobe_path), + source="explicit path", + ) + + desktop_env = os.environ.get("OPENADAPT_DESKTOP_FFMPEG_PATH") + if desktop_env: + resolved = _candidate_is_file(desktop_env) + if resolved is None: + raise FFmpegUnavailableError( + f"OPENADAPT_DESKTOP_FFMPEG_PATH does not name a file: {desktop_env}" + ) + return FFmpegProvision( + resolved, + ffprobe=_resolve_ffprobe(resolved, ffprobe_path), + source="Desktop environment", + ) + + for root in _desktop_data_dirs(): + manifest = _load_desktop_manifest(root) + if manifest is not None: + if not Path(manifest.executable).is_file(): + raise FFmpegUnavailableError( + f"Desktop-provisioned FFmpeg is missing: {manifest.executable}" + ) + manifest_probe = manifest.ffprobe + if manifest_probe is not None and not Path(manifest_probe).is_file(): + raise FFmpegUnavailableError( + f"Desktop-provisioned ffprobe is missing: {manifest_probe}" + ) + return FFmpegProvision( + executable=manifest.executable, + ffprobe=( + _resolve_ffprobe(manifest.executable, ffprobe_path) + if ffprobe_path or os.environ.get("OPENADAPT_FFPROBE_PATH") + else manifest_probe or _resolve_ffprobe(manifest.executable) + ), + codec=manifest.codec, + pixel_format=manifest.pixel_format, + muxer=manifest.muxer, + source=manifest.source, + ) + + path_value = shutil.which("ffmpeg") + if path_value: + resolved = str(Path(path_value).resolve()) + return FFmpegProvision( + resolved, + ffprobe=_resolve_ffprobe(resolved, ffprobe_path), + source="PATH", + ) + + raise FFmpegUnavailableError( + "Video capture requires a separately provisioned FFmpeg executable. " + "Set OPENADAPT_FFMPEG_PATH, pass Recorder(ffmpeg_path=...), provision " + "Desktop's ffmpeg.json user-data manifest, or put ffmpeg on PATH. " + "OpenAdapt Capture does not download or bundle FFmpeg." + ) + + +def _run_checked( + command: Sequence[str], + *, + timeout: float, + cwd: Path | None = None, + input_bytes: bytes | None = None, +) -> subprocess.CompletedProcess[bytes]: + """Run an argument-vector command with bounded, non-shell execution.""" + try: + result = subprocess.run( + list(command), + cwd=cwd, + input=input_bytes, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + shell=False, + check=False, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + raise FFmpegEncodingError(f"FFmpeg timed out after {timeout:g}s: {command[0]}") from exc + except OSError as exc: + raise FFmpegEncodingError(f"Could not execute FFmpeg: {exc}") from exc + if result.returncode != 0: + stderr = result.stderr.decode("utf-8", errors="replace").strip() + raise FFmpegEncodingError( + f"FFmpeg exited with code {result.returncode}: {stderr or '(no stderr)'}" + ) + return result + + +def available_encoders( + provision: FFmpegProvision, + *, + timeout: float = PROBE_TIMEOUT_SECONDS, +) -> set[str]: + """Return video encoder names advertised by the resolved executable.""" + result = _run_checked( + [provision.executable, "-hide_banner", "-encoders"], + timeout=timeout, + ) + output = ( + result.stdout.decode("utf-8", errors="replace") + + "\n" + + result.stderr.decode("utf-8", errors="replace") + ) + return {match.group("name") for match in _ENCODER_LINE.finditer(output)} + + +def _codec_arguments( + codec: str, + *, + crf: int = DEFAULT_CRF, + preset: str = DEFAULT_PRESET, +) -> list[str]: + if codec.startswith(("libx264", "libx265")): + return ["-crf", str(crf), "-preset", preset] + if codec.startswith("libvpx"): + return ["-crf", str(crf), "-b:v", "0"] + if codec == "mpeg4": + return ["-q:v", "2"] + return [] + + +def _decode_first_frame_png( + provision: FFmpegProvision, + video_path: Path, + *, + timeout: float, +) -> bytes: + """Decode one frame to PNG, proving both the video and extraction path.""" + result = _run_checked( + [ + provision.executable, + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-i", + str(video_path), + "-map", + "0:v:0", + "-vf", + "select=eq(n\\,0)", + "-frames:v", + "1", + "-f", + "image2pipe", + "-vcodec", + "png", + "-", + ], + timeout=timeout, + ) + if not result.stdout: + raise FFmpegEncodingError("FFmpeg decoded no PNG verification frame") + try: + with Image.open(io.BytesIO(result.stdout)) as image: + image.verify() + except Exception as exc: + raise FFmpegEncodingError("FFmpeg returned an invalid PNG verification frame") from exc + return result.stdout + + +def _probe_encoder( + provision: FFmpegProvision, + codec: str, + pixel_format: str, + muxer: str, + *, + timeout: float = PROBE_TIMEOUT_SECONDS, +) -> None: + """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") + command = [ + provision.executable, + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-y", + "-i", + str(source), + "-frames:v", + "1", + "-an", + "-c:v", + codec, + "-pix_fmt", + pixel_format, + *_codec_arguments(codec), + "-f", + muxer, + str(output), + ] + _run_checked(command, timeout=timeout) + 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( + provision, + output, + timeout=timeout, + ) + + +def _automatic_codec_candidates() -> list[str]: + candidates: list[str] = [] + if sys.platform == "darwin": + candidates.append("h264_videotoolbox") + elif sys.platform == "win32": + candidates.append("h264_mf") + candidates.extend([DEFAULT_CODEC, "mpeg4"]) + return candidates + + +def require_video_encoder( + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, + codec: str | None = None, + pixel_format: str | None = None, + muxer: str | None = None, +) -> FFmpegProvision: + """Resolve and prove a usable codec before recording starts. + + A caller- or Desktop-declared codec is an exact contract and fails loudly. + With no declared codec, usable platform encoders are tried in order and the + portable LGPL ``mpeg4`` encoder is the final fallback. + """ + provision = resolve_ffmpeg(ffmpeg_path, ffprobe_path) + exact_codec = codec or provision.codec + candidates = [exact_codec] if exact_codec else _automatic_codec_candidates() + candidates = list(dict.fromkeys(candidate for candidate in candidates if candidate)) + encoders = available_encoders(provision) + failures: list[str] = [] + selected_muxer = _validate_option_token( + "muxer", + muxer or provision.muxer or "mp4", + ) + for candidate in candidates: + candidate = _validate_option_token("codec", candidate) + if candidate not in encoders: + failures.append(f"{candidate}: not advertised") + continue + selected_pixel_format = ( + "yuv420p" + if candidate == "mpeg4" or candidate.startswith(("h264_", "hevc_")) + else pixel_format or provision.pixel_format or DEFAULT_PIXEL_FORMAT + ) + selected_pixel_format = _validate_option_token( + "pixel format", + selected_pixel_format, + ) + try: + _probe_encoder( + provision, + candidate, + selected_pixel_format, + selected_muxer, + ) + except FFmpegEncodingError as exc: + failures.append(f"{candidate}: {exc}") + continue + if not exact_codec and candidate != candidates[0]: + logger.info( + f"Selected verified video encoder {candidate!r} after earlier " + "automatic candidates were unavailable" + ) + return FFmpegProvision( + executable=provision.executable, + ffprobe=provision.ffprobe, + codec=candidate, + pixel_format=selected_pixel_format, + muxer=selected_muxer, + source=provision.source, + ) + + requested = f"requested encoder {exact_codec!r}" if exact_codec else "any encoder" + detail = "; ".join(failures) or "no candidates" + raise FFmpegUnavailableError( + f"FFmpeg from {provision.source} could not actually encode and decode " + f"with {requested}: {provision.executable}. {detail}" + ) + + +class FFmpegFrameStage: + """Lossless timestamped frames awaiting transactional FFmpeg encoding.""" + + def __init__( + self, + output_path: str | Path, + stream: FFmpegVideoStream, + provision: FFmpegProvision, + *, + crf: int = DEFAULT_CRF, + preset: str = DEFAULT_PRESET, + timeout_seconds: float = DEFAULT_PROCESS_TIMEOUT_SECONDS, + ) -> None: + self.output_path = Path(output_path).resolve() + self.output_path.parent.mkdir(parents=True, exist_ok=True) + self.stream = stream + self.provision = provision + 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.frames: list[_StagedFrame] = [] + 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]: + command = [ + self.provision.executable, + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-y", + "-f", + "concat", + "-safe", + "1", + "-i", + manifest.name, + "-an", + "-c:v", + self.stream.codec, + "-pix_fmt", + self.stream.pix_fmt, + "-movflags", + "+faststart", + ] + command.extend( + _codec_arguments( + self.stream.codec, + crf=self.crf, + preset=self.preset, + ) + ) + command.extend(["-fps_mode", "vfr"] if fps_mode else ["-vsync", "vfr"]) + command.extend(["-f", self.stream.muxer]) + command.append(str(output)) + return command + + def close(self) -> None: + """Encode, verify, atomically promote, then remove staging data.""" + with self._lock: + if self._closed: + return + if not self.frames: + 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}" + ) + 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, + ) + + if not temp_output.is_file() or temp_output.stat().st_size == 0: + raise FFmpegEncodingError("FFmpeg returned success without a non-empty output") + _decode_first_frame_png( + self.provision, + temp_output, + timeout=min(self.timeout_seconds, EXTRACT_TIMEOUT_SECONDS), + ) + os.replace(temp_output, 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}") + raise + else: + shutil.rmtree(self.stage_dir) + self._closed = True + + +class VideoWriter: + """MP4 writer backed by a separately provisioned FFmpeg executable.""" def __init__( self, output_path: str | Path, width: int, height: int, - fps: int = 24, - codec: str = "libx264", - pix_fmt: str = "yuv444p", - crf: int = 0, - preset: str = "veryslow", + fps: int = DEFAULT_FPS, + codec: str | None = None, + pix_fmt: str | None = None, + muxer: str | None = None, + crf: int = DEFAULT_CRF, + preset: str = DEFAULT_PRESET, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, + timeout_seconds: float = DEFAULT_PROCESS_TIMEOUT_SECONDS, ) -> None: - """Initialize video writer. - - Args: - output_path: Path to output MP4 file. - width: Video width in pixels. - height: Video height in pixels. - fps: Frames per second (default 24). - codec: Video codec (default libx264). - pix_fmt: Pixel format (default yuv444p for full color). - crf: Constant Rate Factor, 0 for lossless (default 0). - preset: Encoding preset (default veryslow for max compression). - """ - + provision = require_video_encoder( + ffmpeg_path=ffmpeg_path, + ffprobe_path=ffprobe_path, + codec=codec, + pixel_format=pix_fmt, + muxer=muxer, + ) self.output_path = Path(output_path) self.width = width self.height = height self.fps = fps - self.codec = codec - self.pix_fmt = pix_fmt + self.codec = provision.codec or codec or DEFAULT_CODEC + self.pix_fmt = provision.pixel_format or pix_fmt or DEFAULT_PIXEL_FORMAT + self.muxer = provision.muxer or muxer or "mp4" self.crf = crf self.preset = preset - - self._container = None - self._stream = None + self._stream = FFmpegVideoStream( + width=width, + height=height, + average_rate=Fraction(fps), + pix_fmt=self.pix_fmt, + codec=self.codec, + muxer=self.muxer, + ) + self._container = FFmpegFrameStage( + output_path, + self._stream, + provision, + crf=crf, + preset=preset, + timeout_seconds=timeout_seconds, + ) self._start_time: float | None = None - self._last_pts: int = -1 - self._last_frame: "Image.Image" | None = None - self._last_frame_timestamp: float | None = None + self._last_pts = -1 self._lock = threading.Lock() - def _init_stream(self) -> None: - """Initialize the video stream.""" - self._container = av.open(str(self.output_path), mode="w") - self._stream = self._container.add_stream(self.codec, rate=self.fps) - self._stream.width = self.width - self._stream.height = self.height - self._stream.pix_fmt = self.pix_fmt - self._stream.options = {"crf": str(self.crf), "preset": self.preset} - @property def start_time(self) -> float | None: - """Get the start time of the video.""" return self._start_time @property def is_open(self) -> bool: - """Check if writer is open.""" - return self._container is not None + return self._start_time is not None and not self._container._closed def write_frame( self, - image: "Image.Image", + image: "PILImage", timestamp: float, force_key_frame: bool = False, ) -> None: - """Write a frame to the video. - - Args: - image: PIL Image to write. - timestamp: Unix timestamp of the frame. - force_key_frame: Force this frame to be a key frame. - """ + del force_key_frame # FFmpeg makes the first encoded frame a key frame. with self._lock: - if self._container is None: - self._init_stream() + if self._start_time is None: self._start_time = timestamp - - # Convert PIL Image to AVFrame - av_frame = av.VideoFrame.from_image(image) - - # Force key frame if requested - if force_key_frame: - av_frame.pict_type = av.video.frame.PictureType.I - - # Calculate PTS based on elapsed time - time_diff = timestamp - self._start_time - pts = int(time_diff * float(Fraction(self._stream.average_rate))) - - # Ensure monotonically increasing PTS - if pts <= self._last_pts: - pts = self._last_pts + 1 - - av_frame.pts = pts - self._last_pts = pts - - # Encode and write - for packet in self._stream.encode(av_frame): - packet.pts = pts - self._container.mux(packet) - - # Track last frame for finalization - self._last_frame = image - self._last_frame_timestamp = timestamp + self._last_pts = write_video_frame( + self._container, + self._stream, + image, + timestamp, + self._start_time, + self._last_pts, + ) def close(self) -> None: - """Close the video writer and finalize the file. - - This method handles the GIL deadlock issue by closing in a separate thread. - """ with self._lock: - if self._container is None: - return - - # Write a final key frame to ensure clean ending - if self._last_frame is not None and self._last_frame_timestamp is not None: - av_frame = av.VideoFrame.from_image(self._last_frame) - # pict_type 1 = I-frame (key frame) - av_frame.pict_type = av.video.frame.PictureType.I - - time_diff = self._last_frame_timestamp - self._start_time - pts = int(time_diff * float(Fraction(self._stream.average_rate))) - if pts <= self._last_pts: - pts = self._last_pts + 1 - av_frame.pts = pts - - for packet in self._stream.encode(av_frame): - packet.pts = pts - self._container.mux(packet) - - # Flush the stream - for packet in self._stream.encode(): - self._container.mux(packet) - - # Close in separate thread to avoid GIL deadlock - # https://github.com/PyAV-Org/PyAV/issues/1053 - container = self._container - - def close_container() -> None: - container.close() - - close_thread = threading.Thread(target=close_container) - close_thread.start() - close_thread.join() - - self._container = None - self._stream = None + self._container.close() def __enter__(self) -> "VideoWriter": - """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: - """Context manager exit.""" self.close() -# ============================================================================= -# Legacy Functional API (copied from legacy OpenAdapt video.py) -# ============================================================================= - - def get_video_file_path(recording_timestamp: float, video_dir: str = None) -> str: - """Generates a file path for a video recording based on a timestamp. - - Args: - recording_timestamp (float): The timestamp of the recording. - video_dir (str): Directory for video files. If None, uses capture dir. - - Returns: - str: The generated file name for the video recording. - """ if video_dir is None: video_dir = os.path.join(os.getcwd(), "video") os.makedirs(video_dir, exist_ok=True) - return os.path.join( - video_dir, f"oa_recording-{recording_timestamp}.mp4" - ) + return os.path.join(video_dir, f"oa_recording-{recording_timestamp}.mp4") def initialize_video_writer( output_path: str, width: int, height: int, - fps: int = 24, - codec: str = config.VIDEO_ENCODING, - pix_fmt: str = config.VIDEO_PIXEL_FORMAT, - crf: int = 0, - preset: str = "veryslow", -) -> tuple[av.container.OutputContainer, av.stream.Stream, float]: - """Initializes video writer and returns the container, stream, and base timestamp. - - Args: - output_path (str): Path to the output video file. - width (int): Width of the video. - height (int): Height of the video. - fps (int, optional): Frames per second of the video. Defaults to 24. - codec (str, optional): Codec used for encoding the video. - Defaults to 'libx264'. - pix_fmt (str, optional): Pixel format of the video. Defaults to 'yuv420p'. - crf (int, optional): Constant Rate Factor for encoding quality. - Defaults to 0 for lossless. - preset (str, optional): Encoding speed/quality trade-off. - Defaults to 'veryslow' for maximum compression. - - Returns: - tuple[av.container.OutputContainer, av.stream.Stream, float]: The initialized - container, stream, and base timestamp. - """ - logger.info("initializing video stream...") - video_container = av.open(output_path, mode="w") - video_stream = video_container.add_stream(codec, rate=fps) - video_stream.width = width - video_stream.height = height - video_stream.pix_fmt = pix_fmt - video_stream.options = {"crf": str(crf), "preset": preset} - - base_timestamp = utils.get_timestamp() - - return video_container, video_stream, base_timestamp + fps: int = DEFAULT_FPS, + codec: str | None = None, + pix_fmt: str | None = None, + muxer: str | None = None, + crf: int = DEFAULT_CRF, + preset: str = DEFAULT_PRESET, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, + timeout_seconds: float | None = None, + preflight_provision: FFmpegProvision | None = None, +) -> tuple[FFmpegFrameStage, FFmpegVideoStream, float]: + if preflight_provision is None: + selected_path = ffmpeg_path or config.VIDEO_FFMPEG_PATH + provision = require_video_encoder( + ffmpeg_path=selected_path, + ffprobe_path=ffprobe_path or config.VIDEO_FFPROBE_PATH, + codec=codec or config.VIDEO_ENCODING, + pixel_format=pix_fmt or config.VIDEO_PIXEL_FORMAT, + muxer=muxer or config.VIDEO_MUXER, + ) + else: + provision = preflight_provision + if not provision.codec or not provision.pixel_format or not provision.muxer: + raise FFmpegUnavailableError( + "Preflight FFmpeg provision must include codec, pixel format, and muxer" + ) + selected_codec = provision.codec or codec or config.VIDEO_ENCODING or DEFAULT_CODEC + selected_pixel_format = ( + provision.pixel_format or pix_fmt or config.VIDEO_PIXEL_FORMAT or DEFAULT_PIXEL_FORMAT + ) + selected_muxer = provision.muxer or muxer or config.VIDEO_MUXER or "mp4" + stream = FFmpegVideoStream( + width=width, + height=height, + average_rate=Fraction(fps), + pix_fmt=selected_pixel_format, + codec=selected_codec, + muxer=selected_muxer, + ) + container = FFmpegFrameStage( + output_path, + stream, + provision, + crf=crf, + preset=preset, + timeout_seconds=timeout_seconds or config.VIDEO_FFMPEG_TIMEOUT_SECONDS, + ) + return container, stream, utils.get_timestamp() def write_video_frame( - video_container: av.container.OutputContainer, - video_stream: av.stream.Stream, - screenshot: "Image.Image", + video_container: FFmpegFrameStage, + video_stream: FFmpegVideoStream, + screenshot: "PILImage", timestamp: float, video_start_timestamp: float, last_pts: int, force_key_frame: bool = False, ) -> int: - """Encodes and writes a video frame to the output container from a given screenshot. - - This function converts a PIL.Image to an AVFrame, - and encodes it for writing to the video stream. It calculates the - presentation timestamp (PTS) for each frame based on the elapsed time since - the base timestamp, ensuring monotonically increasing PTS values. - - Args: - video_container (av.container.OutputContainer): The output container to which - the frame is written. - video_stream (av.stream.Stream): The video stream within the container. - screenshot (Image.Image): The screenshot to be written as a video frame. - timestamp (float): The timestamp of the current frame. - video_start_timestamp (float): The base timestamp from which the video - recording started. - last_pts (int): The PTS of the last written frame. - force_key_frame (bool): Whether to force this frame to be a key frame. - - Returns: - int: The updated last_pts value, to be used for writing the next frame. - - Note: - - It is crucial to maintain monotonically increasing PTS values for the - video stream's consistency and playback. - - The function logs the current timestamp, base timestamp, and - calculated PTS values for debugging purposes. - """ - # Convert the PIL Image to an AVFrame - av_frame = av.VideoFrame.from_image(screenshot) - - # Optionally force a key frame - # TODO: force key frames on active window change? - if force_key_frame: - av_frame.pict_type = av.video.frame.PictureType.I - - # Calculate the time difference in seconds - time_diff = timestamp - video_start_timestamp - - # Calculate PTS, taking into account the fractional average rate - pts = int(time_diff * float(Fraction(video_stream.average_rate))) - - logger.debug( - f"{timestamp=} {video_start_timestamp=} {time_diff=} {pts=} {force_key_frame=}" - ) - - # Ensure monotonically increasing PTS + del force_key_frame + time_diff = max(timestamp - video_start_timestamp, 0.0) + pts = int(time_diff * float(video_stream.average_rate)) if pts <= last_pts: pts = last_pts + 1 - logger.debug(f"incremented {pts=}") - av_frame.pts = pts - last_pts = pts # Update the last_pts - - # Encode and write the frame - for packet in video_stream.encode(av_frame): - packet.pts = pts - video_container.mux(packet) - - return last_pts # Return the updated last_pts for the next call + video_container.stage_frame(screenshot, pts) + return pts def finalize_video_writer( - video_container: av.container.OutputContainer, - video_stream: av.stream.Stream, + video_container: FFmpegFrameStage, + video_stream: FFmpegVideoStream, video_start_timestamp: float, - last_frame: "Image.Image", + last_frame: "PILImage", last_frame_timestamp: float, last_pts: int, video_file_path: str, fix_moov: bool = False, ) -> None: - """Finalizes the video writer, ensuring all buffered frames are encoded and written. - - Args: - video_container (av.container.OutputContainer): The AV container to finalize. - video_stream (av.stream.Stream): The AV stream to finalize. - video_start_timestamp (float): The base timestamp from which the video - recording started. - last_frame (Image.Image): The last frame that was written (to be written again). - last_frame_timestamp (float): The timestamp of the last frame that was written. - last_pts (int): The last presentation timestamp. - video_file_path (str): The path to the video file. - fix_moov (bool): Whether to move the moov atom to the beginning of the file. - Setting this to True will fix a bug when displaying the video in Github - comments causing the video to appear to start a few seconds after 0:00. - However, this causes extract_frames to fail. - """ - # Closing the container in the main thread leads to a GIL deadlock. - # https://github.com/PyAV-Org/PyAV/issues/1053 - - # Write a final key frame - last_pts = write_video_frame( + del video_file_path + write_video_frame( video_container, video_stream, last_frame, @@ -376,234 +877,293 @@ def finalize_video_writer( last_pts, force_key_frame=True, ) - - # Closing in the same thread sometimes hangs, so do it in a different thread: - - # Define a function to close the container - def close_container() -> None: - logger.info("closing video container...") - video_container.close() - - # Create a new thread to close the container - close_thread = threading.Thread(target=close_container) - - # Flush stream - logger.info("flushing video stream...") - for packet in video_stream.encode(): - video_container.mux(packet) - - # Start the thread to close the container - close_thread.start() - - # Wait for the thread to finish execution - close_thread.join() - - # Move moov atom to beginning of file + video_container.close() if fix_moov: - # TODO: fix this - logger.warning(f"{fix_moov=} will cause extract_frames() to fail!!!") - move_moov_atom(video_file_path) + logger.info("FFmpeg encoding already applied +faststart") - logger.info("done") +def move_moov_atom( + input_file: str, + output_file: str = None, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, +) -> None: + provision = resolve_ffmpeg(ffmpeg_path or config.VIDEO_FFMPEG_PATH) + input_path = Path(input_file) + temp_file: Path | None = None + if output_file is None: + temp_file = input_path.with_name(f".{input_path.name}.{uuid.uuid4().hex}.mp4") + output_path = temp_file + else: + output_path = Path(output_file) + _run_checked( + [ + provision.executable, + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-y", + "-i", + str(input_path), + "-codec", + "copy", + "-movflags", + "faststart", + str(output_path), + ], + timeout=DEFAULT_PROCESS_TIMEOUT_SECONDS, + ) + if temp_file is not None: + os.replace(temp_file, input_path) -def move_moov_atom(input_file: str, output_file: str = None) -> None: - """Moves the moov atom to the beginning of the video file using ffmpeg. - If no output file is specified, modifies the input file in place. +def _require_ffprobe(provision: FFmpegProvision) -> str: + if provision.ffprobe is None: + raise FFmpegUnavailableError( + "Exact frame lookup and video metadata require a separately " + "provisioned ffprobe executable. Put ffprobe beside FFmpeg, set " + "OPENADAPT_FFPROBE_PATH, pass ffprobe_path=..., or declare it in " + "Desktop's ffmpeg.json manifest." + ) + return provision.ffprobe + + +def _probe_json( + provision: FFmpegProvision, + command: Sequence[str], + *, + timeout: float = EXTRACT_TIMEOUT_SECONDS, +) -> dict: + result = _run_checked( + [ + _require_ffprobe(provision), + "-v", + "error", + *command, + "-of", + "json", + ], + timeout=timeout, + ) + try: + payload = json.loads(result.stdout.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise FFmpegEncodingError("ffprobe returned invalid JSON") from exc + if not isinstance(payload, dict): + raise FFmpegEncodingError("ffprobe returned a non-object JSON payload") + return payload + + +def _frame_catalog( + video_path: Path, + provision: FFmpegProvision, +) -> list[tuple[int, float]]: + """Return decoded-frame indexes and timestamps in presentation order.""" + payload = _probe_json( + provision, + [ + "-select_streams", + "v:0", + "-show_frames", + "-show_entries", + "frame=best_effort_timestamp_time,pts_time", + str(video_path), + ], + ) + catalog: list[tuple[int, float]] = [] + for index, frame in enumerate(payload.get("frames", [])): + if not isinstance(frame, dict): + continue + raw = frame.get("best_effort_timestamp_time", frame.get("pts_time")) + try: + timestamp = float(raw) + except (TypeError, ValueError): + continue + catalog.append((index, timestamp)) + return catalog + + +def _extract_frame_index_png( + video_path: Path, + frame_index: int, + provision: FFmpegProvision, +) -> "PILImage": + result = _run_checked( + [ + provision.executable, + "-hide_banner", + "-loglevel", + "error", + "-nostdin", + "-i", + str(video_path), + "-vf", + f"select=eq(n\\,{frame_index})", + "-frames:v", + "1", + "-f", + "image2pipe", + "-vcodec", + "png", + "-", + ], + timeout=EXTRACT_TIMEOUT_SECONDS, + ) + if not result.stdout: + raise ValueError(f"No decoded video frame at index {frame_index}") - Args: - input_file (str): The path to the input MP4 file. - output_file (str, optional): The path to the output MP4 file where the moov - atom is at the beginning. If None, modifies the input file in place. - """ - temp_file = None - if output_file is None: - # Create a temporary file - temp_file = tempfile.NamedTemporaryFile( - delete=False, - suffix=".mp4", - dir=os.path.dirname(input_file), - ).name - output_file = temp_file - - command = [ - "ffmpeg", - "-y", # Automatically overwrite files without asking - "-i", - input_file, - "-codec", - "copy", # Avoid re-encoding; just copy streams - "-movflags", - "faststart", # Move the moov atom to the start - output_file, - ] - logger.info(f"{command=}") - subprocess.run(command, check=True) - - if temp_file: - # Replace the original file with the modified one - os.replace(temp_file, input_file) - - -# ============================================================================= -# Frame Extraction -# ============================================================================= + with Image.open(io.BytesIO(result.stdout)) as image: + return image.convert("RGB").copy() def extract_frames( video_path: str | Path, timestamps: list[float], tolerance: float = 0.1, -) -> list["Image.Image"]: - """Extract frames from a video at specified timestamps. - - Args: - video_path: Path to the video file. - timestamps: List of timestamps (in seconds) to extract. - tolerance: Maximum difference between requested and actual frame time. - - Returns: - List of PIL Images at the requested timestamps. - - Raises: - ValueError: If no frame found within tolerance for a timestamp. - """ - - video_container = av.open(str(video_path)) - video_stream = video_container.streams.video[0] - - # Storage for matched frames - frame_by_timestamp: dict[float, "Image.Image" | None] = {t: None for t in timestamps} - frame_differences: dict[float, float] = {t: float("inf") for t in timestamps} - - # Convert PTS to seconds - time_base = float(video_stream.time_base) - - for frame in video_container.decode(video_stream): - frame_timestamp = frame.pts * time_base - - for target_timestamp in timestamps: - difference = abs(frame_timestamp - target_timestamp) - if difference <= tolerance and difference < frame_differences[target_timestamp]: - frame_by_timestamp[target_timestamp] = frame.to_image() - frame_differences[target_timestamp] = difference - - video_container.close() - - # Check for missing frames - missing = [t for t, frame in frame_by_timestamp.items() if frame is None] + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, +) -> list["PILImage"]: + provision = resolve_ffmpeg( + ffmpeg_path or config.VIDEO_FFMPEG_PATH, + ffprobe_path or config.VIDEO_FFPROBE_PATH, + ) + path = Path(video_path) + catalog = _frame_catalog(path, provision) + selected: list[int] = [] + missing: list[float] = [] + for target in timestamps: + nearest = min( + catalog, + key=lambda item: abs(item[1] - target), + default=None, + ) + if nearest is None or abs(nearest[1] - target) > tolerance: + missing.append(target) + else: + selected.append(nearest[0]) if missing: raise ValueError(f"No frame within tolerance for timestamps: {missing}") - # Return in same order as input - return [frame_by_timestamp[t] for t in timestamps] + images_by_index = { + index: _extract_frame_index_png(path, index, provision) for index in dict.fromkeys(selected) + } + return [images_by_index[index].copy() for index in selected] def extract_frame( video_path: str | Path, timestamp: float, tolerance: float = 0.1, -) -> "Image.Image": - """Extract a single frame from a video. - - Args: - video_path: Path to the video file. - timestamp: Timestamp (in seconds) to extract. - tolerance: Maximum difference between requested and actual frame time. - - Returns: - PIL Image at the requested timestamp. - """ - return extract_frames(video_path, [timestamp], tolerance)[0] - - -def get_video_info(video_path: str | Path) -> dict: - """Get information about a video file. - - Args: - video_path: Path to the video file. - - Returns: - Dictionary with video information (duration, width, height, fps, etc). - """ - - video_container = av.open(str(video_path)) - video_stream = video_container.streams.video[0] - - info = { - "duration": float(video_stream.duration * video_stream.time_base) - if video_stream.duration - else None, - "width": video_stream.width, - "height": video_stream.height, - "fps": float(video_stream.average_rate) if video_stream.average_rate else None, - "codec": video_stream.codec_context.codec.name, - "frames": video_stream.frames, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, +) -> "PILImage": + return extract_frames( + video_path, + [timestamp], + tolerance, + ffmpeg_path=ffmpeg_path, + ffprobe_path=ffprobe_path, + )[0] + + +def get_video_info( + video_path: str | Path, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, +) -> dict: + provision = resolve_ffmpeg( + ffmpeg_path or config.VIDEO_FFMPEG_PATH, + ffprobe_path or config.VIDEO_FFPROBE_PATH, + ) + payload = _probe_json( + provision, + [ + "-select_streams", + "v:0", + "-count_frames", + "-show_streams", + "-show_format", + "-show_entries", + ( + "stream=width,height,codec_name,avg_frame_rate,nb_frames," + "nb_read_frames,duration:format=duration" + ), + str(video_path), + ], + ) + streams = payload.get("streams", []) + stream = streams[0] if streams and isinstance(streams[0], dict) else {} + format_info = payload.get("format", {}) + if not isinstance(format_info, dict): + format_info = {} + + def _number(value, cast): + try: + return cast(value) + except (TypeError, ValueError): + return None + + fps = None + rate = stream.get("avg_frame_rate") + if isinstance(rate, str) and rate not in {"0/0", "N/A"}: + try: + fps = float(Fraction(rate)) + except (ValueError, ZeroDivisionError): + pass + duration = _number(stream.get("duration"), float) + if duration is None: + duration = _number(format_info.get("duration"), float) + frames = _number(stream.get("nb_frames"), int) + if frames is None: + frames = _number(stream.get("nb_read_frames"), int) + return { + "duration": duration, + "width": _number(stream.get("width"), int), + "height": _number(stream.get("height"), int), + "fps": fps, + "codec": stream.get("codec_name"), + "frames": frames, } - video_container.close() - return info - - -# ============================================================================= -# Chunked Video Writer (for long captures) -# ============================================================================= - class ChunkedVideoWriter: - """Video writer that automatically chunks output into segments. - - For long captures (hours/days), this splits the video into manageable - segments to avoid huge files and enable recovery from crashes. - - Usage: - writer = ChunkedVideoWriter( - output_dir="capture_abc123/video", - width=1920, height=1080, - chunk_duration=600, # 10 minutes per chunk - ) - writer.write_frame(image, timestamp) - writer.close() - """ + """Video writer that splits output into time-bounded MP4 chunks.""" def __init__( self, output_dir: str | Path, width: int, height: int, - chunk_duration: float = 600.0, # 10 minutes - fps: int = 24, - codec: str = "libx264", - pix_fmt: str = "yuv444p", - crf: int = 0, - preset: str = "veryslow", + chunk_duration: float = 600.0, + fps: int = DEFAULT_FPS, + codec: str | None = None, + pix_fmt: str | None = None, + muxer: str = "mp4", + crf: int = DEFAULT_CRF, + preset: str = DEFAULT_PRESET, + *, + ffmpeg_path: str | os.PathLike[str] | None = None, + ffprobe_path: str | os.PathLike[str] | None = None, + timeout_seconds: float = DEFAULT_PROCESS_TIMEOUT_SECONDS, ) -> None: - """Initialize chunked video writer. - - Args: - output_dir: Directory for video chunks. - width: Video width in pixels. - height: Video height in pixels. - chunk_duration: Duration of each chunk in seconds. - fps: Frames per second. - codec: Video codec. - pix_fmt: Pixel format. - crf: Constant Rate Factor. - preset: Encoding preset. - """ self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) - self.width = width self.height = height self.chunk_duration = chunk_duration self.fps = fps self.codec = codec self.pix_fmt = pix_fmt + self.muxer = muxer self.crf = crf self.preset = preset - + self.ffmpeg_path = ffmpeg_path + self.ffprobe_path = ffprobe_path + self.timeout_seconds = timeout_seconds self._current_writer: VideoWriter | None = None self._chunk_index = 0 self._chunk_start_time: float | None = None @@ -612,80 +1172,62 @@ def __init__( @property def start_time(self) -> float | None: - """Get the start time of the recording.""" return self._start_time @property def chunk_paths(self) -> list[Path]: - """Get list of all chunk file paths.""" return sorted(self.output_dir.glob("chunk_*.mp4")) def _get_chunk_path(self, index: int) -> Path: - """Get path for a chunk by index.""" return self.output_dir / f"chunk_{index:04d}.mp4" def _start_new_chunk(self, timestamp: float) -> None: - """Start a new video chunk.""" if self._current_writer is not None: self._current_writer.close() - - chunk_path = self._get_chunk_path(self._chunk_index) self._current_writer = VideoWriter( - chunk_path, + self._get_chunk_path(self._chunk_index), width=self.width, height=self.height, fps=self.fps, codec=self.codec, pix_fmt=self.pix_fmt, + muxer=self.muxer, crf=self.crf, preset=self.preset, + ffmpeg_path=self.ffmpeg_path, + ffprobe_path=self.ffprobe_path, + timeout_seconds=self.timeout_seconds, ) self._chunk_start_time = timestamp self._chunk_index += 1 def write_frame( self, - image: "Image.Image", + image: "PILImage", timestamp: float, force_key_frame: bool = False, ) -> None: - """Write a frame, automatically starting new chunks as needed. - - Args: - image: PIL Image to write. - timestamp: Unix timestamp of the frame. - force_key_frame: Force this frame to be a key frame. - """ with self._lock: if self._start_time is None: self._start_time = timestamp - - # Check if we need a new chunk - needs_new_chunk = ( - self._current_writer is None - or ( - self._chunk_start_time is not None - and timestamp - self._chunk_start_time >= self.chunk_duration - ) + needs_new_chunk = self._current_writer is None or ( + self._chunk_start_time is not None + and timestamp - self._chunk_start_time >= self.chunk_duration ) - if needs_new_chunk: self._start_new_chunk(timestamp) - force_key_frame = True # First frame of chunk should be key frame - + force_key_frame = True + assert self._current_writer is not None self._current_writer.write_frame(image, timestamp, force_key_frame) def close(self) -> None: - """Close the current chunk and finalize.""" with self._lock: if self._current_writer is not None: self._current_writer.close() self._current_writer = None def __enter__(self) -> "ChunkedVideoWriter": - """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: - """Context manager exit.""" self.close() diff --git a/pyproject.toml b/pyproject.toml index 0514b0c..9ee580b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,6 @@ classifiers = [ dependencies = [ "pydantic>=2.0.0", - "av>=12.0.0", "Pillow>=10.1.0", "fire>=0.7.1", "mss>=6.0.0", diff --git a/scripts/verify_distribution.py b/scripts/verify_distribution.py index c5a4e99..71e845f 100644 --- a/scripts/verify_distribution.py +++ b/scripts/verify_distribution.py @@ -48,6 +48,23 @@ def verify_distribution(path: Path) -> None: assert f"requires-dist: {dependency}" not in metadata, ( f"{path}: forbidden dependency {dependency!r} is in package metadata" ) + assert "requires-dist: av" not in metadata, ( + f"{path}: PyAV must not be in the package dependency closure" + ) + + forbidden_binary_names = ( + "ffmpeg", + "ffprobe", + "avcodec", + "avformat", + "x264", + "x265", + ) + for name in files: + leaf = Path(name).name.lower() + assert not any(token in leaf for token in forbidden_binary_names), ( + f"{path}: bundled video binary violates the external-process boundary: {name}" + ) python_sources = "\n".join( content.decode("utf-8") diff --git a/tests/test_comparison.py b/tests/test_comparison.py index 0bb89a6..0b76921 100644 --- a/tests/test_comparison.py +++ b/tests/test_comparison.py @@ -176,7 +176,7 @@ def test_compare_video_to_images_nonexistent_video(self, tmp_path): img = Image.new("RGB", (100, 100), color="red") images = [(1.0, img)] - with pytest.raises(Exception): # FileNotFoundError or av error + with pytest.raises(Exception): # FileNotFoundError or decoder error compare_video_to_images(tmp_path / "nonexistent.mp4", images) diff --git a/tests/test_highlevel.py b/tests/test_highlevel.py index 163f72f..25322a8 100644 --- a/tests/test_highlevel.py +++ b/tests/test_highlevel.py @@ -3,6 +3,7 @@ Updated for legacy-style SQLAlchemy storage. """ +import multiprocessing import tempfile import threading import time @@ -11,6 +12,7 @@ import pytest from openadapt_capture import recorder as recorder_module +from openadapt_capture import video from openadapt_capture.capture import Capture from openadapt_capture.db import create_db, crud from openadapt_capture.recorder import Recorder @@ -177,6 +179,36 @@ def wait_for_shutdown(): assert time.monotonic() - stop_started < 1 assert recorder.wait_for_ready(timeout=0) is False + def test_spawned_video_writer_failure_surfaces_through_recorder( + self, monkeypatch, tmp_path + ): + """A child finalization failure cannot be reported as a stopped capture.""" + + def record_with_failed_video_writer(**_kwargs): + process = multiprocessing.get_context("spawn").Process( + target=video._run_checked, + args=(["/definitely-missing-openadapt-ffmpeg"],), + kwargs={"timeout": 1}, + ) + process.start() + tasks = {"video_writer": process} + recorder_module._join_tasks(tasks, ["video_writer"]) + recorder_module._raise_for_failed_processes(tasks) + + monkeypatch.setattr( + recorder_module, + "record", + record_with_failed_video_writer, + ) + recorder = Recorder(str(tmp_path / "failed-finalization")) + + with pytest.raises( + RuntimeError, + match=r"video_writer \(exit code 1\)", + ): + with recorder: + pass + class TestCapture: """Tests for Capture/CaptureSession class.""" diff --git a/tests/test_runtime_import_contract.py b/tests/test_runtime_import_contract.py index b46dbd6..6527114 100644 --- a/tests/test_runtime_import_contract.py +++ b/tests/test_runtime_import_contract.py @@ -53,6 +53,21 @@ def test_native_input_observers_do_not_ship_pynput() -> None: assert "pynput" not in package_sources +def test_external_ffmpeg_keeps_core_recorder_video_first_and_pyav_free() -> None: + """Video stays default without linking PyAV/FFmpeg into the MIT package.""" + assert "av" not in _runtime_dependency_names() + package_sources = "\n".join( + path.read_text(encoding="utf-8") + for path in (ROOT / "openadapt_capture").rglob("*.py") + ) + assert "import av" not in package_sources + + from openadapt_capture.config import config + + assert config.RECORD_VIDEO is True + assert config.RECORD_IMAGES is False + + def test_default_install_exposes_recorder() -> None: """Recorder import never fails because a runtime dependency is undeclared.""" result = subprocess.run( @@ -78,3 +93,36 @@ def test_default_install_exposes_recorder() -> None: "Recorder depended on an undeclared default-runtime package.\n" f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" ) + + +def test_default_package_import_and_recorder_construction_do_not_import_av() -> None: + """The video-first API remains importable without in-process PyAV.""" + script = """ +import importlib.abc +import sys + +class BlockAv(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname == "av" or fullname.startswith("av."): + raise ModuleNotFoundError("blocked optional av", name="av") + return None + +sys.meta_path.insert(0, BlockAv()) +from openadapt_capture import Recorder +recorder = Recorder("/tmp/openadapt-capture-core-import-contract") +assert recorder._recording_config.capture_video is None +assert recorder._recording_config.capture_images is None +assert "av" not in sys.modules +print("CORE_RECORDER_OK") +""" + result = subprocess.run( + [sys.executable, "-c", script], + cwd=ROOT, + capture_output=True, + text=True, + ) + assert result.returncode == 0, ( + "The default Recorder path imported optional PyAV.\n" + f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}" + ) + assert "CORE_RECORDER_OK" in result.stdout diff --git a/tests/test_video.py b/tests/test_video.py index 9c552d1..798dcc1 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -1,57 +1,646 @@ -"""Tests for video module.""" +"""Contracts for the external-process video encoder boundary.""" -import tempfile +from __future__ import annotations + +import io +import json +import multiprocessing +import shutil +import subprocess import time +from fractions import Fraction +from pathlib import Path -import av import pytest from PIL import Image -from openadapt_capture import utils -from openadapt_capture.video import ( - initialize_video_writer, - write_video_frame, -) +from openadapt_capture import recorder as recorder_module +from openadapt_capture import utils, video @pytest.fixture(autouse=True) def _init_timestamp(): - """Ensure utils timestamp system is initialized.""" utils.set_start_time(time.time()) -class TestWriteVideoFrame: - """Tests for write_video_frame.""" +def _provision(path: Path, codec: str = "libx264") -> video.FFmpegProvision: + return video.FFmpegProvision(str(path), codec=codec, source="test") - def test_write_frame_basic(self): - """Test writing a basic video frame.""" - with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f: - container, stream, start_ts = initialize_video_writer( - f.name, 100, 100 - ) - img = Image.new("RGB", (100, 100), color="red") - last_pts = write_video_frame( - container, stream, img, start_ts + 0.1, start_ts, 0 - ) - assert last_pts > 0 - container.close() - - def test_write_frame_force_key_frame(self): - """Test writing a video frame with force_key_frame=True.""" - with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f: - container, stream, start_ts = initialize_video_writer( - f.name, 100, 100 - ) - img = Image.new("RGB", (100, 100), color="blue") - last_pts = write_video_frame( - container, stream, img, start_ts + 0.1, start_ts, 0, - force_key_frame=True, + +def _stream(codec: str = "libx264") -> video.FFmpegVideoStream: + return video.FFmpegVideoStream( + width=100, + height=80, + average_rate=Fraction(24), + pix_fmt="yuv444p", + codec=codec, + muxer="mp4", + ) + + +def _png_bytes(color: str = "black") -> bytes: + output = io.BytesIO() + Image.new("RGB", (2, 2), color).save(output, format="PNG") + return output.getvalue() + + +def _spawn_initialize_video_writer( + output_path: str, + provision: video.FFmpegProvision, +) -> None: + """Spawn-safe target for exercising the writer's serialized preflight.""" + utils.set_start_time(time.time()) + container, _, _ = video.initialize_video_writer( + output_path, + 2, + 2, + timeout_seconds=1, + preflight_provision=provision, + ) + container.close() + + +def test_resolve_explicit_path_precedes_other_sources(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + ffprobe = tmp_path / "ffprobe" + executable.write_bytes(b"fake") + ffprobe.write_bytes(b"fake") + monkeypatch.setenv("OPENADAPT_DESKTOP_FFMPEG_PATH", str(tmp_path / "missing")) + + provision = video.resolve_ffmpeg(executable) + + assert provision.executable == str(executable.resolve()) + assert provision.ffprobe == str(ffprobe.resolve()) + assert provision.source == "explicit path" + + +def test_desktop_manifest_declares_codec_pixel_format_and_muxer(tmp_path, monkeypatch): + root = tmp_path / "desktop-data" + binary = root / "bin" / "ffmpeg" + ffprobe = root / "bin" / "ffprobe" + binary.parent.mkdir(parents=True) + binary.write_bytes(b"fake") + ffprobe.write_bytes(b"fake") + (root / "ffmpeg.json").write_text( + json.dumps( + { + "version": 1, + "executable": "bin/ffmpeg", + "ffprobe": "bin/ffprobe", + "codec": "h264_videotoolbox", + "pixel_format": "yuv420p", + "muxer": "mp4", + } + ), + encoding="utf-8", + ) + monkeypatch.delenv("OPENADAPT_FFMPEG_PATH", raising=False) + monkeypatch.delenv("OPENADAPT_DESKTOP_FFMPEG_PATH", raising=False) + monkeypatch.setattr(video, "_desktop_data_dirs", lambda: [root]) + + provision = video.resolve_ffmpeg() + + assert provision.executable == str(binary.resolve()) + assert provision.ffprobe == str(ffprobe.resolve()) + assert provision.codec == "h264_videotoolbox" + assert provision.pixel_format == "yuv420p" + assert provision.muxer == "mp4" + + +def test_desktop_manifest_cannot_escape_user_data_root(tmp_path, monkeypatch): + root = tmp_path / "desktop-data" + root.mkdir() + (root / "ffmpeg.json").write_text( + json.dumps({"version": 1, "executable": "../ffmpeg", "codec": "mpeg4"}), + encoding="utf-8", + ) + monkeypatch.delenv("OPENADAPT_FFMPEG_PATH", raising=False) + monkeypatch.delenv("OPENADAPT_DESKTOP_FFMPEG_PATH", raising=False) + monkeypatch.setattr(video, "_desktop_data_dirs", lambda: [root]) + + with pytest.raises(video.FFmpegUnavailableError, match="escapes"): + video.resolve_ffmpeg() + + +def test_desktop_manifest_rejects_pathlike_muxer_token(tmp_path, monkeypatch): + root = tmp_path / "desktop-data" + binary = root / "bin" / "ffmpeg" + binary.parent.mkdir(parents=True) + binary.write_bytes(b"fake") + (root / "ffmpeg.json").write_text( + json.dumps( + { + "version": 1, + "executable": "bin/ffmpeg", + "codec": "mpeg4", + "muxer": "../../escape", + } + ), + encoding="utf-8", + ) + monkeypatch.delenv("OPENADAPT_FFMPEG_PATH", raising=False) + monkeypatch.delenv("OPENADAPT_DESKTOP_FFMPEG_PATH", raising=False) + monkeypatch.setattr(video, "_desktop_data_dirs", lambda: [root]) + + with pytest.raises(video.FFmpegUnavailableError, match="muxer token"): + video.resolve_ffmpeg() + + +def test_require_video_encoder_probes_exact_selected_codec(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + monkeypatch.setattr( + video, + "available_encoders", + lambda _provision: {"mpeg4", "h264_videotoolbox"}, + ) + probed: list[tuple[str, str]] = [] + monkeypatch.setattr( + video, + "_probe_encoder", + lambda _provision, codec, pixel_format, _muxer: probed.append((codec, pixel_format)), + ) + + provision = video.require_video_encoder( + ffmpeg_path=executable, + codec="mpeg4", + ) + assert provision.codec == "mpeg4" + assert provision.pixel_format == "yuv420p" + assert probed == [("mpeg4", "yuv420p")] + + with pytest.raises(video.FFmpegUnavailableError, match="libx264"): + video.require_video_encoder( + ffmpeg_path=executable, + codec="libx264", + ) + + +def test_automatic_encoder_uses_real_probe_then_mpeg4_fallback(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + monkeypatch.setattr( + video, + "_automatic_codec_candidates", + lambda: ["h264_videotoolbox", "mpeg4"], + ) + monkeypatch.setattr( + video, + "available_encoders", + lambda _provision: {"h264_videotoolbox", "mpeg4"}, + ) + probes: list[tuple[str, str]] = [] + + def probe(_provision, codec, pixel_format, _muxer): + probes.append((codec, pixel_format)) + if codec == "h264_videotoolbox": + raise video.FFmpegEncodingError("VideoToolbox failed: -12908") + + monkeypatch.setattr(video, "_probe_encoder", probe) + + provision = video.require_video_encoder(ffmpeg_path=executable) + + assert provision.codec == "mpeg4" + assert provision.pixel_format == "yuv420p" + assert probes == [ + ("h264_videotoolbox", "yuv420p"), + ("mpeg4", "yuv420p"), + ] + + +def test_explicit_encoder_probe_failure_does_not_silently_change_codec(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + monkeypatch.setattr( + video, + "available_encoders", + lambda _provision: {"h264_videotoolbox", "mpeg4"}, + ) + + def fail(*_args, **_kwargs): + raise video.FFmpegEncodingError("hardware unavailable") + + monkeypatch.setattr(video, "_probe_encoder", fail) + + with pytest.raises(video.FFmpegUnavailableError, match="hardware unavailable"): + video.require_video_encoder( + ffmpeg_path=executable, + codec="h264_videotoolbox", + ) + + +def test_run_checked_is_bounded_non_shell(monkeypatch): + captured = {} + + def fake_run(command, **kwargs): + captured["command"] = command + captured["kwargs"] = kwargs + return subprocess.CompletedProcess(command, 0, stdout=b"ok", stderr=b"") + + monkeypatch.setattr(video.subprocess, "run", fake_run) + result = video._run_checked(["/tmp/ffmpeg", "-version"], timeout=7) + + assert result.stdout == b"ok" + assert captured["command"] == ["/tmp/ffmpeg", "-version"] + assert captured["kwargs"]["shell"] is False + assert captured["kwargs"]["timeout"] == 7 + + +def test_run_checked_timeout_is_fail_loud(monkeypatch): + def timeout(*args, **kwargs): + raise subprocess.TimeoutExpired(args[0], kwargs["timeout"]) + + monkeypatch.setattr(video.subprocess, "run", timeout) + with pytest.raises(video.FFmpegEncodingError, match="timed out"): + video._run_checked(["/tmp/ffmpeg"], timeout=3) + + +def test_staged_pts_become_exact_ffconcat_durations(tmp_path): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _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", + ] + + +def test_successful_encode_is_verified_promoted_and_cleans_stage(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _stream(), + _provision(executable), + ) + stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) + commands: list[list[str]] = [] + + 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.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) + + +def test_failed_encode_retains_stage_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, + _stream(), + _provision(executable), + ) + stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) + + def fail(*_args, **_kwargs): + raise video.FFmpegEncodingError("encoder failed") + + monkeypatch.setattr(video, "_run_checked", fail) + 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 not output.exists() + + +def test_fps_mode_compatibility_fallback_is_narrow_and_transactional(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + executable.write_bytes(b"fake") + output = tmp_path / "capture.mp4" + stage = video.FFmpegFrameStage( + output, + _stream(), + _provision(executable), + ) + stage.stage_frame(Image.new("RGB", (100, 80), "red"), 0) + commands: list[list[str]] = [] + + 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'" ) - assert last_pts > 0 - container.close() - - def test_pict_type_enum(self): - """Test that PictureType.I is valid for pict_type assignment.""" - frame = av.VideoFrame(100, 100, "rgb24") - frame.pict_type = av.video.frame.PictureType.I - assert frame.pict_type == av.video.frame.PictureType.I + 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"") + + monkeypatch.setattr(video, "_run_checked", fake_run) + 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) + + +def test_record_refuses_missing_encoder_before_display_or_listeners(tmp_path, monkeypatch): + display_touched = False + + def fail_preflight(**_kwargs): + raise video.FFmpegUnavailableError("no provisioned encoder") + + def touch_display(): + nonlocal display_touched + display_touched = True + raise AssertionError("display must not be touched") + + monkeypatch.setattr(video, "require_video_encoder", fail_preflight) + monkeypatch.setattr(recorder_module.utils, "take_screenshot", touch_display) + + with pytest.raises(video.FFmpegUnavailableError, match="no provisioned"): + recorder_module.record("test", capture_dir=str(tmp_path / "capture")) + assert display_touched is False + + +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") + provision = video.FFmpegProvision( + executable=str(executable), + ffprobe=None, + codec="mpeg4", + pixel_format="yuv420p", + muxer="mp4", + source="parent preflight", + ) + monkeypatch.setenv("PATH", "") + monkeypatch.setenv("OPENADAPT_PLATFORM_OVERRIDE", "linux") + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "no-desktop-runtime")) + monkeypatch.delenv("OPENADAPT_FFMPEG_PATH", raising=False) + monkeypatch.delenv("OPENADAPT_DESKTOP_FFMPEG_PATH", raising=False) + + process = multiprocessing.get_context("spawn").Process( + target=_spawn_initialize_video_writer, + args=(str(tmp_path / "spawned.mp4"), provision), + ) + process.start() + process.join(timeout=15) + + assert not process.is_alive() + assert process.exitcode == 0 + + +def test_extract_frames_preserves_nearest_within_tolerance_and_order(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + ffprobe = tmp_path / "ffprobe" + video_path = tmp_path / "capture.mp4" + for path in (executable, ffprobe, video_path): + path.write_bytes(b"fake") + provision = video.FFmpegProvision( + str(executable), + ffprobe=str(ffprobe), + source="test", + ) + monkeypatch.setattr(video, "resolve_ffmpeg", lambda *_args: provision) + monkeypatch.setattr( + video, + "_frame_catalog", + lambda *_args: [(0, 0.0), (1, 0.4), (2, 1.0)], + ) + monkeypatch.setattr( + video, + "_extract_frame_index_png", + lambda _path, index, _provision: Image.new("RGB", (2, 2), (index, 0, 0)), + ) + + frames = video.extract_frames( + video_path, + [0.39, 0.01, 0.75], + tolerance=0.26, + ) + + assert [frame.getpixel((0, 0))[0] for frame in frames] == [1, 0, 2] + + with pytest.raises(ValueError, match=r"0\.75"): + video.extract_frames(video_path, [0.75], tolerance=0.24) + + +def test_extract_frames_tie_keeps_earlier_decoded_frame(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + ffprobe = tmp_path / "ffprobe" + video_path = tmp_path / "capture.mp4" + for path in (executable, ffprobe, video_path): + path.write_bytes(b"fake") + provision = video.FFmpegProvision( + str(executable), + ffprobe=str(ffprobe), + source="test", + ) + monkeypatch.setattr(video, "resolve_ffmpeg", lambda *_args: provision) + monkeypatch.setattr( + video, + "_frame_catalog", + lambda *_args: [(0, 0.0), (1, 1.0)], + ) + selected: list[int] = [] + + def extract(_path, index, _provision): + selected.append(index) + return Image.new("RGB", (1, 1)) + + monkeypatch.setattr(video, "_extract_frame_index_png", extract) + + video.extract_frame(video_path, 0.5, tolerance=0.5) + assert selected == [0] + + +def test_get_video_info_preserves_metadata_contract(tmp_path, monkeypatch): + executable = tmp_path / "ffmpeg" + ffprobe = tmp_path / "ffprobe" + video_path = tmp_path / "capture.mp4" + for path in (executable, ffprobe, video_path): + path.write_bytes(b"fake") + provision = video.FFmpegProvision( + str(executable), + ffprobe=str(ffprobe), + source="test", + ) + monkeypatch.setattr(video, "resolve_ffmpeg", lambda *_args: provision) + monkeypatch.setattr( + video, + "_probe_json", + lambda *_args: { + "streams": [ + { + "width": 1920, + "height": 1080, + "codec_name": "mpeg4", + "avg_frame_rate": "24/1", + "duration": "2.5", + "nb_frames": "60", + } + ], + "format": {"duration": "2.6"}, + }, + ) + + assert video.get_video_info(video_path) == { + "duration": 2.5, + "width": 1920, + "height": 1080, + "fps": 24.0, + "codec": "mpeg4", + "frames": 60, + } + + +def _real_ffmpeg_with_codec(codec: str) -> str | None: + executable = shutil.which("ffmpeg") + if not executable: + return None + try: + provision = video.FFmpegProvision(executable, source="test PATH") + if codec not in video.available_encoders(provision): + return None + except (video.FFmpegEncodingError, OSError): + return None + return executable + + +@pytest.mark.skipif( + _real_ffmpeg_with_codec("libx264") is None, + reason="opt-in real FFmpeg/libx264 executable is not available", +) +def test_real_external_ffmpeg_encode_verify_and_extract(tmp_path): + executable = _real_ffmpeg_with_codec("libx264") + assert executable is not None + output = tmp_path / "real.mp4" + container, stream, start = video.initialize_video_writer( + str(output), + 100, + 80, + ffmpeg_path=executable, + timeout_seconds=60, + ) + red = Image.new("RGB", (100, 80), "red") + blue = Image.new("RGB", (100, 80), "blue") + last_pts = video.write_video_frame( + container, + stream, + red, + start, + start, + -1, + ) + last_pts = video.write_video_frame( + container, + stream, + blue, + start + 1, + start, + last_pts, + ) + video.finalize_video_writer( + container, + stream, + start, + blue, + start + 1, + last_pts, + str(output), + ) + + assert output.stat().st_size > 0 + frame = video.extract_frame(output, 0, ffmpeg_path=executable) + assert frame.size == (100, 80) + assert frame.getpixel((10, 10))[0] > frame.getpixel((10, 10))[2] + + +@pytest.mark.skipif( + _real_ffmpeg_with_codec("mpeg4") is None, + reason="opt-in real FFmpeg/mpeg4 executable is not available", +) +def test_real_external_mpeg4_preserves_metadata_and_nearest_frame(tmp_path): + executable = _real_ffmpeg_with_codec("mpeg4") + assert executable is not None + output = tmp_path / "portable.mp4" + container, stream, start = video.initialize_video_writer( + str(output), + 100, + 80, + codec="mpeg4", + ffmpeg_path=executable, + timeout_seconds=60, + ) + assert stream.pix_fmt == "yuv420p" + red = Image.new("RGB", (100, 80), "red") + blue = Image.new("RGB", (100, 80), "blue") + first_pts = video.write_video_frame( + container, + stream, + red, + start, + start, + -1, + ) + last_pts = video.write_video_frame( + container, + stream, + blue, + start + 1, + start, + first_pts, + ) + video.finalize_video_writer( + container, + stream, + start, + blue, + start + 1, + last_pts, + str(output), + ) + + info = video.get_video_info(output, ffmpeg_path=executable) + assert info["codec"] == "mpeg4" + assert info["width"] == 100 + assert info["height"] == 80 + assert info["frames"] >= 2 + frame = video.extract_frame( + output, + 0.8, + tolerance=0.25, + ffmpeg_path=executable, + ) + assert frame.getpixel((10, 10))[2] > frame.getpixel((10, 10))[0]