Add Siamese U-Net training notebook and model inference integration

app/detection_engine.py

@@ -584,14 +584,27 @@ def _ai_fusion_core(img1, img2, sensitivity=0.5):
 
 def ai_deep_learning_method(img1, img2, sensitivity=0.5):
     """
-    Single-pass AI fusion with CVA, SNR weighting, and vegetation/shadow
-    suppression. The suppression maps make the reverse pass unnecessary,
-    halving computation and eliminating OR-induced false positives.
+    Uses the trained Siamese U-Net when available; falls back to the
+    rule-based multi-channel fusion otherwise.
     """
-    change_mask, core_debug = _ai_fusion_core(img1, img2, sensitivity=sensitivity)
+    from .model_inference import is_model_available, predict_change_mask
+
+    if is_model_available():
+        threshold = 0.35 + (1.0 - sensitivity) * 0.3
+        change_mask, score_map = predict_change_mask(img1, img2, threshold=threshold)
+        change_mask = _clean_mask(change_mask, sensitivity=sensitivity)
+        debug = {
+            "method": "AI-Based Deep Learning (Siamese U-Net)",
+            "model": "siamese_unet",
+            "threshold_used": int(threshold * 255),
+            "sensitivity": float(sensitivity),
+        }
+        return change_mask, debug
 
