File size: 6,429 Bytes
53c10a4 | 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 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 | """Small parsing and geometry primitives shared by built-in adapters."""
from __future__ import annotations
import json
import math
import re
from collections.abc import Mapping, Sequence
from typing import Any, Optional
_ANSWER_RE = re.compile(
r"<answer>\s*(.*?)\s*</answer>",
flags=re.DOTALL | re.IGNORECASE,
)
_CANONICAL_ANSWER_RE = re.compile(
r"\A\s*(?:(?:<think>)?.*?</think>\s*)?"
r"<answer>\s*(.*?)\s*</answer>\s*\Z",
flags=re.DOTALL | re.IGNORECASE,
)
_FENCE_RE = re.compile(
r"\A\s*```(?:json)?\s*(.*?)\s*```\s*\Z",
flags=re.DOTALL | re.IGNORECASE,
)
_THINK_BLOCK_RE = re.compile(
r"<think>.*?</think>",
flags=re.DOTALL | re.IGNORECASE,
)
def finite_float(value: Any) -> Optional[float]:
"""Return a finite float, rejecting booleans and invalid values."""
if isinstance(value, bool):
return None
try:
number = float(value)
except (TypeError, ValueError):
return None
return number if math.isfinite(number) else None
def exact_answer_payload(value: Any) -> Optional[str]:
"""Extract a payload only from the canonical final-answer shape."""
match = _CANONICAL_ANSWER_RE.fullmatch(str(value or ""))
if match is None or not match.group(1).strip():
return None
return match.group(1).strip()
def answer_payload(value: Any) -> str:
"""Extract the final answer block when present, otherwise return text."""
text = str(value or "").strip()
matches = _ANSWER_RE.findall(text)
return matches[-1].strip() if matches else text
def final_response_text(value: Any) -> str:
"""Remove a complete reasoning block and an optional answer wrapper."""
text = str(value or "").strip()
text = _THINK_BLOCK_RE.sub("", text).strip()
if "</think>" in text.lower():
text = re.split(r"</think>", text, flags=re.IGNORECASE)[-1].strip()
matches = _ANSWER_RE.findall(text)
return matches[-1].strip() if matches else text
def unfence(value: Any) -> str:
text = str(value or "").strip()
match = _FENCE_RE.fullmatch(text)
return match.group(1).strip() if match is not None else text
def parse_json(value: Any) -> Any:
"""Parse JSON from common answer/fence wrappers without executing code."""
if isinstance(value, (Mapping, list, tuple)):
return value
text = unfence(answer_payload(value))
try:
return json.loads(text)
except (TypeError, ValueError):
pass
decoder = json.JSONDecoder()
for index, character in enumerate(text):
if character not in "[{":
continue
try:
parsed, _ = decoder.raw_decode(text[index:])
except (TypeError, ValueError):
continue
return parsed
return None
def parse_mapping(value: Any) -> Optional[dict[str, Any]]:
payload = parse_json(value)
return dict(payload) if isinstance(payload, Mapping) else None
def canonical_json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
def canonical_answer(value: Any) -> str:
payload = value if isinstance(value, str) else canonical_json(value)
return f"<answer>{payload.strip()}</answer>"
def normalize_box(value: Any, *, reorder: bool = False) -> Optional[list[float]]:
"""Normalize one ``[x1, y1, x2, y2]`` box."""
if isinstance(value, Mapping):
for key in ("bbox_2d", "bbox", "box", "boxes"):
if key in value:
box = normalize_box(value[key], reorder=reorder)
if box is not None:
return box
return None
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)) and len(value) == 4:
box = [finite_float(coordinate) for coordinate in value]
if all(coordinate is not None for coordinate in box):
normalized = [float(coordinate) for coordinate in box]
if reorder:
normalized[0], normalized[2] = sorted((normalized[0], normalized[2]))
normalized[1], normalized[3] = sorted((normalized[1], normalized[3]))
return normalized
if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
for item in value:
box = normalize_box(item, reorder=reorder)
if box is not None:
return box
return None
def normalize_boxes(value: Any) -> dict[str, list[float]]:
"""Normalize frame-keyed boxes using the released integer-key contract."""
if not isinstance(value, Mapping):
return {}
boxes: dict[str, list[float]] = {}
for key, raw_box in value.items():
key_number = finite_float(key)
box = normalize_box(raw_box)
if key_number is None or box is None or not key_number.is_integer():
continue
boxes[str(int(key_number))] = box
return boxes
def box_iou(first: Any, second: Any) -> float:
box_a = normalize_box(first)
box_b = normalize_box(second)
if box_a is None or box_b is None:
return 0.0
x1 = max(box_a[0], box_b[0])
y1 = max(box_a[1], box_b[1])
x2 = min(box_a[2], box_b[2])
y2 = min(box_a[3], box_b[3])
intersection = max(0.0, x2 - x1) * max(0.0, y2 - y1)
area_a = max(0.0, box_a[2] - box_a[0]) * max(0.0, box_a[3] - box_a[1])
area_b = max(0.0, box_b[2] - box_b[0]) * max(0.0, box_b[3] - box_b[1])
union = area_a + area_b - intersection
return intersection / union if union > 0.0 else 0.0
def interval_iou(first: Any, second: Any) -> float:
if not (
isinstance(first, Sequence)
and not isinstance(first, (str, bytes))
and len(first) == 2
and isinstance(second, Sequence)
and not isinstance(second, (str, bytes))
and len(second) == 2
):
return 0.0
first_values = [finite_float(value) for value in first]
second_values = [finite_float(value) for value in second]
if any(value is None for value in first_values + second_values):
return 0.0
a0, a1 = sorted(float(value) for value in first_values)
b0, b1 = sorted(float(value) for value in second_values)
intersection = max(0.0, min(a1, b1) - max(a0, b0))
union = max(a1, b1) - min(a0, b0)
return intersection / union if union > 0.0 else 0.0
def ground_truth(item: Mapping[str, Any]) -> Any:
value = item.get("ground_truth")
return item.get("answer") if value is None else value
|