Spaces:
Running on Zero
Running on Zero
File size: 6,255 Bytes
85300f6 | 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 129 130 131 132 133 134 135 136 137 138 139 | """Typed pilot records, strict target parsing and answer-free prompt construction."""
import ast
import math
import re
SCHEMA_VERSION = "spatial-decision/v2"
EXTENDED_SCHEMA_VERSION = "spatial-decision/v3"
TASKS = ("choice", "scalar", "point")
UNIT_FACTORS = {"m": 1.0, "meter": 1.0, "meters": 1.0,
"cm": 0.01, "centimeter": 0.01, "centimeters": 0.01,
"mm": 0.001, "millimeter": 0.001, "millimeters": 0.001,
"inch": 0.0254, "inches": 0.0254, "ft": 0.3048,
"foot": 0.3048, "feet": 0.3048}
_NUMBER = r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?"
def parse_measurement(answer, question=""):
"""Accept a scalar, optionally with an explicit length unit; never extract CoT numbers."""
match = re.fullmatch(rf"\s*({_NUMBER})\s*([a-zA-Z]+)?\s*\.?\s*", answer)
if not match:
raise ValueError("not_a_scalar_answer")
unit = match.group(2)
if unit is None:
units = re.findall(r"\b(" + "|".join(UNIT_FACTORS) + r")\b", question.lower())
factors = {UNIT_FACTORS[u] for u in units}
if len(factors) != 1:
raise ValueError("missing_or_ambiguous_unit")
unit = units[-1]
unit = unit.lower()
if unit not in UNIT_FACTORS:
raise ValueError("unsupported_unit")
value = float(match.group(1)) * UNIT_FACTORS[unit]
if not math.isfinite(value) or value < 0:
raise ValueError("invalid_measurement")
return value, unit
def parse_single_point(answer):
try:
value = ast.literal_eval(answer.strip())
except (ValueError, SyntaxError) as exc:
raise ValueError("not_a_point_literal") from exc
if not isinstance(value, (list, tuple)) or len(value) != 1:
raise ValueError("not_a_single_point")
point = value[0]
if not isinstance(point, (list, tuple)) or len(point) != 2:
raise ValueError("not_xy")
if any(isinstance(x, bool) or not isinstance(x, (int, float)) for x in point):
raise ValueError("non_numeric_point")
if any(not math.isfinite(x) or not 0 <= x <= 1 for x in point):
raise ValueError("point_out_of_range")
return [float(x) for x in point]
def clean_question(question):
question = question.replace("<image>", "").strip()
question = re.split(r"\s*Your answer should be formatted", question, flags=re.I)[0]
question = re.sub(r"\s*Answer the question using a single word or phrase\.?", "",
question, flags=re.I)
return question.strip()
def canonical_metric_question(question):
question = clean_question(question)
# Do not rewrite numeric lengths in a premise without also converting their values.
if re.search(rf"{_NUMBER}\s*(?:cm|mm|meters?|centimeters?|inches|feet)\b",
question, re.I):
raise ValueError("numeric_unit_in_question_premise")
return re.sub(r"\b(?:centimeters?|cm|millimeters?|mm|inches|feet|foot|ft)\b",
"meters", question, flags=re.I)
def validate_record(record, max_choices=32):
if record.get("schema_version") not in (SCHEMA_VERSION, EXTENDED_SCHEMA_VERSION):
raise ValueError("unsupported_schema")
inp = record["input"]
if not inp["question"].strip() or not inp["media"]:
raise ValueError("empty_input")
if any(m["kind"] != "image" for m in inp["media"]):
raise ValueError("pilot_requires_images")
space, target = inp["answer_space"], record["target"]
kind = space["kind"]
if kind == "choice":
options = space["options"]
ids = [o["id"] for o in options]
if not 2 <= len(options) <= max_choices or len(set(ids)) != len(ids):
raise ValueError("invalid_options")
if target["option_id"] not in ids:
raise ValueError("target_missing_from_options")
elif kind == "scalar":
allowed = ("m", "count", "deg") if record["schema_version"] == EXTENDED_SCHEMA_VERSION else ("m",)
if space["unit"] not in allowed or target["unit"] != space["unit"]:
raise ValueError("pilot_scalar_requires_meters")
if isinstance(target["value"], bool) or not math.isfinite(target["value"]) or target["value"] < 0:
raise ValueError("invalid_scalar")
if space["unit"] == "count":
if space.get("quantity") != "count" or int(target["value"]) != target["value"] or target["value"] > 1023:
raise ValueError("invalid_integer_count")
if space["unit"] == "deg":
if target["value"] > 360 or not space.get("angle_semantics"):
raise ValueError("invalid_angle_semantics_or_range")
elif kind == "point":
if len(inp["media"]) != 1:
raise ValueError("pilot_point_requires_one_image")
if space["coordinate_system"] != "normalized_xy" or space["num_points"] != 1:
raise ValueError("unsupported_point_space")
parse_single_point(repr(target["points"]))
else:
raise ValueError("unsupported_task")
def compile_prompt(inp, permutation=None):
"""Only input is accepted; targets and provenance cannot enter the prompt."""
space = inp["answer_space"]
kind = space["kind"]
text = f"Question: {inp['question']}\n"
option_ids = []
if kind == "choice":
options = space["options"]
order = list(range(len(options))) if permutation is None else list(permutation)
if sorted(order) != list(range(len(options))):
raise ValueError("invalid_permutation")
text += "Options:\n"
for i, index in enumerate(order):
option = options[index]
text += f"{i + 1}. {option['text']}\n"
option_ids.append(option["id"])
text += "Select exactly one option."
elif kind == "scalar":
unit = {"m": "meters", "count": "count units (an integer)", "deg": "degrees"}[space["unit"]]
text += f"Estimate {space['quantity']}. Return a nonnegative value in {unit}."
elif kind == "point":
text += ("Return one valid point (x, y) in the original image, normalized to [0, 1]. "
"The origin is top-left; x increases rightward and y increases downward.")
else:
raise ValueError("unsupported_task")
return text, option_ids
|