"""Fail-closed, task-aware metrics for the GeoIntel Phase 4 benchmark. The evaluator accepts explicitly separated synthetic contract portfolios and governed product-baseline portfolios. Nothing emitted by this module is, by itself, evidence that inference provenance or production release gates passed. """ from __future__ import annotations import hashlib import json import math from collections import defaultdict from pathlib import Path from statistics import mean, median from typing import Any, Callable, Iterable PORTFOLIO_KINDS = {"synthetic_contract", "governed_product_baseline"} EVALUATOR_VERSION = "2.1.0" REPORT_SCHEMA_VERSION = 2 SUBGROUP_MIN_CASE_SUPPORT = 5 CANONICAL_JSON_SPEC = ( "UTF-8 JSON produced with sort_keys=true, separators=(',', ':'), " "ensure_ascii=false and allow_nan=false" ) TASKS = { "object_detection", "footprint_segmentation", "raster_classification", "vector_comparison", "change_detection", "terrain_interpretation", "geospatial_data_validation", } VALID_ANOMALY_SEVERITIES = { "blocker", "critical", "major", "minor", "informational", } ERROR_CODES = { "false_positive": "M-FP-CONFUSER", "false_negative": "M-FN-MISSED", "event_false_positive": "M-FP-CONFUSER", "event_false_negative": "M-FN-MISSED", "validation_false_positive": "D-VALIDATION-FP", "validation_false_negative": "D-VALIDATION-FN", "terrain_missing": "P-PARTIAL", "raster_misclassification": "M-CLASS", "raster_missing": "P-PARTIAL", "boundary_error": "M-BOUNDARY", "area_bias": "M-AREA-BIAS", "miscalibrated": "M-MISCALIBRATED", } EXPECTED_PROTECTED_POLICY = { "operating_point_selection_allowed": False, "diagnostic_curves_select_operating_point": False, "test_feedback_allowed": False, "threshold_selection_source": "pre_registered_configuration_only", } REQUIRED_METADATA_STRING_FIELDS = ( "region", "municipality", "urbanity", "object_size", "source", "sensor", "season", "date", "vegetation", "occlusion", "difficulty", "context", ) LINEAGE_SIDES = ("reference", "prediction") LINEAGE_FIELDS = ("source_id", "source_version", "derivation") NON_MEANINGFUL_TOKENS = {"", "unknown", "n/a", "na", "null", "tbd", "todo"} BELGIUM_SCOPE_BOUNDS = (1.9, 49.4, 7.5, 52.1) FAILURE_TAXONOMY = { "M-FP-CONFUSER": "unmatched object or event prediction", "M-FN-MISSED": "unmatched reference object or event", "M-CLASS": "raster class differs from the valid reference class", "M-BOUNDARY": "matched footprint has a measurable boundary deviation", "M-AREA-BIAS": "matched footprint has a measurable relative area bias", "M-MISCALIBRATED": "confidence diagnostic deviates from observed correctness", "M-OOD": "error observed in an explicitly declared out-of-distribution case", "D-VALIDATION-FP": "anomaly or severity reported without an exact reference match", "D-VALIDATION-FN": "reference anomaly or severity not exactly reported", "P-PARTIAL": "required prediction value is unavailable", } FAILURE_CONTEXTS = { "tile_edge": "case metadata explicitly marks tile-edge context", "high_confidence": ( "false positive meets the highest frozen diagnostic confidence threshold" ), "out_of_distribution": "case metadata explicitly marks OOD context", } def canonical_json_bytes(value: Any) -> bytes: """Return the exact canonical byte representation used by all hashes.""" return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def canonical_hash(value: Any) -> str: return hashlib.sha256(canonical_json_bytes(value)).hexdigest() def file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def safe_rate(numerator: int | float, denominator: int | float) -> float | None: return float(numerator) / float(denominator) if denominator else None def f1_score(precision: float | None, recall: float | None) -> float | None: if precision is None or recall is None or precision + recall == 0: return None return 2 * precision * recall / (precision + recall) def count_metrics(tp: int, fp: int, fn: int) -> dict[str, Any]: precision = safe_rate(tp, tp + fp) recall = safe_rate(tp, tp + fn) return { "true_positive": tp, "false_positive": fp, "false_negative": fn, "prediction_count": tp + fp, "reference_count": tp + fn, "precision": precision, "recall": recall, "f1": f1_score(precision, recall), "false_discovery_rate": safe_rate(fp, tp + fp), "miss_rate": safe_rate(fn, tp + fn), "precision_ci95_wilson": wilson_interval(tp, tp + fp), "recall_ci95_wilson": wilson_interval(tp, tp + fn), } def wilson_interval( successes: int, total: int, z: float = 1.959963984540054 ) -> dict[str, Any]: if total <= 0: return {"status": "undefined", "lower": None, "upper": None, "support": total} proportion = successes / total denominator = 1 + z * z / total centre = (proportion + z * z / (2 * total)) / denominator margin = ( z * math.sqrt((proportion * (1 - proportion) + z * z / (4 * total)) / total) / denominator ) return { "status": "computed", "lower": max(0.0, centre - margin), "upper": min(1.0, centre + margin), "support": total, "method": "Wilson score interval", } def _finite_float(value: Any, label: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{label} must be a finite number") converted = float(value) if not math.isfinite(converted): raise ValueError(f"{label} must be a finite number") return converted def _probability(value: Any, label: str) -> float: converted = _finite_float(value, label) if not 0.0 <= converted <= 1.0: raise ValueError(f"{label} must be between 0 and 1") return converted def _nonempty_mapping(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict) or not value: raise ValueError(f"{label} must be a non-empty object") return value def _meaningful_string( value: Any, label: str, *, allow_not_applicable: bool = False, ) -> str: if not isinstance(value, str) or value.strip().lower() in NON_MEANINGFUL_TOKENS: raise ValueError(f"{label} must be a meaningful non-empty string") normalized = value.strip() if not allow_not_applicable and normalized.lower() == "not_applicable": raise ValueError(f"{label} may not be not_applicable") return normalized def _validate_lineage(case: dict[str, Any]) -> None: sample_id = case["sample_id"] lineage = _nonempty_mapping(case.get("lineage"), f"{sample_id}: lineage") missing_sides = sorted(set(LINEAGE_SIDES) - lineage.keys()) if missing_sides: raise ValueError(f"{sample_id}: lineage missing {missing_sides}") for side in LINEAGE_SIDES: entry = _nonempty_mapping(lineage.get(side), f"{sample_id}: lineage.{side}") missing_fields = sorted(set(LINEAGE_FIELDS) - entry.keys()) if missing_fields: raise ValueError(f"{sample_id}: lineage.{side} missing {missing_fields}") for field in LINEAGE_FIELDS: _meaningful_string( entry.get(field), f"{sample_id}: lineage.{side}.{field}", ) def _validate_case_common(case: dict[str, Any]) -> None: sample_id = case.get("sample_id") if not isinstance(sample_id, str) or not sample_id.strip(): raise ValueError("Every case requires a non-empty sample_id") if case.get("task") not in TASKS: raise ValueError(f"{sample_id}: unsupported task {case.get('task')!r}") metadata = _nonempty_mapping(case.get("metadata"), f"{sample_id}: metadata") missing_metadata = sorted(set(REQUIRED_METADATA_STRING_FIELDS) - metadata.keys()) if missing_metadata: raise ValueError(f"{sample_id}: metadata missing {missing_metadata}") for field in REQUIRED_METADATA_STRING_FIELDS: _meaningful_string( metadata.get(field), f"{sample_id}: metadata.{field}", allow_not_applicable=True, ) resolution = metadata.get("resolution_m") if case["task"] == "geospatial_data_validation" and resolution is None: pass elif _finite_float(resolution, f"{sample_id}: metadata.resolution_m") <= 0: raise ValueError(f"{sample_id}: metadata.resolution_m must be positive") if not isinstance(case.get("config"), dict): raise ValueError(f"{sample_id}: config must be an explicit object") _validate_lineage(case) def _stable_id(item: dict[str, Any], label: str) -> str: identifier = item.get("id") if not isinstance(identifier, str) or not identifier.strip(): raise ValueError(f"{label} requires a non-empty string id") return identifier def _validate_unique_ids(items: list[dict[str, Any]], label: str) -> None: identifiers = [ _stable_id(item, f"{label}[{index}]") for index, item in enumerate(items) ] duplicates = sorted( {identifier for identifier in identifiers if identifiers.count(identifier) > 1} ) if duplicates: raise ValueError(f"{label} contains duplicate ids: {duplicates}") def _validate_bbox(value: Any, label: str) -> list[float]: if not isinstance(value, list) or len(value) != 4: raise ValueError(f"{label} must be [min_x, min_y, max_x, max_y]") bbox = [ _finite_float(coordinate, f"{label}[{index}]") for index, coordinate in enumerate(value) ] if bbox[2] <= bbox[0] or bbox[3] <= bbox[1]: raise ValueError(f"{label} must have positive width and height") return bbox def _validate_classes(case: dict[str, Any]) -> list[str | int]: sample_id = case["sample_id"] classes = case.get("classes") if not isinstance(classes, list) or not classes: raise ValueError(f"{sample_id}: classes must be a non-empty list") if any( isinstance(value, bool) or not isinstance(value, (str, int)) for value in classes ): raise ValueError(f"{sample_id}: classes may contain only strings or integers") if len({canonical_hash(value) for value in classes}) != len(classes): raise ValueError(f"{sample_id}: classes must be unique") if len({type(value) for value in classes}) != 1: raise ValueError(f"{sample_id}: class values must use one JSON scalar type") return classes def _validate_labeled_items( case: dict[str, Any], items: list[dict[str, Any]], label: str, ) -> None: classes = _validate_classes(case) for index, item in enumerate(items): if item.get("class") not in classes: raise ValueError( f"{case['sample_id']}: {label}[{index}].class is outside the " "declared ontology" ) def _class_compatible(left: dict[str, Any], right: dict[str, Any]) -> bool: return left.get("class") == right.get("class") def bbox_iou(left: list[float], right: list[float]) -> float: x1, y1 = max(left[0], right[0]), max(left[1], right[1]) x2, y2 = min(left[2], right[2]), min(left[3], right[3]) intersection = max(0.0, x2 - x1) * max(0.0, y2 - y1) left_area = max(0.0, left[2] - left[0]) * max(0.0, left[3] - left[1]) right_area = max(0.0, right[2] - right[0]) * max(0.0, right[3] - right[1]) union = left_area + right_area - intersection return intersection / union if union > 0 else 0.0 def greedy_match( predictions: list[dict[str, Any]], references: list[dict[str, Any]], overlap: Callable[[dict[str, Any], dict[str, Any]], float], threshold: float, *, class_aware: bool = False, ) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: unmatched = set(range(len(references))) matches: list[dict[str, Any]] = [] false_positives: list[dict[str, Any]] = [] ordered = sorted( predictions, key=lambda item: ( -float(item.get("confidence", 1.0)), _stable_id(item, "prediction"), ), ) for prediction in ordered: candidates: list[tuple[float, str, int]] = [] for index in sorted(unmatched): reference = references[index] if class_aware and not _class_compatible(prediction, reference): continue candidates.append( ( overlap(prediction, reference), _stable_id(reference, "reference"), index, ) ) if candidates: score, _reference_id, index = min( candidates, key=lambda item: (-item[0], item[1]) ) else: score, index = 0.0, -1 if index >= 0 and score >= threshold: unmatched.remove(index) match = { "prediction_id": prediction["id"], "reference_id": references[index]["id"], "overlap": score, "confidence": prediction.get("confidence"), } if class_aware: match["class"] = prediction["class"] matches.append(match) else: false_positives.append(prediction) false_negatives = [references[index] for index in sorted(unmatched)] return matches, false_positives, false_negatives def interpolated_ap( points: list[tuple[float, str, int]], reference_count: int, ) -> float | None: if reference_count <= 0: return None ordered = sorted(points, key=lambda item: (-item[0], item[1])) true_positive = 0 false_positive = 0 curve: list[tuple[float, float]] = [] for _confidence, _stable_prediction_id, correct in ordered: true_positive += correct false_positive += 1 - correct curve.append( ( true_positive / reference_count, true_positive / (true_positive + false_positive), ) ) values = [] for step in range(101): recall_level = step / 100 values.append( max( (precision for recall, precision in curve if recall >= recall_level), default=0.0, ) ) return sum(values) / len(values) def detection_ap( predictions: list[dict[str, Any]], references: list[dict[str, Any]], iou_threshold: float, ) -> float | None: unmatched = set(range(len(references))) points: list[tuple[float, str, int]] = [] ordered = sorted( predictions, key=lambda item: ( -float(item["confidence"]), _stable_id(item, "prediction"), ), ) for prediction in ordered: candidates: list[tuple[float, str, int]] = [] for index in sorted(unmatched): reference = references[index] if not _class_compatible(prediction, reference): continue if prediction.get("_sample_id") != reference.get("_sample_id"): continue candidates.append( ( bbox_iou(prediction["bbox"], reference["bbox"]), _stable_id(reference, "reference"), index, ) ) if candidates: score, _reference_id, index = min( candidates, key=lambda item: (-item[0], item[1]) ) else: score, index = 0.0, -1 correct = int(index >= 0 and score >= iou_threshold) if correct: unmatched.remove(index) points.append((float(prediction["confidence"]), prediction["id"], correct)) return interpolated_ap(points, len(references)) def calibration_metrics( predictions: list[dict[str, Any]], matches: list[dict[str, Any]], bins: int = 5 ) -> dict[str, Any]: matched_ids = {item["prediction_id"] for item in matches} scored = [ (float(item["confidence"]), 1 if item.get("id") in matched_ids else 0) for item in predictions ] if not scored: return {"ece": None, "brier": None, "bins": [], "status": "undefined"} blocks = [] ece = 0.0 for index in range(bins): lower = index / bins upper = (index + 1) / bins selected = [ (confidence, correct) for confidence, correct in scored if lower <= confidence <= upper and (index == bins - 1 or confidence < upper) ] if not selected: blocks.append( { "lower": lower, "upper": upper, "count": 0, "mean_confidence": None, "accuracy": None, } ) continue avg_confidence = mean(item[0] for item in selected) accuracy = mean(item[1] for item in selected) ece += len(selected) / len(scored) * abs(accuracy - avg_confidence) blocks.append( { "lower": lower, "upper": upper, "count": len(selected), "mean_confidence": avg_confidence, "accuracy": accuracy, } ) return { "status": "computed", "ece": ece, "brier": mean((confidence - correct) ** 2 for confidence, correct in scored), "bins": blocks, "binning": "five fixed equal-width bins", } def coverage_risk( predictions: list[dict[str, Any]], references: list[dict[str, Any]], match_iou: float, operating_threshold: float, ) -> list[dict[str, Any]]: """Report retention, reference coverage and risk including every FN.""" thresholds = sorted({0.0, 0.5, 0.7, 0.9, operating_threshold}) rows = [] for threshold in thresholds: retained = [ item for item in predictions if float(item["confidence"]) >= threshold ] matches, false_positives, false_negatives = greedy_match( retained, references, lambda prediction, reference: bbox_iou( prediction["bbox"], reference["bbox"] ), match_iou, class_aware=True, ) tp = len(matches) fp = len(false_positives) fn = len(false_negatives) rows.append( { "threshold": threshold, "retained_prediction_count": len(retained), "total_prediction_count": len(predictions), "retained_prediction_coverage": safe_rate( len(retained), len(predictions) ), "matched_reference_count": tp, "reference_count": len(references), "reference_coverage": safe_rate(tp, len(references)), "false_positive_count": fp, "false_negative_count": fn, "risk": safe_rate(fp + fn, tp + fp + fn), "risk_definition": ( "(false_positive + false_negative) / " "(true_positive + false_positive + false_negative)" ), } ) return rows def _mean_or_none(values: Iterable[float | None]) -> float | None: retained = [value for value in values if value is not None] return mean(retained) if retained else None def _validate_detection_case( case: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: references = case.get("references") predictions = case.get("predictions") if not isinstance(references, list) or not all( isinstance(item, dict) for item in references ): raise ValueError(f"{case['sample_id']}: references must be an object list") if not isinstance(predictions, list) or not all( isinstance(item, dict) for item in predictions ): raise ValueError(f"{case['sample_id']}: predictions must be an object list") _validate_unique_ids(references, f"{case['sample_id']}.references") _validate_unique_ids(predictions, f"{case['sample_id']}.predictions") _validate_labeled_items(case, references, "references") _validate_labeled_items(case, predictions, "predictions") for index, item in enumerate(references): _validate_bbox( item.get("bbox"), f"{case['sample_id']}.references[{index}].bbox", ) for index, item in enumerate(predictions): _validate_bbox( item.get("bbox"), f"{case['sample_id']}.predictions[{index}].bbox", ) _probability( item.get("confidence"), f"{case['sample_id']}.predictions[{index}].confidence", ) return references, predictions def evaluate_object_detection(case: dict[str, Any]) -> dict[str, Any]: references, predictions = _validate_detection_case(case) threshold = _probability( case["config"].get("confidence_threshold"), f"{case['sample_id']}.config.confidence_threshold", ) match_iou = _probability( case["config"].get("match_iou"), f"{case['sample_id']}.config.match_iou", ) diagnostic_thresholds = case["config"].get("fixed_diagnostic_risk_thresholds") if diagnostic_thresholds is not None: if not isinstance(diagnostic_thresholds, list) or not diagnostic_thresholds: raise ValueError( f"{case['sample_id']}: fixed diagnostic risk thresholds must be a " "non-empty list" ) for index, value in enumerate(diagnostic_thresholds): _probability( value, f"{case['sample_id']}.config.fixed_diagnostic_risk_thresholds[{index}]", ) retained = [item for item in predictions if float(item["confidence"]) >= threshold] matches, false_positives, false_negatives = greedy_match( retained, references, lambda prediction, reference: bbox_iou(prediction["bbox"], reference["bbox"]), match_iou, class_aware=True, ) per_class = {} for class_value in case["classes"]: class_predictions = [item for item in retained if item["class"] == class_value] all_class_predictions = [ item for item in predictions if item["class"] == class_value ] class_references = [item for item in references if item["class"] == class_value] class_matches, class_fp, class_fn = greedy_match( class_predictions, class_references, lambda prediction, reference: bbox_iou( prediction["bbox"], reference["bbox"] ), match_iou, class_aware=True, ) class_metrics = count_metrics(len(class_matches), len(class_fp), len(class_fn)) class_metrics["ap50"] = detection_ap( all_class_predictions, class_references, 0.5 ) class_metrics["ap50_95"] = _mean_or_none( detection_ap(all_class_predictions, class_references, 0.5 + step * 0.05) for step in range(10) ) per_class[str(class_value)] = class_metrics metrics = count_metrics(len(matches), len(false_positives), len(false_negatives)) metrics.update( { "operating_confidence": threshold, "match_iou": match_iou, "matched_iou": distribution([item["overlap"] for item in matches]), "ap50": detection_ap(predictions, references, 0.5), "ap50_95": _mean_or_none( detection_ap(predictions, references, 0.5 + step * 0.05) for step in range(10) ), "per_class": per_class, "calibration": calibration_metrics(retained, matches), "coverage_risk": coverage_risk( predictions, references, match_iou, threshold ), "map50": _mean_or_none(item["ap50"] for item in per_class.values()), "map50_95": _mean_or_none(item["ap50_95"] for item in per_class.values()), } ) evaluation = result( case, metrics, matches, false_positives, false_negatives, post_filter_predictions=retained, filter_description={ "applied": True, "field": "confidence", "operator": ">=", "threshold": threshold, }, ) calibration = metrics["calibration"] if calibration["status"] == "computed" and calibration["ece"] > 1e-12: evaluation["failures"].append( failure_entry( case, "miscalibrated", { "ece": calibration["ece"], "brier": calibration["brier"], "prediction_support": len(retained), "claim_boundary": ( "diagnostic deviation only; not a representative " "population-calibration claim" ), }, ) ) return evaluation def distribution(values: list[float]) -> dict[str, Any]: if not values: return {"count": 0, "mean": None, "median": None, "min": None, "max": None} if any(not math.isfinite(value) for value in values): raise ValueError("Distribution values must be finite") ordered = sorted(values) return { "count": len(values), "mean": mean(values), "median": median(values), "min": ordered[0], "max": ordered[-1], } def _bounds_overlap( left: tuple[float, float, float, float], right: tuple[float, float, float, float], ) -> bool: return not ( left[2] < right[0] or left[0] > right[2] or left[3] < right[1] or left[1] > right[3] ) def _validate_metric_spatial_context(case: dict[str, Any]) -> dict[str, Any]: context = _nonempty_mapping( case.get("spatial_context"), f"{case['sample_id']}: spatial_context", ) if context.get("metric") is not True: raise ValueError(f"{case['sample_id']}: spatial_context.metric must be true") if context.get("coordinate_units") != "m": raise ValueError( f"{case['sample_id']}: spatial_context.coordinate_units must be 'm'" ) crs_value = context.get("crs") if not isinstance(crs_value, str) or not crs_value.strip(): raise ValueError(f"{case['sample_id']}: a non-empty CRS is required") try: from pyproj import CRS crs = CRS.from_user_input(crs_value) except Exception as exc: # noqa: BLE001 - invalid CRS must fail closed raise ValueError(f"{case['sample_id']}: invalid CRS {crs_value!r}") from exc if not crs.is_projected: raise ValueError( f"{case['sample_id']}: metric geometry requires a projected CRS" ) if not crs.axis_info or any( axis.unit_conversion_factor is None or not math.isclose( axis.unit_conversion_factor, 1.0, rel_tol=0.0, abs_tol=1e-12, ) for axis in crs.axis_info[:2] ): raise ValueError(f"{case['sample_id']}: CRS axes must use metres") if crs.to_epsg() in {3395, 3857}: raise ValueError( f"{case['sample_id']}: Web/World Mercator is unsuitable for " "benchmark area, boundary and distance metrics" ) area = crs.area_of_use if area is None: raise ValueError( f"{case['sample_id']}: CRS area of use is unavailable; Belgian " "metric suitability cannot be proven" ) crs_bounds = (area.west, area.south, area.east, area.north) if not _bounds_overlap(crs_bounds, BELGIUM_SCOPE_BOUNDS): raise ValueError( f"{case['sample_id']}: CRS area of use does not overlap the " "declared Belgium and Belgian North Sea product scope" ) return context def _validated_ring(coordinates: Any, label: str) -> list[list[float]]: if not isinstance(coordinates, list) or len(coordinates) < 4: raise ValueError( f"{label} must contain a closed ring with at least four points" ) normalized: list[list[float]] = [] for index, point in enumerate(coordinates): if not isinstance(point, list) or len(point) != 2: raise ValueError(f"{label}[{index}] must be an [x, y] pair") normalized.append( [ _finite_float(point[0], f"{label}[{index}][0]"), _finite_float(point[1], f"{label}[{index}][1]"), ] ) if normalized[0] != normalized[-1]: raise ValueError(f"{label} must be explicitly closed") return normalized def _validated_polygon(item: dict[str, Any], label: str): if "geometry" in item and "polygon" in item: raise ValueError(f"{label} may not declare both geometry and polygon") if "geometry" in item: payload = item["geometry"] if not isinstance(payload, dict): raise ValueError(f"{label}.geometry must be a GeoJSON object") geometry_type = payload.get("type") coordinates = payload.get("coordinates") if geometry_type == "Polygon": if not isinstance(coordinates, list) or not coordinates: raise ValueError(f"{label}.geometry Polygon requires rings") normalized_coordinates: Any = [ _validated_ring(ring, f"{label}.geometry.coordinates[{index}]") for index, ring in enumerate(coordinates) ] elif geometry_type == "MultiPolygon": if not isinstance(coordinates, list) or not coordinates: raise ValueError( f"{label}.geometry MultiPolygon requires polygon members" ) normalized_coordinates = [] for polygon_index, polygon_coordinates in enumerate(coordinates): if not isinstance(polygon_coordinates, list) or not polygon_coordinates: raise ValueError( f"{label}.geometry.coordinates[{polygon_index}] requires rings" ) normalized_coordinates.append( [ _validated_ring( ring, f"{label}.geometry.coordinates[{polygon_index}]" f"[{ring_index}]", ) for ring_index, ring in enumerate(polygon_coordinates) ] ) else: raise ValueError(f"{label}.geometry.type must be Polygon or MultiPolygon") normalized_payload = { "type": geometry_type, "coordinates": normalized_coordinates, } else: normalized_payload = { "type": "Polygon", "coordinates": [_validated_ring(item.get("polygon"), f"{label}.polygon")], } from shapely.geometry import shape geometry = shape(normalized_payload) if geometry.is_empty or geometry.area <= 0 or not geometry.is_valid: raise ValueError(f"{label}.geometry must be non-empty, positive-area and valid") return geometry def _validated_polygon_items( case: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: _validate_metric_spatial_context(case) references = case.get("references") predictions = case.get("predictions") if not isinstance(references, list) or not all( isinstance(item, dict) for item in references ): raise ValueError(f"{case['sample_id']}: references must be an object list") if not isinstance(predictions, list) or not all( isinstance(item, dict) for item in predictions ): raise ValueError(f"{case['sample_id']}: predictions must be an object list") _validate_unique_ids(references, f"{case['sample_id']}.references") _validate_unique_ids(predictions, f"{case['sample_id']}.predictions") _validate_labeled_items(case, references, "references") _validate_labeled_items(case, predictions, "predictions") validated_references = [ dict( item, _shape=_validated_polygon(item, f"{case['sample_id']}.references[{index}]"), ) for index, item in enumerate(references) ] validated_predictions = [ dict( item, _shape=_validated_polygon( item, f"{case['sample_id']}.predictions[{index}]" ), ) for index, item in enumerate(predictions) ] return validated_references, validated_predictions def _polygon_overlap(prediction: dict[str, Any], reference: dict[str, Any]) -> float: intersection = prediction["_shape"].intersection(reference["_shape"]).area union = prediction["_shape"].union(reference["_shape"]).area return intersection / union if union > 0 else 0.0 def boundary_f1(prediction, reference, tolerance: float) -> float | None: predicted_boundary = prediction.boundary reference_boundary = reference.boundary predicted_length = predicted_boundary.length reference_length = reference_boundary.length if predicted_length <= 0 or reference_length <= 0: return None precision = ( predicted_boundary.intersection(reference_boundary.buffer(tolerance)).length / predicted_length ) recall = ( reference_boundary.intersection(predicted_boundary.buffer(tolerance)).length / reference_length ) return f1_score(precision, recall) def evaluate_footprint_segmentation(case: dict[str, Any]) -> dict[str, Any]: references, predictions = _validated_polygon_items(case) match_iou = _probability( case["config"].get("match_iou"), f"{case['sample_id']}.config.match_iou", ) tolerance = _finite_float( case["config"].get("boundary_tolerance_m"), f"{case['sample_id']}.config.boundary_tolerance_m", ) if tolerance <= 0: raise ValueError(f"{case['sample_id']}: boundary tolerance must be positive") for index, item in enumerate(predictions): if "confidence" in item: _probability( item["confidence"], f"{case['sample_id']}.predictions[{index}].confidence", ) matches, false_positives, false_negatives = greedy_match( predictions, references, _polygon_overlap, match_iou, class_aware=True, ) by_prediction = {item["id"]: item for item in predictions} by_reference = {item["id"]: item for item in references} dice = [] boundary = [] centroids = [] area_errors = [] for match in matches: pred = by_prediction[match["prediction_id"]]["_shape"] ref = by_reference[match["reference_id"]]["_shape"] intersection = pred.intersection(ref).area dice_value = 2 * intersection / (pred.area + ref.area) boundary_value = boundary_f1(pred, ref, tolerance) centroid_value = pred.centroid.distance(ref.centroid) area_error = (pred.area - ref.area) / ref.area dice.append(dice_value) boundary.append(boundary_value) centroids.append(centroid_value) area_errors.append(area_error) match.update( dice=dice_value, boundary_f1=boundary_value, centroid_distance_m=centroid_value, relative_area_error=area_error, ) metrics = count_metrics(len(matches), len(false_positives), len(false_negatives)) metrics.update( { "mean_iou": _mean_or_none([item["overlap"] for item in matches]), "mean_dice": _mean_or_none(dice), "mean_boundary_f1": _mean_or_none(boundary), "centroid_distance_m": distribution(centroids), "relative_area_error": distribution(area_errors), "topologically_valid_predictions": len(predictions), "topologically_invalid_predictions": 0, "spatial_context_validated": True, } ) evaluation = result( case, metrics, matches, false_positives, false_negatives, ) for match in matches: if match["boundary_f1"] is not None and match["boundary_f1"] < 1 - 1e-12: evaluation["failures"].append( failure_entry( case, "boundary_error", { "prediction_id": match["prediction_id"], "reference_id": match["reference_id"], "boundary_f1": match["boundary_f1"], "boundary_tolerance_m": tolerance, }, ) ) if abs(match["relative_area_error"]) > 1e-12: evaluation["failures"].append( failure_entry( case, "area_bias", { "prediction_id": match["prediction_id"], "reference_id": match["reference_id"], "relative_area_error": match["relative_area_error"], }, ) ) return evaluation def _validate_rectangular_grid( value: Any, label: str, ) -> tuple[list[list[Any]], tuple[int, int]]: if ( not isinstance(value, list) or not value or not all(isinstance(row, list) for row in value) ): raise ValueError(f"{label} must be a non-empty two-dimensional array") width = len(value[0]) if width <= 0 or any(len(row) != width for row in value): raise ValueError(f"{label} must be exactly rectangular") return value, (len(value), width) def _validate_raster_side( case: dict[str, Any], side_name: str, expected_shape: tuple[int, int], ) -> dict[str, Any]: context = _nonempty_mapping( case["raster_context"].get(side_name), f"{case['sample_id']}.raster_context.{side_name}", ) required = {"crs", "transform", "shape", "nodata", "mask"} missing = sorted(required - context.keys()) if missing: raise ValueError( f"{case['sample_id']}.raster_context.{side_name} missing {missing}" ) crs_value = context["crs"] if not isinstance(crs_value, str) or not crs_value.strip(): raise ValueError(f"{case['sample_id']}: raster CRS must be non-empty") try: from pyproj import CRS normalized_crs = CRS.from_user_input(crs_value) except Exception as exc: # noqa: BLE001 - invalid CRS must fail closed raise ValueError( f"{case['sample_id']}: invalid raster CRS {crs_value!r}" ) from exc transform = context["transform"] if not isinstance(transform, list) or len(transform) != 6: raise ValueError( f"{case['sample_id']}: raster transform must contain six numbers" ) normalized_transform = tuple( _finite_float( value, f"{case['sample_id']}.{side_name}.transform[{index}]", ) for index, value in enumerate(transform) ) a, b, _x_offset, d, e, _y_offset = normalized_transform determinant = a * e - b * d linear_scale = max(abs(a), abs(b), abs(d), abs(e), 1.0) if abs(determinant) <= 1e-12 * linear_scale * linear_scale: raise ValueError( f"{case['sample_id']}: raster affine transform is singular or " "numerically invalid" ) shape = context["shape"] if ( not isinstance(shape, list) or len(shape) != 2 or any( isinstance(value, bool) or not isinstance(value, int) or value <= 0 for value in shape ) ): raise ValueError( f"{case['sample_id']}: raster shape must be [positive rows, " "positive columns]" ) if tuple(shape) != expected_shape: raise ValueError( f"{case['sample_id']}: declared {side_name} shape differs from the grid" ) mask, mask_shape = _validate_rectangular_grid( context["mask"], f"{case['sample_id']}.{side_name}.mask", ) if mask_shape != expected_shape or any( not isinstance(value, bool) for row in mask for value in row ): raise ValueError( f"{case['sample_id']}: {side_name} mask must be a boolean grid " "with the exact raster shape" ) nodata = context["nodata"] if nodata is not None: _finite_float(nodata, f"{case['sample_id']}.{side_name}.nodata") return { "crs": normalized_crs, "transform": normalized_transform, "shape": expected_shape, "nodata": nodata, "mask": mask, } def _is_nodata(value: Any, nodata: Any) -> bool: return nodata is not None and value == nodata def _validate_raster_context( case: dict[str, Any], reference_shape: tuple[int, int], prediction_shape: tuple[int, int], ) -> tuple[dict[str, Any], dict[str, Any]]: if reference_shape != prediction_shape: raise ValueError(f"{case['sample_id']}: raster shapes differ") _nonempty_mapping( case.get("raster_context"), f"{case['sample_id']}: raster_context", ) reference = _validate_raster_side(case, "reference", reference_shape) prediction = _validate_raster_side(case, "prediction", prediction_shape) if reference["crs"] != prediction["crs"]: raise ValueError(f"{case['sample_id']}: raster CRS alignment differs") if reference["transform"] != prediction["transform"]: raise ValueError(f"{case['sample_id']}: raster affine alignment differs") if reference["shape"] != prediction["shape"]: raise ValueError(f"{case['sample_id']}: raster shape alignment differs") return reference, prediction def evaluate_raster_classification(case: dict[str, Any]) -> dict[str, Any]: classes = _validate_classes(case) reference_grid, reference_shape = _validate_rectangular_grid( case.get("references"), f"{case['sample_id']}.references", ) prediction_grid, prediction_shape = _validate_rectangular_grid( case.get("predictions"), f"{case['sample_id']}.predictions", ) reference_context, prediction_context = _validate_raster_context( case, reference_shape, prediction_shape, ) labels = [str(value) for value in classes] missing_label = "__prediction_nodata__" confusion = { reference: {prediction: 0 for prediction in [*labels, missing_label]} for reference in labels } failures = [] evaluated_reference_count = 0 prediction_present_count = 0 ignored_reference_count = 0 prediction_outside_reference_count = 0 correct_count = 0 for row_index in range(reference_shape[0]): for column_index in range(reference_shape[1]): reference_value = reference_grid[row_index][column_index] prediction_value = prediction_grid[row_index][column_index] reference_valid = reference_context["mask"][row_index][column_index] prediction_valid = prediction_context["mask"][row_index][column_index] if reference_valid and _is_nodata( reference_value, reference_context["nodata"] ): raise ValueError( f"{case['sample_id']}: reference mask marks nodata as " f"valid at [{row_index}, {column_index}]" ) if prediction_valid and _is_nodata( prediction_value, prediction_context["nodata"] ): raise ValueError( f"{case['sample_id']}: prediction mask marks nodata as " f"valid at [{row_index}, {column_index}]" ) if not reference_valid: ignored_reference_count += 1 if prediction_valid: prediction_outside_reference_count += 1 continue if reference_value not in classes: raise ValueError( f"{case['sample_id']}: reference class outside ontology " f"at [{row_index}, {column_index}]" ) evaluated_reference_count += 1 reference_key = str(reference_value) if not prediction_valid: confusion[reference_key][missing_label] += 1 failures.append( { "pixel_index": row_index * reference_shape[1] + column_index, "row": row_index, "column": column_index, "reference": reference_value, "prediction": None, "reason": "prediction_masked_or_nodata", } ) continue if prediction_value not in classes: raise ValueError( f"{case['sample_id']}: prediction class outside ontology " f"at [{row_index}, {column_index}]" ) prediction_present_count += 1 prediction_key = str(prediction_value) confusion[reference_key][prediction_key] += 1 if reference_value == prediction_value: correct_count += 1 else: failures.append( { "pixel_index": row_index * reference_shape[1] + column_index, "row": row_index, "column": column_index, "reference": reference_value, "prediction": prediction_value, "reason": "class_mismatch", } ) per_class = {} for class_value in classes: class_key = str(class_value) tp = confusion[class_key][class_key] fp = sum( confusion[str(other)][class_key] for other in classes if other != class_value ) fn = sum( confusion[class_key][prediction] for prediction in [*labels, missing_label] if prediction != class_key ) values = count_metrics(tp, fp, fn) values["iou"] = safe_rate(tp, tp + fp + fn) per_class[class_key] = values metrics = { "pixel_count": reference_shape[0] * reference_shape[1], "evaluated_reference_pixel_count": evaluated_reference_count, "prediction_present_pixel_count": prediction_present_count, "ignored_reference_pixel_count": ignored_reference_count, "prediction_outside_reference_count": (prediction_outside_reference_count), "correct_pixel_count": correct_count, "prediction_coverage": safe_rate( prediction_present_count, evaluated_reference_count, ), "prediction_coverage_ci95_wilson": wilson_interval( prediction_present_count, evaluated_reference_count, ), "accuracy": safe_rate(correct_count, evaluated_reference_count), "accuracy_ci95_wilson": wilson_interval( correct_count, evaluated_reference_count, ), "mean_iou": _mean_or_none([value["iou"] for value in per_class.values()]), "macro_f1": _mean_or_none([value["f1"] for value in per_class.values()]), "per_class": per_class, "confusion_matrix": confusion, "alignment_validated": True, } return result(case, metrics, [], failures, []) def evaluate_vector_comparison(case: dict[str, Any]) -> dict[str, Any]: references, predictions = _validated_polygon_items(case) match_iou = _probability( case["config"].get("match_iou"), f"{case['sample_id']}.config.match_iou", ) matches, false_positives, false_negatives = greedy_match( predictions, references, _polygon_overlap, match_iou, class_aware=True, ) metrics = count_metrics(len(matches), len(false_positives), len(false_negatives)) metrics.update( { "mean_iou": _mean_or_none([item["overlap"] for item in matches]), "topologically_valid": True, "spatial_context_validated": True, } ) return result(case, metrics, matches, false_positives, false_negatives) def evaluate_change_detection(case: dict[str, Any]) -> dict[str, Any]: references, predictions = _validate_detection_case(case) match_iou = _probability( case["config"].get("match_iou"), f"{case['sample_id']}.config.match_iou", ) matches = [] false_positives = [] false_negatives = [] event_metrics = {} for event in case["classes"]: event_predictions = [item for item in predictions if item["class"] == event] event_references = [item for item in references if item["class"] == event] event_matches, event_fp, event_fn = greedy_match( event_predictions, event_references, lambda prediction, reference: bbox_iou( prediction["bbox"], reference["bbox"] ), match_iou, class_aware=True, ) matches.extend(dict(item, event=event) for item in event_matches) false_positives.extend(event_fp) false_negatives.extend(event_fn) event_metrics[str(event)] = count_metrics( len(event_matches), len(event_fp), len(event_fn) ) metrics = count_metrics(len(matches), len(false_positives), len(false_negatives)) metrics["event_metrics"] = event_metrics return result(case, metrics, matches, false_positives, false_negatives) def evaluate_terrain(case: dict[str, Any]) -> dict[str, Any]: references = case.get("references") predictions = case.get("predictions") if not isinstance(references, list) or not isinstance(predictions, list): raise ValueError(f"{case['sample_id']}: terrain inputs must be lists") if len(references) != len(predictions): raise ValueError(f"{case['sample_id']}: terrain vector lengths differ") units = case.get("units") if not isinstance(units, str) or not units.strip(): raise ValueError(f"{case['sample_id']}: terrain units must be explicit") normalized_references = [ _finite_float(value, f"{case['sample_id']}.references[{index}]") for index, value in enumerate(references) ] normalized_predictions: list[float | None] = [] for index, value in enumerate(predictions): normalized_predictions.append( None if value is None else _finite_float( value, f"{case['sample_id']}.predictions[{index}]", ) ) pairs = [ (reference, prediction) for reference, prediction in zip( normalized_references, normalized_predictions, strict=True, ) if prediction is not None ] errors = [prediction - reference for reference, prediction in pairs] missing = [ index for index, value in enumerate(normalized_predictions) if value is None ] metrics = { "units": units, "reference_count": len(normalized_references), "evaluated_count": len(pairs), "missing_count": len(missing), "coverage": safe_rate(len(pairs), len(normalized_references)), "coverage_ci95_wilson": wilson_interval(len(pairs), len(normalized_references)), "mae": _mean_or_none([abs(value) for value in errors]), "rmse": ( math.sqrt(mean(value * value for value in errors)) if errors else None ), "bias": _mean_or_none(errors), "error_sum": sum(errors), "absolute_error_sum": sum(abs(value) for value in errors), "squared_error_sum": sum(value * value for value in errors), "error_distribution": distribution(errors), } return result( case, metrics, [], [], [{"missing_index": index} for index in missing], ) def _normalized_anomalies( case: dict[str, Any], field: str, ) -> list[dict[str, str]]: values = case.get(field) if not isinstance(values, list): raise ValueError(f"{case['sample_id']}: {field} must be a list") normalized = [] for index, value in enumerate(values): if not isinstance(value, dict): raise ValueError( f"{case['sample_id']}: {field}[{index}] must include code and severity" ) code = value.get("code") severity = value.get("severity") if not isinstance(code, str) or not code.strip(): raise ValueError(f"{case['sample_id']}: {field}[{index}].code is invalid") if severity not in VALID_ANOMALY_SEVERITIES: raise ValueError( f"{case['sample_id']}: {field}[{index}].severity is invalid" ) normalized.append({"code": code, "severity": severity}) codes = [item["code"] for item in normalized] duplicates = sorted({code for code in codes if codes.count(code) > 1}) if duplicates: raise ValueError(f"{case['sample_id']}: duplicate {field} codes {duplicates}") return normalized def evaluate_validation(case: dict[str, Any]) -> dict[str, Any]: expected_items = _normalized_anomalies(case, "expected_anomalies") observed_items = _normalized_anomalies(case, "observed_anomalies") expected = {item["code"]: item for item in expected_items} observed = {item["code"]: item for item in observed_items} common_codes = sorted(expected.keys() & observed.keys()) matched_codes = [ code for code in common_codes if expected[code]["severity"] == observed[code]["severity"] ] severity_mismatches = [ { "anomaly": code, "expected_severity": expected[code]["severity"], "observed_severity": observed[code]["severity"], "reason": "severity_mismatch", } for code in common_codes if expected[code]["severity"] != observed[code]["severity"] ] matched = [ { "anomaly": code, "expected_severity": expected[code]["severity"], "observed_severity": observed[code]["severity"], } for code in matched_codes ] false_positives = [ {"anomaly": code, "severity": observed[code]["severity"]} for code in sorted(observed.keys() - expected.keys()) ] + [ { "anomaly": item["anomaly"], "severity": item["observed_severity"], "expected_severity": item["expected_severity"], "reason": "severity_mismatch", } for item in severity_mismatches ] false_negatives = [ {"anomaly": code, "severity": expected[code]["severity"]} for code in sorted(expected.keys() - observed.keys()) ] + [ { "anomaly": item["anomaly"], "severity": item["expected_severity"], "observed_severity": item["observed_severity"], "reason": "severity_mismatch", } for item in severity_mismatches ] metrics = count_metrics(len(matched), len(false_positives), len(false_negatives)) metrics["severity_mismatch_count"] = len(severity_mismatches) metrics["severity_mismatches"] = severity_mismatches metrics["blocker_or_critical_miss_count"] = sum( item["severity"] in {"blocker", "critical"} for item in false_negatives ) metrics["misses_by_severity"] = { severity: sum(item["severity"] == severity for item in false_negatives) for severity in sorted(VALID_ANOMALY_SEVERITIES) } return result(case, metrics, matched, false_positives, false_negatives) def _case_payload_fields(case: dict[str, Any]) -> tuple[str, str]: if case["task"] == "geospatial_data_validation": return "expected_anomalies", "observed_anomalies" return "references", "predictions" def result( case: dict[str, Any], metrics: dict[str, Any], matches: list[dict[str, Any]], false_positives: list[dict[str, Any]], false_negatives: list[dict[str, Any]], *, post_filter_predictions: Any | None = None, filter_description: dict[str, Any] | None = None, ) -> dict[str, Any]: reference_field, prediction_field = _case_payload_fields(case) exact_references = serializable(case[reference_field]) exact_predictions = serializable(case[prediction_field]) post_filter = ( exact_predictions if post_filter_predictions is None else serializable(post_filter_predictions) ) case_hash = canonical_hash(case) raw = { "sample_id": case["sample_id"], "task": case["task"], "metadata": serializable(case["metadata"]), "split": case["split"], "config": serializable(case["config"]), "input_lineage": serializable(case["lineage"]), "reference_input_field": reference_field, "prediction_input_field": prediction_field, "classes": serializable(case.get("classes")), "references": exact_references, "predictions_pre_filter": exact_predictions, "predictions_post_filter": post_filter, "filter": filter_description or {"applied": False}, "matches": serializable(matches), "false_positives": serializable(false_positives), "false_negatives": serializable(false_negatives), "input_sha256": case_hash, "hashes": { "case_input_canonical_json_sha256": case_hash, "references_canonical_json_sha256": canonical_hash(exact_references), "predictions_pre_filter_canonical_json_sha256": canonical_hash( exact_predictions ), "predictions_post_filter_canonical_json_sha256": canonical_hash( post_filter ), "canonicalization": CANONICAL_JSON_SPEC, }, } failures = [] for item in raw["false_positives"]: failures.append(failure_entry(case, "false_positive", item)) for item in raw["false_negatives"]: failures.append(failure_entry(case, "false_negative", item)) return { "sample_id": case["sample_id"], "task": case["task"], "metadata": serializable(case["metadata"]), "metrics": metrics, "raw": raw, "failures": failures, } def serializable(value: Any) -> Any: if isinstance(value, dict): return { key: serializable(item) for key, item in value.items() if key != "_shape" } if isinstance(value, list): return [serializable(item) for item in value] if isinstance(value, tuple): return [serializable(item) for item in value] return value def failure_entry( case: dict[str, Any], kind: str, evidence: dict[str, Any] ) -> dict[str, Any]: base_kind = kind if base_kind in {"false_positive", "false_negative"}: if case["task"] == "change_detection": kind = ( "event_false_positive" if base_kind == "false_positive" else "event_false_negative" ) elif case["task"] == "geospatial_data_validation": kind = ( "validation_false_positive" if base_kind == "false_positive" else "validation_false_negative" ) elif case["task"] == "terrain_interpretation": kind = "terrain_missing" elif case["task"] == "raster_classification": kind = ( "raster_misclassification" if evidence.get("reason") == "class_mismatch" else "raster_missing" ) contexts: list[str] = [] secondary_error_codes: list[str] = [] metadata = case.get("metadata", {}) if metadata.get("tile_edge") is True: contexts.append("tile_edge") if metadata.get("ood") is True: contexts.append("out_of_distribution") secondary_error_codes.append("M-OOD") confidence = evidence.get("confidence") diagnostic_thresholds = case.get("config", {}).get( "fixed_diagnostic_risk_thresholds" ) valid_thresholds = ( [ float(value) for value in diagnostic_thresholds if isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) and 0 <= float(value) <= 1 ] if isinstance(diagnostic_thresholds, list) else [] ) if ( base_kind == "false_positive" and isinstance(confidence, (int, float)) and not isinstance(confidence, bool) and valid_thresholds and float(confidence) >= max(valid_thresholds) ): contexts.append("high_confidence") secondary_error_codes.append("M-MISCALIBRATED") normalized_evidence = serializable(evidence) return { "failure_id": canonical_hash( { "sample_id": case["sample_id"], "kind": kind, "evidence": normalized_evidence, } )[:20], "sample_id": case["sample_id"], "task": case["task"], "error_code": ERROR_CODES[kind], "secondary_error_codes": sorted(set(secondary_error_codes)), "contexts": sorted(set(contexts)), "kind": kind, "metadata": serializable(metadata), "evidence": normalized_evidence, } EVALUATORS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { "object_detection": evaluate_object_detection, "footprint_segmentation": evaluate_footprint_segmentation, "raster_classification": evaluate_raster_classification, "vector_comparison": evaluate_vector_comparison, "change_detection": evaluate_change_detection, "terrain_interpretation": evaluate_terrain, "geospatial_data_validation": evaluate_validation, } def _capability( capability_id: str, task: str, implementation_paths: list[str], implementation_kind: str, evaluation_status: str, suitable_metrics: list[str], ) -> dict[str, Any]: return { "capability_id": capability_id, "task": task, "implementation_paths": implementation_paths, "implementation_kind": implementation_kind, "evaluation_status": evaluation_status, "suitable_metrics": suitable_metrics, "claim_boundary": ( "A family mapping is not evidence that this capability has a " "separate representative product benchmark." ), } def task_inventory() -> list[dict[str, Any]]: """Map concrete capabilities to evaluator families without overclaiming.""" synthetic = "synthetic_contract_case_only" covered = "covered_by_family_not_separately_benchmarked" separate = "not_separately_benchmarked" return [ _capability( "model_object_detection", "object_detection", ["backend/app/services/detection_service.py"], "model_inference_pipeline", synthetic, [ "precision", "recall", "F1", "AP50", "AP50-95", "IoU", "ECE", "Brier", "coverage-risk", ], ), _capability( "building_proposal_filtering", "object_detection", [ "scripts/train_building_proposal_classifier.py", "scripts/evaluate_belgium_building_candidate.py", ], "supporting_candidate_classifier", separate, ["candidate precision", "candidate recall", "F1", "calibration"], ), _capability( "footprint_segmentation", "footprint_segmentation", [ "backend/app/services/segmentation_service.py", "backend/app/services/segmentation_adapter.py", ], "model_inference_pipeline", synthetic, [ "object precision", "object recall", "F1", "IoU", "Dice", "boundary F1", "centroid distance", "area error", "topology", ], ), _capability( "detection_and_vector_qa", "vector_comparison", [ "backend/app/services/qa_service.py", "backend/app/services/detection_qa_service.py", "backend/app/services/quality_check_service.py", ], "deterministic_geospatial_comparison", synthetic, ["precision", "recall", "F1", "IoU", "topology", "coverage"], ), _capability( "vector_clip_buffer_intersect", "vector_comparison", [ "backend/app/services/vector_operations_service.py", "backend/app/services/vector_feature_service.py", ], "deterministic_vector_processing", covered, [ "geometry validity", "CRS correctness", "area conservation", "feature counts", "topology", ], ), _capability( "temporal_vector_change", "change_detection", ["backend/app/services/change_detection_service.py"], "deterministic_change_detection", synthetic, ["event precision", "event recall", "event F1", "IoU"], ), _capability( "thematic_raster_interpretation", "raster_classification", ["backend/app/services/thematic_raster_analysis_service.py"], "deterministic_source_interpretation", "synthetic_metric_contract_only_no_generic_learned_classifier_claim", ["pixel accuracy", "per-class F1", "per-class IoU", "mean IoU"], ), _capability( "raster_clip_reproject_indices", "raster_classification", [ "backend/app/services/raster_service.py", "backend/app/services/raster_operations_service.py", ], "deterministic_raster_processing", covered, [ "CRS/transform preservation", "pixel alignment", "nodata", "numeric tolerance", ], ), _capability( "raster_partition_mosaic", "raster_classification", ["backend/app/services/raster_partition_analysis_service.py"], "deterministic_raster_partitioning", separate, ["seam equality", "coverage completeness", "resolution consistency"], ), _capability( "terrain_height_interpretation", "terrain_interpretation", [ "backend/app/services/terrain_analysis_service.py", "backend/app/services/spw_terrain_service.py", ], "deterministic_continuous_raster_analysis", synthetic, ["MAE", "RMSE", "bias", "coverage", "unit integrity"], ), _capability( "flood_hazard_interpretation", "terrain_interpretation", ["backend/app/services/flood_hazard_analysis_service.py"], "deterministic_scenario_raster_analysis", covered, ["depth MAE/RMSE", "hazard-class IoU", "coverage", "scenario identity"], ), _capability( "bathymetry_interpretation", "terrain_interpretation", [ "backend/app/services/bathymetry_raster_analysis_service.py", "backend/app/services/mdk_bathymetry_probe_service.py", ], "deterministic_vertical_reference_analysis", covered, ["MAE", "RMSE", "bias", "coverage", "vertical-datum integrity"], ), _capability( "data_contract_validation_and_scan", "geospatial_data_validation", [ "backend/app/services/data_contract_validation.py", "scripts/run_accuracy_phase3_full_data_scan.py", ], "deterministic_validation", synthetic, ["anomaly precision", "anomaly recall", "anomaly F1", "critical misses"], ), _capability( "aoi_partition_orchestration", "geospatial_data_validation", [ "backend/app/services/aoi_operation_service.py", "backend/app/services/aoi_operation_executor.py", "backend/app/services/aoi_operation_worker.py", ], "deterministic_orchestration", separate, [ "partition completeness", "overlap/gap", "idempotency", "resume correctness", ], ), _capability( "geo_assistant_orchestration", "geospatial_data_validation", ["backend/app/services/geo_assistant_service.py"], "tool_orchestration_interface", "no_independent_accuracy_score_underlying_tool_results_are_authoritative", ["tool-selection correctness", "grounding", "unsupported-claim rate"], ), ] def _numeric_metric(metrics: dict[str, Any], key: str) -> float | None: value = metrics.get(key) if isinstance(value, bool) or not isinstance(value, (int, float)): return None converted = float(value) return converted if math.isfinite(converted) else None def _macro_metrics( values: list[dict[str, Any]], keys: Iterable[str], ) -> dict[str, Any]: output: dict[str, Any] = { "status": "computed", "method": "unweighted mean across cases with defined values", } computed = 0 for key in keys: items = [ value for metrics in values if (value := _numeric_metric(metrics, key)) is not None ] output[key] = mean(items) if items else None output[f"{key}_case_support"] = len(items) computed += bool(items) if not computed: output["status"] = "not_evaluable" return output def _aggregate_count_family( values: list[dict[str, Any]], extra_macro_keys: Iterable[str] = (), ) -> dict[str, Any]: metrics = [item["metrics"] for item in values] micro = count_metrics( sum(int(item["true_positive"]) for item in metrics), sum(int(item["false_positive"]) for item in metrics), sum(int(item["false_negative"]) for item in metrics), ) return { "micro": micro, "macro": _macro_metrics( metrics, ("precision", "recall", "f1", *extra_macro_keys), ), "observation_support": { "references": sum(int(item["reference_count"]) for item in metrics), "predictions": sum(int(item["prediction_count"]) for item in metrics), }, "primary_metric": { "name": "micro.f1", "value": micro["f1"], "direction": "higher_is_better", }, } def _pooled_detection_inputs( values: list[dict[str, Any]], ) -> tuple[ list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[str | int], ]: references: list[dict[str, Any]] = [] predictions_pre_filter: list[dict[str, Any]] = [] predictions_post_filter: list[dict[str, Any]] = [] matches: list[dict[str, Any]] = [] classes_by_hash: dict[str, str | int] = {} for value in sorted(values, key=lambda item: item.get("sample_id", "")): sample_id = value.get("sample_id") if not isinstance(sample_id, str) or not sample_id: raise ValueError("Pooled detection metrics require a sample_id") raw = _nonempty_mapping( value.get("raw"), f"{sample_id}: pooled detection raw evidence" ) required_lists = ( "references", "predictions_pre_filter", "predictions_post_filter", "matches", "classes", ) for field in required_lists: if not isinstance(raw.get(field), list): raise ValueError( f"{sample_id}: pooled detection raw.{field} must be a list" ) for class_value in raw["classes"]: classes_by_hash[canonical_hash(class_value)] = class_value def scoped(item: dict[str, Any], label: str) -> dict[str, Any]: if not isinstance(item, dict): raise ValueError(f"{sample_id}: pooled {label} must contain objects") identifier = _stable_id(item, f"{sample_id}: pooled {label}") return { **item, "id": f"{sample_id}::{identifier}", "_sample_id": sample_id, } references.extend(scoped(item, "references") for item in raw["references"]) predictions_pre_filter.extend( scoped(item, "predictions_pre_filter") for item in raw["predictions_pre_filter"] ) predictions_post_filter.extend( scoped(item, "predictions_post_filter") for item in raw["predictions_post_filter"] ) for match in raw["matches"]: if not isinstance(match, dict): raise ValueError(f"{sample_id}: pooled matches must contain objects") prediction_id = match.get("prediction_id") if not isinstance(prediction_id, str) or not prediction_id: raise ValueError( f"{sample_id}: pooled match prediction_id must be non-empty" ) matches.append({**match, "prediction_id": f"{sample_id}::{prediction_id}"}) return ( references, predictions_pre_filter, predictions_post_filter, matches, [classes_by_hash[key] for key in sorted(classes_by_hash)], ) def _aggregate_detection(values: list[dict[str, Any]]) -> dict[str, Any]: aggregation = _aggregate_count_family(values) references, predictions, retained, matches, classes = _pooled_detection_inputs( values ) per_class = {} for class_value in classes: class_references = [item for item in references if item["class"] == class_value] class_predictions = [ item for item in predictions if item["class"] == class_value ] ap50 = detection_ap(class_predictions, class_references, 0.5) ap50_95 = _mean_or_none( detection_ap(class_predictions, class_references, 0.5 + step * 0.05) for step in range(10) ) per_class[str(class_value)] = { "reference_count": len(class_references), "prediction_count": len(class_predictions), "ap50": ap50, "ap50_95": ap50_95, } aggregation["micro"].update( { "ap50": detection_ap(predictions, references, 0.5), "ap50_95": _mean_or_none( detection_ap(predictions, references, 0.5 + step * 0.05) for step in range(10) ), "map50": _mean_or_none(item["ap50"] for item in per_class.values()), "map50_95": _mean_or_none(item["ap50_95"] for item in per_class.values()), "per_class_ap": per_class, "calibration": calibration_metrics(retained, matches), "ranking_scope": ( "predictions pooled across cases; class- and sample-aware matching; " "no averaging of per-case AP" ), } ) aggregation["observation_support"].update( { "ranking_predictions": len(predictions), "calibration_predictions": len(retained), } ) return aggregation def _aggregate_raster(values: list[dict[str, Any]]) -> dict[str, Any]: metrics = [item["metrics"] for item in values] class_counts: dict[str, dict[str, int]] = defaultdict( lambda: {"tp": 0, "fp": 0, "fn": 0} ) for item in metrics: for class_name, class_metrics in item["per_class"].items(): class_counts[class_name]["tp"] += int(class_metrics["true_positive"]) class_counts[class_name]["fp"] += int(class_metrics["false_positive"]) class_counts[class_name]["fn"] += int(class_metrics["false_negative"]) per_class = {} for class_name, counts in sorted(class_counts.items()): class_metrics = count_metrics(counts["tp"], counts["fp"], counts["fn"]) class_metrics["iou"] = safe_rate( counts["tp"], counts["tp"] + counts["fp"] + counts["fn"], ) per_class[class_name] = class_metrics evaluated = sum(int(item["evaluated_reference_pixel_count"]) for item in metrics) present = sum(int(item["prediction_present_pixel_count"]) for item in metrics) correct = sum(int(item["correct_pixel_count"]) for item in metrics) micro = { "evaluated_reference_pixel_count": evaluated, "prediction_present_pixel_count": present, "correct_pixel_count": correct, "accuracy": safe_rate(correct, evaluated), "accuracy_ci95_wilson": wilson_interval(correct, evaluated), "prediction_coverage": safe_rate(present, evaluated), "prediction_coverage_ci95_wilson": wilson_interval(present, evaluated), "mean_iou": _mean_or_none([item["iou"] for item in per_class.values()]), "macro_f1_across_classes": _mean_or_none( [item["f1"] for item in per_class.values()] ), "per_class": per_class, } return { "micro": micro, "macro": _macro_metrics( metrics, ("accuracy", "mean_iou", "macro_f1"), ), "observation_support": {"evaluated_pixels": evaluated}, "primary_metric": { "name": "micro.mean_iou", "value": micro["mean_iou"], "direction": "higher_is_better", }, } def _aggregate_terrain(values: list[dict[str, Any]]) -> dict[str, Any]: metrics = [item["metrics"] for item in values] units = sorted({str(item["units"]) for item in metrics}) references = sum(int(item["reference_count"]) for item in metrics) evaluated = sum(int(item["evaluated_count"]) for item in metrics) if len(units) != 1: return { "micro": {"status": "not_evaluable", "reason": "mixed units"}, "macro": {"status": "not_evaluable", "reason": "mixed units"}, "observation_support": {"references": references}, "primary_metric": { "name": "micro.rmse", "value": None, "direction": "lower_is_better", }, } absolute_error_sum = sum(float(item["absolute_error_sum"]) for item in metrics) squared_error_sum = sum(float(item["squared_error_sum"]) for item in metrics) error_sum = sum(float(item["error_sum"]) for item in metrics) micro = { "status": "computed" if evaluated else "not_evaluable", "units": units[0], "reference_count": references, "evaluated_count": evaluated, "coverage": safe_rate(evaluated, references), "coverage_ci95_wilson": wilson_interval(evaluated, references), "mae": safe_rate(absolute_error_sum, evaluated), "rmse": (math.sqrt(squared_error_sum / evaluated) if evaluated else None), "bias": safe_rate(error_sum, evaluated), } return { "micro": micro, "macro": _macro_metrics( metrics, ("coverage", "mae", "rmse", "bias"), ), "observation_support": { "references": references, "evaluated": evaluated, }, "primary_metric": { "name": "micro.rmse", "value": micro["rmse"], "direction": "lower_is_better", }, } def _aggregate_task(values: list[dict[str, Any]]) -> dict[str, Any]: task = values[0]["task"] if task == "object_detection": aggregation = _aggregate_detection(values) elif task == "raster_classification": aggregation = _aggregate_raster(values) elif task == "terrain_interpretation": aggregation = _aggregate_terrain(values) else: extras = { "footprint_segmentation": ( "mean_iou", "mean_dice", "mean_boundary_f1", ), "vector_comparison": ("mean_iou",), "change_detection": (), "geospatial_data_validation": ("blocker_or_critical_miss_count",), }[task] aggregation = _aggregate_count_family(values, extras) if task == "geospatial_data_validation": aggregation["micro"]["blocker_or_critical_miss_count"] = sum( int(item["metrics"]["blocker_or_critical_miss_count"]) for item in values ) case_support = len(values) status = ( "evaluable" if case_support >= SUBGROUP_MIN_CASE_SUPPORT and aggregation["primary_metric"]["value"] is not None else "insufficient_support" ) return { "case_support": case_support, "minimum_case_support": SUBGROUP_MIN_CASE_SUPPORT, "status": status, "release_gate_status": "not_evaluable", "release_gate_reason": ( "insufficient_support" if status == "insufficient_support" else "no_frozen_release_target" ), **aggregation, } def _stratum_key(value: Any) -> str: return "__not_applicable__" if value is None else str(value) def _worst_stratum_by_task(strata: dict[str, Any]) -> dict[str, Any]: tasks = sorted( {task for stratum in strata.values() for task in stratum["task_metrics"]} ) output = {} for task in tasks: candidates = [] for stratum_name, stratum in strata.items(): task_metrics = stratum["task_metrics"].get(task) if not task_metrics or task_metrics["status"] != "evaluable": continue primary = task_metrics["primary_metric"] if primary["value"] is not None: candidates.append((stratum_name, primary)) if not candidates: output[task] = { "status": "not_evaluable", "reason": ( "no stratum meets minimum support with a defined primary metric" ), } continue direction = candidates[0][1]["direction"] selected = ( min(candidates, key=lambda item: (item[1]["value"], item[0])) if direction == "higher_is_better" else max( candidates, key=lambda item: (item[1]["value"], item[0]), ) ) output[task] = { "status": "computed", "stratum": selected[0], "primary_metric": selected[1], } return output def subgroup_report(results: list[dict[str, Any]]) -> dict[str, Any]: dimensions = ( "region", "municipality", "urbanity", "object_size", "source", "sensor", "resolution_m", "season", "date", "vegetation", "occlusion", "difficulty", "context", ) dimension_reports = {} any_insufficient = False for dimension in dimensions: groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for item in results: groups[_stratum_key(item["metadata"].get(dimension))].append(item) strata = {} for key, values in sorted(groups.items()): per_task: dict[str, list[dict[str, Any]]] = defaultdict(list) for value in values: per_task[value["task"]].append(value) task_metrics = { task: _aggregate_task(task_values) for task, task_values in sorted(per_task.items()) } insufficient = any( item["status"] == "insufficient_support" for item in task_metrics.values() ) any_insufficient = any_insufficient or insufficient strata[key] = { "case_support": len(values), "tasks": sorted(per_task), "failure_count": sum(len(value["failures"]) for value in values), "task_metrics": task_metrics, "release_gate_status": "not_evaluable", "release_gate_reason": "insufficient_support" if insufficient else "no_frozen_release_target", } dimension_reports[dimension] = { "strata": strata, "worst_stratum_by_task": _worst_stratum_by_task(strata), } return { "schema_version": REPORT_SCHEMA_VERSION, "minimum_case_support": SUBGROUP_MIN_CASE_SUPPORT, "support_unit": ( "independent benchmark cases; pixel/object counts are reported " "separately and do not replace case support" ), "overall_status": ( "not_evaluable" if any_insufficient else "evaluable_no_release_target" ), "dimensions": dimension_reports, } def evaluate_cases(path: Path, allowed_sample_ids: set[str]) -> dict[str, Any]: file_hash = file_sha256(path) portfolio = json.loads(path.read_text(encoding="utf-8")) if not isinstance(portfolio, dict): raise ValueError("The evaluation portfolio must be a JSON object") schema_version = portfolio.get("schema_version") if type(schema_version) is not int or schema_version != REPORT_SCHEMA_VERSION: raise ValueError( f"schema_version must be exactly integer {REPORT_SCHEMA_VERSION}" ) _meaningful_string(portfolio.get("portfolio_id"), "portfolio_id") protected_policy = portfolio.get("protected_policy") if protected_policy != EXPECTED_PROTECTED_POLICY: raise ValueError( "protected_policy must exactly equal the frozen no-selection policy: " f"{EXPECTED_PROTECTED_POLICY}" ) portfolio_kind = portfolio.get("portfolio_kind") if portfolio_kind not in PORTFOLIO_KINDS: raise ValueError(f"portfolio_kind must be one of {sorted(PORTFOLIO_KINDS)}") claim_boundary = portfolio.get("claim_boundary") claim_lower = claim_boundary.lower() if isinstance(claim_boundary, str) else "" if portfolio_kind == "synthetic_contract": if "synthetic" not in claim_lower: raise ValueError( "A synthetic_contract claim boundary must explicitly state that it is synthetic" ) elif "governed product baseline" not in claim_lower or "synthetic" in claim_lower: raise ValueError( "A governed_product_baseline claim boundary must explicitly state " "'governed product baseline' and must not describe the portfolio as synthetic" ) split_roles = portfolio.get("split_roles") if split_roles not in (["test"], ["test", "background-test"]): raise ValueError( "split_roles must be exactly ['test'] or ['test', 'background-test']" ) if "challenge_cases" in portfolio or "challenge_labels" in portfolio: raise ValueError("Challenge cases and labels must remain sealed and absent") selection_policy = _meaningful_string( portfolio.get("selection_policy"), "selection_policy" ) declared_portfolio_lineage = _nonempty_mapping( portfolio.get("portfolio_lineage"), "portfolio_lineage", ) for field in ("origin", "source_path", "version"): _meaningful_string( declared_portfolio_lineage.get(field), f"portfolio_lineage.{field}", ) cases = portfolio.get("cases") if not isinstance(cases, list) or not all(isinstance(item, dict) for item in cases): raise ValueError("The portfolio cases must be an object list") for case in cases: if case.get("split") not in split_roles: raise ValueError( f"{case['sample_id']}: split must be one of {split_roles}; " "challenge remains sealed" ) _validate_case_common(case) identifiers = [item["sample_id"] for item in cases] duplicates = sorted( {identifier for identifier in identifiers if identifiers.count(identifier) > 1} ) if duplicates: raise ValueError(f"Duplicate protected case ids: {duplicates}") case_ids = set(identifiers) observed_tasks = {item["task"] for item in cases} if portfolio_kind == "governed_product_baseline" and observed_tasks != TASKS: raise ValueError( "A governed_product_baseline must cover every evaluator task family: " f"missing={sorted(TASKS - observed_tasks)}, " f"unexpected={sorted(observed_tasks - TASKS)}" ) unexpected = sorted(case_ids - allowed_sample_ids) missing = sorted(allowed_sample_ids - case_ids) if unexpected or missing: raise ValueError( "Protected case identity mismatch: " f"unexpected={unexpected}, missing={missing}" ) portfolio_canonical_hash = canonical_hash(portfolio) results = [ EVALUATORS[item["task"]](item) for item in sorted(cases, key=lambda item: item["sample_id"]) ] portfolio_lineage = { "declared": serializable(declared_portfolio_lineage), "portfolio_id": portfolio.get("portfolio_id"), "portfolio_schema_version": portfolio.get("schema_version"), "portfolio_kind": portfolio_kind, "portfolio_file_sha256": file_hash, "portfolio_canonical_json_sha256": portfolio_canonical_hash, "source_path": declared_portfolio_lineage["source_path"], "split_roles": list(split_roles), "claim_boundary": claim_boundary, } for item in results: item["raw"]["portfolio_lineage"] = portfolio_lineage results_by_task: dict[str, list[dict[str, Any]]] = defaultdict(list) for item in results: results_by_task[item["task"]].append(item) portfolio_metrics = { task: _aggregate_task(task_results) for task, task_results in sorted(results_by_task.items()) } failures = sorted( (failure for item in results for failure in item["failures"]), key=lambda item: item["failure_id"], ) results_hash = canonical_hash(results) return { "schema_version": REPORT_SCHEMA_VERSION, "evaluator_version": EVALUATOR_VERSION, "portfolio_id": portfolio["portfolio_id"], "portfolio_kind": portfolio_kind, "portfolio_file_sha256": file_hash, "portfolio_canonical_json_sha256": portfolio_canonical_hash, "claim_boundary": claim_boundary, "selection_policy": selection_policy, "split_roles": list(split_roles), "protected_policy": serializable(portfolio.get("protected_policy")), "hash_specification": { "algorithm": "SHA-256", "canonical_json": CANONICAL_JSON_SPEC, "portfolio_file_sha256_input": "exact source-file bytes", "portfolio_canonical_json_sha256_input": ("parsed complete portfolio"), "results_canonical_json_sha256_input": ( "complete sample-id-ordered results array" ), }, "task_inventory": task_inventory(), "evaluated_task_families": sorted({item["task"] for item in results}), "task_count": len({item["task"] for item in results}), "case_count": len(results), "results": results, "subgroups": subgroup_report(results), "failures": failures, "failure_taxonomy": { "primary_codes": FAILURE_TAXONOMY, "contexts": FAILURE_CONTEXTS, "claim_boundary": "diagnostic classification; not a release decision", }, "portfolio_metrics": portfolio_metrics, "results_canonical_json_sha256": results_hash, }