From 072d16fedf8d7d6ded39084075bbe34c77680444 Mon Sep 17 00:00:00 2001 From: Mahnoor Zaffar <1999mahnoor@gmail.com> Date: Sun, 16 Aug 2026 13:29:04 +0500 Subject: [PATCH 1/3] feat(benchmarks): add reproducible indexing-latency benchmark with regression detection (#1) * feat(benchmarks): add reproducible indexing-latency benchmark Adds a command that generates synthetic media via FFmpeg testsrc2 and measures per-stage indexing throughput, per-stage wall time, and peak memory across configurable modalities. Supports regression detection against a prior baseline report. - : corpus generation, run orchestrator (drives real run_index/ModelRuntime), per-stage aggregation, baseline comparison - CLI command with --modalities, --videos, --duration-seconds, --resolution, --repetitions, --input-mode (transcript/transcribe), --audio-mode, --baseline, --baseline-tolerance - documents the protocol, output schema, and limitations - 23 unit tests for validation, aggregation, clip command building, baseline comparison, and corpus spec * fix(benchmarks): address CodeRabbit review issues - Reject resolutions with extra components (e.g. 320x180x1) - Accumulate record_counts across repetitions instead of overwriting - Move corpus generation inside try block for proper failure handling - Pass reset parameter through instead of hardcoded True - Validate baseline configuration compatibility before comparison --- docs/benchmarking/README.md | 1 + docs/benchmarking/performance.md | 152 +++++++ src/vidxp/benchmarks/cli.py | 132 ++++++ src/vidxp/benchmarks/latency.py | 703 +++++++++++++++++++++++++++++++ tests/test_benchmark_latency.py | 329 +++++++++++++++ 5 files changed, 1317 insertions(+) create mode 100644 docs/benchmarking/performance.md create mode 100644 src/vidxp/benchmarks/latency.py create mode 100644 tests/test_benchmark_latency.py diff --git a/docs/benchmarking/README.md b/docs/benchmarking/README.md index cfe692fd..8e2e38bc 100644 --- a/docs/benchmarking/README.md +++ b/docs/benchmarking/README.md @@ -18,6 +18,7 @@ installation and product usage, start with the main | HiREST transcript localization | Legacy full result + current smoke | The legacy MiniLM stack scored all 193 validation pairs; current Qwen3 passed a two-video real execution smoke; 776 released test predictions remain unscored because their public bounds are placeholders | | LongVALE combined evaluation | Next | Build the visual-plus-speech adapter and validate one evaluation archive before scheduling the full run | | Actor clustering | Data-gated | The preferred BBT/Buffy evaluation still requires lawful access to the source episodes | +| Indexing latency benchmark | Ready | `vidxp benchmark index-latency` measures throughput, per-stage timings, and peak memory on synthetic FFmpeg media; supports regression detection against baselines | Read [current results](results.md) for the scores, plain-language metric definitions, honest comparisons, and the next benchmark decision. diff --git a/docs/benchmarking/performance.md b/docs/benchmarking/performance.md new file mode 100644 index 00000000..192c0bf1 --- /dev/null +++ b/docs/benchmarking/performance.md @@ -0,0 +1,152 @@ +# Latency benchmark protocol + +Status: Ready + +The latency benchmark (`vidxp benchmark index-latency`) measures indexing +throughput, per-stage timing, and peak memory using synthetic media generated +by FFmpeg on the caller's machine. It is designed for regression detection +between VidXP builds and for evaluating the latency impact of model or +architecture changes. + +## Protocol + +### Corpus generation + +The benchmark generates deterministic synthetic clips using FFmpeg's `lavfi` +source filters: + +| Parameter | Default | Notes | +|---|---|---| +| `--videos` | 1 | Number of synthetic clips | +| `--duration-seconds` | 8.0 | Wall-clock duration of each clip | +| `--fps` | 24 | Frame rate | +| `--resolution` | 320x180 | `WxH` format | +| `--audio-mode` | `none` | `none`, `sine`, or `flite` | +| `--input-mode` | `transcript` | `transcript` or `transcribe` | + +Video is generated via `testsrc2` (colour bars + timestamp). When +`--input-mode transcript` and `dialogue` is enabled, a deterministic +synthetic transcript (seeded PRNG over a fixed English vocabulary) is +supplied without real transcription. When `input-mode transcribe` is +used, `--audio-mode flite` must also be set and libflite must be +available in the ffmpeg build. + +### Indexing measurement + +Each repetition runs the full indexing pipeline via `run_index()` with +`reset=True`. The following stages are timed by the existing manifest +timing infrastructure (`core/manifest.py:record_stage`): + +| Stage | Modality | Measures | +|---|---|---| +| `frame_stream` | (all visual) | Decode throughput (frames/s) | +| `scene` | scene | SigLIP2 embedding (frames/s) | +| `actor` | actor | OpenCV detect + recognise (frames/s) | +| `visual_indexing` | all visual | Combined group wall time | +| `dialogue_indexing` | dialogue | Embedding throughput (phrases/s) | + +Peak RSS is captured via `resource.getrusage(RUSAGE_SELF).ru_maxrss` +(POSIX only; `None` on Windows, reported in bytes on macOS, KiB on +Linux). + +### Repetitions + +When `--repetitions N` > 1, each repetition runs the full cycle +(generate once, index each time after `reset`). Results are reported +as mean, min, and max across all per-video per-repetition samples. + +### Baseline comparison + +Pass `--baseline ` to compare the +current run against a prior report. For each stage present in both, +the delta ratio (`new_mean / old_mean - 1`) is computed. A stage with +a delta exceeding `--baseline-tolerance` (default 0.15 = 15%) is +flagged as a regression. The verdict is `fail` if any stage regressed, +else `pass`. + +### Output + +The benchmark writes its report to `run_directory/report.json` and +invokes `record_adapter_manifest` (embedding the corpus spec, device, +and result classification into the run's `manifest.json`). + +Report schema: + +```json +{ + "schema_version": 1, + "benchmark": "latency", + "run_id": "my-run", + "corpus": { "videos": 1, "duration_seconds": 8.0, ... }, + "modalities": ["scene", "actor"], + "device": "cpu", + "repetitions": 1, + "git": { "commit": "...", "dirty": false }, + "environment": { ... }, + "record_counts": { "scene": 8, "actor": 0 }, + "processed_frames": 8, + "stages": { + "scene": { + "runs": 1, "mean_seconds": 2.1, "min_seconds": 2.1, + "max_seconds": 2.1, "rate_per_second": 3.8 + } + }, + "summary": { + "wall_seconds": { "runs": 1, "mean_seconds": 5.0, ... }, + "peak_rss": { "unit": "bytes", "samples": 1, "value": 123456789 } + }, + "baseline": null | { "stages": {...}, "regressions": [...], "verdict": "pass" } +} +``` + +## Limitations + +- The synthetic video has no semantic scene content, so scene embeddings + are representative of throughput but not retrieval quality. +- Actors are not present in `testsrc2` video; `actor` stage measures + the per-frame face-detection overhead with zero detections. +- When `input_mode=transcript`, no real whisper transcription occurs; + dialogue embedding is measured on a synthetic transcript. +- True transcription latency (`input_mode=transcribe`) requires a + speech source (`--audio-mode flite`) and libflite in the FFmpeg + build; the generated speech is a short fixed sentence and does not + represent naturalistic conversation length or vocabulary. +- Peak RSS measures the whole-process peak, which includes Python + overhead, loaded models, and Chroma state; it is not a pure + indexing-stage measurement. + +## Usage + +```bash +# Default: single 8-second 320x180 clip, scene-only, 1 rep +vidxp benchmark index-latency --run-id my-baseline + +# Scene + actor + dialogue (synthetic transcript), 3 reps, compare with baseline +vidxp benchmark index-latency \ + --run-id v2-compare \ + --modalities scene,actor,dialogue \ + --videos 2 \ + --duration-seconds 12 \ + --repetitions 3 \ + --json \ + --baseline benchmark_runs/latency/synthetic/my-baseline/report.json + +# Real transcription (requires libflite in ffmpeg) +vidxp benchmark index-latency \ + --run-id transcribe-test \ + --modalities dialogue \ + --input-mode transcribe \ + --audio-mode flite \ + --device cpu +``` + +## Adding a new performance benchmark + +1. Define the corpus parameters and any new modality combinations in + the existing `run_latency` entry point. +2. Run the baseline and save its `report.json`. +3. Make your change (model swap, concurrency refactor, etc.). +4. Re-run with `--baseline ` and verify no + regressions. +5. Commit the baseline report to a designated location (e.g. + `docs/benchmarking/baselines/`) if it serves as a team reference. diff --git a/src/vidxp/benchmarks/cli.py b/src/vidxp/benchmarks/cli.py index 34fa5e2d..e8f069dd 100644 --- a/src/vidxp/benchmarks/cli.py +++ b/src/vidxp/benchmarks/cli.py @@ -21,6 +21,7 @@ HIREST_DEFAULT_WINDOW_FRACTION, run_hirest, ) +from vidxp.benchmarks.latency import run_latency from vidxp.benchmarks.prepare import ( PreparationPlan, execute_preparation, @@ -574,3 +575,134 @@ def hirest_command( emit_json(metrics) else: rich_print(metrics) + + +@app.command("index-latency") +def index_latency_command( + ctx: typer.Context, + run_id: Annotated[str, typer.Option(help="Arbitrary label for this run.")], + modalities: Annotated[ + str, + typer.Option( + help="Comma-separated modality names: scene,actor,dialogue." + ), + ] = "scene", + videos: Annotated[ + int, + typer.Option(min=1, help="Number of synthetic clips to generate."), + ] = 1, + duration_seconds: Annotated[ + float, + typer.Option(min=0.1, help="Duration of each synthetic clip."), + ] = 8.0, + fps: Annotated[ + int, + typer.Option(min=1, help="Frame rate of synthetic clips."), + ] = 24, + resolution: Annotated[ + str, + typer.Option( + help="Synthetic clip resolution in WxH format (e.g. 320x180)." + ), + ] = "320x180", + repetitions: Annotated[ + int, + typer.Option(min=1, help="Number of times to repeat the run."), + ] = 1, + input_mode: Annotated[ + Literal["transcript", "transcribe"], + typer.Option( + help=( + "'transcript' supplies a synthetic transcript for dialogue " + "embedding (no real transcription). 'transcribe' runs " + "real whisper on audio (requires --audio-mode flite)." + ) + ), + ] = "transcript", + audio_mode: Annotated[ + Literal["none", "sine", "flite"], + typer.Option( + help=( + "Audio track for synthetic clips: 'none' (no audio), " + "'sine' (tone), or 'flite' (speech synthesis)." + ) + ), + ] = "none", + reset: Annotated[ + bool, + typer.Option(help="Clear any existing index before running."), + ] = False, + baseline: Annotated[ + Path | None, + typer.Option( + exists=True, + dir_okay=False, + help=( + "Path to a previous latency report JSON for regression " + "comparison." + ), + ), + ] = None, + baseline_tolerance: Annotated[ + float, + typer.Option( + min=0.0, + max=5.0, + help=( + "Relative regression tolerance. A stage mean slower by " + "more than this ratio flags as regression." + ), + ), + ] = 0.15, + json_output: Annotated[ + bool, + typer.Option("--json", help="Emit machine-readable JSON."), + ] = False, +) -> None: + """Run a reproducible indexing-latency benchmark on synthetic media.""" + + selected = [item.strip() for item in modalities.split(",") if item.strip()] + if not selected: + raise typer.BadParameter( + "At least one latency modality is required.", param_hint="--modalities" + ) + for modality in selected: + _require_benchmark_dependencies(modality) + + try: + parts = resolution.lower().split("x") + if len(parts) != 2: + raise ValueError + width, height = int(parts[0]), int(parts[1]) + if width <= 0 or height <= 0: + raise ValueError + except (IndexError, ValueError, AttributeError): + raise typer.BadParameter( + f"Invalid resolution: {resolution!r}. Use WxH, e.g. 320x180.", + param_hint="--resolution", + ) + + state = state_from_context(ctx) + report = run_latency( + run_id=run_id, + output_root=state.settings.data_dir / "benchmark_runs", + ffprobe=state.settings.ffprobe_executable, + ffmpeg=state.settings.ffmpeg_executable, + modalities=tuple(selected), + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + repetitions=repetitions, + input_mode=input_mode, + audio_mode=audio_mode, + device=state.settings.runtime_backend, + reset=reset, + baseline_path=baseline, + baseline_tolerance=baseline_tolerance, + ) + if effective_output_format(state, json_output) == OutputFormat.json: + emit_json(report) + else: + rich_print(report) diff --git a/src/vidxp/benchmarks/latency.py b/src/vidxp/benchmarks/latency.py new file mode 100644 index 00000000..87272f57 --- /dev/null +++ b/src/vidxp/benchmarks/latency.py @@ -0,0 +1,703 @@ +from __future__ import annotations + +import json +import random +import subprocess +from dataclasses import dataclass +from pathlib import Path +from statistics import mean +from time import perf_counter +from typing import Any, Literal, Mapping, Sequence + +from vidxp.benchmarks.common import ( + append_failure, + benchmark_generation_id, + benchmark_media_id, + ensure_adapter_outputs, + record_adapter_manifest, +) +from vidxp.capabilities.registry import create_capability_registry +from vidxp.core.contracts import IndexConfig, VideoSource +from vidxp.core.manifest import ManifestStore, write_json_atomic +from vidxp.core.runner import run_index +from vidxp.core.storage import IndexStorage +from vidxp.infrastructure.local_index import LOCAL_INDEX_RUNTIME_CHECKS +from vidxp.media_runtime import inspect_media_runtime +from vidxp.runtime import ModelRuntime +from vidxp.settings import VidXPSettings + + +LATENCY_BENCHMARK = "latency" +LATENCY_SPLIT = "synthetic" +LATENCY_SCHEMA_VERSION = 1 +DEFAULT_CORPUS_SEED = 2026 +SUPPORTED_MODALITIES = ("scene", "actor", "dialogue") + +_STAGE_RATES: Mapping[str, str] = { + "scene": "scene_frames", + "actor": "actor_frames", + "frame_stream": "source_frames_advanced", + "dialogue_indexing": "dialogue_phrases", +} + +_VOCABULARY = ( + "the quick brown fox jumps over the lazy dog honest sunshine light " + "morning river ocean mountain garden flower silver golden copper " + "bright shadow shadow candle lantern window door table chair book " + "letter number station market kitchen garden bakery camera video " + "music voice speech word phrase moment memory journey story world " + "quiet calm gentle peaceful vivid warm cool bright dark soft loud" +).split() + + +def _peak_rss_bytes() -> int | None: + try: + import resource + except ImportError: + return None + return int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) + + +def rss_unit() -> Literal["bytes", "KiB"]: + return "bytes" if _sys_platform() == "darwin" else "KiB" + + +def _sys_platform() -> str: + import sys + + return sys.platform + + +@dataclass(frozen=True) +class SyntheticCorpusSpec: + videos: int + duration_seconds: float + fps: int + width: int + height: int + audio_mode: Literal["none", "sine", "flite"] + seed: int + + def public_record(self) -> dict[str, Any]: + return { + "videos": self.videos, + "duration_seconds": self.duration_seconds, + "fps": self.fps, + "width": self.width, + "height": self.height, + "audio_mode": self.audio_mode, + "seed": self.seed, + } + + +def validate_latency_options( + *, + modalities: Sequence[str], + videos: int, + duration_seconds: float, + fps: int, + width: int, + height: int, + repetitions: int, + input_mode: str, + audio_mode: str, + baseline_tolerance: float, +) -> tuple[str, ...]: + selected = tuple(dict.fromkeys(modalities)) + if not selected: + raise ValueError("At least one latency modality must be selected.") + unsupported = sorted(set(selected) - set(SUPPORTED_MODALITIES)) + if unsupported: + raise ValueError( + "Latency modalities must be a subset of " + + ", ".join(SUPPORTED_MODALITIES) + + "; unsupported: " + + ", ".join(unsupported) + ) + if videos <= 0: + raise ValueError("videos must be greater than zero.") + if duration_seconds <= 0: + raise ValueError("duration_seconds must be greater than zero.") + if fps <= 0: + raise ValueError("fps must be greater than zero.") + if width <= 0 or height <= 0: + raise ValueError("width and height must be greater than zero.") + if repetitions <= 0: + raise ValueError("repetitions must be greater than zero.") + if input_mode not in {"transcript", "transcribe"}: + raise ValueError("input_mode must be 'transcript' or 'transcribe'.") + if audio_mode not in {"none", "sine", "flite"}: + raise ValueError("audio_mode must be 'none', 'sine', or 'flite'.") + if "dialogue" in selected and input_mode == "transcribe": + if audio_mode != "flite": + raise ValueError( + "Real transcription requires a speech audio source; " + "use --audio-mode flite with --input-mode transcribe." + ) + if not 0 <= baseline_tolerance <= 5: + raise ValueError("baseline_tolerance must be between zero and five.") + return selected + + +def synthetic_transcript( + *, + duration_seconds: float, + seed: int, +) -> list[dict[str, Any]]: + generator = random.Random(seed) + strides = max(1, int(duration_seconds / 0.4)) + words = [generator.choice(_VOCABULARY) for _ in range(strides)] + span = duration_seconds / len(words) + word_events = [ + { + "word": word, + "start": round(index * span, 4), + "end": round((index + 1) * span, 4), + } + for index, word in enumerate(words) + ] + return [ + { + "text": " ".join(words), + "start": 0.0, + "end": duration_seconds, + "words": word_events, + } + ] + + +def _flite_text(seed: int) -> str: + generator = random.Random(seed) + words = [generator.choice(_VOCABULARY) for _ in range(24)] + return " ".join(words) + + +def build_clip_command( + *, + spec: SyntheticCorpusSpec, + ffmpeg: str, + destination: Path, +) -> list[str]: + compact = spec.width != 0 and spec.height != 0 + if not compact: + raise ValueError("The synthetic corpus requires positive dimensions.") + command = [ + ffmpeg, + "-y", + "-v", + "error", + "-f", + "lavfi", + "-i", + f"testsrc2=size={spec.width}x{spec.height}:rate={spec.fps}", + ] + if spec.audio_mode == "sine": + command += [ + "-f", + "lavfi", + "-i", + "sine=frequency=440:sample_rate=16000", + ] + elif spec.audio_mode == "flite": + command += [ + "-f", + "lavfi", + "-i", + f"flite=text='{_flite_text(spec.seed)}',sample_rate=16000", + ] + command += [ + "-t", + f"{spec.duration_seconds:g}", + "-c:v", + "libx264", + "-pix_fmt", + "yuv420p", + ] + if spec.audio_mode == "none": + command.append("-an") + else: + command += ["-c:a", "aac"] + command.append(str(destination)) + return command + + +def _probe_duration(ffprobe: str, path: Path) -> float: + completed = subprocess.run( + [ + ffprobe, + "-v", + "error", + "-show_entries", + "format=duration", + "-of", + "json", + str(path), + ], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=60, + ) + if completed.returncode != 0: + raise ValueError( + f"ffprobe could not read a generated clip: {path}" + ) + try: + duration = float(json.loads(completed.stdout)["format"]["duration"]) + except (KeyError, ValueError, TypeError, json.JSONDecodeError) as exc: + raise ValueError( + f"ffprobe returned an invalid duration for {path}." + ) from exc + if duration <= 0: + raise ValueError(f"ffprobe reported a non-positive duration for {path}.") + return duration + + +def generate_synthetic_corpus( + *, + spec: SyntheticCorpusSpec, + directory: str | Path, + ffprobe: str, + ffmpeg: str, + audio_mode: str | None = None, +) -> list[Path]: + runtime_status = inspect_media_runtime( + ffprobe=ffprobe, + ffmpeg=ffmpeg, + ) + if not runtime_status.ready: + raise ValueError( + "The latency benchmark requires FFmpeg and ffprobe to generate " + "the synthetic corpus. Run `vidxp init`, then retry." + ) + destination = Path(directory) + destination.mkdir(parents=True, exist_ok=True) + clips = [] + for index in range(spec.videos): + path = destination / f"clip-{index:03d}.mp4" + command = build_clip_command( + spec=spec, + ffmpeg=ffmpeg, + destination=path, + ) + completed = subprocess.run( + command, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=600, + ) + if completed.returncode != 0: + stderr = (completed.stderr or "").strip() + if spec.audio_mode == "flite" and stderr: + raise ValueError( + "FFmpeg could not apply the flite speech filter " + f"(libflite likely unavailable): {stderr}" + ) + raise ValueError(f"FFmpeg could not generate {path}: {stderr}") + if not path.is_file() or path.stat().st_size == 0: + raise ValueError(f"FFmpeg did not produce {path}.") + _probe_duration(ffprobe, path) + clips.append(path) + return clips + + +def _per_video_stages( + manifest: Mapping[str, Any], + video_id: str, +) -> dict[str, float]: + video = manifest["videos"].get(video_id) + if video is None or video.get("state") == "failed": + return {} + return { + str(stage): float(entry["seconds"]) + for stage, entry in (video.get("stages") or {}).items() + if entry.get("state") != "incomplete" + and float(entry.get("seconds", 0.0)) > 0 + } + + +def _summary_ratio( + manifest: Mapping[str, Any], + video_id: str, + *, + metric: str, + stage: str, +) -> float | None: + video = manifest["videos"].get(video_id) or {} + summary = video.get("summary") or {} + seconds = _per_video_stages(manifest, video_id).get(stage) + count = summary.get(metric) + if seconds is None or count is None or seconds <= 0 or count <= 0: + return None + return float(count) / seconds + + +def aggregate_latency_runs( + manifests: Sequence[Mapping[str, Any]], + *, + wall_seconds: Sequence[float], + peak_rss_samples: Sequence[int | None], +) -> dict[str, Any]: + if len(manifests) != len(wall_seconds): + raise ValueError( + "Every latency repetition requires a wall-clock sample." + ) + stage_samples: dict[str, list[float]] = {} + rate_samples: dict[str, list[float]] = {} + per_video: list[dict[str, Any]] = [] + processed_frames = 0 + record_counts: dict[str, int] = {} + for repetition, manifest in enumerate(manifests): + processed_frames += int(manifest.get("processed_frames", 0)) + for modality, count in (manifest.get("record_counts") or {}).items(): + record_counts[modality] = record_counts.get(modality, 0) + int(count) + for video_id in sorted(manifest.get("videos", {})): + stages = _per_video_stages(manifest, video_id) + if not stages: + continue + for stage, seconds in stages.items(): + stage_samples.setdefault(stage, []).append(seconds) + rate_stages = { + stage: _summary_ratio( + manifest, + video_id, + metric=_STAGE_RATES[stage], + stage=stage, + ) + for stage in _STAGE_RATES + if stage in stages + } + for stage, rate in rate_stages.items(): + if rate is None: + continue + rate_samples.setdefault(stage, []).append(rate) + video = manifest["videos"].get(video_id, {}) + per_video.append( + { + "repetition": repetition, + "video_id": video_id, + "wall_seconds": ( + wall_seconds[repetition] + ), + "stages": dict(sorted(stages.items())), + "summary": video.get("summary", {}), + } + ) + stages: dict[str, dict[str, Any]] = {} + for stage, samples in stage_samples.items(): + values = sorted(samples) + summary: dict[str, Any] = { + "runs": len(values), + "mean_seconds": mean(values), + "min_seconds": values[0], + "max_seconds": values[-1], + } + rates = rate_samples.get(stage) + if rates: + summary["rate_per_second"] = mean(rates) + stages[stage] = summary + approximate_rss = [ + sample for sample in peak_rss_samples if sample is not None + ] + summary = { + "wall_seconds": { + "runs": len(wall_seconds), + "mean_seconds": mean(wall_seconds), + "min_seconds": min(wall_seconds), + "max_seconds": max(wall_seconds), + }, + "peak_rss": { + "unit": rss_unit(), + "samples": len(approximate_rss), + "value": int(max(approximate_rss)) if approximate_rss else None, + }, + } + return { + "per_video": per_video, + "stages": dict(sorted(stages.items())), + "summary": summary, + "processed_frames": processed_frames, + "record_counts": dict(sorted(record_counts.items())), + } + + +def _validate_baseline_compatibility( + report: Mapping[str, Any], + baseline: Mapping[str, Any], +) -> None: + baseline_corpus = baseline.get("corpus") or {} + report_corpus = report.get("corpus") or {} + corpus_keys = ("videos", "duration_seconds", "fps", "width", "height", "audio_mode", "seed") + for key in corpus_keys: + report_val = report_corpus.get(key) + baseline_val = baseline_corpus.get(key) + if report_val is not None and baseline_val is not None and report_val != baseline_val: + raise ValueError( + f"Baseline corpus mismatch: {key!r} is {baseline_val!r}, " + f"current is {report_val!r}." + ) + for key in ("input_mode", "device"): + report_val = report.get(key) + baseline_val = baseline.get(key) + if report_val is not None and baseline_val is not None and report_val != baseline_val: + raise ValueError( + f"Baseline {key!r} mismatch: {baseline_val!r}, current is {report_val!r}." + ) + report_mods = sorted(report.get("modalities") or []) + baseline_mods = sorted(baseline.get("modalities") or []) + if report_mods and baseline_mods and report_mods != baseline_mods: + raise ValueError( + f"Baseline modalities mismatch: {baseline_mods}, current is {report_mods}." + ) + + +def compare_baseline( + report: Mapping[str, Any], + baseline: Mapping[str, Any], + *, + tolerance: float, +) -> dict[str, Any]: + if tolerance < 0: + raise ValueError("Baseline tolerance must be nonnegative.") + comparisons: dict[str, dict[str, Any]] = {} + regressions: list[str] = [] + for stage, previous in (baseline.get("stages") or {}).items(): + current = (report.get("stages") or {}).get(stage) + if current is None or not previous.get("runs"): + continue + old_mean = float(previous["mean_seconds"]) + new_mean = float(current["mean_seconds"]) + if old_mean <= 0: + continue + delta = new_mean / old_mean - 1.0 + comparisons[stage] = { + "old_mean_seconds": old_mean, + "new_mean_seconds": new_mean, + "delta_ratio": delta, + "regressed": delta > tolerance, + } + if delta > tolerance: + regressions.append(stage) + return { + "schema_version": LATENCY_SCHEMA_VERSION, + "tolerance": tolerance, + "stages": dict(sorted(comparisons.items())), + "regressions": regressions, + "verdict": "fail" if regressions else "pass", + } + + +def build_latency_sources( + *, + clips: Sequence[Path], + spec: SyntheticCorpusSpec, + input_mode: Literal["transcript", "transcribe"], +) -> list[VideoSource]: + sources = [] + for index, path in enumerate(clips): + transcript = ( + synthetic_transcript( + duration_seconds=spec.duration_seconds, + seed=spec.seed + index, + ) + if input_mode == "transcript" + else None + ) + sources.append( + VideoSource( + video_id=benchmark_media_id( + LATENCY_BENCHMARK, + f"clip-{index:03d}", + ), + path=path, + source_name=f"clip-{index:03d}.mp4", + transcript=transcript, + ) + ) + return sources + + +def run_latency( + *, + run_id: str, + output_root: str | Path = "benchmark_runs", + ffprobe: str = "ffprobe", + ffmpeg: str = "ffmpeg", + modalities: Sequence[str] = ("scene",), + videos: int = 1, + duration_seconds: float = 8.0, + fps: int = 24, + width: int = 320, + height: int = 180, + repetitions: int = 1, + input_mode: Literal["transcript", "transcribe"] = "transcript", + audio_mode: Literal["none", "sine", "flite"] = "none", + device: str = "cpu", + reset: bool = False, + baseline_path: str | Path | None = None, + baseline_tolerance: float = 0.15, +) -> dict[str, Any]: + selected = validate_latency_options( + modalities=modalities, + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + repetitions=repetitions, + input_mode=input_mode, + audio_mode=audio_mode, + baseline_tolerance=baseline_tolerance, + ) + spec = SyntheticCorpusSpec( + videos=videos, + duration_seconds=duration_seconds, + fps=fps, + width=width, + height=height, + audio_mode=audio_mode, + seed=DEFAULT_CORPUS_SEED, + ) + config = IndexConfig( + dataset=LATENCY_BENCHMARK, + split=LATENCY_SPLIT, + run_id=run_id, + enabled_modalities=selected, + device=device, + output_root=output_root, + generation_id=benchmark_generation_id( + LATENCY_BENCHMARK, + LATENCY_SPLIT, + run_id, + ), + ) + run_directory = config.run_directory + registry = create_capability_registry( + platform_runtime_checks=LOCAL_INDEX_RUNTIME_CHECKS + ) + runtime = ModelRuntime( + VidXPSettings( + repository_root=run_directory, + runtime_backend=device, + ), + allowed_specs=registry.model_specs(), + ) + ensure_adapter_outputs(run_directory) + manifests: list[dict[str, Any]] = [] + wall_samples: list[float] = [] + rss_samples: list[int | None] = [] + try: + clips = generate_synthetic_corpus( + spec=spec, + directory=run_directory / "corpus", + ffprobe=ffprobe, + ffmpeg=ffmpeg, + ) + sources = build_latency_sources( + clips=clips, + spec=spec, + input_mode=input_mode, + ) + for _ in range(repetitions): + started = perf_counter() + with IndexStorage(config) as storage: + manifest = run_index( + sources, + config, + reset=reset, + storage=storage, + manifest_store=ManifestStore( + config, + registry=registry, + runtime=runtime, + ), + registry=registry, + runtime=runtime, + ) + store_size_bytes = storage.size_bytes() + wall_samples.append(perf_counter() - started) + rss_samples.append(_peak_rss_bytes()) + manifests.append( + { + **manifest, + "store_size_bytes_at_commit": store_size_bytes, + } + ) + aggregated = aggregate_latency_runs( + manifests, + wall_seconds=wall_samples, + peak_rss_samples=rss_samples, + ) + report = { + "schema_version": LATENCY_SCHEMA_VERSION, + "benchmark": LATENCY_BENCHMARK, + "run_id": run_id, + "created_at": manifests[-1].get("completed_at"), + "corpus": spec.public_record(), + "input_mode": input_mode, + "modalities": list(selected), + "device": device, + "repetitions": repetitions, + "git": manifests[-1].get("git"), + "environment": manifests[-1].get("environment"), + "config_fingerprint": manifests[-1].get("config_fingerprint"), + "record_counts": aggregated["record_counts"], + "processed_frames": aggregated["processed_frames"], + "summary": aggregated["summary"], + "stages": aggregated["stages"], + "per_video": aggregated["per_video"], + "baseline": None, + } + if baseline_path is not None: + try: + baseline = json.loads( + Path(baseline_path).read_text(encoding="utf-8") + ) + except (OSError, json.JSONDecodeError) as exc: + raise ValueError( + f"Baseline report is not readable JSON: {baseline_path}" + ) from exc + _validate_baseline_compatibility(report, baseline) + report["baseline"] = compare_baseline( + report, + baseline, + tolerance=baseline_tolerance, + ) + write_json_atomic(run_directory / "report.json", report) + record_adapter_manifest( + run_directory, + benchmark=LATENCY_BENCHMARK, + subset={ + "label": f"latency_{run_id}", + "modalities": list(selected), + "video_count": videos, + "duration_seconds": duration_seconds, + "repetitions": repetitions, + }, + artifacts=[], + state="complete", + details={ + "device": device, + "input_mode": input_mode, + "audio_mode": audio_mode, + "corpus": spec.public_record(), + "result_classification": "performance_benchmark_not_quality_score", + }, + ) + return report + except BaseException as error: + append_failure(run_directory, stage="latency_adapter", error=error) + record_adapter_manifest( + run_directory, + benchmark=LATENCY_BENCHMARK, + subset={ + "label": f"latency_{run_id}", + "modalities": list(selected), + }, + artifacts=[], + state="failed", + ) + raise \ No newline at end of file diff --git a/tests/test_benchmark_latency.py b/tests/test_benchmark_latency.py new file mode 100644 index 00000000..6cbd7057 --- /dev/null +++ b/tests/test_benchmark_latency.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import unittest +from pathlib import Path + +from vidxp.benchmarks.latency import ( + SyntheticCorpusSpec, + aggregate_latency_runs, + build_clip_command, + build_latency_sources, + compare_baseline, + synthetic_transcript, + validate_latency_options, +) + + +class LatencyValidationTests(unittest.TestCase): + def test_validates_default_options(self): + selected = validate_latency_options( + modalities=("scene",), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=3, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("scene",)) + + def test_rejects_empty_modalities(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=(), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + + def test_rejects_unsupported_modality(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=("scene", "ocr"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + + def test_rejects_transcribe_without_flite(self): + with self.assertRaises(ValueError): + validate_latency_options( + modalities=("scene", "dialogue"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcribe", + audio_mode="sine", + baseline_tolerance=0.15, + ) + + def test_accepts_transcribe_with_flite(self): + selected = validate_latency_options( + modalities=("dialogue",), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcribe", + audio_mode="flite", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("dialogue",)) + + def test_deduplicates_modalities(self): + selected = validate_latency_options( + modalities=("scene", "scene", "actor"), + videos=1, + duration_seconds=8.0, + fps=24, + width=320, + height=180, + repetitions=1, + input_mode="transcript", + audio_mode="none", + baseline_tolerance=0.15, + ) + self.assertEqual(selected, ("scene", "actor")) + + +class SyntheticTranscriptTests(unittest.TestCase): + def test_returns_one_segment_with_words(self): + transcript = synthetic_transcript(duration_seconds=10.0, seed=42) + self.assertEqual(len(transcript), 1) + segment = transcript[0] + self.assertGreater(len(segment["text"]), 0) + self.assertEqual(segment["start"], 0.0) + self.assertGreater(segment["end"], 0.0) + self.assertGreater(len(segment["words"]), 0) + for word in segment["words"]: + self.assertIn("word", word) + self.assertIsInstance(word["start"], float) + self.assertIsInstance(word["end"], float) + + def test_deterministic_across_calls(self): + first = synthetic_transcript(duration_seconds=5.0, seed=99) + second = synthetic_transcript(duration_seconds=5.0, seed=99) + self.assertEqual(first, second) + + def test_different_seeds_differ(self): + first = synthetic_transcript(duration_seconds=5.0, seed=99) + second = synthetic_transcript(duration_seconds=5.0, seed=100) + self.assertNotEqual(first, second) + + +class BuildClipCommandTests(unittest.TestCase): + def test_no_audio_default(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + self.assertIn("testsrc2=size=320x180:rate=24", command) + self.assertIn("-an", command) + self.assertNotIn("-c:a", command) + + def test_sine_audio_adds_aac(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="sine", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + self.assertIn("sine=frequency=440:sample_rate=16000", command) + self.assertIn("-c:a", command) + self.assertNotIn("-an", command) + + def test_flite_audio_contains_filter_ref(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="flite", seed=42, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) + flite_args = [arg for arg in command if "flite=text=" in arg] + self.assertEqual(len(flite_args), 1) + + def test_duration_is_formatted(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=3.5, fps=30, + width=640, height=480, audio_mode="none", seed=0, + ) + command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("clip.mp4")) + idx = command.index("-t") + self.assertEqual(command[idx + 1], "3.5") + self.assertIn("testsrc2=size=640x480:rate=30", command) + + +class BuildSourcesTests(unittest.TestCase): + def test_transcript_attached_in_input_mode(self): + spec = SyntheticCorpusSpec( + videos=2, duration_seconds=4.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + clips = [Path(f"{i}.mp4") for i in range(2)] + sources = build_latency_sources(clips=clips, spec=spec, input_mode="transcript") + self.assertEqual(len(sources), 2) + for index, source in enumerate(sources): + self.assertIsNotNone(source.transcript) + self.assertIsNotNone(source.path) + self.assertIsNotNone(source.video_id) + self.assertEqual(source.source_name, f"clip-{index:03d}.mp4") + + def test_no_transcript_in_transcribe_mode(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=4.0, fps=24, + width=320, height=180, audio_mode="flite", seed=42, + ) + sources = build_latency_sources(clips=[Path("0.mp4")], spec=spec, input_mode="transcribe") + for source in sources: + self.assertIsNone(source.transcript) + + +class AggregateMetricsTests(unittest.TestCase): + def _sample_manifest(self, scene_seconds, scene_frames, actor_seconds, actor_frames): + return { + "processed_frames": scene_frames, + "record_counts": {"scene": scene_frames, "actor": actor_frames}, + "git": {"commit": "abc", "dirty": False}, + "environment": {"platform": "test"}, + "config_fingerprint": "fp1", + "completed_at": "2026-01-01T00:00:00", + "videos": { + "vid-1": { + "state": "complete", + "summary": { + "scene_frames": scene_frames, + "actor_frames": actor_frames, + "source_frames_advanced": scene_frames + 100, + }, + "stages": { + "scene": {"seconds": scene_seconds, "state": ""}, + "actor": {"seconds": actor_seconds, "state": ""}, + "frame_stream": {"seconds": 0.5, "state": ""}, + }, + } + }, + } + + def test_aggregates_single_manifest(self): + result = aggregate_latency_runs( + [self._sample_manifest(2.0, 8, 1.5, 3)], + wall_seconds=[3.5], + peak_rss_samples=[100000], + ) + self.assertEqual(result["processed_frames"], 8) + self.assertEqual(result["record_counts"], {"actor": 3, "scene": 8}) + self.assertIn("scene", result["stages"]) + self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.0) + self.assertAlmostEqual(result["stages"]["scene"]["rate_per_second"], 4.0) + self.assertAlmostEqual(result["stages"]["actor"]["mean_seconds"], 1.5) + self.assertAlmostEqual(result["summary"]["wall_seconds"]["mean_seconds"], 3.5) + + def test_aggregates_multiple_manifests(self): + m1 = self._sample_manifest(2.0, 8, 1.5, 3) + m2 = self._sample_manifest(2.5, 10, 2.0, 4) + result = aggregate_latency_runs( + [m1, m2], + wall_seconds=[3.5, 4.5], + peak_rss_samples=[100000, 120000], + ) + self.assertEqual(result["processed_frames"], 18) + self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.25) + self.assertAlmostEqual(result["stages"]["scene"]["min_seconds"], 2.0) + self.assertAlmostEqual(result["stages"]["scene"]["max_seconds"], 2.5) + self.assertEqual(len(result["per_video"]), 2) + self.assertAlmostEqual( + result["summary"]["wall_seconds"]["mean_seconds"], + 4.0, + ) + + def test_skips_failed_videos(self): + manifest = { + "processed_frames": 0, + "record_counts": {}, + "git": {}, + "environment": {}, + "config_fingerprint": "fp", + "completed_at": "", + "videos": { + "vid-1": { + "state": "failed", + "summary": {}, + "stages": {}, + } + }, + } + result = aggregate_latency_runs( + [manifest], + wall_seconds=[1.0], + peak_rss_samples=[None], + ) + self.assertEqual(result["processed_frames"], 0) + self.assertEqual(result["stages"], {}) + + +class CompareBaselineTests(unittest.TestCase): + def test_no_baseline_stages_returns_empty(self): + report = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + baseline = {"stages": {}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertEqual(result["stages"], {}) + self.assertEqual(result["regressions"], []) + self.assertEqual(result["verdict"], "pass") + + def test_regression_detected(self): + report = {"stages": {"scene": {"mean_seconds": 3.0, "runs": 1}}} + baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertIn("scene", result["stages"]) + self.assertAlmostEqual( + result["stages"]["scene"]["delta_ratio"], 0.5 + ) + self.assertTrue(result["stages"]["scene"]["regressed"]) + self.assertEqual(result["regressions"], ["scene"]) + self.assertEqual(result["verdict"], "fail") + + def test_improvement_not_regression(self): + report = {"stages": {"scene": {"mean_seconds": 1.5, "runs": 1}}} + baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} + result = compare_baseline(report, baseline, tolerance=0.1) + self.assertFalse(result["stages"]["scene"]["regressed"]) + self.assertEqual(result["regressions"], []) + self.assertEqual(result["verdict"], "pass") + + +class CorpusSpecTests(unittest.TestCase): + def test_public_record_roundtrip(self): + spec = SyntheticCorpusSpec( + videos=2, duration_seconds=8.0, fps=24, + width=320, height=180, audio_mode="none", seed=42, + ) + record = spec.public_record() + self.assertEqual(record["videos"], 2) + self.assertEqual(record["duration_seconds"], 8.0) + self.assertEqual(record["audio_mode"], "none") + + def test_flite_mode_recorded(self): + spec = SyntheticCorpusSpec( + videos=1, duration_seconds=5.0, fps=30, + width=640, height=480, audio_mode="flite", seed=7, + ) + self.assertEqual(spec.public_record()["audio_mode"], "flite") From 09a2a6366ba57f8cf43428b74f66bf1f9ec0adf9 Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sun, 16 Aug 2026 13:40:32 +0500 Subject: [PATCH 2/3] Revert "feat(benchmarks): add reproducible indexing-latency benchmark with regression detection (#1)" This reverts commit 072d16fedf8d7d6ded39084075bbe34c77680444. --- docs/benchmarking/README.md | 1 - docs/benchmarking/performance.md | 152 ------- src/vidxp/benchmarks/cli.py | 132 ------ src/vidxp/benchmarks/latency.py | 703 ------------------------------- tests/test_benchmark_latency.py | 329 --------------- 5 files changed, 1317 deletions(-) delete mode 100644 docs/benchmarking/performance.md delete mode 100644 src/vidxp/benchmarks/latency.py delete mode 100644 tests/test_benchmark_latency.py diff --git a/docs/benchmarking/README.md b/docs/benchmarking/README.md index 8e2e38bc..cfe692fd 100644 --- a/docs/benchmarking/README.md +++ b/docs/benchmarking/README.md @@ -18,7 +18,6 @@ installation and product usage, start with the main | HiREST transcript localization | Legacy full result + current smoke | The legacy MiniLM stack scored all 193 validation pairs; current Qwen3 passed a two-video real execution smoke; 776 released test predictions remain unscored because their public bounds are placeholders | | LongVALE combined evaluation | Next | Build the visual-plus-speech adapter and validate one evaluation archive before scheduling the full run | | Actor clustering | Data-gated | The preferred BBT/Buffy evaluation still requires lawful access to the source episodes | -| Indexing latency benchmark | Ready | `vidxp benchmark index-latency` measures throughput, per-stage timings, and peak memory on synthetic FFmpeg media; supports regression detection against baselines | Read [current results](results.md) for the scores, plain-language metric definitions, honest comparisons, and the next benchmark decision. diff --git a/docs/benchmarking/performance.md b/docs/benchmarking/performance.md deleted file mode 100644 index 192c0bf1..00000000 --- a/docs/benchmarking/performance.md +++ /dev/null @@ -1,152 +0,0 @@ -# Latency benchmark protocol - -Status: Ready - -The latency benchmark (`vidxp benchmark index-latency`) measures indexing -throughput, per-stage timing, and peak memory using synthetic media generated -by FFmpeg on the caller's machine. It is designed for regression detection -between VidXP builds and for evaluating the latency impact of model or -architecture changes. - -## Protocol - -### Corpus generation - -The benchmark generates deterministic synthetic clips using FFmpeg's `lavfi` -source filters: - -| Parameter | Default | Notes | -|---|---|---| -| `--videos` | 1 | Number of synthetic clips | -| `--duration-seconds` | 8.0 | Wall-clock duration of each clip | -| `--fps` | 24 | Frame rate | -| `--resolution` | 320x180 | `WxH` format | -| `--audio-mode` | `none` | `none`, `sine`, or `flite` | -| `--input-mode` | `transcript` | `transcript` or `transcribe` | - -Video is generated via `testsrc2` (colour bars + timestamp). When -`--input-mode transcript` and `dialogue` is enabled, a deterministic -synthetic transcript (seeded PRNG over a fixed English vocabulary) is -supplied without real transcription. When `input-mode transcribe` is -used, `--audio-mode flite` must also be set and libflite must be -available in the ffmpeg build. - -### Indexing measurement - -Each repetition runs the full indexing pipeline via `run_index()` with -`reset=True`. The following stages are timed by the existing manifest -timing infrastructure (`core/manifest.py:record_stage`): - -| Stage | Modality | Measures | -|---|---|---| -| `frame_stream` | (all visual) | Decode throughput (frames/s) | -| `scene` | scene | SigLIP2 embedding (frames/s) | -| `actor` | actor | OpenCV detect + recognise (frames/s) | -| `visual_indexing` | all visual | Combined group wall time | -| `dialogue_indexing` | dialogue | Embedding throughput (phrases/s) | - -Peak RSS is captured via `resource.getrusage(RUSAGE_SELF).ru_maxrss` -(POSIX only; `None` on Windows, reported in bytes on macOS, KiB on -Linux). - -### Repetitions - -When `--repetitions N` > 1, each repetition runs the full cycle -(generate once, index each time after `reset`). Results are reported -as mean, min, and max across all per-video per-repetition samples. - -### Baseline comparison - -Pass `--baseline ` to compare the -current run against a prior report. For each stage present in both, -the delta ratio (`new_mean / old_mean - 1`) is computed. A stage with -a delta exceeding `--baseline-tolerance` (default 0.15 = 15%) is -flagged as a regression. The verdict is `fail` if any stage regressed, -else `pass`. - -### Output - -The benchmark writes its report to `run_directory/report.json` and -invokes `record_adapter_manifest` (embedding the corpus spec, device, -and result classification into the run's `manifest.json`). - -Report schema: - -```json -{ - "schema_version": 1, - "benchmark": "latency", - "run_id": "my-run", - "corpus": { "videos": 1, "duration_seconds": 8.0, ... }, - "modalities": ["scene", "actor"], - "device": "cpu", - "repetitions": 1, - "git": { "commit": "...", "dirty": false }, - "environment": { ... }, - "record_counts": { "scene": 8, "actor": 0 }, - "processed_frames": 8, - "stages": { - "scene": { - "runs": 1, "mean_seconds": 2.1, "min_seconds": 2.1, - "max_seconds": 2.1, "rate_per_second": 3.8 - } - }, - "summary": { - "wall_seconds": { "runs": 1, "mean_seconds": 5.0, ... }, - "peak_rss": { "unit": "bytes", "samples": 1, "value": 123456789 } - }, - "baseline": null | { "stages": {...}, "regressions": [...], "verdict": "pass" } -} -``` - -## Limitations - -- The synthetic video has no semantic scene content, so scene embeddings - are representative of throughput but not retrieval quality. -- Actors are not present in `testsrc2` video; `actor` stage measures - the per-frame face-detection overhead with zero detections. -- When `input_mode=transcript`, no real whisper transcription occurs; - dialogue embedding is measured on a synthetic transcript. -- True transcription latency (`input_mode=transcribe`) requires a - speech source (`--audio-mode flite`) and libflite in the FFmpeg - build; the generated speech is a short fixed sentence and does not - represent naturalistic conversation length or vocabulary. -- Peak RSS measures the whole-process peak, which includes Python - overhead, loaded models, and Chroma state; it is not a pure - indexing-stage measurement. - -## Usage - -```bash -# Default: single 8-second 320x180 clip, scene-only, 1 rep -vidxp benchmark index-latency --run-id my-baseline - -# Scene + actor + dialogue (synthetic transcript), 3 reps, compare with baseline -vidxp benchmark index-latency \ - --run-id v2-compare \ - --modalities scene,actor,dialogue \ - --videos 2 \ - --duration-seconds 12 \ - --repetitions 3 \ - --json \ - --baseline benchmark_runs/latency/synthetic/my-baseline/report.json - -# Real transcription (requires libflite in ffmpeg) -vidxp benchmark index-latency \ - --run-id transcribe-test \ - --modalities dialogue \ - --input-mode transcribe \ - --audio-mode flite \ - --device cpu -``` - -## Adding a new performance benchmark - -1. Define the corpus parameters and any new modality combinations in - the existing `run_latency` entry point. -2. Run the baseline and save its `report.json`. -3. Make your change (model swap, concurrency refactor, etc.). -4. Re-run with `--baseline ` and verify no - regressions. -5. Commit the baseline report to a designated location (e.g. - `docs/benchmarking/baselines/`) if it serves as a team reference. diff --git a/src/vidxp/benchmarks/cli.py b/src/vidxp/benchmarks/cli.py index e8f069dd..34fa5e2d 100644 --- a/src/vidxp/benchmarks/cli.py +++ b/src/vidxp/benchmarks/cli.py @@ -21,7 +21,6 @@ HIREST_DEFAULT_WINDOW_FRACTION, run_hirest, ) -from vidxp.benchmarks.latency import run_latency from vidxp.benchmarks.prepare import ( PreparationPlan, execute_preparation, @@ -575,134 +574,3 @@ def hirest_command( emit_json(metrics) else: rich_print(metrics) - - -@app.command("index-latency") -def index_latency_command( - ctx: typer.Context, - run_id: Annotated[str, typer.Option(help="Arbitrary label for this run.")], - modalities: Annotated[ - str, - typer.Option( - help="Comma-separated modality names: scene,actor,dialogue." - ), - ] = "scene", - videos: Annotated[ - int, - typer.Option(min=1, help="Number of synthetic clips to generate."), - ] = 1, - duration_seconds: Annotated[ - float, - typer.Option(min=0.1, help="Duration of each synthetic clip."), - ] = 8.0, - fps: Annotated[ - int, - typer.Option(min=1, help="Frame rate of synthetic clips."), - ] = 24, - resolution: Annotated[ - str, - typer.Option( - help="Synthetic clip resolution in WxH format (e.g. 320x180)." - ), - ] = "320x180", - repetitions: Annotated[ - int, - typer.Option(min=1, help="Number of times to repeat the run."), - ] = 1, - input_mode: Annotated[ - Literal["transcript", "transcribe"], - typer.Option( - help=( - "'transcript' supplies a synthetic transcript for dialogue " - "embedding (no real transcription). 'transcribe' runs " - "real whisper on audio (requires --audio-mode flite)." - ) - ), - ] = "transcript", - audio_mode: Annotated[ - Literal["none", "sine", "flite"], - typer.Option( - help=( - "Audio track for synthetic clips: 'none' (no audio), " - "'sine' (tone), or 'flite' (speech synthesis)." - ) - ), - ] = "none", - reset: Annotated[ - bool, - typer.Option(help="Clear any existing index before running."), - ] = False, - baseline: Annotated[ - Path | None, - typer.Option( - exists=True, - dir_okay=False, - help=( - "Path to a previous latency report JSON for regression " - "comparison." - ), - ), - ] = None, - baseline_tolerance: Annotated[ - float, - typer.Option( - min=0.0, - max=5.0, - help=( - "Relative regression tolerance. A stage mean slower by " - "more than this ratio flags as regression." - ), - ), - ] = 0.15, - json_output: Annotated[ - bool, - typer.Option("--json", help="Emit machine-readable JSON."), - ] = False, -) -> None: - """Run a reproducible indexing-latency benchmark on synthetic media.""" - - selected = [item.strip() for item in modalities.split(",") if item.strip()] - if not selected: - raise typer.BadParameter( - "At least one latency modality is required.", param_hint="--modalities" - ) - for modality in selected: - _require_benchmark_dependencies(modality) - - try: - parts = resolution.lower().split("x") - if len(parts) != 2: - raise ValueError - width, height = int(parts[0]), int(parts[1]) - if width <= 0 or height <= 0: - raise ValueError - except (IndexError, ValueError, AttributeError): - raise typer.BadParameter( - f"Invalid resolution: {resolution!r}. Use WxH, e.g. 320x180.", - param_hint="--resolution", - ) - - state = state_from_context(ctx) - report = run_latency( - run_id=run_id, - output_root=state.settings.data_dir / "benchmark_runs", - ffprobe=state.settings.ffprobe_executable, - ffmpeg=state.settings.ffmpeg_executable, - modalities=tuple(selected), - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - repetitions=repetitions, - input_mode=input_mode, - audio_mode=audio_mode, - device=state.settings.runtime_backend, - reset=reset, - baseline_path=baseline, - baseline_tolerance=baseline_tolerance, - ) - if effective_output_format(state, json_output) == OutputFormat.json: - emit_json(report) - else: - rich_print(report) diff --git a/src/vidxp/benchmarks/latency.py b/src/vidxp/benchmarks/latency.py deleted file mode 100644 index 87272f57..00000000 --- a/src/vidxp/benchmarks/latency.py +++ /dev/null @@ -1,703 +0,0 @@ -from __future__ import annotations - -import json -import random -import subprocess -from dataclasses import dataclass -from pathlib import Path -from statistics import mean -from time import perf_counter -from typing import Any, Literal, Mapping, Sequence - -from vidxp.benchmarks.common import ( - append_failure, - benchmark_generation_id, - benchmark_media_id, - ensure_adapter_outputs, - record_adapter_manifest, -) -from vidxp.capabilities.registry import create_capability_registry -from vidxp.core.contracts import IndexConfig, VideoSource -from vidxp.core.manifest import ManifestStore, write_json_atomic -from vidxp.core.runner import run_index -from vidxp.core.storage import IndexStorage -from vidxp.infrastructure.local_index import LOCAL_INDEX_RUNTIME_CHECKS -from vidxp.media_runtime import inspect_media_runtime -from vidxp.runtime import ModelRuntime -from vidxp.settings import VidXPSettings - - -LATENCY_BENCHMARK = "latency" -LATENCY_SPLIT = "synthetic" -LATENCY_SCHEMA_VERSION = 1 -DEFAULT_CORPUS_SEED = 2026 -SUPPORTED_MODALITIES = ("scene", "actor", "dialogue") - -_STAGE_RATES: Mapping[str, str] = { - "scene": "scene_frames", - "actor": "actor_frames", - "frame_stream": "source_frames_advanced", - "dialogue_indexing": "dialogue_phrases", -} - -_VOCABULARY = ( - "the quick brown fox jumps over the lazy dog honest sunshine light " - "morning river ocean mountain garden flower silver golden copper " - "bright shadow shadow candle lantern window door table chair book " - "letter number station market kitchen garden bakery camera video " - "music voice speech word phrase moment memory journey story world " - "quiet calm gentle peaceful vivid warm cool bright dark soft loud" -).split() - - -def _peak_rss_bytes() -> int | None: - try: - import resource - except ImportError: - return None - return int(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) - - -def rss_unit() -> Literal["bytes", "KiB"]: - return "bytes" if _sys_platform() == "darwin" else "KiB" - - -def _sys_platform() -> str: - import sys - - return sys.platform - - -@dataclass(frozen=True) -class SyntheticCorpusSpec: - videos: int - duration_seconds: float - fps: int - width: int - height: int - audio_mode: Literal["none", "sine", "flite"] - seed: int - - def public_record(self) -> dict[str, Any]: - return { - "videos": self.videos, - "duration_seconds": self.duration_seconds, - "fps": self.fps, - "width": self.width, - "height": self.height, - "audio_mode": self.audio_mode, - "seed": self.seed, - } - - -def validate_latency_options( - *, - modalities: Sequence[str], - videos: int, - duration_seconds: float, - fps: int, - width: int, - height: int, - repetitions: int, - input_mode: str, - audio_mode: str, - baseline_tolerance: float, -) -> tuple[str, ...]: - selected = tuple(dict.fromkeys(modalities)) - if not selected: - raise ValueError("At least one latency modality must be selected.") - unsupported = sorted(set(selected) - set(SUPPORTED_MODALITIES)) - if unsupported: - raise ValueError( - "Latency modalities must be a subset of " - + ", ".join(SUPPORTED_MODALITIES) - + "; unsupported: " - + ", ".join(unsupported) - ) - if videos <= 0: - raise ValueError("videos must be greater than zero.") - if duration_seconds <= 0: - raise ValueError("duration_seconds must be greater than zero.") - if fps <= 0: - raise ValueError("fps must be greater than zero.") - if width <= 0 or height <= 0: - raise ValueError("width and height must be greater than zero.") - if repetitions <= 0: - raise ValueError("repetitions must be greater than zero.") - if input_mode not in {"transcript", "transcribe"}: - raise ValueError("input_mode must be 'transcript' or 'transcribe'.") - if audio_mode not in {"none", "sine", "flite"}: - raise ValueError("audio_mode must be 'none', 'sine', or 'flite'.") - if "dialogue" in selected and input_mode == "transcribe": - if audio_mode != "flite": - raise ValueError( - "Real transcription requires a speech audio source; " - "use --audio-mode flite with --input-mode transcribe." - ) - if not 0 <= baseline_tolerance <= 5: - raise ValueError("baseline_tolerance must be between zero and five.") - return selected - - -def synthetic_transcript( - *, - duration_seconds: float, - seed: int, -) -> list[dict[str, Any]]: - generator = random.Random(seed) - strides = max(1, int(duration_seconds / 0.4)) - words = [generator.choice(_VOCABULARY) for _ in range(strides)] - span = duration_seconds / len(words) - word_events = [ - { - "word": word, - "start": round(index * span, 4), - "end": round((index + 1) * span, 4), - } - for index, word in enumerate(words) - ] - return [ - { - "text": " ".join(words), - "start": 0.0, - "end": duration_seconds, - "words": word_events, - } - ] - - -def _flite_text(seed: int) -> str: - generator = random.Random(seed) - words = [generator.choice(_VOCABULARY) for _ in range(24)] - return " ".join(words) - - -def build_clip_command( - *, - spec: SyntheticCorpusSpec, - ffmpeg: str, - destination: Path, -) -> list[str]: - compact = spec.width != 0 and spec.height != 0 - if not compact: - raise ValueError("The synthetic corpus requires positive dimensions.") - command = [ - ffmpeg, - "-y", - "-v", - "error", - "-f", - "lavfi", - "-i", - f"testsrc2=size={spec.width}x{spec.height}:rate={spec.fps}", - ] - if spec.audio_mode == "sine": - command += [ - "-f", - "lavfi", - "-i", - "sine=frequency=440:sample_rate=16000", - ] - elif spec.audio_mode == "flite": - command += [ - "-f", - "lavfi", - "-i", - f"flite=text='{_flite_text(spec.seed)}',sample_rate=16000", - ] - command += [ - "-t", - f"{spec.duration_seconds:g}", - "-c:v", - "libx264", - "-pix_fmt", - "yuv420p", - ] - if spec.audio_mode == "none": - command.append("-an") - else: - command += ["-c:a", "aac"] - command.append(str(destination)) - return command - - -def _probe_duration(ffprobe: str, path: Path) -> float: - completed = subprocess.run( - [ - ffprobe, - "-v", - "error", - "-show_entries", - "format=duration", - "-of", - "json", - str(path), - ], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=60, - ) - if completed.returncode != 0: - raise ValueError( - f"ffprobe could not read a generated clip: {path}" - ) - try: - duration = float(json.loads(completed.stdout)["format"]["duration"]) - except (KeyError, ValueError, TypeError, json.JSONDecodeError) as exc: - raise ValueError( - f"ffprobe returned an invalid duration for {path}." - ) from exc - if duration <= 0: - raise ValueError(f"ffprobe reported a non-positive duration for {path}.") - return duration - - -def generate_synthetic_corpus( - *, - spec: SyntheticCorpusSpec, - directory: str | Path, - ffprobe: str, - ffmpeg: str, - audio_mode: str | None = None, -) -> list[Path]: - runtime_status = inspect_media_runtime( - ffprobe=ffprobe, - ffmpeg=ffmpeg, - ) - if not runtime_status.ready: - raise ValueError( - "The latency benchmark requires FFmpeg and ffprobe to generate " - "the synthetic corpus. Run `vidxp init`, then retry." - ) - destination = Path(directory) - destination.mkdir(parents=True, exist_ok=True) - clips = [] - for index in range(spec.videos): - path = destination / f"clip-{index:03d}.mp4" - command = build_clip_command( - spec=spec, - ffmpeg=ffmpeg, - destination=path, - ) - completed = subprocess.run( - command, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=600, - ) - if completed.returncode != 0: - stderr = (completed.stderr or "").strip() - if spec.audio_mode == "flite" and stderr: - raise ValueError( - "FFmpeg could not apply the flite speech filter " - f"(libflite likely unavailable): {stderr}" - ) - raise ValueError(f"FFmpeg could not generate {path}: {stderr}") - if not path.is_file() or path.stat().st_size == 0: - raise ValueError(f"FFmpeg did not produce {path}.") - _probe_duration(ffprobe, path) - clips.append(path) - return clips - - -def _per_video_stages( - manifest: Mapping[str, Any], - video_id: str, -) -> dict[str, float]: - video = manifest["videos"].get(video_id) - if video is None or video.get("state") == "failed": - return {} - return { - str(stage): float(entry["seconds"]) - for stage, entry in (video.get("stages") or {}).items() - if entry.get("state") != "incomplete" - and float(entry.get("seconds", 0.0)) > 0 - } - - -def _summary_ratio( - manifest: Mapping[str, Any], - video_id: str, - *, - metric: str, - stage: str, -) -> float | None: - video = manifest["videos"].get(video_id) or {} - summary = video.get("summary") or {} - seconds = _per_video_stages(manifest, video_id).get(stage) - count = summary.get(metric) - if seconds is None or count is None or seconds <= 0 or count <= 0: - return None - return float(count) / seconds - - -def aggregate_latency_runs( - manifests: Sequence[Mapping[str, Any]], - *, - wall_seconds: Sequence[float], - peak_rss_samples: Sequence[int | None], -) -> dict[str, Any]: - if len(manifests) != len(wall_seconds): - raise ValueError( - "Every latency repetition requires a wall-clock sample." - ) - stage_samples: dict[str, list[float]] = {} - rate_samples: dict[str, list[float]] = {} - per_video: list[dict[str, Any]] = [] - processed_frames = 0 - record_counts: dict[str, int] = {} - for repetition, manifest in enumerate(manifests): - processed_frames += int(manifest.get("processed_frames", 0)) - for modality, count in (manifest.get("record_counts") or {}).items(): - record_counts[modality] = record_counts.get(modality, 0) + int(count) - for video_id in sorted(manifest.get("videos", {})): - stages = _per_video_stages(manifest, video_id) - if not stages: - continue - for stage, seconds in stages.items(): - stage_samples.setdefault(stage, []).append(seconds) - rate_stages = { - stage: _summary_ratio( - manifest, - video_id, - metric=_STAGE_RATES[stage], - stage=stage, - ) - for stage in _STAGE_RATES - if stage in stages - } - for stage, rate in rate_stages.items(): - if rate is None: - continue - rate_samples.setdefault(stage, []).append(rate) - video = manifest["videos"].get(video_id, {}) - per_video.append( - { - "repetition": repetition, - "video_id": video_id, - "wall_seconds": ( - wall_seconds[repetition] - ), - "stages": dict(sorted(stages.items())), - "summary": video.get("summary", {}), - } - ) - stages: dict[str, dict[str, Any]] = {} - for stage, samples in stage_samples.items(): - values = sorted(samples) - summary: dict[str, Any] = { - "runs": len(values), - "mean_seconds": mean(values), - "min_seconds": values[0], - "max_seconds": values[-1], - } - rates = rate_samples.get(stage) - if rates: - summary["rate_per_second"] = mean(rates) - stages[stage] = summary - approximate_rss = [ - sample for sample in peak_rss_samples if sample is not None - ] - summary = { - "wall_seconds": { - "runs": len(wall_seconds), - "mean_seconds": mean(wall_seconds), - "min_seconds": min(wall_seconds), - "max_seconds": max(wall_seconds), - }, - "peak_rss": { - "unit": rss_unit(), - "samples": len(approximate_rss), - "value": int(max(approximate_rss)) if approximate_rss else None, - }, - } - return { - "per_video": per_video, - "stages": dict(sorted(stages.items())), - "summary": summary, - "processed_frames": processed_frames, - "record_counts": dict(sorted(record_counts.items())), - } - - -def _validate_baseline_compatibility( - report: Mapping[str, Any], - baseline: Mapping[str, Any], -) -> None: - baseline_corpus = baseline.get("corpus") or {} - report_corpus = report.get("corpus") or {} - corpus_keys = ("videos", "duration_seconds", "fps", "width", "height", "audio_mode", "seed") - for key in corpus_keys: - report_val = report_corpus.get(key) - baseline_val = baseline_corpus.get(key) - if report_val is not None and baseline_val is not None and report_val != baseline_val: - raise ValueError( - f"Baseline corpus mismatch: {key!r} is {baseline_val!r}, " - f"current is {report_val!r}." - ) - for key in ("input_mode", "device"): - report_val = report.get(key) - baseline_val = baseline.get(key) - if report_val is not None and baseline_val is not None and report_val != baseline_val: - raise ValueError( - f"Baseline {key!r} mismatch: {baseline_val!r}, current is {report_val!r}." - ) - report_mods = sorted(report.get("modalities") or []) - baseline_mods = sorted(baseline.get("modalities") or []) - if report_mods and baseline_mods and report_mods != baseline_mods: - raise ValueError( - f"Baseline modalities mismatch: {baseline_mods}, current is {report_mods}." - ) - - -def compare_baseline( - report: Mapping[str, Any], - baseline: Mapping[str, Any], - *, - tolerance: float, -) -> dict[str, Any]: - if tolerance < 0: - raise ValueError("Baseline tolerance must be nonnegative.") - comparisons: dict[str, dict[str, Any]] = {} - regressions: list[str] = [] - for stage, previous in (baseline.get("stages") or {}).items(): - current = (report.get("stages") or {}).get(stage) - if current is None or not previous.get("runs"): - continue - old_mean = float(previous["mean_seconds"]) - new_mean = float(current["mean_seconds"]) - if old_mean <= 0: - continue - delta = new_mean / old_mean - 1.0 - comparisons[stage] = { - "old_mean_seconds": old_mean, - "new_mean_seconds": new_mean, - "delta_ratio": delta, - "regressed": delta > tolerance, - } - if delta > tolerance: - regressions.append(stage) - return { - "schema_version": LATENCY_SCHEMA_VERSION, - "tolerance": tolerance, - "stages": dict(sorted(comparisons.items())), - "regressions": regressions, - "verdict": "fail" if regressions else "pass", - } - - -def build_latency_sources( - *, - clips: Sequence[Path], - spec: SyntheticCorpusSpec, - input_mode: Literal["transcript", "transcribe"], -) -> list[VideoSource]: - sources = [] - for index, path in enumerate(clips): - transcript = ( - synthetic_transcript( - duration_seconds=spec.duration_seconds, - seed=spec.seed + index, - ) - if input_mode == "transcript" - else None - ) - sources.append( - VideoSource( - video_id=benchmark_media_id( - LATENCY_BENCHMARK, - f"clip-{index:03d}", - ), - path=path, - source_name=f"clip-{index:03d}.mp4", - transcript=transcript, - ) - ) - return sources - - -def run_latency( - *, - run_id: str, - output_root: str | Path = "benchmark_runs", - ffprobe: str = "ffprobe", - ffmpeg: str = "ffmpeg", - modalities: Sequence[str] = ("scene",), - videos: int = 1, - duration_seconds: float = 8.0, - fps: int = 24, - width: int = 320, - height: int = 180, - repetitions: int = 1, - input_mode: Literal["transcript", "transcribe"] = "transcript", - audio_mode: Literal["none", "sine", "flite"] = "none", - device: str = "cpu", - reset: bool = False, - baseline_path: str | Path | None = None, - baseline_tolerance: float = 0.15, -) -> dict[str, Any]: - selected = validate_latency_options( - modalities=modalities, - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - repetitions=repetitions, - input_mode=input_mode, - audio_mode=audio_mode, - baseline_tolerance=baseline_tolerance, - ) - spec = SyntheticCorpusSpec( - videos=videos, - duration_seconds=duration_seconds, - fps=fps, - width=width, - height=height, - audio_mode=audio_mode, - seed=DEFAULT_CORPUS_SEED, - ) - config = IndexConfig( - dataset=LATENCY_BENCHMARK, - split=LATENCY_SPLIT, - run_id=run_id, - enabled_modalities=selected, - device=device, - output_root=output_root, - generation_id=benchmark_generation_id( - LATENCY_BENCHMARK, - LATENCY_SPLIT, - run_id, - ), - ) - run_directory = config.run_directory - registry = create_capability_registry( - platform_runtime_checks=LOCAL_INDEX_RUNTIME_CHECKS - ) - runtime = ModelRuntime( - VidXPSettings( - repository_root=run_directory, - runtime_backend=device, - ), - allowed_specs=registry.model_specs(), - ) - ensure_adapter_outputs(run_directory) - manifests: list[dict[str, Any]] = [] - wall_samples: list[float] = [] - rss_samples: list[int | None] = [] - try: - clips = generate_synthetic_corpus( - spec=spec, - directory=run_directory / "corpus", - ffprobe=ffprobe, - ffmpeg=ffmpeg, - ) - sources = build_latency_sources( - clips=clips, - spec=spec, - input_mode=input_mode, - ) - for _ in range(repetitions): - started = perf_counter() - with IndexStorage(config) as storage: - manifest = run_index( - sources, - config, - reset=reset, - storage=storage, - manifest_store=ManifestStore( - config, - registry=registry, - runtime=runtime, - ), - registry=registry, - runtime=runtime, - ) - store_size_bytes = storage.size_bytes() - wall_samples.append(perf_counter() - started) - rss_samples.append(_peak_rss_bytes()) - manifests.append( - { - **manifest, - "store_size_bytes_at_commit": store_size_bytes, - } - ) - aggregated = aggregate_latency_runs( - manifests, - wall_seconds=wall_samples, - peak_rss_samples=rss_samples, - ) - report = { - "schema_version": LATENCY_SCHEMA_VERSION, - "benchmark": LATENCY_BENCHMARK, - "run_id": run_id, - "created_at": manifests[-1].get("completed_at"), - "corpus": spec.public_record(), - "input_mode": input_mode, - "modalities": list(selected), - "device": device, - "repetitions": repetitions, - "git": manifests[-1].get("git"), - "environment": manifests[-1].get("environment"), - "config_fingerprint": manifests[-1].get("config_fingerprint"), - "record_counts": aggregated["record_counts"], - "processed_frames": aggregated["processed_frames"], - "summary": aggregated["summary"], - "stages": aggregated["stages"], - "per_video": aggregated["per_video"], - "baseline": None, - } - if baseline_path is not None: - try: - baseline = json.loads( - Path(baseline_path).read_text(encoding="utf-8") - ) - except (OSError, json.JSONDecodeError) as exc: - raise ValueError( - f"Baseline report is not readable JSON: {baseline_path}" - ) from exc - _validate_baseline_compatibility(report, baseline) - report["baseline"] = compare_baseline( - report, - baseline, - tolerance=baseline_tolerance, - ) - write_json_atomic(run_directory / "report.json", report) - record_adapter_manifest( - run_directory, - benchmark=LATENCY_BENCHMARK, - subset={ - "label": f"latency_{run_id}", - "modalities": list(selected), - "video_count": videos, - "duration_seconds": duration_seconds, - "repetitions": repetitions, - }, - artifacts=[], - state="complete", - details={ - "device": device, - "input_mode": input_mode, - "audio_mode": audio_mode, - "corpus": spec.public_record(), - "result_classification": "performance_benchmark_not_quality_score", - }, - ) - return report - except BaseException as error: - append_failure(run_directory, stage="latency_adapter", error=error) - record_adapter_manifest( - run_directory, - benchmark=LATENCY_BENCHMARK, - subset={ - "label": f"latency_{run_id}", - "modalities": list(selected), - }, - artifacts=[], - state="failed", - ) - raise \ No newline at end of file diff --git a/tests/test_benchmark_latency.py b/tests/test_benchmark_latency.py deleted file mode 100644 index 6cbd7057..00000000 --- a/tests/test_benchmark_latency.py +++ /dev/null @@ -1,329 +0,0 @@ -from __future__ import annotations - -import unittest -from pathlib import Path - -from vidxp.benchmarks.latency import ( - SyntheticCorpusSpec, - aggregate_latency_runs, - build_clip_command, - build_latency_sources, - compare_baseline, - synthetic_transcript, - validate_latency_options, -) - - -class LatencyValidationTests(unittest.TestCase): - def test_validates_default_options(self): - selected = validate_latency_options( - modalities=("scene",), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=3, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("scene",)) - - def test_rejects_empty_modalities(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=(), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - - def test_rejects_unsupported_modality(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=("scene", "ocr"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - - def test_rejects_transcribe_without_flite(self): - with self.assertRaises(ValueError): - validate_latency_options( - modalities=("scene", "dialogue"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcribe", - audio_mode="sine", - baseline_tolerance=0.15, - ) - - def test_accepts_transcribe_with_flite(self): - selected = validate_latency_options( - modalities=("dialogue",), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcribe", - audio_mode="flite", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("dialogue",)) - - def test_deduplicates_modalities(self): - selected = validate_latency_options( - modalities=("scene", "scene", "actor"), - videos=1, - duration_seconds=8.0, - fps=24, - width=320, - height=180, - repetitions=1, - input_mode="transcript", - audio_mode="none", - baseline_tolerance=0.15, - ) - self.assertEqual(selected, ("scene", "actor")) - - -class SyntheticTranscriptTests(unittest.TestCase): - def test_returns_one_segment_with_words(self): - transcript = synthetic_transcript(duration_seconds=10.0, seed=42) - self.assertEqual(len(transcript), 1) - segment = transcript[0] - self.assertGreater(len(segment["text"]), 0) - self.assertEqual(segment["start"], 0.0) - self.assertGreater(segment["end"], 0.0) - self.assertGreater(len(segment["words"]), 0) - for word in segment["words"]: - self.assertIn("word", word) - self.assertIsInstance(word["start"], float) - self.assertIsInstance(word["end"], float) - - def test_deterministic_across_calls(self): - first = synthetic_transcript(duration_seconds=5.0, seed=99) - second = synthetic_transcript(duration_seconds=5.0, seed=99) - self.assertEqual(first, second) - - def test_different_seeds_differ(self): - first = synthetic_transcript(duration_seconds=5.0, seed=99) - second = synthetic_transcript(duration_seconds=5.0, seed=100) - self.assertNotEqual(first, second) - - -class BuildClipCommandTests(unittest.TestCase): - def test_no_audio_default(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - self.assertIn("testsrc2=size=320x180:rate=24", command) - self.assertIn("-an", command) - self.assertNotIn("-c:a", command) - - def test_sine_audio_adds_aac(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="sine", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - self.assertIn("sine=frequency=440:sample_rate=16000", command) - self.assertIn("-c:a", command) - self.assertNotIn("-an", command) - - def test_flite_audio_contains_filter_ref(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="flite", seed=42, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("out.mp4")) - flite_args = [arg for arg in command if "flite=text=" in arg] - self.assertEqual(len(flite_args), 1) - - def test_duration_is_formatted(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=3.5, fps=30, - width=640, height=480, audio_mode="none", seed=0, - ) - command = build_clip_command(spec=spec, ffmpeg="ffmpeg", destination=Path("clip.mp4")) - idx = command.index("-t") - self.assertEqual(command[idx + 1], "3.5") - self.assertIn("testsrc2=size=640x480:rate=30", command) - - -class BuildSourcesTests(unittest.TestCase): - def test_transcript_attached_in_input_mode(self): - spec = SyntheticCorpusSpec( - videos=2, duration_seconds=4.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - clips = [Path(f"{i}.mp4") for i in range(2)] - sources = build_latency_sources(clips=clips, spec=spec, input_mode="transcript") - self.assertEqual(len(sources), 2) - for index, source in enumerate(sources): - self.assertIsNotNone(source.transcript) - self.assertIsNotNone(source.path) - self.assertIsNotNone(source.video_id) - self.assertEqual(source.source_name, f"clip-{index:03d}.mp4") - - def test_no_transcript_in_transcribe_mode(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=4.0, fps=24, - width=320, height=180, audio_mode="flite", seed=42, - ) - sources = build_latency_sources(clips=[Path("0.mp4")], spec=spec, input_mode="transcribe") - for source in sources: - self.assertIsNone(source.transcript) - - -class AggregateMetricsTests(unittest.TestCase): - def _sample_manifest(self, scene_seconds, scene_frames, actor_seconds, actor_frames): - return { - "processed_frames": scene_frames, - "record_counts": {"scene": scene_frames, "actor": actor_frames}, - "git": {"commit": "abc", "dirty": False}, - "environment": {"platform": "test"}, - "config_fingerprint": "fp1", - "completed_at": "2026-01-01T00:00:00", - "videos": { - "vid-1": { - "state": "complete", - "summary": { - "scene_frames": scene_frames, - "actor_frames": actor_frames, - "source_frames_advanced": scene_frames + 100, - }, - "stages": { - "scene": {"seconds": scene_seconds, "state": ""}, - "actor": {"seconds": actor_seconds, "state": ""}, - "frame_stream": {"seconds": 0.5, "state": ""}, - }, - } - }, - } - - def test_aggregates_single_manifest(self): - result = aggregate_latency_runs( - [self._sample_manifest(2.0, 8, 1.5, 3)], - wall_seconds=[3.5], - peak_rss_samples=[100000], - ) - self.assertEqual(result["processed_frames"], 8) - self.assertEqual(result["record_counts"], {"actor": 3, "scene": 8}) - self.assertIn("scene", result["stages"]) - self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.0) - self.assertAlmostEqual(result["stages"]["scene"]["rate_per_second"], 4.0) - self.assertAlmostEqual(result["stages"]["actor"]["mean_seconds"], 1.5) - self.assertAlmostEqual(result["summary"]["wall_seconds"]["mean_seconds"], 3.5) - - def test_aggregates_multiple_manifests(self): - m1 = self._sample_manifest(2.0, 8, 1.5, 3) - m2 = self._sample_manifest(2.5, 10, 2.0, 4) - result = aggregate_latency_runs( - [m1, m2], - wall_seconds=[3.5, 4.5], - peak_rss_samples=[100000, 120000], - ) - self.assertEqual(result["processed_frames"], 18) - self.assertAlmostEqual(result["stages"]["scene"]["mean_seconds"], 2.25) - self.assertAlmostEqual(result["stages"]["scene"]["min_seconds"], 2.0) - self.assertAlmostEqual(result["stages"]["scene"]["max_seconds"], 2.5) - self.assertEqual(len(result["per_video"]), 2) - self.assertAlmostEqual( - result["summary"]["wall_seconds"]["mean_seconds"], - 4.0, - ) - - def test_skips_failed_videos(self): - manifest = { - "processed_frames": 0, - "record_counts": {}, - "git": {}, - "environment": {}, - "config_fingerprint": "fp", - "completed_at": "", - "videos": { - "vid-1": { - "state": "failed", - "summary": {}, - "stages": {}, - } - }, - } - result = aggregate_latency_runs( - [manifest], - wall_seconds=[1.0], - peak_rss_samples=[None], - ) - self.assertEqual(result["processed_frames"], 0) - self.assertEqual(result["stages"], {}) - - -class CompareBaselineTests(unittest.TestCase): - def test_no_baseline_stages_returns_empty(self): - report = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - baseline = {"stages": {}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertEqual(result["stages"], {}) - self.assertEqual(result["regressions"], []) - self.assertEqual(result["verdict"], "pass") - - def test_regression_detected(self): - report = {"stages": {"scene": {"mean_seconds": 3.0, "runs": 1}}} - baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertIn("scene", result["stages"]) - self.assertAlmostEqual( - result["stages"]["scene"]["delta_ratio"], 0.5 - ) - self.assertTrue(result["stages"]["scene"]["regressed"]) - self.assertEqual(result["regressions"], ["scene"]) - self.assertEqual(result["verdict"], "fail") - - def test_improvement_not_regression(self): - report = {"stages": {"scene": {"mean_seconds": 1.5, "runs": 1}}} - baseline = {"stages": {"scene": {"mean_seconds": 2.0, "runs": 1}}} - result = compare_baseline(report, baseline, tolerance=0.1) - self.assertFalse(result["stages"]["scene"]["regressed"]) - self.assertEqual(result["regressions"], []) - self.assertEqual(result["verdict"], "pass") - - -class CorpusSpecTests(unittest.TestCase): - def test_public_record_roundtrip(self): - spec = SyntheticCorpusSpec( - videos=2, duration_seconds=8.0, fps=24, - width=320, height=180, audio_mode="none", seed=42, - ) - record = spec.public_record() - self.assertEqual(record["videos"], 2) - self.assertEqual(record["duration_seconds"], 8.0) - self.assertEqual(record["audio_mode"], "none") - - def test_flite_mode_recorded(self): - spec = SyntheticCorpusSpec( - videos=1, duration_seconds=5.0, fps=30, - width=640, height=480, audio_mode="flite", seed=7, - ) - self.assertEqual(spec.public_record()["audio_mode"], "flite") From 3257c99b7e5ccb37b745eb3587e4e16cabf8f7eb Mon Sep 17 00:00:00 2001 From: Mahnoor-Zaffar <1999mahnoor@gmail.com> Date: Sun, 20 Sep 2026 17:42:20 +0500 Subject: [PATCH 3/3] fix: allow clear to recover from incompatible index schemas Clear and reindex both validated the active snapshot before proceeding, so an older index_schema_version manifest raised IndexSchemaError and blocked recovery. Index clear now reads the active snapshot without per-generation manifest validation and also discards the Chroma collections so a rebuild recreates them with the current embedding dimensions. Source media is preserved; generated index data is removed. Internal-only; no automatic migration of old indexes. --- src/vidxp/cli_commands/index.py | 13 +- src/vidxp/infrastructure/local_index.py | 36 +++- src/vidxp/infrastructure/local_snapshots.py | 23 ++- src/vidxp/infrastructure/sql_snapshots.py | 19 ++- tests/test_local_snapshots.py | 180 ++++++++++++++++++-- 5 files changed, 248 insertions(+), 23 deletions(-) diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index d51a1593..14fee847 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -478,12 +478,14 @@ def index_clear( typer.Option("--json", help="Emit machine-readable JSON."), ] = False, ) -> None: - """Publish an empty active snapshot without deleting retained generations.""" + """Remove generated index data and publish an empty active snapshot.""" state = state_from_context(ctx) if not yes: typer.confirm( - f"Clear the active index at {state.service.index_directory}?", + f"Clear the active index at {state.service.index_directory}? " + "This removes generated index data and vector collections; " + "imported source media is preserved.", abort=True, ) cleared = state.service.clear_index() @@ -493,4 +495,9 @@ def index_clear( if effective_output_format(state, json_output) == OutputFormat.json: emit_json(payload) else: - typer.echo("Index cleared." if cleared else "No index was found.") + typer.echo( + "Index cleared. Generated index data and vector collections " + "were removed; imported source media is preserved." + if cleared + else "No index was found." + ) diff --git a/src/vidxp/infrastructure/local_index.py b/src/vidxp/infrastructure/local_index.py index 982e4199..3fcc4cda 100644 --- a/src/vidxp/infrastructure/local_index.py +++ b/src/vidxp/infrastructure/local_index.py @@ -582,4 +582,38 @@ def clear(self, config: IndexConfig) -> bool: self._require_index_directory(config.index_directory) repository = self.repository with repository.lease(): - return repository.clear() + cleared = repository.clear() + self._discard_vector_collections( + repository, + config, + client_factory=self.chroma_clients, + ) + return cleared + + def _discard_vector_collections( + self, + repository: LocalSnapshotRepository, + config: IndexConfig, + *, + client_factory: ChromaClientFactory | None = None, + ) -> None: + clients = client_factory or ChromaClientFactory() + cleanup_config = replace( + config, + storage_directory=repository.store, + generation_directory=None, + video_id=None, + generation_id=None, + snapshot_id=None, + snapshot_sha256=None, + ) + if clients.remote or repository.store.is_dir(): + try: + with IndexStorage( + cleanup_config, + create=False, + client_factory=clients, + ) as storage: + storage.clear() + except FileNotFoundError: + pass diff --git a/src/vidxp/infrastructure/local_snapshots.py b/src/vidxp/infrastructure/local_snapshots.py index e364f27f..8c2221ad 100644 --- a/src/vidxp/infrastructure/local_snapshots.py +++ b/src/vidxp/infrastructure/local_snapshots.py @@ -116,14 +116,23 @@ def generation_directory(self, generation_id: str) -> Path: def _snapshot_path(self, snapshot_id: str) -> Path: return self.snapshots / f"{snapshot_id}.json" - def read_active(self, *, required: bool = False) -> IndexSnapshot | None: - resolved = self._read_active(required=required) + def read_active( + self, + *, + required: bool = False, + validate_generations: bool = True, + ) -> IndexSnapshot | None: + resolved = self._read_active( + required=required, + validate_generations=validate_generations, + ) return None if resolved is None else resolved[1] def _read_active( self, *, required: bool = False, + validate_generations: bool = True, ) -> tuple[ActiveSnapshotPointer, IndexSnapshot] | None: if not self.active_pointer.is_file(): if required: @@ -138,6 +147,7 @@ def _read_active( snapshot = self.read_snapshot( pointer.snapshot_id, expected_sha256=pointer.snapshot_sha256, + validate_generations=validate_generations, ) return pointer, snapshot except IndexSchemaError: @@ -152,6 +162,7 @@ def read_snapshot( snapshot_id: str, *, expected_sha256: str | None = None, + validate_generations: bool = True, ) -> IndexSnapshot: snapshot_path = self._snapshot_path(snapshot_id) if not snapshot_path.is_file(): @@ -177,7 +188,8 @@ def read_snapshot( raise IndexSchemaError( "The snapshot filename and document identifier differ." ) - self._validate_generations(snapshot) + if validate_generations: + self._validate_generations(snapshot) return snapshot def _validate_generations(self, snapshot: IndexSnapshot) -> None: @@ -338,9 +350,10 @@ def remove(self, media_id: str) -> bool: return True def clear(self) -> bool: - active = self.read_active() - if active is None or not active.generations: + resolved = self._read_active(validate_generations=False) + if resolved is None or not resolved[1].generations: return False + active = resolved[1] self._publish( generations={}, config_fingerprint=active.config_fingerprint, diff --git a/src/vidxp/infrastructure/sql_snapshots.py b/src/vidxp/infrastructure/sql_snapshots.py index 43a19473..4dc7dfd9 100644 --- a/src/vidxp/infrastructure/sql_snapshots.py +++ b/src/vidxp/infrastructure/sql_snapshots.py @@ -104,7 +104,12 @@ def _ensure_index_state(connection: Connection) -> None: except IntegrityError: pass - def read_active(self, *, required: bool = False) -> IndexSnapshot | None: + def read_active( + self, + *, + required: bool = False, + validate_generations: bool = True, + ) -> IndexSnapshot | None: with self.engine.connect() as connection: row = connection.execute( select( @@ -122,6 +127,7 @@ def read_active(self, *, required: bool = False) -> IndexSnapshot | None: connection, row.active_snapshot_id, expected_sha256=row.active_snapshot_sha256, + validate_generations=validate_generations, ) def read_snapshot( @@ -129,12 +135,14 @@ def read_snapshot( snapshot_id: str, *, expected_sha256: str | None = None, + validate_generations: bool = True, ) -> IndexSnapshot: with self.engine.connect() as connection: return self._read_snapshot( connection, snapshot_id, expected_sha256=expected_sha256, + validate_generations=validate_generations, ) def _read_snapshot( @@ -143,6 +151,7 @@ def _read_snapshot( snapshot_id: str, *, expected_sha256: str | None, + validate_generations: bool = True, ) -> IndexSnapshot: row = connection.execute( select( @@ -166,7 +175,8 @@ def _read_snapshot( raise IndexSchemaError( f"Index snapshot {snapshot_id} failed integrity validation." ) - self._validate_generations(snapshot) + if validate_generations: + self._validate_generations(snapshot) return snapshot def validate_generation( @@ -247,7 +257,7 @@ def remove(self, media_id: str) -> bool: return True def clear(self) -> bool: - active = self.read_active() + active = self.read_active(validate_generations=False) if active is None or not active.generations: return False self._publish( @@ -255,6 +265,7 @@ def clear(self) -> bool: remove_media_id="*", config_fingerprint=active.config_fingerprint, configuration=dict(active.configuration), + validate_generations=False, ) return True @@ -265,6 +276,7 @@ def _publish( remove_media_id: str | None, config_fingerprint: str, configuration: dict[str, Any], + validate_generations: bool = True, ) -> IndexSnapshot: with self.engine.begin() as connection: self._ensure_index_state(connection) @@ -283,6 +295,7 @@ def _publish( connection, state.active_snapshot_id, expected_sha256=state.active_snapshot_sha256, + validate_generations=validate_generations, ) ) generations = dict(active.generations) if active is not None else {} diff --git a/tests/test_local_snapshots.py b/tests/test_local_snapshots.py index 091060c8..ce198e47 100644 --- a/tests/test_local_snapshots.py +++ b/tests/test_local_snapshots.py @@ -16,8 +16,9 @@ IndexSchemaError, StorageRecord, ) +from vidxp.core.manifest import MANIFEST_FILE, sha256_file, write_json_atomic +from vidxp.core.snapshots import GenerationReference from vidxp.core.storage import IndexStorage -from vidxp.core.manifest import MANIFEST_FILE, write_json_atomic from vidxp.infrastructure.local_snapshots import LocalSnapshotRepository from vidxp.infrastructure.local_index import LocalIndexBackend @@ -66,11 +67,95 @@ def write_generation( input_sha: str, record_counts: dict[str, int] | None = None, store_size_bytes_at_commit: int | None = 123, + ): + manifest = self._manifest_payload( + config, + media_id=media_id, + input_sha=input_sha, + record_counts=record_counts, + store_size_bytes_at_commit=store_size_bytes_at_commit, + schema_version=INDEX_SCHEMA_VERSION, + ) + write_json_atomic( + config.run_directory / MANIFEST_FILE, + manifest, + ) + return self.repository.generation_reference( + generation_id=str(config.generation_id), + media_id=media_id, + ) + + def legacy_generation( + self, + media_id: str, + *, + input_sha: str, + schema_version: int, + ): + generation_id = self.repository.new_generation_id() + config = replace( + self.config, + video_id=media_id, + generation_id=generation_id, + generation_directory=self.repository.generation_directory( + generation_id + ), + ) + return config, self.write_legacy_generation( + config, + media_id=media_id, + input_sha=input_sha, + schema_version=schema_version, + ) + + def write_legacy_generation( + self, + config: IndexConfig, + *, + media_id: str, + input_sha: str, + schema_version: int, + record_counts: dict[str, int] | None = None, + store_size_bytes_at_commit: int | None = 123, + ): + manifest = self._manifest_payload( + config, + media_id=media_id, + input_sha=input_sha, + record_counts=record_counts, + store_size_bytes_at_commit=store_size_bytes_at_commit, + schema_version=schema_version, + ) + manifest_path = config.run_directory / MANIFEST_FILE + write_json_atomic(manifest_path, manifest) + return GenerationReference( + generation_id=str(config.generation_id), + media_id=media_id, + manifest_sha256=sha256_file(manifest_path), + input_sha256=input_sha, + config_fingerprint=config.fingerprint(), + modalities=tuple(config.enabled_modalities), + record_counts={ + modality: manifest["record_counts"][modality] + for modality in config.enabled_modalities + }, + store_size_bytes_at_commit=store_size_bytes_at_commit, + ) + + def _manifest_payload( + self, + config: IndexConfig, + *, + media_id: str, + input_sha: str, + record_counts: dict[str, int] | None, + store_size_bytes_at_commit: int | None, + schema_version: int, ): now = datetime.now(timezone.utc).isoformat() - manifest = { + return { "manifest_schema_version": MANIFEST_SCHEMA_VERSION, - "index_schema_version": INDEX_SCHEMA_VERSION, + "index_schema_version": schema_version, "dataset": config.dataset, "split": config.split, "run_id": config.run_id, @@ -114,14 +199,6 @@ def write_generation( }, "store_size_bytes_at_commit": store_size_bytes_at_commit, } - write_json_atomic( - config.run_directory / MANIFEST_FILE, - manifest, - ) - return self.repository.generation_reference( - generation_id=str(config.generation_id), - media_id=media_id, - ) def test_unknown_store_size_round_trips_through_snapshot_metadata(self): config, reference = self.generation( @@ -185,6 +262,28 @@ def test_add_reindex_remove_and_clear_publish_immutable_snapshots(self): ) self.assertFalse(self.repository.clear()) + def test_clear_recovers_from_incompatible_generation_schema(self): + config, reference = self.legacy_generation( + "a", + input_sha="a" * 64, + schema_version=INDEX_SCHEMA_VERSION - 1, + ) + self.repository._publish( + generations={"a": reference}, + config_fingerprint=config.fingerprint(), + configuration=self.repository.snapshot_configuration(config), + ) + + with self.assertRaisesRegex(IndexSchemaError, "invalid"): + self.repository.read_active(required=True) + + self.assertTrue(self.repository.clear()) + self.assertEqual( + self.repository.status()["state"], + "empty", + ) + self.assertFalse(self.repository.clear()) + def test_pointer_failure_preserves_previous_active_snapshot(self): config_a1, a1 = self.generation("a", input_sha="a" * 64) previous = self.repository.publish_generation(a1, config_a1) @@ -793,6 +892,65 @@ def test_real_chroma_reader_remains_pinned_across_reindex(self): {second_reference.generation_id}, ) + def test_clear_discards_incompatible_vector_collections(self): + generation_id = self.repository.new_generation_id() + config = replace( + self.config, + video_id="a", + generation_id=generation_id, + generation_directory=self.repository.generation_directory( + generation_id + ), + ) + with IndexStorage(config) as storage: + storage.upsert( + "scene", + [ + StorageRecord( + source_id=f"source-{generation_id}", + embedding=[1.0, 0.0], + metadata={ + **config.record_identity( + "scene", + f"source-{generation_id}", + ), + }, + ) + ], + batch_size=1, + cancellation=CancellationToken(), + ) + reference = self.write_legacy_generation( + config, + media_id="a", + input_sha="a" * 64, + schema_version=INDEX_SCHEMA_VERSION - 1, + record_counts={"scene": 1}, + ) + self.repository._publish( + generations={"a": reference}, + config_fingerprint=config.fingerprint(), + configuration=self.repository.snapshot_configuration(config), + ) + runtime = Mock() + runtime.backends.torch_device = "cpu" + backend = LocalIndexBackend( + Mock(), + runtime, + self.repository.layout, + ) + clear_config = replace( + self.config, + storage_directory=self.repository.indexes, + ) + + self.assertTrue(backend.clear(clear_config)) + + self.assertEqual(self.repository.status()["state"], "empty") + with self.assertRaises(FileNotFoundError): + with IndexStorage(self.config, create=False): + pass + if __name__ == "__main__": unittest.main()