Improve detection accuracy: TTA for AdaptFormer, hysteresis thresholding, robust logits and LBP borders.

app/detection_engine.py

@@ -506,18 +506,48 @@ def compute_ssim_change_map(img1, img2, win_size=11):
 # ---------------------------------------------------------------------------
 
 def compute_lbp(gray, radius=1, n_points=8):
-    """Compute simplified Local Binary Pattern texture descriptor."""
-    h, w = gray.shape
+    """Compute simplified Local Binary Pattern texture descriptor.
+
+    Uses reflect-padded shifts (not np.roll) so opposite image borders do not
+    wrap into each other and create spurious texture-change along the frame.
+    """
+    pad = max(1, int(radius))
+    padded = cv2.copyMakeBorder(gray, pad, pad, pad, pad, cv2.BORDER_REFLECT)
     lbp = np.zeros_like(gray, dtype=np.float32)
+    h, w = gray.shape
     for i in range(n_points):
         angle = 2 * np.pi * i / n_points
         dx = int(round(radius * np.cos(angle)))
         dy = int(round(-radius * np.sin(angle)))
-        shifted = np.roll(np.roll(gray, dy, axis=0), dx, axis=1)
+        shifted = padded[pad + dy:pad + dy + h, pad + dx:pad + dx + w]
         lbp += (shifted >= gray).astype(np.float32)
     return lbp / n_points
 
 
+def _hysteresis_threshold(score, high_thr, low_thr):
+    """Two-level (hysteresis) thresholding on a [0,1] score map.
+
+    Keeps every low-threshold connected component that contains at least one
+    high-threshold "seed" pixel, and drops the rest. This recovers complete
+    change blobs (less fragmentation) while removing isolated weak speckle that
+    a single hard threshold would either cut or admit.
+    """
+    score = score.astype(np.float32)
+    high = (score >= high_thr).astype(np.uint8)
+    if int(high.sum()) == 0:
+        return (high * 255).astype(np.uint8)
+    low = (score >= min(low_thr, high_thr)).astype(np.uint8)
+    num, labels = cv2.connectedComponents(low, connectivity=8)
+    if num <= 1:
+        return (high * 255).astype(np.uint8)
+    seed_labels = np.unique(labels[high > 0])
+    seed_labels = seed_labels[seed_labels != 0]
+    if seed_labels.size == 0:
+        return (high * 255).astype(np.uint8)
+    keep = np.isin(labels, seed_labels)
+    return np.where(keep, 255, 0).astype(np.uint8)
+
+
 def compute_texture_change(img1, img2):
     """Compute texture difference using LBP."""
     gray1 = cv2.cvtColor(img1, cv2.COLOR_RGB2GRAY).astype(np.float32)
