tests/test_coco_evaluator.py

from __future__ import annotations

import importlib
import importlib.util
import unittest
from unittest.mock import patch


class CocoEvaluatorRegressionTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        package = "hcl"
        cls.evaluator_module = importlib.import_module(f"{package}.evaluator")
        cls.pipeline_module = importlib.import_module(f"{package}.pipeline")

    @staticmethod
    def _example(task_id: str, task_type: str, answer: object) -> dict[str, object]:
        return {
            "task_id": task_id,
            "task_name": task_type,
            "task_type": task_type,
            "answer": answer,
            "visible_context": {
                "width": 100,
                "height": 100,
                "bbox_coordinate_system": "qwen_0_1000",
            },
            "metadata": {
                "hcl_task": task_type,
                "answer_bbox_coordinate_system": "pixel_xyxy",
                "prediction_bbox_coordinate_system": "qwen_0_1000",
            },
        }

    @unittest.skipUnless(importlib.util.find_spec("pycocotools"), "pycocotools is optional")
    def test_detection_uses_micro_f1_and_also_reports_coco_ap(self) -> None:
        example = self._example(
            "det-1",
            "coco_detection",
            [{"label": "person", "bbox": [10, 10, 50, 50]}],
        )
        prediction = {
            "task_id": "det-1",
            "answer": [
                {"label": "person", "bbox": [600, 600, 900, 900], "score": 0.9},
                {"label": "person", "bbox": [100, 100, 500, 500], "score": 0.8},
            ],
        }
        metrics = self.evaluator_module.ExactMatchEvaluator().evaluate(
            [example], [prediction]
        )["metrics"]

        self.assertNotIn("accuracy", metrics)
        self.assertEqual(metrics["detection_metric_source"], "pycocotools.COCOeval")
        self.assertAlmostEqual(metrics["detection_coco_ap"], 0.5)
        self.assertAlmostEqual(metrics["detection_coco_ap50"], 0.5)
        self.assertAlmostEqual(metrics["detection_micro_precision_iou50"], 0.5)
        self.assertAlmostEqual(metrics["detection_micro_recall_iou50"], 1.0)
        self.assertAlmostEqual(metrics["detection_micro_f1_iou50"], 2.0 / 3.0)
        self.assertEqual(metrics["primary_metric_name"], "detection_micro_f1_iou50")
        self.assertAlmostEqual(metrics["primary_score"], 2.0 / 3.0)

    def test_grounding_converts_qwen_coordinates_and_reports_mean_iou(self) -> None:
        example = self._example(
            "ref-1", "refcoco_grounding", {"bbox": [10, 20, 50, 80]}
        )
        prediction = {"task_id": "ref-1", "answer": {"bbox": [100, 200, 500, 800]}}
        metrics = self.evaluator_module.ExactMatchEvaluator().evaluate(
            [example], [prediction]
        )["metrics"]

        self.assertNotIn("accuracy", metrics)
        self.assertEqual(metrics["grounding_accuracy_iou50"], 1.0)
        self.assertAlmostEqual(metrics["grounding_mean_iou"], 1.0)
        self.assertEqual(metrics["primary_metric_name"], "grounding_accuracy_iou50")

    @unittest.skipUnless(importlib.util.find_spec("pycocotools"), "pycocotools is optional")
    def test_gate_top_level_is_current_task_and_anchors_stay_per_task(self) -> None:
        current = self._example(
            "ref-1", "refcoco_grounding", {"bbox": [10, 20, 50, 80]}
        )
        anchor = self._example(
            "det-1",
            "coco_detection",
            [{"label": "person", "bbox": [10, 10, 50, 50]}],
        )
        predictions = [
            {"task_id": "ref-1", "answer": {"bbox": [100, 200, 500, 800]}},
            {
                "task_id": "det-1",
                "answer": [
                    {"label": "person", "bbox": [100, 100, 500, 500]},
                    {"label": "dog", "bbox": [600, 600, 900, 900]},
                ],
            },
        ]
        pipeline = self.pipeline_module.HCLPipeline.__new__(self.pipeline_module.HCLPipeline)
        pipeline.evaluator = self.evaluator_module.ExactMatchEvaluator()
        result = pipeline._evaluate_gate(
            [current, anchor],
            predictions,
            current_val_examples=[current],
            historical_anchor_examples=[anchor],
            split="val",
        )

        self.assertEqual(result["metric_scope"], "current_task")
        self.assertEqual(result["metrics"]["primary_score"], 1.0)
        self.assertEqual(
            result["combined_gate_metrics"]["primary_metric_name"],
            "mixed_tasks_not_comparable",
        )
        self.assertIn("coco_detection", result["historical_anchor_task_metrics"])
        self.assertEqual(
            result["historical_anchor_metrics"]["primary_metric_name"],
            "historical_tasks_reported_separately",
        )
        self.assertIsNone(result["historical_anchor_metrics"]["primary_score"])
        self.assertIsNone(result["combined_gate_metrics"]["mean_example_score"])

    def test_caption_composite_is_named_and_not_exposed_as_accuracy(self) -> None:
        example = self._example("cap-1", "coco_caption", "a red bus")
        prediction = {"task_id": "cap-1", "answer": "a bus"}
        fake_caption_eval = (
            {"cap-1": 0.42},
            {
                "caption_metric_source": "test",
                "caption_count": 1,
                "caption_primary_score": 0.42,
                "caption_CIDEr": 0.7,
            },
        )
        with patch.object(
            self.evaluator_module,
            "_official_caption_eval_scores",
            return_value=fake_caption_eval,
        ):
            metrics = self.evaluator_module.ExactMatchEvaluator().evaluate(
                [example], [prediction]
            )["metrics"]

        self.assertNotIn("accuracy", metrics)
        self.assertEqual(metrics["primary_metric_name"], "caption_composite_score")
        self.assertEqual(metrics["primary_score"], 0.42)
        self.assertEqual(metrics["caption_CIDEr"], 0.7)

    def test_checkpoint_signatures_include_evaluation_schema(self) -> None:
        version = self.evaluator_module.EVALUATION_SCHEMA_VERSION
        train_signature = self.pipeline_module._train_checkpoint_signature(
            split="train",
            examples=[],
            val_examples=[],
            batch_count=0,
            batchsize=None,
        )
        test_signature = self.pipeline_module._test_checkpoint_signature(
            split="test", examples=[]
        )
        continual_signature = self.pipeline_module._continual_checkpoint_signature(
            task_stream_path="stream.json",
            task_order=[],
            train_split="train",
            val_split="validation",
            test_split="test",
            train_limit_per_task=None,
            validation_limit_per_task=None,
            test_limit_per_task=None,
            batchsize=None,
        )
        self.assertEqual(train_signature["evaluation_schema_version"], version)
        self.assertEqual(test_signature["evaluation_schema_version"], version)
        self.assertEqual(continual_signature["evaluation_schema_version"], version)

    def test_paper_anchor_thresholds_and_schema_validity(self) -> None:
        caption = self._example("cap", "coco_caption", "a red bus")
        vqa = self._example("vqa", "vqav2", "red")
        vqa["metadata"]["answers"] = ["red", "red", "blue"]

        self.assertEqual(self.evaluator_module._correct_threshold(caption), 0.5)
        self.assertEqual(self.evaluator_module._correct_threshold(vqa), 1.0)

        malformed_grounding = self._example(
            "ground", "refcoco_grounding", {"bbox": [10, 20, 50, 80]}
        )
        result = self.evaluator_module.ExactMatchEvaluator().evaluate(
            [malformed_grounding],
            [{"task_id": "ground", "answer": "not a box"}],
        )["metrics"]
        self.assertEqual(result["correct"], 0)
        self.assertEqual(result["format_compliant_count"], 0)


if __name__ == "__main__":
    unittest.main()