geointel/scripts/build_mol_operational_benchmark_report.py

from __future__ import annotations

import argparse
from collections import defaultdict
from datetime import datetime, timezone
import json
from pathlib import Path
from statistics import fmean
from typing import Any


CandidateKey = tuple[str, int, int, float]


def load_json(path: Path) -> dict[str, Any]:
    if not path.is_file():
        raise SystemExit(f"Input is not readable: {path}")
    try:
        payload = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, json.JSONDecodeError) as exc:
        raise SystemExit(f"Input is not valid JSON: {path}: {exc}") from exc
    if not isinstance(payload, dict):
        raise SystemExit(f"Input root must be an object: {path}")
    return payload


def candidate_key(item: dict[str, Any]) -> CandidateKey:
    model_asset_id = str(item.get("model_asset_id") or "").strip()
    if not model_asset_id:
        raise SystemExit("Benchmark item is missing model_asset_id")
    try:
        tile_size = int(item["tile_size"])
        tile_overlap = int(item["tile_overlap"])
        threshold = round(float(item["threshold"]), 8)
    except (KeyError, TypeError, ValueError) as exc:
        raise SystemExit(f"Benchmark item has an invalid model configuration: {item}") from exc
    return model_asset_id, tile_size, tile_overlap, threshold


def candidate_label(key: CandidateKey) -> str:
    return f"{key[0]}|{key[1]}|{key[2]}|{key[3]:.8g}"


def number(item: dict[str, Any], key: str) -> float:
    value = item.get(key)
    if value is None:
        raise SystemExit(f"Positive benchmark item is missing {key}: {item.get('sample_slug')}")
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise SystemExit(f"Positive benchmark item has invalid {key}: {item.get('sample_slug')}") from exc


def integer(item: dict[str, Any], key: str) -> int:
    value = item.get(key)
    if value is None:
        raise SystemExit(f"Benchmark item is missing {key}: {item.get('sample_slug')}")
    try:
        return int(value)
    except (TypeError, ValueError) as exc:
        raise SystemExit(f"Benchmark item has invalid {key}: {item.get('sample_slug')}") from exc


def coverage_ratio(item: dict[str, Any]) -> float:
    explicit = item.get("reference_coverage_ratio")
    if explicit is not None:
        return float(explicit)
    raw_count = integer(item, "reference_raw_count")
    evaluated_count = integer(item, "reference_evaluated_count")
    if raw_count < 1:
        raise SystemExit(f"Positive benchmark item has no raw references: {item.get('sample_slug')}")
    return evaluated_count / raw_count


def gate(name: str, passed: bool, observed: Any, required: str) -> dict[str, Any]:
    return {"name": name, "passed": bool(passed), "observed": observed, "required": required}


def zone_row(item: dict[str, Any], manifest_samples: dict[str, dict[str, Any]]) -> dict[str, Any]:
    slug = str(item.get("sample_slug") or "").strip().lower()
    if not slug:
        raise SystemExit("Positive benchmark item is missing sample_slug")
    manifest = manifest_samples.get(slug, {})
    matches = integer(item, "matches")
    false_positives = integer(item, "false_positives")
    false_negatives = integer(item, "false_negatives")
    strict_matches = item.get("strict_matches")
    if strict_matches is not None and int(strict_matches) != matches:
        raise SystemExit(f"Diagnostic strict match count drifted from canonical matches for {slug}")
    return {
        "sample_slug": slug,
        "display_name": item.get("sample_display_name") or manifest.get("display_name") or slug,
        "municipality": item.get("municipality") or manifest.get("municipality"),
        "operational_zone": item.get("operational_zone") or manifest.get("operational_zone"),
        "recommended_split": item.get("recommended_split") or manifest.get("recommended_split"),
        "project_id": item.get("project_id"),
        "area_id": item.get("area_id"),
        "analysis_run_id": item.get("analysis_run_id"),
        "quality_check_id": item.get("quality_check_id"),
        "detection_count": integer(item, "detection_count"),
        "precision": number(item, "precision"),
        "recall": number(item, "recall"),
        "f1_score": number(item, "f1_score"),
        "mean_iou": number(item, "mean_iou"),
        "matches": matches,
        "false_positives": false_positives,
        "false_negatives": false_negatives,
        "coverage_applied": item.get("coverage_applied") is True,
        "coverage_mode": item.get("coverage_mode"),
        "coverage_tile_count": integer(item, "coverage_tile_count"),
        "reference_raw_count": integer(item, "reference_raw_count"),
        "reference_evaluated_count": integer(item, "reference_evaluated_count"),
        "reference_excluded_outside_count": integer(item, "reference_excluded_outside_count"),
        "reference_clipped_boundary_count": integer(item, "reference_clipped_boundary_count"),
        "reference_coverage_ratio": coverage_ratio(item),
        "candidate_raw_count": integer(item, "candidate_raw_count"),
        "candidate_evaluated_count": integer(item, "candidate_evaluated_count"),
        "candidate_excluded_outside_count": integer(item, "candidate_excluded_outside_count"),
        "candidate_clipped_boundary_count": integer(item, "candidate_clipped_boundary_count"),
        "diagnostic_only": item.get("diagnostic_only") is True,
        "strict_matches": int(strict_matches) if strict_matches is not None else None,
        "envelope_matches": integer(item, "envelope_matches"),
        "possible_box_to_footprint_mismatch_count": integer(item, "possible_box_to_footprint_mismatch_count"),
        "envelope_precision": number(item, "envelope_precision"),
        "envelope_recall": number(item, "envelope_recall"),
        "envelope_f1_score": number(item, "envelope_f1_score"),
    }


