315 lines
13 KiB
Python
315 lines
13 KiB
Python
#!/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)
|
|
parser.add_argument(
|
|
"--default-positive-tile-size",
|
|
type=int,
|
|
default=None,
|
|
help="Tile size to use for positive portfolio runs that do not record tile_size.",
|
|
)
|
|
parser.add_argument(
|
|
"--default-positive-tile-overlap",
|
|
type=int,
|
|
default=None,
|
|
help="Tile overlap to use for positive portfolio runs that do not record tile_overlap.",
|
|
)
|
|
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],
|
|
*,
|
|
fallback_model_asset_id: str | None = None,
|
|
fallback_tile_size: int | None = None,
|
|
fallback_tile_overlap: int | None = None,
|
|
) -> CandidateKey | None:
|
|
model_asset_id = item.get("model_asset_id") or item.get("model_request") or fallback_model_asset_id
|
|
if not model_asset_id:
|
|
return None
|
|
try:
|
|
return CandidateKey(
|
|
model_asset_id=str(model_asset_id),
|
|
tile_size=int(item.get("tile_size") if item.get("tile_size") is not None else fallback_tile_size),
|
|
tile_overlap=int(
|
|
item.get("tile_overlap") if item.get("tile_overlap") is not None else fallback_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],
|
|
*,
|
|
default_tile_size: int | None = None,
|
|
default_tile_overlap: int | None = None,
|
|
) -> dict[CandidateKey, list[dict[str, Any]]]:
|
|
best_by_candidate_sample: dict[tuple[CandidateKey, str], dict[str, Any]] = {}
|
|
portfolio_model_asset_id = portfolio.get("model_asset_id")
|
|
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,
|
|
fallback_model_asset_id=str(portfolio_model_asset_id) if portfolio_model_asset_id else None,
|
|
fallback_tile_size=default_tile_size,
|
|
fallback_tile_overlap=default_tile_overlap,
|
|
)
|
|
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,
|
|
default_tile_size=args.default_positive_tile_size,
|
|
default_tile_overlap=args.default_positive_tile_overlap,
|
|
)
|
|
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()
|