Add persisted false-positive visual review gate
GeoIntel CI / docs-smoke (push) Canceled after 0s
GeoIntel CI / contract-smoke (push) Canceled after 0s

This commit is contained in:
Codex
2026-07-13 17:12:59 +02:00
parent 2bd38556e8
commit 1322a5d66a
13 changed files with 1379 additions and 0 deletions
+9
View File
@@ -7,6 +7,15 @@
# Changelog # Changelog
## Sprint 176 Detection false-positive visual review gate (2026-07-13)
- Enriched existing QA evidence GeoJSON with persisted detection and segmentation provenance without changing its endpoint or canonical envelope.
- Added a read-only, storage-root-confined false-positive contact-sheet renderer with deterministic AOI/area/confidence stratification and persisted reference overlays.
- Added an explicit five-state operator review contract; incomplete reviews fail the completion gate and no decision is inferred.
- Exported only manually confirmed model false-positives as possible review input, keeping reference gaps, QA alignment issues and uncertain cases separate.
- Added focused provenance, visual rendering, path-confinement, incomplete-review and confirmed-export regression coverage plus all-in-one/readiness wiring.
- Kept database migrations, model activation, inference, training and provider behavior unchanged.
## Sprint 175 Detection result scale and false-positive evidence review (2026-07-13) ## Sprint 175 Detection result scale and false-positive evidence review (2026-07-13)
- Bounded Detection Lab table rendering with client-side 25/50/100-row pagination while preserving the complete persisted detection set for MapLibre and QA/QC. - Bounded Detection Lab table rendering with client-side 25/50/100-row pagination while preserving the complete persisted detection set for MapLibre and QA/QC.
@@ -194,6 +194,7 @@ class QualityEvidenceService:
"iou": evidence.get("iou"), "iou": evidence.get("iou"),
} }
) )
properties.update(QualityEvidenceService._row_provenance(row))
return { return {
"type": "Feature", "type": "Feature",
@@ -201,3 +202,29 @@ class QualityEvidenceService:
"geometry": mapping(geometry), "geometry": mapping(geometry),
"properties": properties, "properties": properties,
} }
@staticmethod
def _row_provenance(row: Any) -> dict[str, Any]:
if isinstance(row, Detection):
return {
"detection_id": str(row.id),
"job_id": str(row.job_id) if row.job_id else None,
"confidence": row.confidence,
"model_name": row.model_name,
"model_version": row.model_version,
"source_tile_path": row.source_tile_path,
"bbox_json": row.bbox_json,
}
if isinstance(row, Segmentation):
return {
"segmentation_id": str(row.id),
"job_id": str(row.job_id) if row.job_id else None,
"confidence": row.confidence,
"model_name": row.model_name,
"model_version": row.model_version,
"source_tile_path": row.source_tile_path,
"bbox_json": row.bbox_json,
"mask_path": row.mask_path,
"area_m2": row.area_m2,
}
return {}
@@ -79,6 +79,9 @@ def test_all_in_one_dockerfile_copies_operator_scripts_for_runtime_use() -> None
"assemble_detection_calibration_evidence_portfolio.sh", "assemble_detection_calibration_evidence_portfolio.sh",
"build_fixed_threshold_evidence_portfolio_inputs.py", "build_fixed_threshold_evidence_portfolio_inputs.py",
"audit_detection_false_negative_evidence.py", "audit_detection_false_negative_evidence.py",
"audit_detection_false_positive_evidence.py",
"render_detection_false_positive_review_contact_sheets.py",
"validate_detection_false_positive_review_decisions.py",
"run_operator_hard_negative_detection_matrix.sh", "run_operator_hard_negative_detection_matrix.sh",
"run_background_corpus_split_matrix.sh", "run_background_corpus_split_matrix.sh",
"build_background_corpus_split_report.py", "build_background_corpus_split_report.py",
@@ -0,0 +1,442 @@
from __future__ import annotations
import csv
import json
import subprocess
import sys
from pathlib import Path
from uuid import uuid4
import numpy as np
import rasterio
from geoalchemy2.shape import from_shape
from PIL import Image
from rasterio.transform import from_bounds
from shapely.geometry import box, mapping
from app.models import Detection, QualityCheck
from app.services.quality_evidence_service import QualityEvidenceService
ROOT = Path(__file__).resolve().parents[2]
class FakeQuery:
def __init__(self, rows):
self.rows = list(rows)
def filter(self, *criteria):
for criterion in criteria:
left = getattr(criterion, "left", None)
right = getattr(criterion, "right", None)
operator = getattr(criterion, "operator", None)
name = getattr(left, "name", None)
value = getattr(right, "value", right)
if name and operator and operator.__name__ == "eq":
self.rows = [row for row in self.rows if getattr(row, name) == value]
return self
def all(self):
return list(self.rows)
class FakeSession:
def __init__(self, objects=None, query_rows=None) -> None:
self.objects = objects or {}
self.query_rows = query_rows or {}
def get(self, model, item_id):
return self.objects.get((model, item_id))
def query(self, model):
return FakeQuery(self.query_rows.get(model, []))
def test_detection_quality_evidence_exposes_persisted_detection_provenance() -> None:
project_id = uuid4()
dataset_id = uuid4()
reference_dataset_id = uuid4()
analysis_run_id = uuid4()
quality_check_id = uuid4()
detection = Detection(
id=uuid4(),
project_id=project_id,
dataset_id=dataset_id,
analysis_run_id=analysis_run_id,
job_id=uuid4(),
model_name="yolo-configured",
model_version="review-model",
class_name="building",
confidence=0.73,
geometry=from_shape(box(5.0, 51.0, 5.001, 51.001), srid=4326),
bbox_json={"x_min": 12.0, "y_min": 18.0, "x_max": 42.0, "y_max": 51.0},
source_tile_path="/app/storage/tiles/review/tile_0003.tif",
properties_json={"class_id": 0, "tile_index": 3},
)
quality_check = QualityCheck(
id=quality_check_id,
project_id=project_id,
analysis_run_id=analysis_run_id,
candidate_dataset_id=dataset_id,
reference_dataset_id=reference_dataset_id,
check_type="detections_vs_reference",
status="ok",
findings_json={
"false_positive_evidence": [{"candidate_feature_id": str(detection.id)}]
},
)
db = FakeSession(
objects={(QualityCheck, quality_check_id): quality_check},
query_rows={Detection: [detection]},
)
result = QualityEvidenceService.evidence_geojson(
db,
project_id=project_id,
quality_check_id=quality_check_id,
)
properties = result["geojson"]["features"][0]["properties"]
assert properties["qa_evidence_role"] == "false_positive"
assert properties["detection_id"] == str(detection.id)
assert properties["confidence"] == 0.73
assert properties["model_name"] == "yolo-configured"
assert properties["model_version"] == "review-model"
assert properties["source_tile_path"] == "/app/storage/tiles/review/tile_0003.tif"
assert properties["bbox_json"] == {
"x_min": 12.0,
"y_min": 18.0,
"x_max": 42.0,
"y_max": 51.0,
}
assert properties["tile_index"] == 3
def _write_geotiff(path: Path, *, seed: int) -> None:
rng = np.random.default_rng(seed)
data = rng.integers(35, 190, size=(3, 128, 128), dtype=np.uint8)
data[:, 32:92, 38:98] = np.array([190, 180, 165], dtype=np.uint8)[:, None, None]
path.parent.mkdir(parents=True, exist_ok=True)
with rasterio.open(
path,
"w",
driver="GTiff",
width=128,
height=128,
count=3,
dtype="uint8",
crs="EPSG:4326",
transform=from_bounds(5.0, 51.0, 5.01, 51.01, 128, 128),
) as dataset:
dataset.write(data)
def _feature(
role: str,
feature_id: str,
geometry: dict,
*,
tile_path: Path | None = None,
confidence: float | None = None,
bbox: dict | None = None,
) -> dict:
properties = {
"qa_evidence_role": role,
"feature_id": feature_id,
"candidate_feature_id": feature_id if role in {"false_positive", "match_candidate"} else None,
"reference_feature_id": feature_id if role in {"false_negative", "match_reference"} else None,
"analysis_run_id": "run-review",
"quality_check_id": "quality-review",
"feature_class": "building",
}
if tile_path is not None:
properties.update(
{
"detection_id": feature_id,
"confidence": confidence,
"model_name": "yolo-configured",
"model_version": "review-model",
"source_tile_path": str(tile_path),
"bbox_json": bbox,
"tile_index": 0,
}
)
return {
"type": "Feature",
"id": f"{role}:{feature_id}",
"properties": properties,
"geometry": geometry,
}
def _write_review_portfolio(tmp_path: Path, *, unsafe_tile: bool = False) -> tuple[Path, Path]:
storage_root = tmp_path / "storage"
samples = []
for sample_index, sample_slug in enumerate(("geel", "turnhout")):
tile_path = storage_root / sample_slug / "tile_0000.tif"
_write_geotiff(tile_path, seed=sample_index + 1)
selected_tile = (tmp_path / "outside.tif") if unsafe_tile and sample_slug == "geel" else tile_path
if unsafe_tile and sample_slug == "geel":
_write_geotiff(selected_tile, seed=99)
features = [
_feature(
"false_positive",
f"{sample_slug}-low-small",
mapping(box(5.001, 51.001, 5.0014, 51.0014)),
tile_path=selected_tile,
confidence=0.22,
bbox={"x_min": 18, "y_min": 22, "x_max": 35, "y_max": 39},
),
_feature(
"false_positive",
f"{sample_slug}-mid-medium",
mapping(box(5.003, 51.003, 5.004, 51.004)),
tile_path=tile_path,
confidence=0.48,
bbox={"x_min": 45, "y_min": 48, "x_max": 76, "y_max": 79},
),
_feature(
"false_positive",
f"{sample_slug}-high-large",
mapping(box(5.005, 51.005, 5.007, 51.007)),
tile_path=tile_path,
confidence=0.81,
bbox={"x_min": 70, "y_min": 18, "x_max": 111, "y_max": 62},
),
_feature(
"match_reference",
f"{sample_slug}-reference",
mapping(box(5.002, 51.002, 5.003, 51.003)),
),
_feature(
"false_negative",
f"{sample_slug}-missed-reference",
mapping(box(5.006, 51.002, 5.007, 51.003)),
),
]
evidence_dir = tmp_path / "portfolio" / "samples" / sample_slug / "evidence"
evidence_dir.mkdir(parents=True)
evidence_path = evidence_dir / "calibration_evidence.geojson"
evidence_path.write_text(
json.dumps({"type": "FeatureCollection", "features": features}),
encoding="utf-8",
)
samples.append(
{
"sample_slug": sample_slug,
"aoi_label": sample_slug.title(),
"role_counts": {
"false_positive": 3,
"match_reference": 1,
"false_negative": 1,
},
"evidence_geojson_path": str(evidence_path),
}
)
portfolio_path = tmp_path / "portfolio" / "calibration_evidence_portfolio.json"
portfolio_path.write_text(
json.dumps(
{
"model_asset_id": "model-review",
"model_sha256": "abc123",
"samples": samples,
}
),
encoding="utf-8",
)
return portfolio_path, storage_root
def test_false_positive_visual_review_is_stratified_and_requires_manual_decisions(
tmp_path: Path,
) -> None:
renderer = ROOT / "scripts" / "render_detection_false_positive_review_contact_sheets.py"
validator = ROOT / "scripts" / "validate_detection_false_positive_review_decisions.py"
readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8")
dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(
encoding="utf-8"
)
assert renderer.exists()
assert validator.exists()
assert "py_compile scripts/render_detection_false_positive_review_contact_sheets.py" in readiness
assert "py_compile scripts/validate_detection_false_positive_review_decisions.py" in readiness
assert "COPY scripts/render_detection_false_positive_review_contact_sheets.py" in dockerfile
assert "COPY scripts/validate_detection_false_positive_review_decisions.py" in dockerfile
portfolio_path, storage_root = _write_review_portfolio(tmp_path)
output_dir = tmp_path / "review"
result = subprocess.run(
[
sys.executable,
str(renderer),
"--portfolio",
str(portfolio_path),
"--storage-root",
str(storage_root),
"--output-dir",
str(output_dir),
"--sample-slugs",
"geel,turnhout",
"--max-features",
"4",
"--columns",
"2",
"--cards-per-sheet",
"4",
"--thumb-size",
"128",
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
report = json.loads(
(output_dir / "detection_false_positive_review_summary.json").read_text(
encoding="utf-8"
)
)
assert report["status"] == "review_required"
assert report["population_count"] == 6
assert report["selected_feature_count"] == 4
assert report["selected_sample_slugs"] == ["geel", "turnhout"]
assert report["missing_provenance_count"] == 0
assert report["missing_tile_count"] == 0
assert report["reference_overlay_feature_count"] > 0
assert set(report["selected_area_buckets"])
assert set(report["selected_confidence_bands"])
assert "review required" in result.stdout.lower()
sheet_path = output_dir / report["contact_sheets"][0]["path"]
sheet = Image.open(sheet_path).convert("RGB")
assert sheet.width >= 256
assert sheet.height >= 256
assert len(sheet.getcolors(maxcolors=1_000_000) or []) > 20
decisions_path = output_dir / "false_positive_review_decisions.csv"
with decisions_path.open(newline="", encoding="utf-8") as handle:
rows = list(csv.DictReader(handle))
assert len(rows) == 4
assert {row["review_decision"] for row in rows} == {"unreviewed"}
assert all(row["candidate_feature_id"] for row in rows)
assert all(row["source_tile_path"] for row in rows)
incomplete_dir = tmp_path / "incomplete"
incomplete = subprocess.run(
[
sys.executable,
str(validator),
"--review-summary",
str(output_dir / "detection_false_positive_review_summary.json"),
"--decisions-csv",
str(decisions_path),
"--output-dir",
str(incomplete_dir),
"--require-complete",
],
cwd=ROOT,
check=False,
capture_output=True,
text=True,
)
assert incomplete.returncode == 2
incomplete_validation = json.loads(
(incomplete_dir / "detection_false_positive_review_validation.json").read_text(
encoding="utf-8"
)
)
assert incomplete_validation["status"] == "review_required"
assert incomplete_validation["decision_counts"]["unreviewed"] == 4
incomplete_confirmed = json.loads(
(incomplete_dir / "confirmed_model_false_positives.geojson").read_text(
encoding="utf-8"
)
)
assert incomplete_confirmed["features"] == []
decisions = (
"confirmed_model_false_positive",
"reference_gap_or_change",
"qa_alignment_mismatch",
"uncertain",
)
for row, decision in zip(rows, decisions, strict=True):
row["review_decision"] = decision
row["review_notes"] = f"reviewed as {decision}"
with decisions_path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
writer.writeheader()
writer.writerows(rows)
validation_dir = tmp_path / "validated"
subprocess.run(
[
sys.executable,
str(validator),
"--review-summary",
str(output_dir / "detection_false_positive_review_summary.json"),
"--decisions-csv",
str(decisions_path),
"--output-dir",
str(validation_dir),
"--require-complete",
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
validation = json.loads(
(validation_dir / "detection_false_positive_review_validation.json").read_text(
encoding="utf-8"
)
)
assert validation["status"] == "complete"
assert validation["decision_counts"] == {
"confirmed_model_false_positive": 1,
"qa_alignment_mismatch": 1,
"reference_gap_or_change": 1,
"uncertain": 1,
"unreviewed": 0,
}
confirmed = json.loads(
(validation_dir / "confirmed_model_false_positives.geojson").read_text(
encoding="utf-8"
)
)
assert len(confirmed["features"]) == 1
assert confirmed["features"][0]["properties"]["review_decision"] == (
"confirmed_model_false_positive"
)
def test_false_positive_visual_review_rejects_tiles_outside_storage_root(
tmp_path: Path,
) -> None:
renderer = ROOT / "scripts" / "render_detection_false_positive_review_contact_sheets.py"
portfolio_path, storage_root = _write_review_portfolio(tmp_path, unsafe_tile=True)
result = subprocess.run(
[
sys.executable,
str(renderer),
"--portfolio",
str(portfolio_path),
"--storage-root",
str(storage_root),
"--output-dir",
str(tmp_path / "review"),
"--sample-slugs",
"geel",
"--max-features",
"3",
],
cwd=ROOT,
check=False,
capture_output=True,
text=True,
)
assert result.returncode != 0
assert "outside storage root" in result.stderr
+2
View File
@@ -84,6 +84,8 @@ COPY scripts/assemble_detection_calibration_evidence_portfolio.sh /app/scripts/a
COPY scripts/build_fixed_threshold_evidence_portfolio_inputs.py /app/scripts/build_fixed_threshold_evidence_portfolio_inputs.py COPY scripts/build_fixed_threshold_evidence_portfolio_inputs.py /app/scripts/build_fixed_threshold_evidence_portfolio_inputs.py
COPY scripts/audit_detection_false_negative_evidence.py /app/scripts/audit_detection_false_negative_evidence.py COPY scripts/audit_detection_false_negative_evidence.py /app/scripts/audit_detection_false_negative_evidence.py
COPY scripts/audit_detection_false_positive_evidence.py /app/scripts/audit_detection_false_positive_evidence.py COPY scripts/audit_detection_false_positive_evidence.py /app/scripts/audit_detection_false_positive_evidence.py
COPY scripts/render_detection_false_positive_review_contact_sheets.py /app/scripts/render_detection_false_positive_review_contact_sheets.py
COPY scripts/validate_detection_false_positive_review_decisions.py /app/scripts/validate_detection_false_positive_review_decisions.py
COPY scripts/run_operator_hard_negative_detection_matrix.sh /app/scripts/run_operator_hard_negative_detection_matrix.sh COPY scripts/run_operator_hard_negative_detection_matrix.sh /app/scripts/run_operator_hard_negative_detection_matrix.sh
COPY scripts/run_background_corpus_split_matrix.sh /app/scripts/run_background_corpus_split_matrix.sh COPY scripts/run_background_corpus_split_matrix.sh /app/scripts/run_background_corpus_split_matrix.sh
COPY scripts/build_background_corpus_split_report.py /app/scripts/build_background_corpus_split_report.py COPY scripts/build_background_corpus_split_report.py /app/scripts/build_background_corpus_split_report.py
+21
View File
@@ -129,6 +129,27 @@ candidates with EPSG:4326 geometry IoU greater than or equal to
raw, persisted and suppressed detection counts so calibration evidence remains raw, persisted and suppressed detection counts so calibration evidence remains
auditable. auditable.
### Persisted false-positive visual review
Detection QA labels a candidate as a false-positive only relative to the
selected persisted reference dataset and matching tolerance. That finding is
not automatically a model error: the reference can be incomplete or stale, and
alignment can be wrong. GeoIntel therefore exposes persisted detection
confidence/model/tile/bbox provenance in the existing QA evidence GeoJSON and
provides a read-only contact-sheet workflow.
The operator must explicitly select one of:
- `confirmed_model_false_positive`;
- `reference_gap_or_change`;
- `qa_alignment_mismatch`;
- `uncertain`;
- `unreviewed`.
Only records explicitly marked `confirmed_model_false_positive` are emitted by
the validator as possible hard-negative review input. The workflow does not
train a model, mutate QA persistence, fetch data or infer review decisions.
### Local model asset catalog ### Local model asset catalog
GeoIntel can list local runtime model files mounted into the backend model GeoIntel can list local runtime model files mounted into the backend model
+8
View File
@@ -1215,6 +1215,14 @@ Response:
`false_positive` or `false_negative`. Missing persisted feature ids are reported `false_positive` or `false_negative`. Missing persisted feature ids are reported
in `warnings`; no fake geometries are produced. in `warnings`; no fake geometries are produced.
Detection-backed candidate evidence also exposes provenance read from the
persisted `detections` row: `detection_id`, `job_id`, `confidence`,
`model_name`, `model_version`, `source_tile_path` and `bbox_json`. Existing
`properties_json` fields such as `tile_index` remain present. Segmentation-backed
candidate evidence exposes the equivalent persisted model/source fields plus
`segmentation_id`, `mask_path` and `area_m2`. These are additive GeoJSON
properties; the canonical envelope and endpoint path are unchanged.
## Exports ## Exports
### POST `/api/v1/exports/geojson` ### POST `/api/v1/exports/geojson`
+35
View File
@@ -7189,3 +7189,38 @@ Open:
## Next recommended pass ## Next recommended pass
- Visually classify a stratified false-positive sample from Turnhout, Herentals and Geel before deciding whether any confirmed examples belong in a new hard-negative corpus. Review the remaining 5,838 persistent false negatives in the same evidence-led pass; do not start another blind training run. - Visually classify a stratified false-positive sample from Turnhout, Herentals and Geel before deciding whether any confirmed examples belong in a new hard-negative corpus. Review the remaining 5,838 persistent false negatives in the same evidence-led pass; do not start another blind training run.
# Sprint 176 - Detection false-positive visual review gate
## Persisted provenance
- Extended the existing read-only QA evidence GeoJSON conversion so detection-backed evidence carries the persisted detection id, job id, confidence, model name/version, source tile path and pixel bbox.
- Added equivalent persisted segmentation provenance fields without changing the endpoint, canonical envelope, ORM or migration chain.
- Historical `QualityCheck` evidence can be re-exported against existing persisted `Detection` rows; no QA rerun or data rewrite is required.
## Manual visual review
- Added a storage-root-confined contact-sheet renderer for persisted detection false-positive evidence.
- The renderer validates portfolio role counts, polygon geometry, source imagery and persisted provenance, then selects deterministically across AOI, WGS84 area bucket and confidence band.
- Source imagery is rendered with the candidate pixel bbox plus persisted matched-reference and missed-reference overlays.
- Added an explicit five-state review CSV: `confirmed_model_false_positive`, `reference_gap_or_change`, `qa_alignment_mismatch`, `uncertain` and `unreviewed`.
- Added a separate validator that rejects missing, duplicate, unexpected or invalid decisions. `--require-complete` exits with code `2` while any record remains unreviewed.
- Only explicitly confirmed model false-positives are emitted to `confirmed_model_false_positives.geojson`; no QA result is automatically converted into a model label or training artifact.
## Validation
- `python -m compileall backend/app`: passed.
- `python -m pytest`: 478 passed.
- Focused provenance/render/path-confinement/incomplete-review/export tests: passed.
- `python -m ruff check` for changed Python services, scripts and tests: passed.
- Generated fixture contact sheet was visually inspected at 128 px thumbnails; candidate/reference/missed-reference overlays and header provenance remained readable.
- `npm run typecheck`: passed.
- `npm run build`: passed; app bundle `217.00 kB`, MapLibre bundle `801.82 kB` before gzip.
- `bash scripts/run_readiness_check.sh`: passed with 478 tests and the new operator-script compile gates.
- `python -m alembic heads`: one head, `202606120900`.
- `python -m alembic upgrade head --sql`: complete migration chain rendered successfully.
- Local Docker validation remains unavailable because Docker CLI is not installed on the Windows host; live all-in-one/PostGIS validation follows on Tower after deployment.
## Next recommended pass
- Re-export the seven-AOI evidence portfolio from the deployed backend, render the Turnhout/Herentals/Geel sheets and inspect the real orthophoto evidence. Keep all CSV decisions `unreviewed` until an operator makes an explicit visual classification; do not start another model training run yet.
+1
View File
@@ -139,6 +139,7 @@ This file now starts with the current implementation status. Older preparation/b
- [x] Complete rebuild/restart and browser/runtime smoke for the guarded promoted V1 building detector activation. - [x] Complete rebuild/restart and browser/runtime smoke for the guarded promoted V1 building detector activation.
- [x] Expand focused small-building training evidence after reviewing persistent false negatives, train one inactive candidate and pass it through positive, pure-empty and fixed-reference promotion evidence before guarded activation. - [x] Expand focused small-building training evidence after reviewing persistent false negatives, train one inactive candidate and pass it through positive, pure-empty and fixed-reference promotion evidence before guarded activation.
- [x] Audit the promoted model's increased false-positive load from persisted seven-AOI QA evidence, including geodetic area buckets, AOI-qualified tile hotspots and combined review GeoJSON. - [x] Audit the promoted model's increased false-positive load from persisted seven-AOI QA evidence, including geodetic area buckets, AOI-qualified tile hotspots and combined review GeoJSON.
- [x] Add persisted detection provenance, stratified visual contact sheets and an explicit manual-decision gate for false-positive review.
- [ ] Visually classify representative false-positive evidence from Turnhout, Herentals and Geel and review the remaining 5,838 persistent false negatives before any further model training; do not start another blind run. - [ ] Visually classify representative false-positive evidence from Turnhout, Herentals and Geel and review the remaining 5,838 persistent false negatives before any further model training; do not start another blind run.
## Sprint 8 status ## Sprint 8 status
+47
View File
@@ -811,6 +811,53 @@ evidence properties and geometry are preserved. Confidence statistics are only
computed when confidence is actually present in persisted evidence; missing computed when confidence is actually present in persisted evidence; missing
coverage is reported explicitly and never inferred from the run threshold. coverage is reported explicitly and never inferred from the run threshold.
Render a deterministic, stratified visual review over persisted false-positive
evidence. Static portfolios created before detection provenance was added must
first be re-exported from the current backend; existing `QualityCheck` and
`Detection` rows do not need to be rerun:
```bash
docker exec \
-e CALIBRATION_EVIDENCE_MODE=all \
-e CALIBRATION_PORTFOLIO_OUTPUT_DIR=/app/storage/operator-data/model-review/small-building-candidate/evidence-portfolio-enriched \
geointel bash /app/scripts/assemble_detection_calibration_evidence_portfolio.sh \
http://127.0.0.1 \
/app/storage/operator-data/model-review/small-building-candidate/evidence-inputs/calibration-evidence-portfolio-manifest.json
docker exec geointel /opt/geointel/venv/bin/python \
/app/scripts/render_detection_false_positive_review_contact_sheets.py \
--portfolio /app/storage/operator-data/model-review/small-building-candidate/evidence-portfolio-enriched/calibration_evidence_portfolio.json \
--storage-root /app/storage \
--output-dir /app/storage/operator-data/model-review/small-building-candidate/false-positive-visual-review \
--sample-slugs turnhout,herentals,geel \
--max-features 48 \
--columns 4 \
--cards-per-sheet 16 \
--thumb-size 256
```
The renderer validates source paths against `--storage-root`, checks persisted
confidence/bbox/tile provenance, samples across AOI, WGS84 area bucket and
confidence band, and overlays persisted matched/missed reference polygons. It
writes PNG sheets, a JSON/Markdown summary and
`false_positive_review_decisions.csv` with every row set to `unreviewed`.
After manual inspection, validate the edited CSV:
```bash
docker exec geointel /opt/geointel/venv/bin/python \
/app/scripts/validate_detection_false_positive_review_decisions.py \
--review-summary /app/storage/operator-data/model-review/small-building-candidate/false-positive-visual-review/detection_false_positive_review_summary.json \
--decisions-csv /app/storage/operator-data/model-review/small-building-candidate/false-positive-visual-review/false_positive_review_decisions.csv \
--output-dir /app/storage/operator-data/model-review/small-building-candidate/false-positive-visual-review/validated \
--require-complete
```
`--require-complete` exits with code `2` while any record is still `unreviewed`.
Only explicit `confirmed_model_false_positive` decisions are written to
`confirmed_model_false_positives.geojson`; the tool never promotes generic QA
false-positives into training labels.
Docker images install only the GIS runtime by default. To build a local/Tower Docker images install only the GIS runtime by default. To build a local/Tower
image with PyTorch/Ultralytics available for the configured-YOLO preflight and image with PyTorch/Ultralytics available for the configured-YOLO preflight and
runtime path, set: runtime path, set:
@@ -0,0 +1,586 @@
#!/usr/bin/env python3
"""Render persisted detection QA evidence for explicit operator review.
The script is read-only. It never infers whether a QA false-positive is a
model error and never mutates application persistence or source imagery.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import math
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
JSON_NAME = "detection_false_positive_review_summary.json"
MARKDOWN_NAME = "detection_false_positive_review.md"
DECISIONS_NAME = "false_positive_review_decisions.csv"
CONFIDENCE_BANDS = (
("low_lt_0_30", 0.0, 0.30),
("mid_0_30_0_60", 0.30, 0.60),
("high_gte_0_60", 0.60, math.inf),
)
DECISION_FIELDS = (
"candidate_feature_id",
"evidence_feature_id",
"sample_slug",
"confidence",
"area_m2",
"area_bucket",
"confidence_band",
"analysis_run_id",
"quality_check_id",
"source_tile_path",
"review_decision",
"review_notes",
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Render persisted detection false-positive evidence for manual review."
)
parser.add_argument("--portfolio", required=True, type=Path)
parser.add_argument("--storage-root", default="/app/storage", type=Path)
parser.add_argument("--output-dir", required=True, type=Path)
parser.add_argument(
"--sample-slugs",
default="",
help="Optional comma-separated AOI slugs. All portfolio samples are used by default.",
)
parser.add_argument("--max-features", type=int, default=48)
parser.add_argument("--columns", type=int, default=4)
parser.add_argument("--cards-per-sheet", type=int, default=16)
parser.add_argument("--thumb-size", type=int, default=256)
return parser.parse_args()
def load_json(path: Path) -> dict[str, Any]:
if not path.is_file():
raise SystemExit(f"JSON input is not readable: {path}")
payload = json.loads(path.read_text(encoding="utf-8-sig"))
if not isinstance(payload, dict):
raise SystemExit(f"JSON input must be an object: {path}")
return payload
def resolve_evidence_path(portfolio_path: Path, sample: dict[str, Any]) -> Path:
raw = str(sample.get("evidence_geojson_path") or "").strip()
configured = Path(raw).expanduser()
sample_slug = str(sample.get("sample_slug") or "").strip().lower()
candidates = [configured]
if raw and not configured.is_absolute():
candidates.append(portfolio_path.parent / configured)
candidates.append(
portfolio_path.parent
/ "samples"
/ sample_slug
/ "evidence"
/ "calibration_evidence.geojson"
)
for candidate in candidates:
if candidate.is_file():
return candidate.resolve()
raise SystemExit(f"Evidence GeoJSON is not readable for {sample_slug}: {raw}")
def confidence_band(confidence: float) -> str:
for label, minimum, maximum in CONFIDENCE_BANDS:
if minimum <= confidence < maximum:
return label
raise SystemExit(f"Detection confidence is outside [0, 1]: {confidence}")
def stable_sort_key(record: dict[str, Any]) -> str:
identity = f"{record['sample_slug']}:{record['candidate_feature_id']}"
return hashlib.sha256(identity.encode("utf-8")).hexdigest()
def stratified_selection(
records: list[dict[str, Any]], limit: int
) -> list[dict[str, Any]]:
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list)
for record in records:
grouped[
(
record["sample_slug"],
record["area_bucket"],
record["confidence_band"],
)
].append(record)
for values in grouped.values():
values.sort(key=stable_sort_key)
selected: list[dict[str, Any]] = []
keys = sorted(grouped)
depth = 0
while len(selected) < limit:
added = False
for key in keys:
values = grouped[key]
if depth < len(values):
selected.append(values[depth])
added = True
if len(selected) == limit:
break
if not added:
break
depth += 1
return selected
def require_dependencies() -> dict[str, Any]:
try:
import numpy
import rasterio
from PIL import Image, ImageDraw, ImageFont
from pyproj import Geod, Transformer
from shapely.geometry import box, shape
from shapely.ops import transform
except ImportError as exc:
raise SystemExit(
"False-positive visual review requires the backend GIS/raster extras"
) from exc
return {
"numpy": numpy,
"rasterio": rasterio,
"Image": Image,
"ImageDraw": ImageDraw,
"ImageFont": ImageFont,
"Geod": Geod,
"Transformer": Transformer,
"box": box,
"shape": shape,
"transform": transform,
}
def resolve_source_tile(raw: str, storage_root: Path) -> Path:
storage_root = storage_root.expanduser().resolve()
candidate = Path(raw).expanduser()
if not candidate.is_absolute():
candidate = storage_root / candidate
candidate = candidate.resolve()
try:
candidate.relative_to(storage_root)
except ValueError as exc:
raise SystemExit(f"Detection source tile is outside storage root: {candidate}") from exc
if not candidate.is_file():
raise SystemExit(f"Detection source tile is not readable: {candidate}")
return candidate
def normalize_raster(data: Any, numpy: Any) -> Any:
if data.shape[0] == 1:
data = numpy.repeat(data, 3, axis=0)
elif data.shape[0] >= 3:
data = data[:3]
else:
data = numpy.vstack([data, data[-1:]])[:3]
if data.dtype == numpy.uint8:
return numpy.moveaxis(data, 0, 2)
output = numpy.zeros(data.shape, dtype=numpy.uint8)
for index, band in enumerate(data):
finite = band[numpy.isfinite(band)]
if not finite.size:
continue
low, high = numpy.percentile(finite, (2, 98))
if high <= low:
high = low + 1
output[index] = numpy.clip((band - low) * 255 / (high - low), 0, 255)
return numpy.moveaxis(output, 0, 2)
def polygon_rings(geometry: Any) -> Iterable[Any]:
if geometry.geom_type == "Polygon":
yield geometry.exterior
yield from geometry.interiors
elif geometry.geom_type == "MultiPolygon":
for polygon in geometry.geoms:
yield polygon.exterior
yield from polygon.interiors
def draw_geometry(
draw: Any,
geometry: Any,
inverse_transform: Any,
scale_x: float,
scale_y: float,
color: tuple[int, int, int],
) -> None:
for ring in polygon_rings(geometry):
points = []
for x, y in ring.coords:
column, row = inverse_transform * (x, y)
points.append((column * scale_x, row * scale_y))
if len(points) >= 2:
draw.line(points, fill=color, width=2, joint="curve")
def render_card(
record: dict[str, Any],
references: list[dict[str, Any]],
thumb_size: int,
dependencies: dict[str, Any],
) -> tuple[Any, int]:
rasterio = dependencies["rasterio"]
Image = dependencies["Image"]
ImageDraw = dependencies["ImageDraw"]
ImageFont = dependencies["ImageFont"]
numpy = dependencies["numpy"]
Transformer = dependencies["Transformer"]
shape = dependencies["shape"]
transform_geometry = dependencies["transform"]
box = dependencies["box"]
header_height = 88
with rasterio.open(record["resolved_source_tile_path"]) as source:
pixels = normalize_raster(source.read(), numpy)
image = Image.fromarray(pixels, mode="RGB").resize(
(thumb_size, thumb_size), Image.Resampling.BILINEAR
)
card = Image.new(
"RGB", (thumb_size, thumb_size + header_height), color=(242, 245, 247)
)
card.paste(image, (0, header_height))
draw = ImageDraw.Draw(card)
font = ImageFont.load_default()
draw.rectangle((0, 0, thumb_size, header_height), fill=(22, 29, 38))
draw.text(
(6, 6),
f"{record['sample_slug']} conf {record['confidence']:.2f}",
fill=(255, 255, 255),
font=font,
)
draw.text(
(6, 23),
f"{record['area_bucket'].split('_', 1)[0]} | {record['area_m2']:.1f} m2",
fill=(197, 215, 231),
font=font,
)
draw.text(
(6, 40),
Path(record["source_tile_path"]).name[:24],
fill=(197, 215, 231),
font=font,
)
draw.text(
(6, 56), "red: candidate", fill=(235, 238, 241), font=font
)
draw.text(
(6, 72), "green ref | blue miss", fill=(235, 238, 241), font=font
)
scale_x = thumb_size / source.width
scale_y = thumb_size / source.height
bbox = record["bbox_json"]
draw.rectangle(
(
max(0, float(bbox["x_min"]) * scale_x),
header_height + max(0, float(bbox["y_min"]) * scale_y),
min(thumb_size - 1, float(bbox["x_max"]) * scale_x),
header_height + min(thumb_size - 1, float(bbox["y_max"]) * scale_y),
),
outline=(231, 76, 60),
width=3,
)
overlay_count = 0
if source.crs:
transformer = Transformer.from_crs("EPSG:4326", source.crs, always_xy=True)
bounds = box(*source.bounds)
overlay = Image.new("RGBA", (thumb_size, thumb_size), (0, 0, 0, 0))
overlay_draw = ImageDraw.Draw(overlay)
for reference in references:
geometry = transform_geometry(transformer.transform, shape(reference["geometry"]))
if geometry.is_empty or not geometry.intersects(bounds):
continue
color = (
(39, 174, 96)
if reference["role"] == "match_reference"
else (52, 152, 219)
)
draw_geometry(
overlay_draw,
geometry,
~source.transform,
scale_x,
scale_y,
color,
)
overlay_count += 1
card.paste(overlay, (0, header_height), overlay)
return card, overlay_count
def build_contact_sheet(cards: list[Any], columns: int, output_path: Path, Image: Any) -> None:
gap = 12
rows = math.ceil(len(cards) / columns)
width = columns * cards[0].width + (columns + 1) * gap
height = rows * cards[0].height + (rows + 1) * gap
sheet = Image.new("RGB", (width, height), color=(220, 226, 231))
for index, card in enumerate(cards):
column = index % columns
row = index // columns
sheet.paste(
card,
(gap + column * (card.width + gap), gap + row * (card.height + gap)),
)
sheet.save(output_path)
def read_population(
portfolio_path: Path,
selected_slugs: set[str],
storage_root: Path,
dependencies: dict[str, Any],
) -> tuple[list[dict[str, Any]], dict[str, list[dict[str, Any]]]]:
portfolio = load_json(portfolio_path)
samples = portfolio.get("samples") or []
if not isinstance(samples, list):
raise SystemExit("Portfolio samples must be a list")
geod = dependencies["Geod"](ellps="WGS84")
shape = dependencies["shape"]
from audit_detection_false_negative_evidence import area_bucket
population: list[dict[str, Any]] = []
references: dict[str, list[dict[str, Any]]] = defaultdict(list)
for sample in samples:
if not isinstance(sample, dict):
raise SystemExit("Portfolio contains a non-object sample")
slug = str(sample.get("sample_slug") or "").strip().lower()
if not slug or (selected_slugs and slug not in selected_slugs):
continue
evidence_path = resolve_evidence_path(portfolio_path, sample)
evidence = load_json(evidence_path)
if evidence.get("type") != "FeatureCollection":
raise SystemExit(f"Evidence must be a FeatureCollection: {evidence_path}")
false_positive_count = 0
for feature in evidence.get("features") or []:
if not isinstance(feature, dict):
raise SystemExit(f"Evidence contains a non-object feature: {evidence_path}")
properties = feature.get("properties") or {}
role = str(properties.get("qa_evidence_role") or "")
if role in {"match_reference", "false_negative"}:
reference_geometry = shape(feature.get("geometry"))
if (
reference_geometry.is_empty
or not reference_geometry.is_valid
or reference_geometry.geom_type not in {"Polygon", "MultiPolygon"}
):
raise SystemExit(
f"Reference evidence has invalid polygon geometry: {feature.get('id')}"
)
references[slug].append(
{"role": role, "geometry": feature.get("geometry")}
)
continue
if role != "false_positive":
continue
false_positive_count += 1
missing = [
key
for key in ("detection_id", "confidence", "source_tile_path", "bbox_json")
if properties.get(key) in (None, "")
]
if missing:
raise SystemExit(
f"False-positive evidence {feature.get('id')} lacks persisted detection provenance: "
+ ", ".join(missing)
+ ". Re-export the evidence portfolio with the current backend."
)
geometry = shape(feature.get("geometry"))
if geometry.is_empty or not geometry.is_valid or geometry.geom_type not in {
"Polygon",
"MultiPolygon",
}:
raise SystemExit(f"False-positive evidence has invalid polygon geometry: {feature.get('id')}")
confidence = float(properties["confidence"])
if not 0 <= confidence <= 1:
raise SystemExit(f"Detection confidence is outside [0, 1]: {confidence}")
bbox = properties["bbox_json"]
if not isinstance(bbox, dict) or any(
key not in bbox for key in ("x_min", "y_min", "x_max", "y_max")
):
raise SystemExit(f"Detection bbox_json is invalid: {feature.get('id')}")
try:
x_min, y_min, x_max, y_max = (
float(bbox[key]) for key in ("x_min", "y_min", "x_max", "y_max")
)
except (TypeError, ValueError) as exc:
raise SystemExit(
f"Detection bbox_json must contain numeric values: {feature.get('id')}"
) from exc
if not (x_min < x_max and y_min < y_max):
raise SystemExit(f"Detection bbox_json is not ordered: {feature.get('id')}")
source_tile_path = str(properties["source_tile_path"])
resolved_tile = resolve_source_tile(source_tile_path, storage_root)
area_m2 = abs(float(geod.geometry_area_perimeter(geometry)[0]))
candidate_id = str(
properties.get("candidate_feature_id")
or properties.get("detection_id")
or feature.get("id")
)
population.append(
{
"candidate_feature_id": candidate_id,
"evidence_feature_id": str(feature.get("id") or candidate_id),
"sample_slug": slug,
"confidence": confidence,
"confidence_band": confidence_band(confidence),
"area_m2": area_m2,
"area_bucket": area_bucket(area_m2),
"analysis_run_id": properties.get("analysis_run_id"),
"quality_check_id": properties.get("quality_check_id"),
"source_tile_path": source_tile_path,
"resolved_source_tile_path": str(resolved_tile),
"bbox_json": bbox,
"model_name": properties.get("model_name"),
"model_version": properties.get("model_version"),
"geometry": feature.get("geometry"),
"properties": properties,
}
)
declared = (sample.get("role_counts") or {}).get("false_positive")
if declared is not None and int(declared) != false_positive_count:
raise SystemExit(
f"Portfolio role count drift for {slug}: declared {declared}, found {false_positive_count}"
)
if not population:
raise SystemExit("No false-positive evidence was found for the selected samples")
return population, references
def write_decisions(selected: list[dict[str, Any]], output_path: Path) -> None:
with output_path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=DECISION_FIELDS)
writer.writeheader()
for record in selected:
writer.writerow(
{
key: record.get(key, "")
for key in DECISION_FIELDS
if key not in {"review_decision", "review_notes"}
}
| {"review_decision": "unreviewed", "review_notes": ""}
)
def write_markdown(report: dict[str, Any], output_dir: Path) -> None:
lines = [
"# Detection false-positive visual review",
"",
"> A QA false-positive is not automatically a model error. Review each selected detection against the imagery and reference context.",
"",
f"- Status: `{report['status']}`",
f"- Population: {report['population_count']}",
f"- Selected for manual review: {report['selected_feature_count']}",
f"- AOIs: {', '.join(report['selected_sample_slugs'])}",
"",
"## Allowed decisions",
"",
"- `confirmed_model_false_positive`: imagery confirms that the model detection is wrong.",
"- `reference_gap_or_change`: imagery supports the detection but the reference is missing or stale.",
"- `qa_alignment_mismatch`: CRS, geometry or matching tolerance caused the QA result.",
"- `uncertain`: available evidence is insufficient.",
"- `unreviewed`: no operator decision has been made.",
"",
"Only `confirmed_model_false_positive` records may be exported as possible hard-negative candidates.",
"",
"## Contact sheets",
"",
]
for sheet in report["contact_sheets"]:
lines.extend([f"![{sheet['path']}]({sheet['path']})", ""])
(output_dir / MARKDOWN_NAME).write_text("\n".join(lines) + "\n", encoding="utf-8")
def run(args: argparse.Namespace) -> dict[str, Any]:
if min(args.max_features, args.columns, args.cards_per_sheet, args.thumb_size) <= 0:
raise SystemExit("Review limits, columns and thumbnail size must be positive")
dependencies = require_dependencies()
portfolio_path = args.portfolio.expanduser().resolve()
storage_root = args.storage_root.expanduser().resolve()
requested_slugs = {
value.strip().lower() for value in args.sample_slugs.split(",") if value.strip()
}
population, references = read_population(
portfolio_path, requested_slugs, storage_root, dependencies
)
available_slugs = {record["sample_slug"] for record in population}
missing_slugs = requested_slugs - available_slugs
if missing_slugs:
raise SystemExit("Selected sample slug is absent from the portfolio: " + ", ".join(sorted(missing_slugs)))
selected = stratified_selection(population, min(args.max_features, len(population)))
output_dir = args.output_dir.expanduser().resolve()
output_dir.mkdir(parents=True, exist_ok=True)
cards = []
overlay_count = 0
for record in selected:
card, count = render_card(
record,
references.get(record["sample_slug"], []),
args.thumb_size,
dependencies,
)
cards.append(card)
overlay_count += count
contact_sheets = []
for index in range(0, len(cards), args.cards_per_sheet):
batch = cards[index : index + args.cards_per_sheet]
path = output_dir / f"false_positive_review_{index // args.cards_per_sheet + 1:03d}.png"
build_contact_sheet(batch, args.columns, path, dependencies["Image"])
contact_sheets.append({"path": path.name, "feature_count": len(batch)})
portfolio = load_json(portfolio_path)
serializable_selected = [
{key: value for key, value in record.items() if key != "resolved_source_tile_path"}
for record in selected
]
report = {
"generated_at": datetime.now(timezone.utc).isoformat(),
"schema_version": 1,
"status": "review_required",
"portfolio_path": str(portfolio_path),
"model_asset_id": portfolio.get("model_asset_id"),
"model_sha256": portfolio.get("model_sha256"),
"population_count": len(population),
"selected_feature_count": len(selected),
"selected_sample_slugs": sorted({record["sample_slug"] for record in selected}),
"selected_area_buckets": sorted({record["area_bucket"] for record in selected}),
"selected_confidence_bands": sorted({record["confidence_band"] for record in selected}),
"missing_provenance_count": 0,
"missing_tile_count": 0,
"reference_overlay_feature_count": overlay_count,
"contact_sheets": contact_sheets,
"selected_features": serializable_selected,
}
(output_dir / JSON_NAME).write_text(
json.dumps(report, indent=2, sort_keys=True), encoding="utf-8"
)
write_decisions(selected, output_dir / DECISIONS_NAME)
write_markdown(report, output_dir)
return report
def main() -> int:
args = parse_args()
report = run(args)
print("Detection false-positive review required")
print(f"Selected features: {report['selected_feature_count']}")
print(f"Summary: {args.output_dir.expanduser().resolve() / JSON_NAME}")
print(f"Decisions: {args.output_dir.expanduser().resolve() / DECISIONS_NAME}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+2
View File
@@ -50,6 +50,8 @@ ${PYTHON_BIN} -m py_compile scripts/build_background_corpus_split_report.py
${PYTHON_BIN} -m py_compile scripts/build_fixed_threshold_evidence_portfolio_inputs.py ${PYTHON_BIN} -m py_compile scripts/build_fixed_threshold_evidence_portfolio_inputs.py
${PYTHON_BIN} -m py_compile scripts/audit_detection_false_negative_evidence.py ${PYTHON_BIN} -m py_compile scripts/audit_detection_false_negative_evidence.py
${PYTHON_BIN} -m py_compile scripts/audit_detection_false_positive_evidence.py ${PYTHON_BIN} -m py_compile scripts/audit_detection_false_positive_evidence.py
${PYTHON_BIN} -m py_compile scripts/render_detection_false_positive_review_contact_sheets.py
${PYTHON_BIN} -m py_compile scripts/validate_detection_false_positive_review_decisions.py
${PYTHON_BIN} -m py_compile scripts/activate_promoted_yolo_candidate.py ${PYTHON_BIN} -m py_compile scripts/activate_promoted_yolo_candidate.py
${PYTHON_BIN} -m py_compile scripts/cleanup_demo_artifacts.py ${PYTHON_BIN} -m py_compile scripts/cleanup_demo_artifacts.py
${PYTHON_BIN} -m py_compile backend/scripts/cleanup_demo_artifacts.py ${PYTHON_BIN} -m py_compile backend/scripts/cleanup_demo_artifacts.py
@@ -0,0 +1,196 @@
#!/usr/bin/env python3
"""Validate explicit operator decisions for detection QA false-positives."""
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_positive_review_validation.json"
MARKDOWN_NAME = "detection_false_positive_review_validation.md"
CONFIRMED_NAME = "confirmed_model_false_positives.geojson"
DECISIONS = (
"confirmed_model_false_positive",
"qa_alignment_mismatch",
"reference_gap_or_change",
"uncertain",
"unreviewed",
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Validate manual false-positive 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 = {"candidate_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:
candidate_id = str(feature.get("candidate_feature_id") or "").strip()
if not candidate_id or candidate_id in expected:
raise SystemExit("Review summary contains a missing or duplicate candidate_feature_id")
expected[candidate_id] = feature
provided: dict[str, dict[str, str]] = {}
for row in rows:
candidate_id = str(row.get("candidate_feature_id") or "").strip()
if not candidate_id or candidate_id in provided:
raise SystemExit("Review decisions contain a missing or duplicate candidate_feature_id")
decision = str(row.get("review_decision") or "").strip()
if decision not in DECISIONS:
raise SystemExit(
f"Invalid review decision for {candidate_id}: {decision}. "
+ "Allowed values: "
+ ", ".join(DECISIONS)
)
provided[candidate_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 candidate_id, feature in expected.items():
row = provided[candidate_id]
decision = row["review_decision"].strip()
counts[decision] += 1
if decision != "confirmed_model_false_positive":
continue
source_properties = feature.get("properties") or {}
properties = dict(source_properties) if isinstance(source_properties, dict) else {}
properties.update(
{
"candidate_feature_id": candidate_id,
"sample_slug": feature.get("sample_slug"),
"confidence": feature.get("confidence"),
"area_m2": feature.get("area_m2"),
"area_bucket": feature.get("area_bucket"),
"confidence_band": feature.get("confidence_band"),
"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_positive:{candidate_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_positive_count": len(confirmed_features),
"safety_rule": (
"Only explicit confirmed_model_false_positive decisions are exported; "
"QA false-positives are never inferred as model errors."
),
}
return report, {"type": "FeatureCollection", "features": confirmed_features}
def write_markdown(report: dict[str, Any], output_dir: Path) -> None:
lines = [
"# Detection false-positive review validation",
"",
f"- Status: `{report['status']}`",
f"- Selected records: {report['selected_feature_count']}",
f"- Confirmed model false-positives: {report['confirmed_model_false_positive_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-positive 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())