[faces_sub_image] - add overlap patch extraction and boundary blending reconstruction
Replace non-overlapping stride=32 with stride=16 (50% overlap) throughout. Add assemble_overlapping() which accumulates reconstructed patches into a float buffer and averages contributions per pixel, eliminating blocking artefacts. Update load_image_patches to return nx_steps/ny_steps for correct reassembly. Add start parameter to load_patches for held-out test image selection. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,7 @@ from rbm.matrix import Mat, np, convert
|
|||||||
|
|
||||||
DATA_DIR = '/media/jens/cifs/bilder/MachineVision/Caltech_WebFaces/'
|
DATA_DIR = '/media/jens/cifs/bilder/MachineVision/Caltech_WebFaces/'
|
||||||
PATCH = 32
|
PATCH = 32
|
||||||
|
STRIDE = 16 # 50% overlap; set equal to PATCH for non-overlapping
|
||||||
N_CH = 3
|
N_CH = 3
|
||||||
N_VIS = N_CH * PATCH * PATCH # 3072
|
N_VIS = N_CH * PATCH * PATCH # 3072
|
||||||
N_HID = 128
|
N_HID = 128
|
||||||
@@ -32,67 +33,85 @@ class TestModel(Model):
|
|||||||
return self.unit1.reconstruct(x)
|
return self.unit1.reconstruct(x)
|
||||||
|
|
||||||
|
|
||||||
def pad_to_multiple(img: numpy.ndarray, m: int = PATCH) -> numpy.ndarray:
|
def pad_to_stride(img: numpy.ndarray) -> numpy.ndarray:
|
||||||
|
"""Pad so (H - PATCH) and (W - PATCH) are divisible by STRIDE (required by SubImage.check)."""
|
||||||
h, w = img.shape[:2]
|
h, w = img.shape[:2]
|
||||||
return numpy.pad(img, ((0, (-h) % m), (0, (-w) % m), (0, 0)), mode='constant')
|
ph = (-h) % STRIDE if h >= PATCH else PATCH - h
|
||||||
|
pw = (-w) % STRIDE if w >= PATCH else PATCH - w
|
||||||
|
return numpy.pad(img, ((0, ph), (0, pw), (0, 0)), mode='constant')
|
||||||
|
|
||||||
|
|
||||||
def extract_patches_numpy(img_chw: numpy.ndarray, patch: int = PATCH) -> numpy.ndarray:
|
def extract_patches_numpy(img_chw: numpy.ndarray) -> numpy.ndarray:
|
||||||
"""Extract non-overlapping patches (step=patch) from (C,H,W) array; returns (N,C*P*P)."""
|
"""Extract overlapping patches (stride=STRIDE) from (C,H,W) array; returns (N, C*P*P)."""
|
||||||
C, H, W = img_chw.shape
|
C, H, W = img_chw.shape
|
||||||
rows = [img_chw[:, x:x+patch, y:y+patch].flatten()
|
rows = [img_chw[:, x:x+PATCH, y:y+PATCH].flatten()
|
||||||
for x in range(0, H - patch + 1, patch)
|
for x in range(0, H - PATCH + 1, STRIDE)
|
||||||
for y in range(0, W - patch + 1, patch)]
|
for y in range(0, W - PATCH + 1, STRIDE)]
|
||||||
return numpy.stack(rows, axis=0)
|
return numpy.stack(rows, axis=0)
|
||||||
|
|
||||||
|
|
||||||
def load_patches(data_dir: str, n_images: int = N_IMAGES):
|
def load_patches(data_dir: str, n_images: int = N_IMAGES, start: int = 0):
|
||||||
"""Load RGB images, extract non-overlapping 32×32 patches, return normalised cupy array."""
|
"""Load RGB images, extract overlapping patches, return normalised cupy array."""
|
||||||
all_patches = []
|
all_patches = []
|
||||||
files = sorted(f for f in os.listdir(data_dir) if f.endswith('.jpg'))[:n_images]
|
files = sorted(f for f in os.listdir(data_dir) if f.endswith('.jpg'))[start:start + n_images]
|
||||||
for fname in files:
|
for fname in files:
|
||||||
img = cv2.imread(os.path.join(data_dir, fname))
|
img = cv2.imread(os.path.join(data_dir, fname))
|
||||||
if img is None or img.shape[0] < PATCH or img.shape[1] < PATCH:
|
if img is None or img.shape[0] < PATCH or img.shape[1] < PATCH:
|
||||||
continue
|
continue
|
||||||
img = pad_to_multiple(img)
|
img = pad_to_stride(img)
|
||||||
img_f = img[:, :, ::-1].astype(numpy.float64) / 255.0 # BGR→RGB, [0,1]
|
img_f = img[:, :, ::-1].astype(numpy.float64) / 255.0
|
||||||
img_chw = numpy.ascontiguousarray(numpy.transpose(img_f, (2, 0, 1)))
|
img_chw = numpy.ascontiguousarray(numpy.transpose(img_f, (2, 0, 1)))
|
||||||
all_patches.append(extract_patches_numpy(img_chw))
|
all_patches.append(extract_patches_numpy(img_chw))
|
||||||
|
|
||||||
patches_np = numpy.concatenate(all_patches, axis=0)
|
patches_np = numpy.concatenate(all_patches, axis=0)
|
||||||
|
|
||||||
# Drop near-constant patches (std ≈ 0 would give NaN after normalize)
|
|
||||||
std = numpy.std(patches_np, axis=1)
|
std = numpy.std(patches_np, axis=1)
|
||||||
patches_np = patches_np[std > 0.01]
|
patches_np = patches_np[std > 0.01]
|
||||||
|
return normalize(np.array(patches_np))
|
||||||
patches = np.array(patches_np)
|
|
||||||
return normalize(patches)
|
|
||||||
|
|
||||||
|
|
||||||
def load_image_patches(path: str):
|
def load_image_patches(path: str):
|
||||||
"""Extract normalised patches from a single image using SubImage; return (patches, nx, ny)."""
|
"""Extract normalised patches from a single image via SubImage; return (patches, nx_steps, ny_steps)."""
|
||||||
sub = SubImage(PATCH, PATCH, PATCH, PATCH)
|
sub = SubImage(PATCH, PATCH, STRIDE, STRIDE)
|
||||||
img = cv2.imread(path)
|
img = cv2.imread(path)
|
||||||
img = pad_to_multiple(img)
|
img = pad_to_stride(img)
|
||||||
h, w = img.shape[:2]
|
h, w = img.shape[:2]
|
||||||
nx, ny = h // PATCH, w // PATCH
|
nx_steps = (h - PATCH) // STRIDE + 1
|
||||||
|
ny_steps = (w - PATCH) // STRIDE + 1
|
||||||
|
|
||||||
img_f = img[:, :, ::-1].astype(numpy.float64) / 255.0
|
img_f = img[:, :, ::-1].astype(numpy.float64) / 255.0
|
||||||
img_nchw = np.array(numpy.ascontiguousarray(numpy.transpose(img_f, (2, 0, 1))[numpy.newaxis]))
|
img_nchw = np.array(numpy.ascontiguousarray(numpy.transpose(img_f, (2, 0, 1))[numpy.newaxis]))
|
||||||
patches_nchw = sub(img_nchw) # (nx*ny, 3, 32, 32)
|
patches_nchw = sub(img_nchw)
|
||||||
patches = patches_nchw.reshape(-1, N_VIS)
|
patches = patches_nchw.reshape(-1, N_VIS)
|
||||||
|
|
||||||
mean = np.mean(patches, axis=1, keepdims=True)
|
mean = np.mean(patches, axis=1, keepdims=True)
|
||||||
std = np.std(patches, axis=1, keepdims=True)
|
std = np.std(patches, axis=1, keepdims=True)
|
||||||
std = np.where(std < 0.01, np.ones_like(std), std)
|
std = np.where(std < 0.01, np.ones_like(std), std)
|
||||||
return (patches - mean) / std, nx, ny
|
return (patches - mean) / std, nx_steps, ny_steps
|
||||||
|
|
||||||
|
|
||||||
|
def assemble_overlapping(patch_list, nx_steps: int, ny_steps: int) -> numpy.ndarray:
|
||||||
|
"""Reconstruct image from overlapping patches by averaging contributions per pixel."""
|
||||||
|
H = (nx_steps - 1) * STRIDE + PATCH
|
||||||
|
W = (ny_steps - 1) * STRIDE + PATCH
|
||||||
|
accum = numpy.zeros((N_CH, H, W), dtype=numpy.float64)
|
||||||
|
count = numpy.zeros((1, H, W), dtype=numpy.float64)
|
||||||
|
|
||||||
|
for k, p in enumerate(patch_list):
|
||||||
|
x = (k // ny_steps) * STRIDE
|
||||||
|
y = (k % ny_steps) * STRIDE
|
||||||
|
arr = convert(p).reshape(N_CH, PATCH, PATCH)
|
||||||
|
accum[:, x:x+PATCH, y:y+PATCH] += arr
|
||||||
|
count[0, x:x+PATCH, y:y+PATCH] += 1.0
|
||||||
|
|
||||||
|
img = (accum / numpy.maximum(count, 1)).transpose(1, 2, 0) # (H, W, C)
|
||||||
|
lo, hi = img.min(), img.max()
|
||||||
|
return numpy.clip((img - lo) / (hi - lo + 1e-8), 0, 1)
|
||||||
|
|
||||||
|
|
||||||
def show_filters(entity: Entity, rows: int = 8, cols: int = 16):
|
def show_filters(entity: Entity, rows: int = 8, cols: int = 16):
|
||||||
W = convert(entity.state.w_hv) # (N_VIS, N_HID) numpy
|
W = convert(entity.state.w_hv)
|
||||||
fig, axes = plt.subplots(rows, cols, figsize=(cols * 1.2, rows * 1.2))
|
fig, axes = plt.subplots(rows, cols, figsize=(cols * 1.2, rows * 1.2))
|
||||||
fig.suptitle(f'Learned GB-RBM filters ({N_HID} hidden units, {PATCH}×{PATCH} RGB patches)',
|
fig.suptitle(f'Learned GB-RBM filters ({N_HID} hidden units, {PATCH}×{PATCH} RGB patches)', fontsize=10)
|
||||||
fontsize=10)
|
|
||||||
for j in range(rows * cols):
|
for j in range(rows * cols):
|
||||||
ax = axes[j // cols][j % cols]
|
ax = axes[j // cols][j % cols]
|
||||||
ax.axis('off')
|
ax.axis('off')
|
||||||
@@ -126,21 +145,15 @@ def show_reconstructions(model: TestModel, patches: Mat, n_show: int = 10):
|
|||||||
|
|
||||||
|
|
||||||
def show_image_reconstruction(model: TestModel, img_path: str):
|
def show_image_reconstruction(model: TestModel, img_path: str):
|
||||||
patches, nx, ny = load_image_patches(img_path)
|
patches, nx_steps, ny_steps = load_image_patches(img_path)
|
||||||
recons = [model.reconstruct(model.forward(patches[i])) for i in range(patches.shape[0])]
|
recons = [model.reconstruct(model.forward(patches[i])) for i in range(patches.shape[0])]
|
||||||
|
|
||||||
def assemble(patch_list):
|
orig_grid = assemble_overlapping([patches[i] for i in range(len(recons))], nx_steps, ny_steps)
|
||||||
grid = numpy.zeros((nx * PATCH, ny * PATCH, N_CH))
|
recon_grid = assemble_overlapping(recons, nx_steps, ny_steps)
|
||||||
for k, p in enumerate(patch_list):
|
|
||||||
x, y = k // ny, k % ny
|
|
||||||
arr = convert(p).reshape(N_CH, PATCH, PATCH).transpose(1, 2, 0)
|
|
||||||
lo, hi = arr.min(), arr.max()
|
|
||||||
grid[x*PATCH:(x+1)*PATCH, y*PATCH:(y+1)*PATCH] = numpy.clip((arr - lo) / (hi - lo + 1e-8), 0, 1)
|
|
||||||
return grid
|
|
||||||
|
|
||||||
fig, axes = plt.subplots(1, 2, figsize=(12, 6))
|
fig, axes = plt.subplots(1, 2, figsize=(12, 6))
|
||||||
axes[0].imshow(assemble([patches[i] for i in range(len(recons))])); axes[0].set_title('Original (normalised)'); axes[0].axis('off')
|
axes[0].imshow(orig_grid); axes[0].set_title('Original (normalised)'); axes[0].axis('off')
|
||||||
axes[1].imshow(assemble(recons)); axes[1].set_title('Reconstructed'); axes[1].axis('off')
|
axes[1].imshow(recon_grid); axes[1].set_title('Reconstructed (overlap blended)'); axes[1].axis('off')
|
||||||
fig.suptitle(os.path.basename(img_path), fontsize=9)
|
fig.suptitle(os.path.basename(img_path), fontsize=9)
|
||||||
plt.tight_layout()
|
plt.tight_layout()
|
||||||
|
|
||||||
@@ -164,7 +177,7 @@ if __name__ == '__main__':
|
|||||||
model.load()
|
model.load()
|
||||||
|
|
||||||
if args.do_train:
|
if args.do_train:
|
||||||
print(f'Loading patches from {N_IMAGES} images...')
|
print(f'Loading {N_IMAGES} training images (stride={STRIDE})...')
|
||||||
train_patches = load_patches(DATA_DIR, N_IMAGES)
|
train_patches = load_patches(DATA_DIR, N_IMAGES)
|
||||||
print(f'Training on {train_patches.shape[0]} patches ({N_VIS}-dim each)')
|
print(f'Training on {train_patches.shape[0]} patches ({N_VIS}-dim each)')
|
||||||
model.train(train_patches)
|
model.train(train_patches)
|
||||||
@@ -175,13 +188,12 @@ if __name__ == '__main__':
|
|||||||
|
|
||||||
# Show patch-level reconstructions on held-out images
|
# Show patch-level reconstructions on held-out images
|
||||||
print('Loading test patches...')
|
print('Loading test patches...')
|
||||||
test_patches = load_patches(DATA_DIR, n_images=N_IMAGES + 10)
|
test_patches = load_patches(DATA_DIR, n_images=10, start=N_IMAGES)
|
||||||
test_patches = test_patches[N_IMAGES * 80:] # approximate held-out offset
|
|
||||||
if test_patches.shape[0] < 10:
|
if test_patches.shape[0] < 10:
|
||||||
test_patches = load_patches(DATA_DIR, n_images=10)
|
test_patches = load_patches(DATA_DIR, n_images=10)
|
||||||
show_reconstructions(model, test_patches)
|
show_reconstructions(model, test_patches)
|
||||||
|
|
||||||
# Show full image reconstruction
|
# Show full image reconstruction with overlap blending
|
||||||
test_img = os.path.join(DATA_DIR, sorted(os.listdir(DATA_DIR))[60])
|
test_img = os.path.join(DATA_DIR, sorted(os.listdir(DATA_DIR))[60])
|
||||||
show_image_reconstruction(model, test_img)
|
show_image_reconstruction(model, test_img)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user