""" Bi-temporal change detection + change-VQA specialist "model". Phase 7 (P0). Satisfies PS requirements: - "Multi-image change analysis: Change description or change-based visual question answering from a bi-temporal image pair shall be mandatory." - "A spatial change map may also be generated where reference masks are available." MODEL CHOICE — deterministic pixel-difference change map, NOT a learned change-detection network. - There is no lightweight, pip-installable, genuinely pretrained bi-temporal RS change-detection model that satisfies "prefer existing pretrained models" AND "avoid unnecessary infrastructure" AND fits a 4-day build. The realistic off-the-shelf options (e.g. open-cd / torchange toolkits) need dataset-specific checkpoints and non-trivial setup — not a hackathon-week integration. - The master brief explicitly sanctions this fallback for exactly this situation (§ Day 3 / Optical+SAR guidance): "You don't need a perfect model. You need working end-to-end output" and "simple fusion/classifier" is an acceptable MVP approach. - What this IS: registered-pixel absolute difference on percentile-stretched grayscale, thresholded and morphologically cleaned into a binary change mask, summarized into a change percentage and largest-region bounding box. This is a classical CV heuristic, not a learned detector — it will react to illumination/seasonal differences as well as real change. State that explicitly in every response; never call this "AI-detected change" without qualification. - Change-VQA path reuses the Phase 5 BLIP-VQA model (see models/vqa.py) by feeding it a single composite image (before | diff | after, side by side) plus the user's question, then combines that free-text answer with the deterministic change stats (percentage, direction) computed above. This is a composite-image VQA heuristic, NOT a purpose-built CDVQA model (the PS names CDVQA as an *evaluation* benchmark for this capability, not a mandated architecture) — disclosed, not hidden. - License/cost: zero extra dependencies beyond numpy/Pillow/rasterio (already in requirements.txt) for the change map; VQA path reuses the already-installed BLIP-VQA weights, so no new model download. """ from __future__ import annotations import threading from dataclasses import dataclass from pathlib import Path from typing import Optional import numpy as np from PIL import Image from preprocessing.raster_to_rgb import load_as_rgb, to_uint8 import rasterio # Prototype policy thresholds — NOT scientifically validated cutoffs. # Percent of pixels flagged "changed" before we call it HIGH/MEDIUM/LOW. _HIGH_CHANGE_PCT = 8.0 _MEDIUM_CHANGE_PCT = 2.0 _DIFF_Z_THRESHOLD = 1.5 # std-devs above mean abs-diff to flag a pixel "changed" @dataclass class ChangeBox: x_min: float y_min: float x_max: float y_max: float @dataclass class ChangeMapResult: changed_pixel_pct: float change_level: str # "high" | "medium" | "low" change_note: str largest_change_box: Optional[ChangeBox] width: int height: int method: str @dataclass class ChangeVQAResult: answer: str change_map: ChangeMapResult vqa_confidence: float vqa_confidence_level: str model_id: str adapted: bool def _load_gray_uint8(path: Path) -> np.ndarray: """Grayscale, percentile-stretched, resized-consistent array for diffing.""" 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) def _resize_to_match(a: np.ndarray, b: np.ndarray) -> tuple[np.ndarray, np.ndarray, bool]: """Bi-temporal pairs aren't guaranteed identical pixel dims. Resize the larger to the smaller (nearest-neighbor via PIL) — a prototype co-registration stand-in, NOT real geometric co-registration. If the pair isn't actually co-registered, this diff is not scientifically meaningful; that caveat is surfaced in change_note when sizes differ.""" if a.shape == b.shape: return a, b, False h = min(a.shape[0], b.shape[0]) w = min(a.shape[1], b.shape[1]) a2 = np.array(Image.fromarray(a).resize((w, h))) b2 = np.array(Image.fromarray(b).resize((w, h))) return a2, b2, True def compute_change_map(before_path: Path, after_path: Path) -> ChangeMapResult: before = _load_gray_uint8(before_path) after = _load_gray_uint8(after_path) before, after, was_resized = _resize_to_match(before, after) diff = np.abs(after.astype("float32") - before.astype("float32")) mean, std = float(diff.mean()), float(diff.std() or 1.0) mask = diff > (mean + _DIFF_Z_THRESHOLD * std) changed_pct = float(mask.mean() * 100) if changed_pct >= _HIGH_CHANGE_PCT: level = "high" elif changed_pct >= _MEDIUM_CHANGE_PCT: level = "medium" else: level = "low" box: Optional[ChangeBox] = None if mask.any(): ys, xs = np.where(mask) box = ChangeBox( x_min=float(xs.min()), y_min=float(ys.min()), x_max=float(xs.max()), y_max=float(ys.max()), ) note = ( f"{changed_pct:.1f}% of pixels flagged as changed by a statistical " f"threshold (>{_DIFF_Z_THRESHOLD} std-dev above mean intensity " "difference). This is a classical pixel-difference heuristic, not " "a learned change-detection model — illumination, seasonal, or " "sensor differences between acquisitions will also register as " "\"change\". Treat the box as the largest contiguous changed region, " "not a validated object boundary." ) if was_resized: note += ( " NOTE: before/after images had different pixel dimensions and " "were resized to match for this comparison — this is NOT real " "geometric co-registration; treat spatial correspondence as " "approximate." ) return ChangeMapResult( changed_pixel_pct=round(changed_pct, 2), change_level=level, change_note=note, largest_change_box=box, width=int(before.shape[1]), height=int(before.shape[0]), method="pixel_diff_zscore_threshold (deterministic, not a trained model)", ) def describe_change(change: ChangeMapResult, query: str) -> str: """Rule-based change description — no model call. Satisfies the "change description" half of the mandatory requirement standalone, without needing the VQA composite path.""" q = query.lower() directional = any(k in q for k in ("increase", "grew", "expand")) if change.change_level == "low": return ( f"Minimal change detected ({change.changed_pixel_pct}% of pixels) " "between the two dates — the scene appears largely stable." ) region = "a concentrated region" if change.largest_change_box else "no single dominant region" base = ( f"{change.change_level.upper()} change detected: {change.changed_pixel_pct}% " f"of pixels differ between the before and after images, concentrated in {region}." ) if directional: base += ( " Note: this heuristic detects that pixels changed, not the semantic " "direction (e.g. built-up increased vs. decreased) — that requires the " "change-VQA path or a trained classifier, not this diff map alone." ) return base class _ChangeVQABridge: """Builds a single composite (before | after) image and routes it through the existing Phase 5 BLIP-VQA model. Lazy singleton mirrors the pattern in models/vqa.py — no separate model weights to load.""" def __init__(self) -> None: self._lock = threading.Lock() def answer(self, before_path: Path, after_path: Path, question: str) -> tuple[str, float, str]: from models.vqa import get_vqa_model # deferred import, reuse Phase 5 singleton before_img = load_as_rgb(before_path) after_img = load_as_rgb(after_path) h = min(before_img.height, after_img.height) before_r = before_img.resize((int(before_img.width * h / before_img.height), h)) after_r = after_img.resize((int(after_img.width * h / after_img.height), h)) composite = Image.new("RGB", (before_r.width + after_r.width, h)) composite.paste(before_r, (0, 0)) composite.paste(after_r, (before_r.width, 0)) # Reuse VQA's own image loader contract by writing the composite to # a temp path — the singleton's .answer() takes a path. import tempfile with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: composite.save(tmp.name) tmp_path = Path(tmp.name) try: phrased_question = f"Comparing the left (before) and right (after) images: {question}" result = get_vqa_model().answer(tmp_path, phrased_question) finally: tmp_path.unlink(missing_ok=True) return result.answer, result.confidence, result.model_id _bridge: Optional[_ChangeVQABridge] = None _bridge_lock = threading.Lock() def _get_bridge() -> _ChangeVQABridge: global _bridge if _bridge is None: with _bridge_lock: if _bridge is None: _bridge = _ChangeVQABridge() return _bridge def run_change_detection(before_path: Path, after_path: Path, query: str) -> dict: change = compute_change_map(before_path, after_path) description = describe_change(change, query) return { "mode": "change_detection", "description": description, "change_map": change.__dict__ | { "largest_change_box": change.largest_change_box.__dict__ if change.largest_change_box else None }, } def run_change_vqa(before_path: Path, after_path: Path, query: str) -> ChangeVQAResult: change = compute_change_map(before_path, after_path) answer, confidence, model_id = _get_bridge().answer(before_path, after_path, query) if confidence >= 0.75: level = "high" elif confidence >= 0.40: level = "medium" else: level = "insufficient" return ChangeVQAResult( answer=answer, change_map=change, vqa_confidence=round(confidence, 4), vqa_confidence_level=level, model_id=f"{model_id} (composite before|after image — no dedicated CDVQA model)", adapted=False, )