Spaces:
Paused
Paused
File size: 2,806 Bytes
a358495 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """Request-scoped water evidence shared by surface specialists."""
from pathlib import Path
import numpy as np
import rasterio
class SurfaceContext:
def __init__(self, output: Path):
self.output = output
self.results = {}
def water(self, source, *, largest=False, strict=False):
from satquery_engine.services.measurements import measure_cover
from satquery_engine.services.quality_gate import validate_result_evidence
key = (str(Path(source).resolve()), largest, strict)
if key not in self.results:
folder = self.output / f"water_{len(self.results)}"
result = measure_cover(source, folder, "water", largest=largest, strict=strict)
validate_result_evidence(result, folder)
self.results[key] = result
return self.results[key]
def exclude_water_instances(labels, valid, source, water_result, minimum_pixels):
"""Reject water-dominated instances; trim shoreline overlap without new IDs."""
if water_result.get("quality_gate_passed") is False:
raise ValueError("Water evidence failed validation; building count withheld.")
paths = [Path(p) for p in water_result.get("paths", []) if Path(p).name == "water_mask.tif"]
if len(paths) != 1:
raise ValueError("A unique canonical water mask is required for building exclusion.")
with rasterio.open(source) as src, rasterio.open(paths[0]) as mask:
if ((mask.height, mask.width) != labels.shape or mask.shape != src.shape
or mask.transform != src.transform or mask.crs != src.crs):
raise ValueError("Water and building evidence must use the same source grid.")
water = mask.read(1) > 0
if np.any(water & ~(mask.read_masks(1) > 0)) or np.any(water & ~valid):
raise ValueError("Water exclusion evidence contains invalid pixels.")
sizes = np.bincount(labels.ravel())
overlaps = np.bincount(labels[water], minlength=len(sizes))
reject = (overlaps >= sizes * .5) | ((sizes - overlaps) < minimum_pixels)
reject[0] = False
filtered = np.where(water | reject[labels] | ~valid, 0, labels)
ids = np.unique(filtered)
ids = ids[ids > 0]
mapping = np.zeros(len(sizes), dtype="int32")
mapping[ids] = np.arange(1, len(ids) + 1)
return mapping[filtered], {
"status": "applied", "water_model": water_result.get("model_id", water_result.get("method")),
"water_overlap_pixels": int(overlaps[1:].sum()),
"rejected_instances": int(np.count_nonzero(reject[1:] & (sizes[1:] > 0))),
"removed_building_pixels": int(np.count_nonzero(labels) - np.count_nonzero(filtered)),
"policy": "Reject instances with at least 50% water overlap; trim other water pixels and remove undersized remnants.",
}
|