def evaluate_candidate(
    key: CandidateKey,
    positive_items: list[dict[str, Any]],
    background_items: list[dict[str, Any]],
    manifest_samples: dict[str, dict[str, Any]],
    args: argparse.Namespace,
) -> dict[str, Any]:
    zones = [zone_row(item, manifest_samples) for item in positive_items]
    zone_slugs = [row["sample_slug"] for row in zones]
    if len(zone_slugs) != len(set(zone_slugs)):
        raise SystemExit(f"Candidate has duplicate positive sample rows: {candidate_label(key)}")
    background_slugs = [str(item.get("sample_slug") or "").strip().lower() for item in background_items]
    if any(not slug for slug in background_slugs) or len(background_slugs) != len(set(background_slugs)):
        raise SystemExit(f"Candidate has invalid or duplicate background rows: {candidate_label(key)}")

    f1_values = [row["f1_score"] for row in zones]
    precision_values = [row["precision"] for row in zones]
    recall_values = [row["recall"] for row in zones]
    coverage_ratios = [row["reference_coverage_ratio"] for row in zones]
    background_detections = [integer(item, "detection_count") for item in background_items]
    total_matches = sum(row["matches"] for row in zones)
    total_fp = sum(row["false_positives"] for row in zones)
    total_fn = sum(row["false_negatives"] for row in zones)
    micro_precision = total_matches / (total_matches + total_fp) if total_matches + total_fp else None
    micro_recall = total_matches / (total_matches + total_fn) if total_matches + total_fn else None
    micro_f1 = None
    if micro_precision is not None and micro_recall is not None:
        micro_f1 = 2 * micro_precision * micro_recall / (micro_precision + micro_recall) if micro_precision + micro_recall else 0.0

    gates = [
        gate("positive_sample_count", len(zones) >= args.min_positive_samples, len(zones), f">={args.min_positive_samples}"),
        gate("background_sample_count", len(background_items) >= args.min_background_samples, len(background_items), f">={args.min_background_samples}"),
        gate("coverage_provenance", all(row["coverage_applied"] for row in zones), sum(row["coverage_applied"] for row in zones), "all positive runs"),
        gate("coverage_ratio", bool(coverage_ratios) and min(coverage_ratios) >= args.min_reference_coverage_ratio, min(coverage_ratios) if coverage_ratios else None, f">={args.min_reference_coverage_ratio}"),
        gate("diagnostic_separation", all(row["diagnostic_only"] for row in zones), sum(row["diagnostic_only"] for row in zones), "all positive runs"),
        gate("mean_f1", bool(f1_values) and fmean(f1_values) >= args.min_mean_f1, fmean(f1_values) if f1_values else None, f">={args.min_mean_f1}"),
        gate("minimum_zone_f1", bool(f1_values) and min(f1_values) >= args.min_zone_f1, min(f1_values) if f1_values else None, f">={args.min_zone_f1}"),
        gate("background_false_positive_pressure", bool(background_detections) and max(background_detections) <= args.max_background_detections, max(background_detections) if background_detections else None, f"<={args.max_background_detections} per sample"),
    ]
    failed_gates = [item["name"] for item in gates if not item["passed"]]
    return {
        "candidate_key": candidate_label(key),
        "model_asset_id": key[0],
        "tile_size": key[1],
        "tile_overlap": key[2],
        "threshold": key[3],
        "decision": "operationally_accepted" if not failed_gates else "review_required",
        "failed_gates": failed_gates,
        "gates": gates,
        "positive_sample_count": len(zones),
        "background_sample_count": len(background_items),
        "mean_precision": fmean(precision_values) if precision_values else None,
        "mean_recall": fmean(recall_values) if recall_values else None,
        "mean_f1": fmean(f1_values) if f1_values else None,
        "minimum_zone_f1": min(f1_values) if f1_values else None,
        "micro_precision": micro_precision,
        "micro_recall": micro_recall,
        "micro_f1": micro_f1,
        "total_matches": total_matches,
        "total_false_positives": total_fp,
        "total_false_negatives": total_fn,
        "total_background_detections": sum(background_detections),
        "max_background_detections": max(background_detections) if background_detections else None,
        "minimum_reference_coverage_ratio": min(coverage_ratios) if coverage_ratios else None,
        "total_references_raw": sum(row["reference_raw_count"] for row in zones),
        "total_references_evaluated": sum(row["reference_evaluated_count"] for row in zones),
        "total_references_excluded_outside": sum(row["reference_excluded_outside_count"] for row in zones),
        "total_references_clipped_boundary": sum(row["reference_clipped_boundary_count"] for row in zones),
        "total_candidates_raw": sum(row["candidate_raw_count"] for row in zones),
        "total_candidates_evaluated": sum(row["candidate_evaluated_count"] for row in zones),
        "total_box_to_footprint_mismatch_count": sum(row["possible_box_to_footprint_mismatch_count"] for row in zones),
        "zones": zones,
        "background_controls": [
            {
                "sample_slug": str(item.get("sample_slug") or "").strip().lower(),
                "background_category": item.get("background_category"),
                "project_id": item.get("project_id"),
                "area_id": item.get("area_id"),
                "analysis_run_id": item.get("analysis_run_id"),
                "tile_count": item.get("tile_count"),
                "detection_count": integer(item, "detection_count"),
            }
            for item in background_items
        ],
    }


