""" Multi-Method Verification — a second, independently-computed deterministic check that cross-validates the primary change-detection / optical-SAR fusion result. Phase 15. Closes a gap flagged in the Phase 14 review (master brief §13B / uisuggestions.docx "Multi-Model Verification: 3/3 models support the detected change"): PHASE12.md and evidence/fusion.py both explicitly deferred this — fusion.py's CONFLICTING check only cross-checks two numbers computed by the SAME pipeline; nothing independent existed. HONESTY CONTRACT — read before presenting this to a judge: - This is NOT a second trained model, and it is NOT the "3/3 models" multi-model consensus sketched in the master brief. There is exactly one image-processing pipeline behind each task (see models/change.py, models/optical_sar.py) — no second neural network was integrated in Phase 15. - What this IS: a second classical CV heuristic that uses a genuinely different algorithm/statistic/granularity than the primary method, so the two do not share a code path and can actually disagree: * Change detection — primary method is a per-PIXEL z-score threshold (models/change.py). The independent check here is a per-BLOCK (patch-mean) Tukey-IQR outlier rule — different spatial scale, different outlier statistic. * Optical+SAR fusion — primary method is a fixed-percentile threshold per modality (models/optical_sar.py). The independent check here is Otsu's automatic histogram threshold (Otsu, 1979) — a different, data-adaptive thresholding rule. - If a judge asks "are these two independent AI models", the honest answer is no — they are two different deterministic CV heuristics cross-checking the same uploaded pixels. Real multi-model (i.e. multiple independently-trained models) verification is a larger, separate feature and is still not built. - Agreement/disagreement here is informational, exactly like the existing CONFLICTING state — it never blocks a response, only adds a trace line and a `verification` object to `tool_output` so the disagreement is visible, not hidden. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path import numpy as np import rasterio from PIL import Image from preprocessing.raster_to_rgb import to_uint8 _BLOCK = 16 # patch size (pixels) for the block-level change check _HIGH_CHANGE_PCT = 8.0 _MEDIUM_CHANGE_PCT = 2.0 @dataclass class VerificationResult: method_a: str method_a_level: str method_b: str method_b_level: str agree: bool note: str def _load_gray(path: Path) -> np.ndarray: with rasterio.open(path) as src: band = src.read(1).astype("float32") if src.count >= 3: g = src.read(2).astype("float32") b = src.read(3).astype("float32") band = (band + g + b) / 3.0 return to_uint8(band).astype("float32") def _resize_to_match(a: np.ndarray, b: np.ndarray) -> tuple[np.ndarray, np.ndarray]: if a.shape == b.shape: return a, b h, w = min(a.shape[0], b.shape[0]), min(a.shape[1], b.shape[1]) a2 = np.array(Image.fromarray(a.astype("uint8")).resize((w, h))).astype("float32") b2 = np.array(Image.fromarray(b.astype("uint8")).resize((w, h))).astype("float32") return a2, b2 def _level_from_pct(pct: float) -> str: if pct >= _HIGH_CHANGE_PCT: return "high" if pct >= _MEDIUM_CHANGE_PCT: return "medium" return "low" def verify_change(before_path: Path, after_path: Path, primary_level: str, primary_pct: float) -> VerificationResult: """Independent check for change_detection / change_vqa results.""" before = _load_gray(before_path) after = _load_gray(after_path) before, after = _resize_to_match(before, after) h, w = before.shape bh, bw = h // _BLOCK, w // _BLOCK if bh == 0 or bw == 0: # Image too small to block at this patch size — whole-image # fallback so this never crashes on tiny demo/benchmark images. pct = float(abs(after.mean() - before.mean()) > 1.0) * 100.0 else: crop_h, crop_w = bh * _BLOCK, bw * _BLOCK b_blocks = before[:crop_h, :crop_w].reshape(bh, _BLOCK, bw, _BLOCK).mean(axis=(1, 3)) a_blocks = after[:crop_h, :crop_w].reshape(bh, _BLOCK, bw, _BLOCK).mean(axis=(1, 3)) diff = np.abs(a_blocks - b_blocks) q1, q3 = np.percentile(diff, 25), np.percentile(diff, 75) iqr = (q3 - q1) or 1.0 fence = q3 + 1.5 * iqr # Tukey's outlier fence — deliberately not a z-score pct = float((diff > fence).mean() * 100) level = _level_from_pct(pct) agree = level == primary_level note = ( f"Primary (pixel-level z-score, models/change.py): {primary_pct:.1f}% of pixels -> " f"{primary_level.upper()}. Independent check (block-level Tukey-outlier, {_BLOCK}x{_BLOCK}px " f"patches): {pct:.1f}% of patches -> {level.upper()}. " + ("Both methods agree." if agree else "The two methods DISAGREE — treat this result with extra caution.") ) return VerificationResult( method_a="pixel_diff_zscore_threshold", method_a_level=primary_level, method_b="block_mean_tukey_outlier", method_b_level=level, agree=agree, note=note, ) def _otsu_threshold(arr: np.ndarray) -> float: """Classic Otsu (1979) automatic bimodal threshold from the 0-255 histogram. Deliberately independent of models/optical_sar.py's fixed 25th/85th-percentile rule — this one adapts to each image's own intensity distribution instead of a hardcoded cutoff.""" hist, _ = np.histogram(arr, bins=256, range=(0, 255)) hist = hist.astype("float64") total = hist.sum() if total == 0: return 127.0 sum_total = float(np.dot(np.arange(256), hist)) sum_b = w_b = max_var = 0.0 threshold = 127 for i in range(256): w_b += hist[i] if w_b == 0: continue w_f = total - w_b if w_f == 0: break sum_b += i * hist[i] m_b = sum_b / w_b m_f = (sum_total - sum_b) / w_f var_between = w_b * w_f * (m_b - m_f) ** 2 if var_between > max_var: max_var = var_between threshold = i return float(threshold) def verify_optical_sar( optical_path: Path, sar_path: Path, target: str, primary_agreement_pct: float, primary_level: str ) -> VerificationResult: """Independent check for optical_sar fusion results.""" optical = _load_gray(optical_path) sar = _load_gray(sar_path) optical, sar = _resize_to_match(optical, sar) opt_t = _otsu_threshold(optical) sar_t = _otsu_threshold(sar) if target == "water": opt_mask = optical <= opt_t sar_mask = sar <= sar_t else: opt_mask = optical >= opt_t sar_mask = sar >= sar_t agreement_pct = float((opt_mask & sar_mask).mean() * 100) if agreement_pct >= 3.0: level = "high" elif agreement_pct >= 0.5: level = "medium" else: level = "low" agree = level == primary_level note = ( f"Primary (fixed-percentile threshold, models/optical_sar.py): {primary_agreement_pct:.1f}% " f"agreement -> {primary_level.upper()}. Independent check (Otsu automatic threshold): " f"{agreement_pct:.1f}% agreement -> {level.upper()}. " + ("Both methods agree." if agree else "The two methods DISAGREE on confidence level — treat with extra caution.") ) return VerificationResult( method_a="percentile_threshold_agreement", method_a_level=primary_level, method_b="otsu_automatic_threshold_agreement", method_b_level=level, agree=agree, note=note, )