Add detection model promotion report

scripts/build_detection_model_promotion_report.py

@@ -0,0 +1,279 @@
+#!/usr/bin/env python3
+"""Build an operator-only detection model promotion report.
+
+The report combines persisted positive-AOI QA portfolio evidence with
+hard-negative/background detection-count summaries. It does not run inference,
+mutate application state, download models or change active model configuration.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from statistics import mean
+from typing import Any
+
+
+@dataclass(frozen=True)
+class CandidateKey:
+    model_asset_id: str
+    tile_size: int
+    tile_overlap: int
+    threshold: float
+
+    @property
+    def label(self) -> str:
+        threshold = ("%f" % self.threshold).rstrip("0").rstrip(".")
+        return f"{self.model_asset_id}|{self.tile_size}|{self.tile_overlap}|{threshold}"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Build a detection model promotion report from positive and hard-negative evidence."
+    )
+    parser.add_argument("--positive-portfolio", required=True, help="Path to calibration_evidence_portfolio.json")
+    parser.add_argument(
+        "--hard-negative-summary",
+        action="append",
+        required=True,
+        help="Path to hard_negative_matrix_summary.json. May be supplied multiple times.",
+    )
+    parser.add_argument("--output-dir", required=True, help="Directory for JSON and Markdown report output.")
+    parser.add_argument("--min-positive-samples", type=int, default=3)
+    parser.add_argument("--min-background-samples", type=int, default=3)
+    parser.add_argument("--min-mean-f1", type=float, default=0.25)
+    parser.add_argument("--max-background-detections-per-sample", type=int, default=0)
+    return parser.parse_args()
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    if not path.is_file():
+        raise SystemExit(f"JSON input is not readable: {path}")
+    return json.loads(path.read_text(encoding="utf-8-sig"))
+
+
+def as_candidate_key(item: dict[str, Any]) -> CandidateKey | None:
+    model_asset_id = item.get("model_asset_id") or item.get("model_request")
+    if not model_asset_id:
+        return None
+    try:
+        return CandidateKey(
+            model_asset_id=str(model_asset_id),
+            tile_size=int(item.get("tile_size")),
+            tile_overlap=int(item.get("tile_overlap")),
+            threshold=float(item.get("threshold")),
+        )
+    except (TypeError, ValueError):
+        return None
+
+
+def numeric(item: dict[str, Any], key: str) -> float | None:
+    value = item.get(key)
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def collect_positive_runs(portfolio: dict[str, Any]) -> dict[CandidateKey, list[dict[str, Any]]]:
+    best_by_candidate_sample: dict[tuple[CandidateKey, str], dict[str, Any]] = {}
+    for sample in portfolio.get("samples") or []:
+        sample_slug = str(sample.get("sample_slug") or "unknown")
+        for run in sample.get("runs") or []:
+            key = as_candidate_key(run)
+            f1 = numeric(run, "f1_score")
+            if key is None or f1 is None:
+                continue
+            enriched = dict(run)
+            enriched["sample_slug"] = sample_slug
+            existing = best_by_candidate_sample.get((key, sample_slug))
+            existing_f1 = numeric(existing or {}, "f1_score")
+            if existing is None or existing_f1 is None or f1 > existing_f1:
+                best_by_candidate_sample[(key, sample_slug)] = enriched
+
+    grouped: dict[CandidateKey, list[dict[str, Any]]] = defaultdict(list)
+    for (key, _sample_slug), run in best_by_candidate_sample.items():
+        grouped[key].append(run)
+    return grouped
+
+
+def collect_background_runs(summary_paths: list[Path]) -> dict[CandidateKey, list[dict[str, Any]]]:
+    best_by_candidate_sample: dict[tuple[CandidateKey, str], dict[str, Any]] = {}
+    for path in summary_paths:
+        payload = read_json(path)
+        for item in payload.get("items") or []:
+            key = as_candidate_key(item)
+            sample_slug = str(item.get("sample_slug") or "unknown")
+            if key is None:
+                continue
+            enriched = dict(item)
+            enriched["source_summary_path"] = str(path)
+            detection_count = int(enriched.get("detection_count") or 0)
+            existing = best_by_candidate_sample.get((key, sample_slug))
+            existing_count = int((existing or {}).get("detection_count") or 0)
+            if existing is None or detection_count > existing_count:
+                best_by_candidate_sample[(key, sample_slug)] = enriched
+
+    grouped: dict[CandidateKey, list[dict[str, Any]]] = defaultdict(list)
+    for (key, _sample_slug), item in best_by_candidate_sample.items():
+        grouped[key].append(item)
+    return grouped
+
+
+def average_metric(runs: list[dict[str, Any]], metric: str) -> float | None:
+    values = [numeric(run, metric) for run in runs]
+    values = [value for value in values if value is not None]
+    return mean(values) if values else None
+
+
+def build_decisions(args: argparse.Namespace) -> dict[str, Any]:
+    portfolio_path = Path(args.positive_portfolio)
+    positive_portfolio = read_json(portfolio_path)
+    background_paths = [Path(path) for path in args.hard_negative_summary]
+    positive = collect_positive_runs(positive_portfolio)
+    background = collect_background_runs(background_paths)
+
+    decisions = []
+    for key in sorted(set(positive) | set(background), key=lambda item: item.label):
+        positive_runs = positive.get(key, [])
+        background_runs = background.get(key, [])
+        f1_values = [numeric(run, "f1_score") for run in positive_runs]
+        f1_values = [value for value in f1_values if value is not None]
+        background_counts = [int(run.get("detection_count") or 0) for run in background_runs]
+
+        mean_f1 = mean(f1_values) if f1_values else None
+        min_f1 = min(f1_values) if f1_values else None
+        total_background = sum(background_counts)
+        max_background = max(background_counts, default=None)
+
+        rejection_reasons: list[str] = []
+        if len(positive_runs) < args.min_positive_samples:
+            rejection_reasons.append("insufficient_positive_samples")
+        if len(background_runs) < args.min_background_samples:
+            rejection_reasons.append("insufficient_background_samples")
+        if mean_f1 is None or mean_f1 < args.min_mean_f1:
+            rejection_reasons.append("positive_mean_f1_below_gate")
+        if max_background is None or max_background > args.max_background_detections_per_sample:
+            rejection_reasons.append("background_false_positive_pressure")
+
+        decisions.append(
+            {
+                "candidate_key": key.label,
+                "model_asset_id": key.model_asset_id,
+                "tile_size": key.tile_size,
+                "tile_overlap": key.tile_overlap,
+                "threshold": key.threshold,
+                "promotion_status": "reject" if rejection_reasons else "promote_candidate",
+                "rejection_reasons": rejection_reasons,
+                "positive_sample_count": len(positive_runs),
+                "background_sample_count": len(background_runs),
+                "mean_f1": mean_f1,
+                "min_f1": min_f1,
+                "mean_precision": average_metric(positive_runs, "precision"),
+                "mean_recall": average_metric(positive_runs, "recall"),
+                "total_background_detections": total_background,
+                "max_background_detections": max_background,
+                "positive_samples": sorted(str(run.get("sample_slug")) for run in positive_runs),
+                "background_samples": sorted(
+                    {
+                        str(run.get("sample_slug")): int(run.get("detection_count") or 0)
+                        for run in background_runs
+                    }.items()
+                ),
+            }
+        )
+
+    promoted = [item for item in decisions if item["promotion_status"] == "promote_candidate"]
+    recommended = max(
+        promoted,
+        key=lambda item: (
+            item["mean_f1"] if item["mean_f1"] is not None else float("-inf"),
+            -(item["total_background_detections"] or 0),
+        ),
+        default=None,
+    )
+
+    return {
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "positive_portfolio_path": str(portfolio_path),
+        "hard_negative_summary_paths": [str(path) for path in background_paths],
+        "gates": {
+            "min_positive_samples": args.min_positive_samples,
+            "min_background_samples": args.min_background_samples,
+            "min_mean_f1": args.min_mean_f1,
+            "max_background_detections_per_sample": args.max_background_detections_per_sample,
+        },
+        "candidate_count": len(decisions),
+        "recommended_candidate": recommended,
+        "candidate_decisions": decisions,
+    }
+
+
+def write_markdown(report: dict[str, Any], path: Path) -> None:
+    lines = [
+        "# Detection Model Promotion Report",
+        "",
+        f"- Generated: {report['generated_at']}",
+        f"- Positive portfolio: `{report['positive_portfolio_path']}`",
+        f"- Hard-negative summaries: {len(report['hard_negative_summary_paths'])}",
+        f"- Candidates: {report['candidate_count']}",
+        "",
+        "## Gates",
+        "",
+    ]
+    for key, value in report["gates"].items():
+        lines.append(f"- {key}: `{value}`")
+    lines.extend(["", "## Recommendation", ""])
+    recommended = report.get("recommended_candidate")
+    if recommended:
+        lines.append(f"- Promote candidate for operator review: `{recommended['candidate_key']}`")
+    else:
+        lines.append("- No candidate passed all positive and hard-negative gates.")
+    lines.extend(["", "## Candidate Decisions", ""])
+    for item in report["candidate_decisions"]:
+        reasons = ", ".join(item["rejection_reasons"]) or "none"
+        lines.extend(
+            [
+                f"### `{item['candidate_key']}`",
+                "",
+                f"- Status: `{item['promotion_status']}`",
+                f"- Rejection reasons: `{reasons}`",
+                f"- Positive samples: `{item['positive_sample_count']}`",
+                f"- Background samples: `{item['background_sample_count']}`",
+                f"- Mean F1: `{item['mean_f1']}`",
+                f"- Mean precision: `{item['mean_precision']}`",
+                f"- Mean recall: `{item['mean_recall']}`",
+                f"- Max background detections: `{item['max_background_detections']}`",
+                f"- Total background detections: `{item['total_background_detections']}`",
+                "",
+            ]
+        )
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def main() -> None:
+    args = parse_args()
+    output_dir = Path(args.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    report = build_decisions(args)
+    report_path = output_dir / "detection_model_promotion_report.json"
+    markdown_path = output_dir / "detection_model_promotion_report.md"
+    report_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")
+    write_markdown(report, markdown_path)
+
+    print("Detection model promotion report passed")
+    print(f"Report JSON: {report_path}")
+    print(f"Report Markdown: {markdown_path}")
+    recommended = report.get("recommended_candidate")
+    print(f"Recommended candidate: {recommended['candidate_key'] if recommended else 'none'}")
+
+
+if __name__ == "__main__":
+    main()