File size: 4,711 Bytes
c10636a | 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 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Create an image-only runtime package from local image/metadata pairs."""
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
from pathlib import Path
from PIL import Image
from eviscene.eval.release_adapter import (
DATA_PIPELINE_DECISION_POLICY_VERSION,
INPUT_MODE,
RELEASE_SCHEMA_VERSION,
RUNTIME_CASE_KEYS,
RUNTIME_CONTRACT_VERSION,
)
def prepare_runtime_input(
metadata: Path, image_root: Path, output: Path, *, release_id: str
) -> int:
"""Whitelist runtime fields; validate all sources before creating output."""
if not release_id.strip():
raise ValueError("release_id must not be empty")
source_root = image_root.resolve(strict=True)
output = output.resolve()
if output.exists():
raise FileExistsError("output already exists; choose a new directory")
sources = []
seen = set()
with metadata.open(encoding="utf-8") as handle:
for number, line in enumerate(handle, 1):
if not line.strip():
continue
row = json.loads(line)
if not isinstance(row, dict):
raise ValueError(f"line {number}: expected an object")
case_id = row.get("case_id")
if not isinstance(case_id, str) or not case_id.strip() or case_id in seen:
raise ValueError(f"line {number}: empty or duplicate case_id")
raw_path = row.get("image_path")
if not isinstance(raw_path, str) or not raw_path.strip():
raise ValueError(f"line {number}: missing image_path")
relative = Path(raw_path)
if relative.is_absolute() or relative.drive:
raise ValueError(f"line {number}: image_path must be relative")
source = (source_root / relative).resolve(strict=True)
if not source.is_relative_to(source_root) or not source.is_file():
raise ValueError(
f"line {number}: image_path escapes image_root or is not a file"
)
with Image.open(source) as image:
image.verify()
seen.add(case_id)
sources.append((case_id, source))
if not sources:
raise ValueError("metadata contains no cases")
output.mkdir(parents=True, exist_ok=False)
runtime = output / "runtime_input"
images = runtime / "images"
images.mkdir(parents=True)
rows = []
for index, (case_id, source) in enumerate(sources, 1):
image_path = Path("images") / f"{index:06d}{source.suffix.lower()}"
target = runtime / image_path
shutil.copyfile(source, target)
with target.open("rb") as handle:
digest = hashlib.file_digest(handle, "sha256").hexdigest()
rows.append(
{
"case_id": case_id,
"image_path": image_path.as_posix(),
"image_sha256": digest,
}
)
(runtime / "cases.jsonl").write_text(
"".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
encoding="utf-8",
)
# Publish the manifest last: an interrupted copy is not a valid release.
manifest = {
"schema_version": RELEASE_SCHEMA_VERSION,
"runtime_contract_version": RUNTIME_CONTRACT_VERSION,
"input_mode": INPUT_MODE,
"decision_policy_version": DATA_PIPELINE_DECISION_POLICY_VERSION,
"release_id": release_id,
"release_stage": "custom_image_only",
"runtime_contract": {
"allowed_keys": sorted(RUNTIME_CASE_KEYS),
"private_keys_absent": True,
},
"artifacts": {"agent_input": "runtime_input/cases.jsonl"},
"source_access_policy": {"active": False},
"case_count": len(rows),
}
(output / "manifest.json").write_text(
json.dumps(manifest, indent=2) + "\n", encoding="utf-8"
)
return len(rows)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--metadata", type=Path, required=True, help="JSONL with case_id and image_path"
)
parser.add_argument("--image-root", type=Path, required=True)
parser.add_argument(
"--output", type=Path, required=True, help="New directory; never overwritten"
)
parser.add_argument(
"--release-id",
required=True,
help="Your dataset version, not the frozen EviLens release ID",
)
args = parser.parse_args()
count = prepare_runtime_input(
args.metadata, args.image_root, args.output, release_id=args.release_id
)
print(f"Prepared {count} image-only cases")
if __name__ == "__main__":
main()
|