| |
| """Deterministic (metamorphic) experiment sets for the four dimensions. |
| |
| Transformations are grouped by the relation they test: |
| |
| * ``scene-synonym`` -- invariance: replace a scene word with a same-scene synonym |
| (ocean -> sea). The official pipeline matches prompt keywords inside the Tag2Text |
| caption, so this is exactly where it is expected to be brittle. |
| * ``action-swap`` -- text-side sensitivity plus the ``other`` protocol: replace the |
| prompt's action with a different Kinetics-400 class, or with an out-of-vocabulary |
| action that must be reported as ``other`` instead of being forced into a class. |
| * ``spatial-mirror`` -- sensitivity: mirror the video horizontally/vertically, so the |
| left/right (above/below) verdict must flip. |
| * ``objects-occlusion`` -- sensitivity: progressively occlude one target object. |
| |
| Prompt-side sets are built here; video-side transforms are emitted as an ffmpeg/occlusion |
| plan that runs on the scoring host (``--emit-video-plan``). |
| |
| Usage:: |
| |
| uv run --no-sync python scripts/build_deterministic_sets.py --captions <scene captions.jsonl> |
| uv run --no-sync python scripts/build_deterministic_sets.py --captions <...> --emit-video-plan |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| from collections import Counter |
| import json |
| from pathlib import Path |
| import sys |
| from typing import Any, Sequence |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT / "src")) |
|
|
| from vbench_prompts_compile import records as R |
| from vbench_prompts_compile import sources as S |
|
|
| |
| |
| SCENE_SYNONYMS: dict[str, str] = { |
| "ocean": "sea", |
| "sea": "ocean", |
| "beach": "seaside", |
| "seaside": "beach", |
| "forest": "woods", |
| "woods": "forest", |
| "cave": "cavern", |
| "cavern": "cave", |
| "garden": "yard", |
| "yard": "garden", |
| "street": "road", |
| "road": "street", |
| "shop": "store", |
| "store": "shop", |
| "bathroom": "washroom", |
| "washroom": "bathroom", |
| "bedroom": "bedchamber", |
| "bedchamber": "bedroom", |
| } |
|
|
|
|
| def substitute_synonym(prompt: str) -> tuple[str, str] | None: |
| """Replace the first synonym-table word; return ``(new_prompt, replaced_pair)``.""" |
| words = prompt.split() |
| for index, word in enumerate(words): |
| key = R.normalize_phrase(word) |
| if key in SCENE_SYNONYMS: |
| replacement = SCENE_SYNONYMS[key] |
| words[index] = replacement |
| return " ".join(words), f"{key}->{replacement}" |
| return None |
|
|
|
|
| def build_scene_synonym(captions_path: Path, *, from_gold: bool = False) -> list[dict[str, Any]]: |
| """Build synonym pairs; with ``from_gold`` every pair keeps its annotated label.""" |
| rows: list[dict[str, Any]] = [] |
| seen: set[tuple[str, str]] = set() |
| for line in captions_path.read_text(encoding="utf-8").splitlines(): |
| if not line.strip(): |
| continue |
| row = json.loads(line) |
| if from_gold: |
| row = { |
| "prompt": row["input"]["prompt"], |
| "caption": row["input"]["caption"], |
| "video_path": row.get("sample_id"), |
| "frame_path": None, |
| "generator": row.get("meta", {}).get("generator"), |
| "split": row.get("meta", {}).get("split"), |
| "gold_label": row.get("target"), |
| } |
| elif row.get("status") != "ok": |
| continue |
| result = substitute_synonym(row["prompt"]) |
| if result is None: |
| continue |
| synonym_prompt, pair = result |
| key = (row["prompt"], str(row.get("video_path"))) |
| if key in seen: |
| continue |
| seen.add(key) |
| rows.append( |
| { |
| "family": "scene-synonym", |
| "pair": pair, |
| "original_prompt": row["prompt"], |
| "variant_prompt": synonym_prompt, |
| "caption": row["caption"], |
| "video_path": row["video_path"], |
| "frame_path": row.get("frame_path"), |
| "generator": row.get("generator"), |
| "split": row.get("split"), |
| **({"gold_label": row["gold_label"]} if row.get("gold_label") else {}), |
| } |
| ) |
| return rows |
|
|
|
|
| def build_action_swap(action_prompts: list[str], vocab: S.K400Vocabulary) -> list[dict[str, Any]]: |
| """Replace the prompt's action class with another class, and add an OOV variant.""" |
| rows: list[dict[str, Any]] = [] |
| labels = list(vocab.labels) |
| for index, prompt in enumerate(action_prompts): |
| original = None |
| for label in labels: |
| if label in prompt.lower(): |
| original = label |
| break |
| if original is None: |
| continue |
| other = labels[(labels.index(original) + 97) % len(labels)] |
| if other == original: |
| continue |
| rows.append( |
| { |
| "family": "action-swap", |
| "kind": "class_swap", |
| "original_prompt": prompt, |
| "variant_prompt": prompt.lower().replace(original, other), |
| "original_action": original, |
| "variant_action": other, |
| } |
| ) |
| oov = "assembling furniture" |
| if oov in prompt.lower(): |
| continue |
| rows.append( |
| { |
| "family": "action-swap", |
| "kind": "out_of_vocabulary", |
| "original_prompt": prompt, |
| "variant_prompt": prompt.lower().replace(original, oov), |
| "original_action": original, |
| "variant_action": "other", |
| } |
| ) |
| return rows |
|
|
|
|
| def build_video_plan(captions_path: Path) -> dict[str, list[dict[str, Any]]]: |
| """Mirror- and occlusion-side plans (executed on the scoring host).""" |
| scene_rows = [] |
| for line in captions_path.read_text(encoding="utf-8").splitlines(): |
| if not line.strip(): |
| continue |
| row = json.loads(line) |
| if row.get("status") == "ok": |
| scene_rows.append({"video_path": row["video_path"], "prompt": row["prompt"], "generator": row.get("generator")}) |
| return { |
| "spatial-mirror": [ |
| {"video_path": row["video_path"], "prompt": row["prompt"], "transforms": ["hflip", "vflip"]} for row in scene_rows |
| ], |
| "objects-occlusion": [ |
| {"video_path": row["video_path"], "prompt": row["prompt"], "levels": [0.0, 0.25, 0.5, 0.75, 1.0]} for row in scene_rows |
| ], |
| } |
|
|
|
|
| def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument("--captions", required=True, help="caption_frames.py output (real frames + official Tag2Text captions)") |
| parser.add_argument("--action-prompts", default=None, help="JSONL with action prompts (defaults to the K400 template corpus)") |
| parser.add_argument("--out-dir", default=str(ROOT / "data" / "deterministic")) |
| parser.add_argument("--emit-video-plan", action="store_true") |
| parser.add_argument("--synonyms-from-gold", action="store_true", help="treat --captions as a scene gold file so every pair has an annotated label") |
| return parser.parse_args(argv) |
|
|
|
|
| def main(argv: Sequence[str] | None = None) -> int: |
| args = parse_args(argv) |
| captions_path = Path(args.captions) |
| out_dir = Path(args.out_dir) |
| out_dir.mkdir(parents=True, exist_ok=True) |
|
|
| scene_rows = build_scene_synonym(captions_path, from_gold=args.synonyms_from_gold) |
| R.write_jsonl(out_dir / "scene-synonym.jsonl", scene_rows) |
| print(json.dumps({"scene_synonym_pairs": len(scene_rows), "pairs": dict(Counter(row["pair"] for row in scene_rows))}, ensure_ascii=False)) |
|
|
| action_prompts: list[str] = [] |
| action_path = Path(args.action_prompts) if args.action_prompts else ROOT / "data" / "processed" / "clean-0002" / "candidates" / "action.jsonl" |
| if action_path.exists(): |
| for line in action_path.read_text(encoding="utf-8").splitlines(): |
| if line.strip(): |
| action_prompts.append(json.loads(line)["input"]["prompt"]) |
| action_rows = build_action_swap(action_prompts, S.load_k400()) |
| R.write_jsonl(out_dir / "action-swap.jsonl", action_rows) |
| print(json.dumps({"action_swap_pairs": len(action_rows), "kinds": dict(Counter(row["kind"] for row in action_rows))}, ensure_ascii=False)) |
|
|
| if args.emit_video_plan: |
| plan = build_video_plan(captions_path) |
| (out_dir / "video-plan.json").write_text(json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") |
| print(json.dumps({"video_plan_families": {k: len(v) for k, v in plan.items()}}, ensure_ascii=False)) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