def render_markdown(report: dict[str, Any]) -> str:
    lines = [
        "# Mol operational detection benchmark",
        "",
        f"- Generated: `{report['generated_at']}`",
        f"- Status: `{report['status']}`",
        f"- Recommendation: `{report['recommendation']}`",
        f"- Candidate configurations: `{report['candidate_count']}`",
        "",
        "Canonical metrics remain footprint-IoU based. Envelope results are diagnostic only.",
        "",
    ]
    for decision in report["candidate_decisions"]:
        lines.extend(
            [
                f"## {decision['candidate_key']}",
                "",
                f"- Decision: `{decision['decision']}`",
                f"- Mean/minimum F1: `{decision['mean_f1']:.4f}` / `{decision['minimum_zone_f1']:.4f}`",
                f"- Micro precision/recall/F1: `{decision['micro_precision']:.4f}` / `{decision['micro_recall']:.4f}` / `{decision['micro_f1']:.4f}`",
                f"- Canonical matches / FP / FN: `{decision['total_matches']}` / `{decision['total_false_positives']}` / `{decision['total_false_negatives']}`",
                f"- Reference coverage: `{decision['total_references_evaluated']}` / `{decision['total_references_raw']}` evaluated; `{decision['total_references_excluded_outside']}` outside; `{decision['total_references_clipped_boundary']}` boundary-clipped",
                f"- Diagnostic box-to-footprint gap: `{decision['total_box_to_footprint_mismatch_count']}`",
                f"- Background detections: `{decision['total_background_detections']}`",
                f"- Failed gates: `{', '.join(decision['failed_gates']) or 'none'}`",
                "",
                "| Zone | Precision | Recall | F1 | Coverage | Strict | Envelope | Gap |",
                "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
            ]
        )
        for zone in decision["zones"]:
            lines.append(
                f"| {zone['display_name']} | {zone['precision']:.4f} | {zone['recall']:.4f} | {zone['f1_score']:.4f} | "
                f"{zone['reference_evaluated_count']}/{zone['reference_raw_count']} | {zone['matches']} | {zone['envelope_matches']} | "
                f"{zone['possible_box_to_footprint_mismatch_count']} |"
            )
        lines.append("")
    return "\n".join(lines) + "\n"


