File size: 3,078 Bytes
2847d0b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | #!/usr/bin/env python3
import argparse
import hashlib
import json
import os
from pathlib import Path
def read_nonempty(path: Path) -> list[str]:
return [
line.strip()
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
def atomic_write(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.tmp.{os.getpid()}")
temporary.write_text(text, encoding="utf-8")
os.replace(temporary, path)
def main() -> None:
parser = argparse.ArgumentParser(
description="Create the fixed MovieGenVideoBench prompts 101-200 split."
)
parser.add_argument(
"--source",
type=Path,
default=Path("prompts/MovieGenVideoBench_extended.txt"),
)
parser.add_argument(
"--training-cases",
type=Path,
default=Path("data/self_forcing_predictor_v4_1000_seed0/cases.jsonl"),
)
parser.add_argument(
"--output",
type=Path,
default=Path("prompts/MovieGenVideoBench_extended_101_200.txt"),
)
parser.add_argument(
"--manifest",
type=Path,
default=Path("prompts/MovieGenVideoBench_extended_101_200.json"),
)
args = parser.parse_args()
source_prompts = read_nonempty(args.source)
if len(source_prompts) < 200:
raise ValueError(
f"{args.source} has only {len(source_prompts)} non-empty prompts"
)
selected = source_prompts[100:200]
if len(selected) != 100 or len(set(selected)) != 100:
raise ValueError("prompts 101-200 must contain 100 unique prompts")
training_rows = [
json.loads(line)
for line in args.training_cases.read_text(encoding="utf-8").splitlines()
if line.strip()
]
training_prompts = {row["prompt"] for row in training_rows}
overlap = sorted(set(selected).intersection(training_prompts))
if overlap:
raise ValueError(
f"validation split overlaps {len(overlap)} exact training prompts"
)
prompts_text = "".join(f"{prompt}\n" for prompt in selected)
prompt_sha256 = hashlib.sha256(prompts_text.encode("utf-8")).hexdigest()
manifest = {
"source": str(args.source.resolve()),
"source_nonempty_count": len(source_prompts),
"selection": {
"zero_based_slice": [100, 200],
"one_based_prompt_range": [101, 200],
"count": len(selected),
"unique_count": len(set(selected)),
},
"training_cases": str(args.training_cases.resolve()),
"training_prompt_count": len(training_rows),
"exact_training_overlap_count": 0,
"seed": 0,
"output": str(args.output.resolve()),
"output_sha256": prompt_sha256,
}
atomic_write(args.output, prompts_text)
atomic_write(
args.manifest,
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
)
print(json.dumps(manifest, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()
|