rpx_benchmark.vqa.evaluate

 1"""Run any callable VQA model with the published RPX prompts and scoring."""
 2from __future__ import annotations
 3
 4import hashlib
 5import json
 6import time
 7from dataclasses import asdict
 8from pathlib import Path
 9from typing import Callable
10
11from ..exceptions import AdapterError, ConfigError, ManifestError
12from .contract import load_manifest
13from .hub_rgb import fetch_images
14from .metrics import score_predictions
15from .outputs import ParsedOutput, parse_output
16from .prompts import build_prompt
17
18
19def evaluate_vqa(
20    manifest_path: str | Path,
21    predict: Callable[[tuple[Path, ...], str, int, str], str],
22    *,
23    model_name: str,
24    output_dir: str | Path,
25    image_cache: str | Path,
26    model_revision: str,
27) -> dict:
28    """Evaluate a local model or API callable, without changing the model roster.
29
30    ``predict(image_paths, prompt, max_new_tokens, output_kind)`` returns raw
31    text. Only input images and the public prompt reach the callable; answers,
32    evidence and target boxes remain inside the evaluator. Two-image requests
33    preserve reference-then-target order and use the verified reference crop.
34
35    The generic bbox contract is one JSON object with ``label`` and ``bbox``
36    (XYXY, normalized 0–1000). Binary questions require ``yes`` or ``no``.
37    Malformed outputs and inference exceptions score as failures and remain
38    in the denominator. Dataset download errors abort rather than dropping rows.
39    Use a fresh output directory for each run; no silent overwrites or resume.
40    """
41    if not model_name.strip() or not model_revision.strip():
42        raise ConfigError("model_name and model_revision must identify the evaluated model")
43    manifest_path = Path(manifest_path)
44    samples = load_manifest(manifest_path)
45    if not samples:
46        raise ManifestError("VQA manifest contains no samples")
47    output = Path(output_dir)
48    if output.exists() and any(output.iterdir()):
49        raise ConfigError(f"Use an empty output directory: {output}")
50    output.mkdir(parents=True, exist_ok=True)
51    config = {
52        "model": model_name,
53        "model_revision": model_revision,
54        "manifest_sha256": hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
55        "num_samples": len(samples),
56        "protocol": "generic_single_call",
57        "timing": "end_to_end_callable_wall_ms_including_API_transport",
58    }
59    (output / "run_config.json").write_text(json.dumps(config, indent=2) + "\n")
60    scored = []
61    with (output / "predictions.jsonl").open("w", encoding="utf-8") as handle:
62        for sample in samples:
63            images = fetch_images(sample, image_cache)
64            prompt = build_prompt(sample, "custom")
65            started = time.perf_counter()
66            raw = None
67            error = None
68            try:
69                raw = predict(images, prompt.text, prompt.max_new_tokens, prompt.output_kind)
70                if not isinstance(raw, str):
71                    raise AdapterError("The VQA callable must return raw text")
72                parsed = parse_output(sample, raw, "custom")
73            except Exception as exc:
74                error = f"{type(exc).__name__}: {exc}"
75                parsed = ParsedOutput(False, error=error)
76            elapsed_ms = (time.perf_counter() - started) * 1000
77            scored.append((sample, parsed))
78            row = {
79                "sample_id": sample.sample_id,
80                "raw_output": raw if isinstance(raw, str) else None,
81                "parsed": asdict(parsed),
82                "inference_error": error,
83                "callable_wall_ms": elapsed_ms,
84            }
85            handle.write(json.dumps(row, allow_nan=False) + "\n")
86            handle.flush()
87    report = {**config, "metrics": score_predictions(scored)}
88    (output / "result.json").write_text(json.dumps(report, indent=2, allow_nan=False) + "\n")
89    return report