From 1cb4df0f6d4c07e3d3186bcf062a08a46e914c97 Mon Sep 17 00:00:00 2001 From: Tao Wen Date: Sun, 9 Aug 2026 20:11:52 +0800 Subject: [PATCH] feat(media): add ffprobe inspection - execute ffprobe without a shell and map missing, timeout, process, and parse failures - parse exact duration, container and stream codecs, bitrate, frame rate, rotation, color, audio, and creation metadata - expose scriptable semanticvideo inspect JSON output with timeout and executable controls - document the inspection boundary and update the generated JSON Schema and roadmap - validate fixture variants and the real CLI with a synthetic FFmpeg video across 57 tests --- CHANGELOG.md | 3 +- README.md | 28 +- ROADMAP.md | 21 +- docs/adr/0006-ffprobe-inspection-boundary.md | 31 ++ docs/architecture.md | 15 +- docs/media-inspection.md | 75 ++++ pyproject.toml | 4 + semanticvideo.schema.json | 49 ++- src/semanticvideo/__init__.py | 3 + src/semanticvideo/cli/__init__.py | 1 + src/semanticvideo/cli/main.py | 86 ++++ src/semanticvideo/errors.py | 31 ++ src/semanticvideo/media/__init__.py | 5 + src/semanticvideo/media/ffprobe.py | 371 ++++++++++++++++++ src/semanticvideo/schema/media.py | 5 +- tests/fixtures/ffprobe/standard_mp4.json | 56 +++ .../ffprobe/video_only_duration_ts.json | 28 ++ tests/test_cli.py | 85 ++++ tests/test_ffprobe.py | 276 +++++++++++++ tests/test_media_integration.py | 64 +++ 20 files changed, 1210 insertions(+), 27 deletions(-) create mode 100644 docs/adr/0006-ffprobe-inspection-boundary.md create mode 100644 docs/media-inspection.md create mode 100644 src/semanticvideo/cli/__init__.py create mode 100644 src/semanticvideo/cli/main.py create mode 100644 src/semanticvideo/errors.py create mode 100644 src/semanticvideo/media/__init__.py create mode 100644 src/semanticvideo/media/ffprobe.py create mode 100644 tests/fixtures/ffprobe/standard_mp4.json create mode 100644 tests/fixtures/ffprobe/video_only_duration_ts.json create mode 100644 tests/test_cli.py create mode 100644 tests/test_ffprobe.py create mode 100644 tests/test_media_integration.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 86f65b9..bbe177c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,4 +12,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - Repository foundation, documentation, tests, and continuous integration. - Initial provider-neutral SemanticVideo schema and JSON Schema export. - Example Japan trip semantic manifest. - +- Deterministic `ffprobe` media inspection and `semanticvideo inspect` CLI. +- Fixture-driven parser coverage and a synthetic video integration test. diff --git a/README.md b/README.md index 0954d22..11fc811 100644 --- a/README.md +++ b/README.md @@ -44,21 +44,24 @@ came from. ## Current scope -Milestones 0 and 1 establish the repository and the schema foundation: +Milestones 0 through 2 establish the schema foundation and technical inspection: - Pydantic models for media, streams, exact time, segments, annotations, entities, evidence, and provenance - JSON serialization and JSON Schema export - cross-reference and temporal validation - an example `.semantic.json` manifest +- deterministic `ffprobe` inspection for real video files +- a scriptable `semanticvideo inspect` command with JSON output - tests, linting, typing, CI, documentation, and architectural decisions -Media analysis, AI providers, search, EditPlan, OpenTimelineIO, and FFmpeg -rendering are intentionally scheduled for later milestones. +Shot detection, semantic AI providers, search, EditPlan, OpenTimelineIO, and +FFmpeg rendering are intentionally scheduled for later milestones. ## Quick start -Requirements: Python 3.12+ and [`uv`](https://docs.astral.sh/uv/). +Requirements: Python 3.12+, [`uv`](https://docs.astral.sh/uv/), and FFmpeg's +`ffprobe` executable for media inspection. ```bash uv sync --all-groups @@ -68,6 +71,17 @@ uv run mypy src uv run semanticvideo-schema --output semanticvideo.schema.json ``` +Inspect a real video without invoking an AI model: + +```bash +uv run semanticvideo inspect GX010231.MP4 +uv run semanticvideo inspect GX010231.MP4 --output GX010231.inspect.json +``` + +The command reports source identity, exact duration, container, bitrate, video/audio/ +subtitle streams, codecs, dimensions, frame rate, time base, rotation, color +metadata, audio layout, language, timestamps, and filesystem facts as JSON. + Load and validate a manifest: ```python @@ -81,8 +95,9 @@ document = SemanticVideoDocument.model_validate_json( print(document.media.duration.seconds) ``` -See [the semantic format](docs/semantic-format.md), -[architecture](docs/architecture.md), and [roadmap](ROADMAP.md) for details. +See [media inspection](docs/media-inspection.md), +[the semantic format](docs/semantic-format.md), [architecture](docs/architecture.md), +and [roadmap](ROADMAP.md) for details. ## Project status @@ -98,4 +113,3 @@ example `feat(schema): add exact time ranges`. ## License Licensed under the [Apache License 2.0](LICENSE). - diff --git a/ROADMAP.md b/ROADMAP.md index 1be3231..262cdf1 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -10,19 +10,21 @@ must not compromise the provider-neutral schema foundation. - **Milestone 1 — Core schema:** exact time, media identity, streams, structural segments, typed annotations, entities, evidence, provenance, document validation, JSON serialization, and JSON Schema export. +- **Milestone 2 — Media inspection:** safe `ffprobe` execution, pure JSON + parsing, filesystem identity, technical stream metadata, scriptable CLI, + fixtures, and a synthetic video integration test. ## Next milestones -1. **Media inspection:** parse deterministic `ffprobe` output. -2. **Vertical editing slice:** manually authored manifests to a minimal +1. **Vertical editing slice:** manually authored manifests to a minimal EditPlan and deterministic FFmpeg cut/concatenate renderer. -3. **Frame extraction and shot detection:** modular sampling and boundaries. -4. **Representative frames and signal quality:** local deterministic metrics. -5. **Timed transcription:** provider interface and one reference adapter. -6. **Structured visual semantics:** provider-neutral VLM adapter contracts. -7. **Semantic retrieval:** embeddings behind a replaceable local index. -8. **Editorial interchange:** validated EditPlan and OpenTimelineIO export. -9. **Japan trip demo:** semantic selection and a human-reviewable rough cut. +2. **Frame extraction and shot detection:** modular sampling and boundaries. +3. **Representative frames and signal quality:** local deterministic metrics. +4. **Timed transcription:** provider interface and one reference adapter. +5. **Structured visual semantics:** provider-neutral VLM adapter contracts. +6. **Semantic retrieval:** embeddings behind a replaceable local index. +7. **Editorial interchange:** validated EditPlan and OpenTimelineIO export. +8. **Japan trip demo:** semantic selection and a human-reviewable rough cut. ## Future exploration @@ -30,4 +32,3 @@ OCR, face and speaker identity, GPS fusion, landmark recognition, audio events, story segmentation, visual similarity, rights policies, content credentials, professional editor integrations, an MCP server, and embedded container metadata remain future ideas, not current commitments. - diff --git a/docs/adr/0006-ffprobe-inspection-boundary.md b/docs/adr/0006-ffprobe-inspection-boundary.md new file mode 100644 index 0000000..2c4ddc4 --- /dev/null +++ b/docs/adr/0006-ffprobe-inspection-boundary.md @@ -0,0 +1,31 @@ +# 0006: Separate ffprobe execution from metadata parsing + +- Status: Accepted +- Date: 2026-08-09 + +## Context + +Technical inspection depends on an external executable, untrusted media paths, +and ffprobe JSON that varies by container and stream. Tests should not require +large binary fixtures or a particular local FFmpeg installation. + +## Decision + +Invoke ffprobe without a shell, with fixed arguments, captured diagnostics, and +a timeout. Keep its JSON-to-schema conversion in a pure parser. Combine local +filesystem facts only in the higher-level inspection function. + +## Alternatives + +- Parse human-readable ffprobe console output +- Couple subprocess execution and schema conversion in one function +- Use OpenCV as the authoritative technical metadata source +- Require a Python FFmpeg wrapper dependency + +## Consequences + +Fixture tests are fast and deterministic, command injection through filenames +is avoided, and other transports can reuse the parser. The project still +depends on an installed ffprobe executable for real inspection and must handle +version-specific fields conservatively. + diff --git a/docs/architecture.md b/docs/architecture.md index e9c3f75..a9b6886 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -30,6 +30,20 @@ Replaceable analyzers ---> SemanticVideo manifest None of these representations substitutes for the others. +## Media inspection boundary + +Milestone 2 separates three responsibilities: + +1. `run_ffprobe` invokes a configured executable with a fixed argument list, + no shell, a timeout, and captured diagnostics. +2. `parse_ffprobe_json` is a deterministic pure parser that can be tested from + checked-in fixtures without an FFmpeg installation. +3. `inspect_media` validates the local path, combines filesystem identity with + parsed metadata, and returns a core `MediaInfo` model. + +The command-line interface only formats this validated model. It does not +duplicate media parsing rules. + ## Core document `SemanticVideoDocument` is the aggregate consistency boundary for one source @@ -72,4 +86,3 @@ artifact references instead of bloating human-readable JSON. Breaking core changes require a schema version change, fixtures, migration notes, and an ADR. Provider-specific fields do not belong in core models. - diff --git a/docs/media-inspection.md b/docs/media-inspection.md new file mode 100644 index 0000000..c013057 --- /dev/null +++ b/docs/media-inspection.md @@ -0,0 +1,75 @@ +# Media inspection + +Milestone 2 extracts deterministic technical metadata from a real media file. +It does not detect shots or describe the visible content. + +## Requirements + +Install FFmpeg and ensure `ffprobe` is on `PATH`, or pass its executable path: + +```bash +semanticvideo inspect input.mp4 --ffprobe /path/to/ffprobe +``` + +SemanticVideo invokes ffprobe with a fixed argument list and without a shell: + +```text +ffprobe -v error -print_format json -show_format -show_streams INPUT +``` + +The default timeout is 60 seconds and can be changed with `--timeout`. + +## Usage + +Pretty JSON is written to standard output: + +```bash +semanticvideo inspect GX010231.MP4 +``` + +For pipelines or files: + +```bash +semanticvideo inspect GX010231.MP4 --compact +semanticvideo inspect GX010231.MP4 --output GX010231.inspect.json +``` + +Expected media, ffprobe, parse, and output errors use exit code 1 and a concise +diagnostic on standard error. Argument errors use argparse's exit code 2. + +## Extracted information + +The validated `MediaInfo` result includes: + +- deterministic asset ID derived from the resolved file URI +- original input URI, file size, and filesystem modification time +- exact positive duration represented as integer ticks +- embedded creation time, container format, and bitrate when present +- video codec and bitrate, dimensions, pixel format, frame rate, time base, display + rotation, sample aspect ratio, color metadata, and a conservative VFR hint +- audio codec and bitrate, sample rate, channel count/layout, language, and time base +- subtitle codec and language +- string-valued container tags for later evidence or identity decisions + +Data and attachment streams are ignored in v0.1 because the core schema does +not yet model them. + +## Duration fallback + +The parser uses container duration when positive. If unavailable, it selects +the longest positive stream duration. A stream may express duration directly +in seconds or as `duration_ts * time_base`. Missing or non-positive duration is +an error because the core media schema requires a bounded timeline. + +## Frame-rate limitation + +`variable_frame_rate` compares ffprobe's average and nominal frame rates. A +difference is useful evidence of VFR, but equality does not prove that every +frame interval is constant. Later analysis may inspect packet timestamps when +an editing workflow requires stronger guarantees. + +## Identity and hashing + +The asset ID is deterministic for a resolved local URI; it is not a content +hash. M2 deliberately avoids hashing large video files. A future cache strategy +will define fast fingerprints and optional SHA-256 verification separately. diff --git a/pyproject.toml b/pyproject.toml index 324afa9..345f4cb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,7 @@ Issues = "https://github.com/TristinOrg/SemanticVideo/issues" Repository = "https://github.com/TristinOrg/SemanticVideo.git" [project.scripts] +semanticvideo = "semanticvideo.cli.main:main" semanticvideo-schema = "semanticvideo.schema.export:main" [dependency-groups] @@ -49,6 +50,9 @@ packages = ["src/semanticvideo"] [tool.pytest.ini_options] addopts = "--strict-config --strict-markers --cov=semanticvideo --cov-report=term-missing --cov-fail-under=90" testpaths = ["tests"] +markers = [ + "integration: requires ffmpeg and ffprobe executables", +] [tool.ruff] line-length = 88 diff --git a/semanticvideo.schema.json b/semanticvideo.schema.json index cabbb66..eb74bb0 100644 --- a/semanticvideo.schema.json +++ b/semanticvideo.schema.json @@ -141,6 +141,19 @@ "additionalProperties": false, "description": "Technical metadata for one audio stream.", "properties": { + "bit_rate": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Bit Rate" + }, "channel_layout": { "anyOf": [ { @@ -826,6 +839,19 @@ "additionalProperties": false, "description": "Identity and technical facts for the source media asset.", "properties": { + "bit_rate": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Bit Rate" + }, "checksum": { "anyOf": [ { @@ -1914,6 +1940,19 @@ "additionalProperties": false, "description": "Technical metadata for one video stream.", "properties": { + "bit_rate": { + "anyOf": [ + { + "minimum": 0, + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Bit Rate" + }, "codec": { "minLength": 1, "title": "Codec", @@ -2002,14 +2041,10 @@ }, "rotation_degrees": { "default": 0, - "enum": [ - 0, - 90, - 180, - 270 - ], + "exclusiveMaximum": 360, + "minimum": 0, "title": "Rotation Degrees", - "type": "integer" + "type": "number" }, "sample_aspect_ratio": { "anyOf": [ diff --git a/src/semanticvideo/__init__.py b/src/semanticvideo/__init__.py index 730b0ff..3857b61 100644 --- a/src/semanticvideo/__init__.py +++ b/src/semanticvideo/__init__.py @@ -1,5 +1,6 @@ """Public API for the SemanticVideo reference implementation.""" +from semanticvideo.media import inspect_media, parse_ffprobe_json from semanticvideo.schema import ( AnalysisRun, Annotation, @@ -74,6 +75,8 @@ "SubtitleStream", "TimeRange", "VideoStream", + "inspect_media", + "parse_ffprobe_json", ] __version__ = "0.1.0" diff --git a/src/semanticvideo/cli/__init__.py b/src/semanticvideo/cli/__init__.py new file mode 100644 index 0000000..2ceec0e --- /dev/null +++ b/src/semanticvideo/cli/__init__.py @@ -0,0 +1 @@ +"""Command-line interface package.""" diff --git a/src/semanticvideo/cli/main.py b/src/semanticvideo/cli/main.py new file mode 100644 index 0000000..4678bd1 --- /dev/null +++ b/src/semanticvideo/cli/main.py @@ -0,0 +1,86 @@ +"""Scriptable SemanticVideo command-line interface.""" + +from __future__ import annotations + +import argparse +import sys +from collections.abc import Sequence +from pathlib import Path + +from semanticvideo import __version__ +from semanticvideo.errors import SemanticVideoError +from semanticvideo.media import inspect_media + + +def build_parser() -> argparse.ArgumentParser: + """Build the root CLI parser.""" + + parser = argparse.ArgumentParser( + prog="semanticvideo", + description="Inspect and describe video as reusable semantic metadata.", + ) + parser.add_argument("--version", action="version", version=__version__) + commands = parser.add_subparsers(dest="command", required=True) + + inspect_parser = commands.add_parser( + "inspect", help="Extract deterministic technical media metadata." + ) + inspect_parser.add_argument("input", type=Path, help="Input media file.") + inspect_parser.add_argument( + "--ffprobe", + default="ffprobe", + help="ffprobe executable name or path (default: ffprobe).", + ) + inspect_parser.add_argument( + "--timeout", + type=_positive_float, + default=60.0, + metavar="SECONDS", + help="Maximum ffprobe runtime (default: 60).", + ) + inspect_parser.add_argument( + "-o", + "--output", + type=Path, + help="Write JSON to this file instead of stdout.", + ) + inspect_parser.add_argument( + "--compact", + action="store_true", + help="Emit compact JSON instead of indented output.", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the CLI and convert expected failures into concise diagnostics.""" + + args = build_parser().parse_args(argv) + try: + if args.command == "inspect": + media = inspect_media( + args.input, + executable=args.ffprobe, + timeout_seconds=args.timeout, + ) + rendered = media.model_dump_json(indent=None if args.compact else 2) + if args.output is None: + sys.stdout.write(f"{rendered}\n") + else: + args.output.write_text(f"{rendered}\n", encoding="utf-8") + return 0 + except (SemanticVideoError, OSError) as error: + sys.stderr.write(f"error: {error}\n") + return 1 + return 2 + + +def _positive_float(value: str) -> float: + parsed = float(value) + if parsed <= 0: + raise argparse.ArgumentTypeError("must be greater than zero") + return parsed + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/semanticvideo/errors.py b/src/semanticvideo/errors.py new file mode 100644 index 0000000..9103086 --- /dev/null +++ b/src/semanticvideo/errors.py @@ -0,0 +1,31 @@ +"""Public exception hierarchy for SemanticVideo operations.""" + + +class SemanticVideoError(Exception): + """Base class for expected, user-actionable SemanticVideo failures.""" + + +class MediaInspectionError(SemanticVideoError): + """Base class for failures while inspecting a media asset.""" + + +class MediaNotFoundError(MediaInspectionError): + """The requested media path does not identify a readable file.""" + + +class FFprobeNotFoundError(MediaInspectionError): + """The configured ffprobe executable could not be started.""" + + +class FFprobeExecutionError(MediaInspectionError): + """ffprobe ran but returned an unsuccessful exit code.""" + + def __init__(self, returncode: int, stderr: str) -> None: + detail = stderr.strip() or "ffprobe returned no diagnostic output" + super().__init__(f"ffprobe failed with exit code {returncode}: {detail}") + self.returncode = returncode + self.stderr = stderr + + +class FFprobeParseError(MediaInspectionError): + """ffprobe output was invalid or lacked required media information.""" diff --git a/src/semanticvideo/media/__init__.py b/src/semanticvideo/media/__init__.py new file mode 100644 index 0000000..e631e1c --- /dev/null +++ b/src/semanticvideo/media/__init__.py @@ -0,0 +1,5 @@ +"""Deterministic media inspection utilities.""" + +from semanticvideo.media.ffprobe import inspect_media, parse_ffprobe_json, run_ffprobe + +__all__ = ["inspect_media", "parse_ffprobe_json", "run_ffprobe"] diff --git a/src/semanticvideo/media/ffprobe.py b/src/semanticvideo/media/ffprobe.py new file mode 100644 index 0000000..31a26c5 --- /dev/null +++ b/src/semanticvideo/media/ffprobe.py @@ -0,0 +1,371 @@ +"""Execute ffprobe safely and parse its JSON output into core schema models.""" + +from __future__ import annotations + +import json +import subprocess +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from decimal import Decimal, InvalidOperation +from fractions import Fraction +from pathlib import Path +from typing import Any +from uuid import NAMESPACE_URL, uuid5 + +from pydantic import ValidationError + +from semanticvideo.errors import ( + FFprobeExecutionError, + FFprobeNotFoundError, + FFprobeParseError, + MediaNotFoundError, +) +from semanticvideo.schema.media import ( + AudioStream, + MediaInfo, + Stream, + SubtitleStream, + VideoStream, +) +from semanticvideo.schema.time import RationalRate, RationalTime + +FFPROBE_ARGUMENTS = ( + "-v", + "error", + "-print_format", + "json", + "-show_format", + "-show_streams", +) + + +def run_ffprobe( + path: Path, + *, + executable: str = "ffprobe", + timeout_seconds: float = 60, +) -> dict[str, Any]: + """Run ffprobe without a shell and return its decoded JSON object.""" + + command = [executable, *FFPROBE_ARGUMENTS, str(path)] + try: + completed = subprocess.run( + command, + capture_output=True, + check=False, + encoding="utf-8", + errors="replace", + timeout=timeout_seconds, + ) + except FileNotFoundError as error: + raise FFprobeNotFoundError( + f"ffprobe executable was not found: {executable!r}" + ) from error + except subprocess.TimeoutExpired as error: + raise FFprobeExecutionError( + -1, f"ffprobe timed out after {timeout_seconds:g} seconds" + ) from error + + if completed.returncode != 0: + raise FFprobeExecutionError(completed.returncode, completed.stderr) + + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError as error: + raise FFprobeParseError(f"ffprobe returned invalid JSON: {error}") from error + if not isinstance(payload, dict): + raise FFprobeParseError("ffprobe JSON root must be an object") + return payload + + +def inspect_media( + path: str | Path, + *, + executable: str = "ffprobe", + timeout_seconds: float = 60, +) -> MediaInfo: + """Inspect one local media file and return validated technical metadata.""" + + media_path = Path(path) + if not media_path.is_file(): + raise MediaNotFoundError(f"media file does not exist: {media_path}") + + payload = run_ffprobe( + media_path, + executable=executable, + timeout_seconds=timeout_seconds, + ) + stat = media_path.stat() + return parse_ffprobe_json( + payload, + uri=str(media_path), + asset_id=_asset_id(media_path), + file_size=stat.st_size, + modified_at=datetime.fromtimestamp(stat.st_mtime, tz=UTC), + ) + + +def parse_ffprobe_json( + payload: Mapping[str, Any], + *, + uri: str, + asset_id: str | None = None, + file_size: int | None = None, + modified_at: datetime | None = None, +) -> MediaInfo: + """Purely parse an ffprobe payload without starting external processes.""" + + raw_streams = payload.get("streams") + if not isinstance(raw_streams, Sequence) or isinstance(raw_streams, (str, bytes)): + raise FFprobeParseError("ffprobe output must contain a streams array") + + streams: list[Stream] = [] + for raw_stream in raw_streams: + if not isinstance(raw_stream, Mapping): + raise FFprobeParseError("each ffprobe stream must be an object") + try: + parsed = _parse_stream(raw_stream) + except ValidationError as error: + raise FFprobeParseError( + f"ffprobe stream failed schema validation: {error}" + ) from error + if parsed is not None: + streams.append(parsed) + + raw_format = payload.get("format") + format_info = raw_format if isinstance(raw_format, Mapping) else {} + duration = _parse_duration(format_info, raw_streams) + tags = _string_mapping(format_info.get("tags")) + + effective_size = file_size + if effective_size is None: + effective_size = _optional_int(format_info.get("size")) + + try: + return MediaInfo( + id=asset_id or _asset_id_from_uri(uri), + uri=uri, + duration=duration, + file_size=effective_size, + modified_at=modified_at, + created_at=_find_creation_time(tags, raw_streams), + container_format=_optional_text(format_info.get("format_name")), + bit_rate=_optional_int(format_info.get("bit_rate")), + streams=tuple(streams), + metadata=tags, + ) + except ValidationError as error: + raise FFprobeParseError( + f"ffprobe metadata failed schema validation: {error}" + ) from error + + +def _parse_stream(raw: Mapping[str, Any]) -> Stream | None: + codec_type = _optional_text(raw.get("codec_type")) + index = _required_int(raw, "index") + stream_id = f"stream.{codec_type or 'unknown'}.{index}" + codec = ( + _optional_text(raw.get("codec_name")) + or _optional_text(raw.get("codec_long_name")) + or "unknown" + ) + tags = _string_mapping(raw.get("tags")) + + if codec_type == "video": + return VideoStream( + id=stream_id, + index=index, + codec=codec, + bit_rate=_optional_int(raw.get("bit_rate")), + width=_required_int(raw, "width", positive=True), + height=_required_int(raw, "height", positive=True), + pixel_format=_optional_text(raw.get("pix_fmt")), + frame_rate=_parse_ratio(raw.get("avg_frame_rate")), + time_base=_parse_ratio(raw.get("time_base")), + rotation_degrees=_parse_rotation(raw, tags), + sample_aspect_ratio=_parse_ratio(raw.get("sample_aspect_ratio")), + color_primaries=_optional_text(raw.get("color_primaries")), + color_transfer=_optional_text(raw.get("color_transfer")), + color_space=_optional_text(raw.get("color_space")), + variable_frame_rate=_detect_variable_frame_rate(raw), + ) + if codec_type == "audio": + return AudioStream( + id=stream_id, + index=index, + codec=codec, + bit_rate=_optional_int(raw.get("bit_rate")), + sample_rate=_required_int(raw, "sample_rate", positive=True), + channels=_required_int(raw, "channels", positive=True), + channel_layout=_optional_text(raw.get("channel_layout")), + language=tags.get("language"), + time_base=_parse_ratio(raw.get("time_base")), + ) + if codec_type == "subtitle": + return SubtitleStream( + id=stream_id, + index=index, + codec=codec, + language=tags.get("language"), + ) + return None + + +def _parse_duration( + format_info: Mapping[str, Any], raw_streams: Sequence[Any] +) -> RationalTime: + format_duration = _decimal_time(format_info.get("duration")) + if format_duration is not None and format_duration.value > 0: + return format_duration + + candidates: list[RationalTime] = [] + for raw in raw_streams: + if not isinstance(raw, Mapping): + continue + direct = _decimal_time(raw.get("duration")) + if direct is not None and direct.value > 0: + candidates.append(direct) + continue + duration_ts = _optional_int(raw.get("duration_ts")) + time_base = _parse_ratio(raw.get("time_base")) + if duration_ts is not None and duration_ts > 0 and time_base is not None: + fraction = Fraction( + duration_ts * time_base.numerator, time_base.denominator + ) + candidates.append(_fraction_time(fraction)) + + if not candidates: + raise FFprobeParseError("ffprobe output does not contain a positive duration") + return max(candidates, key=lambda item: item.fraction) + + +def _parse_ratio(value: Any) -> RationalRate | None: + text = _optional_text(value) + if text is None or text in {"0/0", "N/A"}: + return None + separator = "/" if "/" in text else ":" if ":" in text else None + if separator is None: + try: + fraction = Fraction(text) + except (ValueError, ZeroDivisionError) as error: + raise FFprobeParseError(f"invalid rational value: {text!r}") from error + else: + left, right = text.split(separator, maxsplit=1) + try: + fraction = Fraction(int(left), int(right)) + except (ValueError, ZeroDivisionError) as error: + raise FFprobeParseError(f"invalid rational value: {text!r}") from error + if fraction <= 0: + return None + return RationalRate(numerator=fraction.numerator, denominator=fraction.denominator) + + +def _decimal_time(value: Any) -> RationalTime | None: + text = _optional_text(value) + if text is None or text == "N/A": + return None + try: + decimal = Decimal(text) + except InvalidOperation as error: + raise FFprobeParseError(f"invalid duration value: {text!r}") from error + if not decimal.is_finite() or decimal < 0: + raise FFprobeParseError(f"invalid duration value: {text!r}") + return _fraction_time(Fraction(decimal)) + + +def _fraction_time(value: Fraction) -> RationalTime: + return RationalTime(value=value.numerator, rate=value.denominator) + + +def _required_int(raw: Mapping[str, Any], key: str, *, positive: bool = False) -> int: + value = _optional_int(raw.get(key)) + if value is None or (positive and value <= 0): + qualifier = "positive " if positive else "" + raise FFprobeParseError(f"stream field {key!r} must be a {qualifier}integer") + return value + + +def _optional_int(value: Any) -> int | None: + if value is None or isinstance(value, bool): + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _optional_text(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + +def _string_mapping(value: Any) -> dict[str, str]: + if not isinstance(value, Mapping): + return {} + return {str(key): str(item) for key, item in value.items() if item is not None} + + +def _parse_datetime(value: str | None) -> datetime | None: + if value is None: + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=UTC) + return parsed + + +def _find_creation_time( + format_tags: Mapping[str, str], raw_streams: Sequence[Any] +) -> datetime | None: + candidates = [format_tags.get("creation_time")] + candidates.extend( + _string_mapping(raw.get("tags")).get("creation_time") + for raw in raw_streams + if isinstance(raw, Mapping) + ) + for candidate in candidates: + parsed = _parse_datetime(candidate) + if parsed is not None: + return parsed + return None + + +def _parse_rotation(raw: Mapping[str, Any], tags: Mapping[str, str]) -> float: + values: list[Any] = [] + side_data = raw.get("side_data_list") + if isinstance(side_data, Sequence) and not isinstance(side_data, (str, bytes)): + for item in side_data: + if isinstance(item, Mapping) and "rotation" in item: + values.append(item["rotation"]) + if "rotate" in tags: + values.append(tags["rotate"]) + for value in values: + try: + rotation = float(value) % 360 + except (TypeError, ValueError): + continue + return 0.0 if rotation == 360 else rotation + return 0.0 + + +def _detect_variable_frame_rate(raw: Mapping[str, Any]) -> bool | None: + average = _parse_ratio(raw.get("avg_frame_rate")) + nominal = _parse_ratio(raw.get("r_frame_rate")) + if average is None or nominal is None: + return None + return average.fraction != nominal.fraction + + +def _asset_id(path: Path) -> str: + return _asset_id_from_uri(path.resolve().as_uri()) + + +def _asset_id_from_uri(uri: str) -> str: + # UUIDv5 makes the ID deterministic without hashing the potentially huge media file. + digest = uuid5(NAMESPACE_URL, uri).hex + return f"asset.{digest}" diff --git a/src/semanticvideo/schema/media.py b/src/semanticvideo/schema/media.py index 17dfa53..85ce9ad 100644 --- a/src/semanticvideo/schema/media.py +++ b/src/semanticvideo/schema/media.py @@ -27,12 +27,13 @@ class VideoStream(SemanticModel): id: Identifier index: int = Field(ge=0) codec: str = Field(min_length=1) + bit_rate: int | None = Field(default=None, ge=0) width: int = Field(gt=0) height: int = Field(gt=0) pixel_format: str | None = None frame_rate: RationalRate | None = None time_base: RationalRate | None = None - rotation_degrees: Literal[0, 90, 180, 270] = 0 + rotation_degrees: float = Field(default=0, ge=0, lt=360) sample_aspect_ratio: RationalRate | None = None color_primaries: str | None = None color_transfer: str | None = None @@ -47,6 +48,7 @@ class AudioStream(SemanticModel): id: Identifier index: int = Field(ge=0) codec: str = Field(min_length=1) + bit_rate: int | None = Field(default=None, ge=0) sample_rate: int = Field(gt=0) channels: int = Field(gt=0) channel_layout: str | None = None @@ -81,6 +83,7 @@ class MediaInfo(SemanticModel): created_at: datetime | None = None checksum: Checksum | None = None container_format: str | None = None + bit_rate: int | None = Field(default=None, ge=0) streams: tuple[Stream, ...] metadata: dict[str, str] = Field(default_factory=dict) diff --git a/tests/fixtures/ffprobe/standard_mp4.json b/tests/fixtures/ffprobe/standard_mp4.json new file mode 100644 index 0000000..8aae2d2 --- /dev/null +++ b/tests/fixtures/ffprobe/standard_mp4.json @@ -0,0 +1,56 @@ +{ + "streams": [ + { + "index": 0, + "codec_name": "h264", + "codec_long_name": "H.264 / AVC", + "codec_type": "video", + "bit_rate": "11800000", + "width": 3840, + "height": 2160, + "pix_fmt": "yuv420p", + "r_frame_rate": "30000/1001", + "avg_frame_rate": "30000/1001", + "time_base": "1/30000", + "sample_aspect_ratio": "1:1", + "color_space": "bt709", + "color_transfer": "bt709", + "color_primaries": "bt709", + "side_data_list": [ + { + "side_data_type": "Display Matrix", + "rotation": -90 + } + ] + }, + { + "index": 1, + "codec_name": "aac", + "codec_type": "audio", + "bit_rate": "128000", + "sample_rate": "48000", + "channels": 2, + "channel_layout": "stereo", + "time_base": "1/48000", + "tags": { + "language": "jpn" + } + }, + { + "index": 2, + "codec_name": "bin_data", + "codec_type": "data" + } + ], + "format": { + "filename": "GX010231.MP4", + "format_name": "mov,mp4,m4a,3gp,3g2,mj2", + "duration": "12.345000", + "size": "183728192", + "bit_rate": "11906168", + "tags": { + "major_brand": "isom", + "creation_time": "2026-05-18T12:35:10.000000Z" + } + } +} diff --git a/tests/fixtures/ffprobe/video_only_duration_ts.json b/tests/fixtures/ffprobe/video_only_duration_ts.json new file mode 100644 index 0000000..5462947 --- /dev/null +++ b/tests/fixtures/ffprobe/video_only_duration_ts.json @@ -0,0 +1,28 @@ +{ + "streams": [ + { + "index": 0, + "codec_name": "hevc", + "codec_type": "video", + "width": 1920, + "height": 1080, + "r_frame_rate": "25/1", + "avg_frame_rate": "24/1", + "time_base": "1/30000", + "duration_ts": 90000, + "sample_aspect_ratio": "N/A", + "tags": { + "rotate": "90", + "creation_time": "2026-05-18T12:35:10Z" + } + } + ], + "format": { + "format_name": "matroska,webm", + "size": "2048", + "bit_rate": "5461", + "tags": { + "creation_time": "not-a-date" + } + } +} diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..c0d1e5d --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,85 @@ +"""Command-line interface tests.""" + +import json +from pathlib import Path + +import pytest + +from semanticvideo.cli.main import build_parser, main +from semanticvideo.errors import MediaNotFoundError +from semanticvideo.schema import MediaInfo, RationalTime, VideoStream + + +def media_info() -> MediaInfo: + return MediaInfo( + id="asset.test", + uri="clip.mp4", + duration=RationalTime(value=3, rate=1), + streams=( + VideoStream( + id="stream.video.0", + index=0, + codec="h264", + width=1920, + height=1080, + ), + ), + ) + + +def test_inspect_prints_scriptable_json( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr( + "semanticvideo.cli.main.inspect_media", lambda *_args, **_kwargs: media_info() + ) + + assert main(["inspect", "clip.mp4", "--compact"]) == 0 + captured = capsys.readouterr() + assert json.loads(captured.out)["id"] == "asset.test" + assert captured.err == "" + + +def test_inspect_writes_output_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + output = tmp_path / "inspection.json" + monkeypatch.setattr( + "semanticvideo.cli.main.inspect_media", lambda *_args, **_kwargs: media_info() + ) + + assert main(["inspect", "clip.mp4", "--output", str(output)]) == 0 + assert json.loads(output.read_text(encoding="utf-8"))["duration"] == { + "value": 3, + "rate": 1, + } + + +def test_inspect_reports_expected_error( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + def fail(*_args: object, **_kwargs: object) -> None: + raise MediaNotFoundError("missing clip") + + monkeypatch.setattr("semanticvideo.cli.main.inspect_media", fail) + + assert main(["inspect", "missing.mp4"]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err == "error: missing clip\n" + + +def test_output_os_error_is_reported( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr( + "semanticvideo.cli.main.inspect_media", lambda *_args, **_kwargs: media_info() + ) + + assert main(["inspect", "clip.mp4", "-o", str(tmp_path)]) == 1 + assert "error:" in capsys.readouterr().err + + +def test_timeout_must_be_positive() -> None: + with pytest.raises(SystemExit): + build_parser().parse_args(["inspect", "clip.mp4", "--timeout", "0"]) diff --git a/tests/test_ffprobe.py b/tests/test_ffprobe.py new file mode 100644 index 0000000..c658ae9 --- /dev/null +++ b/tests/test_ffprobe.py @@ -0,0 +1,276 @@ +"""Unit tests for ffprobe execution and pure parsing.""" + +from __future__ import annotations + +import json +import subprocess +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from semanticvideo.errors import ( + FFprobeExecutionError, + FFprobeNotFoundError, + FFprobeParseError, + MediaNotFoundError, +) +from semanticvideo.media.ffprobe import inspect_media, parse_ffprobe_json, run_ffprobe +from semanticvideo.schema import VideoStream + +FIXTURES = Path(__file__).parent / "fixtures" / "ffprobe" + + +def load_fixture(name: str) -> dict[str, Any]: + parsed = json.loads((FIXTURES / name).read_text(encoding="utf-8")) + assert isinstance(parsed, dict) + return cast(dict[str, Any], parsed) + + +def test_parse_standard_mp4() -> None: + media = parse_ffprobe_json( + load_fixture("standard_mp4.json"), + uri="GX010231.MP4", + asset_id="asset.GX010231", + modified_at=datetime(2026, 8, 9, tzinfo=UTC), + ) + + assert media.id == "asset.GX010231" + assert media.duration.seconds == pytest.approx(12.345) + assert media.file_size == 183_728_192 + assert media.bit_rate == 11_906_168 + assert media.created_at == datetime(2026, 5, 18, 12, 35, 10, tzinfo=UTC) + assert media.container_format == "mov,mp4,m4a,3gp,3g2,mj2" + assert media.metadata["major_brand"] == "isom" + assert len(media.streams) == 2 + + video = media.streams[0] + assert video.kind == "video" + assert video.frame_rate is not None + assert (video.frame_rate.numerator, video.frame_rate.denominator) == (30_000, 1001) + assert video.rotation_degrees == 270 + assert video.bit_rate == 11_800_000 + assert video.variable_frame_rate is False + + audio = media.streams[1] + assert audio.kind == "audio" + assert audio.sample_rate == 48_000 + assert audio.bit_rate == 128_000 + assert audio.language == "jpn" + + +def test_parse_video_only_with_duration_ts_fallback() -> None: + media = parse_ffprobe_json( + load_fixture("video_only_duration_ts.json"), uri="clip.mkv" + ) + + assert media.duration.seconds == 3 + assert media.file_size == 2048 + assert media.created_at == datetime(2026, 5, 18, 12, 35, 10, tzinfo=UTC) + assert media.bit_rate == 5461 + video = media.streams[0] + assert isinstance(video, VideoStream) + assert video.rotation_degrees == 90 + assert video.variable_frame_rate is True + + +def test_parse_uses_long_codec_name_and_direct_stream_duration() -> None: + payload = { + "streams": [ + { + "index": "0", + "codec_long_name": "Example codec", + "codec_type": "video", + "width": "640", + "height": "360", + "duration": "1.25", + "avg_frame_rate": "0/0", + "r_frame_rate": "0/0", + "time_base": "1/1000", + } + ] + } + + media = parse_ffprobe_json(payload, uri="clip.example") + + assert media.duration.seconds == 1.25 + video = media.streams[0] + assert isinstance(video, VideoStream) + assert video.codec == "Example codec" + assert video.frame_rate is None + assert video.variable_frame_rate is None + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ({}, "streams array"), + ({"streams": ["invalid"]}, "stream must be an object"), + ( + {"streams": [{"index": 0, "codec_type": "video", "width": 1, "height": 1}]}, + "positive duration", + ), + ( + { + "streams": [ + { + "index": 0, + "codec_type": "audio", + "sample_rate": 48000, + "channels": 2, + "duration": 1, + } + ] + }, + "at least one video stream", + ), + ( + { + "streams": [ + { + "index": 0, + "codec_type": "video", + "width": 0, + "height": 1, + "duration": 1, + } + ] + }, + "positive integer", + ), + ( + { + "streams": [ + { + "index": 0, + "codec_type": "video", + "width": 1, + "height": 1, + "duration": "invalid", + } + ] + }, + "invalid duration", + ), + ( + { + "format": {"duration": 1}, + "streams": [ + { + "index": 0, + "codec_type": "video", + "width": 1, + "height": 1, + "avg_frame_rate": "broken", + } + ], + }, + "invalid rational", + ), + ( + { + "format": {"duration": 1}, + "streams": [ + { + "index": -1, + "codec_type": "video", + "width": 1, + "height": 1, + } + ], + }, + "stream failed schema validation", + ), + ], +) +def test_invalid_ffprobe_payloads_are_rejected( + payload: dict[str, Any], message: str +) -> None: + with pytest.raises(FFprobeParseError, match=message): + parse_ffprobe_json(payload, uri="invalid.mp4") + + +def test_run_ffprobe_decodes_json(monkeypatch: pytest.MonkeyPatch) -> None: + observed: list[str] = [] + + def fake_run(command: list[str], **_: Any) -> SimpleNamespace: + observed.extend(command) + return SimpleNamespace(returncode=0, stdout='{"streams": []}', stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert run_ffprobe(Path("clip.mp4"), executable="custom-ffprobe") == {"streams": []} + assert observed[0] == "custom-ffprobe" + assert observed[-1] == "clip.mp4" + + +def test_run_ffprobe_maps_expected_failures(monkeypatch: pytest.MonkeyPatch) -> None: + def missing(*_: Any, **__: Any) -> None: + raise FileNotFoundError + + monkeypatch.setattr(subprocess, "run", missing) + with pytest.raises(FFprobeNotFoundError, match="was not found"): + run_ffprobe(Path("clip.mp4")) + + def timeout(*_: Any, **__: Any) -> None: + raise subprocess.TimeoutExpired("ffprobe", 2) + + monkeypatch.setattr(subprocess, "run", timeout) + with pytest.raises(FFprobeExecutionError, match="timed out"): + run_ffprobe(Path("clip.mp4"), timeout_seconds=2) + + +@pytest.mark.parametrize( + ("result", "error_type", "message"), + [ + ( + SimpleNamespace(returncode=7, stdout="", stderr="bad input"), + FFprobeExecutionError, + "bad input", + ), + ( + SimpleNamespace(returncode=0, stdout="not-json", stderr=""), + FFprobeParseError, + "invalid JSON", + ), + ( + SimpleNamespace(returncode=0, stdout="[]", stderr=""), + FFprobeParseError, + "root must be an object", + ), + ], +) +def test_run_ffprobe_rejects_bad_results( + monkeypatch: pytest.MonkeyPatch, + result: SimpleNamespace, + error_type: type[Exception], + message: str, +) -> None: + monkeypatch.setattr(subprocess, "run", lambda *_args, **_kwargs: result) + with pytest.raises(error_type, match=message): + run_ffprobe(Path("clip.mp4")) + + +def test_inspect_media_requires_file(tmp_path: Path) -> None: + with pytest.raises(MediaNotFoundError, match="does not exist"): + inspect_media(tmp_path / "missing.mp4") + + +def test_inspect_media_adds_filesystem_identity( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + media_path = tmp_path / "clip.mp4" + media_path.write_bytes(b"fixture") + payload = load_fixture("standard_mp4.json") + monkeypatch.setattr( + "semanticvideo.media.ffprobe.run_ffprobe", lambda *_args, **_kwargs: payload + ) + + media = inspect_media(media_path) + + assert media.file_size == 7 + assert media.modified_at is not None + assert media.id.startswith("asset.") + assert inspect_media(media_path).id == media.id diff --git a/tests/test_media_integration.py b/tests/test_media_integration.py new file mode 100644 index 0000000..cc602df --- /dev/null +++ b/tests/test_media_integration.py @@ -0,0 +1,64 @@ +"""Small end-to-end media inspection test using local FFmpeg tools.""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +from semanticvideo.cli.main import main +from semanticvideo.media import inspect_media + + +@pytest.mark.integration +def test_inspect_synthetic_video( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + ffmpeg = shutil.which("ffmpeg") + ffprobe = shutil.which("ffprobe") + if ffmpeg is None or ffprobe is None: + pytest.skip("FFmpeg tools are not installed") + + output = tmp_path / "synthetic.mp4" + subprocess.run( + [ + ffmpeg, + "-v", + "error", + "-y", + "-f", + "lavfi", + "-i", + "testsrc2=size=160x90:rate=25", + "-f", + "lavfi", + "-i", + "sine=frequency=1000:sample_rate=48000", + "-t", + "0.4", + "-c:v", + "mpeg4", + "-c:a", + "aac", + "-shortest", + str(output), + ], + check=True, + capture_output=True, + ) + + media = inspect_media(output, executable=ffprobe) + + assert media.duration.seconds == pytest.approx(0.4, abs=0.05) + assert [stream.kind for stream in media.streams] == ["video", "audio"] + assert media.streams[0].kind == "video" + assert (media.streams[0].width, media.streams[0].height) == (160, 90) + + assert main(["inspect", str(output), "--ffprobe", ffprobe, "--compact"]) == 0 + cli_result = json.loads(capsys.readouterr().out) + assert cli_result["duration"] == {"value": 2, "rate": 5} + assert [stream["kind"] for stream in cli_result["streams"]] == [ + "video", + "audio", + ]