| |
| """Evaluate Triadic Object-State Consistency (TOSC). |
| |
| The evaluator consumes captions generated for origin / removed / replaced |
| images, extracts canonical COCO objects with local synonym and lemmatization |
| rules, and computes TOSC metrics. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import ast |
| import json |
| import re |
| from dataclasses import dataclass |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| REPO_ROOT = Path(__file__).resolve().parents[1] |
| DEFAULT_DATASET_FILE = REPO_ROOT / "benchmark" / "TOSC_dataset.jsonl" |
| DEFAULT_INSERTIONS_FILE = REPO_ROOT / "benchmark" / "insertions.jsonl" |
|
|
| STATE_SUFFIXES = { |
| "origin": "_origin", |
| "removed": "_removed", |
| "replaced": "_replaced", |
| } |
| |
| |
| |
| INDEX_ID_RE = re.compile(r"^(\d+)") |
| MASKED_OBJECT_RE = re.compile(r"_masked_(.+)$") |
|
|
|
|
| try: |
| from nltk.stem import WordNetLemmatizer |
|
|
| _LEMMATIZER = WordNetLemmatizer() |
| except Exception: |
| _LEMMATIZER = None |
|
|
|
|
| @dataclass(frozen=True) |
| class DatasetSample: |
| sample_id: str |
| state: str |
| image_path: str |
| question: str | None |
| base_id: str | None |
| coco_id: str |
|
|
|
|
| @dataclass(frozen=True) |
| class Triplet: |
| base_id: str |
| coco_id: str |
| original_object: str |
| replacement_object: str |
| origin: DatasetSample |
| removed: DatasetSample |
| replaced: DatasetSample |
|
|
|
|
| class ObjectNormalizer: |
| def __init__(self) -> None: |
| self.synonym_to_canonical: dict[str, str] = {} |
| self.phrase_to_canonicals: dict[str, set[str]] = {} |
| self.canonical_objects: set[str] = set() |
| self._load_eval_masked_synonyms() |
|
|
| def _load_eval_masked_synonyms(self) -> None: |
| synonyms_path = Path(__file__).with_name("eval_masked_obj_generative.py") |
| payload = ast.parse(synonyms_path.read_text(encoding="utf-8")) |
| object_synonyms: dict[str, list[str]] | None = None |
| for node in payload.body: |
| if not isinstance(node, ast.Assign): |
| continue |
| if not any(isinstance(target, ast.Name) and target.id == "OBJECT_SYNONYMS" for target in node.targets): |
| continue |
| object_synonyms = ast.literal_eval(node.value) |
| break |
| if not object_synonyms: |
| raise ValueError(f"Could not load OBJECT_SYNONYMS from {synonyms_path}") |
|
|
| for canonical, synonyms in object_synonyms.items(): |
| self._add_synonym_group(canonical, [canonical, *synonyms], overwrite=True) |
|
|
| def _add_synonym_group(self, canonical: str, synonyms: list[str], *, overwrite: bool) -> None: |
| canonical_norm = normalize_phrase(canonical) |
| if not canonical_norm: |
| return |
| self.canonical_objects.add(canonical_norm) |
| for synonym in synonyms: |
| self._add_synonym(synonym, canonical_norm, overwrite=overwrite) |
|
|
| def _add_synonym(self, synonym: str, canonical: str, *, overwrite: bool) -> None: |
| canonical_norm = normalize_phrase(canonical) |
| if not canonical_norm: |
| return |
| self.canonical_objects.add(canonical_norm) |
| variants = { |
| normalize_phrase(synonym), |
| lemmatize_phrase(normalize_phrase(synonym)), |
| simple_singular_phrase(normalize_phrase(synonym)), |
| } |
| for variant in variants: |
| if not variant: |
| continue |
| self.phrase_to_canonicals.setdefault(variant, set()).add(canonical_norm) |
| if overwrite or variant not in self.synonym_to_canonical: |
| self.synonym_to_canonical[variant] = canonical_norm |
|
|
| def canonicalize(self, phrase: str) -> str | None: |
| normalized = normalize_phrase(phrase) |
| if not normalized: |
| return None |
| for candidate in ( |
| normalized, |
| lemmatize_phrase(normalized), |
| simple_singular_phrase(normalized), |
| ): |
| if candidate in self.synonym_to_canonical: |
| return self.synonym_to_canonical[candidate] |
| if normalized in self.canonical_objects: |
| return normalized |
| return None |
|
|
| def normalize_objects(self, raw_objects: list[str]) -> tuple[list[str], list[str]]: |
| canonical: set[str] = set() |
| unmapped: list[str] = [] |
| seen_unmapped: set[str] = set() |
| for raw_object in raw_objects: |
| mapped = self.canonicalize(raw_object) |
| if mapped: |
| canonical.add(mapped) |
| continue |
| cleaned = normalize_phrase(raw_object) |
| if cleaned and cleaned not in seen_unmapped: |
| unmapped.append(cleaned) |
| seen_unmapped.add(cleaned) |
| return sorted(canonical), unmapped |
|
|
| def extract_caption_objects(self, caption: str) -> dict[str, Any]: |
| caption_variants = caption_text_variants(caption) |
| raw_objects: set[str] = set() |
| normalized_objects: set[str] = set() |
|
|
| for phrase, canonicals in self.phrase_to_canonicals.items(): |
| if not phrase: |
| continue |
| if any(contains_token_phrase(text, phrase) for text in caption_variants): |
| raw_objects.add(phrase) |
| normalized_objects.update(canonicals) |
|
|
| return { |
| "raw_objects": sorted(raw_objects), |
| "normalized_objects": sorted(normalized_objects), |
| "unmapped_raw_objects": [], |
| "extraction_source": "local_synonym_match", |
| } |
|
|
|
|
| def normalize_phrase(value: str) -> str: |
| value = str(value).lower().strip() |
| value = value.replace("_", " ").replace("-", " ") |
| value = re.sub(r"[^a-z0-9\s]", " ", value) |
| value = re.sub(r"\b(a|an|the)\b", " ", value) |
| return re.sub(r"\s+", " ", value).strip() |
|
|
|
|
| def lemmatize_word(word: str) -> str: |
| if not word: |
| return word |
| if _LEMMATIZER is None: |
| return simple_singular_word(word) |
| try: |
| return _LEMMATIZER.lemmatize(word) |
| except Exception: |
| return simple_singular_word(word) |
|
|
|
|
| def lemmatize_phrase(phrase: str) -> str: |
| return " ".join(lemmatize_word(part) for part in phrase.split()) |
|
|
|
|
| def simple_singular_word(word: str) -> str: |
| if len(word) > 3 and word.endswith("ies"): |
| return word[:-3] + "y" |
| if len(word) > 3 and word.endswith("es") and not word.endswith(("ses", "ies")): |
| return word[:-2] |
| if len(word) > 3 and word.endswith("s") and not word.endswith("ss"): |
| return word[:-1] |
| return word |
|
|
|
|
| def simple_singular_phrase(phrase: str) -> str: |
| return " ".join(simple_singular_word(part) for part in phrase.split()) |
|
|
|
|
| def caption_text_variants(caption: str) -> set[str]: |
| normalized = normalize_phrase(caption) |
| variants = { |
| normalized, |
| lemmatize_phrase(normalized), |
| simple_singular_phrase(normalized), |
| simple_singular_phrase(lemmatize_phrase(normalized)), |
| } |
| return {variant for variant in variants if variant} |
|
|
|
|
| def contains_token_phrase(text: str, phrase: str) -> bool: |
| return f" {phrase} " in f" {text} " |
|
|
|
|
| def read_json_or_jsonl(path: Path) -> Any: |
| if path.suffix == ".jsonl": |
| with path.open("r", encoding="utf-8") as handle: |
| return [json.loads(line) for line in handle if line.strip()] |
| with path.open("r", encoding="utf-8") as handle: |
| return json.load(handle) |
|
|
|
|
| def write_json(path: Path, payload: Any) -> None: |
| path.parent.mkdir(parents=True, exist_ok=True) |
| with path.open("w", encoding="utf-8") as handle: |
| json.dump(payload, handle, ensure_ascii=False, indent=2) |
| handle.write("\n") |
|
|
|
|
| def get_record_id(record: dict[str, Any]) -> str | None: |
| if "question_id" in record: |
| return str(record["question_id"]) |
| if "image_id" in record: |
| return str(record["image_id"]) |
| return None |
|
|
|
|
| def get_caption(record: dict[str, Any]) -> str | None: |
| for key in ("text", "caption", "answer"): |
| value = record.get(key) |
| if isinstance(value, str) and value.strip(): |
| return value.strip() |
| return None |
|
|
|
|
| def parse_dataset_sample(record: dict[str, Any]) -> DatasetSample | None: |
| sample_id = str(record.get("image_id", "")) |
| image_path = str(record.get("image_path", "")) |
| if not sample_id or not image_path: |
| return None |
|
|
| state = "" |
| stem = sample_id |
| for candidate_state, suffix in STATE_SUFFIXES.items(): |
| if sample_id.endswith(suffix): |
| state = candidate_state |
| stem = sample_id[: -len(suffix)] |
| break |
| if not state: |
| return None |
|
|
| if state == "origin": |
| coco_id = stem |
| base_id = None |
| else: |
| base_id = stem |
| match = INDEX_ID_RE.match(base_id) |
| if not match: |
| return None |
| coco_id = match.group(1) |
|
|
| return DatasetSample( |
| sample_id=sample_id, |
| state=state, |
| image_path=image_path, |
| question=record.get("question"), |
| base_id=base_id, |
| coco_id=coco_id, |
| ) |
|
|
|
|
| def parse_original_object(base_id: str) -> str: |
| match = MASKED_OBJECT_RE.search(base_id) |
| if not match: |
| raise ValueError(f"Could not parse original object from base id: {base_id}") |
| return match.group(1).replace("_", " ") |
|
|
|
|
| def parse_replacement_from_path(path: str) -> str | None: |
| stem = Path(path).stem |
| if "_insert_" not in stem: |
| return None |
| return stem.split("_insert_", 1)[1].replace("_", " ") |
|
|
|
|
| def load_insertions(path: Path) -> dict[str, dict[str, Any]]: |
| if not path.exists(): |
| return {} |
| records: dict[str, dict[str, Any]] = {} |
| for item in read_json_or_jsonl(path): |
| image_id = item.get("image_id") |
| if image_id: |
| records[str(image_id)] = item |
| return records |
|
|
|
|
| def load_dataset_triplets(dataset_file: Path, insertions_file: Path) -> tuple[list[Triplet], list[dict[str, Any]]]: |
| rows = read_json_or_jsonl(dataset_file) |
| origin_by_coco: dict[str, DatasetSample] = {} |
| removed_by_base: dict[str, DatasetSample] = {} |
| replaced_by_base: dict[str, DatasetSample] = {} |
| skipped: list[dict[str, Any]] = [] |
|
|
| for row in rows: |
| sample = parse_dataset_sample(row) |
| if sample is None: |
| skipped.append({"reason": "invalid_dataset_sample", "sample": row}) |
| continue |
| if sample.state == "origin": |
| origin_by_coco[sample.coco_id] = sample |
| elif sample.state == "removed" and sample.base_id: |
| removed_by_base[sample.base_id] = sample |
| elif sample.state == "replaced" and sample.base_id: |
| replaced_by_base[sample.base_id] = sample |
|
|
| insertions = load_insertions(insertions_file) |
| triplets: list[Triplet] = [] |
| for base_id, replaced in sorted(replaced_by_base.items()): |
| coco_match = INDEX_ID_RE.match(base_id) |
| if not coco_match: |
| skipped.append({"reason": "invalid_replaced_base_id", "base_id": base_id}) |
| continue |
| coco_id = coco_match.group(1) |
| origin = origin_by_coco.get(coco_id) |
| removed = removed_by_base.get(base_id) |
| insertion = insertions.get(base_id, {}) |
| replacement_object = insertion.get("replacement_object") or parse_replacement_from_path(replaced.image_path) |
|
|
| if origin is None or removed is None or not replacement_object: |
| skipped.append( |
| { |
| "reason": "incomplete_triplet", |
| "base_id": base_id, |
| "has_origin": origin is not None, |
| "has_removed": removed is not None, |
| "has_replacement_object": bool(replacement_object), |
| } |
| ) |
| continue |
|
|
| triplets.append( |
| Triplet( |
| base_id=base_id, |
| coco_id=coco_id, |
| original_object=insertion.get("original_object") or parse_original_object(base_id), |
| replacement_object=str(replacement_object), |
| origin=origin, |
| removed=removed, |
| replaced=replaced, |
| ) |
| ) |
| return triplets, skipped |
|
|
|
|
| def load_captions(inference_file: Path) -> dict[str, dict[str, Any]]: |
| captions: dict[str, dict[str, Any]] = {} |
| for row in read_json_or_jsonl(inference_file): |
| sample_id = get_record_id(row) |
| caption = get_caption(row) |
| if not sample_id or not caption: |
| continue |
| captions[sample_id] = {"caption": caption, "record": row} |
| return captions |
|
|
|
|
| def evaluate_tosc( |
| *, |
| inference_file: Path, |
| dataset_file: Path, |
| insertions_file: Path, |
| save_file: Path | None, |
| ) -> dict[str, Any]: |
| normalizer = ObjectNormalizer() |
| triplets, dataset_skipped = load_dataset_triplets(dataset_file, insertions_file) |
| captions = load_captions(inference_file) |
|
|
| missing_caption_skipped: list[dict[str, Any]] = [] |
| for triplet in triplets: |
| for state, sample in ( |
| ("origin", triplet.origin), |
| ("removed", triplet.removed), |
| ("replaced", triplet.replaced), |
| ): |
| if sample.sample_id not in captions: |
| missing_caption_skipped.append( |
| { |
| "reason": "missing_caption", |
| "base_id": triplet.base_id, |
| "state": state, |
| "sample_id": sample.sample_id, |
| } |
| ) |
| continue |
|
|
| details: list[dict[str, Any]] = [] |
| skipped: list[dict[str, Any]] = [*dataset_skipped, *missing_caption_skipped] |
| seen_missing_caption_triplets = {item["base_id"] for item in missing_caption_skipped} |
|
|
| for triplet in triplets: |
| if triplet.base_id in seen_missing_caption_triplets: |
| continue |
|
|
| state_outputs: dict[str, dict[str, Any]] = {} |
| for state, sample in ( |
| ("origin", triplet.origin), |
| ("removed", triplet.removed), |
| ("replaced", triplet.replaced), |
| ): |
| caption = captions[sample.sample_id]["caption"] |
| extraction = normalizer.extract_caption_objects(caption) |
| state_outputs[state] = { |
| "sample_id": sample.sample_id, |
| "image_path": sample.image_path, |
| "caption": caption, |
| **extraction, |
| } |
|
|
| original_canonical = normalizer.canonicalize(triplet.original_object) or normalize_phrase(triplet.original_object) |
| replacement_canonical = normalizer.canonicalize(triplet.replacement_object) or normalize_phrase( |
| triplet.replacement_object |
| ) |
|
|
| origin_objects = set(state_outputs["origin"]["normalized_objects"]) |
| removed_objects = set(state_outputs["removed"]["normalized_objects"]) |
| replaced_objects = set(state_outputs["replaced"]["normalized_objects"]) |
|
|
| m_orig_o = int(original_canonical in origin_objects) |
| m_rem_o = int(original_canonical in removed_objects) |
| m_rep_o = int(original_canonical in replaced_objects) |
| m_rep_r = int(replacement_canonical in replaced_objects) |
|
|
| details.append( |
| { |
| "base_id": triplet.base_id, |
| "coco_id": triplet.coco_id, |
| "original_object": triplet.original_object, |
| "original_object_canonical": original_canonical, |
| "replacement_object": triplet.replacement_object, |
| "replacement_object_canonical": replacement_canonical, |
| "states": state_outputs, |
| "indicators": { |
| "m_orig_o": m_orig_o, |
| "m_rem_o": m_rem_o, |
| "m_rep_o": m_rep_o, |
| "m_rep_r": m_rep_r, |
| }, |
| "contributions": { |
| "OPA": m_orig_o, |
| "RCA": 1 - m_rem_o, |
| "RUA": m_rep_r * (1 - m_rep_o), |
| "TOSC": m_orig_o * (1 - m_rem_o) * m_rep_r * (1 - m_rep_o), |
| "OldPersist_rem": m_rem_o, |
| "OldPersist_rep": m_rep_o, |
| "RepFail": 1 - m_rep_r, |
| "CC": (1 - m_rep_o) * (1 - m_rep_r), |
| "MixConf": m_rep_o * m_rep_r, |
| }, |
| } |
| ) |
|
|
| metrics = compute_metrics(details) |
| metrics.update( |
| { |
| "dataset_triplets": len(triplets), |
| "evaluated_triplets": len(details), |
| "skipped_triplets": len(skipped), |
| "object_extraction_source": "local_synonym_match", |
| } |
| ) |
| output = { |
| "metrics": metrics, |
| "detailed_results": details, |
| "skipped": skipped, |
| } |
| if save_file: |
| write_json(save_file, output) |
| summary_path = save_file.with_name(save_file.stem + "_summary.json") |
| write_json(summary_path, metrics) |
| return output |
|
|
|
|
| def compute_metrics(details: list[dict[str, Any]]) -> dict[str, Any]: |
| total = len(details) |
| metric_names = [ |
| "OPA", |
| "RCA", |
| "RUA", |
| "TOSC", |
| "OldPersist_rem", |
| "OldPersist_rep", |
| "RepFail", |
| "CC", |
| "MixConf", |
| ] |
| sums = {name: sum(item["contributions"][name] for item in details) for name in metric_names} |
| return { |
| "total_samples": total, |
| **{f"{name}_count": sums[name] for name in metric_names}, |
| **{name: (sums[name] / total if total else 0.0) for name in metric_names}, |
| **{f"{name}_percent": ((sums[name] / total * 100) if total else 0.0) for name in metric_names}, |
| } |
|
|
|
|
| def print_metrics(metrics: dict[str, Any]) -> None: |
| print("\n" + "=" * 70) |
| print("Triadic Object-State Consistency Benchmark") |
| print("=" * 70) |
| print(f"Dataset Triplets: {metrics.get('dataset_triplets', 0)}") |
| print(f"Evaluated Triplets: {metrics.get('evaluated_triplets', 0)}") |
| print(f"Skipped Triplets: {metrics.get('skipped_triplets', 0)}") |
| print("-" * 70) |
| for name in ("OPA", "RCA", "RUA", "TOSC", "OldPersist_rem", "OldPersist_rep", "RepFail", "CC", "MixConf"): |
| print(f"{name:16s}: {metrics.get(name, 0.0) * 100:6.2f}%") |
| print("=" * 70 + "\n") |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| parser = argparse.ArgumentParser(description="Evaluate Triadic Object-State Consistency.") |
| parser.add_argument("--inference_file", type=Path, required=True, help="Caption inference JSONL.") |
| parser.add_argument("--dataset_file", type=Path, default=DEFAULT_DATASET_FILE) |
| parser.add_argument("--insertions_file", type=Path, default=DEFAULT_INSERTIONS_FILE) |
| parser.add_argument("--save_file", type=Path, default=None) |
| return parser.parse_args() |
|
|
|
|
| def main() -> int: |
| args = parse_args() |
| output = evaluate_tosc( |
| inference_file=args.inference_file, |
| dataset_file=args.dataset_file, |
| insertions_file=args.insertions_file, |
| save_file=args.save_file, |
| ) |
| print_metrics(output["metrics"]) |
| return 0 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