+    # Fallback: rule-based fusion
+    change_mask, core_debug = _ai_fusion_core(img1, img2, sensitivity=sensitivity)
     debug = {
-        "method": "AI-Based Deep Learning",
+        "method": "AI-Based Deep Learning (rule-based fallback)",
         "threshold_used": core_debug.get("threshold_used"),
         "sensitivity": float(sensitivity),
         "core": core_debug,

app/model_inference.py

@@ -0,0 +1,114 @@
+"""
+Siamese U-Net inference for satellite change detection.
+
+Loads a TorchScript model exported from the training notebook and runs
+tile-based inference on arbitrary-size image pairs, producing a binary
+change mask compatible with the rest of the detection pipeline.
+
+Set CHANGE_MODEL_PATH env var to the .pt file location.
+Falls back to the rule-based AI fusion when no model is available.
+"""
+import logging
+import os
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+_MODEL = None
+_MODEL_PATH = os.environ.get("CHANGE_MODEL_PATH", "data/siamese_unet.pt")
+_TILE_SIZE = 256
+_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
+_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
+
+
+def _get_torch():
+    """Lazy import torch — only when model exists."""
+    try:
+        import torch
+        return torch
+    except ImportError:
+        return None
+
+
+def is_model_available():
+    """Check if a trained model file exists and torch is installed."""
+    return Path(_MODEL_PATH).is_file() and _get_torch() is not None
+
+
+def _load_model():
+    global _MODEL
+    if _MODEL is not None:
+        return _MODEL
+    torch = _get_torch()
+    if torch is None:
+        raise RuntimeError("PyTorch is not installed")
+    path = Path(_MODEL_PATH)
+    if not path.is_file():
+        raise FileNotFoundError(f"Model not found at {path}")
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    _MODEL = torch.jit.load(str(path), map_location=device)
+    _MODEL.eval()
+    logger.info("Loaded Siamese U-Net from %s on %s", path, device)
+    return _MODEL
+
+
+def _preprocess_tile(tile):
+    """Normalize a (H, W, 3) uint8 RGB tile to (1, 3, H, W) float tensor."""
+    torch = _get_torch()
+    img = tile.astype(np.float32) / 255.0
+    img = (img - _MEAN) / _STD
+    tensor = torch.from_numpy(img.transpose(2, 0, 1)).unsqueeze(0)
+    return tensor
+
+
+def predict_change_mask(img1, img2, threshold=0.5):
+    """
+    Run Siamese U-Net inference on two RGB numpy arrays (H, W, 3).
+    Images are split into overlapping tiles, predicted individually,
+    and stitched back into a full-resolution binary mask.
+
+    Returns a uint8 mask (0 or 255) at the input resolution.
+    """
+    torch = _get_torch()
+    model = _load_model()
+    device = next(model.parameters()).device
+
+    if img1.shape != img2.shape:
+        img2 = cv2.resize(img2, (img1.shape[1], img1.shape[0]))
+
+    h, w = img1.shape[:2]
+    tile = _TILE_SIZE
+    stride = tile * 3 // 4  # 75% overlap for smoother stitching
+
+    # Pad to make dimensions divisible by tile size
+    pad_h = (tile - h % tile) % tile
+    pad_w = (tile - w % tile) % tile
+    if pad_h or pad_w:
+        img1 = np.pad(img1, ((0, pad_h), (0, pad_w), (0, 0)), mode="reflect")
+        img2 = np.pad(img2, ((0, pad_h), (0, pad_w), (0, 0)), mode="reflect")
+
+    ph, pw = img1.shape[:2]
+    score_sum = np.zeros((ph, pw), dtype=np.float32)
+    count = np.zeros((ph, pw), dtype=np.float32)
+
+    with torch.no_grad():
+        for y0 in range(0, ph - tile + 1, stride):
+            for x0 in range(0, pw - tile + 1, stride):
+                t1 = _preprocess_tile(img1[y0:y0+tile, x0:x0+tile])
+                t2 = _preprocess_tile(img2[y0:y0+tile, x0:x0+tile])
+                logits = model(t1.to(device), t2.to(device))
+                prob = torch.sigmoid(logits).squeeze().cpu().numpy()
+                score_sum[y0:y0+tile, x0:x0+tile] += prob
+                count[y0:y0+tile, x0:x0+tile] += 1.0
+
+    count = np.maximum(count, 1.0)
+    avg_score = score_sum / count
+
+    # Crop back to original size
+    avg_score = avg_score[:h, :w]
+
+    mask = (avg_score >= threshold).astype(np.uint8) * 255
+    return mask, avg_score

train_change_detection_model.ipynb

@@ -0,0 +1,586 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Satellite Change Detection — Siamese U-Net Training\n",
+    "\n",
+    "This notebook trains a **Siamese U-Net** on the **LEVIR-CD** dataset for pixel-level\n",
+    "satellite image change detection. The exported model plugs directly into the\n",
+    "AI Change Detection web app.\n",
+    "\n",
+    "**Run in Google Colab** with a GPU runtime (Runtime → Change runtime type → T4 GPU).\n",
+    "\n",
+    "Training takes ~2-3 hours on a free T4."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 1. Install Dependencies"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "!pip install -q torch torchvision segmentation-models-pytorch albumentations gdown tqdm matplotlib"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 2. Download LEVIR-CD Dataset\n",
+    "\n",
+    "LEVIR-CD contains 637 pairs of 1024×1024 Google Earth images with pixel-level\n",
+    "building change annotations. We download the pre-cut 256×256 patch version."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import os, zipfile, gdown\n",
+    "\n",
+    "DATA_ROOT = \"./levir_cd_256\"\n",
+    "\n",
+    "if not os.path.isdir(DATA_ROOT):\n",
+    "    # LEVIR-CD 256x256 patches (hosted mirror)\n",
+    "    url = \"https://drive.google.com/uc?id=1RUFY5Z4Bf5LAoYXC8Iq9k5GloJmVDmfF\"\n",
+    "    zip_path = \"levir_cd_256.zip\"\n",
+    "    print(\"Downloading LEVIR-CD 256×256 patches (~540 MB)...\")\n",
+    "    gdown.download(url, zip_path, quiet=False)\n",
+    "    print(\"Extracting...\")\n",
+    "    with zipfile.ZipFile(zip_path, \"r\") as z:\n",
+    "        z.extractall(\".\")\n",
+    "    os.remove(zip_path)\n",
+    "    print(\"Done. Dataset at:\", DATA_ROOT)\n",
+    "else:\n",
+    "    print(\"Dataset already present at\", DATA_ROOT)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Verify structure — adjust paths if your zip extracts differently\n",
+    "for split in [\"train\", \"val\", \"test\"]:\n",
+    "    for sub in [\"A\", \"B\", \"label\"]:\n",
+    "        p = os.path.join(DATA_ROOT, split, sub)\n",
+    "        if os.path.isdir(p):\n",
+    "            n = len(os.listdir(p))\n",
+    "            print(f\"{split}/{sub}: {n} files\")\n",
+    "        else:\n",
+    "            print(f\"WARNING: {p} not found — check extracted folder name\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 3. Dataset & DataLoader"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import numpy as np\n",
+    "from PIL import Image\n",
+    "from torch.utils.data import Dataset, DataLoader\n",
+    "import albumentations as A\n",
+    "from albumentations.pytorch import ToTensorV2\n",
+    "\n",
+    "\n",
+    "class LEVIRCDDataset(Dataset):\n",
+    "    \"\"\"LEVIR-CD dataset: before (A), after (B), binary label.\"\"\"\n",
+    "\n",
+    "    def __init__(self, root, split=\"train\", transform=None):\n",
+    "        self.dir_a = os.path.join(root, split, \"A\")\n",
+    "        self.dir_b = os.path.join(root, split, \"B\")\n",
+    "        self.dir_label = os.path.join(root, split, \"label\")\n",
+    "        self.fnames = sorted(os.listdir(self.dir_a))\n",
+    "        self.transform = transform\n",
+    "\n",
+    "    def __len__(self):\n",
+    "        return len(self.fnames)\n",
+    "\n",
+    "    def __getitem__(self, idx):\n",
+    "        name = self.fnames[idx]\n",
+    "        img_a = np.array(Image.open(os.path.join(self.dir_a, name)).convert(\"RGB\"))\n",
+    "        img_b = np.array(Image.open(os.path.join(self.dir_b, name)).convert(\"RGB\"))\n",
+    "        label = np.array(Image.open(os.path.join(self.dir_label, name)).convert(\"L\"))\n",
+    "        label = (label > 127).astype(np.float32)\n",
+    "\n",
+    "        if self.transform:\n",
+    "            # Apply same spatial transform to all three\n",
+    "            aug = self.transform(\n",
+    "                image=img_a,\n",
+    "                image_b=img_b,\n",
+    "                mask=label,\n",
+    "            )\n",
+    "            img_a = aug[\"image\"]        # (3, H, W) tensor\n",
+    "            img_b = aug[\"image_b\"]      # (3, H, W) tensor\n",
+    "            label = aug[\"mask\"].unsqueeze(0)  # (1, H, W)\n",
+    "        return img_a, img_b, label\n",
+    "\n",
+    "\n",
+    "train_transform = A.Compose(\n",
+    "    [\n",
+    "        A.HorizontalFlip(p=0.5),\n",
+    "        A.VerticalFlip(p=0.5),\n",
+    "        A.RandomRotate90(p=0.5),\n",
+    "        A.RandomBrightnessContrast(p=0.3, brightness_limit=0.15, contrast_limit=0.15),\n",
+    "        A.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)),\n",
+    "        ToTensorV2(),\n",
+    "    ],\n",
+    "    additional_targets={\"image_b\": \"image\"},\n",
+    ")\n",
+    "\n",
+    "val_transform = A.Compose(\n",
+    "    [\n",
+    "        A.Normalize(mean=(0.485, 0.456, 0.406), std=(0.229, 0.224, 0.225)),\n",
+    "        ToTensorV2(),\n",
+    "    ],\n",
+    "    additional_targets={\"image_b\": \"image\"},\n",
+    ")\n",
+    "\n",
+    "train_ds = LEVIRCDDataset(DATA_ROOT, \"train\", train_transform)\n",
+    "val_ds = LEVIRCDDataset(DATA_ROOT, \"val\", val_transform)\n",
+    "test_ds = LEVIRCDDataset(DATA_ROOT, \"test\", val_transform)\n",
+    "\n",
+    "BATCH = 8\n",
+    "train_dl = DataLoader(train_ds, batch_size=BATCH, shuffle=True, num_workers=2, pin_memory=True)\n",
+    "val_dl = DataLoader(val_ds, batch_size=BATCH, shuffle=False, num_workers=2, pin_memory=True)\n",
+    "test_dl = DataLoader(test_ds, batch_size=BATCH, shuffle=False, num_workers=2, pin_memory=True)\n",
+    "\n",
+    "print(f\"Train: {len(train_ds)}, Val: {len(val_ds)}, Test: {len(test_ds)}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 4. Siamese U-Net Model\n",
+    "\n",
+    "Architecture:\n",
+    "- **Shared encoder** (ResNet34, ImageNet pretrained) processes both images\n",
+    "- Feature maps from both branches are **concatenated** at each decoder level\n",
+    "- Standard U-Net decoder produces a binary change mask"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import torch\n",
+    "import torch.nn as nn\n",
+    "import segmentation_models_pytorch as smp\n",
+    "\n",
+    "\n",
+    "class SiameseUNet(nn.Module):\n",
+    "    \"\"\"\n",
+    "    Siamese U-Net for change detection.\n",
+    "    Shared encoder extracts features from both images;\n",
+    "    concatenated features are decoded into a binary change mask.\n",
+    "    \"\"\"\n",
+    "\n",
+    "    def __init__(self, encoder_name=\"resnet34\", pretrained=True):\n",
+    "        super().__init__()\n",
+    "        # Build a standard U-Net to reuse its encoder and decoder pieces\n",
+    "        aux = smp.Unet(\n",
+    "            encoder_name=encoder_name,\n",
+    "            encoder_weights=\"imagenet\" if pretrained else None,\n",
+    "            in_channels=3,\n",
+    "            classes=1,\n",
+    "        )\n",
+    "        self.encoder = aux.encoder\n",
+    "\n",
+    "        # The decoder expects concatenated features (2x channels at each level)\n",
+    "        encoder_channels = self.encoder.out_channels  # e.g. (3,64,64,128,256,512)\n",
+    "        doubled = tuple(c * 2 for c in encoder_channels)\n",
+    "\n",
+    "        self.decoder = smp.decoders.unet.decoder.UnetDecoder(\n",
+    "            encoder_channels=doubled,\n",
+    "            decoder_channels=(256, 128, 64, 32, 16),\n",
+    "            n_blocks=5,\n",
+    "            use_batchnorm=True,\n",
+    "            attention_type=None,\n",
+    "        )\n",
+    "\n",
+    "        self.head = nn.Conv2d(16, 1, kernel_size=1)\n",
+    "\n",
+    "    def forward(self, img_a, img_b):\n",
+    "        # Shared encoder for both temporal images\n",
+    "        feats_a = self.encoder(img_a)\n",
+    "        feats_b = self.encoder(img_b)\n",
+    "\n",
+    "        # Concatenate features at every level\n",
+    "        feats_cat = [torch.cat([fa, fb], dim=1) for fa, fb in zip(feats_a, feats_b)]\n",
+    "\n",
+    "        decoded = self.decoder(*feats_cat)\n",
+    "        logits = self.head(decoded)\n",
+    "        return logits\n",
+    "\n",
+    "\n",
+    "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n",
+    "model = SiameseUNet(encoder_name=\"resnet34\", pretrained=True).to(device)\n",
+    "\n",
+    "total_params = sum(p.numel() for p in model.parameters()) / 1e6\n",
+    "print(f\"Model on {device}, {total_params:.1f}M parameters\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 5. Loss Function & Metrics\n",
+    "\n",
+    "Combined **BCE + Dice** loss handles class imbalance (most pixels are unchanged)."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "class BCEDiceLoss(nn.Module):\n",
+    "    def __init__(self, bce_weight=0.5):\n",
+    "        super().__init__()\n",
+    "        self.bce = nn.BCEWithLogitsLoss()\n",
+    "        self.bce_weight = bce_weight\n",
+    "\n",
+    "    def forward(self, logits, targets):\n",
+    "        bce_loss = self.bce(logits, targets)\n",
+    "        probs = torch.sigmoid(logits)\n",
+    "        smooth = 1.0\n",
+    "        intersection = (probs * targets).sum()\n",
+    "        dice = (2.0 * intersection + smooth) / (probs.sum() + targets.sum() + smooth)\n",
+    "        dice_loss = 1.0 - dice\n",
+    "        return self.bce_weight * bce_loss + (1 - self.bce_weight) * dice_loss\n",
+    "\n",
+    "\n",
+    "def compute_metrics(preds, targets, threshold=0.5):\n",
+    "    \"\"\"Compute precision, recall, F1, and IoU.\"\"\"\n",
+    "    preds_bin = (preds > threshold).float()\n",
+    "    tp = (preds_bin * targets).sum().item()\n",
+    "    fp = (preds_bin * (1 - targets)).sum().item()\n",
+    "    fn = ((1 - preds_bin) * targets).sum().item()\n",
+    "    precision = tp / (tp + fp + 1e-8)\n",
+    "    recall = tp / (tp + fn + 1e-8)\n",
+    "    f1 = 2 * precision * recall / (precision + recall + 1e-8)\n",
+    "    iou = tp / (tp + fp + fn + 1e-8)\n",
+    "    return {\"precision\": precision, \"recall\": recall, \"f1\": f1, \"iou\": iou}"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 6. Training Loop"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from tqdm.auto import tqdm\n",
+    "\n",
+    "NUM_EPOCHS = 50\n",
+    "LR = 1e-4\n",
+    "\n",
+    "criterion = BCEDiceLoss(bce_weight=0.5)\n",
+    "optimizer = torch.optim.Adam(model.parameters(), lr=LR, weight_decay=1e-4)\n",
+    "scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=NUM_EPOCHS, eta_min=1e-6)\n",
+    "\n",
+    "best_f1 = 0.0\n",
+    "history = {\"train_loss\": [], \"val_loss\": [], \"val_f1\": [], \"val_iou\": []}\n",
+    "\n",
+    "for epoch in range(1, NUM_EPOCHS + 1):\n",
+    "    # --- Train ---\n",
+    "    model.train()\n",
+    "    running_loss = 0.0\n",
+    "    for img_a, img_b, label in tqdm(train_dl, desc=f\"Epoch {epoch}/{NUM_EPOCHS} [train]\", leave=False):\n",
+    "        img_a = img_a.to(device)\n",
+    "        img_b = img_b.to(device)\n",
+    "        label = label.to(device)\n",
+    "\n",
+    "        logits = model(img_a, img_b)\n",
+    "        loss = criterion(logits, label)\n",
+    "\n",
+    "        optimizer.zero_grad()\n",
+    "        loss.backward()\n",
+    "        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)\n",
+    "        optimizer.step()\n",
+    "        running_loss += loss.item() * img_a.size(0)\n",
+    "\n",
+    "    train_loss = running_loss / len(train_ds)\n",
+    "    scheduler.step()\n",
+    "\n",
+    "    # --- Validate ---\n",
+    "    model.eval()\n",
+    "    val_loss_sum = 0.0\n",
+    "    all_preds, all_targets = [], []\n",
+    "    with torch.no_grad():\n",
+    "        for img_a, img_b, label in val_dl:\n",
+    "            img_a = img_a.to(device)\n",
+    "            img_b = img_b.to(device)\n",
+    "            label = label.to(device)\n",
+    "\n",
+    "            logits = model(img_a, img_b)\n",
+    "            val_loss_sum += criterion(logits, label).item() * img_a.size(0)\n",
+    "            all_preds.append(torch.sigmoid(logits).cpu())\n",
+    "            all_targets.append(label.cpu())\n",
+    "\n",
+    "    val_loss = val_loss_sum / len(val_ds)\n",
+    "    preds_cat = torch.cat(all_preds)\n",
+    "    targets_cat = torch.cat(all_targets)\n",
+    "    metrics = compute_metrics(preds_cat, targets_cat)\n",
+    "\n",
+    "    history[\"train_loss\"].append(train_loss)\n",
+    "    history[\"val_loss\"].append(val_loss)\n",
+    "    history[\"val_f1\"].append(metrics[\"f1\"])\n",
+    "    history[\"val_iou\"].append(metrics[\"iou\"])\n",
+    "\n",
+    "    print(\n",
+    "        f\"Epoch {epoch:02d} | \"\n",
+    "        f\"train_loss={train_loss:.4f} | \"\n",
+    "        f\"val_loss={val_loss:.4f} | \"\n",
+    "        f\"F1={metrics['f1']:.4f} | \"\n",
+    "        f\"IoU={metrics['iou']:.4f} | \"\n",
+    "        f\"P={metrics['precision']:.4f} R={metrics['recall']:.4f}\"\n",
+    "    )\n",
+    "\n",
+    "    if metrics[\"f1\"] > best_f1:\n",
+    "        best_f1 = metrics[\"f1\"]\n",
+    "        torch.save(model.state_dict(), \"best_siamese_unet.pth\")\n",
+    "        print(f\"  >> Saved best model (F1={best_f1:.4f})\")\n",
+    "\n",
+    "print(f\"\\nTraining complete. Best val F1: {best_f1:.4f}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 7. Training Curves"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import matplotlib.pyplot as plt\n",
+    "\n",
+    "fig, axes = plt.subplots(1, 3, figsize=(15, 4))\n",
+    "\n",
+    "axes[0].plot(history[\"train_loss\"], label=\"Train\")\n",
+    "axes[0].plot(history[\"val_loss\"], label=\"Val\")\n",
+    "axes[0].set_title(\"Loss\")\n",
+    "axes[0].legend()\n",
+    "\n",
+    "axes[1].plot(history[\"val_f1\"])\n",
+    "axes[1].set_title(\"Val F1 Score\")\n",
+    "\n",
+    "axes[2].plot(history[\"val_iou\"])\n",
+    "axes[2].set_title(\"Val IoU\")\n",
+    "\n",
+    "for ax in axes:\n",
+    "    ax.set_xlabel(\"Epoch\")\n",
+    "    ax.grid(True, alpha=0.3)\n",
+    "\n",
+    "plt.tight_layout()\n",
+    "plt.show()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 8. Evaluate on Test Set"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Load best checkpoint\n",
+    "model.load_state_dict(torch.load(\"best_siamese_unet.pth\", map_location=device))\n",
+    "model.eval()\n",
+    "\n",
+    "all_preds, all_targets = [], []\n",
+    "with torch.no_grad():\n",
+    "    for img_a, img_b, label in tqdm(test_dl, desc=\"Testing\"):\n",
+    "        logits = model(img_a.to(device), img_b.to(device))\n",
+    "        all_preds.append(torch.sigmoid(logits).cpu())\n",
+    "        all_targets.append(label)\n",
+    "\n",
+    "preds = torch.cat(all_preds)\n",
+    "targets = torch.cat(all_targets)\n",
+    "test_metrics = compute_metrics(preds, targets)\n",
+    "\n",
+    "print(f\"\\nTest Results:\")\n",
+    "print(f\"  F1 Score:  {test_metrics['f1']:.4f}\")\n",
+    "print(f\"  IoU:       {test_metrics['iou']:.4f}\")\n",
+    "print(f\"  Precision: {test_metrics['precision']:.4f}\")\n",
+    "print(f\"  Recall:    {test_metrics['recall']:.4f}\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 9. Visualize Predictions"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "MEAN = np.array([0.485, 0.456, 0.406])\n",
+    "STD = np.array([0.229, 0.224, 0.225])\n",
+    "\n",
+    "def denorm(tensor):\n",
+    "    img = tensor.permute(1, 2, 0).numpy()\n",
+    "    img = img * STD + MEAN\n",
+    "    return np.clip(img, 0, 1)\n",
+    "\n",
+    "fig, axes = plt.subplots(4, 4, figsize=(16, 16))\n",
+    "sample_indices = np.random.choice(len(test_ds), 4, replace=False)\n",
+    "\n",
+    "for row, idx in enumerate(sample_indices):\n",
+    "    img_a, img_b, label = test_ds[idx]\n",
+    "    with torch.no_grad():\n",
+    "        logit = model(img_a.unsqueeze(0).to(device), img_b.unsqueeze(0).to(device))\n",
+    "        pred = (torch.sigmoid(logit) > 0.5).squeeze().cpu().numpy()\n",
+    "\n",
+    "    axes[row, 0].imshow(denorm(img_a))\n",
+    "    axes[row, 0].set_title(\"Before\")\n",
+    "    axes[row, 1].imshow(denorm(img_b))\n",
+    "    axes[row, 1].set_title(\"After\")\n",
+    "    axes[row, 2].imshow(label.squeeze(), cmap=\"gray\")\n",
+    "    axes[row, 2].set_title(\"Ground Truth\")\n",
+    "    axes[row, 3].imshow(pred, cmap=\"gray\")\n",
+    "    axes[row, 3].set_title(\"Prediction\")\n",
+    "\n",
+    "for ax in axes.flat:\n",
+    "    ax.axis(\"off\")\n",
+    "plt.tight_layout()\n",
+    "plt.show()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 10. Export Model for Deployment\n",
+    "\n",
+    "Export as TorchScript for the web app. Download the `.pt` file and place it in\n",
+    "your app's `data/` folder, then set the environment variable:\n",
+    "\n",
+    "```\n",
+    "CHANGE_MODEL_PATH=data/siamese_unet.pt\n",
+    "```"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "model.eval()\n",
+    "model_cpu = model.cpu()\n",
+    "\n",
+    "# Trace with example inputs\n",
+    "example_a = torch.randn(1, 3, 256, 256)\n",
+    "example_b = torch.randn(1, 3, 256, 256)\n",
+    "traced = torch.jit.trace(model_cpu, (example_a, example_b))\n",
+    "\n",
+    "export_path = \"siamese_unet.pt\"\n",
+    "traced.save(export_path)\n",
+    "size_mb = os.path.getsize(export_path) / 1e6\n",
+    "print(f\"Exported TorchScript model: {export_path} ({size_mb:.1f} MB)\")\n",
+    "print(\"\\nDownload this file and place it in your app's data/ directory.\")\n",
+    "print('Then set: CHANGE_MODEL_PATH=data/siamese_unet.pt')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Quick sanity check: verify exported model produces same output\n",
+    "loaded = torch.jit.load(export_path)\n",
+    "with torch.no_grad():\n",
+    "    out_orig = model_cpu(example_a, example_b)\n",
+    "    out_loaded = loaded(example_a, example_b)\n",
+    "    diff = (out_orig - out_loaded).abs().max().item()\n",
+    "    print(f\"Max diff between original and exported: {diff:.8f}\")\n",
+    "    assert diff < 1e-5, \"Export verification failed!\"\n",
+    "    print(\"Export verified successfully.\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 11. Download from Colab\n",
+    "\n",
+    "Run this cell to trigger a browser download of the model file."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "try:\n",
+    "    from google.colab import files\n",
+    "    files.download(\"siamese_unet.pt\")\n",
+    "    files.download(\"best_siamese_unet.pth\")\n",
+    "except ImportError:\n",
+    "    print(\"Not running in Colab. Files saved locally:\")\n",
+    "    print(f\"  - {export_path}\")\n",
+    "    print(f\"  - best_siamese_unet.pth\")"
+   ]
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.11.0"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}