Spaces:
Paused
Paused
| """ | |
| Optical + SAR cross-modal fusion specialist "model". | |
| Phase 8 (P0). Satisfies PS requirements: | |
| - "Cross-modal pair analysis: The system must extract complementary | |
| information from a co-registered optical/multispectral and SAR image | |
| pair." | |
| - This is also the backend half of the master brief's PRIMARY HERO | |
| FEATURE (cloud obstruction -> evidence insufficiency -> SAR | |
| activation -> multimodal evidence fusion). The Evidence Sufficiency | |
| Engine that decides WHEN to activate this is Phase 9 scope (P1/P2 per | |
| brief §12) — this phase only builds the fusion analysis itself so it | |
| exists to be activated. | |
| MODEL CHOICE — deterministic per-pixel thresholding + fusion, NOT a | |
| trained multimodal fusion network. | |
| - Same constraint as Phase 7: no pip-installable, genuinely pretrained | |
| optical+SAR fusion model exists that meets "prefer existing pretrained | |
| models" + 4-day integration. The master brief explicitly allows this | |
| ("Optical features + SAR features -> simple fusion/classifier -> | |
| prediction" — Day 3 guidance). | |
| - Optical channel: brightness-based water/built-up heuristic (NOT NDWI — | |
| that needs a NIR band this pipeline doesn't guarantee is present/known; | |
| using a labeled "brightness heuristic", not claiming spectral-index | |
| validity). | |
| - SAR channel: backscatter-intensity threshold (low backscatter ~ open | |
| water / flooded surfaces; high backscatter ~ built-up/rough surfaces). | |
| This is the standard qualitative SAR interpretation heuristic, not a | |
| calibrated sigma-naught classifier — no radiometric calibration is | |
| performed on the uploaded raster. | |
| - Fusion rule: simple agreement logic per pixel (BOTH indicate water -> | |
| high-confidence water; only one indicates it -> low-confidence; | |
| neither -> not flagged). This directly demonstrates "extract | |
| complementary information" without inventing a black-box model. | |
| - Every number in the output is computed from the actual uploaded pixels | |
| at request time — nothing here is a canned/demo constant. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from typing import Literal | |
| import numpy as np | |
| import rasterio | |
| from preprocessing.raster_to_rgb import to_uint8 | |
| _WATER_DARK_PCT = 25 # optical: bottom 25th percentile brightness -> "dark/possible water" | |
| _SAR_LOW_BACKSCATTER_PCT = 25 # SAR: bottom 25th percentile intensity -> "low backscatter/possible water or flood" | |
| _BUILTUP_BRIGHT_PCT = 85 # optical: top 15% brightness -> "possible built-up/bare surface" | |
| _SAR_HIGH_BACKSCATTER_PCT = 85 # SAR: top 15% intensity -> "possible built-up/rough surface" | |
| class FusionRegionStats: | |
| label: str | |
| optical_only_pct: float | |
| sar_only_pct: float | |
| agreement_pct: float # both modalities agree -> high-confidence | |
| confidence_level: Literal["high", "medium", "low"] | |
| class OpticalSarResult: | |
| target: str # "water" | "built_up" | |
| stats: FusionRegionStats | |
| answer: str | |
| width: int | |
| height: int | |
| method: 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, b = src.read(2).astype("float32"), 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, bool]: | |
| if a.shape == b.shape: | |
| return a, b, False | |
| from PIL import Image | |
| 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, True | |
| def _target_from_query(query: str) -> str: | |
| q = query.lower() | |
| if any(k in q for k in ("water", "flood", "flooded", "river", "lake")): | |
| return "water" | |
| return "built_up" # default — matches PS representative query "identify built-up and water-covered regions" | |
| def run_optical_sar(optical_path: Path, sar_path: Path, query: str) -> OpticalSarResult: | |
| optical = _load_gray(optical_path) | |
| sar = _load_gray(sar_path) | |
| optical, sar, was_resized = _resize_to_match(optical, sar) | |
| target = _target_from_query(query) | |
| if target == "water": | |
| opt_thresh = np.percentile(optical, _WATER_DARK_PCT) | |
| sar_thresh = np.percentile(sar, _SAR_LOW_BACKSCATTER_PCT) | |
| opt_mask = optical <= opt_thresh # dark in optical | |
| sar_mask = sar <= sar_thresh # low backscatter in SAR | |
| label = "water / flooded surface" | |
| else: | |
| opt_thresh = np.percentile(optical, _BUILTUP_BRIGHT_PCT) | |
| sar_thresh = np.percentile(sar, _SAR_HIGH_BACKSCATTER_PCT) | |
| opt_mask = optical >= opt_thresh # bright in optical | |
| sar_mask = sar >= sar_thresh # high backscatter in SAR | |
| label = "built-up / bare surface" | |
| agreement = opt_mask & sar_mask | |
| optical_only = opt_mask & ~sar_mask | |
| sar_only = sar_mask & ~opt_mask | |
| total = optical.size | |
| agreement_pct = float(agreement.sum() / total * 100) | |
| optical_only_pct = float(optical_only.sum() / total * 100) | |
| sar_only_pct = float(sar_only.sum() / total * 100) | |
| if agreement_pct >= 3.0: | |
| level: Literal["high", "medium", "low"] = "high" | |
| elif agreement_pct >= 0.5: | |
| level = "medium" | |
| else: | |
| level = "low" | |
| stats = FusionRegionStats( | |
| label=label, | |
| optical_only_pct=round(optical_only_pct, 2), | |
| sar_only_pct=round(sar_only_pct, 2), | |
| agreement_pct=round(agreement_pct, 2), | |
| confidence_level=level, | |
| ) | |
| answer = ( | |
| f"Both optical and SAR evidence agree on {label} in {stats.agreement_pct}% of the " | |
| f"scene ({level.upper()} confidence). Optical alone flagged an additional " | |
| f"{stats.optical_only_pct}% (possible optical-only signal — e.g. shadow or cloud edge " | |
| f"for water, bare soil for built-up); SAR alone flagged {stats.sar_only_pct}% (possible " | |
| f"SAR-only signal — e.g. surface roughness unrelated to the target, or optical cloud " | |
| f"obstruction hiding a real feature)." | |
| ) | |
| if was_resized: | |
| answer += ( | |
| " NOTE: optical and SAR images had different pixel dimensions and were resized to " | |
| "match — this is NOT true geometric co-registration; treat spatial agreement as " | |
| "approximate, not pixel-exact." | |
| ) | |
| return OpticalSarResult( | |
| target=target, | |
| stats=stats, | |
| answer=answer, | |
| width=int(optical.shape[1]), | |
| height=int(optical.shape[0]), | |
| method="percentile_threshold_agreement (deterministic, not a trained fusion model)", | |
| ) | |