@@ -768,7 +798,9 @@ def fuse_dl_and_classical(dl_score, classical_score, img1, img2, sensitivity=0.5
     final_score = np.where(veg_boost, np.maximum(final_score, classical_score), final_score)
 
     fused_thr = 0.45 + (1.0 - sens) * 0.15
-    change_mask = (final_score >= fused_thr).astype(np.uint8) * 255
+    # Hysteresis grow keeps complete blobs while pruning isolated weak responses
+    fused_low = max(0.25, fused_thr - 0.12)
+    change_mask = _hysteresis_threshold(final_score, fused_thr, fused_low)
     change_mask = _clean_mask(change_mask, sensitivity=sens)
 
     debug = {
@@ -777,6 +809,7 @@ def fuse_dl_and_classical(dl_score, classical_score, img1, img2, sensitivity=0.5
         "T_cl_percentile_q": q,
         "T_cl_score": T_cl,
         "fused_threshold": fused_thr,
+        "fused_low_threshold": fused_low,
         "dl_changed_px": int(np.sum(dl_score >= med_dl)),
         "classical_changed_px": int(np.sum(classical_score >= T_cl)),
         "fused_changed_px": int(np.sum(change_mask > 127)),
@@ -793,7 +826,10 @@ def _ai_fusion_core(img1, img2, sensitivity=0.5, registration_ok=True):
     # Looser percentile than gated fusion — keeps recall for multi-region detection
     q = float(np.clip(0.93 - (sens - 0.5) * 0.06, 0.85, 0.96))
     thr_score = float(np.quantile(classical_score, q))
-    change_mask = (classical_score >= thr_score).astype(np.uint8) * 255
+    # Hysteresis: grow seeds down to a lower percentile to recover full change blobs
+    q_low = float(np.clip(q - 0.06, 0.78, q))
+    low_score = float(np.quantile(classical_score, q_low))
+    change_mask = _hysteresis_threshold(classical_score, thr_score, low_score)
     change_mask = _clean_mask(change_mask, sensitivity=sens)
 
     debug = {
@@ -801,6 +837,8 @@ def _ai_fusion_core(img1, img2, sensitivity=0.5, registration_ok=True):
         "threshold_used": int(thr_score * 255),
         "threshold_percentile_q": q,
         "threshold_score": thr_score,
+        "hysteresis_low_q": q_low,
+        "hysteresis_low_score": low_score,
         "sensitivity": float(sensitivity),
         "channel_weights": {
             "color": round(weights[0], 4),

app/model_inference.py

@@ -112,23 +112,68 @@ def get_model_status() -> dict:
         "available": available,
         "detectionMode": mode,
         "device": str(_DEVICE) if _DEVICE is not None else None,
+        "tta": [op or "identity" for op in _resolve_tta_ops()],
         "error": _LOAD_ERROR,
     }
 
 
-def predict_change_mask(img1, img2, threshold=0.5):
-    """
-    Run AdaptFormer inference on two RGB numpy arrays (H, W, 3).
-    Returns (uint8 mask [0 or 255], float32 score map [0-1]).
-    Use threshold > 1.0 to obtain score map only (empty mask).
+def _logits_to_change_prob(logits, torch):
+    """Robustly convert model logits to a single-channel change probability tile.
+
+    Handles 1-channel (sigmoid), 2+-channel (softmax, change=last channel) and
+    already-2D outputs so a change in the upstream model head does not silently
+    break inference.
     """
+    t = logits
+    if t.dim() == 4:  # (N, C, H, W)
+        c = t.shape[1]
+        if c == 1:
+            return torch.sigmoid(t)[0, 0]
+        return torch.softmax(t, dim=1)[0, -1]
+    if t.dim() == 3:  # (C, H, W) or (N, H, W)
+        if t.shape[0] == 1:
+            return torch.sigmoid(t)[0]
+        # ambiguous: treat as (N,H,W) single map
+        return torch.sigmoid(t[0])
+    return torch.sigmoid(t)
+
+
+def _resolve_tta_ops():
+    """Choose test-time augmentation flips. Env DETECTION_TTA: off|0 | hflip | full | auto."""
+    mode = os.environ.get("DETECTION_TTA", "auto").strip().lower()
+    if mode in ("0", "off", "none", "false"):
+        return [None]
+    if mode in ("h", "hflip"):
+        return [None, "h"]
+    if mode in ("full", "all"):
+        return [None, "h", "v"]
+    # auto: lighter on CPU (hflip only), fuller on GPU
+    on_cuda = _DEVICE is not None and str(_DEVICE).startswith("cuda")
+    return [None, "h", "v"] if on_cuda else [None, "h"]
+
+
+def _apply_flip(arr, op):
+    if op == "h":
+        return arr[:, ::-1]
+    if op == "v":
+        return arr[::-1, :]
+    return arr
+
+
+def _unflip_map(score, op):
+    if op == "h":
+        return score[:, ::-1]
+    if op == "v":
+        return score[::-1, :]
+    return score
+
+
+def _infer_score_map(img1, img2):
+    """Single-pass tiled inference returning a float32 change-probability map at (h, w)."""
     torch, _, _ = _try_import()
     model, processor = _load_model()
     from PIL import Image as PILImage
 
-    if img1.shape != img2.shape:
-        img2 = cv2.resize(img2, (img1.shape[1], img1.shape[0]))
-
     h, w = img1.shape[:2]
     tile = _TILE_SIZE
     overlap = tile // 4
@@ -152,8 +197,8 @@ def predict_change_mask(img1, img2, threshold=0.5):
     with torch.no_grad():
         for y0 in range(0, ph - tile + 1, stride):
             for x0 in range(0, pw - tile + 1, stride):
-                t1 = img1[y0:y0+tile, x0:x0+tile]
-                t2 = img2[y0:y0+tile, x0:x0+tile]
+                t1 = np.ascontiguousarray(img1[y0:y0+tile, x0:x0+tile])
+                t2 = np.ascontiguousarray(img2[y0:y0+tile, x0:x0+tile])
 
                 pil1 = PILImage.fromarray(t1)
                 pil2 = PILImage.fromarray(t2)
@@ -163,9 +208,7 @@ def predict_change_mask(img1, img2, threshold=0.5):
 
                 outputs = model(**inputs)
                 logits = outputs.logits
-                probs = torch.softmax(logits, dim=1)
-
-                prob_map = probs[0, 1].cpu().numpy()
+                prob_map = _logits_to_change_prob(logits, torch).cpu().numpy()
 
                 out_h, out_w = prob_map.shape
                 if out_h != tile or out_w != tile:
@@ -177,7 +220,41 @@ def predict_change_mask(img1, img2, threshold=0.5):
 
     count = np.maximum(count, 1e-6)
     avg_score = score_sum / count
-    avg_score = avg_score[:h, :w]
+    return avg_score[:h, :w]
+
+
+def predict_change_mask(img1, img2, threshold=0.5):
+    """
+    Run AdaptFormer inference on two RGB numpy arrays (H, W, 3).
+    Averages predictions over test-time augmentation flips (DETECTION_TTA) for
+    higher-accuracy, less boundary-sensitive change maps.
+    Returns (uint8 mask [0 or 255], float32 score map [0-1]).
+    Use threshold > 1.0 to obtain score map only (empty mask).
+    """
+    _load_model()
+
+    if img1.shape != img2.shape:
+        img2 = cv2.resize(img2, (img1.shape[1], img1.shape[0]))
+
+    h, w = img1.shape[:2]
+    ops = _resolve_tta_ops()
+
+    acc = np.zeros((h, w), dtype=np.float32)
+    n = 0
+    for op in ops:
+        a1 = np.ascontiguousarray(_apply_flip(img1, op))
+        a2 = np.ascontiguousarray(_apply_flip(img2, op))
+        try:
+            s = _infer_score_map(a1, a2)
+        except Exception as exc:
+            logger.warning("TTA pass %s failed: %s", op, exc)
+            continue
+        acc += _unflip_map(s, op)
+        n += 1
+
+    if n == 0:
+        raise RuntimeError("AdaptFormer inference produced no predictions")
+    avg_score = acc / float(n)
 
     mask = (avg_score >= threshold).astype(np.uint8) * 255
     return mask, avg_score