Pooling and Downsampling

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

import numpy as np
 
def avg_pool_forward(X, pool_size=2, stride=2):
    """
    Forward pass for average pooling.
    
    Args:
        X: Input tensor of shape (batch, channels, height, width)
        pool_size: Size of pooling window (assumes square)
        stride: Stride for pooling operation
    
    Returns:
        out: Pooled output
    """
    N, C, H, W = X.shape
    H_out = (H - pool_size) // stride + 1
    W_out = (W - pool_size) // stride + 1
    
    out = np.zeros((N, C, H_out, W_out))
    
    for i in range(H_out):
        for j in range(W_out):
            h_start = i * stride
            w_start = j * stride
            
            # Extract pooling region and compute mean
            region = X[:, :, h_start:h_start+pool_size, 
                               w_start:w_start+pool_size]
            out[:, :, i, j] = np.mean(region, axis=(2, 3))
    
    return out
 
def avg_pool_backward(dout, input_shape, pool_size=2, stride=2):
    """
    Backward pass for average pooling.
    
    Gradients are distributed uniformly to all positions in the 
    pooling region. Each position receives 1/(pool_size^2) of the
    upstream gradient.
    """
    N, C, H, W = input_shape
    dX = np.zeros(input_shape)
    
    _, _, H_out, W_out = dout.shape
    scale = 1.0 / (pool_size * pool_size)
    
    for i in range(H_out):
        for j in range(W_out):
            h_start = i * stride
            w_start = j * stride
            
            # Distribute gradient uniformly
            for m in range(pool_size):
                for n in range(pool_size):
                    dX[:, :, h_start+m, w_start+n] += dout[:, :, i, j] * scale
    
    return dX
 
# Comparison of gradient behavior
def compare_gradient_behavior():
    """
    Demonstrate the difference between max and average pooling gradients.
    """
    # Create a 2x2 region with varying activations
    region = np.array([[1.0, 3.0], 
                       [2.0, 0.5]])
    upstream_grad = 1.0  # Gradient from the layer above
    
    # Max pooling: gradient goes only to maximum (3.0 at position [0,1])
    max_grad = np.array([[0.0, 1.0], 
                         [0.0, 0.0]])
    
    # Average pooling: gradient distributed uniformly
    avg_grad = np.array([[0.25, 0.25], 
                         [0.25, 0.25]])
    
    print("Input region:")
    print(region)
    print(f"\nMax pooling output: {np.max(region):.2f}")
    print(f"Average pooling output: {np.mean(region):.2f}")
    print(f"\nMax pooling gradient:\n{max_grad}")
    print(f"\nAverage pooling gradient:\n{avg_grad}")
    
compare_gradient_behavior()

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

import numpy as np
import torch
import torch.nn.functional as F
 
def demonstrate_noise_reduction():
    """
    Demonstrate how average pooling reduces noise while max pooling
    can amplify noise peaks.
    """
    np.random.seed(42)
    
    # Create a clean signal: a centered blob of activation
    H, W = 8, 8
    clean_signal = np.zeros((1, 1, H, W))
    
    # Create a smooth Gaussian-like feature
    for i in range(H):
        for j in range(W):
            dist_sq = (i - 3.5)**2 + (j - 3.5)**2
            clean_signal[0, 0, i, j] = np.exp(-dist_sq / 4.0)
    
    # Add Gaussian noise
    noise = np.random.normal(0, 0.3, clean_signal.shape)
    noisy_signal = clean_signal + noise
    
    # Clip any negative values (post-ReLU assumption)
    noisy_signal = np.maximum(noisy_signal, 0)
    
    # Convert to torch tensors
    clean_t = torch.tensor(clean_signal, dtype=torch.float32)
    noisy_t = torch.tensor(noisy_signal, dtype=torch.float32)
    
    # Apply pooling
    max_clean = F.max_pool2d(clean_t, 2, 2)
    max_noisy = F.max_pool2d(noisy_t, 2, 2)
    avg_clean = F.avg_pool2d(clean_t, 2, 2)
    avg_noisy = F.avg_pool2d(noisy_t, 2, 2)
    
    # Measure distortion from clean reference
    max_dist = torch.mean((max_noisy - max_clean)**2).item()
    avg_dist = torch.mean((avg_noisy - avg_clean)**2).item()
    
    print("Noise-induced distortion (MSE from clean pooled output):")
    print(f"  Max pooling: {max_dist:.6f}")
    print(f"  Average pooling: {avg_dist:.6f}")
    print(f"  Ratio (max/avg): {max_dist/avg_dist:.2f}x")
    
    # Result: Average pooling shows lower distortion because noise is averaged
    # rather than selected when it happens to be the maximum
 
