feat: close measured model review evidence
This commit is contained in:
Codex
@@ -0,0 +1,208 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate explicit operator decisions for detection QA false-negatives."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
JSON_NAME = "detection_false_negative_review_validation.json"
|
||||
MARKDOWN_NAME = "detection_false_negative_review_validation.md"
|
||||
CONFIRMED_NAME = "confirmed_model_false_negatives.geojson"
|
||||
DECISIONS = (
|
||||
"confirmed_model_false_negative",
|
||||
"qa_alignment_mismatch",
|
||||
"reference_gap_or_change",
|
||||
"imagery_obscured_or_uncertain",
|
||||
"unreviewed",
|
||||
)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Validate manual false-negative review decisions without inferring labels."
|
||||
)
|
||||
parser.add_argument("--review-summary", required=True, type=Path)
|
||||
parser.add_argument("--decisions-csv", required=True, type=Path)
|
||||
parser.add_argument("--output-dir", required=True, type=Path)
|
||||
parser.add_argument(
|
||||
"--require-complete",
|
||||
action="store_true",
|
||||
help="Return a non-zero exit code while any selected record remains unreviewed.",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def load_json(path: Path) -> dict[str, Any]:
|
||||
if not path.is_file():
|
||||
raise SystemExit(f"Review summary is not readable: {path}")
|
||||
payload = json.loads(path.read_text(encoding="utf-8-sig"))
|
||||
if not isinstance(payload, dict):
|
||||
raise SystemExit(f"Review summary must be a JSON object: {path}")
|
||||
return payload
|
||||
|
||||
|
||||
def load_decisions(path: Path) -> list[dict[str, str]]:
|
||||
if not path.is_file():
|
||||
raise SystemExit(f"Review decisions CSV is not readable: {path}")
|
||||
with path.open(newline="", encoding="utf-8-sig") as handle:
|
||||
reader = csv.DictReader(handle)
|
||||
required = {"reference_feature_id", "review_decision", "review_notes"}
|
||||
missing = required - set(reader.fieldnames or [])
|
||||
if missing:
|
||||
raise SystemExit(
|
||||
"Review decisions CSV lacks required columns: "
|
||||
+ ", ".join(sorted(missing))
|
||||
)
|
||||
return [dict(row) for row in reader]
|
||||
|
||||
|
||||
def validate(
|
||||
summary: dict[str, Any], rows: list[dict[str, str]]
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
features = summary.get("selected_features") or []
|
||||
if not isinstance(features, list) or not all(
|
||||
isinstance(item, dict) for item in features
|
||||
):
|
||||
raise SystemExit("Review summary selected_features must be a list of objects")
|
||||
|
||||
expected: dict[str, dict[str, Any]] = {}
|
||||
for feature in features:
|
||||
reference_id = str(feature.get("reference_feature_id") or "").strip()
|
||||
if not reference_id or reference_id in expected:
|
||||
raise SystemExit(
|
||||
"Review summary contains a missing or duplicate reference_feature_id"
|
||||
)
|
||||
expected[reference_id] = feature
|
||||
|
||||
provided: dict[str, dict[str, str]] = {}
|
||||
for row in rows:
|
||||
reference_id = str(row.get("reference_feature_id") or "").strip()
|
||||
if not reference_id or reference_id in provided:
|
||||
raise SystemExit(
|
||||
"Review decisions contain a missing or duplicate reference_feature_id"
|
||||
)
|
||||
decision = str(row.get("review_decision") or "").strip()
|
||||
if decision not in DECISIONS:
|
||||
raise SystemExit(
|
||||
f"Invalid review decision for {reference_id}: {decision}. "
|
||||
+ "Allowed values: "
|
||||
+ ", ".join(DECISIONS)
|
||||
)
|
||||
provided[reference_id] = row
|
||||
|
||||
missing_ids = set(expected) - set(provided)
|
||||
extra_ids = set(provided) - set(expected)
|
||||
if missing_ids or extra_ids:
|
||||
details = []
|
||||
if missing_ids:
|
||||
details.append("missing: " + ", ".join(sorted(missing_ids)))
|
||||
if extra_ids:
|
||||
details.append("unexpected: " + ", ".join(sorted(extra_ids)))
|
||||
raise SystemExit(
|
||||
"Review decisions do not match the selected evidence ("
|
||||
+ "; ".join(details)
|
||||
+ ")"
|
||||
)
|
||||
|
||||
counts = {decision: 0 for decision in DECISIONS}
|
||||
confirmed_features: list[dict[str, Any]] = []
|
||||
for reference_id, feature in expected.items():
|
||||
row = provided[reference_id]
|
||||
decision = row["review_decision"].strip()
|
||||
counts[decision] += 1
|
||||
if decision != "confirmed_model_false_negative":
|
||||
continue
|
||||
source_properties = feature.get("properties") or {}
|
||||
properties = (
|
||||
dict(source_properties) if isinstance(source_properties, dict) else {}
|
||||
)
|
||||
properties.update(
|
||||
{
|
||||
"reference_feature_id": reference_id,
|
||||
"sample_slug": feature.get("sample_slug"),
|
||||
"area_m2": feature.get("area_m2"),
|
||||
"area_bucket": feature.get("area_bucket"),
|
||||
"source_tile_path": feature.get("source_tile_path"),
|
||||
"review_decision": decision,
|
||||
"review_notes": row.get("review_notes", "").strip(),
|
||||
}
|
||||
)
|
||||
confirmed_features.append(
|
||||
{
|
||||
"type": "Feature",
|
||||
"id": f"confirmed_model_false_negative:{reference_id}",
|
||||
"geometry": feature.get("geometry"),
|
||||
"properties": properties,
|
||||
}
|
||||
)
|
||||
|
||||
status = "complete" if counts["unreviewed"] == 0 else "review_required"
|
||||
report = {
|
||||
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||||
"schema_version": 1,
|
||||
"status": status,
|
||||
"review_summary_path": str(summary.get("portfolio_path") or ""),
|
||||
"selected_feature_count": len(features),
|
||||
"decision_counts": counts,
|
||||
"confirmed_model_false_negative_count": len(confirmed_features),
|
||||
"safety_rule": (
|
||||
"Only explicit confirmed_model_false_negative decisions are exported; "
|
||||
"QA false-negatives are never inferred as model training labels."
|
||||
),
|
||||
}
|
||||
return report, {"type": "FeatureCollection", "features": confirmed_features}
|
||||
|
||||
|
||||
def write_markdown(report: dict[str, Any], output_dir: Path) -> None:
|
||||
lines = [
|
||||
"# Detection false-negative review validation",
|
||||
"",
|
||||
f"- Status: `{report['status']}`",
|
||||
f"- Selected records: {report['selected_feature_count']}",
|
||||
f"- Confirmed model false-negatives: {report['confirmed_model_false_negative_count']}",
|
||||
"",
|
||||
"## Decision counts",
|
||||
"",
|
||||
]
|
||||
lines.extend(
|
||||
f"- `{decision}`: {count}"
|
||||
for decision, count in report["decision_counts"].items()
|
||||
)
|
||||
lines.extend(["", f"> {report['safety_rule']}", ""])
|
||||
(output_dir / MARKDOWN_NAME).write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
summary_path = args.review_summary.expanduser().resolve()
|
||||
decisions_path = args.decisions_csv.expanduser().resolve()
|
||||
output_dir = args.output_dir.expanduser().resolve()
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
report, confirmed = validate(load_json(summary_path), load_decisions(decisions_path))
|
||||
report["review_summary_path"] = str(summary_path)
|
||||
report["decisions_csv_path"] = str(decisions_path)
|
||||
(output_dir / JSON_NAME).write_text(
|
||||
json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
|
||||
)
|
||||
(output_dir / CONFIRMED_NAME).write_text(
|
||||
json.dumps(confirmed, indent=2, sort_keys=True), encoding="utf-8"
|
||||
)
|
||||
write_markdown(report, output_dir)
|
||||
print(f"False-negative review status: {report['status']}")
|
||||
print(f"Validation: {output_dir / JSON_NAME}")
|
||||
print(f"Confirmed evidence: {output_dir / CONFIRMED_NAME}")
|
||||
if args.require_complete and report["status"] != "complete":
|
||||
print("Review remains incomplete", flush=True)
|
||||
return 2
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user