Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Select on calibration evidence and apply fixed, fail-closed release gates."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def load(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def find_threshold(report: dict[str, Any], threshold: float) -> dict[str, Any]:
|
|
for item in report["sweeps"]:
|
|
if abs(float(item["threshold"]) - threshold) < 1e-9:
|
|
return item
|
|
raise ValueError(f"Threshold {threshold} is absent from {report.get('summary')}")
|
|
|
|
|
|
def select_calibration_threshold(report: dict[str, Any]) -> dict[str, Any]:
|
|
eligible = [item for item in report["sweeps"] if item["pure_empty_false_positives"] == 0]
|
|
if not eligible:
|
|
eligible = report["sweeps"]
|
|
return max(
|
|
eligible,
|
|
key=lambda item: (
|
|
min(region["f1"] for region in item["regions"].values()),
|
|
item["aggregate"]["f1"],
|
|
-item["pure_empty_false_positives"],
|
|
),
|
|
)
|
|
|
|
|
|
INFERENCE_CONFIG_FIELDS = (
|
|
"model", "match_iou", "test_time_augmentation", "inference_imgsz", "inference_batch",
|
|
"max_detections_per_tile", "nms_iou", "containment_nms", "box_scale",
|
|
"box_offset_x", "box_offset_y", "additional_model", "ensemble_mode",
|
|
"ensemble_match_iou", "proposal_classifier", "proposal_classifier_threshold",
|
|
"proposal_crop_scale", "proposal_classifier_batch", "regional_models",
|
|
)
|
|
|
|
|
|
def assert_same_inference_config(calibration: dict[str, Any], report: dict[str, Any], role: str) -> None:
|
|
differences = [
|
|
field for field in INFERENCE_CONFIG_FIELDS
|
|
if calibration.get(field) != report.get(field)
|
|
]
|
|
if differences:
|
|
raise ValueError(f"{role} inference configuration differs from calibration: {differences}")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--calibration", type=Path, required=True)
|
|
parser.add_argument("--test", type=Path, required=True)
|
|
parser.add_argument("--background", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--min-aggregate-f1", type=float, default=0.55)
|
|
parser.add_argument("--min-region-f1", type=float, default=0.45)
|
|
parser.add_argument("--min-region-precision", type=float, default=0.5)
|
|
parser.add_argument("--min-region-recall", type=float, default=0.4)
|
|
parser.add_argument("--max-pure-empty-fp", type=int, default=0)
|
|
parser.add_argument(
|
|
"--selected-threshold", type=float,
|
|
help="Previously frozen calibration threshold; omit to select by worst-region F1.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
calibration = load(args.calibration)
|
|
chosen = (
|
|
find_threshold(calibration, args.selected_threshold)
|
|
if args.selected_threshold is not None
|
|
else select_calibration_threshold(calibration)
|
|
)
|
|
threshold = float(chosen["threshold"])
|
|
test_report = load(args.test)
|
|
background_report = load(args.background)
|
|
assert_same_inference_config(calibration, test_report, "test")
|
|
assert_same_inference_config(calibration, background_report, "background")
|
|
test = find_threshold(test_report, threshold)
|
|
background = find_threshold(background_report, threshold)
|
|
failures: list[str] = []
|
|
if test["aggregate"]["f1"] < args.min_aggregate_f1:
|
|
failures.append("test_aggregate_f1_below_gate")
|
|
for region, values in test["regions"].items():
|
|
if values["f1"] < args.min_region_f1:
|
|
failures.append(f"test_{region}_f1_below_gate")
|
|
if values["precision"] < args.min_region_precision:
|
|
failures.append(f"test_{region}_precision_below_gate")
|
|
if values["recall"] < args.min_region_recall:
|
|
failures.append(f"test_{region}_recall_below_gate")
|
|
if background["pure_empty_false_positives"] > args.max_pure_empty_fp:
|
|
failures.append("pure_empty_false_positive_gate_failed")
|
|
payload = {
|
|
"schema_version": 1,
|
|
"status": "training_complete" if not failures else "continue_training_loop",
|
|
"threshold_selection_source": "calibration_only",
|
|
"selected_threshold": threshold,
|
|
"gates": {
|
|
"min_aggregate_f1": args.min_aggregate_f1,
|
|
"min_region_f1": args.min_region_f1,
|
|
"min_region_precision": args.min_region_precision,
|
|
"min_region_recall": args.min_region_recall,
|
|
"max_pure_empty_false_positives": args.max_pure_empty_fp,
|
|
},
|
|
"calibration": chosen,
|
|
"test": test,
|
|
"background": background,
|
|
"failures": failures,
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
print(json.dumps(payload, indent=2))
|
|
return 0 if not failures else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|