204 lines
9.7 KiB
Python
204 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from geoalchemy2.shape import to_shape
|
|
from shapely.geometry import mapping
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models import Detection, QualityCheck, Segmentation, VectorFeature
|
|
|
|
|
|
class QualityEvidenceService:
|
|
@staticmethod
|
|
def evidence_geojson(db: Session, *, project_id: UUID, quality_check_id: UUID) -> 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 {}
|
|
features: list[dict[str, Any]] = []
|
|
warnings: list[str] = []
|
|
candidate_index = QualityEvidenceService._candidate_feature_index(db, quality_check)
|
|
reference_index = QualityEvidenceService._reference_feature_index(db, quality_check)
|
|
|
|
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"))
|
|
iou = evidence.get("iou")
|
|
if candidate_id:
|
|
row = candidate_index.get(candidate_id)
|
|
if row is not None:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="match_candidate",
|
|
quality_check=quality_check,
|
|
evidence=evidence,
|
|
)
|
|
)
|
|
else:
|
|
warnings.append(f"Candidate evidence feature not found: {candidate_id}")
|
|
if reference_id:
|
|
row = reference_index.get(reference_id)
|
|
if row is not None:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="match_reference",
|
|
quality_check=quality_check,
|
|
evidence={"candidate_feature_id": candidate_id, "reference_feature_id": reference_id, "iou": iou},
|
|
)
|
|
)
|
|
else:
|
|
warnings.append(f"Reference evidence feature not found: {reference_id}")
|
|
|
|
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
|
|
row = candidate_index.get(candidate_id)
|
|
if row is not None:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="false_positive",
|
|
quality_check=quality_check,
|
|
evidence=evidence,
|
|
)
|
|
)
|
|
else:
|
|
warnings.append(f"False-positive evidence feature not found: {candidate_id}")
|
|
|
|
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
|
|
row = reference_index.get(reference_id)
|
|
if row is not None:
|
|
features.append(
|
|
QualityEvidenceService._row_to_feature(
|
|
row,
|
|
role="false_negative",
|
|
quality_check=quality_check,
|
|
evidence=evidence,
|
|
)
|
|
)
|
|
else:
|
|
warnings.append(f"False-negative evidence feature not found: {reference_id}")
|
|
|
|
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),
|
|
"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 _candidate_feature_index(db: Session, quality_check: QualityCheck) -> dict[str, Any]:
|
|
index: dict[str, Any] = {}
|
|
if quality_check.candidate_dataset_id:
|
|
for row in db.query(VectorFeature).filter(VectorFeature.dataset_id == quality_check.candidate_dataset_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Detection).filter(Detection.dataset_id == quality_check.candidate_dataset_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Segmentation).filter(Segmentation.dataset_id == quality_check.candidate_dataset_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
if quality_check.analysis_run_id:
|
|
for row in db.query(Detection).filter(Detection.analysis_run_id == quality_check.analysis_run_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Segmentation).filter(Segmentation.analysis_run_id == quality_check.analysis_run_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
elif quality_check.analysis_run_id:
|
|
for row in db.query(Detection).filter(Detection.analysis_run_id == quality_check.analysis_run_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
for row in db.query(Segmentation).filter(Segmentation.analysis_run_id == quality_check.analysis_run_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
return index
|
|
|
|
@staticmethod
|
|
def _reference_feature_index(db: Session, quality_check: QualityCheck) -> dict[str, Any]:
|
|
index: dict[str, Any] = {}
|
|
for row in db.query(VectorFeature).filter(VectorFeature.dataset_id == quality_check.reference_dataset_id).all():
|
|
QualityEvidenceService._add_index_keys(index, row)
|
|
return index
|
|
|
|
@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"),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"type": "Feature",
|
|
"id": f"{role}:{row.id}",
|
|
"geometry": mapping(geometry),
|
|
"properties": properties,
|
|
}
|