From 8aa33a01ba9ada6bb7f3b72c7c2db26da18ed452 Mon Sep 17 00:00:00 2001 From: Armando Anaya Date: Mon, 27 Jul 2026 03:19:27 -0700 Subject: [PATCH] =?UTF-8?q?feat(kernel):=20VideoProcessor=20=E2=80=94=20ff?= =?UTF-8?q?mpeg=20probing,=20oriented=20dimensions,=20deterministic=20fram?= =?UTF-8?q?e=20extraction=20(#17)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Declares `ports/video_processor.py` beside #16's `image_processor.py` — one file per port, each declared by the task that implements it — and ships `FfmpegVideoProcessor` behind it. `probe` reports as-displayed dimensions, source rate, duration and codec; `frames` streams PNG frames carrying `(index, timestamp)` off a running ffmpeg, one at a time. The two media errors are reused rather than extended: a container ffmpeg never opens is `UnsupportedMedia`, a clip that decodes for a while and then runs out is `CorruptMedia`. A missing binary is `MediaToolUnavailable`, deliberately outside that family — no file is at fault, so an ingest must not record it against five thousand innocent ones. `workspace.video_processor` is the fifth port, composed the EventBus way. No migration: FORMAT_VERSION stays 6, VERSION stays 0.0.1.dev0, no Asset field. --- docs/README.md | 2 +- docs/media.md | 183 +++++- docs/workspaces.md | 24 +- src/visionset/kernel/__init__.py | 2 + src/visionset/kernel/adapters/__init__.py | 2 + .../kernel/adapters/ffmpeg_video_processor.py | 521 ++++++++++++++++ src/visionset/kernel/domain/__init__.py | 9 +- src/visionset/kernel/domain/media.py | 84 ++- src/visionset/kernel/errors.py | 19 + src/visionset/kernel/ports/__init__.py | 8 + src/visionset/kernel/ports/video_processor.py | 106 ++++ .../kernel/services/workspace_service.py | 67 +- tests/fixtures/media.py | 125 +++- tests/fixtures/test_media.py | 33 +- tests/kernel/test_video_processor.py | 571 ++++++++++++++++++ 15 files changed, 1688 insertions(+), 68 deletions(-) create mode 100644 src/visionset/kernel/adapters/ffmpeg_video_processor.py create mode 100644 src/visionset/kernel/ports/video_processor.py create mode 100644 tests/kernel/test_video_processor.py diff --git a/docs/README.md b/docs/README.md index 76dd1f09..82364ccb 100644 --- a/docs/README.md +++ b/docs/README.md @@ -12,7 +12,7 @@ contracts (kernel purity, headless annotator) are described there and enforced i | [batches.md](batches.md) | The unit of annotation work: the state machine, membership frozen at approval, the schema pin, and the exact partition into jobs | | [jobs.md](jobs.md) | Annotation jobs: the job and per-asset progress machines, what counts as settled, ordered `next_pending`, and derived progress | | [annotations.md](annotations.md) | The labels themselves: the one door, the batch's pinned version, the five hard rejects, attribute values, and progress derived from the annotations | -| [media.md](media.md) | Decoding raw media: the two processor ports, the accepted formats, the orientation policy, pinned thumbnails and what their determinism does and does not promise | +| [media.md](media.md) | Decoding raw media: the two processor ports, the accepted image formats, the orientation policy for stills and clips, pinned thumbnails and seek-free frame extraction, and what their determinism does and does not promise | | [datasets.md](datasets.md) | The curated trunk: promotion from a completed batch, what `skipped` keeps out, curation without a `confirm=`, and the append-only change log | | [releases.md](releases.md) | The immutable artifact: what a manifest is and is not, why two publishes agree byte for byte, hash verification, and the seeded split recipe | | [events.md](events.md) | Domain events: subscribing by type, why emission follows the commit, at-most-once delivery, and what an isolated subscriber failure does | diff --git a/docs/media.md b/docs/media.md index 264d3318..c498a253 100644 --- a/docs/media.md +++ b/docs/media.md @@ -5,12 +5,19 @@ decoded. That is what the media ports are for: hand them bytes, and they either the picture is or refuse it with a sentence naming the file. ```python +from pathlib import Path + from visionset.kernel.services import WorkspaceService with WorkspaceService.open("./road-signs") as workspace: with open("photos/IMG_0043.jpg", "rb") as handle: metadata = workspace.image_processor.probe(handle) # ImageMetadata(width=..., ...) preview = workspace.image_processor.thumbnail(handle) # opaque JPEG bytes + + clip = Path("footage/dashcam.mp4") + info = workspace.video_processor.probe(clip) # VideoMetadata(fps=29.97, ...) + for frame in workspace.video_processor.frames(clip, fps=1): + ... # frame.index, frame.timestamp, frame.content — a PNG ``` ## Two protocols, not one @@ -21,8 +28,22 @@ has no thumbnail to serve and a Pillow adapter has no frames to iterate. A share force each implementation to declare the other's methods and raise from them — a runtime failure where a compile-time absence was available. -Each is declared by the task that implements it. `ImageProcessor` and the Pillow adapter behind -it exist today; `VideoProcessor` arrives with the ffmpeg one. +Each is declared by the task that implements it, in its own file, and both exist today: +`ImageProcessor` behind `PillowImageProcessor`, `VideoProcessor` behind `FfmpegVideoProcessor`. + +They also take different inputs, and that asymmetry is deliberate rather than an inconsistency +waiting to be tidied. `ImageProcessor` takes a **stream**; `VideoProcessor` takes a **path**. A +video decoder is an out-of-process program that seeks: handed a pipe it cannot say how long a +clip is without decoding all of it, and cannot revisit a byte it has already read. Nothing is +lost by requiring a file, because no caller has video bytes without one — a source is a path on +disk, and a blob in the default blob store is a path too. + +| | `ImageProcessor` | `VideoProcessor` | +| --- | --- | --- | +| Input | an open binary stream | a `Path` | +| Reads | `probe` → `ImageMetadata` | `probe` → `VideoMetadata` | +| Produces | `thumbnail` → JPEG bytes | `frames` → an iterator of `VideoFrame` | +| Needs | Pillow, a dependency | ffmpeg, a binary on `PATH` | ## The accepted list is two formats, and widening it is a decision @@ -47,6 +68,19 @@ unsupported. **The bytes decide the format, not the filename.** A `.png` holding JPEG bytes reports `jpeg`. A suffix is a hint for choosing which files to look at; what a file *is* comes from decoding it. +### There is no `VideoFormat`, on purpose + +The video side has no curated list at all: it accepts whatever ffmpeg's demuxer opens, and +`VideoMetadata.codec` is a plain `str` recording what that turned out to be. + +The asymmetry is the argument. **An image is an asset; a video is a source.** Curating +`ImageFormat` buys something real, because those exact bytes enter the dataset and the promise +above is made about them. A video's bytes never do — they leave the decoder as PNG frames — so a +closed list of codecs would gate nothing while going stale every time a camera vendor ships a new +profile. `codec` therefore *records* what was read instead of *deciding* what may be read, which +is the same split as `DatasetChange.operation` being a `str` while `DatasetOperation` is the enum +a writer picks from. + ## Orientation is applied, not reported `ImageMetadata.width` and `.height` are **as displayed**. A 32×24 JPEG carrying EXIF orientation @@ -66,6 +100,72 @@ stored dimensions, and that is the ordinary path rather than a fallback. The policy is **format-independent**. PNG carries EXIF in an `eXIf` chunk and is oriented on exactly the same terms as JPEG, which is the case a hand-rolled tag-274 reader gets wrong. +It is **modality-independent** too. A clip carries its turn in a display matrix rather than in an +EXIF tag — a phone held upright writes a landscape stream plus a quarter turn — and ffmpeg +applies that matrix when it decodes. So `VideoMetadata` reports the swapped edges, and the frames +`frames()` yields come out at exactly those dimensions. Reporting the stored numbers would +describe a picture nobody will ever see, and would put every extracted frame at odds with the +metadata stamped beside it. ffprobe spells the same quarter turn as `90`, as `-90` and as `270` +depending on the file and the build, so the rule is arithmetic on the angle (`rotation % 180`) +and never a membership test against three literals. + +## Frames: the extraction is pinned, and it does not seek + +`frames()` streams one frame at a time off a running ffmpeg — nothing buffers a clip — and every +argument that decides what comes out is fixed in one tuple in the adapter. Four of them are worth +knowing about: + +- **`fps=N:round=up`** is the extraction grid, and it is what makes `frame.timestamp` honest. The + filter maps each input frame onto an output slot and the last one to land there wins; under the + default `near` rounding the winner is the frame nearest the slot's *midpoint*, so at 1 fps a + clip shot at 10 yields the pictures from 0.4 s and 1.4 s while labelling them 0.0 s and 1.0 s. + `up` makes the frame at the grid point itself the winner, which puts the reported timestamp + within one *source* frame of the pixels — the best any resampler can do. Frame counts are + identical either way. +- **There is no seek.** Input seeking lands on a keyframe and is approximate; output seeking + interacts with the filter. Extraction reads the clip from the start every time, which is + exactly what makes it reproducible. +- **`-xerror`** is the least obvious and the most load-bearing. Without it ffmpeg reports a + truncated clip on stderr, exits **zero**, and hands back the frames it managed to decode — so a + damaged file would ingest as a merely short one and nothing would ever say so. Its opposite + number, `-err_detect explode`, is deliberately not used: it rejects perfectly good files. +- **`-pred none -compression_level 6` and `+bitexact`** pin the PNG encoder, whose bytes become + the content hash of the asset each frame turns into. + +Two consequences worth stating outright: + +- **A frame's `index` counts the extracted sequence, not the source.** A source frame number + means nothing for a variable-rate stream and cannot be reproduced without knowing the rate the + file was shot at. `timestamp` is the locator that survives — it says where in the clip to look, + whatever rate the next decomposition runs at. +- **Asking for a higher rate than the clip has duplicates frames.** This is documented rather + than clamped: clamping would mean probing inside `frames()`, and an ingest content-addresses + what it stores, so the duplicates collapse into one asset anyway. The honest fix is not to ask. + +### The iterator owns a running program + +`frames()` is not itself a generator — it validates its arguments, checks for ffmpeg and returns +an inner one — so a missing binary or a negative `fps` is reported at the call rather than at the +first iteration inside whatever loop happened to consume it. + +What comes back holds a live decoder until it is exhausted or closed. A `for` loop that runs to +the end is fine, and so is one that `break`s (closing the generator terminates ffmpeg); a caller +that stashes the iterator somewhere should close it, `contextlib.closing`-style. + +### What video determinism costs, and who pays + +The same caveat as thumbnails, with a bigger consequence. Identical clip, identical `fps`, +identical frames — across runs and across processes, **on one machine with one installed +ffmpeg**. It does *not* hold across ffmpeg builds. + +Frames are content-addressed, so this propagates: **video-derived asset identity is reproducible +within an ffmpeg build, not across one.** Re-ingesting the same clip after an ffmpeg upgrade +yields different hashes and therefore new assets beside the old ones. Images do not have this +property — a JPEG's bytes are its bytes — and video does, because the asset is something we +computed rather than something we were given. The practical rule is the familiar one: assert +repeatability, never a hardcoded hash, and do not build a cross-machine identity claim on a +frame. + ## Thumbnails: one encoding, pinned Every `thumbnail()` returns a **JPEG**, at most 256 pixels on its longest edge unless the caller @@ -115,8 +215,8 @@ failure against the item it was reading, and carry on with the next file. | Error | Means | Remedy | | --- | --- | --- | -| `UnsupportedMedia` | not an image at all, an image in a format outside `ImageFormat`, or one declaring more pixels than the decoder will take | filter the input, convert the file, or ask for the format to be accepted | -| `CorruptMedia` | an accepted format whose bytes will not decode | re-fetch or re-export the file | +| `UnsupportedMedia` | not an image at all, an image in a format outside `ImageFormat`, one declaring more pixels than the decoder will take, or a file ffmpeg never opened | filter the input, convert the file, or ask for the format to be accepted | +| `CorruptMedia` | an accepted format whose bytes will not decode, or a clip that decoded for a while and then ran out | re-fetch or re-export the file | The split is by **remedy**, which is the only thing an error hierarchy should branch on. "Your input folder contains a README" and "one of your JPEGs is truncated" have opposite answers, and @@ -129,7 +229,31 @@ fails exactly when the decode fails. And splitting `UnsupportedMedia` into "not and an exotic one both come back as *I do not know what this is*. The family is named for **media**, not images, because the video processor raises the same two -for the same reasons. +for the same reasons — the split is by remedy, and remedies have no modality. For video the two +are separated by *when* the decoder gave up, and the order of the calls is what keeps that +honest. A container ffmpeg cannot open at all is `UnsupportedMedia`, and that includes a clip +whose index went missing with its tail; a clip that opens, yields frames and then runs out is +`CorruptMedia`, and it hands back the frames that did decode before it raises. + +### `MediaToolUnavailable` is not in the family + +The default video adapter needs ffmpeg on the machine, and its absence is a `MediaToolUnavailable` +— deliberately **not** a `MediaError`, so `except MediaError` will not catch it. + +Every error in that family answers *what is wrong with this file?*; this one answers *what is +wrong with this machine?*. An ingest catches the media family per item and carries on, so if this +were in it a missing decoder would be recorded five thousand times against five thousand innocent +files, and the run would report a data problem it does not have. It is the single fatal cause an +ingest job records beside its per-file report, and its message carries an install hint, because +the remedy is a package manager. + +It is also why nothing checks for ffmpeg at import or when a workspace opens: a machine with no +ffmpeg still opens workspaces and still ingests images perfectly well, so the check belongs to +the call that actually needs to decode a video. + +Two more refusals sit outside the family on purpose, for the same reason `max_edge < 1` does: a +missing file is a `FileNotFoundError` and a non-positive `fps` is a `ValueError`. Neither is a +property of any media. ### `name` and `reason` @@ -177,21 +301,42 @@ second call sees nothing. A test says so, rather than leaving it to be discovere ## Composition -The image processor is the fourth port on `WorkspaceService`, reached as -`workspace.image_processor` and built by a zero-argument `image_processor_factory` on both -`init` and `open` — the shape the [event bus](events.md) uses, because neither is derived from -the workspace path. One per open workspace, never a module-level singleton, and nothing to -close: the decoder holds no state at all. +The two media processors are the fourth and fifth ports on `WorkspaceService`, reached as +`workspace.image_processor` and `workspace.video_processor` and built by zero-argument +`image_processor_factory` / `video_processor_factory` on both `init` and `open` — the shape the +[event bus](events.md) uses, because none of them is derived from the workspace path. One of each +per open workspace, never a module-level singleton, and nothing to close: the decoders hold no +state at all. A live frame iterator *does* own a running program, but that belongs to whoever +asked for it, not to the workspace. + +A new port is appended **last** to `WorkspaceService.__init__`, never inserted: both classmethods +bind those arguments positionally, so a parameter added in the middle silently re-binds every one +after it. + +No service below the composition point ever names `PillowImageProcessor` or +`FfmpegVideoProcessor`. That is the rule [workspaces.md](workspaces.md) describes, and it is what +makes swapping a decoder a change to two functions and to nowhere else. + +## ffmpeg is a binary; Pillow is a dependency + +Pillow is in `[project].dependencies` and arrives with the wheel. ffmpeg cannot: it is a program, +not a package, so `pip install visionset` does not put one on the machine and a user may +legitimately never need it. Hence the lazy check, the install hint, and the fact that video tests +**skip** locally when it is missing — with one guard, because a silently skipped video test looks +exactly like a passing one. CI installs ffmpeg and sets `VISIONSET_REQUIRE_FFMPEG=1`, which turns +that skip into a hard error, so a broken install step goes red rather than quietly shrinking the +suite. -No service below the composition point ever names `PillowImageProcessor`. That is the rule -[workspaces.md](workspaces.md) describes, and it is what makes swapping a decoder a change to -two functions and to nowhere else. +Neither library needs an import-linter change. The contracts forbid *frameworks* inside the +kernel — FastAPI, Typer, MCP, uvicorn — not third-party libraries, and ffmpeg is reached through +`subprocess` and is not an import at all. ## What is deliberately not here yet -- **No `Asset` field.** `ImageMetadata` is returned, not stored; putting `format` and origin on - the asset row belongs with the ingest pipeline. -- **No blob write.** `thumbnail()` hands back bytes. Storing them content-addressed and - recording a `thumbnail_hash` is the thumbnail-cache task. -- **No video.** `VideoProcessor`, fps probing and frame extraction arrive with the ffmpeg - adapter. +- **No `Asset` field.** `ImageMetadata` and `VideoMetadata` are returned, not stored; putting + `format` and origin on the asset row belongs with the ingest pipeline. +- **No blob write.** `thumbnail()` and `frames()` hand back bytes. Storing them + content-addressed, recording a `thumbnail_hash` and writing a frame's `index`/`timestamp` onto + an asset are the ingest and thumbnail-cache tasks. +- **No `Source`.** Nothing yet records that a clip was registered, at what original rate, or with + what decomposition parameters. `VideoMetadata.fps` is what that record will be built from. diff --git a/docs/workspaces.md b/docs/workspaces.md index 7eebdfe5..f3ee620f 100644 --- a/docs/workspaces.md +++ b/docs/workspaces.md @@ -15,15 +15,20 @@ are the whole format — if WAL is ever adopted, `visionset.db-wal` and `-shm` b of it and this page has to say so, because a user who copies only `visionset.db` under WAL loses committed data. -Two of the four ports have no line in that layout, and that is the point: the [event -bus](events.md) is in-process and the [image processor](media.md) is a decoder, so neither -leaves anything behind. They are composed here anyway, because a workspace is what services +Three of the five ports have no line in that layout, and that is the point: the [event +bus](events.md) is in-process and the two [media processors](media.md) are decoders, so none of +them leaves anything behind. They are composed here anyway, because a workspace is what services are handed and every port has to arrive with it. One of each per open workspace, built by -`event_bus_factory` and `image_processor_factory` — never a module-level singleton, which two -workspaces open at once must not share. +`event_bus_factory`, `image_processor_factory` and `video_processor_factory` — never a +module-level singleton, which two workspaces open at once must not share. + +A new port is appended **last** to `WorkspaceService.__init__`, never inserted: `init` and `open` +bind those arguments positionally, so a parameter added in the middle silently re-binds every one +after it. `WorkspaceService` is the only place in the kernel that names `SqliteMetadataStore`, -`FilesystemBlobStore`, `InProcessEventBus` or `PillowImageProcessor`. Everything above it — later +`FilesystemBlobStore`, `InProcessEventBus`, `PillowImageProcessor` or `FfmpegVideoProcessor`. +Everything above it — later surface, the CLI, MCP — gets an open service and reaches the ports through it, so swapping an adapter is a change to two functions and to nowhere else. @@ -224,9 +229,10 @@ Four habits that keep the boundary honest: workspace-level rules with it. - One `unit_of_work()` per operation, and do the whole operation inside it. - Reach the ports through the handle — `workspace.metadata_store`, `workspace.blob_store`, - `workspace.event_bus`, `workspace.image_processor`. No service other than `workspace_service` - should name `SqliteMetadataStore`, `FilesystemBlobStore`, `InProcessEventBus` or - `PillowImageProcessor` — if a second one does, the composition point has stopped being single. + `workspace.event_bus`, `workspace.image_processor`, `workspace.video_processor`. No service + other than `workspace_service` should name `SqliteMetadataStore`, `FilesystemBlobStore`, + `InProcessEventBus`, `PillowImageProcessor` or `FfmpegVideoProcessor` — if a second one does, + the composition point has stopped being single. - Publish [events](events.md) *after* the `unit_of_work()` block, never inside it. An announcement is about work that committed, and a subscriber that raises must have nothing left to roll back. diff --git a/src/visionset/kernel/__init__.py b/src/visionset/kernel/__init__.py index 5da2e815..947db614 100644 --- a/src/visionset/kernel/__init__.py +++ b/src/visionset/kernel/__init__.py @@ -34,6 +34,7 @@ JobNotFound, LabelClassNotInSchema, MediaError, + MediaToolUnavailable, MissingRequiredAttribute, NoSplitRecipe, NotAWorkspace, @@ -83,6 +84,7 @@ "JobNotFound", "LabelClassNotInSchema", "MediaError", + "MediaToolUnavailable", "MissingRequiredAttribute", "NoSplitRecipe", "NotAWorkspace", diff --git a/src/visionset/kernel/adapters/__init__.py b/src/visionset/kernel/adapters/__init__.py index 24d1b0dd..ac55300f 100644 --- a/src/visionset/kernel/adapters/__init__.py +++ b/src/visionset/kernel/adapters/__init__.py @@ -1,11 +1,13 @@ """Default adapters for the kernel ports (filesystem + SQLite, local-first).""" +from visionset.kernel.adapters.ffmpeg_video_processor import FfmpegVideoProcessor from visionset.kernel.adapters.filesystem_blob_store import FilesystemBlobStore from visionset.kernel.adapters.in_process_event_bus import InProcessEventBus from visionset.kernel.adapters.pillow_image_processor import PillowImageProcessor from visionset.kernel.adapters.sqlite_metadata_store import SqliteMetadataStore __all__ = [ + "FfmpegVideoProcessor", "FilesystemBlobStore", "InProcessEventBus", "PillowImageProcessor", diff --git a/src/visionset/kernel/adapters/ffmpeg_video_processor.py b/src/visionset/kernel/adapters/ffmpeg_video_processor.py new file mode 100644 index 00000000..8d8ac618 --- /dev/null +++ b/src/visionset/kernel/adapters/ffmpeg_video_processor.py @@ -0,0 +1,521 @@ +"""Default VideoProcessor adapter: ffprobe for metadata, ffmpeg for frames.""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import tempfile +from collections.abc import Iterator, Mapping +from pathlib import Path +from typing import IO, Final + +from visionset.kernel.domain import VideoFrame, VideoMetadata +from visionset.kernel.errors import CorruptMedia, MediaToolUnavailable, UnsupportedMedia +from visionset.kernel.ports.video_processor import DEFAULT_EXTRACTION_FPS + +#: The two programs this adapter shells out to. They ship together in every +#: distribution of ffmpeg, and are still checked separately: a method asks for +#: what it is about to run, so a broken install fails where it is used. +_FFMPEG: Final = "ffmpeg" +_FFPROBE: Final = "ffprobe" + +#: Appended to :class:`MediaToolUnavailable`, because "not installed" without a +#: remedy tells an operator nothing they had not already worked out. Worded to +#: match ``tests/fixtures/media.py``, which has to say the same thing to explain +#: a skipped test; the two are duplicated rather than shared because the kernel +#: does not import from ``tests``. +_INSTALL_HINT: Final = ( + "install it with `brew install ffmpeg` (macOS) or `sudo apt-get install ffmpeg` (Debian/Ubuntu)" +) + +#: Everything ffprobe is asked for, in one invocation. +#: +#: ``-select_streams v:0`` because a clip's second video stream is a cover image +#: or a thumbnail track, never the picture; ``-show_format`` because a container +#: often knows its duration when the stream does not. +_PROBE_ARGS: Final = ( + "-v", "error", + "-print_format", "json", + "-show_streams", + "-show_format", + "-select_streams", "v:0", +) # fmt: skip + +#: Every ffmpeg argument that decides what comes out, pinned. Each one earns it: +#: +#: ``-xerror`` is load-bearing and the least obvious. Without it ffmpeg reports a +#: truncated clip on stderr, exits **zero**, and hands back the frames it managed +#: to decode — so a damaged file would ingest as a short one and nothing would +#: ever say so. Its opposite number, ``-err_detect explode``, is deliberately not +#: here: it rejects perfectly good files. +#: +#: ``fps=...:round=up`` is the extraction grid *and* the reason +#: ``timestamp = index / fps`` is honest. The filter maps each input frame onto an +#: output slot and the last one to land there wins; under the default ``near`` the +#: winner is the frame nearest the slot's midpoint, so at 1 fps a clip shot at 10 +#: yields the pictures from 0.4 s and 1.4 s while claiming 0.0 s and 1.0 s. +#: ``up`` makes the frame at the grid point itself the winner, which puts the +#: reported timestamp within one *source* frame of the pixels — the best any +#: resampler can do. Frame counts are identical either way. +#: +#: There is **no seek**. Input seeking lands on a keyframe and is approximate, +#: output seeking interacts with the filter; extraction reads the clip from the +#: start every time, which is what makes it reproducible. +#: +#: ``-pred`` and ``-compression_level`` pin the PNG encoder, whose bytes are the +#: content hash of the asset each frame becomes; ``+bitexact`` strips the encoder +#: version string that would otherwise change with every ffmpeg upgrade. Together +#: they buy determinism *within one installed ffmpeg* — never across builds. +_EXTRACTION_ARGS: Final = ( + "-an", + "-f", "image2pipe", + "-c:v", "png", + "-pred", "none", + "-compression_level", "6", + "-fflags", "+bitexact", + "-flags:v", "+bitexact", +) # fmt: skip + +#: ``\x89PNG\r\n\x1a\n``. Every frame on the pipe starts with it. +_PNG_SIGNATURE: Final = b"\x89PNG\r\n\x1a\n" + +#: A PNG chunk header is a four-byte big-endian length plus a four-byte type, +#: and the payload is followed by a four-byte CRC. Walking those is how frames +#: are split apart; searching for the signature would risk matching a byte +#: sequence inside compressed pixel data. +_CHUNK_HEADER: Final = 8 +_CHUNK_CRC: Final = 4 +_END_CHUNK: Final = b"IEND" + +#: How long a decoder gets to exit after it is asked politely, before it is +#: killed. It is being told to stop reading a local file; a second is generous. +_STOP_TIMEOUT: Final = 1.0 + +#: How much of the decoder's own complaint to quote back. Enough to name the +#: cause, short enough that ``MediaError.reason`` stays a table cell. +_DIAGNOSTIC_LINES: Final = 2 + +#: What a path becomes when it is taken out of a quoted diagnostic. ffmpeg names +#: the file it was given in most of its errors, and ``MediaError.reason`` is not +#: allowed to: the name is a column of the report, not a prefix on every sentence +#: in it. Redacted rather than dropped, so the sentence still parses and it is +#: obvious something was removed. +_REDACTED: Final = "" + + +def _require_tool(program: str) -> str: + """The program's path, or refuse before anything is attempted. + + Looked up per call rather than cached on the instance, which keeps the + adapter as stateless as its image sibling. The cost is one PATH scan per + clip — not per frame — against a decode of the whole file. + """ + found = shutil.which(program) + if found is None: + raise MediaToolUnavailable(f"{program} is not installed or not on PATH; {_INSTALL_HINT}") + return found + + +def _require_file(source: Path) -> None: + """A plain ``FileNotFoundError``, deliberately outside the ``MediaError`` family. + + Nothing is wrong with the media: there is no media. Handing this to an ingest + as a per-file media error would file "the operator deleted it mid-run" under + the same heading as "this codec is not supported", which are not the same + conversation. + """ + if not source.is_file(): + raise FileNotFoundError(f"no video file at {source}") + + +def _clip_name(source: Path, name: str | None) -> str: + """What to call this clip in an error: the caller's word, then the path. + + Unlike the image processor's version this never answers ``None``. There is + always a path — that is what taking a ``Path`` instead of a stream buys — so + an unnamed clip is not a case that exists. + """ + return name if name is not None else str(source) + + +def _rational(value: object) -> float | None: + """ffprobe's ``"30000/1001"`` as 29.97. Also accepts a plain number. + + ``0/0`` is ffprobe's way of saying it does not know, and comes back as + ``None`` rather than as a division error. + """ + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) if value > 0 else None + if not isinstance(value, str): + return None + numerator, separator, denominator = value.partition("/") + try: + result = float(numerator) / float(denominator) if separator else float(numerator) + except (ValueError, ZeroDivisionError): + return None + return result if result > 0 else None + + +def _positive_number(value: object) -> float | None: + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value) if value > 0 else None + if not isinstance(value, str): + return None + try: + parsed = float(value) + except ValueError: + return None + return parsed if parsed > 0 else None + + +def _integer(value: object) -> int | None: + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + return round(value) + if isinstance(value, str): + try: + return round(float(value)) + except ValueError: + return None + return None + + +def _text(mapping: Mapping[str, object], key: str) -> str | None: + value = mapping.get(key) + return value if isinstance(value, str) else None + + +def _rotation(stream: Mapping[str, object]) -> int: + """How far the container says to turn this picture before showing it. + + Two places to look, because the vocabulary moved. Modern files carry a + display matrix in the stream's side data; older ones carry a ``rotate`` tag, + which recent ffmpeg no longer even writes but still reads. Absent from both + is the ordinary case and means upright. + """ + side_data = stream.get("side_data_list") + if isinstance(side_data, list): + for entry in side_data: + if isinstance(entry, dict) and _text(entry, "side_data_type") == "Display Matrix": + degrees = _integer(entry.get("rotation")) + if degrees is not None: + return degrees % 360 + tags = stream.get("tags") + if isinstance(tags, dict): + degrees = _integer(tags.get("rotate")) + if degrees is not None: + return degrees % 360 + return 0 + + +def _oriented(width: int, height: int, rotation: int) -> tuple[int, int]: + """The edges as a viewer sees them: swapped by a quarter turn, not by a half. + + Pure, and separately swept by a test, because it is the one line where a + dataset silently acquires sideways dimensions. ``% 180`` rather than a + membership test so that -90, 270 and 630 all mean the same thing, which is + what ffprobe's mix of conventions requires. + """ + return (height, width) if rotation % 180 == 90 else (width, height) + + +def _read_exactly(stream: IO[bytes], count: int) -> bytes: + """``count`` bytes, or everything left if the stream ends first. + + A pipe hands back what has arrived, not what was asked for, so every read + here loops. Short output is not an error at this level: it means the decoder + stopped, and its exit code is what says whether that was the end of the clip + or the end of the file. + """ + chunks: list[bytes] = [] + remaining = count + while remaining > 0: + chunk = stream.read(remaining) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) + + +def _png_frames(stream: IO[bytes]) -> Iterator[bytes]: + """Split a concatenated PNG stream into whole images, by walking its chunks. + + ``image2pipe`` writes one complete PNG after another with nothing between + them, so the frames have to be found rather than delimited. Chunk lengths are + declared, which makes this exact; scanning for the signature would be a guess + that happens to work until a compressed scanline reproduces those eight bytes. + + A truncated frame is dropped rather than yielded — half a picture is not a + picture — and the caller's exit-code check is what turns that into an error. + """ + while True: + signature = _read_exactly(stream, len(_PNG_SIGNATURE)) + if signature != _PNG_SIGNATURE: + return + parts = [signature] + while True: + header = _read_exactly(stream, _CHUNK_HEADER) + if len(header) < _CHUNK_HEADER: + return + payload = _read_exactly(stream, int.from_bytes(header[:4], "big") + _CHUNK_CRC) + parts.append(header) + parts.append(payload) + if header[4:] == _END_CHUNK: + break + yield b"".join(parts) + + +def _stop(process: subprocess.Popen[bytes]) -> None: + """Leave no decoder running, however the iteration ended. + + Reached on the ordinary path (where the process has already exited and this + does nothing) and on the one that matters: a caller that breaks out of the + loop closes the generator, which raises ``GeneratorExit`` at the ``yield`` + and lands here with ffmpeg still decoding a file nobody is reading. + """ + if process.poll() is not None: + return + if process.stdout is not None: + process.stdout.close() + process.terminate() + try: + process.wait(timeout=_STOP_TIMEOUT) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + +def _redacted(detail: str, *names: str) -> str: + """One line of decoder output with every mention of the file taken out. + + ``MediaError.reason`` never repeats the name — that is what makes an ingest + report a table instead of five thousand sentences — and ffmpeg quotes the + path it was handed in nearly everything it complains about. Whitespace is + squeezed afterwards, because the quoted line may have been wrapped. + """ + for name in names: + if name: + detail = detail.replace(name, _REDACTED) + return " ".join(detail.split()) + + +def _last_lines(output: str, *names: str) -> str: + """The last thing the decoder said, as one line fit for an error report.""" + lines = [line.strip() for line in output.splitlines() if line.strip()] + tail = [_redacted(line, *names) for line in lines[-_DIAGNOSTIC_LINES:]] + return "; ".join(tail) if tail else "the decoder gave no reason" + + +def _diagnostic(diagnostics: IO[bytes], *names: str) -> str: + diagnostics.seek(0) + return _last_lines(diagnostics.read().decode("utf-8", "replace"), *names) + + +class FfmpegVideoProcessor: + """Probes clips with ffprobe and decomposes them into frames with ffmpeg. + + Holds no state at all — no handle, no cache, no configuration — so + ``WorkspaceService`` builds one per workspace from a zero-argument factory + and never closes it. What *does* need closing is an iterator: a live + :meth:`frames` owns a running decoder, and the workspace knows nothing about + it. + + **ffmpeg is a binary, not a dependency.** It is not installable from + ``pyproject.toml`` and a machine can perfectly well run VisionSet's image + half without it, so its absence is discovered here rather than at import and + reported as ``MediaToolUnavailable`` — which is not a ``MediaError``, because + no file is at fault. + + **Validation is the decode**, as it is for images: :meth:`probe` reads what + the container declares, and :meth:`frames` refuses the clip whose pixels + actually fail. The two refusals divide by remedy rather than by symptom, and + the division is enforced by the order they run in. A container ffmpeg cannot + open at all is ``UnsupportedMedia`` — including a clip whose index went + missing with its tail, which is why probing before extracting matters. A + clip that opens and then runs out mid-decode is ``CorruptMedia``. + + **The extracted frames are the product, not the video.** Nothing here stores + or re-encodes the clip; it is a source, and what enters the dataset are PNGs + in the port's ``FRAME_FORMAT``. That is why there is no accepted-codec list + to keep up to date — see ``domain/media.py``. + """ + + def probe(self, source: Path, *, name: str | None = None) -> VideoMetadata: + """What this clip is, at the dimensions a viewer would show. + + Raises: + MediaToolUnavailable: ffprobe is not installed. + FileNotFoundError: there is nothing at ``source``. + UnsupportedMedia: not a video ffprobe can open, or nothing in it is + an identifiable video stream. + CorruptMedia: it opened, and what it says about itself is unusable. + """ + ffprobe = _require_tool(_FFPROBE) + _require_file(source) + clip = _clip_name(source, name) + + document = _run_ffprobe(ffprobe, source, clip) + stream = _video_stream(document, clip) + + width = _integer(stream.get("width")) + height = _integer(stream.get("height")) + if width is None or height is None or width < 1 or height < 1: + raise CorruptMedia("the video stream declares no usable dimensions", name=clip) + + codec = _text(stream, "codec_name") + if codec is None: + raise UnsupportedMedia("the video stream carries no identifiable codec", name=clip) + + # avg_frame_rate is frames over duration; r_frame_rate is the base rate + # ffmpeg guessed, which doubles on interlaced content. Prefer the honest + # one and fall back only when the container declined to compute it. + fps = _rational(stream.get("avg_frame_rate")) or _rational(stream.get("r_frame_rate")) + if fps is None: + raise CorruptMedia("the video stream declares no frame rate", name=clip) + + duration = _positive_number(stream.get("duration")) + if duration is None: + container = document.get("format") + if isinstance(container, Mapping): + duration = _positive_number(container.get("duration")) + if duration is None: + raise CorruptMedia( + "neither the stream nor the container declares a duration", name=clip + ) + + oriented_width, oriented_height = _oriented(width, height, _rotation(stream)) + return VideoMetadata( + width=oriented_width, + height=oriented_height, + fps=fps, + duration_seconds=duration, + codec=codec, + ) + + def frames( + self, + source: Path, + *, + fps: float = DEFAULT_EXTRACTION_FPS, + name: str | None = None, + ) -> Iterator[VideoFrame]: + """Frames taken off ``source`` at ``fps``, one at a time, in order. + + Not a generator itself, on purpose. A generator's body does not run until + something asks it for a value, so writing it that way would report a + missing ffmpeg — or a negative ``fps`` — at the first iteration, in + whatever loop happened to consume it, rather than here where the mistake + was made. + + The iterator owns a running decoder until it is exhausted or closed. + + Raises: + ValueError: ``fps`` is not positive. A programming error rather than + a media one, so it is not translated into the ``MediaError`` + family. + MediaToolUnavailable: ffmpeg is not installed. + FileNotFoundError: there is nothing at ``source``. + UnsupportedMedia: ffmpeg never opened the clip. Raised on first + iteration, because that is when the decoder has said so. + CorruptMedia: ffmpeg opened the clip and its bytes ran out. Raised + after the frames that did decode have been yielded. + """ + if fps <= 0: + raise ValueError(f"fps must be greater than zero, got {fps}") + ffmpeg = _require_tool(_FFMPEG) + _require_file(source) + return _extract(ffmpeg, source, fps, _clip_name(source, name)) + + +def _run_ffprobe(ffprobe: str, source: Path, clip: str) -> Mapping[str, object]: + """One ffprobe run, or the refusal that says this is not a video.""" + result = subprocess.run( + [ffprobe, *_PROBE_ARGS, str(source)], + capture_output=True, + stdin=subprocess.DEVNULL, + ) + if result.returncode != 0: + detail = _last_lines(result.stderr.decode("utf-8", "replace"), str(source), clip) + raise UnsupportedMedia(f"not a video ffmpeg can open ({detail})", name=clip) + try: + document = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise UnsupportedMedia(f"the probe returned nothing readable ({exc})", name=clip) from exc + if not isinstance(document, dict): + raise UnsupportedMedia("the probe returned no description of this file", name=clip) + return document + + +def _video_stream(document: Mapping[str, object], clip: str) -> Mapping[str, object]: + """The first video stream, or the refusal for a file that has none. + + An audio file, or a container holding only subtitles, gets here rather than + failing the probe: ffprobe reads it happily and reports no picture. The + remedy is the operator's, so it is ``UnsupportedMedia``. + """ + streams = document.get("streams") + if isinstance(streams, list): + for entry in streams: + if isinstance(entry, dict) and _text(entry, "codec_type") == "video": + return entry + raise UnsupportedMedia("the file holds no video stream", name=clip) + + +def _extract(ffmpeg: str, source: Path, fps: float, clip: str) -> Iterator[VideoFrame]: + """Stream frames off one ffmpeg process, and account for how it ended. + + stderr goes to a temporary **file** rather than to a pipe, which is not + fastidiousness: reading stdout while an unread stderr pipe fills its buffer + deadlocks, and a damaged clip — the exact input this has to survive — is what + makes ffmpeg talkative. + """ + command = [ + ffmpeg, + "-nostdin", + "-loglevel", "error", + "-xerror", + "-i", str(source), + "-vf", f"fps=fps={fps}:round=up", + *_EXTRACTION_ARGS, + "-", + ] # fmt: skip + + with tempfile.TemporaryFile() as diagnostics: + process = subprocess.Popen( + command, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=diagnostics, + ) + stdout = process.stdout + if stdout is None: # unreachable: the call above asked for a pipe + _stop(process) + raise RuntimeError("ffmpeg was started without a readable stdout") + + produced = 0 + try: + for index, content in enumerate(_png_frames(stdout)): + produced = index + 1 + # The grid, not the source's own presentation timestamps: with + # round=up the filter emits the frame sitting at each grid point, + # so this is the arithmetic ffmpeg would have produced anyway, + # without a second stream of text to parse. It counts from the + # start of the stream, so a clip whose first frame is not at zero + # is reported relative to its own beginning. + yield VideoFrame(index=index, timestamp=index / fps, content=content) + returncode = process.wait() + finally: + _stop(process) + + if returncode != 0: + detail = _diagnostic(diagnostics, str(source), clip) + if produced == 0: + # It never got a picture out, so there was nothing here to break: + # an intact file that is not one we can read. + raise UnsupportedMedia(f"ffmpeg could not decode this video ({detail})", name=clip) + raise CorruptMedia( + f"the video is damaged or truncated after {produced} frames ({detail})", name=clip + ) diff --git a/src/visionset/kernel/domain/__init__.py b/src/visionset/kernel/domain/__init__.py index 4aac9833..ba752e0d 100644 --- a/src/visionset/kernel/domain/__init__.py +++ b/src/visionset/kernel/domain/__init__.py @@ -33,7 +33,12 @@ PolygonGeometry, ) from visionset.kernel.domain.ingest import IngestJob, IngestState -from visionset.kernel.domain.media import ImageFormat, ImageMetadata +from visionset.kernel.domain.media import ( + ImageFormat, + ImageMetadata, + VideoFrame, + VideoMetadata, +) from visionset.kernel.domain.names import normalize_name from visionset.kernel.domain.partition import ( BySegments, @@ -140,6 +145,8 @@ "SplitAssignment", "SplitRecipe", "TaskGroup", + "VideoFrame", + "VideoMetadata", "Workspace", "assign_split", "canonical_bytes", diff --git a/src/visionset/kernel/domain/media.py b/src/visionset/kernel/domain/media.py index 29eda82d..7aafbf25 100644 --- a/src/visionset/kernel/domain/media.py +++ b/src/visionset/kernel/domain/media.py @@ -1,4 +1,4 @@ -# usage: from visionset.kernel.domain import ImageFormat, ImageMetadata +# usage: from visionset.kernel.domain import ImageFormat, ImageMetadata, VideoFrame, VideoMetadata """What a media file turns out to be, once something has decoded it. These are the *result* types of the media ports. They live in the domain rather @@ -7,8 +7,9 @@ REST surface serializes one. A port's vocabulary may be domain; a domain model may never be a port's. -``VideoMetadata`` joins this module when the video processor lands, which is why -the file is named for the concept and not for one modality. +Both modalities live here, which is why the file is named for the concept and not +for one of them: :class:`ImageMetadata` for a still, :class:`VideoMetadata` and +:class:`VideoFrame` for a clip and the frames taken out of it. **The accepted list is** :class:`ImageFormat` **, and nothing else.** Extending it is two adjacent edits — a member here, and the decoder's own spelling of it in @@ -22,6 +23,16 @@ WEBP is the obvious next member. It is not here yet because a format with no generated fixture is a format nobody is testing. + +**There is deliberately no ``VideoFormat`` beside it.** The asymmetry is the +point: an image is an asset, a video is a source. Curating :class:`ImageFormat` +buys something, because those exact bytes enter the dataset and the promise above +is made about them. A video's bytes never do — they leave the decoder as frames, +which are :attr:`ImageFormat.PNG` like any other still — so a closed list of +codecs would gate nothing while going stale every time a camera vendor ships a +profile. :attr:`VideoMetadata.codec` therefore *records* what was read instead of +*deciding* what may be read, the way ``DatasetChange.operation`` is a ``str`` +while ``DatasetOperation`` is the enum a writer picks from. """ from __future__ import annotations @@ -73,3 +84,70 @@ class ImageMetadata(BaseModel): width: int = Field(ge=1) height: int = Field(ge=1) format: ImageFormat + + +class VideoMetadata(BaseModel): + """What one clip turns out to be: how big, how fast, how long, and in what codec. + + **Dimensions are as displayed**, on exactly the terms :class:`ImageMetadata` + states. A video carries its rotation in a display matrix rather than in an + EXIF tag, and a phone shooting in portrait writes a landscape stream plus a + quarter turn; ffmpeg applies that turn when it decodes, so reporting the + stored numbers would describe a picture nobody will ever see. There is no + ``rotation_applied`` flag, for the reason there is no ``orientation_applied`` + one: a caller that could branch on it is a caller who was handed the + un-normalized case after all. + + :attr:`fps` is the *source* rate, which is provenance and not a decision — + what a decomposition ran at is a parameter the caller chose and the ingest + records. It is a ``float`` rather than a rational because 30000/1001 is going + to be reported as 29.97 by every surface that shows it, and carrying the + fraction only to divide it at the edge buys nothing. + + :attr:`codec` is a plain ``str``. See the module docstring: this file has no + ``VideoFormat`` enum on purpose. + + **There is no frame count.** For a variable-rate stream it would be a + guess, for a constant-rate one it is ``fps * duration_seconds``, and neither + is the number an ingest actually needs — that one is how many frames the + extraction produced, which only the caller doing the extraction can count. + + Frozen, like every other value in the domain that is a pure function of some + bytes. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + width: int = Field(ge=1) + height: int = Field(ge=1) + fps: float = Field(gt=0) + duration_seconds: float = Field(gt=0) + codec: str = Field(min_length=1) + + +class VideoFrame(BaseModel): + """One still lifted out of a clip, with enough provenance to say where from. + + Transient, unlike its neighbours here: nothing stores a ``VideoFrame``. The + :attr:`content` goes to the blob store and the two numbers go onto an + ``Asset``, so this type exists to keep them together for the length of one + loop iteration. It lives in the domain anyway because the kernel passes it + between a port and a service, which is the whole test for belonging here. + + :attr:`index` counts **the extracted sequence**, not the source. A source + frame number means nothing for a variable-rate stream and cannot be + reproduced without knowing the rate the file was shot at; the extracted index + is what orders the assets and what names them. :attr:`timestamp` is the + locator that survives — it says where in the clip to look, whatever rate the + next decomposition runs at. + + :attr:`content` is a complete, self-contained image in the port's + ``FRAME_FORMAT``. Hashing it is what gives the resulting asset its identity, + which is why the encoder producing it is pinned rather than left to a default. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + index: int = Field(ge=0) + timestamp: float = Field(ge=0) + content: bytes = Field(min_length=1) diff --git a/src/visionset/kernel/errors.py b/src/visionset/kernel/errors.py index 694ea2fb..1d2c7fd8 100644 --- a/src/visionset/kernel/errors.py +++ b/src/visionset/kernel/errors.py @@ -520,3 +520,22 @@ class CorruptMedia(MediaError): whose bytes nobody can read until a training run trips over it. That is why probing pays for a full decode per file instead of sniffing. """ + + +class MediaToolUnavailable(VisionSetError): + """A media adapter needs an external program, and it is not installed. + + Deliberately **not** a ``MediaError``, and the distance is the whole point. + Every error in that family answers "what is wrong with this file?"; this one + answers "what is wrong with this machine?". An ingest catches the media + family per item and carries on, so if this were in it, a missing decoder + would be recorded five thousand times against five thousand innocent files + and the run would report a data problem it does not have. It is the *fatal + cause* an ingest job records once, next to the per-file report. + + Named for the tool rather than for the program, because the kernel does not + know which program: only the adapter does, and only the adapter spells its + name. The message is where that lands, and it carries an install hint — the + remedy here is a package manager, so an error that merely says "unavailable" + has told the operator nothing they did not already suspect. + """ diff --git a/src/visionset/kernel/ports/__init__.py b/src/visionset/kernel/ports/__init__.py index 2746fcd7..6d552039 100644 --- a/src/visionset/kernel/ports/__init__.py +++ b/src/visionset/kernel/ports/__init__.py @@ -21,9 +21,16 @@ UnitOfWork, ) from visionset.kernel.ports.model_provider import ModelProvider +from visionset.kernel.ports.video_processor import ( + DEFAULT_EXTRACTION_FPS, + FRAME_FORMAT, + VideoProcessor, +) __all__ = [ + "DEFAULT_EXTRACTION_FPS", "DEFAULT_THUMBNAIL_MAX_EDGE", + "FRAME_FORMAT", "THUMBNAIL_FORMAT", "UNINITIALIZED", "AuthProvider", @@ -36,4 +43,5 @@ "ModelProvider", "Repository", "UnitOfWork", + "VideoProcessor", ] diff --git a/src/visionset/kernel/ports/video_processor.py b/src/visionset/kernel/ports/video_processor.py new file mode 100644 index 00000000..d5f29c94 --- /dev/null +++ b/src/visionset/kernel/ports/video_processor.py @@ -0,0 +1,106 @@ +"""Video: say what a clip is, and take frames out of it at a rate someone chose. + +The video half of what the ``MediaProcessor`` placeholder was going to be. It is +declared apart from the image processor because the two have almost nothing in +common at the type level: an ffmpeg adapter has no thumbnail to serve and a +Pillow adapter has no frames to iterate. One shared protocol would have forced +each implementation to declare the other's methods and raise from them, which is +a runtime failure where a compile-time absence was available. + +Both methods take a :class:`~pathlib.Path`, where ``ImageProcessor`` takes a +stream, and that divergence is deliberate rather than an oversight to be tidied +up later. A video decoder is an out-of-process program that seeks: handed a pipe +it cannot ask how long the clip is without decoding all of it, and cannot revisit +a byte it has already read. Nothing is lost by requiring a file, because there is +no caller that has video bytes and no file — a source is a path on disk and a +blob in the default blob store is a path too. +""" + +from collections.abc import Iterator +from pathlib import Path +from typing import Final, Protocol, runtime_checkable + +from visionset.kernel.domain import ImageFormat, VideoFrame, VideoMetadata + +#: How many frames a second to take out of a clip, unless a caller says otherwise. +#: +#: One. A dataset wants pictures that differ from each other, and consecutive +#: frames of the same clip mostly do not — at a source rate of thirty, twenty-nine +#: of every thirty frames cost a decode, a hash and a row to say almost exactly +#: what their neighbour said. A caller that wants more asks for more. +DEFAULT_EXTRACTION_FPS: Final = 1.0 + +#: The encoding every extracted frame arrives in. +#: +#: PNG, because a frame is ground truth: an annotator draws boxes on these pixels +#: and an exporter ships them, so a lossy re-encode would put compression +#: artefacts under the labels for a saving nobody asked for. Fixed by the port +#: rather than chosen per call, for the reason ``THUMBNAIL_FORMAT`` is — the bytes +#: are content-addressed, and a per-call format would give one frame several +#: equally correct hashes. +#: +#: That this is an ``ImageFormat`` member is load-bearing and not decoration: it +#: is what lets an ingest hand an extracted frame straight to an ``ImageProcessor`` +#: and stamp the result on an ``Asset``, with no branch for where the picture came +#: from. +FRAME_FORMAT: Final = ImageFormat.PNG + + +@runtime_checkable +class VideoProcessor(Protocol): + """Reads video: what a clip is, and the frames a caller wants out of it. + + Two methods rather than one, for the reason ``ImageProcessor`` has two: the + callers are different. Registering a source wants the rate and the duration + and no pixels at all; an ingest run wants the pixels and already knows the + rate. A combined call would make the first pay for a full decode. + + Four rules an implementation owes its callers: + + - **Dimensions are as displayed.** Whatever rotation the container carries + has been applied by the time the numbers come back, and the frames + :meth:`frames` yields have those same dimensions. See + :class:`~visionset.kernel.domain.VideoMetadata`. + - **Extraction is deterministic.** The same file and the same ``fps`` produce + the same frames, byte for byte — which is what lets content addressing + dedup a re-ingest instead of doubling a dataset. The promise holds *within + one installed decoder*: a different ffmpeg build may encode the same + picture to different bytes, so a caller asserts repeatability and never a + literal hash. + - **Frames arrive lazily, and the iterator owns a running program.** Nothing + buffers a whole clip. A caller either exhausts the iterator or closes it + (``contextlib.closing``, or letting a ``for`` loop go out of scope), and an + implementation must not leave a decoder running when it is abandoned. + - **Every frame is a complete image in** :data:`FRAME_FORMAT`, at the + dimensions :meth:`probe` reported. + + ``name`` is what the caller calls this clip, and it exists only so a refusal + can say which item failed; an implementation may fall back to the file's own + name, but it never invents one. The convention for naming a frame further + down the pipeline is ``clip.mp4#frame=42`` — see ``MediaError``. + + **Asking for more frames a second than the clip has duplicates them.** This + is documented rather than clamped: clamping would mean probing inside + :meth:`frames`, and an ingest content-addresses what it stores, so the + duplicates collapse into one asset anyway. The honest fix is not to ask. + + Raises: + MediaToolUnavailable: the decoder this implementation shells out to is + not installed. Raised before any work starts, and deliberately not a + ``MediaError`` — no file is at fault. + UnsupportedMedia: the bytes are not a video the decoder can open. + CorruptMedia: the decoder opened the clip and its bytes ran out. + FileNotFoundError: there is nothing at ``source``. + ValueError: ``fps`` is not positive. A programming error rather than a + media one, so it is not translated into the ``MediaError`` family. + """ + + def probe(self, source: Path, *, name: str | None = None) -> VideoMetadata: ... + + def frames( + self, + source: Path, + *, + fps: float = DEFAULT_EXTRACTION_FPS, + name: str | None = None, + ) -> Iterator[VideoFrame]: ... diff --git a/src/visionset/kernel/services/workspace_service.py b/src/visionset/kernel/services/workspace_service.py index 6d0c5f83..9e78f7b4 100644 --- a/src/visionset/kernel/services/workspace_service.py +++ b/src/visionset/kernel/services/workspace_service.py @@ -4,8 +4,8 @@ Every kernel operation happens in the context of exactly one workspace, so this module is also the **single composition point** for the default adapters. It is the only place in the kernel that names ``SqliteMetadataStore``, -``FilesystemBlobStore``, ``InProcessEventBus`` or ``PillowImageProcessor``; -everything above it — later +``FilesystemBlobStore``, ``InProcessEventBus``, ``PillowImageProcessor`` or +``FfmpegVideoProcessor``; everything above it — later services, the REST surface, the CLI, MCP — receives an open ``WorkspaceService`` and reaches the ports through it. Swapping an adapter is therefore a change to two functions here and to nowhere else. @@ -15,10 +15,10 @@ /visionset.db the metadata store; holds format_version /blobs/ the content-addressed blob store -Two of the four ports have no line in that layout, and that is the point: the -event bus is in-process and the image processor is a decoder, so neither leaves -anything behind. They are composed here anyway, because a workspace is what -services are handed and every port has to arrive with it. +Three of the five ports have no line in that layout, and that is the point: the +event bus is in-process and the two media processors are decoders, so none of +them leaves anything behind. They are composed here anyway, because a workspace +is what services are handed and every port has to arrive with it. There is no sidecar file carrying the format version. It lives inside the database it describes, for the same reason there is no alembic ledger: a second @@ -50,6 +50,7 @@ from uuid import UUID from visionset.kernel.adapters import ( + FfmpegVideoProcessor, FilesystemBlobStore, InProcessEventBus, PillowImageProcessor, @@ -70,6 +71,7 @@ ImageProcessor, MetadataStore, UnitOfWork, + VideoProcessor, ) #: The metadata store. Its presence is what makes a directory a workspace. @@ -90,6 +92,11 @@ #: decoder has no state at all. It is composed here anyway, because this module is #: the only one allowed to name an adapter. type ImageProcessorFactory = Callable[[], ImageProcessor] +#: Zero-argument, like its image sibling. The video decoder needs an external +#: program rather than a library, but that is the adapter's problem and not the +#: workspace's: a missing ffmpeg is discovered by the call that needs it, so a +#: machine without one still opens workspaces and still ingests images. +type VideoProcessorFactory = Callable[[], VideoProcessor] def _resolved(path: Path | str) -> Path: @@ -103,7 +110,7 @@ def _resolved(path: Path | str) -> Path: class WorkspaceService: - """One open workspace: its identity, its directory, and its four ports. + """One open workspace: its identity, its directory, and its five ports. Instances come from :meth:`init` and :meth:`open`. Constructing one directly is the injection seam — hand it ports and nothing here touches a disk. @@ -113,11 +120,15 @@ class WorkspaceService: *factories* on ``init``/``open`` carry them instead: no module-level singleton, no import from a delivery module, and each default named once. - The event bus and the image processor have no path to be derived from and - could have been plain defaults, but they take the same shape anyway — one + The event bus and the two media processors have no path to be derived from + and could have been plain defaults, but they take the same shape anyway — one place naming each default, one of each per open workspace. A module-level default would be a singleton shared by every workspace in the process, which is precisely the thing two workspaces open at once must not have. + + A new port is appended **last** to this signature, never inserted. Both + classmethods below bind these arguments positionally, so a parameter added in + the middle silently re-binds every one after it. """ def __init__( @@ -128,6 +139,7 @@ def __init__( blob_store: BlobStore, event_bus: EventBus, image_processor: ImageProcessor, + video_processor: VideoProcessor, ) -> None: self._root = root self._workspace = workspace @@ -135,6 +147,7 @@ def __init__( self._blob_store = blob_store self._event_bus = event_bus self._image_processor = image_processor + self._video_processor = video_processor # --- composition: the only two ways to get one ------------------------ @@ -148,6 +161,7 @@ def init( blob_store_factory: BlobStoreFactory = FilesystemBlobStore, event_bus_factory: EventBusFactory = InProcessEventBus, image_processor_factory: ImageProcessorFactory = PillowImageProcessor, + video_processor_factory: VideoProcessorFactory = FfmpegVideoProcessor, ) -> WorkspaceService: """Create a workspace at ``path`` and return it open. @@ -195,9 +209,9 @@ def init( metadata_store.close() _undo_init(root, created_root=created_root) raise - # Both remaining factories are zero-argument and cannot touch the disk, so - # they run outside the block that would undo a half-made workspace. A - # future port whose construction can fail moves inside it. + # The three remaining factories are zero-argument and cannot touch the + # disk, so they run outside the block that would undo a half-made + # workspace. A future port whose construction can fail moves inside it. return cls( root, workspace, @@ -205,6 +219,7 @@ def init( blob_store, event_bus_factory(), image_processor_factory(), + video_processor_factory(), ) @classmethod @@ -216,6 +231,7 @@ def open( blob_store_factory: BlobStoreFactory = FilesystemBlobStore, event_bus_factory: EventBusFactory = InProcessEventBus, image_processor_factory: ImageProcessorFactory = PillowImageProcessor, + video_processor_factory: VideoProcessorFactory = FfmpegVideoProcessor, ) -> WorkspaceService: """Open the workspace at ``path``, migrating it forward if it is behind. @@ -270,6 +286,7 @@ def open( blob_store, event_bus_factory(), image_processor_factory(), + video_processor_factory(), ) # --- what the surfaces and the later services read -------------------- @@ -312,7 +329,7 @@ def event_bus(self) -> EventBus: def image_processor(self) -> ImageProcessor: """Decoding, dimensions and thumbnails for still images. - Reached through the handle like the other three ports, which is what lets + Reached through the handle like the other four ports, which is what lets an ingest service take a ``WorkspaceService`` and still name no adapter — the rule this module exists to keep, and the reason a decoder is composed here rather than defaulted in the service that uses it. @@ -325,6 +342,23 @@ def image_processor(self) -> ImageProcessor: """ return self._image_processor + @property + def video_processor(self) -> VideoProcessor: + """Probing and frame extraction for video. + + Composed on exactly the terms the image processor is, including the part + that looks like it should be an exception: the default adapter needs + ffmpeg on the machine, and building one still cannot fail. That is + deliberate. Checking for the binary here would mean a workspace full of + JPEGs refuses to open on a laptop with no ffmpeg installed, so the check + belongs to the call that actually needs to decode something. + + The one way this port differs in use: what it returns owns a running + program. A workspace has no say in that lifetime — the iterator does — so + :meth:`close` has nothing to do here either. + """ + return self._video_processor + @property def format_version(self) -> int: return self._metadata_store.format_version @@ -386,10 +420,11 @@ def require_project_name( def close(self) -> None: """Release the metadata store's connections. Safe to call twice. - The other three ports are not closed and have nothing to close: the blob + The other four ports are not closed and have nothing to close: the blob store addresses files by hash and opens them per call, the event bus holds - a list of callables, and the image processor holds nothing whatsoever. - Only the database keeps a connection. + a list of callables, and the two media processors hold nothing whatsoever. + Only the database keeps a connection. A frame iterator does own a running + decoder, but it belongs to whoever asked for it, not to the workspace. """ self._metadata_store.close() diff --git a/tests/fixtures/media.py b/tests/fixtures/media.py index b6b5b956..b1b5c2bf 100644 --- a/tests/fixtures/media.py +++ b/tests/fixtures/media.py @@ -210,6 +210,12 @@ def frame_count(self) -> int: return round(self.fps * self.duration_seconds) +def _run_ffmpeg(command: list[str], path: Path) -> None: + result = subprocess.run(command, capture_output=True, text=True) + if result.returncode != 0: + raise RuntimeError(f"ffmpeg failed generating {path.name}:\n{result.stderr}") + + def write_video( path: Path, *, @@ -222,27 +228,43 @@ def write_video( `-fflags/-flags:v +bitexact` strip the encoder version and timestamps that would otherwise make two runs of the same command differ. + + `-movflags +faststart` puts the index at the front of the file. Nothing in an intact clip + cares, but it is what makes `write_corrupt_video` able to produce a *partially* readable + file rather than an unopenable one — see there. """ require_ffmpeg() width, height = size path.parent.mkdir(parents=True, exist_ok=True) - command = [ - "ffmpeg", - "-nostdin", - "-loglevel", "error", - "-f", "lavfi", - "-i", f"testsrc=size={width}x{height}:rate={fps}:duration={duration_seconds}", - "-pix_fmt", "yuv420p", - "-c:v", "libx264", - "-preset", "ultrafast", - "-g", str(fps), - "-fflags", "+bitexact", - "-flags:v", "+bitexact", - "-y", str(path), - ] # fmt: skip - result = subprocess.run(command, capture_output=True, text=True) - if result.returncode != 0: - raise RuntimeError(f"ffmpeg failed generating {path.name}:\n{result.stderr}") + _run_ffmpeg( + [ + "ffmpeg", + "-nostdin", + "-loglevel", + "error", + "-f", + "lavfi", + "-i", + f"testsrc=size={width}x{height}:rate={fps}:duration={duration_seconds}", + "-pix_fmt", + "yuv420p", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-g", + str(fps), + "-movflags", + "+faststart", + "-fflags", + "+bitexact", + "-flags:v", + "+bitexact", + "-y", + str(path), + ], # fmt: skip + path, + ) return GeneratedVideo( path=path, width=width, @@ -250,3 +272,72 @@ def write_video( fps=fps, duration_seconds=duration_seconds, ) + + +def write_corrupt_video( + path: Path, + *, + size: tuple[int, int] = DEFAULT_VIDEO_SIZE, + fps: int = 10, + duration_seconds: float = 2.0, +) -> GeneratedVideo: + """A clip that opens, decodes for a while, and then runs out — the `CorruptMedia` case. + + The trick is the faststart index `write_video` already asks for. With the index at the + front, truncating the tail leaves a file ffprobe describes perfectly well and ffmpeg fails + *partway through*, which is what separates "this file is broken" from "this is not a file we + can read". Truncate a clip whose index sits at the end instead and you get the latter: an + unopenable container, which is `write_unsupported_file`'s job and not this one's. + + The returned `GeneratedVideo` describes what was asked for, not what survived — the whole + point is that the survivor is shorter. + """ + clip = write_video(path, size=size, fps=fps, duration_seconds=duration_seconds) + intact = path.read_bytes() + path.write_bytes(intact[: len(intact) // 2]) + return clip + + +def write_rotated_video( + path: Path, + *, + size: tuple[int, int] = DEFAULT_VIDEO_SIZE, + fps: int = 10, + duration_seconds: float = 2.0, + rotation: int = 90, +) -> GeneratedVideo: + """A clip carrying a display matrix — what a phone writes when it is held upright. + + `-display_rotation` on the *input* plus a stream copy, because the older spelling + (`-metadata:s:v:0 rotate=`) is deprecated and recent ffmpeg drops it silently, which would + give a fixture that generates without error and tests nothing. + + The returned `GeneratedVideo` carries the **stored** size, deliberately: it is what the file + holds, and the test's whole subject is that a probe reports something else. + """ + source = path.with_name(f"upright-{path.name}") + clip = write_video(source, size=size, fps=fps, duration_seconds=duration_seconds) + _run_ffmpeg( + [ + "ffmpeg", + "-nostdin", + "-loglevel", + "error", + "-display_rotation", + str(rotation), + "-i", + str(source), + "-c", + "copy", + "-y", + str(path), + ], # fmt: skip + path, + ) + return GeneratedVideo( + path=path, + width=clip.width, + height=clip.height, + fps=fps, + duration_seconds=duration_seconds, + ) diff --git a/tests/fixtures/test_media.py b/tests/fixtures/test_media.py index db27b719..c5dc1bed 100644 --- a/tests/fixtures/test_media.py +++ b/tests/fixtures/test_media.py @@ -2,8 +2,9 @@ What is pinned here is the part later tasks will *rely on*: that equal arguments give equal bytes (#20's dedup and idempotency), that the EXIF fixture really is rotated (#16), that a -corrupt file really fails to decode (#16), and that a clip carries the frame count it claims -(#17). +corrupt file really fails to decode (#16), that a clip carries the frame count it claims, and +that the two damaged/rotated video generators really produce what their names say (#17) — both +of those lean on ffmpeg behaviour that is easy to get subtly, silently wrong. """ import hashlib @@ -15,15 +16,18 @@ from tests.fixtures import media from tests.fixtures.media import ( DEFAULT_IMAGE_SIZE, + DEFAULT_VIDEO_SIZE, FFMPEG_REQUIRED_ENV, GeneratedVideo, require_ffmpeg, write_corrupt_image, + write_corrupt_video, write_exif_rotated_image, write_image, write_image_in_unsupported_format, write_images, write_multi_picture_jpeg, + write_rotated_video, write_unsupported_file, write_video, ) @@ -167,6 +171,31 @@ def test_two_runs_of_the_same_clip_are_byte_identical(tmp_path: Path) -> None: assert _digest(first.path) == _digest(second.path) +def test_a_corrupt_clip_is_still_readable_enough_to_describe(tmp_path: Path) -> None: + """The whole trick of the fixture: the faststart index survives, so ffprobe still answers. + + A clip whose index went with its tail is unopenable, which is a different refusal (#17 maps + it to `UnsupportedMedia`) and `write_unsupported_file`'s job. This one has to break *during* + a decode, not before one. + """ + require_ffmpeg() + broken = write_corrupt_video(tmp_path / "broken.mp4") + intact = write_video(tmp_path / "intact.mp4") + + assert broken.path.stat().st_size < intact.path.stat().st_size + assert _probe(broken.path, "stream=codec_name") == ["h264"] + + +def test_a_rotated_clip_carries_a_display_matrix(tmp_path: Path) -> None: + """Guards the trap that made this fixture worth a helper: `-metadata:s:v rotate=` is dropped + silently by recent ffmpeg, which would generate a file that tests nothing and fails nowhere.""" + require_ffmpeg() + rotated = write_rotated_video(tmp_path / "portrait.mp4") + + assert _probe(rotated.path, "stream_side_data=rotation") == ["90"] + assert (rotated.width, rotated.height) == DEFAULT_VIDEO_SIZE + + def test_the_generated_video_record_is_immutable() -> None: """No ffmpeg needed: this is about the record, not the file it describes.""" video = GeneratedVideo(path=Path("clip.mp4"), width=64, height=48, fps=10, duration_seconds=1.5) diff --git a/tests/kernel/test_video_processor.py b/tests/kernel/test_video_processor.py new file mode 100644 index 00000000..5e1bb706 --- /dev/null +++ b/tests/kernel/test_video_processor.py @@ -0,0 +1,571 @@ +"""The video processor: what it reports, what it refuses, and what it promises. + +Four properties are pinned here, and they are worth keeping apart. + +The first is **the frame count**. Extraction at 1, 5 and 10 fps from a 10 fps two-second clip +must give exactly 2, 10 and 20 frames. That is the acceptance criterion, and it is also the +canary for the extraction filter: almost any change to the command moves one of those numbers. + +The second is **rotation**, and it is the same policy the image processor applies to EXIF, on a +different mechanism. A clip carrying a quarter turn probes with its edges swapped, and the frames +that come out have those same dimensions — a probe and a decode that disagreed would be a dataset +whose labels are ninety degrees off. + +The third is **determinism**, with the caveat `tests/fixtures/media.py` already records for +ffmpeg: two runs on one machine agree byte for byte, two ffmpeg builds need not. So every +assertion here is about repeatability and none is about a literal hash. + +The fourth is **the refusals**, and their split is by remedy rather than by symptom. A file +ffmpeg never opens is `UnsupportedMedia`; a clip that decodes for a while and then runs out is +`CorruptMedia` — and it yields the frames it managed before it says so. `MediaToolUnavailable` +is in neither family and is not a `MediaError` at all, because no file is at fault. + +Two practical notes for anyone adding a case. There is **no module-level skip**: the composition +tests at the bottom need no ffmpeg and must run on a laptop without one, so the requirement +arrives through the `clip` fixture instead — `write_video` calls `require_ffmpeg` itself, which +skips locally and raises under `VISIONSET_REQUIRE_FFMPEG=1` on CI. And the composition tests live +here rather than in `test_workspace_service.py`, following the event bus and the image processor: +the port's own file is where "can this be injected?" is asked. +""" + +from __future__ import annotations + +import hashlib +import io +import shutil +import subprocess +from collections.abc import Iterator +from itertools import islice +from pathlib import Path + +import pytest +from tests.fixtures.media import ( + DEFAULT_VIDEO_SIZE, + GeneratedVideo, + write_corrupt_video, + write_rotated_video, + write_unsupported_file, + write_video, +) + +from visionset.kernel.adapters import ( + FfmpegVideoProcessor, + PillowImageProcessor, + ffmpeg_video_processor, +) +from visionset.kernel.adapters.ffmpeg_video_processor import ( + _EXTRACTION_ARGS, + _oriented, + _rational, + _rotation, +) +from visionset.kernel.domain import ImageFormat, VideoFrame, VideoMetadata +from visionset.kernel.errors import ( + CorruptMedia, + MediaError, + MediaToolUnavailable, + UnsupportedMedia, + VisionSetError, +) +from visionset.kernel.ports import DEFAULT_EXTRACTION_FPS, FRAME_FORMAT, VideoProcessor +from visionset.kernel.services import WorkspaceService + + +@pytest.fixture +def clip(tmp_path: Path) -> GeneratedVideo: + """A 64x48 clip at 10 fps for 2 s: 20 frames, so 1/5/10 fps divide it exactly. + + Also the ffmpeg gate for this module. `write_video` calls `require_ffmpeg`, so a test that + asks for a clip skips without the binary and fails loudly with `VISIONSET_REQUIRE_FFMPEG=1`. + """ + return write_video(tmp_path / "clip.mp4") + + +def _frames(video: Path, **kwargs: float) -> list[VideoFrame]: + return list(FfmpegVideoProcessor().frames(video, **kwargs)) + + +def _dimensions(content: bytes) -> tuple[int, int]: + metadata = PillowImageProcessor().probe(io.BytesIO(content)) + return metadata.width, metadata.height + + +# --- what a clip reports ------------------------------------------------------ + + +def test_a_clip_reports_its_dimensions_rate_duration_and_codec(clip: GeneratedVideo) -> None: + metadata = FfmpegVideoProcessor().probe(clip.path) + + assert (metadata.width, metadata.height) == (clip.width, clip.height) + assert metadata.fps == pytest.approx(clip.fps) + assert metadata.duration_seconds == pytest.approx(clip.duration_seconds) + assert metadata.codec == "h264" + + +def test_a_probe_reads_the_container_and_decodes_nothing(clip: GeneratedVideo) -> None: + """The reason there are two methods: registering a source must not cost a full decode.""" + assert FfmpegVideoProcessor().probe(clip.path) == FfmpegVideoProcessor().probe(clip.path) + + +def test_a_clip_at_another_size_and_rate_reports_those(tmp_path: Path) -> None: + other = write_video(tmp_path / "short.mp4", size=(32, 32), fps=5, duration_seconds=1.0) + + metadata = FfmpegVideoProcessor().probe(other.path) + + assert (metadata.width, metadata.height) == (32, 32) + assert metadata.fps == pytest.approx(5.0) + + +# --- rotation ----------------------------------------------------------------- + + +def test_a_rotated_clip_reports_swapped_edges(tmp_path: Path) -> None: + """The video spelling of the EXIF rule: a 64x48 file turned a quarter is a 48x64 picture.""" + rotated = write_rotated_video(tmp_path / "portrait.mp4") + + metadata = FfmpegVideoProcessor().probe(rotated.path) + + assert (rotated.width, rotated.height) == DEFAULT_VIDEO_SIZE + assert (metadata.width, metadata.height) == (rotated.height, rotated.width) + + +def test_the_frames_of_a_rotated_clip_match_what_was_probed(tmp_path: Path) -> None: + """A probe and a decode that disagreed is a dataset whose labels are ninety degrees off.""" + rotated = write_rotated_video(tmp_path / "portrait.mp4") + processor = FfmpegVideoProcessor() + + metadata = processor.probe(rotated.path) + frames = list(processor.frames(rotated.path, fps=5)) + + assert _dimensions(frames[0].content) == (metadata.width, metadata.height) + + +def test_a_half_turn_leaves_the_edges_alone(tmp_path: Path) -> None: + upside_down = write_rotated_video(tmp_path / "flipped.mp4", rotation=180) + + metadata = FfmpegVideoProcessor().probe(upside_down.path) + + assert (metadata.width, metadata.height) == DEFAULT_VIDEO_SIZE + + +@pytest.mark.parametrize( + ("rotation", "expected"), + [ + (0, (64, 48)), + (90, (48, 64)), + (180, (64, 48)), + (270, (48, 64)), + (-90, (48, 64)), + (-180, (64, 48)), + (360, (64, 48)), + (450, (48, 64)), + ], + ids=lambda value: str(value), +) +def test_only_a_quarter_turn_swaps_the_edges(rotation: int, expected: tuple[int, int]) -> None: + """Swept directly, because this one line is where a dataset silently goes sideways. + + ffprobe reports the same turn as 90, as -90 and as 270 depending on the file and the build, + so the rule is arithmetic on the angle and never a membership test against three literals. + """ + assert _oriented(64, 48, rotation) == expected + + +def test_a_clip_with_no_rotation_at_all_is_upright(clip: GeneratedVideo) -> None: + """The ordinary path rather than a fallback: most video carries no display matrix.""" + assert _rotation({}) == 0 + assert _rotation({"tags": {"language": "und"}}) == 0 + + +def test_the_older_rotate_tag_is_still_read() -> None: + """Recent ffmpeg no longer writes it; files written by older ones are still out there.""" + assert _rotation({"tags": {"rotate": "270"}}) == 270 + assert _rotation({"side_data_list": [{"side_data_type": "Display Matrix", "rotation": -90}]}) + + +# --- frame counts ------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("fps", "expected"), [(1, 2), (5, 10), (10, 20)], ids=lambda value: str(value) +) +def test_extraction_at_a_target_rate_gives_the_expected_frame_count( + clip: GeneratedVideo, fps: int, expected: int +) -> None: + """The acceptance criterion, and the canary for the whole extraction command.""" + assert len(_frames(clip.path, fps=fps)) == expected + + +def test_a_rate_below_one_frame_a_second_is_allowed(clip: GeneratedVideo) -> None: + """Long clips are the case this exists for: one frame every two seconds, not every tenth.""" + assert len(_frames(clip.path, fps=0.5)) == 1 + + +def test_the_default_rate_is_the_one_the_port_declares(clip: GeneratedVideo) -> None: + assert DEFAULT_EXTRACTION_FPS == 1.0 + assert _frames(clip.path) == _frames(clip.path, fps=DEFAULT_EXTRACTION_FPS) + + +def test_asking_for_more_frames_than_the_clip_has_duplicates_them(clip: GeneratedVideo) -> None: + """Documented rather than clamped — and content addressing collapses the duplicates.""" + frames = _frames(clip.path, fps=20) + + assert len(frames) == 40 + assert len({hashlib.sha256(frame.content).digest() for frame in frames}) == 20 + + +# --- frame origin ------------------------------------------------------------- + + +def test_every_frame_carries_its_position_in_the_extracted_sequence( + clip: GeneratedVideo, +) -> None: + """The acceptance criterion: origin metadata attached to each frame, not inferred later.""" + frames = _frames(clip.path, fps=5) + + assert [frame.index for frame in frames] == list(range(10)) + + +def test_a_frames_timestamp_is_its_place_on_the_requested_grid(clip: GeneratedVideo) -> None: + frames = _frames(clip.path, fps=5) + + assert [frame.timestamp for frame in frames] == pytest.approx( + [index / 5 for index in range(10)] + ) + + +def test_the_same_moment_is_named_the_same_at_any_extraction_rate(clip: GeneratedVideo) -> None: + """What makes a timestamp the locator and the index merely the ordering: `round=up`. + + Under the filter's default rounding the frame taken for a given grid point drifts with the + rate, so one second into the clip would be a different picture at 1 fps than at 5. + """ + slow = _frames(clip.path, fps=1) + quick = _frames(clip.path, fps=5) + + assert slow[1].timestamp == quick[5].timestamp == 1.0 + assert slow[1].content == quick[5].content + + +def test_every_frame_is_an_image_the_image_processor_accepts(clip: GeneratedVideo) -> None: + """The point of `FRAME_FORMAT` being an `ImageFormat` member, asserted end to end.""" + frame = _frames(clip.path, fps=1)[0] + + metadata = PillowImageProcessor().probe(io.BytesIO(frame.content)) + + assert metadata.format is FRAME_FORMAT is ImageFormat.PNG + assert (metadata.width, metadata.height) == (clip.width, clip.height) + + +def test_consecutive_frames_of_a_moving_clip_differ(clip: GeneratedVideo) -> None: + """Guards the failure where a pinned command quietly starts emitting one frame repeatedly.""" + frames = _frames(clip.path, fps=10) + + assert len({hashlib.sha256(frame.content).digest() for frame in frames}) == len(frames) + + +# --- determinism -------------------------------------------------------------- + + +def test_two_extractions_of_one_clip_are_byte_identical(clip: GeneratedVideo) -> None: + """The acceptance criterion. Repeatability, never a hardcoded hash — see the docstring.""" + assert _frames(clip.path, fps=5) == _frames(clip.path, fps=5) + + +def test_two_identical_clips_extract_to_the_same_hashes(tmp_path: Path) -> None: + """The property a re-ingest dedups on: same bytes in, same asset identities out.""" + first = write_video(tmp_path / "a.mp4") + second = write_video(tmp_path / "b.mp4") + + def digests(video: Path) -> list[bytes]: + return [hashlib.sha256(frame.content).digest() for frame in _frames(video, fps=5)] + + assert digests(first.path) == digests(second.path) + + +def test_two_processors_agree_on_one_clip(clip: GeneratedVideo) -> None: + """No instance state, so a workspace-scoped decoder is uniformity and not isolation.""" + assert list(FfmpegVideoProcessor().frames(clip.path, fps=5)) == list( + FfmpegVideoProcessor().frames(clip.path, fps=5) + ) + + +# --- streaming ---------------------------------------------------------------- + + +def test_frames_arrive_lazily(clip: GeneratedVideo) -> None: + """Nothing buffers a clip: a caller that wants three frames does not decode twenty.""" + stream = FfmpegVideoProcessor().frames(clip.path, fps=10) + + taken = list(islice(stream, 3)) + + assert [frame.index for frame in taken] == [0, 1, 2] + stream.close() # type: ignore[attr-defined] + + +def test_an_abandoned_iterator_leaves_no_decoder_running( + clip: GeneratedVideo, monkeypatch: pytest.MonkeyPatch +) -> None: + """A caller that breaks out of the loop must not leak an ffmpeg reading a file nobody wants. + + Asserted on the process object rather than by counting what is running, which would be a + guess on a machine doing anything else at the time. + """ + started: list[subprocess.Popen[bytes]] = [] + spawn = subprocess.Popen + + def record(*args: object, **kwargs: object) -> subprocess.Popen[bytes]: + process = spawn(*args, **kwargs) # type: ignore[arg-type, misc] + started.append(process) + return process + + monkeypatch.setattr(ffmpeg_video_processor.subprocess, "Popen", record) + + stream = FfmpegVideoProcessor().frames(clip.path, fps=10) + next(iter(stream)) + stream.close() # type: ignore[attr-defined] + + assert len(started) == 1 + assert started[0].poll() is not None + + +# --- refusals ----------------------------------------------------------------- + + +def test_a_file_that_is_not_a_video_is_refused_by_the_probe(tmp_path: Path) -> None: + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(UnsupportedMedia, match="not a video ffmpeg can open"): + FfmpegVideoProcessor().probe(path) + + +def test_a_file_that_is_not_a_video_is_refused_by_the_extraction_too(tmp_path: Path) -> None: + """The two methods cannot disagree about what is readable.""" + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(UnsupportedMedia): + _frames(path, fps=1) + + +def test_a_truncated_clip_yields_what_decoded_and_then_refuses(tmp_path: Path) -> None: + """The remedy split, end to end: it opened, so this is damage and not an unsupported file.""" + broken = write_corrupt_video(tmp_path / "broken.mp4") + stream = FfmpegVideoProcessor().frames(broken.path, fps=10) + + decoded = [] + with pytest.raises(CorruptMedia, match="damaged or truncated"): + for frame in stream: + decoded.append(frame) + + assert 0 < len(decoded) < broken.frame_count + + +def test_a_truncated_clip_still_probes(tmp_path: Path) -> None: + """Deliberate: the index survived, so the container can still say what it holds.""" + broken = write_corrupt_video(tmp_path / "broken.mp4") + + assert FfmpegVideoProcessor().probe(broken.path).codec == "h264" + + +def test_a_refusal_names_the_file_it_was_given(tmp_path: Path) -> None: + """A per-file error has to say which of five thousand files.""" + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(UnsupportedMedia) as caught: + FfmpegVideoProcessor().probe(path) + + assert caught.value.name == str(path) + + +def test_an_explicit_name_wins_over_the_path(tmp_path: Path) -> None: + path = write_unsupported_file(tmp_path / "blob-ab12cd") + + with pytest.raises(UnsupportedMedia) as caught: + FfmpegVideoProcessor().probe(path, name="dashcam/2026-07-27.mp4") + + assert caught.value.name == "dashcam/2026-07-27.mp4" + + +def test_a_refusal_reason_does_not_repeat_the_name(tmp_path: Path) -> None: + """ffmpeg quotes the path in nearly everything it says; a report is a table, not sentences.""" + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(UnsupportedMedia) as caught: + FfmpegVideoProcessor().probe(path) + + assert str(path) not in caught.value.reason + assert path.name not in caught.value.reason + assert str(path) in str(caught.value) + + +def test_a_refusal_quotes_what_the_decoder_actually_said(tmp_path: Path) -> None: + """Redacted, not discarded: without the cause an operator has nothing to act on.""" + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(UnsupportedMedia) as caught: + FfmpegVideoProcessor().probe(path) + + assert "Invalid data" in caught.value.reason + + +def test_both_video_refusals_are_the_image_family(tmp_path: Path) -> None: + """Reused rather than duplicated: the split is by remedy, and remedies have no modality.""" + assert issubclass(UnsupportedMedia, MediaError) + assert issubclass(CorruptMedia, MediaError) + + +@pytest.mark.parametrize("method", ["probe", "frames"], ids=str) +def test_a_missing_file_is_not_a_media_error(tmp_path: Path, method: str) -> None: + """Nothing is wrong with the media: there is no media.""" + processor = FfmpegVideoProcessor() + + with pytest.raises(FileNotFoundError, match="no video file at"): + getattr(processor, method)(tmp_path / "absent.mp4") + + +@pytest.mark.parametrize("fps", [0, -1, -0.5], ids=lambda value: str(value)) +def test_a_rate_that_is_not_positive_is_a_programming_error( + clip: GeneratedVideo, fps: float +) -> None: + """A `ValueError`, deliberately outside the `MediaError` family an ingest catches.""" + with pytest.raises(ValueError, match="fps must be greater than zero"): + _frames(clip.path, fps=fps) + + +def test_no_decoder_exception_ever_escapes(tmp_path: Path) -> None: + """The sibling of the image processor's: ffmpeg's failures stop at the adapter.""" + path = write_unsupported_file(tmp_path / "notes.txt") + + with pytest.raises(VisionSetError): + FfmpegVideoProcessor().probe(path) + + +# --- the tool this adapter does not ship -------------------------------------- + + +@pytest.mark.parametrize("method", ["probe", "frames"], ids=str) +def test_a_missing_binary_is_reported_with_an_install_hint( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, method: str +) -> None: + """Needs no ffmpeg to run, which is the point: this is the path a fresh machine takes.""" + monkeypatch.setattr(shutil, "which", lambda _program: None) + + with pytest.raises(MediaToolUnavailable, match="apt-get install ffmpeg") as caught: + getattr(FfmpegVideoProcessor(), method)(tmp_path / "clip.mp4") + + assert "not on PATH" in str(caught.value) + + +def test_a_missing_binary_is_reported_before_the_file_is_looked_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """`frames` is not a generator, so "up front" means at the call and not at the first frame.""" + monkeypatch.setattr(shutil, "which", lambda _program: None) + + with pytest.raises(MediaToolUnavailable): + FfmpegVideoProcessor().frames(tmp_path / "does-not-exist.mp4") + + +def test_a_missing_binary_is_not_a_media_error() -> None: + """No file is at fault, so an ingest must not record it against five thousand of them.""" + assert not issubclass(MediaToolUnavailable, MediaError) + assert issubclass(MediaToolUnavailable, VisionSetError) + + +# --- the port ----------------------------------------------------------------- + + +def test_the_default_video_processor_satisfies_the_port() -> None: + assert isinstance(FfmpegVideoProcessor(), VideoProcessor) + + +def test_the_extraction_arguments_are_pinned() -> None: + """A change detector on purpose: moving a value here moves every frame hash ever stored.""" + assert _EXTRACTION_ARGS == ( + "-an", + "-f", "image2pipe", + "-c:v", "png", + "-pred", "none", + "-compression_level", "6", + "-fflags", "+bitexact", + "-flags:v", "+bitexact", + ) # fmt: skip + + +@pytest.mark.parametrize( + ("value", "expected"), + [("10/1", 10.0), ("30000/1001", 29.97002997002997), ("0/0", None), ("", None), (25, 25.0)], + ids=["cfr", "ntsc", "unknown", "blank", "number"], +) +def test_ffprobes_rational_rates_are_read_as_numbers(value: object, expected: float | None) -> None: + """`0/0` is how ffprobe says it does not know, and must not surface as a division error.""" + assert _rational(value) == expected + + +# --- composition -------------------------------------------------------------- + + +class _NullVideoProcessor: + """A stand-in that decodes nothing — enough to prove the seam, and nothing more.""" + + def probe(self, source: Path, *, name: str | None = None) -> VideoMetadata: + return VideoMetadata(width=1, height=1, fps=1.0, duration_seconds=1.0, codec="none") + + def frames( + self, + source: Path, + *, + fps: float = DEFAULT_EXTRACTION_FPS, + name: str | None = None, + ) -> Iterator[VideoFrame]: + return iter(()) + + +def test_a_workspace_exposes_a_video_processor_by_default(tmp_path: Path) -> None: + with WorkspaceService.init(tmp_path / "ws") as workspace: + assert isinstance(workspace.video_processor, VideoProcessor) + assert isinstance(workspace.video_processor, FfmpegVideoProcessor) + + +def test_a_workspace_opens_on_a_machine_with_no_ffmpeg( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The reason the binary check is not in the constructor: images must still work.""" + monkeypatch.setattr(shutil, "which", lambda _program: None) + + with WorkspaceService.init(tmp_path / "ws") as workspace: + assert isinstance(workspace.video_processor, FfmpegVideoProcessor) + + +def test_each_open_workspace_gets_its_own_processor(tmp_path: Path) -> None: + with ( + WorkspaceService.init(tmp_path / "one") as one, + WorkspaceService.init(tmp_path / "two") as two, + ): + assert one.video_processor is not two.video_processor + + +def test_a_video_processor_can_be_injected_at_init(tmp_path: Path) -> None: + with WorkspaceService.init( + tmp_path / "ws", video_processor_factory=_NullVideoProcessor + ) as workspace: + assert isinstance(workspace.video_processor, _NullVideoProcessor) + + +def test_a_video_processor_can_be_injected_at_open(tmp_path: Path) -> None: + """The seam an embedder actually reaches for: a workspace it did not create.""" + WorkspaceService.init(tmp_path / "ws").close() + + with WorkspaceService.open( + tmp_path / "ws", video_processor_factory=_NullVideoProcessor + ) as workspace: + assert isinstance(workspace.video_processor, _NullVideoProcessor) + + +def test_injecting_a_video_processor_leaves_the_other_ports_alone(tmp_path: Path) -> None: + """The parameter is appended last, so binding it must not shift anything before it.""" + with WorkspaceService.init( + tmp_path / "ws", video_processor_factory=_NullVideoProcessor + ) as workspace: + assert isinstance(workspace.image_processor, PillowImageProcessor) + assert workspace.root == (tmp_path / "ws").resolve()