demonstrate_noise_reduction()

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

import torch
import torch.nn as nn
import torch.nn.functional as F
 
class ClassicClassificationHead(nn.Module):
    """
    Classic approach: Flatten + Fully Connected layers.
    Requires fixed input size, many parameters.
    """
    def __init__(self, in_channels, spatial_size, num_classes):
        super().__init__()
        flat_size = in_channels * spatial_size * spatial_size
        self.fc = nn.Sequential(
            nn.Flatten(),
            nn.Linear(flat_size, 4096),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(4096, 4096),
            nn.ReLU(),
            nn.Dropout(0.5),
            nn.Linear(4096, num_classes)
        )
        
    def forward(self, x):
        return self.fc(x)
 
 
class GAPClassificationHead(nn.Module):
    """
    Modern approach: Global Average Pooling + single linear layer.
    Works with any input size, far fewer parameters.
    """
    def __init__(self, in_channels, num_classes):
        super().__init__()
        self.gap = nn.AdaptiveAvgPool2d(1)  # Output: (batch, channels, 1, 1)
        self.fc = nn.Linear(in_channels, num_classes)
        
    def forward(self, x):
        x = self.gap(x)  # (batch, channels, 1, 1)
        x = x.view(x.size(0), -1)  # (batch, channels)
        return self.fc(x)
 
 
# Parameter comparison
def compare_parameters():
    in_channels = 512
    spatial = 7
    num_classes = 1000
    
    classic = ClassicClassificationHead(in_channels, spatial, num_classes)
    gap_based = GAPClassificationHead(in_channels, num_classes)
    
    classic_params = sum(p.numel() for p in classic.parameters())
    gap_params = sum(p.numel() for p in gap_based.parameters())
    
    print("Parameter Comparison:")
    print(f"  Classic FC head: {classic_params:,} parameters")
    print(f"  GAP-based head: {gap_params:,} parameters")
    print(f"  Reduction: {classic_params / gap_params:.1f}x fewer parameters")
 
compare_parameters()
# Output:
#   Classic FC head: ~119,000,000 parameters  
#   GAP-based head: ~513,000 parameters
#   Reduction: ~232x fewer parameters

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

import torch
import torch.nn as nn
 
class ConcatPooling(nn.Module):
    """
    Concatenate max and average pooling outputs.
    Used in fastai and other frameworks for classification.
    
    This captures both peak activations (max) and aggregate 
    statistics (avg), giving the classifier richer information.
    """
    def __init__(self):
        super().__init__()
        self.gap = nn.AdaptiveAvgPool2d(1)
        self.gmp = nn.AdaptiveMaxPool2d(1)  # Global Max Pooling
        
    def forward(self, x):
        avg_out = self.gap(x)  # (batch, C, 1, 1)
        max_out = self.gmp(x)  # (batch, C, 1, 1)
        return torch.cat([avg_out, max_out], dim=1)  # (batch, 2C, 1, 1)
 
 
class SpatialPyramidPooling(nn.Module):
    """
    Spatial Pyramid Pooling (SPP) combines pooling at multiple scales.
    Enables fixed-size output regardless of input dimensions.
    
    Each level uses average pooling at different granularities.
    """
    def __init__(self, levels=[1, 2, 4]):
        super().__init__()
        self.levels = levels
        
    def forward(self, x):
        batch_size, channels, H, W = x.shape
        outputs = []
        
        for level in self.levels:
            # Pool to level × level grid
            pool = nn.AdaptiveAvgPool2d(level)
            out = pool(x)  # (batch, C, level, level)
            out = out.view(batch_size, -1)  # Flatten spatial dims
            outputs.append(out)
        
        # Concatenate all levels
        return torch.cat(outputs, dim=1)
 
 
class MixedPoolingBlock(nn.Module):
    """
    Inception-style mixed pooling: parallel max and avg pooling
    branches that are concatenated.
    """
    def __init__(self, in_channels):
        super().__init__()
        self.max_pool = nn.MaxPool2d(3, stride=1, padding=1)
        self.avg_pool = nn.AvgPool2d(3, stride=1, padding=1)
        self.conv_reduce = nn.Conv2d(in_channels * 2, in_channels, 1)
        
    def forward(self, x):
        max_out = self.max_pool(x)
        avg_out = self.avg_pool(x)
        concat = torch.cat([max_out, avg_out], dim=1)
        return self.conv_reduce(concat)
 
 
