"""
Change Detection using Siamese Networks.
"""
from typing import Tuple, Optional
import numpy as np
import structlog
from geosight.config import settings
logger = structlog.get_logger(__name__)
class ChangeDetector:
"""Siamese network for bi-temporal change detection."""
def __init__(self, device: Optional[str] = None):
self.device = device or settings.model.inference_device
self._model = None
async def detect(
self,
before_image: np.ndarray,
after_image: np.ndarray,
threshold: float = 0.3,
) -> Tuple[np.ndarray, np.ndarray]:
"""
Detect changes between two images.
Returns:
Tuple of (change_mask, change_magnitude)
"""
return self._compute_change(before_image, after_image, threshold)
def _compute_change(
self,
before: np.ndarray,
after: np.ndarray,
threshold: float,
) -> Tuple[np.ndarray, np.ndarray]:
"""Compute change using image differencing."""
# Ensure same shape
if before.shape != after.shape:
min_h = min(before.shape[0], after.shape[0])
min_w = min(before.shape[1], after.shape[1])
before = before[:min_h, :min_w]
after = after[:min_h, :min_w]
# Handle multi-band images
if before.ndim == 3:
before = np.mean(before, axis=2)
if after.ndim == 3:
after = np.mean(after, axis=2)
# Normalize
before = (before - before.mean()) / (before.std() + 1e-8)
after = (after - after.mean()) / (after.std() + 1e-8)
# Compute difference
diff = np.abs(after - before)
# Normalize to 0-1
diff = (diff - diff.min()) / (diff.max() - diff.min() + 1e-8)
# Create binary mask
change_mask = (diff > threshold).astype(np.uint8)
return change_mask, diff.astype(np.float32)