Spaces:
Running on Zero
Running on Zero
File size: 5,021 Bytes
e7c2ed2 | 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 | """Shared manifest ordering and response validation for the human evaluation."""
import hashlib
import json
import os
import random
import re
import uuid
from pathlib import Path
DIMENSIONS = {
"edit_a_applied",
"edit_b_applied",
"composition_success",
"source_preservation",
"absence_unintended_changes",
"visual_realism",
"overall",
}
MAX_REQUEST_BYTES = 2 * 1024 * 1024
RATER_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{2,63}$")
def read_json(path):
with Path(path).open("r", encoding="utf-8") as handle:
return json.load(handle)
def write_json_atomic(value, path):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_name(f".{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
temporary.write_text(
json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
)
temporary.replace(path)
def stable_seed(*parts):
payload = ":".join(str(part) for part in parts)
return int.from_bytes(hashlib.sha256(payload.encode("utf-8")).digest()[:8], "big")
def validate_rater_id(rater_id):
if not isinstance(rater_id, str) or not RATER_ID_PATTERN.fullmatch(rater_id):
raise ValueError(
"Participant code must be 3-64 characters using letters, numbers, "
"underscores, or hyphens."
)
def public_manifest(study_root, mode, rater_id):
if mode != "main":
raise ValueError("mode must be main")
validate_rater_id(rater_id)
manifest = read_json(Path(study_root) / f"{mode}_public.json")
assignments = manifest.pop("assignment_orders", {})
by_id = {item["blind_id"]: item for item in manifest["items"]}
order = assignments.get(rater_id)
if order is None:
order = list(by_id)
random.Random(stable_seed(manifest["study_id"], mode, rater_id)).shuffle(order)
manifest["items"] = [by_id[item_id] for item_id in order]
manifest["rater_id"] = rater_id
return manifest
def validate_submission(payload, manifest):
if payload.get("schema_version") != manifest["schema_version"]:
raise ValueError("Response schema version does not match the study")
if payload.get("study_id") != manifest["study_id"]:
raise ValueError("Response study id does not match")
if payload.get("set") != manifest["set"]:
raise ValueError("Response set does not match")
validate_rater_id(payload.get("rater_id"))
if payload.get("rater_id") != manifest["rater_id"]:
raise ValueError("Response rater code does not match")
tutorial_version = payload.get("tutorial_version", 0)
if (
isinstance(tutorial_version, bool)
or not isinstance(tutorial_version, int)
or not 0 <= tutorial_version <= 100
):
raise ValueError("tutorial_version must be an integer between 0 and 100")
allowed_ids = {item["blind_id"] for item in manifest["items"]}
seen = set()
responses = payload.get("responses")
if not isinstance(responses, list):
raise ValueError("responses must be a list")
for response in responses:
blind_id = response.get("blind_id")
if blind_id not in allowed_ids or blind_id in seen:
raise ValueError(f"Invalid or repeated blinded item id: {blind_id!r}")
seen.add(blind_id)
ratings = response.get("ratings")
unclear = response.get("unclear")
if not isinstance(ratings, dict) or set(ratings) != DIMENSIONS:
raise ValueError(f"Item {blind_id} has an invalid rating key set")
if not isinstance(unclear, dict) or set(unclear) != {"edit_a", "edit_b"}:
raise ValueError(f"Item {blind_id} has invalid unclear flags")
if not all(isinstance(value, bool) for value in unclear.values()):
raise ValueError(f"Item {blind_id} unclear flags must be booleans")
comment = response.get("comment", "")
if not isinstance(comment, str) or len(comment) > 1000:
raise ValueError(f"Item {blind_id} has an invalid optional comment")
for dimension, value in ratings.items():
may_be_null = (
dimension == "edit_a_applied" and unclear["edit_a"]
) or (dimension == "edit_b_applied" and unclear["edit_b"])
if value is None and may_be_null:
continue
if (
isinstance(value, bool)
or not isinstance(value, int)
or not 1 <= value <= 5
):
raise ValueError(f"Item {blind_id} has invalid {dimension} rating")
elapsed = response.get("elapsed_seconds")
if elapsed is not None and (
isinstance(elapsed, bool)
or not isinstance(elapsed, (int, float))
or elapsed < 0
):
raise ValueError(f"Item {blind_id} has invalid elapsed_seconds")
if payload.get("complete") and seen != allowed_ids:
raise ValueError("A completed response must contain every assigned item")
|