rpx_benchmark.vqa.canonical

 1"""Canonicalization for detecting semantic (not just literal-string)
 2duplicate facts at *sampling* time -- used only by the acceptance-manifest
 3and benchmark-plan builders to avoid spending two scarce slots on what is,
 4after normalization, the same underlying question. Mirrors the alias tables
 5data/mask_annotation/visual_grounding_gt/vqa_gt/gt_incontext.py applies
 6during ground-truth reference matching (``_canon_color``/``_canon_function``)
 7-- never used to alter a stored question, answer, or fact field."""
 8
 9from __future__ import annotations
10
11import re
12
13_COLOR_ALIASES = {"gray": "grey"}
14_FUNCTION_ALIASES = {"to drink": "drink", "to eat": "eat", "to snack": "snack"}
15
16
17def normalize_value(value: str) -> str:
18    return " ".join(value.strip().lower().split())
19
20
21def canon_color(value: str | None) -> str | None:
22    if value is None:
23        return None
24    norm = normalize_value(value)
25    return _COLOR_ALIASES.get(norm, norm)
26
27
28def canon_function(value: str | None) -> str | None:
29    if value is None:
30        return None
31    norm = normalize_value(value)
32    return _FUNCTION_ALIASES.get(norm, norm)
33
34
35def canon_material(value: str | None) -> str | None:
36    return None if value is None else normalize_value(value)
37
38
39_COMPOSITION_RE = re.compile(r"made of (.+?) that is used for (.+?)\??$", re.IGNORECASE)
40_ODD_ONE_OUT_RE = re.compile(r"NOT made of (.+?)\??$", re.IGNORECASE)
41
42
43def parse_composition_question(question: str) -> tuple[str, str] | None:
44    match = _COMPOSITION_RE.search(question)
45    return None if match is None else (match.group(1).strip(), match.group(2).strip())
46
47
48def parse_odd_one_out_question(question: str) -> str | None:
49    match = _ODD_ONE_OUT_RE.search(question)
50    return None if match is None else match.group(1).strip()
51
52
53def semantic_key(row: dict) -> tuple:
54    """A (scene, kind, phase, frame, type, target, reference, canon-value...)
55    tuple such that two rows sharing a key are the same underlying question
56    after alias normalization, even if their literal answer/question text
57    spells the shared attribute value differently (e.g. "drink" vs
58    "to drink"). Works for both normal rows (attr_value / question-derived
59    material+function) and in-context rows (attribute_value / attribute_
60    material / attribute_function columns)."""
61    qtype = row["type"]
62    target = row.get("target_source_catalog_id") or row.get("target_oid")
63    reference = row.get("reference_source_catalog_id")
64    base = (row["scene_id"], row["kind"], row.get("phase"), row["frame"], qtype, target, reference)
65    if qtype in {"attr_single_color", "inctx_attr_single_color"}:
66        return base + (canon_color(row.get("attribute_value") or row.get("attr_value")),)
67    if qtype in {"attr_single_function", "inctx_attr_single_function"}:
68        return base + (canon_function(row.get("attribute_value") or row.get("attr_value")),)
69    if qtype in {"attr_single_material", "inctx_attr_single_material"}:
70        return base + (canon_material(row.get("attribute_value") or row.get("attr_value")),)
71    if qtype in {"attr_composition", "inctx_attr_composition"}:
72        material = row.get("attribute_material")
73        function = row.get("attribute_function")
74        if material is None or function is None:
75            parsed = parse_composition_question(row["question"])
76            material, function = parsed if parsed else (material, function)
77        return base + (canon_material(material), canon_function(function))
78    if qtype in {"attr_odd_one_out", "inctx_attr_odd_one_out"}:
79        material = row.get("attribute_material")
80        if material is None:
81            material = parse_odd_one_out_question(row["question"])
82        return base + (canon_material(material),)
83    # Spatial and any other type: no alias-prone attribute value, identity
84    # is already the base tuple plus the raw answer.
85    return base + (row.get("answer"),)