Spaces:
Paused
Paused
| """Fail closed before a specialist output becomes a finding or downloadable evidence.""" | |
| import math | |
| import json | |
| import numpy as np | |
| import rasterio | |
| from rasterio.features import rasterize | |
| from rasterio.transform import Affine | |
| from scipy import ndimage | |
| from shapely.geometry import shape | |
| def validate_result_evidence(result,output): | |
| for path in result.get("paths",[]): | |
| if not path.resolve().is_relative_to(output.resolve()) or not path.is_file() or not path.stat().st_size: | |
| raise ValueError("A required evidence artifact is missing or outside its analysis directory.") | |
| score=result.get("evidence_strength",0) | |
| if not math.isfinite(score) or not 0<=score<=1: raise ValueError("Invalid evidence strength; finding withheld.") | |
| features=result.get("features") | |
| if features is not None: | |
| ids=[f.get("id") for f in features] | |
| if len(ids)!=len(set(ids)): raise ValueError("Duplicate evidence feature IDs; finding withheld.") | |
| for key in ("count", "building_count"): | |
| if key in result and result[key]!=len(features): raise ValueError("Count does not match final features.") | |
| for feature in features: | |
| geom=shape(feature["geometry"]) | |
| if not geom.is_valid or geom.is_empty or geom.area<=0: raise ValueError("Invalid evidence geometry.") | |
| for path in result.get("paths",[]): | |
| if path.suffix==".geojson" and path.stem not in {"water_gain","water_loss","vegetation_gain","vegetation_loss","built_up_gain","built_up_loss"} and json.loads(path.read_text())["features"]!=json.loads(json.dumps(features)): | |
| raise ValueError("Exported GeoJSON differs from canonical features.") | |
| for key in ("area_m2","selected_pixels","count","region_count"): | |
| value=result.get(key) | |
| if value is not None and (not math.isfinite(value) or value<0): raise ValueError("Invalid spatial measurement.") | |
| if result.get("quality_gate_passed") is False and not result.get("count_reliability"): | |
| raise ValueError("Specialist quality validation failed.") | |
| finals = [p for p in result.get("paths",[]) if p.name in {"water_mask.tif","buildings_labels.tif","land_cover_map.tif"}] | |
| for path in finals: | |
| with rasterio.open(path) as src: | |
| labels = src.read(1); selected = labels > 0; valid = src.read_masks(1) > 0 | |
| if np.any(selected & ~valid): raise ValueError("Final mask includes NoData pixels.") | |
| if result.get("selected_pixels") != int(selected.sum()): raise ValueError("Final mask and pixel count disagree.") | |
| if result.get("area_m2") is not None and src.crs is None: raise ValueError("Area is unavailable without a CRS.") | |
| if features is not None: | |
| polygon_pixels = sum(f["properties"].get("area_pixels",0) for f in features) | |
| if not math.isclose(polygon_pixels,float(selected.sum()),abs_tol=1e-5): | |
| raise ValueError("Final mask and polygon areas disagree.") | |
| # Equal totals cannot detect shifted footprints or swapped instance IDs. | |
| expected_labels = ndimage.label(selected)[0] if path.name == "water_mask.tif" else labels | |
| identifiers, counts = np.unique(expected_labels[selected], return_counts=True) | |
| pixels_by_id = dict(zip(identifiers.tolist(), counts.tolist())) | |
| label_ids = set(pixels_by_id) | |
| if label_ids != {f["id"] for f in features}: | |
| raise ValueError("Final mask and feature IDs disagree.") | |
| geometries = [] | |
| for feature in features: | |
| pixel_geometry = feature["properties"].get("pixel_geometry") | |
| if pixel_geometry is None: | |
| raise ValueError("Pixel geometry is required to verify evidence alignment.") | |
| pixel_shape = shape(pixel_geometry) | |
| pixel_count = pixels_by_id[feature["id"]] | |
| if (not pixel_shape.is_valid or pixel_shape.is_empty | |
| or not math.isclose(pixel_shape.area, pixel_count, abs_tol=1e-5) | |
| or not math.isclose(feature["properties"].get("area_pixels", -1), pixel_count, abs_tol=1e-5)): | |
| raise ValueError("Final mask and individual feature areas disagree.") | |
| geometries.append((pixel_geometry, feature["id"])) | |
| reconstructed = (rasterize(geometries, out_shape=labels.shape, transform=Affine.identity(), dtype="int32") | |
| if geometries else np.zeros_like(labels)) | |
| if not np.array_equal(reconstructed, expected_labels): | |
| raise ValueError("Final mask and polygon pixel alignment disagree.") | |
| if path.name == "land_cover_map.tif": | |
| classes = json.loads(src.tags().get("classes", "{}")) | |
| breakdown = result.get("breakdown", {}) | |
| if set(classes.values()) != set(breakdown) or np.any(valid & ~selected): | |
| raise ValueError("Land-cover classes do not cover the valid grid.") | |
| for ident, name in classes.items(): | |
| pixels = int(np.count_nonzero(labels == int(ident))) | |
| part = breakdown[name] | |
| if (part.get("pixels") != pixels or not valid.any() | |
| or not math.isclose(part.get("percent", -1), 100*pixels/int(valid.sum()), abs_tol=.001)): | |
| raise ValueError("Land-cover class measurements disagree with the final mask.") | |
| if "coverage_percent" in result: | |
| denominator = result.get("valid_pixels", int(valid.sum())) | |
| if not denominator or not math.isclose(result["coverage_percent"],100*selected.sum()/denominator,abs_tol=.001): | |
| raise ValueError("Final mask and coverage percentage disagree.") | |
| if "breakdown" in result: | |
| parts = result["breakdown"].values() | |
| if not math.isclose(sum(p["percent"] for p in parts),100,abs_tol=.001): | |
| raise ValueError("Land-cover class percentages do not sum to 100%.") | |