File size: 2,874 Bytes
78ac017 | 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 | """Official-protocol per-query correctness helpers for TVS analyses."""
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import Any
OVO_FORWARD_TASKS = {"REC", "SSR", "CRR"}
def _mcq_letter(value: Any) -> str | None:
text = str(value or "").strip()
match = re.match(r"^[\(\[]?([A-D])(?:[\)\].:\s-]|$)", text, re.IGNORECASE)
return match.group(1).upper() if match else None
def load_gold(benchmark: str, annotation: Path) -> dict[str, dict[str, Any]]:
data = json.loads(annotation.read_text())
gold: dict[str, dict[str, Any]] = {}
if benchmark == "streamingbench":
for video_index, entry in enumerate(data):
for question_index, question in enumerate(entry.get("questions") or []):
uid = f"streaming_{video_index}_{question_index}"
gold[uid] = {
"kind": "streaming_mcq",
"answer": str(question["answer"]),
"task": str(question.get("task_type") or ""),
}
return gold
if benchmark != "ovo":
raise ValueError(f"unsupported benchmark: {benchmark}")
for item in data:
task = str(item.get("task") or "").upper()
if task not in OVO_FORWARD_TASKS:
uid = f"ovo_{item['id']}"
gold[uid] = {
"kind": "ovo_mcq",
"answer": chr(ord("A") + int(item["gt"])),
"task": task,
}
continue
for index, test_info in enumerate(item.get("test_info") or []):
uid = f"ovo_{item['id']}_{index}"
if task == "REC":
gold[uid] = {
"kind": "ovo_rec",
"answer": str(test_info["count"]),
"task": task,
}
else:
gold[uid] = {
"kind": "ovo_binary",
"answer": int(test_info["type"]),
"task": task,
}
return gold
def is_correct(prediction: Any, gold: dict[str, Any]) -> bool:
text = str(prediction or "")
kind = gold["kind"]
if kind == "streaming_mcq":
letter = _mcq_letter(text)
response = letter if letter is not None else text
return response == str(gold["answer"])
if kind == "ovo_mcq":
letter = _mcq_letter(text)
response = letter if letter is not None else text
return str(gold["answer"]) in response
if kind == "ovo_rec":
return "".join(re.findall(r"\d+", text)) == str(gold["answer"])
if kind == "ovo_binary":
positive = int(gold["answer"]) == 1
if (text == "Y" and positive) or (text == "N" and not positive):
return True
return ("Yes" if positive else "No") in text
raise ValueError(f"unknown gold kind: {kind}")
|