From 2f91f78f5ed82546e0d684c8cec63eb9babc3e7a Mon Sep 17 00:00:00 2001 From: Armando Anaya Date: Mon, 27 Jul 2026 07:44:42 -0700 Subject: [PATCH] =?UTF-8?q?feat(examples):=20ingest=20end-to-end=20?= =?UTF-8?q?=E2=80=94=20a=20synthetic=20clip=20to=20an=20approved,=20partit?= =?UTF-8?q?ioned=20batch=20(#23)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M2's exit criterion made executable, the way examples/sdk_end_to_end.py served M1's. A ten-second testsrc clip and a folder of stills go in; fifty hash-deduplicated assets in an approved batch of two jobs come out, with a pollable progress row and a per-file report of what could not be read. Nothing is annotated and nothing is released — the SDK example already covers that half of the cycle, and this one spends its length on where assets come from. Four properties demonstrated rather than described: - A clip cannot state its total and a directory can. processed climbs to 50 while total stays NULL, because VideoMetadata carries no frame count by design; the image directory states 4 before reading its first file. - One file registered at two rates is two sources whose frames are one set. Decomposition parameters live on the source, so 5 fps and 1 fps over the same path are two origins — and the coarse run still creates nothing, because identity is content and round=up lands both grids on whole seconds. That alignment is a property of this extractor, not a promise the port makes. - A file that is not an image is reported, not skipped. notes.txt yields one IngestFailure and the run still ends completed. - Re-ingesting a source creates nothing: created=0, deduplicated=50, into a new batch because the first froze at approval. The clip is 160x120 rather than the fixtures' 64x48, and that is load-bearing. testsrc moves a little between frames; below roughly 96x72 that movement falls under what the scaler and encoder still resolve, consecutive frames come out byte-identical, and content addressing collapses them — a ten-second clip at 5 fps then yields forty assets, the feature working and reading as a shortfall. Two deliberate divergences from the SDK example, both stated in the module docstring: the clip is generated by shelling out to ffmpeg, because a video is a container wrapped around a codec and the only honest way to write one is the tool that reads it; and the stills are Pillow's work rather than a second copy of M1's hand-rolled PNG encoder, Pillow having been a dependency since #16. The generation command is duplicated from tests/fixtures/media.py rather than imported — that module is a test fixture, it imports pytest, and its answer to a missing binary is pytest.skip, which means nothing in a script. The example checks shutil.which("ffmpeg") before writing anything and exits with an install hint instead. CI runs it twice, as the SDK example is: the smoke test calls main() and asserts on the returned Summary, and a new "Ingest end-to-end example" step runs the file as a plain script, which is the only thing that proves it works from a clean checkout. The smoke test gates on tests/fixtures/media.require_ffmpeg() — a skip locally, an error under VISIONSET_REQUIRE_FFMPEG=1. No new service, domain model, error, event or migration. FORMAT_VERSION stays 10, VERSION stays 0.0.1.dev0, openapi.json unchanged. 885 tests, up from 877. --- .github/workflows/ci.yml | 5 + README.md | 5 + docs/README.md | 2 +- docs/examples.md | 111 ++++++- docs/ingest.md | 3 + examples/README.md | 42 ++- examples/ingest_end_to_end.py | 399 +++++++++++++++++++++++ tests/examples/test_ingest_end_to_end.py | 140 ++++++++ 8 files changed, 690 insertions(+), 17 deletions(-) create mode 100644 examples/ingest_end_to_end.py create mode 100644 tests/examples/test_ingest_end_to_end.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 342af94a..79d809d1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -50,6 +50,11 @@ jobs: - name: SDK end-to-end example run: uv run python examples/sdk_end_to_end.py + # The ingest half of the same argument, and the one that needs the ffmpeg + # installed above: this example generates its own clip. + - name: Ingest end-to-end example + run: uv run python examples/ingest_end_to_end.py + frontend: runs-on: ubuntu-latest steps: diff --git a/README.md b/README.md index 6fbdebd7..3c2400f6 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,11 @@ empty directory to a hash-verified release in one pass, generating its own image no CLI, nothing to download. Run it with `uv run python examples/sdk_end_to_end.py`; the walkthrough is in [docs/examples.md](docs/examples.md). +For where the assets themselves come from, +[`examples/ingest_end_to_end.py`](examples/ingest_end_to_end.py) turns a generated ten-second clip +into 50 deduplicated assets in an approved batch, then shows a re-run creating nothing. It needs +ffmpeg. + ## Monorepo map ``` diff --git a/docs/README.md b/docs/README.md index 7d53df51..96864973 100644 --- a/docs/README.md +++ b/docs/README.md @@ -19,4 +19,4 @@ contracts (kernel purity, headless annotator) are described there and enforced i | [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 | | [persistence.md](persistence.md) | The metadata store: repositories, unit of work, table layout, migrations and `format_version` | -| [examples.md](examples.md) | The runnable end-to-end example: the whole cycle in one pass, and what it is built to demonstrate | +| [examples.md](examples.md) | The two runnable examples: the whole cycle in one pass, ingest on its own, and what each is built to demonstrate | diff --git a/docs/examples.md b/docs/examples.md index 2e5082f9..be8a5961 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -1,17 +1,31 @@ # Examples -Every other document here explains one thing the kernel does. This one is about the single -runnable file that does all of them at once: [`examples/sdk_end_to_end.py`](../examples/sdk_end_to_end.py) -takes an empty directory and leaves behind a release whose every byte can be re-hashed and -checked — with nothing but `import visionset`. +Every other document here explains one thing the kernel does. This one is about the runnable +files that do several of them at once. There are two, one per milestone: + +| Example | What it drives | Milestone | +| --- | --- | --- | +| [`sdk_end_to_end.py`](../examples/sdk_end_to_end.py) | an empty directory to a release whose every byte can be re-hashed and checked | M1 | +| [`ingest_end_to_end.py`](../examples/ingest_end_to_end.py) | a ten-second clip and a folder of stills to an approved, partitioned batch | M2 | ```bash uv run python examples/sdk_end_to_end.py +uv run python examples/ingest_end_to_end.py # needs ffmpeg ``` -It is the milestone's exit criterion made executable, and it runs in CI twice: once as a -[pytest smoke test](../tests/examples/test_sdk_end_to_end.py) that asserts on outcomes, and once -as a plain script, which is the only way to prove it still works from a clean checkout. +Each is its milestone's exit criterion made executable, and each runs in CI **twice**: once as a +pytest smoke test ([M1](../tests/examples/test_sdk_end_to_end.py), +[M2](../tests/examples/test_ingest_end_to_end.py)) that asserts on outcomes, and once as a plain +script, which is the only way to prove it still works from a clean checkout. + +The two overlap by design and neither subsumes the other. The SDK example walks the whole cycle +and treats ingest as one stage of thirteen; the ingest example stops at an approved batch and +spends its length on where assets come from — two sources over one file, dedup, progress and the +per-file report. + +--- + +# The SDK example ## The cycle @@ -70,7 +84,8 @@ the ingest is what makes the batch now. The frames are still generated by the example's own six-line PNG encoder rather than by Pillow, which is a dependency these days. Their bytes are fixed, and that is the point: an asset's identity is the SHA-256 of its content and the split keys on content hash, so the same pictures -land in the same folds on every machine. +land in the same folds on every machine. (The [ingest example](#the-ingest-example) does reach +for Pillow — it is not carrying a determinism argument about folds.) ## Why three classes for "two classes" @@ -83,7 +98,79 @@ attribute is *required* (`occlusion` on `stop-sign`), which is what makes ## No committed media, ever The frames are built by a short PNG encoder using only `zlib` and `struct`: a signature, an IHDR -chunk, zlib-compressed filter-0 scanlines in IDAT, and IEND. M1 has no image library to lean on -— Pillow arrives with the media processor (#16) — and v1 of this product shipped 929 MB of -images into git history, which is why `**/workspace-data/` is ignored and why an example that -needs pictures makes its own. +chunk, zlib-compressed filter-0 scanlines in IDAT, and IEND. It was written when M1 had no image +library to lean on, and it stays for the reason in the section above rather than out of nostalgia. +v1 of this product shipped 929 MB of images into git history, which is why `**/workspace-data/` +is ignored and why an example that needs pictures makes its own. + +--- + +# The ingest example + +Where [`sdk_end_to_end.py`](../examples/sdk_end_to_end.py) treats ingest as one stage, +[`ingest_end_to_end.py`](../examples/ingest_end_to_end.py) is about nothing else. It generates a +ten-second clip, registers it twice at two different rates, ingests a folder of stills with one +file that is not an image, and stops at an approved batch cut into two jobs. Nothing is annotated +and nothing is released. + +## What it does + +| Stage | What happens | Owned by | +| --- | --- | --- | +| Project | `ProjectService.create` + `SchemaService.create_version` — one class, because approval needs a version to pin | [projects.md](projects.md), [schemas.md](schemas.md) | +| Clip | ten seconds of `testsrc` at 10 fps, written by ffmpeg | — | +| Source | `register_video(..., extraction_fps=5.0)` — the rate is part of *what the source is* | [sources.md](sources.md) | +| Assets | `IngestService.ingest` → **50 assets** in a draft batch, each with a frame index and timestamp | [ingest.md](ingest.md) | +| Progress | `IngestService.get(job_id)` → `processed=50`, `total=None` | [ingest.md](ingest.md) | +| Batch | `approve(BySize(size=25))` → 2 jobs of 25, schema pinned | [batches.md](batches.md) | +| Re-run | the same source again → `created=0`, `deduplicated=50` | [ingest.md](ingest.md) | +| Second rate | `register_video(..., extraction_fps=1.0)` → a *different* source, 10 frames, none new | [sources.md](sources.md) | +| Stills | three PNGs and a `notes.txt` → `total=4`, `created=3`, one `IngestFailure` | [ingest.md](ingest.md) | +| Previews | every asset carries a `thumbnail_hash` | [media.md](media.md) | + +## Four things it is built to demonstrate + +**A clip cannot state its total, and a directory can.** `IngestJob.processed` climbs to 50 while +`total` stays NULL, because `VideoMetadata` deliberately carries no frame count — it would be a +guess for a variable-rate clip, and the number an ingest actually wants is what extraction +produced. The image directory *can* be listed, so it states `4` before reading the first file. +Both numbers are written to the row as the run goes, which is what makes them pollable from +another process rather than a return value dressed up as progress. + +**One file registered at two rates is two sources whose frames are one set.** Decomposition +parameters live on the source, so `extraction_fps=5.0` and `extraction_fps=1.0` over the same +path are two origins — "the same source yields the same assets" only means something if the +parameters deciding those assets are part of what the source *is*. And yet the coarse run creates +nothing: identity is content, and the ten frames it cuts are byte-for-byte frames the finer run +already stored. Their recorded origin stays the first sighting's, because origin is provenance +and provenance is never rewritten. + +That alignment is a property of *this* extractor and not a promise the port makes. The fps filter +rounds **up** onto the grid, so both rates land on whole seconds; under the default rounding a +1 fps pass would take the picture from 0.4 s and label it 0.0. + +**A file that is not an image is reported, not skipped.** `notes.txt` produces one +`IngestFailure` — `name`, `kind`, `reason`, where the reason never repeats the name so a surface +can group by kind instead of reading prose — and the run still ends `completed`. Guessing which +files an operator meant to offer is a policy the kernel would be inventing. Failure splits by +remedy, which is also why a missing ffmpeg would fail the whole job instead: one broken machine +is not five thousand broken files. + +**The clip is 160×120, and that is load-bearing.** `testsrc` moves a little between frames; below +roughly 96×72 that movement falls under what the scaler and encoder still resolve, and +consecutive frames come out byte-identical. Content addressing then does exactly what it promises +and collapses them — a ten-second clip at 5 fps yields *forty* assets, the feature working and +reading as a shortfall. The example says so in a comment where the constant is declared. + +## Why this one needs ffmpeg + +The SDK example boasts of needing nothing. This one checks `shutil.which("ffmpeg")` before it +writes anything and exits with an install hint if the binary is absent, because a video is a +container wrapped around a codec and the only honest way to write one is the tool that reads it. +CI installs ffmpeg for exactly this reason, and the smoke test gates on +`tests/fixtures/media.require_ffmpeg()` — a skip locally, an error under `VISIONSET_REQUIRE_FFMPEG=1`. + +The generation command is `tests/fixtures/media.write_video`'s, duplicated rather than imported: +that module is a test fixture, it imports pytest, and its answer to a missing binary is +`pytest.skip`, which means nothing in a script. The stills, by contrast, are Pillow's work — a +real dependency since #16, so a second hand-rolled PNG encoder beside it would be archaeology. diff --git a/docs/ingest.md b/docs/ingest.md index f1c61a6d..897b6cae 100644 --- a/docs/ingest.md +++ b/docs/ingest.md @@ -5,6 +5,9 @@ the bytes once, records what the decoder made of them, and puts the result in a [batch](batches.md) somebody can approve. Nothing else in the kernel creates an `Asset` — `examples/sdk_end_to_end.py` used to, and no longer does. +Everything below is executed by [`examples/ingest_end_to_end.py`](../examples/ingest_end_to_end.py), +which is walked through in [examples.md](examples.md). + ```python source = sources.register_images(project.id, Path("~/dashcam/monday").expanduser()) result = ingest.ingest(source.id, batch_name="monday") diff --git a/examples/README.md b/examples/README.md index 5d6370fc..e94e83e4 100644 --- a/examples/README.md +++ b/examples/README.md @@ -6,8 +6,9 @@ and `**/workspace-data/` is git-ignored by design. | Example | What it shows | | --- | --- | | [`sdk_end_to_end.py`](sdk_end_to_end.py) | The whole cycle in one pass: workspace → project → schema → synthetic frames → batch → jobs → annotations → curated trunk → verified release | +| [`ingest_end_to_end.py`](ingest_end_to_end.py) | Where assets come from: a generated 10 s clip → source at 5 fps → 50 deduplicated assets → approved batch of 2 jobs, plus a re-run that creates nothing, the same clip at a second rate, and a folder of stills with one unreadable file. **Needs ffmpeg.** | -## Running the end-to-end example +## Running the SDK end-to-end example ```bash uv run python examples/sdk_end_to_end.py # into examples/workspace-data/sdk-e2e @@ -15,8 +16,8 @@ uv run python examples/sdk_end_to_end.py ./scratch # or wherever you like ``` It needs nothing but the package: no server, no CLI, no file the repository had to ship. The -six 64×48 frames come from a short PNG encoder built out of `zlib` and `struct`, because M1 has -no image library to lean on (Pillow arrives with the media processor in M2, #16). +six 64×48 frames come from a short PNG encoder built out of `zlib` and `struct` — fixed bytes, +so the release's split folds are the same on every machine. With no argument it writes into `examples/workspace-data/sdk-e2e` and clears a previous run first — but only after confirming that directory holds nothing except a workspace. A @@ -34,4 +35,37 @@ with WorkspaceService.open("examples/workspace-data/sdk-e2e") as workspace: print(release.tag, release.asset_count, release.manifest_hash) ``` -[`docs/examples.md`](../docs/examples.md) walks through what each stage does and why. +## Running the ingest example + +```bash +uv run python examples/ingest_end_to_end.py # into examples/workspace-data/ingest-e2e +uv run python examples/ingest_end_to_end.py ./scratch # or wherever you like +``` + +Same destination rules as above, with one extra requirement: **ffmpeg must be on `PATH`**, because +this one generates its own ten-second clip. A video is a container wrapped around a codec, and the +only honest way to write one is the tool that reads it — so the example checks for the binary +before it writes anything and exits with an install hint if it is missing: + +```bash +brew install ffmpeg # macOS +sudo apt-get install ffmpeg # Debian/Ubuntu +``` + +It leaves a project holding 53 assets: 50 frames cut from the clip at 5 fps into an approved batch +of two jobs, and three stills from `incoming/` — where a fourth file is deliberately not an image, +so the run's per-file report has something in it. + +```python +from visionset.kernel.services import IngestService, ProjectService, SourceService, WorkspaceService + +with WorkspaceService.open("examples/workspace-data/ingest-e2e") as workspace: + project = ProjectService(workspace).list()[0] + ingest = IngestService(workspace) + for source in SourceService(workspace).list(project.id): + for job in ingest.list(source.id): + print(source.kind.value, job.state.value, job.processed, job.total, job.failures) +``` + +[`docs/examples.md`](../docs/examples.md) walks through what each stage of both examples does and +why. diff --git a/examples/ingest_end_to_end.py b/examples/ingest_end_to_end.py new file mode 100644 index 00000000..d4f4a2ca --- /dev/null +++ b/examples/ingest_end_to_end.py @@ -0,0 +1,399 @@ +"""Raw media to an approved, partitioned batch — M2's exit criterion in one pass. + +A ten-second clip and a small folder of stills go in; fifty hash-deduplicated +assets in an approved batch of two jobs come out, with a pollable progress row +and a per-file report of what could not be read. Nothing here is annotated and +nothing is released — [`sdk_end_to_end.py`](sdk_end_to_end.py) covers that half +of the cycle. This one is about where assets *come from*. + +Run it:: + + uv run python examples/ingest_end_to_end.py [DESTINATION] + +**This example needs ffmpeg.** Its sibling boasts of needing nothing: it builds +its PNGs out of ``zlib`` and ``struct`` and hands them to the SDK. A video +cannot be made that way — a clip is a container wrapped around a codec, and the +only honest way to write one is the tool that reads it. So the clip below is +generated by the same pinned ffmpeg command ``tests/fixtures/media.py`` uses, +duplicated rather than imported: that module is a *test* fixture, it imports +pytest, and its missing-binary answer is ``pytest.skip``, which means nothing in +a script. + +The stills are Pillow's work rather than a second copy of the sibling's PNG +encoder. That encoder exists because M1 predates any image library; Pillow has +been a dependency since #16, and re-deriving a PNG by hand next to it would be +archaeology. +""" + +from __future__ import annotations + +import shutil +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from uuid import UUID + +from PIL import Image + +from visionset.kernel.domain import ( + Asset, + AssetProgress, + BySize, + DomainEvent, + GeometryType, + IngestJob, + IngestResult, + LabelClass, +) +from visionset.kernel.services import ( + BatchService, + IngestService, + JobService, + ProjectService, + SchemaService, + SourceService, + WorkspaceService, +) + +#: Where the example puts its workspace unless told otherwise. Under +#: ``workspace-data/``, which the repository ignores by design. +DEFAULT_DEST = Path(__file__).resolve().parent / "workspace-data" / "ingest-e2e" + +#: Bigger than the 64x48 the test fixtures use, and deliberately so. ``testsrc`` +#: draws a pattern that moves a little between frames; below roughly 96x72 that +#: movement falls under what the scaler and the encoder can still resolve, and +#: consecutive frames come out **byte-identical**. Content addressing then does +#: exactly what it promises and collapses them, so a ten-second clip at five +#: frames a second yields forty assets rather than fifty — the feature working, +#: reported as a shortfall. 160x120 leaves a wide margin. +CLIP_SIZE = (160, 120) +CLIP_FPS = 10 +CLIP_SECONDS = 10.0 + +#: Ten seconds at five frames a second is fifty assets — the number M2's exit +#: criterion names, and the reason the clip above is exactly ten seconds long. +EXTRACTION_FPS = 5.0 + +#: The same clip, decomposed coarsely. Ten frames, none of them new. +COARSE_FPS = 1.0 + +#: Fifty assets cut into two jobs. The partition is exact — disjoint, and their +#: union is the batch — which ``partition_assets`` guarantees rather than hopes. +JOB_SIZE = 25 + +STILL_COUNT = 3 +STILL_SIZE = (48, 32) + +#: What a folder of photographs really looks like: some of it is not a +#: photograph. This file is reported, not skipped — see stage (7). +STRAY_FILE = "notes.txt" + +FFMPEG_MISSING = ( + "ffmpeg is not on PATH, and this example decomposes a video.\n" + "Install it with `brew install ffmpeg` (macOS) or " + "`sudo apt-get install ffmpeg` (Debian/Ubuntu), then run this again." +) + +#: One class is enough. A batch cannot be approved by a project with no schema — +#: approval pins the active version forever — but this example writes no labels, +#: so the contract only has to exist. +CLASSES: tuple[LabelClass, ...] = ( + LabelClass(name="vehicle", geometry=GeometryType.BBOX, color="#2a9d8f"), +) + + +@dataclass(frozen=True) +class Summary: + """What the run produced, for a reader and for the smoke test alike.""" + + project_id: UUID + schema_version: int + clip_source_id: UUID + clip_batch_id: UUID + asset_ids: tuple[UUID, ...] + clip_progress: IngestJob + job_sizes: tuple[int, ...] + rerun: IngestResult + coarse_source_id: UUID + coarse: IngestResult + images: IngestResult + image_progress: IngestJob + asset_count: int + thumbnailed: int + events: tuple[str, ...] + + +# --- synthetic media ------------------------------------------------------ + + +def write_clip(path: Path) -> Path: + """A ``testsrc`` clip, byte-identical on every run of one ffmpeg build. + + The flags are ``tests/fixtures/media.write_video``'s, and each earns its + place: ``+bitexact`` on both the container and the video stream strips the + encoder version and the creation time that would otherwise make two runs of + this command differ, and ``-g`` pins the keyframe interval so the decode is + not at the mercy of a rate-control decision. + + Determinism holds *within* one ffmpeg build and not across versions. That is + why nothing here — and nothing in the smoke test — asserts a hardcoded hash. + """ + width, height = CLIP_SIZE + path.parent.mkdir(parents=True, exist_ok=True) + command = [ + "ffmpeg", + "-nostdin", + "-loglevel", "error", + "-f", "lavfi", + "-i", f"testsrc=size={width}x{height}:rate={CLIP_FPS}:duration={CLIP_SECONDS}", + "-pix_fmt", "yuv420p", + "-c:v", "libx264", + "-preset", "ultrafast", + "-g", str(CLIP_FPS), + "-movflags", "+faststart", + "-fflags", "+bitexact", + "-flags:v", "+bitexact", + "-y", str(path), + ] # fmt: skip + result = subprocess.run(command, capture_output=True, text=True) + if result.returncode != 0: + raise SystemExit(f"ffmpeg could not generate {path.name}:\n{result.stderr}") + return path + + +def write_stills(directory: Path) -> Path: + """A few photographs and one thing that is not one. + + Each image gets its own pixels, so no pair deduplicates by accident and the + three assets the ingest reports created are genuinely three. + """ + directory.mkdir(parents=True, exist_ok=True) + width, height = STILL_SIZE + for index in range(STILL_COUNT): + pixels = bytes( + channel + for y in range(height) + for x in range(width) + for channel in ((x * 5 + index * 61) % 256, (y * 7) % 256, (x + y + index * 23) % 256) + ) + Image.frombytes("RGB", STILL_SIZE, pixels).save( + directory / f"still-{index:03d}.png", format="PNG" + ) + (directory / STRAY_FILE).write_text("shot on the coast road, tuesday\n") + return directory + + +# --- the cycle ------------------------------------------------------------ + + +def main(dest: Path) -> Summary: + """Drive an empty directory to an approved batch, and report what happened. + + ``dest`` must not already be a workspace: this creates one. Everything the + run produces lives under it. + """ + if shutil.which("ffmpeg") is None: + # Checked before anything is written, so a machine without the binary + # leaves no half-made workspace behind. The kernel raises + # MediaToolUnavailable for the same reason, with the same hint. + raise SystemExit(FFMPEG_MISSING) + + seen: list[DomainEvent] = [] + + with WorkspaceService.init(dest, name="ingest-end-to-end") as workspace: + workspace.event_bus.subscribe(DomainEvent, seen.append) + + projects = ProjectService(workspace) + schemas = SchemaService(workspace) + sources = SourceService(workspace) + ingest = IngestService(workspace) + batches = BatchService(workspace) + jobs = JobService(workspace) + + # (1) A project, its 1:1 dataset, and a labeling contract. Nothing here + # writes a label, but a batch cannot be approved without a schema to pin. + project = projects.create("dashcam", description="Ingest end-to-end demo") + schema = schemas.create_version(project.id, CLASSES) + _say(f"project {project.name!r} ({project.id}) with schema v{schema.version}") + + # (2) Ten seconds of video, registered as an origin. The extraction rate + # is part of *what the source is*, not a per-run flag — "the same source + # yields the same assets" only means something if the parameters that + # decide those assets are recorded with it. + clip = write_clip(dest / "clips" / "road.mp4") + clip_source = sources.register_video(project.id, clip, extraction_fps=EXTRACTION_FPS) + probed = clip_source.require_video().metadata + _say( + f"source registered: {probed.width}x{probed.height} {probed.codec} at " + f"{probed.fps} fps for {probed.duration_seconds}s, " + f"to be decomposed at {EXTRACTION_FPS} fps" + ) + + # (3) Decode, hash, store once, and materialize into a draft batch. The + # decode happens outside every transaction: an out-of-process decoder + # held inside a write transaction is how a single-writer SQLite store + # starts reporting "database is locked". + ingested = ingest.ingest(clip_source.id, batch_name="road-5fps") + batch = batches.get(ingested.batch_id) + first = ingested.assets[0] + _say( + f"{ingested.created} assets in batch {batch.name!r} ({batch.state.value}); " + f"first is {first.uri.rsplit('/', 1)[-1]} at t={first.frame_timestamp}s" + ) + + # (4) Progress is a row, not a return value. It was written once per + # item while the run was in flight, which is what makes it pollable from + # another process — the contract the HTTP API and the UI will reuse. + # 'total' is NULL for a clip: VideoMetadata carries no frame count by + # design, and the number an ingest wants is what extraction produced. + clip_progress = ingest.get(ingested.job_id) + _say( + f"job {clip_progress.state.value}: processed {clip_progress.processed}, " + f"total {clip_progress.total if clip_progress.total is not None else 'unknown'}" + ) + + # (5) Approval freezes membership, pins the schema version forever, and + # cuts the batch into jobs. The partition is exact — disjoint, union is + # the batch — which is a domain guarantee, not a service's arithmetic. + batch = batches.approve(batch.id, BySize(size=JOB_SIZE)) + batch_jobs = batches.jobs(batch.id) + outstanding = jobs.batch_progress(batch.id)[AssetProgress.UNANNOTATED] + _say( + f"approved against schema v{batch.schema_version} into {len(batch_jobs)} jobs " + f"of {'/'.join(str(len(job.progress)) for job in batch_jobs)}; " + f"{outstanding} assets awaiting an annotator" + ) + + # (6) The idempotency property, demonstrated rather than asserted: the + # same source ingested again creates nothing at all. It lands in a new + # batch because the first one froze at approval — a batch is an + # ephemeral unit of work and two may name the same assets. + rerun = ingest.ingest(clip_source.id, batch_name="road-5fps-again") + _say( + f"re-ingested: {rerun.created} created, {rerun.deduplicated} already known " + f"(same assets: {set(rerun.asset_ids) == set(ingested.asset_ids)})" + ) + + # (7) The same clip at another rate is a *different source* — the + # parameters live on the source, so one file registered twice at two + # rates is two origins. Its ten frames are nonetheless already in the + # project: identity is content, and content does not care which source + # produced it. Their recorded origin stays the first sighting's. + coarse_source = sources.register_video(project.id, clip, extraction_fps=COARSE_FPS) + coarse = ingest.ingest(coarse_source.id, batch_name="road-1fps") + _say( + f"same clip at {COARSE_FPS} fps is source {coarse_source.id} " + f"(≠ {clip_source.id}): {len(coarse.assets)} frames, {coarse.created} new" + ) + + # (8) The other path. A directory can be listed, so the job states its + # total before the first file — and the file that is not an image is + # reported rather than skipped, because guessing which files an operator + # meant to offer is a policy the kernel would be inventing. + stills = write_stills(dest / "incoming") + image_source = sources.register_images(project.id, stills) + images = ingest.ingest(image_source.id, batch_name="stills") + image_progress = ingest.get(images.job_id) + # The report names each item and says what is wrong with it separately, + # which is what lets a surface group by kind instead of reading prose. + reported = ", ".join( + f"{Path(failure.name).name} ({failure.kind.value}: {failure.reason})" + for failure in images.failures + ) + _say( + f"directory: {image_progress.processed} of {image_progress.total} read, " + f"{images.created} created, {images.failed} reported — {reported}" + ) + _say(f"the run still {image_progress.state.value}: an unreadable file is not a failed run") + + # (9) Every asset got a preview at ingest, cached in the same blob store + # as its content. A thumbnail hash is a cache key and not an identity: + # it is in no release manifest, and nothing recomputes it to verify one. + everything = _all_assets(ingested, rerun, coarse, images) + thumbnailed = sum(1 for asset in everything.values() if asset.thumbnail_hash is not None) + _say(f"{thumbnailed} of {len(everything)} assets carry a cached preview") + + _say(f"events seen, in order: {', '.join(event.name for event in seen)}") + + return Summary( + project_id=project.id, + schema_version=schema.version, + clip_source_id=clip_source.id, + clip_batch_id=batch.id, + asset_ids=ingested.asset_ids, + clip_progress=clip_progress, + job_sizes=tuple(len(job.progress) for job in batch_jobs), + rerun=rerun, + coarse_source_id=coarse_source.id, + coarse=coarse, + images=images, + image_progress=image_progress, + asset_count=len(everything), + thumbnailed=thumbnailed, + events=tuple(event.name for event in seen), + ) + + +def _all_assets(*results: IngestResult) -> dict[UUID, Asset]: + """Every distinct asset the run produced, keyed by id. + + Four ingests over three sources, and only fifty-three assets between them — + which is the point. A dict rather than a list because the runs overlap by + design, and counting the overlap twice would describe a project that does + not exist. + """ + return {asset.id: asset for result in results for asset in result.assets} + + +def _say(message: str) -> None: + print(f" · {message}") + + +# --- running it ----------------------------------------------------------- + + +def _clear_previous_run(dest: Path) -> None: + """Remove a previous run of this example, and refuse to remove anything else. + + Only ever called for :data:`DEFAULT_DEST`. A directory that holds anything + other than what this example writes is not ours to delete, so it stops + instead of guessing. + """ + if not dest.exists(): + return + if not dest.is_dir(): + raise SystemExit(f"refusing to run: {dest} exists and is not a directory") + ours = {"visionset.db", "blobs", "clips", "incoming"} + stray = {entry.name for entry in dest.iterdir()} - ours + if stray: + raise SystemExit( + f"refusing to remove {dest}: it holds {', '.join(sorted(stray))}, " + f"which this example did not write" + ) + shutil.rmtree(dest) + + +def _run() -> None: + if len(sys.argv) > 2: + raise SystemExit(f"usage: {Path(sys.argv[0]).name} [DESTINATION]") + if len(sys.argv) == 2: + # A destination someone named is never removed automatically; if it is + # already a workspace, WorkspaceService says so and stops. + dest = Path(sys.argv[1]).resolve() + else: + dest = DEFAULT_DEST + _clear_previous_run(dest) + + print(f"VisionSet ingest end-to-end · {dest}") + summary = main(dest) + print( + f"\nDone. {len(summary.asset_ids)} assets from a " + f"{CLIP_SECONDS:.0f}s clip in {len(summary.job_sizes)} jobs, " + f"{summary.asset_count} in the project altogether.\n" + f"Workspace left at {dest} — open it again with WorkspaceService.open()." + ) + + +if __name__ == "__main__": + _run() diff --git a/tests/examples/test_ingest_end_to_end.py b/tests/examples/test_ingest_end_to_end.py new file mode 100644 index 00000000..87ebeba3 --- /dev/null +++ b/tests/examples/test_ingest_end_to_end.py @@ -0,0 +1,140 @@ +"""The ingest example, run as a smoke test. + +M2's exit criterion turned into a regression guard, the way +``test_sdk_end_to_end.py`` guards M1's: if a source stops registering, a decode +stops deduplicating, or a progress counter stops being written, this fails long +before anyone runs the example by hand. The assertions are about *outcomes* — +how many assets a ten-second clip yields, what a re-run creates, what the job +row says — never about the printed narration, which is free to change. + +The example is not part of the ``visionset`` package (it demonstrates the SDK +from outside it), so it is loaded from its path rather than imported by name. +""" + +from __future__ import annotations + +import importlib.util +import sys +from collections.abc import Iterator +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest +from tests.fixtures.media import require_ffmpeg + +from visionset.kernel.domain import IngestFailureKind, IngestState + +# The example generates its own clip and needs the binary to do it. Locally that +# is a skip; in CI, where VISIONSET_REQUIRE_FFMPEG=1, it is an error — a +# silently skipped video test looks exactly like a passing one. This is the +# right use of the fixture module: a *test* may reach into `tests/`, while the +# example it drives deliberately may not. +require_ffmpeg() + +EXAMPLE = Path(__file__).resolve().parents[2] / "examples" / "ingest_end_to_end.py" + + +@pytest.fixture(scope="module") +def example() -> Iterator[ModuleType]: + spec = importlib.util.spec_from_file_location("ingest_end_to_end", EXAMPLE) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + # Registered while it executes so dataclasses and pydantic can resolve the + # module by name; removed afterwards so the test leaves sys.modules as found. + sys.modules[spec.name] = module + try: + spec.loader.exec_module(module) + yield module + finally: + del sys.modules[spec.name] + + +@pytest.fixture(scope="module") +def summary(example: ModuleType, tmp_path_factory: pytest.TempPathFactory) -> Any: + return example.main(tmp_path_factory.mktemp("workspace") / "ingest-e2e") + + +def test_ten_seconds_at_five_frames_a_second_is_fifty_assets(summary: Any) -> None: + """The milestone's headline number, and it is a count of *distinct* frames. + + Fifty extraction slots and fifty assets only agree because the clip is large + enough for consecutive frames to differ — content addressing collapses any + that do not, which is why the example does not use the fixtures' 64x48. + """ + assert len(summary.asset_ids) == 50 + assert len(set(summary.asset_ids)) == 50 + + +def test_a_clip_reports_its_progress_without_knowing_its_total(summary: Any) -> None: + """`total` is NULL for a clip by design: `VideoMetadata` carries no frame count.""" + progress = summary.clip_progress + assert progress.state is IngestState.COMPLETED + assert progress.processed == 50 + assert progress.total is None + assert progress.failures == () + + +def test_the_batch_partitions_into_two_equal_jobs(summary: Any) -> None: + """An exact partition: disjoint, and their union is the batch.""" + assert summary.job_sizes == (25, 25) + assert sum(summary.job_sizes) == len(summary.asset_ids) + + +def test_re_ingesting_the_same_source_creates_nothing(summary: Any) -> None: + """Idempotency is a consequence of content addressing, not of bookkeeping.""" + rerun = summary.rerun + assert rerun.created == 0 + assert rerun.deduplicated == 50 + assert rerun.failed == 0 + assert set(rerun.asset_ids) == set(summary.asset_ids) + # A second batch, because the first froze at approval. A batch is an + # ephemeral unit of work and two of them may name the same assets. + assert rerun.batch_id != summary.clip_batch_id + + +def test_the_same_clip_at_another_rate_is_a_new_source_of_known_frames(summary: Any) -> None: + """Decomposition parameters belong to the source; identity belongs to the bytes. + + One clip registered at two rates is two sources — and the coarser one's ten + frames land exactly on grid points the finer run already produced, so the + project gains nothing. That alignment holds because the fps filter rounds + *up* onto the grid (#17); it is a property of this extractor, not a promise + the port makes about every rate pair. + """ + assert summary.coarse_source_id != summary.clip_source_id + assert len(summary.coarse.assets) == 10 + assert summary.coarse.created == 0 + assert set(summary.coarse.asset_ids) <= set(summary.asset_ids) + + +def test_a_directory_states_its_total_and_reports_what_it_could_not_read(summary: Any) -> None: + """The other path: countable up front, and one stray file does not fail the run.""" + progress = summary.image_progress + assert progress.state is IngestState.COMPLETED + assert progress.total == 4 + assert progress.processed == 4 + + images = summary.images + assert images.created == 3 + assert images.failed == 1 + (failure,) = images.failures + assert failure.kind is IngestFailureKind.UNSUPPORTED + assert failure.name.endswith("notes.txt") + # The report is a table, not a list of sentences: the reason never repeats + # the name, so a surface can group by kind. + assert "notes.txt" not in failure.reason + + +def test_every_asset_carries_a_cached_preview(summary: Any) -> None: + """#21's thumbnails, filled at ingest on both paths — frames included.""" + assert summary.asset_count == 53 # 50 frames + 3 stills, counted once each + assert summary.thumbnailed == summary.asset_count + + +def test_every_ingest_announced_itself_and_the_first_preceded_the_approval(summary: Any) -> None: + """Emission follows the commit, so the bus has the whole story in order.""" + events = summary.events + assert events.count("ingest_completed") == 4 + assert events.count("batch_approved") == 1 + assert events.index("ingest_completed") < events.index("batch_approved")