mirror of
https://github.com/MLSysBook/TinyTorch.git
synced 2026-07-20 07:58:34 -05:00
- Remove circular imports where modules imported from themselves - Convert tinytorch.core imports to sys.path relative imports - Only import dependencies that are actually used in each module - Preserve documentation imports in markdown cells - Use consistent relative path pattern across all modules - Remove hardcoded absolute paths in favor of relative imports Affected modules: 02_activations, 03_layers, 04_losses, 06_optimizers, 07_training, 09_spatial, 12_attention, 17_quantization
82 KiB
82 KiB
In [ ]:
#| default_exp core.spatial
#| export
import numpy as np
import sys
import os
import time
# Smart import system for development and production compatibility
if 'tinytorch' in sys.modules:
# Production: Import from installed package
from tinytorch.core.tensor import Tensor
from tinytorch.core.layers import Module
else:
# Development: Use simplified local implementations to avoid import loops
# Simplified Tensor class for development
class Tensor:
"""Simplified tensor for spatial operations development."""
def __init__(self, data, requires_grad=False):
self.data = np.array(data, dtype=np.float32)
self.shape = self.data.shape
self.requires_grad = requires_grad
self.grad = None
def __repr__(self):
return f"Tensor(shape={self.shape}, data=\n{self.data})"
def __add__(self, other):
if isinstance(other, Tensor):
return Tensor(self.data + other.data)
return Tensor(self.data + other)
def __mul__(self, other):
if isinstance(other, Tensor):
return Tensor(self.data * other.data)
return Tensor(self.data * other)
def sum(self):
return Tensor(np.sum(self.data))
def mean(self):
return Tensor(np.mean(self.data))
# Create a simple Module base class for inheritance
class Module:
"""Simple base class for neural network modules."""
def __init__(self):
pass
def forward(self, x):
raise NotImplementedError("Subclasses must implement forward()")
def parameters(self):
"""Return list of parameters for this module."""
params = []
for attr_name in dir(self):
attr = getattr(self, attr_name)
if hasattr(attr, 'data') and hasattr(attr, 'requires_grad'):
params.append(attr)
return paramsIn [ ]:
#| export
class Conv2d(Module):
"""
2D Convolution layer for spatial feature extraction.
Implements convolution with explicit loops to demonstrate
computational complexity and memory access patterns.
Args:
in_channels: Number of input channels
out_channels: Number of output feature maps
kernel_size: Size of convolution kernel (int or tuple)
stride: Stride of convolution (default: 1)
padding: Zero-padding added to input (default: 0)
bias: Whether to add learnable bias (default: True)
"""
def __init__(self, in_channels, out_channels, kernel_size, stride=1, padding=0, bias=True):
"""
Initialize Conv2d layer with proper weight initialization.
TODO: Complete Conv2d initialization
APPROACH:
1. Store hyperparameters (channels, kernel_size, stride, padding)
2. Initialize weights using He initialization for ReLU compatibility
3. Initialize bias (if enabled) to zeros
4. Use proper shapes: weight (out_channels, in_channels, kernel_h, kernel_w)
WEIGHT INITIALIZATION:
- He init: std = sqrt(2 / (in_channels * kernel_h * kernel_w))
- This prevents vanishing/exploding gradients with ReLU
HINT: Convert kernel_size to tuple if it's an integer
"""
super().__init__()
### BEGIN SOLUTION
self.in_channels = in_channels
self.out_channels = out_channels
# Handle kernel_size as int or tuple
if isinstance(kernel_size, int):
self.kernel_size = (kernel_size, kernel_size)
else:
self.kernel_size = kernel_size
self.stride = stride
self.padding = padding
# He initialization for ReLU networks
kernel_h, kernel_w = self.kernel_size
fan_in = in_channels * kernel_h * kernel_w
std = np.sqrt(2.0 / fan_in)
# Weight shape: (out_channels, in_channels, kernel_h, kernel_w)
self.weight = Tensor(np.random.normal(0, std,
(out_channels, in_channels, kernel_h, kernel_w)))
# Bias initialization
if bias:
self.bias = Tensor(np.zeros(out_channels))
else:
self.bias = None
### END SOLUTION
def forward(self, x):
"""
Forward pass through Conv2d layer.
TODO: Implement convolution with explicit loops
APPROACH:
1. Extract input dimensions and validate
2. Calculate output dimensions
3. Apply padding if needed
4. Implement 6 nested loops for full convolution
5. Add bias if present
LOOP STRUCTURE:
for batch in range(batch_size):
for out_ch in range(out_channels):
for out_h in range(out_height):
for out_w in range(out_width):
for k_h in range(kernel_height):
for k_w in range(kernel_width):
for in_ch in range(in_channels):
# Accumulate: out += input * weight
EXAMPLE:
>>> conv = Conv2d(3, 16, kernel_size=3, padding=1)
>>> x = Tensor(np.random.randn(2, 3, 32, 32)) # batch=2, RGB, 32x32
>>> out = conv(x)
>>> print(out.shape) # Should be (2, 16, 32, 32)
HINTS:
- Handle padding by creating padded input array
- Watch array bounds in inner loops
- Accumulate products for each output position
"""
### BEGIN SOLUTION
# Input validation and shape extraction
if len(x.shape) != 4:
raise ValueError(f"Expected 4D input (batch, channels, height, width), got {x.shape}")
batch_size, in_channels, in_height, in_width = x.shape
out_channels = self.out_channels
kernel_h, kernel_w = self.kernel_size
# Calculate output dimensions
out_height = (in_height + 2 * self.padding - kernel_h) // self.stride + 1
out_width = (in_width + 2 * self.padding - kernel_w) // self.stride + 1
# Apply padding if needed
if self.padding > 0:
padded_input = np.pad(x.data,
((0, 0), (0, 0), (self.padding, self.padding), (self.padding, self.padding)),
mode='constant', constant_values=0)
else:
padded_input = x.data
# Initialize output
output = np.zeros((batch_size, out_channels, out_height, out_width))
# Explicit 6-nested loop convolution to show complexity
for b in range(batch_size):
for out_ch in range(out_channels):
for out_h in range(out_height):
for out_w in range(out_width):
# Calculate input region for this output position
in_h_start = out_h * self.stride
in_w_start = out_w * self.stride
# Accumulate convolution result
conv_sum = 0.0
for k_h in range(kernel_h):
for k_w in range(kernel_w):
for in_ch in range(in_channels):
# Get input and weight values
input_val = padded_input[b, in_ch,
in_h_start + k_h,
in_w_start + k_w]
weight_val = self.weight.data[out_ch, in_ch, k_h, k_w]
# Accumulate
conv_sum += input_val * weight_val
# Store result
output[b, out_ch, out_h, out_w] = conv_sum
# Add bias if present
if self.bias is not None:
# Broadcast bias across spatial dimensions
for out_ch in range(out_channels):
output[:, out_ch, :, :] += self.bias.data[out_ch]
return Tensor(output)
### END SOLUTION
def parameters(self):
"""Return trainable parameters."""
params = [self.weight]
if self.bias is not None:
params.append(self.bias)
return params
def __call__(self, x):
"""Enable model(x) syntax."""
return self.forward(x)In [ ]:
def test_unit_conv2d():
"""🔬 Test Conv2d implementation with multiple configurations."""
print("🔬 Unit Test: Conv2d...")
# Test 1: Basic convolution without padding
print(" Testing basic convolution...")
conv1 = Conv2d(in_channels=3, out_channels=16, kernel_size=3)
x1 = Tensor(np.random.randn(2, 3, 32, 32))
out1 = conv1(x1)
expected_h = (32 - 3) + 1 # 30
expected_w = (32 - 3) + 1 # 30
assert out1.shape == (2, 16, expected_h, expected_w), f"Expected (2, 16, 30, 30), got {out1.shape}"
# Test 2: Convolution with padding (same size)
print(" Testing convolution with padding...")
conv2 = Conv2d(in_channels=3, out_channels=8, kernel_size=3, padding=1)
x2 = Tensor(np.random.randn(1, 3, 28, 28))
out2 = conv2(x2)
# With padding=1, output should be same size as input
assert out2.shape == (1, 8, 28, 28), f"Expected (1, 8, 28, 28), got {out2.shape}"
# Test 3: Convolution with stride
print(" Testing convolution with stride...")
conv3 = Conv2d(in_channels=1, out_channels=4, kernel_size=3, stride=2)
x3 = Tensor(np.random.randn(1, 1, 16, 16))
out3 = conv3(x3)
expected_h = (16 - 3) // 2 + 1 # 7
expected_w = (16 - 3) // 2 + 1 # 7
assert out3.shape == (1, 4, expected_h, expected_w), f"Expected (1, 4, 7, 7), got {out3.shape}"
# Test 4: Parameter counting
print(" Testing parameter counting...")
conv4 = Conv2d(in_channels=64, out_channels=128, kernel_size=3, bias=True)
params = conv4.parameters()
# Weight: (128, 64, 3, 3) = 73,728 parameters
# Bias: (128,) = 128 parameters
# Total: 73,856 parameters
weight_params = 128 * 64 * 3 * 3
bias_params = 128
total_params = weight_params + bias_params
actual_weight_params = np.prod(conv4.weight.shape)
actual_bias_params = np.prod(conv4.bias.shape) if conv4.bias is not None else 0
actual_total = actual_weight_params + actual_bias_params
assert actual_total == total_params, f"Expected {total_params} parameters, got {actual_total}"
assert len(params) == 2, f"Expected 2 parameter tensors, got {len(params)}"
# Test 5: No bias configuration
print(" Testing no bias configuration...")
conv5 = Conv2d(in_channels=3, out_channels=16, kernel_size=5, bias=False)
params5 = conv5.parameters()
assert len(params5) == 1, f"Expected 1 parameter tensor (no bias), got {len(params5)}"
assert conv5.bias is None, "Bias should be None when bias=False"
print("✅ Conv2d works correctly!")
if __name__ == "__main__":
test_unit_conv2d()In [ ]:
#| export
class MaxPool2d(Module):
"""
2D Max Pooling layer for spatial dimension reduction.
Applies maximum operation over spatial windows, preserving
the strongest activations while reducing computational load.
Args:
kernel_size: Size of pooling window (int or tuple)
stride: Stride of pooling operation (default: same as kernel_size)
padding: Zero-padding added to input (default: 0)
"""
def __init__(self, kernel_size, stride=None, padding=0):
"""
Initialize MaxPool2d layer.
TODO: Store pooling parameters
APPROACH:
1. Convert kernel_size to tuple if needed
2. Set stride to kernel_size if not provided (non-overlapping)
3. Store padding parameter
HINT: Default stride equals kernel_size for non-overlapping windows
"""
super().__init__()
### BEGIN SOLUTION
# Handle kernel_size as int or tuple
if isinstance(kernel_size, int):
self.kernel_size = (kernel_size, kernel_size)
else:
self.kernel_size = kernel_size
# Default stride equals kernel_size (non-overlapping)
if stride is None:
self.stride = self.kernel_size[0]
else:
self.stride = stride
self.padding = padding
### END SOLUTION
def forward(self, x):
"""
Forward pass through MaxPool2d layer.
TODO: Implement max pooling with explicit loops
APPROACH:
1. Extract input dimensions
2. Calculate output dimensions
3. Apply padding if needed
4. Implement nested loops for pooling windows
5. Find maximum value in each window
LOOP STRUCTURE:
for batch in range(batch_size):
for channel in range(channels):
for out_h in range(out_height):
for out_w in range(out_width):
# Find max in window [in_h:in_h+k_h, in_w:in_w+k_w]
max_val = -infinity
for k_h in range(kernel_height):
for k_w in range(kernel_width):
max_val = max(max_val, input[...])
EXAMPLE:
>>> pool = MaxPool2d(kernel_size=2, stride=2)
>>> x = Tensor(np.random.randn(1, 3, 8, 8))
>>> out = pool(x)
>>> print(out.shape) # Should be (1, 3, 4, 4)
HINTS:
- Initialize max_val to negative infinity
- Handle stride correctly when accessing input
- No parameters to update (pooling has no weights)
"""
### BEGIN SOLUTION
# Input validation and shape extraction
if len(x.shape) != 4:
raise ValueError(f"Expected 4D input (batch, channels, height, width), got {x.shape}")
batch_size, channels, in_height, in_width = x.shape
kernel_h, kernel_w = self.kernel_size
# Calculate output dimensions
out_height = (in_height + 2 * self.padding - kernel_h) // self.stride + 1
out_width = (in_width + 2 * self.padding - kernel_w) // self.stride + 1
# Apply padding if needed
if self.padding > 0:
padded_input = np.pad(x.data,
((0, 0), (0, 0), (self.padding, self.padding), (self.padding, self.padding)),
mode='constant', constant_values=-np.inf)
else:
padded_input = x.data
# Initialize output
output = np.zeros((batch_size, channels, out_height, out_width))
# Explicit nested loop max pooling
for b in range(batch_size):
for c in range(channels):
for out_h in range(out_height):
for out_w in range(out_width):
# Calculate input region for this output position
in_h_start = out_h * self.stride
in_w_start = out_w * self.stride
# Find maximum in window
max_val = -np.inf
for k_h in range(kernel_h):
for k_w in range(kernel_w):
input_val = padded_input[b, c,
in_h_start + k_h,
in_w_start + k_w]
max_val = max(max_val, input_val)
# Store result
output[b, c, out_h, out_w] = max_val
return Tensor(output)
### END SOLUTION
def parameters(self):
"""Return empty list (pooling has no parameters)."""
return []
def __call__(self, x):
"""Enable model(x) syntax."""
return self.forward(x)In [ ]:
#| export
class AvgPool2d(Module):
"""
2D Average Pooling layer for spatial dimension reduction.
Applies average operation over spatial windows, smoothing
features while reducing computational load.
Args:
kernel_size: Size of pooling window (int or tuple)
stride: Stride of pooling operation (default: same as kernel_size)
padding: Zero-padding added to input (default: 0)
"""
def __init__(self, kernel_size, stride=None, padding=0):
"""
Initialize AvgPool2d layer.
TODO: Store pooling parameters (same as MaxPool2d)
APPROACH:
1. Convert kernel_size to tuple if needed
2. Set stride to kernel_size if not provided
3. Store padding parameter
"""
super().__init__()
### BEGIN SOLUTION
# Handle kernel_size as int or tuple
if isinstance(kernel_size, int):
self.kernel_size = (kernel_size, kernel_size)
else:
self.kernel_size = kernel_size
# Default stride equals kernel_size (non-overlapping)
if stride is None:
self.stride = self.kernel_size[0]
else:
self.stride = stride
self.padding = padding
### END SOLUTION
def forward(self, x):
"""
Forward pass through AvgPool2d layer.
TODO: Implement average pooling with explicit loops
APPROACH:
1. Similar structure to MaxPool2d
2. Instead of max, compute average of window
3. Divide sum by window area for true average
LOOP STRUCTURE:
for batch in range(batch_size):
for channel in range(channels):
for out_h in range(out_height):
for out_w in range(out_width):
# Compute average in window
window_sum = 0
for k_h in range(kernel_height):
for k_w in range(kernel_width):
window_sum += input[...]
avg_val = window_sum / (kernel_height * kernel_width)
HINT: Remember to divide by window area to get true average
"""
### BEGIN SOLUTION
# Input validation and shape extraction
if len(x.shape) != 4:
raise ValueError(f"Expected 4D input (batch, channels, height, width), got {x.shape}")
batch_size, channels, in_height, in_width = x.shape
kernel_h, kernel_w = self.kernel_size
# Calculate output dimensions
out_height = (in_height + 2 * self.padding - kernel_h) // self.stride + 1
out_width = (in_width + 2 * self.padding - kernel_w) // self.stride + 1
# Apply padding if needed
if self.padding > 0:
padded_input = np.pad(x.data,
((0, 0), (0, 0), (self.padding, self.padding), (self.padding, self.padding)),
mode='constant', constant_values=0)
else:
padded_input = x.data
# Initialize output
output = np.zeros((batch_size, channels, out_height, out_width))
# Explicit nested loop average pooling
for b in range(batch_size):
for c in range(channels):
for out_h in range(out_height):
for out_w in range(out_width):
# Calculate input region for this output position
in_h_start = out_h * self.stride
in_w_start = out_w * self.stride
# Compute sum in window
window_sum = 0.0
for k_h in range(kernel_h):
for k_w in range(kernel_w):
input_val = padded_input[b, c,
in_h_start + k_h,
in_w_start + k_w]
window_sum += input_val
# Compute average
avg_val = window_sum / (kernel_h * kernel_w)
# Store result
output[b, c, out_h, out_w] = avg_val
return Tensor(output)
### END SOLUTION
def parameters(self):
"""Return empty list (pooling has no parameters)."""
return []
def __call__(self, x):
"""Enable model(x) syntax."""
return self.forward(x)In [ ]:
def test_unit_pooling():
"""🔬 Test MaxPool2d and AvgPool2d implementations."""
print("🔬 Unit Test: Pooling Operations...")
# Test 1: MaxPool2d basic functionality
print(" Testing MaxPool2d...")
maxpool = MaxPool2d(kernel_size=2, stride=2)
x1 = Tensor(np.random.randn(1, 3, 8, 8))
out1 = maxpool(x1)
expected_shape = (1, 3, 4, 4) # 8/2 = 4
assert out1.shape == expected_shape, f"MaxPool expected {expected_shape}, got {out1.shape}"
# Test 2: AvgPool2d basic functionality
print(" Testing AvgPool2d...")
avgpool = AvgPool2d(kernel_size=2, stride=2)
x2 = Tensor(np.random.randn(2, 16, 16, 16))
out2 = avgpool(x2)
expected_shape = (2, 16, 8, 8) # 16/2 = 8
assert out2.shape == expected_shape, f"AvgPool expected {expected_shape}, got {out2.shape}"
# Test 3: MaxPool vs AvgPool on known data
print(" Testing max vs avg behavior...")
# Create simple test case with known values
test_data = np.array([[[[1, 2, 3, 4],
[5, 6, 7, 8],
[9, 10, 11, 12],
[13, 14, 15, 16]]]], dtype=np.float32)
x3 = Tensor(test_data)
maxpool_test = MaxPool2d(kernel_size=2, stride=2)
avgpool_test = AvgPool2d(kernel_size=2, stride=2)
max_out = maxpool_test(x3)
avg_out = avgpool_test(x3)
# For 2x2 windows:
# Top-left: max([1,2,5,6]) = 6, avg = 3.5
# Top-right: max([3,4,7,8]) = 8, avg = 5.5
# Bottom-left: max([9,10,13,14]) = 14, avg = 11.5
# Bottom-right: max([11,12,15,16]) = 16, avg = 13.5
expected_max = np.array([[[[6, 8], [14, 16]]]])
expected_avg = np.array([[[[3.5, 5.5], [11.5, 13.5]]]])
assert np.allclose(max_out.data, expected_max), f"MaxPool values incorrect: {max_out.data} vs {expected_max}"
assert np.allclose(avg_out.data, expected_avg), f"AvgPool values incorrect: {avg_out.data} vs {expected_avg}"
# Test 4: Overlapping pooling (stride < kernel_size)
print(" Testing overlapping pooling...")
overlap_pool = MaxPool2d(kernel_size=3, stride=1)
x4 = Tensor(np.random.randn(1, 1, 5, 5))
out4 = overlap_pool(x4)
# Output: (5-3)/1 + 1 = 3
expected_shape = (1, 1, 3, 3)
assert out4.shape == expected_shape, f"Overlapping pool expected {expected_shape}, got {out4.shape}"
# Test 5: No parameters in pooling layers
print(" Testing parameter counts...")
assert len(maxpool.parameters()) == 0, "MaxPool should have no parameters"
assert len(avgpool.parameters()) == 0, "AvgPool should have no parameters"
print("✅ Pooling operations work correctly!")
if __name__ == "__main__":
test_unit_pooling()In [ ]:
def analyze_convolution_complexity():
"""📊 Analyze convolution computational complexity across different configurations."""
print("📊 Analyzing Convolution Complexity...")
# Test configurations optimized for educational demonstration (smaller sizes)
configs = [
{"input": (1, 3, 16, 16), "conv": (8, 3, 3), "name": "Small (16×16)"},
{"input": (1, 3, 24, 24), "conv": (12, 3, 3), "name": "Medium (24×24)"},
{"input": (1, 3, 32, 32), "conv": (16, 3, 3), "name": "Large (32×32)"},
{"input": (1, 3, 16, 16), "conv": (8, 3, 5), "name": "Large Kernel (5×5)"},
]
print(f"{'Configuration':<20} {'FLOPs':<15} {'Memory (MB)':<12} {'Time (ms)':<10}")
print("-" * 70)
for config in configs:
# Create convolution layer
in_ch = config["input"][1]
out_ch, k_size = config["conv"][0], config["conv"][1]
conv = Conv2d(in_ch, out_ch, kernel_size=k_size, padding=k_size//2)
# Create input tensor
x = Tensor(np.random.randn(*config["input"]))
# Calculate theoretical FLOPs
batch, in_channels, h, w = config["input"]
out_channels, kernel_size = config["conv"][0], config["conv"][1]
# Each output element requires in_channels * kernel_size² multiply-adds
flops_per_output = in_channels * kernel_size * kernel_size * 2 # 2 for MAC
total_outputs = batch * out_channels * h * w # Assuming same size with padding
total_flops = flops_per_output * total_outputs
# Measure memory usage
input_memory = np.prod(config["input"]) * 4 # float32 = 4 bytes
weight_memory = out_channels * in_channels * kernel_size * kernel_size * 4
output_memory = batch * out_channels * h * w * 4
total_memory = (input_memory + weight_memory + output_memory) / (1024 * 1024) # MB
# Measure execution time
start_time = time.time()
_ = conv(x)
end_time = time.time()
exec_time = (end_time - start_time) * 1000 # ms
print(f"{config['name']:<20} {total_flops:<15,} {total_memory:<12.2f} {exec_time:<10.2f}")
print("\n💡 Key Insights:")
print("🔸 FLOPs scale as O(H×W×C_in×C_out×K²) - quadratic in spatial and kernel size")
print("🔸 Memory scales linearly with spatial dimensions and channels")
print("🔸 Large kernels dramatically increase computational cost")
print("🚀 This motivates depthwise separable convolutions and attention mechanisms")
# Analysis will be called in main executionIn [ ]:
def analyze_pooling_effects():
"""📊 Analyze pooling's impact on spatial dimensions and features."""
print("\n📊 Analyzing Pooling Effects...")
# Create sample input with spatial structure
# Simple edge pattern that pooling should preserve differently
pattern = np.zeros((1, 1, 8, 8))
pattern[0, 0, :, 3:5] = 1.0 # Vertical edge
pattern[0, 0, 3:5, :] = 1.0 # Horizontal edge
x = Tensor(pattern)
print("Original 8×8 pattern:")
print(x.data[0, 0])
# Test different pooling strategies
pools = [
(MaxPool2d(2, stride=2), "MaxPool 2×2"),
(AvgPool2d(2, stride=2), "AvgPool 2×2"),
(MaxPool2d(4, stride=4), "MaxPool 4×4"),
(AvgPool2d(4, stride=4), "AvgPool 4×4"),
]
print(f"\n{'Operation':<15} {'Output Shape':<15} {'Feature Preservation'}")
print("-" * 60)
for pool_op, name in pools:
result = pool_op(x)
# Measure how much of the original pattern is preserved
preservation = np.sum(result.data > 0.1) / np.prod(result.shape)
print(f"{name:<15} {str(result.shape):<15} {preservation:<.2%}")
print(f" Output:")
print(f" {result.data[0, 0]}")
print()
print("💡 Key Insights:")
print("🔸 MaxPool preserves sharp features better (edge detection)")
print("🔸 AvgPool smooths features (noise reduction)")
print("🔸 Larger pooling windows lose more spatial detail")
print("🚀 Choice depends on task: classification vs detection vs segmentation")
# Analysis will be called in main executionIn [ ]:
#| export
class SimpleCNN(Module):
"""
Simple CNN demonstrating spatial operations integration.
Architecture:
- Conv2d(3→16, 3×3) + ReLU + MaxPool(2×2)
- Conv2d(16→32, 3×3) + ReLU + MaxPool(2×2)
- Flatten + Linear(features→num_classes)
"""
def __init__(self, num_classes=10):
"""
Initialize SimpleCNN.
TODO: Build CNN architecture with spatial and dense layers
APPROACH:
1. Conv layer 1: 3 → 16 channels, 3×3 kernel, padding=1
2. Pool layer 1: 2×2 max pooling
3. Conv layer 2: 16 → 32 channels, 3×3 kernel, padding=1
4. Pool layer 2: 2×2 max pooling
5. Calculate flattened size and add final linear layer
HINT: For 32×32 input → 32→16→8→4 spatial reduction
Final feature size: 32 channels × 4×4 = 512 features
"""
super().__init__()
### BEGIN SOLUTION
# Convolutional layers
self.conv1 = Conv2d(in_channels=3, out_channels=16, kernel_size=3, padding=1)
self.pool1 = MaxPool2d(kernel_size=2, stride=2)
self.conv2 = Conv2d(in_channels=16, out_channels=32, kernel_size=3, padding=1)
self.pool2 = MaxPool2d(kernel_size=2, stride=2)
# Calculate flattened size
# Input: 32×32 → Conv1+Pool1: 16×16 → Conv2+Pool2: 8×8
# Wait, let's recalculate: 32×32 → Pool1: 16×16 → Pool2: 8×8
# Final: 32 channels × 8×8 = 2048 features
self.flattened_size = 32 * 8 * 8
# Import Linear layer (we'll implement a simple version)
# For now, we'll use a placeholder that we can replace
# This represents the final classification layer
self.num_classes = num_classes
self.flattened_size = 32 * 8 * 8 # Will be used when we add Linear layer
### END SOLUTION
def forward(self, x):
"""
Forward pass through SimpleCNN.
TODO: Implement CNN forward pass
APPROACH:
1. Apply conv1 → ReLU → pool1
2. Apply conv2 → ReLU → pool2
3. Flatten spatial dimensions
4. Apply final linear layer (when available)
For now, return features before final linear layer
since we haven't imported Linear from layers module yet.
"""
### BEGIN SOLUTION
# First conv block
x = self.conv1(x)
x = self.relu(x) # ReLU activation
x = self.pool1(x)
# Second conv block
x = self.conv2(x)
x = self.relu(x) # ReLU activation
x = self.pool2(x)
# Flatten for classification (reshape to 2D)
batch_size = x.shape[0]
x_flat = x.data.reshape(batch_size, -1)
# Return flattened features
# In a complete implementation, this would go through a Linear layer
return Tensor(x_flat)
### END SOLUTION
def relu(self, x):
"""Simple ReLU implementation for CNN."""
return Tensor(np.maximum(0, x.data))
def parameters(self):
"""Return all trainable parameters."""
params = []
params.extend(self.conv1.parameters())
params.extend(self.conv2.parameters())
# Linear layer parameters would be added here
return params
def __call__(self, x):
"""Enable model(x) syntax."""
return self.forward(x)In [ ]:
def test_unit_simple_cnn():
"""🔬 Test SimpleCNN integration with spatial operations."""
print("🔬 Unit Test: SimpleCNN Integration...")
# Test 1: Forward pass with CIFAR-10 sized input
print(" Testing forward pass...")
model = SimpleCNN(num_classes=10)
x = Tensor(np.random.randn(2, 3, 32, 32)) # Batch of 2, RGB, 32×32
features = model(x)
# Expected: 2 samples, 32 channels × 8×8 spatial = 2048 features
expected_shape = (2, 2048)
assert features.shape == expected_shape, f"Expected {expected_shape}, got {features.shape}"
# Test 2: Parameter counting
print(" Testing parameter counting...")
params = model.parameters()
# Conv1: (16, 3, 3, 3) + bias (16,) = 432 + 16 = 448
# Conv2: (32, 16, 3, 3) + bias (32,) = 4608 + 32 = 4640
# Total: 448 + 4640 = 5088 parameters
conv1_params = 16 * 3 * 3 * 3 + 16 # weights + bias
conv2_params = 32 * 16 * 3 * 3 + 32 # weights + bias
expected_total = conv1_params + conv2_params
actual_total = sum(np.prod(p.shape) for p in params)
assert actual_total == expected_total, f"Expected {expected_total} parameters, got {actual_total}"
# Test 3: Different input sizes
print(" Testing different input sizes...")
# Test with different spatial dimensions
x_small = Tensor(np.random.randn(1, 3, 16, 16))
features_small = model(x_small)
# 16×16 → 8×8 → 4×4, so 32 × 4×4 = 512 features
expected_small = (1, 512)
assert features_small.shape == expected_small, f"Expected {expected_small}, got {features_small.shape}"
# Test 4: Batch processing
print(" Testing batch processing...")
x_batch = Tensor(np.random.randn(8, 3, 32, 32))
features_batch = model(x_batch)
expected_batch = (8, 2048)
assert features_batch.shape == expected_batch, f"Expected {expected_batch}, got {features_batch.shape}"
print("✅ SimpleCNN integration works correctly!")
if __name__ == "__main__":
test_unit_simple_cnn()In [ ]:
def test_module():
"""
Comprehensive test of entire spatial module functionality.
This final test runs before module summary to ensure:
- All unit tests pass
- Functions work together correctly
- Module is ready for integration with TinyTorch
"""
print("🧪 RUNNING MODULE INTEGRATION TEST")
print("=" * 50)
# Run all unit tests
print("Running unit tests...")
test_unit_conv2d()
test_unit_pooling()
test_unit_simple_cnn()
print("\nRunning integration scenarios...")
# Test realistic CNN workflow
print("🔬 Integration Test: Complete CNN pipeline...")
# Create a mini CNN for CIFAR-10
conv1 = Conv2d(3, 8, kernel_size=3, padding=1)
pool1 = MaxPool2d(2, stride=2)
conv2 = Conv2d(8, 16, kernel_size=3, padding=1)
pool2 = AvgPool2d(2, stride=2)
# Process batch of images
batch_images = Tensor(np.random.randn(4, 3, 32, 32))
# Forward pass through spatial layers
x = conv1(batch_images) # (4, 8, 32, 32)
x = pool1(x) # (4, 8, 16, 16)
x = conv2(x) # (4, 16, 16, 16)
features = pool2(x) # (4, 16, 8, 8)
# Validate shapes at each step
assert x.shape[0] == 4, f"Batch size should be preserved, got {x.shape[0]}"
assert features.shape == (4, 16, 8, 8), f"Final features shape incorrect: {features.shape}"
# Test parameter collection across all layers
all_params = []
all_params.extend(conv1.parameters())
all_params.extend(conv2.parameters())
# Pooling has no parameters
assert len(pool1.parameters()) == 0
assert len(pool2.parameters()) == 0
# Verify we have the right number of parameter tensors
assert len(all_params) == 4, f"Expected 4 parameter tensors (2 conv × 2 each), got {len(all_params)}"
print("✅ Complete CNN pipeline works!")
# Test memory efficiency comparison
print("🔬 Integration Test: Memory efficiency analysis...")
# Compare different pooling strategies (reduced size for faster execution)
input_data = Tensor(np.random.randn(1, 16, 32, 32))
# No pooling: maintain spatial size
conv_only = Conv2d(16, 32, kernel_size=3, padding=1)
no_pool_out = conv_only(input_data)
no_pool_size = np.prod(no_pool_out.shape) * 4 # float32 bytes
# With pooling: reduce spatial size
conv_with_pool = Conv2d(16, 32, kernel_size=3, padding=1)
pool = MaxPool2d(2, stride=2)
pool_out = pool(conv_with_pool(input_data))
pool_size = np.prod(pool_out.shape) * 4 # float32 bytes
memory_reduction = no_pool_size / pool_size
assert memory_reduction == 4.0, f"2×2 pooling should give 4× memory reduction, got {memory_reduction:.1f}×"
print(f" Memory reduction with pooling: {memory_reduction:.1f}×")
print("✅ Memory efficiency analysis complete!")
print("\n" + "=" * 50)
print("🎉 ALL TESTS PASSED! Module ready for export.")
print("Run: tito module complete 09")In [ ]:
# Run comprehensive module test
if __name__ == "__main__":
test_module()