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
486 lines
21 KiB
Python
486 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from geoalchemy2.shape import to_shape
|
|
from shapely.geometry import mapping
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models import Detection, DetectionReview, QualityCheck, Segmentation, VectorFeature
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EvidenceItem:
|
|
role: str
|
|
candidate_feature_id: str | None
|
|
reference_feature_id: str | None
|
|
evidence: dict[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class EvidencePlan:
|
|
"""Which evidence to draw, decided before any geometry is fetched."""
|
|
|
|
items: list[EvidenceItem]
|
|
candidate_ids: set[str]
|
|
reference_ids: set[str]
|
|
total_feature_count: int
|
|
role_counts: dict[str, int]
|
|
truncated: bool
|
|
|
|
|
|
class QualityEvidenceService:
|
|
DEFAULT_EVIDENCE_LIMIT = 5_000
|
|
|
|
@staticmethod
|
|
def plan_evidence(findings: dict[str, Any], *, limit: int) -> EvidencePlan:
|
|
"""Decide what to draw before resolving a single geometry.
|
|
|
|
Building every feature and then discarding most of them meant fetching
|
|
130k geometries to draw 5k, with an ``IN`` clause holding every
|
|
identifier in the check. Planning first makes the work proportional to
|
|
what is returned.
|
|
|
|
A match contributes two features and is kept or dropped as a pair; half
|
|
a match is not reviewable evidence.
|
|
"""
|
|
|
|
matches: list[EvidenceItem] = []
|
|
false_positives: list[EvidenceItem] = []
|
|
false_negatives: list[EvidenceItem] = []
|
|
role_counts: dict[str, int] = {}
|
|
|
|
def count(role: str) -> None:
|
|
role_counts[role] = role_counts.get(role, 0) + 1
|
|
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("match_evidence")):
|
|
candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
|
|
reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
|
|
if not candidate_id and not reference_id:
|
|
continue
|
|
if candidate_id:
|
|
count("match_candidate")
|
|
if reference_id:
|
|
count("match_reference")
|
|
matches.append(EvidenceItem("match", candidate_id, reference_id, evidence))
|
|
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("false_positive_evidence")):
|
|
candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
|
|
if not candidate_id:
|
|
continue
|
|
count("false_positive")
|
|
false_positives.append(EvidenceItem("false_positive", candidate_id, None, evidence))
|
|
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("false_negative_evidence")):
|
|
reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
|
|
if not reference_id:
|
|
continue
|
|
count("false_negative")
|
|
false_negatives.append(EvidenceItem("false_negative", None, reference_id, evidence))
|
|
|
|
total_feature_count = sum(role_counts.values())
|
|
|
|
# Both error classes must be visible. Strict priority would mean a check
|
|
# with 50.000 misses and three false positives never shows one, so the
|
|
# budget is split between them in proportion to their populations, with
|
|
# at least one of each. Confirmations fill whatever is left.
|
|
selected: list[EvidenceItem] = []
|
|
truncated = False
|
|
|
|
if limit <= 0:
|
|
selected = false_negatives + false_positives + matches
|
|
else:
|
|
actionable = len(false_negatives) + len(false_positives)
|
|
if actionable <= limit:
|
|
miss_budget, false_positive_budget = len(false_negatives), len(false_positives)
|
|
else:
|
|
miss_budget = round(limit * len(false_negatives) / actionable)
|
|
miss_budget = min(len(false_negatives), max(1 if false_negatives else 0, miss_budget))
|
|
false_positive_budget = min(len(false_positives), limit - miss_budget)
|
|
if false_positives and false_positive_budget == 0:
|
|
false_positive_budget = 1
|
|
miss_budget = min(miss_budget, limit - 1)
|
|
|
|
selected = false_negatives[:miss_budget] + false_positives[:false_positive_budget]
|
|
truncated = miss_budget < len(false_negatives) or false_positive_budget < len(false_positives)
|
|
|
|
drawn = len(selected)
|
|
for item in matches:
|
|
cost = sum(1 for value in (item.candidate_feature_id, item.reference_feature_id) if value)
|
|
if drawn + cost > limit:
|
|
truncated = True
|
|
continue
|
|
selected.append(item)
|
|
drawn += cost
|
|
|
|
return EvidencePlan(
|
|
items=selected,
|
|
candidate_ids={item.candidate_feature_id for item in selected if item.candidate_feature_id},
|
|
reference_ids={item.reference_feature_id for item in selected if item.reference_feature_id},
|
|
total_feature_count=total_feature_count,
|
|
role_counts=role_counts,
|
|
truncated=truncated,
|
|
)
|
|
|
|
@staticmethod
|
|
def summarize_missing(*, candidate_ids: list[str], reference_ids: list[str]) -> list[str]:
|
|
"""One statement instead of one warning per unresolvable identifier.
|
|
|
|
A run whose detection rows were removed produced tens of thousands of
|
|
identically shaped strings, which buries every other warning.
|
|
"""
|
|
|
|
if not candidate_ids and not reference_ids:
|
|
return []
|
|
return [
|
|
f"{len(candidate_ids)} kandidaat- en {len(reference_ids)} referentieobjecten uit dit bewijs zijn niet "
|
|
"meer als geometrie terug te vinden; ze staan wel in de bewaarde telling."
|
|
]
|
|
|
|
@staticmethod
|
|
def evidence_geojson(
|
|
db: Session,
|
|
*,
|
|
project_id: UUID,
|
|
quality_check_id: UUID,
|
|
limit: int | None = None,
|
|
) -> dict[str, Any]:
|
|
quality_check = db.get(QualityCheck, quality_check_id)
|
|
if not quality_check or quality_check.project_id != project_id:
|
|
raise AppError(code="QUALITY_CHECK_NOT_FOUND", message="Quality check not found", status_code=404)
|
|
|
|
findings = quality_check.findings_json or {}
|
|
warnings: list[str] = []
|
|
missing_candidates: list[str] = []
|
|
missing_references: list[str] = []
|
|
|
|
resolved_limit = (
|
|
QualityEvidenceService.DEFAULT_EVIDENCE_LIMIT if limit is None else max(0, int(limit))
|
|
)
|
|
# Decide what to draw first, then fetch only those geometries.
|
|
plan = QualityEvidenceService.plan_evidence(findings, limit=resolved_limit)
|
|
candidate_index = QualityEvidenceService._candidate_feature_index(db, quality_check, plan.candidate_ids)
|
|
reference_index = QualityEvidenceService._reference_feature_index(db, quality_check, plan.reference_ids)
|
|
|
|
features: list[dict[str, Any]] = []
|
|
for item in plan.items:
|
|
if item.role == "match":
|
|
if item.candidate_feature_id:
|
|
row = candidate_index.get(item.candidate_feature_id)
|
|
if row is None:
|
|
missing_candidates.append(item.candidate_feature_id)
|
|
else:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="match_candidate",
|
|
quality_check=quality_check,
|
|
evidence=item.evidence,
|
|
)
|
|
)
|
|
if item.reference_feature_id:
|
|
row = reference_index.get(item.reference_feature_id)
|
|
if row is None:
|
|
missing_references.append(item.reference_feature_id)
|
|
else:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="match_reference",
|
|
quality_check=quality_check,
|
|
evidence={
|
|
"candidate_feature_id": item.candidate_feature_id,
|
|
"reference_feature_id": item.reference_feature_id,
|
|
"iou": item.evidence.get("iou"),
|
|
},
|
|
)
|
|
)
|
|
continue
|
|
|
|
if item.role == "false_positive" and item.candidate_feature_id:
|
|
row = candidate_index.get(item.candidate_feature_id)
|
|
if row is None:
|
|
missing_candidates.append(item.candidate_feature_id)
|
|
else:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="false_positive",
|
|
quality_check=quality_check,
|
|
evidence=item.evidence,
|
|
)
|
|
)
|
|
elif item.role == "false_negative" and item.reference_feature_id:
|
|
row = reference_index.get(item.reference_feature_id)
|
|
if row is None:
|
|
missing_references.append(item.reference_feature_id)
|
|
else:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="false_negative",
|
|
quality_check=quality_check,
|
|
evidence=item.evidence,
|
|
)
|
|
)
|
|
|
|
warnings.extend(
|
|
QualityEvidenceService.summarize_missing(
|
|
candidate_ids=missing_candidates,
|
|
reference_ids=missing_references,
|
|
)
|
|
)
|
|
role_counts = plan.role_counts
|
|
total_feature_count = plan.total_feature_count
|
|
truncated = plan.truncated
|
|
if truncated:
|
|
warnings.append(
|
|
f"Dit overzicht toont {len(features)} van {total_feature_count} bewijsobjecten: gemiste en "
|
|
"onterecht gevonden objecten eerst. De tellingen in de kwaliteitscontrole blijven volledig."
|
|
)
|
|
|
|
QualityEvidenceService._annotate_reviews(db, quality_check, features)
|
|
|
|
return {
|
|
"quality_check_id": str(quality_check.id),
|
|
"project_id": str(quality_check.project_id),
|
|
"candidate_dataset_id": str(quality_check.candidate_dataset_id) if quality_check.candidate_dataset_id else None,
|
|
"reference_dataset_id": str(quality_check.reference_dataset_id),
|
|
"analysis_run_id": str(quality_check.analysis_run_id) if quality_check.analysis_run_id else None,
|
|
"feature_count": len(features),
|
|
"total_feature_count": total_feature_count,
|
|
"role_counts": role_counts,
|
|
"truncated": truncated,
|
|
"limit": resolved_limit,
|
|
"warnings": warnings,
|
|
"geojson": {
|
|
"type": "FeatureCollection",
|
|
"features": features,
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def _evidence_items(value: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
return [item for item in value if isinstance(item, dict)]
|
|
|
|
@staticmethod
|
|
def _string_value(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
@staticmethod
|
|
def _evidence_identifiers(findings: dict[str, Any]) -> tuple[set[str], set[str]]:
|
|
candidate_ids: set[str] = set()
|
|
reference_ids: set[str] = set()
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("match_evidence")):
|
|
candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
|
|
reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
|
|
if candidate_id:
|
|
candidate_ids.add(candidate_id)
|
|
if reference_id:
|
|
reference_ids.add(reference_id)
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("false_positive_evidence")):
|
|
candidate_id = QualityEvidenceService._string_value(evidence.get("candidate_feature_id"))
|
|
if candidate_id:
|
|
candidate_ids.add(candidate_id)
|
|
for evidence in QualityEvidenceService._evidence_items(findings.get("false_negative_evidence")):
|
|
reference_id = QualityEvidenceService._string_value(evidence.get("reference_feature_id"))
|
|
if reference_id:
|
|
reference_ids.add(reference_id)
|
|
return candidate_ids, reference_ids
|
|
|
|
@staticmethod
|
|
def _uuid_identifiers(identifiers: set[str]) -> list[UUID]:
|
|
values: list[UUID] = []
|
|
for identifier in identifiers:
|
|
try:
|
|
values.append(UUID(identifier))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return values
|
|
|
|
@staticmethod
|
|
def _vector_feature_rows(db: Session, dataset_id: UUID, identifiers: set[str]) -> list[VectorFeature]:
|
|
if not identifiers:
|
|
return []
|
|
conditions = [VectorFeature.source_feature_id.in_(identifiers)]
|
|
uuid_identifiers = QualityEvidenceService._uuid_identifiers(identifiers)
|
|
if uuid_identifiers:
|
|
conditions.append(VectorFeature.id.in_(uuid_identifiers))
|
|
return (
|
|
db.query(VectorFeature)
|
|
.filter(VectorFeature.dataset_id == dataset_id, or_(*conditions))
|
|
.all()
|
|
)
|
|
|
|
@staticmethod
|
|
def _candidate_feature_index(
|
|
db: Session,
|
|
quality_check: QualityCheck,
|
|
identifiers: set[str],
|
|
) -> dict[str, Any]:
|
|
index: dict[str, Any] = {}
|
|
if not identifiers:
|
|
return index
|
|
uuid_identifiers = QualityEvidenceService._uuid_identifiers(identifiers)
|
|
if quality_check.candidate_dataset_id:
|
|
for row in QualityEvidenceService._vector_feature_rows(db, quality_check.candidate_dataset_id, identifiers):
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
if uuid_identifiers:
|
|
for row in db.query(Detection).filter(
|
|
Detection.dataset_id == quality_check.candidate_dataset_id,
|
|
Detection.id.in_(uuid_identifiers),
|
|
).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Segmentation).filter(
|
|
Segmentation.dataset_id == quality_check.candidate_dataset_id,
|
|
Segmentation.id.in_(uuid_identifiers),
|
|
).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
if quality_check.analysis_run_id and uuid_identifiers:
|
|
for row in db.query(Detection).filter(
|
|
Detection.analysis_run_id == quality_check.analysis_run_id,
|
|
Detection.id.in_(uuid_identifiers),
|
|
).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Segmentation).filter(
|
|
Segmentation.analysis_run_id == quality_check.analysis_run_id,
|
|
Segmentation.id.in_(uuid_identifiers),
|
|
).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
return index
|
|
|
|
@staticmethod
|
|
def _reference_feature_index(
|
|
db: Session,
|
|
quality_check: QualityCheck,
|
|
identifiers: set[str],
|
|
) -> dict[str, Any]:
|
|
index: dict[str, Any] = {}
|
|
for row in QualityEvidenceService._vector_feature_rows(db, quality_check.reference_dataset_id, identifiers):
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
return index
|
|
|
|
@staticmethod
|
|
def _annotate_reviews(
|
|
db: Session,
|
|
quality_check: QualityCheck,
|
|
features: list[dict[str, Any]],
|
|
) -> None:
|
|
if quality_check.check_type != "detections_vs_reference":
|
|
return
|
|
reviews = db.query(DetectionReview).filter(DetectionReview.quality_check_id == quality_check.id).all()
|
|
review_index = {(row.evidence_role, row.evidence_feature_id): row for row in reviews}
|
|
for feature in features:
|
|
properties = feature.get("properties")
|
|
if not isinstance(properties, dict):
|
|
continue
|
|
role = QualityEvidenceService._string_value(properties.get("qa_evidence_role"))
|
|
if role == "false_positive":
|
|
evidence_id = QualityEvidenceService._string_value(properties.get("candidate_feature_id"))
|
|
elif role == "false_negative":
|
|
evidence_id = QualityEvidenceService._string_value(properties.get("reference_feature_id"))
|
|
else:
|
|
continue
|
|
review = review_index.get((role, evidence_id or ""))
|
|
properties.update(
|
|
{
|
|
"review_decision": review.decision if review else "unreviewed",
|
|
"review_notes": review.notes if review else None,
|
|
"reviewed_by": review.reviewed_by if review else None,
|
|
"reviewed_at": review.updated_at.isoformat() if review and review.updated_at else None,
|
|
}
|
|
)
|
|
|
|
@staticmethod
|
|
def _add_index_keys(index: dict[str, Any], row: Any) -> None:
|
|
for key in QualityEvidenceService._row_identifiers(row):
|
|
index.setdefault(key, row)
|
|
|
|
@staticmethod
|
|
def _row_identifiers(row: Any) -> set[str]:
|
|
identifiers = {str(row.id)}
|
|
source_feature_id = getattr(row, "source_feature_id", None)
|
|
if source_feature_id:
|
|
identifiers.add(str(source_feature_id))
|
|
properties = getattr(row, "properties_json", None) or {}
|
|
if isinstance(properties, dict):
|
|
for property_key in ("vector_feature_id", "source_feature_id", "detection_id", "segmentation_id", "id", "name"):
|
|
value = properties.get(property_key)
|
|
if value is not None:
|
|
identifiers.add(str(value))
|
|
return identifiers
|
|
|
|
@staticmethod
|
|
def _row_to_feature(row: Any, *, role: str, quality_check: QualityCheck, evidence: dict[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
geometry = to_shape(row.geometry)
|
|
except Exception as exc:
|
|
raise AppError(
|
|
code="INVALID_QA_EVIDENCE_GEOMETRY",
|
|
message="Persisted QA evidence geometry could not be converted to GeoJSON",
|
|
details={"feature_id": str(getattr(row, "id", ""))},
|
|
status_code=500,
|
|
) from exc
|
|
|
|
properties = dict(getattr(row, "properties_json", None) or {})
|
|
properties.update(
|
|
{
|
|
"qa_evidence_role": role,
|
|
"quality_check_id": str(quality_check.id),
|
|
"project_id": str(quality_check.project_id),
|
|
"candidate_dataset_id": str(quality_check.candidate_dataset_id) if quality_check.candidate_dataset_id else None,
|
|
"reference_dataset_id": str(quality_check.reference_dataset_id),
|
|
"analysis_run_id": str(quality_check.analysis_run_id) if quality_check.analysis_run_id else None,
|
|
"feature_id": str(row.id),
|
|
"dataset_id": str(getattr(row, "dataset_id", "")) if getattr(row, "dataset_id", None) else None,
|
|
"source_feature_id": getattr(row, "source_feature_id", None),
|
|
"feature_class": getattr(row, "feature_class", None) or getattr(row, "class_name", None),
|
|
"candidate_feature_id": QualityEvidenceService._string_value(evidence.get("candidate_feature_id")),
|
|
"reference_feature_id": QualityEvidenceService._string_value(evidence.get("reference_feature_id")),
|
|
"iou": evidence.get("iou"),
|
|
}
|
|
)
|
|
properties.update(QualityEvidenceService._row_provenance(row))
|
|
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"{role}:{row.id}",
|
|
"geometry": mapping(geometry),
|
|
"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 {}
|