#!/usr/bin/env python3 """Audit V* benchmark preparation and analyze errors against target scale.""" from __future__ import annotations import argparse import csv import json import math import re from collections import Counter, defaultdict from pathlib import Path from statistics import mean, median from PIL import Image def read_jsonl(path: Path) -> list[dict]: return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] def quantile(values: list[float], probability: float) -> float: ordered = sorted(values) if not ordered: raise ValueError("Cannot compute a quantile of an empty list") position = (len(ordered) - 1) * probability lower = math.floor(position) upper = math.ceil(position) if lower == upper: return float(ordered[lower]) return float(ordered[lower] * (upper - position) + ordered[upper] * (position - lower)) def summary(values: list[float]) -> dict[str, float]: return { "min": min(values), "p10": quantile(values, 0.10), "p25": quantile(values, 0.25), "median": median(values), "mean": mean(values), "p75": quantile(values, 0.75), "p90": quantile(values, 0.90), "max": max(values), } def normalize_text(value: str) -> str: return re.sub(r"[^a-z0-9]+", " ", value.lower()).strip() def accuracy_group(rows: list[dict]) -> dict: samples = len(rows) correct = sum(bool(row["correct"]) for row in rows) accuracy = correct / samples if rows else None lower = None upper = None if samples: z = 1.959963984540054 denominator = 1 + z * z / samples center = (accuracy + z * z / (2 * samples)) / denominator margin = ( z * math.sqrt(accuracy * (1 - accuracy) / samples + z * z / (4 * samples * samples)) / denominator ) lower = center - margin upper = center + margin return { "samples": samples, "correct": correct, "accuracy": accuracy, "wilson_95ci": [lower, upper], } def grouped(rows: list[dict], key: str) -> dict[str, dict]: groups: dict[str, list[dict]] = defaultdict(list) for row in rows: groups[str(row[key])].append(row) return {name: accuracy_group(group) for name, group in sorted(groups.items())} def binned(rows: list[dict], key: str, boundaries: list[float]) -> dict[str, dict]: groups: dict[str, list[dict]] = defaultdict(list) for row in rows: value = float(row[key]) lower = -math.inf label = "" for upper in boundaries: if value <= upper: label = f"({lower if math.isfinite(lower) else '-inf'}, {upper}]" break lower = upper if not label: label = f"({boundaries[-1]}, inf)" groups[label].append(row) return {name: accuracy_group(group) for name, group in groups.items()} def mcnemar_exact(discordant_a_only: int, discordant_b_only: int) -> float: trials = discordant_a_only + discordant_b_only if trials == 0: return 1.0 lower = min(discordant_a_only, discordant_b_only) probability = sum(math.comb(trials, index) for index in range(lower + 1)) / (2**trials) return min(1.0, 2 * probability) def paired_comparison(rows: list[dict], first_key: str, second_key: str) -> dict: first_only = sum(bool(row[first_key]) and not bool(row[second_key]) for row in rows) second_only = sum(not bool(row[first_key]) and bool(row[second_key]) for row in rows) both_correct = sum(bool(row[first_key]) and bool(row[second_key]) for row in rows) both_wrong = sum(not bool(row[first_key]) and not bool(row[second_key]) for row in rows) first_accuracy = sum(bool(row[first_key]) for row in rows) / len(rows) second_accuracy = sum(bool(row[second_key]) for row in rows) / len(rows) return { "samples": len(rows), "first_accuracy": first_accuracy, "second_accuracy": second_accuracy, "difference_percentage_points": (second_accuracy - first_accuracy) * 100, "both_correct": both_correct, "both_wrong": both_wrong, "first_only_correct": first_only, "second_only_correct": second_only, "mcnemar_exact_two_sided_p": mcnemar_exact(first_only, second_only), } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--manifest", type=Path, required=True) parser.add_argument("--results", type=Path, required=True) parser.add_argument("--dynamic-results", type=Path) parser.add_argument("--oracle-results", type=Path) parser.add_argument("--raw-root", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--eval-width", type=int, default=448) parser.add_argument("--eval-height", type=int, default=448) args = parser.parse_args() manifest = read_jsonl(args.manifest) results = read_jsonl(args.results) manifest_by_id = {str(row["sample_id"]): row for row in manifest} results_by_id = {str(row["sample_id"]): row for row in results} if len(manifest_by_id) != len(manifest) or len(results_by_id) != len(results): raise ValueError("Duplicate sample IDs detected") if manifest_by_id.keys() != results_by_id.keys(): raise ValueError("Manifest/result sample IDs differ") diagnostic_results = {} for name, path in (("dynamic", args.dynamic_results), ("oracle", args.oracle_results)): if path is None: continue experiment_rows = read_jsonl(path) experiment_by_id = {str(row["sample_id"]): row for row in experiment_rows} if len(experiment_by_id) != len(experiment_rows): raise ValueError(f"Duplicate sample IDs in {name} results") if experiment_by_id.keys() != manifest_by_id.keys(): raise ValueError(f"Manifest/{name} result sample IDs differ") diagnostic_results[name] = experiment_by_id rows: list[dict] = [] mapping_failures: list[dict] = [] dimension_mismatches: list[dict] = [] for sample_id in sorted(manifest_by_id, key=int): item = manifest_by_id[sample_id] result = results_by_id[sample_id] image_path = Path(item["images"][0]) annotation_path = args.raw_root / str(item["config"]) / f"{image_path.stem}.json" annotation = json.loads(annotation_path.read_text(encoding="utf-8")) with Image.open(image_path) as image: width, height = image.size recorded_width, recorded_height = item["image_sizes"][0] if (width, height) != (recorded_width, recorded_height): dimension_mismatches.append( { "sample_id": sample_id, "actual": [width, height], "manifest": [recorded_width, recorded_height], } ) expected_index = ord(str(item["answer"]).upper()) - ord("A") expected_choice = str(item["choices"][expected_index]) official_correct_sentence = str(annotation["options"][0]) choice_norm = normalize_text(expected_choice) sentence_norm = normalize_text(official_correct_sentence) mapping_valid = choice_norm in sentence_norm or sentence_norm in choice_norm if not mapping_valid: mapping_failures.append( { "sample_id": sample_id, "choice": expected_choice, "official_correct_sentence": official_correct_sentence, } ) scale_x = args.eval_width / width scale_y = args.eval_height / height boxes = annotation["bbox"] projected_boxes = [ [float(x) * scale_x, float(y) * scale_y, float(w) * scale_x, float(h) * scale_y] for x, y, w, h in boxes ] min_projected_side = min(min(box[2], box[3]) for box in projected_boxes) min_projected_area = min(box[2] * box[3] for box in projected_boxes) min_area_fraction = min((float(w) * float(h)) / (width * height) for _, _, w, h in boxes) target_centers = [ ((float(x) + float(w) / 2) / width, (float(y) + float(h) / 2) / height) for x, y, w, h in boxes ] max_center_distance = max( math.sqrt(((cx - 0.5) / 0.5) ** 2 + ((cy - 0.5) / 0.5) ** 2) for cx, cy in target_centers ) predicted = result.get("predicted") predicted_choice = None if predicted: predicted_index = ord(str(predicted).upper()) - ord("A") if 0 <= predicted_index < len(item["choices"]): predicted_choice = item["choices"][predicted_index] question = str(item["question"]).lower() if "color" in question: question_type = "color" elif "material" in question: question_type = "material" elif str(item["config"]) == "relative_position": question_type = "relative_position" elif "kind of" in question or "what animal" in question: question_type = "object_kind" else: question_type = "other_attribute" output_row = { "sample_id": sample_id, "config": item["config"], "question_type": question_type, "question": item["question"], "image_path": str(image_path), "image_width": width, "image_height": height, "image_megapixels": width * height / 1_000_000, "target_count": len(boxes), "target_objects": " | ".join(annotation["target_object"]), "min_target_area_fraction": min_area_fraction, "min_projected_side_448": min_projected_side, "min_projected_area_448": min_projected_area, "max_center_distance": max_center_distance, "choice_count": len(item["choices"]), "expected": result["expected"], "expected_choice": expected_choice, "predicted": predicted, "predicted_choice": predicted_choice, "correct": bool(result["correct"]), "mapping_valid": mapping_valid, "error": result.get("error"), } for experiment_name, experiment_by_id in diagnostic_results.items(): experiment = experiment_by_id[sample_id] output_row[f"{experiment_name}_predicted"] = experiment.get("predicted") output_row[f"{experiment_name}_correct"] = bool(experiment.get("correct")) output_row[f"{experiment_name}_error"] = experiment.get("error") rows.append(output_row) labels = Counter(row["expected"] for row in rows) predictions = Counter(row["predicted"] for row in rows) confusion: dict[str, Counter] = defaultdict(Counter) for row in rows: confusion[str(row["expected"])][str(row["predicted"])] += 1 task_groups: dict[str, list[dict]] = defaultdict(list) for row in rows: task_groups[str(row["config"])].append(row) chance_by_task = { task: mean(1 / int(row["choice_count"]) for row in task_rows) for task, task_rows in sorted(task_groups.items()) } overall_random_chance = mean(1 / int(row["choice_count"]) for row in rows) tiny_threshold_counts = {} for threshold in (4, 8, 12, 16, 24, 32): subset = [row for row in rows if float(row["min_projected_side_448"]) <= threshold] tiny_threshold_counts[f"le_{threshold}px"] = accuracy_group(subset) scale_by_config = {} for config, config_rows in sorted(task_groups.items()): correct_rows = [row for row in config_rows if row["correct"]] wrong_rows = [row for row in config_rows if not row["correct"]] scale_by_config[config] = { "accuracy_by_projected_min_side_bin": binned( config_rows, "min_projected_side_448", [4, 8, 12, 16, 24, 32] ), "min_projected_side_correct": summary( [float(row["min_projected_side_448"]) for row in correct_rows] ), "min_projected_side_wrong": summary( [float(row["min_projected_side_448"]) for row in wrong_rows] ), } wrong_smallest = sorted( (row for row in rows if not row["correct"]), key=lambda row: float(row["min_projected_side_448"]), )[:20] correct_smallest = sorted( (row for row in rows if row["correct"]), key=lambda row: float(row["min_projected_side_448"]), )[:10] report = { "scope": { "samples": len(rows), "manifest": str(args.manifest.resolve()), "results": str(args.results.resolve()), "raw_root": str(args.raw_root.resolve()), "simulated_eval_size": [args.eval_width, args.eval_height], }, "integrity": { "unique_manifest_ids": len(manifest_by_id), "unique_result_ids": len(results_by_id), "id_sets_equal": manifest_by_id.keys() == results_by_id.keys(), "dimension_mismatches": dimension_mismatches, "label_mapping_failures": mapping_failures, "inference_errors": sum(bool(row["error"]) for row in rows), "unparsed": sum(row["predicted"] is None for row in rows), }, "performance": { "overall": accuracy_group(rows), "by_config": grouped(rows, "config"), "by_question_type": grouped(rows, "question_type"), "random_choice_chance": overall_random_chance, "random_choice_chance_by_config": chance_by_task, "expected_label_distribution": dict(sorted(labels.items())), "predicted_label_distribution": dict(sorted(predictions.items())), "confusion": { expected: dict(sorted(predicted.items())) for expected, predicted in sorted(confusion.items()) }, }, "image_and_target_scale": { "image_width": summary([float(row["image_width"]) for row in rows]), "image_height": summary([float(row["image_height"]) for row in rows]), "image_megapixels": summary([float(row["image_megapixels"]) for row in rows]), "min_projected_target_side_448": summary( [float(row["min_projected_side_448"]) for row in rows] ), "min_projected_target_area_448": summary( [float(row["min_projected_area_448"]) for row in rows] ), "min_target_area_fraction": summary( [float(row["min_target_area_fraction"]) for row in rows] ), "accuracy_by_projected_min_side_bin": binned( rows, "min_projected_side_448", [4, 8, 12, 16, 24, 32] ), "accuracy_by_image_megapixels": binned(rows, "image_megapixels", [2, 3, 4, 6]), "accuracy_by_target_center_distance": binned( rows, "max_center_distance", [0.5, 0.8, 1.0, 1.2] ), "cumulative_tiny_target_thresholds": tiny_threshold_counts, "by_config": scale_by_config, }, "smallest_wrong_samples": wrong_smallest, "smallest_correct_samples": correct_smallest, } if diagnostic_results: sensitivity = {} for experiment_name in diagnostic_results: correct_key = f"{experiment_name}_correct" sensitivity[experiment_name] = { "overall": accuracy_group( [{**row, "correct": row[correct_key]} for row in rows] ), "by_config": { config: accuracy_group( [{**row, "correct": row[correct_key]} for row in config_rows] ) for config, config_rows in sorted(task_groups.items()) }, "accuracy_by_baseline_projected_min_side_bin": binned( [{**row, "correct": row[correct_key]} for row in rows], "min_projected_side_448", [4, 8, 12, 16, 24, 32], ), "errors": sum(bool(row[f"{experiment_name}_error"]) for row in rows), } if "dynamic" in diagnostic_results: sensitivity["baseline_vs_dynamic"] = paired_comparison( rows, "correct", "dynamic_correct" ) if "oracle" in diagnostic_results: sensitivity["baseline_vs_oracle"] = paired_comparison( rows, "correct", "oracle_correct" ) if "dynamic" in diagnostic_results and "oracle" in diagnostic_results: sensitivity["dynamic_vs_oracle"] = paired_comparison( rows, "dynamic_correct", "oracle_correct" ) report["sensitivity_experiments"] = sensitivity args.output_dir.mkdir(parents=True, exist_ok=True) (args.output_dir / "diagnosis.json").write_text( json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) fieldnames = list(rows[0]) with (args.output_dir / "sample_diagnostics.csv").open( "w", encoding="utf-8", newline="" ) as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True)) if __name__ == "__main__": main()