362 lines
19 KiB
Python
362 lines
19 KiB
Python
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())
|