Spaces:
Sleeping
Sleeping
File size: 4,321 Bytes
d70361b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """
KPCA-based unsupervised change detection (KPCAMNet-inspired).
Fits Kernel PCA filters on randomly sampled patches from both images (weight-
shared/siamese, no labels needed), projects the whole image through those
filters, then detects change via the L2 norm of the feature difference. This
is a single-stage reconstruction of the described KPCAMNet algorithm (not a
port of the original repo's code, which isn't part of this project) — scoped
down for CPU feasibility: bounded processing resolution and a modest training
patch count, since KernelPCA.transform cost scales with
n_query_patches x n_train_patches.
"""
import cv2
import numpy as np
from numpy.lib.stride_tricks import sliding_window_view
from sklearn.decomposition import KernelPCA
def _extract_all_patches(img: np.ndarray, patch_size: int) -> np.ndarray:
"""Every overlapping patch_size x patch_size patch, one per pixel (reflect-padded)."""
half = patch_size // 2
padded = np.pad(img, half, mode="reflect")
windows = sliding_window_view(padded, (patch_size, patch_size))
h, w = img.shape
return windows.reshape(h, w, -1)
def _project_in_chunks(kpca: KernelPCA, flat_patches: np.ndarray, row_pixels: int,
n_components: int) -> np.ndarray:
"""KernelPCA.transform in row-chunks so memory stays bounded on large images."""
chunks = []
step = max(1, 200_000 // max(1, row_pixels)) * row_pixels
for i in range(0, flat_patches.shape[0], step):
chunks.append(kpca.transform(flat_patches[i:i + step]))
return np.concatenate(chunks, axis=0)
def kpca_change_mask(img1: np.ndarray, img2: np.ndarray, patch_size: int = 5,
n_components: int = 8, gamma: float = 5e-4,
n_train_patches: int = 300, max_side: int = 768):
"""
Returns (magnitude_map float32 [0,1] at processed resolution, debug dict).
Caller handles thresholding/cleanup/resize-back so this stays testable in
isolation without pulling in the rest of the detection pipeline.
"""
if img1.shape != img2.shape:
img2 = cv2.resize(img2, (img1.shape[1], img1.shape[0]))
h0, w0 = img1.shape[:2]
scale = min(1.0, max_side / max(h0, w0, 1))
if scale < 1.0:
nh, nw = max(1, int(h0 * scale)), max(1, int(w0 * scale))
img1_s = cv2.resize(img1, (nw, nh), interpolation=cv2.INTER_AREA)
img2_s = cv2.resize(img2, (nw, nh), interpolation=cv2.INTER_AREA)
else:
img1_s, img2_s = img1, img2
gray1 = cv2.cvtColor(img1_s, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0
gray2 = cv2.cvtColor(img2_s, cv2.COLOR_RGB2GRAY).astype(np.float32) / 255.0
# Per-image z-score standardization (matches the described preprocessing)
gray1 = (gray1 - gray1.mean()) / (gray1.std() + 1e-8)
gray2 = (gray2 - gray2.mean()) / (gray2.std() + 1e-8)
h, w = gray1.shape
patches1 = _extract_all_patches(gray1, patch_size)
patches2 = _extract_all_patches(gray2, patch_size)
flat1 = patches1.reshape(-1, patches1.shape[-1])
flat2 = patches2.reshape(-1, patches2.shape[-1])
rng = np.random.default_rng(42)
half_n = max(1, n_train_patches // 2)
idx1 = rng.choice(flat1.shape[0], size=min(half_n, flat1.shape[0]), replace=False)
idx2 = rng.choice(flat2.shape[0], size=min(half_n, flat2.shape[0]), replace=False)
train_patches = np.concatenate([flat1[idx1], flat2[idx2]], axis=0)
n_components_eff = max(1, min(n_components, train_patches.shape[0] - 1))
kpca = KernelPCA(n_components=n_components_eff, kernel="rbf", gamma=gamma)
kpca.fit(train_patches)
feat1 = _project_in_chunks(kpca, flat1, w, n_components_eff).reshape(h, w, n_components_eff)
feat2 = _project_in_chunks(kpca, flat2, w, n_components_eff).reshape(h, w, n_components_eff)
diff = feat1 - feat2
magnitude = np.linalg.norm(diff, axis=-1)
mag_norm = magnitude / (magnitude.max() + 1e-8)
if scale < 1.0:
mag_norm = cv2.resize(mag_norm.astype(np.float32), (w0, h0), interpolation=cv2.INTER_LINEAR)
debug = {
"n_components": int(n_components_eff),
"n_train_patches": int(train_patches.shape[0]),
"processedSide": int(max(h, w)),
"patchSize": int(patch_size),
"gamma": float(gamma),
}
return mag_norm.astype(np.float32), debug
|