Upload 8 files

models/__init__.py

@@ -0,0 +1,4 @@
+from .cloud_detector import CloudDetector
+from .change_detector import ChangeDetector
+
+__all__ = ["CloudDetector", "ChangeDetector"]

models/change_detector.py

@@ -0,0 +1,129 @@
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+from transformers import ViTModel, ViTImageProcessor
+import numpy as np
+from typing import Tuple, Optional
+import cv2
+
+
+class ChangeDetector:
+    """
+    Change detection model using Siamese ViT architecture.
+
+    Detects changes between two temporal satellite images.
+    Produces spatial 2D confidence maps and masks.
+    """
+
+    def __init__(
+        self,
+        model_name: str = "google/vit-base-patch16-224",
+        device: Optional[str] = None
+    ):
+        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+        self.model_name = model_name
+
+        self.processor = ViTImageProcessor.from_pretrained(model_name)
+        self.encoder = ViTModel.from_pretrained(model_name)
+        self.encoder.to(self.device)
+        self.encoder.eval()
+
+        hidden_size = self.encoder.config.hidden_size
+        # Lightweight head to score each patch token
+        self.patch_head = nn.Sequential(
+            nn.Linear(hidden_size * 2, 256),
+            nn.ReLU(),
+            nn.Dropout(0.2),
+            nn.Linear(256, 1),
+            nn.Sigmoid()
+        ).to(self.device)
+
+    def _encode_patches(self, image: np.ndarray) -> torch.Tensor:
+        """
+        Encodes the image and return ALL patch tokens (not just CLS).
+
+        Returns:
+            Tensor of shape (num_patches, hidden_size)
+        """
+        # Ensure uint8 [0,255] for processor
+        if image.dtype != np.uint8:
+            img_uint8 = (np.clip(image, 0, 1) * 255).astype(np.uint8)
+        else:
+            img_uint8 = image
+
+        inputs = self.processor(images=img_uint8, return_tensors="pt").to(self.device)
+
+        with torch.no_grad():
+            outputs = self.encoder(**inputs)
+            # last_hidden_state: (1, 1+num_patches, hidden_size)
+            # index 0 is CLS, 1: are patch tokens
+            patch_tokens = outputs.last_hidden_state[0, 1:, :]  # (num_patches, H)
+
+        return patch_tokens
+
+    def detect_changes(
+        self,
+        before_image: np.ndarray,
+        after_image: np.ndarray,
+        threshold: float = 0.5
+    ) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        Detect changes between two temporal images.
+
+        Returns:
+            change_mask: 2D binary array (H, W) matching input image size
+            confidence_map: 2D float array (H, W) in [0, 1]
+        """
+        h, w = before_image.shape[:2]
+
+        before_patches = self._encode_patches(before_image)  # (N, D)
+        after_patches = self._encode_patches(after_image)    # (N, D)
+
+        # Concatenate patch-wise features
+        combined = torch.cat([before_patches, after_patches], dim=-1)  # (N, 2D)
+
+        with torch.no_grad():
+            patch_scores = self.patch_head(combined).squeeze(-1)  # (N,)
+
+        patch_scores_np = patch_scores.cpu().numpy()  # shape (num_patches,)
+
+        # ViT-base/16 on 224x224 → 14x14 = 196 patches
+        n = patch_scores_np.shape[0]
+        grid = int(np.sqrt(n))
+        if grid * grid != n:
+            # Fallback: pad to nearest square
+            grid = int(np.ceil(np.sqrt(n)))
+            pad = grid * grid - n
+            patch_scores_np = np.concatenate([patch_scores_np, np.zeros(pad)])
+
+        patch_map = patch_scores_np.reshape(grid, grid)
+
+        # Upsample patch-level map to original image size
+        confidence_map = cv2.resize(
+            patch_map.astype(np.float32),
+            (w, h),
+            interpolation=cv2.INTER_LINEAR
+        )
+        confidence_map = np.clip(confidence_map, 0.0, 1.0)
+
+        # Threshold to binary mask
+        change_mask = (confidence_map > threshold).astype(np.uint8)
+
+        # Morphological cleanup to reduce noise
+        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+        change_mask = cv2.morphologyEx(change_mask, cv2.MORPH_OPEN, kernel, iterations=1)
+
+        return change_mask, confidence_map
+
+    def batch_detect_changes(
+        self,
+        before_images: np.ndarray,
+        after_images: np.ndarray,
+        threshold: float = 0.5
+    ) -> Tuple[np.ndarray, np.ndarray]:
+        masks, confidences = [], []
+        for b, a in zip(before_images, after_images):
+            mask, conf = self.detect_changes(b, a, threshold)
+            masks.append(mask)
+            confidences.append(conf)
+        return np.array(masks), np.array(confidences)

models/cloud_detector.py

@@ -0,0 +1,86 @@
+import numpy as np
+from typing import Tuple, Optional
+import cv2
+
+
+class CloudDetector:
+    """
+    Cloud detection for satellite imagery using brightness and saturation analysis.
+
+    For RGB satellite images, clouds are detected based on:
+    - High brightness (RGB values close to 255)
+    - Low saturation (near-white appearance)
+    - Spatial clustering (using morphological operations)
+    """
+
+    def __init__(self, device: Optional[str] = None):
+        self.device = device or "cpu"
+
+    def detect_clouds(
+        self,
+        image: np.ndarray,
+        threshold: float = 0.5
+    ) -> Tuple[np.ndarray, np.ndarray]:
+        """
+        Detect clouds in satellite image.
+
+        Args are:
+            image: Input image (H, W, 3), uint8 [0,255] or float [0,1]
+            threshold: Cloud confidence threshold (0-1)
+
+        Returns:
+            cloud_mask: 2D binary array (H, W)
+            cloud_confidence: 2D float array (H, W) in [0,1]
+        """
+        # Normalise to float [0, 1]
+        if image.dtype == np.uint8:
+            img = image.astype(np.float32) / 255.0
+        else:
+            img = np.clip(image, 0, 1).astype(np.float32)
+
+        # Handle grayscale
+        if img.ndim == 2:
+            img = np.stack([img, img, img], axis=-1)
+        elif img.shape[2] == 1:
+            img = np.concatenate([img, img, img], axis=-1)
+
+        red   = img[:, :, 0]
+        green = img[:, :, 1]
+        blue  = img[:, :, 2]
+
+        # Brightness: mean of RGB channels
+        brightness = (red + green + blue) / 3.0
+
+        # Saturation (HSV-style for RGB)
+        max_rgb = np.maximum(np.maximum(red, green), blue)
+        min_rgb = np.minimum(np.minimum(red, green), blue)
+        saturation = np.where(
+            max_rgb > 0,
+            (max_rgb - min_rgb) / (max_rgb + 1e-8),
+            0.0
+        )
+
+        # Cloud score: high brightness + low saturation
+        brightness_score = np.clip((brightness - 0.4) / 0.6, 0, 1)
+        saturation_score = np.clip((0.4 - saturation) / 0.4, 0, 1)
+
+        cloud_confidence = (0.6 * brightness_score + 0.4 * saturation_score).astype(np.float32)
+
+        # Binary mask + morphological cleanup
+        cloud_mask = (cloud_confidence > threshold).astype(np.uint8)
+        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3))
+        cloud_mask = cv2.morphologyEx(cloud_mask, cv2.MORPH_OPEN, kernel, iterations=1)
+
+        return cloud_mask, cloud_confidence
+
+    def batch_detect(
+        self,
+        images: np.ndarray,
+        threshold: float = 0.5
+    ) -> Tuple[np.ndarray, np.ndarray]:
+        masks, confidences = [], []
+        for image in images:
+            mask, conf = self.detect_clouds(image, threshold)
+            masks.append(mask)
+            confidences.append(conf)
+        return np.array(masks), np.array(confidences)

utils/__init__.py

@@ -0,0 +1,16 @@
+"""Utility modules for satellite change detection."""
+
+from .preprocessing import preprocess_image, mask_clouds
+from .visualization import create_overlay, visualize_predictions
+from .evaluation import calculate_metrics
+from .metrics import calculate_change_statistics, compare_with_without_masking
+
+__all__ = [
+    "preprocess_image",
+    "mask_clouds",
+    "create_overlay",
+    "visualize_predictions",
+    "calculate_metrics",
+    "calculate_change_statistics",
+    "compare_with_without_masking",
+]

utils/evaluation.py

@@ -0,0 +1,54 @@
+"""Evaluation metrics for change and cloud detection."""
+
+import numpy as np
+from typing import Dict, Optional
+
+
+def calculate_metrics(
+    pred_mask: np.ndarray,
+    gt_mask: np.ndarray,
+    threshold: float = 0.5
+) -> Dict[str, float]:
+    """
+    Calculate pixel-level classification metrics.
+
+    Args:
+        pred_mask: Predicted binary mask (H, W) or confidence map (H, W)
+        gt_mask: Ground truth binary mask (H, W)
+        threshold: Threshold to binarise pred_mask if it's a confidence map
+
+    Returns:
+        Dict with keys: accuracy, precision, recall, f1, iou
+    """
+    # Binarise predictions if needed
+    if pred_mask.dtype != np.uint8 or pred_mask.max() > 1:
+        pred = (pred_mask > threshold).astype(np.uint8)
+    else:
+        pred = pred_mask.astype(np.uint8)
+
+    gt = (gt_mask > 0).astype(np.uint8)
+
+    tp = int(np.sum((pred == 1) & (gt == 1)))
+    tn = int(np.sum((pred == 0) & (gt == 0)))
+    fp = int(np.sum((pred == 1) & (gt == 0)))
+    fn = int(np.sum((pred == 0) & (gt == 1)))
+
+    total = tp + tn + fp + fn
+    accuracy  = (tp + tn) / total if total > 0 else 0.0
+    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
+    recall    = tp / (tp + fn) if (tp + fn) > 0 else 0.0
+    f1        = (2 * precision * recall / (precision + recall)
+                 if (precision + recall) > 0 else 0.0)
+    iou       = tp / (tp + fp + fn) if (tp + fp + fn) > 0 else 0.0
+
+    return {
+        "accuracy": accuracy,
+        "precision": precision,
+        "recall": recall,
+        "f1": f1,
+        "iou": iou,
+        "tp": tp,
+        "tn": tn,
+        "fp": fp,
+        "fn": fn,
+    }

utils/metrics.py

@@ -0,0 +1,95 @@
+"""Advanced comparison metrics for change detection evaluation."""
+
+import numpy as np
+from typing import Dict, Optional
+from .evaluation import calculate_metrics
+
+
+def calculate_change_statistics(
+    change_mask: np.ndarray,
+    change_confidence: np.ndarray
+) -> Dict:
+    """
+    Calculate statistics from a 2D change mask and confidence map.
+
+    Args:
+        change_mask: Binary 2D array (H, W), 1 = changed
+        change_confidence: Float 2D array (H, W) in [0, 1]
+
+    Returns:
+        Dict with total_pixels, changed_pixels, unchanged_pixels,
+        change_percentage, mean_confidence, min_confidence, max_confidence,
+        change_confidence_mean (mean conf among changed pixels only)
+    """
+    # Ensure 2D arrays
+    mask = change_mask.astype(np.uint8)
+    conf = change_confidence.astype(np.float32)
+
+    total_pixels     = int(mask.size)
+    changed_pixels   = int(np.sum(mask == 1))
+    unchanged_pixels = total_pixels - changed_pixels
+    change_percentage = 100.0 * changed_pixels / total_pixels if total_pixels > 0 else 0.0
+
+    mean_confidence = float(conf.mean())
+    min_confidence  = float(conf.min())
+    max_confidence  = float(conf.max())
+
+    # Mean confidence among changed pixels only
+    if changed_pixels > 0:
+        change_confidence_mean = float(conf[mask == 1].mean())
+    else:
+        change_confidence_mean = 0.0
+
+    return {
+        "total_pixels":          total_pixels,
+        "changed_pixels":        changed_pixels,
+        "unchanged_pixels":      unchanged_pixels,
+        "change_percentage":     change_percentage,
+        "mean_confidence":       mean_confidence,
+        "min_confidence":        min_confidence,
+        "max_confidence":        max_confidence,
+        "change_confidence_mean": change_confidence_mean,
+    }
+
+
+def compare_with_without_masking(
+    pred_with_mask: np.ndarray,
+    pred_without_mask: np.ndarray,
+    gt_mask: Optional[np.ndarray] = None
+) -> Dict:
+    """
+    Compare detection results with and without cloud masking.
+
+    Args:
+        pred_with_mask: Change mask produced WITH cloud masking (H, W)
+        pred_without_mask: Change mask produced WITHOUT cloud masking (H, W)
+        gt_mask: Optional ground truth mask for metric computation (H, W)
+
+    Returns:
+        Dict with pixel-level comparison and optional metric differences
+    """
+    agreement = int(np.sum(pred_with_mask == pred_without_mask))
+    total     = int(pred_with_mask.size)
+    agreement_pct = 100.0 * agreement / total if total > 0 else 0.0
+
+    result = {
+        "agreement_pixels":     agreement,
+        "total_pixels":         total,
+        "agreement_percentage": agreement_pct,
+        "changed_with_mask":    int(np.sum(pred_with_mask)),
+        "changed_without_mask": int(np.sum(pred_without_mask)),
+    }
+
+    if gt_mask is not None:
+        metrics_with    = calculate_metrics(pred_with_mask, gt_mask)
+        metrics_without = calculate_metrics(pred_without_mask, gt_mask)
+
+        result["iou_with_mask"]     = metrics_with["iou"]
+        result["iou_without_mask"]  = metrics_without["iou"]
+        result["iou_improvement"]   = metrics_with["iou"] - metrics_without["iou"]
+
+        result["f1_with_mask"]      = metrics_with["f1"]
+        result["f1_without_mask"]   = metrics_without["f1"]
+        result["f1_improvement"]    = metrics_with["f1"] - metrics_without["f1"]
+
+    return result

utils/preprocessing.py

@@ -0,0 +1,79 @@
+"""Preprocessing utilities for satellite imagery."""
+
+import numpy as np
+import cv2
+from typing import Optional
+
+
+def preprocess_image(
+    image: np.ndarray,
+    target_size: Optional[tuple] = None,
+    normalize: bool = True
+) -> np.ndarray:
+    """
+    Preprocess a satellite image for model input.
+
+    Args:
+        image: Input image (H, W, C), uint8 or float
+        target_size: Optional (width, height) to resize to
+        normalize: If True, output is float32 in [0, 1]
+
+    Returns:
+        Preprocessed image as float32 [0,1] or uint8 [0,255]
+    """
+    if image is None:
+        raise ValueError("Input image is None")
+
+    img = image.copy()
+
+    # Ensure 3-channel
+    if img.ndim == 2:
+        img = np.stack([img, img, img], axis=-1)
+    elif img.shape[2] == 1:
+        img = np.concatenate([img, img, img], axis=-1)
+    elif img.shape[2] > 3:
+        img = img[:, :, :3]
+
+    # Resize if requested
+    if target_size is not None:
+        img = cv2.resize(img, target_size, interpolation=cv2.INTER_LINEAR)
+
+    # Normalise
+    if normalize:
+        if img.dtype == np.uint8:
+            img = img.astype(np.float32) / 255.0
+        else:
+            img = np.clip(img, 0, 1).astype(np.float32)
+    else:
+        if img.dtype != np.uint8:
+            img = (np.clip(img, 0, 1) * 255).astype(np.uint8)
+
+    return img
+
+
+def mask_clouds(
+    image: np.ndarray,
+    cloud_mask: np.ndarray,
+    fill_value: float = 0.0
+) -> np.ndarray:
+    """
+    Apply cloud mask to image, replacing cloud pixels with fill_value.
+
+    Args:
+        image: Input image (H, W, C)
+        cloud_mask: Binary mask (H, W), 1 = cloud
+        fill_value: Value to fill masked pixels with
+
+    Returns:
+        Masked image same dtype as input
+    """
+    masked = image.copy().astype(np.float32)
+    mask_bool = cloud_mask.astype(bool)
+
+    for c in range(masked.shape[2]):
+        masked[:, :, c][mask_bool] = fill_value
+
+    if image.dtype == np.uint8:
+        masked = np.clip(masked, 0, 255).astype(np.uint8)
+
+    return masked

utils/visualization.py

@@ -0,0 +1,96 @@
+"""Visualization utilities for satellite change detection."""
+
+import numpy as np
+import cv2
+from typing import Tuple, Optional
+
+
+def create_overlay(
+    image: np.ndarray,
+    mask: np.ndarray,
+    alpha: float = 0.5,
+    color: Tuple[int, int, int] = (255, 0, 0)
+) -> np.ndarray:
+    """
+    Overlay a binary mask on an image with transparency.
+
+    Args:
+        image: Base image (H, W, 3), uint8 or float [0,1]
+        mask: Binary mask (H, W), values 0 or 1
+        alpha: Overlay transparency (0 = invisible, 1 = opaque)
+        color: RGB color for the overlay
+
+    Returns:
+        Blended image as uint8 (H, W, 3)
+    """
+    # Convert image to uint8
+    if image.dtype != np.uint8:
+        base = (np.clip(image, 0, 1) * 255).astype(np.uint8)
+    else:
+        base = image.copy()
+
+    # Ensure 3 channels
+    if base.ndim == 2:
+        base = cv2.cvtColor(base, cv2.COLOR_GRAY2RGB)
+
+    overlay = base.copy()
+    mask_bool = mask.astype(bool)
+
+    # Apply colour to masked region
+    overlay[mask_bool] = [color[0], color[1], color[2]]
+
+    # Blend
+    result = cv2.addWeighted(overlay, alpha, base, 1 - alpha, 0)
+    return result.astype(np.uint8)
+
+
+def visualize_predictions(
+    image: np.ndarray,
+    pred_mask: np.ndarray,
+    gt_mask: Optional[np.ndarray] = None,
+    confidence: Optional[np.ndarray] = None
+) -> np.ndarray:
+    """
+    Create a side-by-side visualization of image, prediction, and (optionally) ground truth.
+
+    Args:
+        image: Original image (H, W, 3)
+        pred_mask: Predicted binary mask (H, W)
+        gt_mask: Optional ground truth mask (H, W)
+        confidence: Optional confidence map (H, W)
+
+    Returns:
+        Combined visualization as uint8 (H, W*N, 3)
+    """
+    if image.dtype != np.uint8:
+        img_u8 = (np.clip(image, 0, 1) * 255).astype(np.uint8)
+    else:
+        img_u8 = image.copy()
+
+    h, w = img_u8.shape[:2]
+    panels = [img_u8]
+
+    # Prediction overlay (red)
+    pred_overlay = create_overlay(img_u8, pred_mask, alpha=0.5, color=(255, 0, 0))
+    panels.append(pred_overlay)
+
+    # Ground truth overlay (green)
+    if gt_mask is not None:
+        gt_overlay = create_overlay(img_u8, gt_mask, alpha=0.5, color=(0, 255, 0))
+        panels.append(gt_overlay)
+
+    # Confidence heatmap
+    if confidence is not None:
+        heatmap = (np.clip(confidence, 0, 1) * 255).astype(np.uint8)
+        heatmap_color = cv2.applyColorMap(heatmap, cv2.COLORMAP_JET)
+        heatmap_color = cv2.cvtColor(heatmap_color, cv2.COLOR_BGR2RGB)
+        panels.append(heatmap_color)
+
+    # Resize all panels to same height
+    panels_resized = [
+        cv2.resize(p, (w, h), interpolation=cv2.INTER_LINEAR)
+        if p.shape[:2] != (h, w) else p
+        for p in panels
+    ]
+
+    return np.concatenate(panels_resized, axis=1)