# Usage example
def test_pooling_variants():
    x = torch.randn(8, 256, 14, 14)
    
    concat_pool = ConcatPooling()
    spp = SpatialPyramidPooling([1, 2, 4])
    mixed = MixedPoolingBlock(256)
    
    print(f"Input shape: {x.shape}")
    print(f"ConcatPooling output: {concat_pool(x).shape}")  # (8, 512, 1, 1)
    print(f"SPP output: {spp(x).shape}")  # (8, 256*(1+4+16)) = (8, 5376)
    print(f"MixedPooling output: {mixed(x).shape}")  # (8, 256, 14, 14)
 
test_pooling_variants()

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
99
100
101
102
103
104

import torch
import torch.nn as nn
import tensorflow as tf
 
# ===========================================
# PyTorch Average Pooling
# ===========================================
 
# Standard 2D average pooling
x = torch.randn(32, 64, 28, 28)  # NCHW format
 
# Functional API
out = torch.nn.functional.avg_pool2d(
    x,
    kernel_size=2,
    stride=2,
    padding=0,
    ceil_mode=False,
    count_include_pad=True,  # Include padding zeros in average
    divisor_override=None    # Custom divisor (not recommended)
)
 
# Module API
avg_pool = nn.AvgPool2d(
    kernel_size=2,
    stride=2,
    padding=0,
    ceil_mode=False,
    count_include_pad=True
)
 
# Adaptive pooling (variable input size → fixed output size)
adaptive_avg = nn.AdaptiveAvgPool2d(output_size=(7, 7))  # Any input → 7×7
global_avg = nn.AdaptiveAvgPool2d(output_size=1)  # GAP
 
print(f"Standard AvgPool2d output: {avg_pool(x).shape}")  # (32, 64, 14, 14)
print(f"Adaptive 7×7 output: {adaptive_avg(x).shape}")  # (32, 64, 7, 7)
print(f"Global average output: {global_avg(x).shape}")  # (32, 64, 1, 1)
 
# 1D average pooling (for sequences/temporal data)
x_1d = torch.randn(32, 64, 100)  # (batch, channels, length)
avg_pool_1d = nn.AvgPool1d(kernel_size=4, stride=2)
print(f"AvgPool1d output: {avg_pool_1d(x_1d).shape}")  # (32, 64, 49)
 
# 3D average pooling (for video/volumetric data)
x_3d = torch.randn(2, 16, 8, 28, 28)  # (batch, C, D, H, W)
avg_pool_3d = nn.AvgPool3d(kernel_size=(2, 2, 2), stride=2)
print(f"AvgPool3d output: {avg_pool_3d(x_3d).shape}")  # (2, 16, 4, 14, 14)
 
 
# ===========================================
# TensorFlow / Keras Average Pooling
# ===========================================
 
x_tf = tf.random.normal([32, 28, 28, 64])  # NHWC format (TF default)
 
# Functional API
out_tf = tf.nn.avg_pool(
    x_tf,
    ksize=[1, 2, 2, 1],
    strides=[1, 2, 2, 1],
    padding='VALID'
)
print(f"TF avg_pool output: {out_tf.shape}")  # (32, 14, 14, 64)
 
# Keras Layer API
keras_avg = tf.keras.layers.AveragePooling2D(
    pool_size=(2, 2),
    strides=(2, 2),
    padding='valid'
)
 
# Global average pooling in Keras
keras_gap = tf.keras.layers.GlobalAveragePooling2D()
gap_output = keras_gap(x_tf)
print(f"Keras GAP output: {gap_output.shape}")  # (32, 64)
 
 
# ===========================================
# count_include_pad behavior
# ===========================================
 
def demonstrate_count_include_pad():
    """
    When padding is used, count_include_pad determines whether
    padding zeros are included in the average calculation.
    """
    x = torch.tensor([[[[1., 2.], [3., 4.]]]])  # 1×1×2×2
    
    # With padding=1, we get a 4×4 padded input
    # count_include_pad=True: divide by all positions including padding
    # count_include_pad=False: divide only by valid positions
    
    pool_include = nn.AvgPool2d(2, stride=1, padding=1, count_include_pad=True)
    pool_exclude = nn.AvgPool2d(2, stride=1, padding=1, count_include_pad=False)
    
    print("Input:")
    print(x.squeeze())
    print("\nWith count_include_pad=True (includes zeros in denominator):")
    print(pool_include(x).squeeze())
    print("\nWith count_include_pad=False (excludes zeros from denominator):")  
    print(pool_exclude(x).squeeze())
 
demonstrate_count_include_pad()