def build_report(args: argparse.Namespace) -> dict[str, Any]:
    positive = load_json(args.positive_summary)
    background = load_json(args.background_summary)
    manifest = load_json(args.manifest_path)
    positive_items = positive.get("items") or []
    background_items = background.get("items") or []
    if not isinstance(positive_items, list) or not positive_items:
        raise SystemExit("Positive summary contains no benchmark items")
    if not isinstance(background_items, list) or not background_items:
        raise SystemExit("Background summary contains no benchmark items")
    manifest_rows = manifest.get("samples") or []
    manifest_samples = {
        str(item.get("sample_slug") or "").strip().lower(): item
        for item in manifest_rows
        if isinstance(item, dict) and item.get("sample_slug")
    }

    grouped_positive: dict[CandidateKey, list[dict[str, Any]]] = defaultdict(list)
    grouped_background: dict[CandidateKey, list[dict[str, Any]]] = defaultdict(list)
    for item in positive_items:
        if not isinstance(item, dict):
            raise SystemExit("Positive summary contains a non-object item")
        grouped_positive[candidate_key(item)].append(item)
    for item in background_items:
        if not isinstance(item, dict):
            raise SystemExit("Background summary contains a non-object item")
        grouped_background[candidate_key(item)].append(item)

    decisions = [
        evaluate_candidate(key, items, grouped_background.get(key, []), manifest_samples, args)
        for key, items in sorted(grouped_positive.items(), key=lambda entry: candidate_label(entry[0]))
    ]
    accepted = [item for item in decisions if item["decision"] == "operationally_accepted"]
    recommended = max(
        accepted,
        key=lambda item: (item["mean_f1"], item["minimum_zone_f1"], item["mean_precision"]),
        default=None,
    )
    report = {
        "schema_version": 1,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "status": "accepted" if recommended else "review_required",
        "recommendation": "retain_or_promote_candidate" if recommended else "do_not_promote_retraining_review_required",
        "base_url": args.base_url,
        "positive_summary_path": str(args.positive_summary),
        "background_summary_path": str(args.background_summary),
        "operator_sample_manifest_path": str(args.manifest_path),
        "gate_configuration": {
            "min_positive_samples": args.min_positive_samples,
            "min_background_samples": args.min_background_samples,
            "min_mean_f1": args.min_mean_f1,
            "min_zone_f1": args.min_zone_f1,
            "min_reference_coverage_ratio": args.min_reference_coverage_ratio,
            "max_background_detections": args.max_background_detections,
        },
        "candidate_count": len(decisions),
        "recommended_candidate": recommended,
        "candidate_decisions": decisions,
    }
    return report


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build a coverage-aware Mol operational detection benchmark report.")
    parser.add_argument("--positive-summary", type=Path, required=True)
    parser.add_argument("--background-summary", type=Path, required=True)
    parser.add_argument("--manifest-path", type=Path, required=True)
    parser.add_argument("--output-dir", type=Path, required=True)
    parser.add_argument("--base-url", default="")
    parser.add_argument("--min-positive-samples", type=int, default=4)
    parser.add_argument("--min-background-samples", type=int, default=1)
    parser.add_argument("--min-mean-f1", type=float, default=0.25)
    parser.add_argument("--min-zone-f1", type=float, default=0.10)
    parser.add_argument("--min-reference-coverage-ratio", type=float, default=0.90)
    parser.add_argument("--max-background-detections", type=int, default=0)
    args = parser.parse_args()
    if args.min_positive_samples < 1 or args.min_background_samples < 1:
        parser.error("sample gates must be at least 1")
    if not 0 <= args.min_mean_f1 <= 1 or not 0 <= args.min_zone_f1 <= 1:
        parser.error("F1 gates must be between 0 and 1")
    if not 0 < args.min_reference_coverage_ratio <= 1:
        parser.error("coverage ratio gate must be greater than 0 and at most 1")
    if args.max_background_detections < 0:
        parser.error("background detection gate cannot be negative")
    return args


def main() -> int:
    args = parse_args()
    report = build_report(args)
    args.output_dir.mkdir(parents=True, exist_ok=True)
    json_path = args.output_dir / "mol_operational_benchmark_report.json"
    markdown_path = args.output_dir / "mol_operational_benchmark_report.md"
    json_path.write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")
    markdown_path.write_text(render_markdown(report), encoding="utf-8")
    print("Mol operational benchmark report passed")
    print(f"Status: {report['status']}")
    print(f"Recommendation: {report['recommendation']}")
    print(f"JSON: {json_path}")
    print(f"Markdown: {markdown_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())