202 KiB
202 KiB
Regular NNs don’t scale well to full images
As an example, consider
an image of size 32\times 32\times 3 (32 wide, 32 high, 3 color channels), so a
single fully-connected neuron in a first hidden layer of a regular
Neural Network would have 32\times 32\times 3 = 3072 weights. This amount still
seems manageable, but clearly this fully-connected structure does not
scale to larger images. For example, an image of more respectable
size, say 200\times 200\times 3, would lead to neurons that have
200\times 200\times 3 = 120,000 weights.
We could have several such neurons, and the parameters would add up quickly! Clearly, this full connectivity is wasteful and the huge number of parameters would quickly lead to possible overfitting.
Figure 1: A regular 3-layer Neural Network.
3D volumes of neurons
Convolutional Neural Networks take advantage of the fact that the input consists of images and they constrain the architecture in a more sensible way.
In particular, unlike a regular Neural Network, the layers of a CNN have neurons arranged in 3 dimensions: width, height, depth. (Note that the word depth here refers to the third dimension of an activation volume, not to the depth of a full Neural Network, which can refer to the total number of layers in a network.)
To understand it better, the above example of an image
with an input volume of
activations has dimensions 32\times 32\times 3 (width, height,
depth respectively).
The neurons in a layer will
only be connected to a small region of the layer before it, instead of
all of the neurons in a fully-connected manner. Moreover, the final
output layer could for this specific image have dimensions 1\times 1 \times 10,
because by the
end of the CNN architecture we will reduce the full image into a
single vector of class scores, arranged along the depth
dimension.
Figure 1: A CNN arranges its neurons in three dimensions (width, height, depth), as visualized in one of the layers. Every layer of a CNN transforms the 3D input volume to a 3D output volume of neuron activations. In this example, the red input layer holds the image, so its width and height would be the dimensions of the image, and the depth would be 3 (Red, Green, Blue channels).
In [1]:
%matplotlib inline
# import necessary packages
import numpy as np
import matplotlib.pyplot as plt
from sklearn import datasets
# ensure the same random numbers appear every time
np.random.seed(0)
# display images in notebook
%matplotlib inline
plt.rcParams['figure.figsize'] = (12,12)
# download MNIST dataset
digits = datasets.load_digits()
# define inputs and labels
inputs = digits.images
labels = digits.target
# RGB images have a depth of 3
# our images are grayscale so they should have a depth of 1
inputs = inputs[:,:,:,np.newaxis]
print("inputs = (n_inputs, pixel_width, pixel_height, depth) = " + str(inputs.shape))
print("labels = (n_inputs) = " + str(labels.shape))
# choose some random images to display
n_inputs = len(inputs)
indices = np.arange(n_inputs)
random_indices = np.random.choice(indices, size=5)
for i, image in enumerate(digits.images[random_indices]):
plt.subplot(1, 5, i+1)
plt.axis('off')
plt.imshow(image, cmap=plt.cm.gray_r, interpolation='nearest')
plt.title("Label: %d" % digits.target[random_indices[i]])
plt.show()In [2]:
from tensorflow.keras import datasets, layers, models
from tensorflow.keras.layers import Input
from tensorflow.keras.models import Sequential #This allows appending layers to existing models
from tensorflow.keras.layers import Dense #This allows defining the characteristics of a particular layer
from tensorflow.keras import optimizers #This allows using whichever optimiser we want (sgd,adam,RMSprop)
from tensorflow.keras import regularizers #This allows using whichever regularizer we want (l1,l2,l1_l2)
from tensorflow.keras.utils import to_categorical #This allows using categorical cross entropy as the cost function
#from tensorflow.keras import Conv2D
#from tensorflow.keras import MaxPooling2D
#from tensorflow.keras import Flatten
from sklearn.model_selection import train_test_split
# representation of labels
labels = to_categorical(labels)
# split into train and test data
# one-liner from scikit-learn library
train_size = 0.8
test_size = 1 - train_size
X_train, X_test, Y_train, Y_test = train_test_split(inputs, labels, train_size=train_size,
test_size=test_size)In [3]:
def create_convolutional_neural_network_keras(input_shape, receptive_field,
n_filters, n_neurons_connected, n_categories,
eta, lmbd):
model = Sequential()
model.add(layers.Conv2D(n_filters, (receptive_field, receptive_field), input_shape=input_shape, padding='same',
activation='relu', kernel_regularizer=regularizers.l2(lmbd)))
model.add(layers.MaxPooling2D(pool_size=(2, 2)))
model.add(layers.Flatten())
model.add(layers.Dense(n_neurons_connected, activation='relu', kernel_regularizer=regularizers.l2(lmbd)))
model.add(layers.Dense(n_categories, activation='softmax', kernel_regularizer=regularizers.l2(lmbd)))
sgd = optimizers.SGD(learning_rate=eta)
model.compile(loss='categorical_crossentropy', optimizer=sgd, metrics=['accuracy'])
return model
epochs = 100
batch_size = 100
input_shape = X_train.shape[1:4]
receptive_field = 3
n_filters = 10
n_neurons_connected = 50
n_categories = 10
eta_vals = np.logspace(-5, 1, 7)
lmbd_vals = np.logspace(-5, 1, 7)In [4]:
CNN_keras = np.zeros((len(eta_vals), len(lmbd_vals)), dtype=object)
for i, eta in enumerate(eta_vals):
for j, lmbd in enumerate(lmbd_vals):
CNN = create_convolutional_neural_network_keras(input_shape, receptive_field,
n_filters, n_neurons_connected, n_categories,
eta, lmbd)
CNN.fit(X_train, Y_train, epochs=epochs, batch_size=batch_size, verbose=0)
scores = CNN.evaluate(X_test, Y_test)
CNN_keras[i][j] = CNN
print("Learning rate = ", eta)
print("Lambda = ", lmbd)
print("Test accuracy: %.3f" % scores[1])
print()In [5]:
# visual representation of grid search
# uses seaborn heatmap, could probably do this in matplotlib
import seaborn as sns
sns.set()
train_accuracy = np.zeros((len(eta_vals), len(lmbd_vals)))
test_accuracy = np.zeros((len(eta_vals), len(lmbd_vals)))
for i in range(len(eta_vals)):
for j in range(len(lmbd_vals)):
CNN = CNN_keras[i][j]
train_accuracy[i][j] = CNN.evaluate(X_train, Y_train)[1]
test_accuracy[i][j] = CNN.evaluate(X_test, Y_test)[1]
fig, ax = plt.subplots(figsize = (10, 10))
sns.heatmap(train_accuracy, annot=True, ax=ax, cmap="viridis")
ax.set_title("Training Accuracy")
ax.set_ylabel("$\eta$")
ax.set_xlabel("$\lambda$")
plt.show()
fig, ax = plt.subplots(figsize = (10, 10))
sns.heatmap(test_accuracy, annot=True, ax=ax, cmap="viridis")
ax.set_title("Test Accuracy")
ax.set_ylabel("$\eta$")
ax.set_xlabel("$\lambda$")
plt.show()In [6]:
import tensorflow as tf
from tensorflow.keras import datasets, layers, models
import matplotlib.pyplot as plt
# We import the data set
(train_images, train_labels), (test_images, test_labels) = datasets.cifar10.load_data()
# Normalize pixel values to be between 0 and 1 by dividing by 255.
train_images, test_images = train_images / 255.0, test_images / 255.0In [7]:
class_names = ['airplane', 'automobile', 'bird', 'cat', 'deer',
'dog', 'frog', 'horse', 'ship', 'truck']
plt.figure(figsize=(10,10))
for i in range(25):
plt.subplot(5,5,i+1)
plt.xticks([])
plt.yticks([])
plt.grid(False)
plt.imshow(train_images[i], cmap=plt.cm.binary)
# The CIFAR labels happen to be arrays,
# which is why you need the extra index
plt.xlabel(class_names[train_labels[i][0]])
plt.show()In [8]:
model = models.Sequential()
model.add(layers.Conv2D(32, (3, 3), activation='relu', input_shape=(32, 32, 3)))
model.add(layers.MaxPooling2D((2, 2)))
model.add(layers.Conv2D(64, (3, 3), activation='relu'))
model.add(layers.MaxPooling2D((2, 2)))
model.add(layers.Conv2D(64, (3, 3), activation='relu'))
# Let's display the architecture of our model so far.
model.summary()In [9]:
model.add(layers.Flatten())
model.add(layers.Dense(64, activation='relu'))
model.add(layers.Dense(10))
# Here's the complete architecture of our model
model.summary()In [10]:
model.compile(optimizer='adam',
loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy'])
history = model.fit(train_images, train_labels, epochs=10,
validation_data=(test_images, test_labels))In [11]:
plt.plot(history.history['accuracy'], label='accuracy')
plt.plot(history.history['val_accuracy'], label = 'val_accuracy')
plt.xlabel('Epoch')
plt.ylabel('Accuracy')
plt.ylim([0.5, 1])
plt.legend(loc='lower right')
test_loss, test_acc = model.evaluate(test_images, test_labels, verbose=2)
print(test_acc)In [12]:
import autograd.numpy as np
class Scheduler:
"""
Abstract class for Schedulers
"""
def __init__(self, eta):
self.eta = eta
# should be overwritten
def update_change(self, gradient):
raise NotImplementedError
# overwritten if needed
def reset(self):
pass
class Constant(Scheduler):
def __init__(self, eta):
super().__init__(eta)
def update_change(self, gradient):
return self.eta * gradient
def reset(self):
pass
class Momentum(Scheduler):
def __init__(self, eta: float, momentum: float):
super().__init__(eta)
self.momentum = momentum
self.change = 0
def update_change(self, gradient):
self.change = self.momentum * self.change + self.eta * gradient
return self.change
def reset(self):
pass
class Adagrad(Scheduler):
def __init__(self, eta):
super().__init__(eta)
self.G_t = None
def update_change(self, gradient):
delta = 1e-8 # avoid division ny zero
if self.G_t is None:
self.G_t = np.zeros((gradient.shape[0], gradient.shape[0]))
self.G_t += gradient @ gradient.T
G_t_inverse = 1 / (
delta + np.sqrt(np.reshape(np.diagonal(self.G_t), (self.G_t.shape[0], 1)))
)
return self.eta * gradient * G_t_inverse
def reset(self):
self.G_t = None
class AdagradMomentum(Scheduler):
def __init__(self, eta, momentum):
super().__init__(eta)
self.G_t = None
self.momentum = momentum
self.change = 0
def update_change(self, gradient):
delta = 1e-8 # avoid division ny zero
if self.G_t is None:
self.G_t = np.zeros((gradient.shape[0], gradient.shape[0]))
self.G_t += gradient @ gradient.T
G_t_inverse = 1 / (
delta + np.sqrt(np.reshape(np.diagonal(self.G_t), (self.G_t.shape[0], 1)))
)
self.change = self.change * self.momentum + self.eta * gradient * G_t_inverse
return self.change
def reset(self):
self.G_t = None
class RMS_prop(Scheduler):
def __init__(self, eta, rho):
super().__init__(eta)
self.rho = rho
self.second = 0.0
def update_change(self, gradient):
delta = 1e-8 # avoid division ny zero
self.second = self.rho * self.second + (1 - self.rho) * gradient * gradient
return self.eta * gradient / (np.sqrt(self.second + delta))
def reset(self):
self.second = 0.0
class Adam(Scheduler):
def __init__(self, eta, rho, rho2):
super().__init__(eta)
self.rho = rho
self.rho2 = rho2
self.moment = 0
self.second = 0
self.n_epochs = 1
def update_change(self, gradient):
delta = 1e-8 # avoid division ny zero
self.moment = self.rho * self.moment + (1 - self.rho) * gradient
self.second = self.rho2 * self.second + (1 - self.rho2) * gradient * gradient
moment_corrected = self.moment / (1 - self.rho**self.n_epochs)
second_corrected = self.second / (1 - self.rho2**self.n_epochs)
return self.eta * moment_corrected / (np.sqrt(second_corrected + delta))
def reset(self):
self.n_epochs += 1
self.moment = 0
self.second = 0In [13]:
momentum_scheduler = Momentum(eta=1e-3, momentum=0.9)
adam_scheduler = Adam(eta=1e-3, rho=0.9, rho2=0.999)In [14]:
weights = np.ones((3,3))
print(f"Before scheduler:\n{weights=}")
epochs = 10
for e in range(epochs):
gradient = np.random.rand(3, 3)
change = adam_scheduler.update_change(gradient)
weights = weights - change
adam_scheduler.reset()
print(f"\nAfter scheduler:\n{weights=}")In [15]:
def CostOLS(target):
"""
Return OLS function valued only at X, so
that it may be easily differentiated
"""
def func(X):
return (1.0 / target.shape[0]) * np.sum((target - X) ** 2)
return func
def CostLogReg(target):
"""
Return Logistic Regression cost function
valued only at X, so that it may be easily differentiated
"""
def func(X):
return -(1.0 / target.shape[0]) * np.sum(
(target * np.log(X + 10e-10)) + ((1 - target) * np.log(1 - X + 10e-10))
)
return func
def CostCrossEntropy(target):
"""
Return cross entropy cost function valued only at X, so
that it may be easily differentiated
"""
def func(X):
return -(1.0 / target.size) * np.sum(target * np.log(X + 10e-10))
return funcIn [16]:
from autograd import grad
target = np.array([[1, 2, 3]]).T
a = np.array([[4, 5, 6]]).T
cost_func = CostCrossEntropy
cost_func_derivative = grad(cost_func(target))
valued_at_a = cost_func_derivative(a)
print(f"Derivative of cost function {cost_func.__name__} valued at a:\n{valued_at_a}")In [17]:
import autograd.numpy as np
from autograd import elementwise_grad
def identity(X):
return X
def sigmoid(X):
try:
return 1.0 / (1 + np.exp(-X))
except FloatingPointError:
return np.where(X > np.zeros(X.shape), np.ones(X.shape), np.zeros(X.shape))
def softmax(X):
X = X - np.max(X, axis=-1, keepdims=True)
delta = 10e-10
return np.exp(X) / (np.sum(np.exp(X), axis=-1, keepdims=True) + delta)
def RELU(X):
return np.where(X > np.zeros(X.shape), X, np.zeros(X.shape))
def LRELU(X):
delta = 10e-4
return np.where(X > np.zeros(X.shape), X, delta * X)
def derivate(func):
if func.__name__ == "RELU":
def func(X):
return np.where(X > 0, 1, 0)
return func
elif func.__name__ == "LRELU":
def func(X):
delta = 10e-4
return np.where(X > 0, 1, delta)
return func
else:
return elementwise_grad(func)In [18]:
z = np.array([[4, 5, 6]]).T
print(f"Input to activation function:\n{z}")
act_func = sigmoid
a = act_func(z)
print(f"\nOutput from {act_func.__name__} activation function:\n{a}")
act_func_derivative = derivate(act_func)
valued_at_z = act_func_derivative(a)
print(f"\nDerivative of {act_func.__name__} activation function valued at z:\n{valued_at_z}")In [19]:
import numpy as np
def padding(image, kernel):
# calculate r and c
r = (kernel.shape[0] // 2) * 2
c = (kernel.shape[1] // 2) * 2
# padded image dimensions
padded_height = image.shape[0] + r
padded_width = image.shape[1] + c
# for more readable code
k_half_height = kernel.shape[0] // 2
k_half_width = kernel.shape[1] // 2
# zero matrix with padded dimensions
padded_img = np.zeros((padded_height, padded_width))
# place image into zero matrix
padded_img[k_half_height : padded_height - k_half_height,
k_half_width : padded_width - k_half_width] = image[:, :]
return padded_img
def convolve(original_image, padded_image, kernel, stride=1):
# rotate kernel by 180 degrees
kernel = np.rot90(np.rot90(kernel))
# note that kernel height // 2 is written as 'm'
# and kernel width // 2 as 'n' in the mathematical notation
m = kernel.shape[0] // 2
n = kernel.shape[1] // 2
r = (kernel.shape[0] // 2) * 2
c = (kernel.shape[1] // 2) * 2
# initialize output array
convolved_image = np.zeros(original_image.shape)
image_height = original_image.shape[0]
image_width = original_image.shape[1]
# the convolution
for i in range(m, image_height + m, stride):
for j in range(n, image_width + n, stride):
convolved_image[i-m, j-n] = np.sum(
padded_image[i : i + m, j : j + n]
* kernel
)
return convolved_image
def convolve(image, kernel, stride=1):
for i in range(2):
kernel = np.rot90(kernel)
k_half_height = kernel.shape[0] // 2
k_half_width = kernel.shape[0] // 2
conv_image = np.zeros(image.shape)
pad_image = padding(image, kernel)
for i in range(k_half_height, conv_image.shape[0] + k_half_height, stride):
for j in range(k_half_width, conv_image.shape[1] + k_half_width, stride):
conv_image[i - k_half_height, j - k_half_width] = np.sum(
pad_image[
i - k_half_height : i + k_half_height + 1, j - k_half_width : j + k_half_width + 1
]
* kernel
)
return conv_imageIn [20]:
original_image = np.array([[4, 1, 2, 9, 8, 6],
[9, 5, 9, 5, 8, 5],
[1, 5, 9, 7, 6, 4],
[2, 9, 8, 3, 7, 1],
[8, 1, 6, 4, 2, 2],
[1, 0, 5, 7, 8, 2]])
kernel = (1/9)*np.ones((3,3))
print(f"{original_image.shape=}")
# note that convolve() performs padding
convolved_image = convolve(original_image, kernel, stride=1)
print(f"{convolved_image.shape=}")In [21]:
# Now an example using a real image and first a gaussian low-pass filter and then a Sobel filter
import numpy as np
import imageio.v3 as imageio
import matplotlib.pyplot as plt
import time
def generate_gauss_mask(sigma, K=1):
side = np.ceil(1 + 8 * sigma)
y, x = np.mgrid[-side // 2 + 1 : (side // 2) + 1, -side // 2 + 1 : (side // 2) + 1]
ker_coef = K / (2 * np.pi * sigma**2)
g = np.exp(-((x**2 + y**2) / (2.0 * sigma**2)))
return g, ker_coef
img_path = "data/IMG-2167.JPG"
image_of_cute_dog = imageio.imread(img_path, mode='L')
plt.imshow(image_of_cute_dog, cmap="gray", vmin=0, vmax=255, aspect="auto")
plt.title("Original image")
plt.show()
gauss, kernel = generate_gauss_mask(sigma=6)
gauss_kernel = gauss*kernel
filtered_image = convolve(image_of_cute_dog, gauss_kernel)
plt.imshow(filtered_image, cmap="gray", vmin=0, vmax=255, aspect="auto")
plt.title("Result of convolution with gauss kernel (blurring filter)")
plt.show()
sobel_kernel = np.array([[1, 2, 1],
[0, 0, 0],
[-1, -2, -1]])
filtered_image = convolve(image_of_cute_dog, sobel_kernel)
plt.imshow(filtered_image, cmap="gray", vmin=0, vmax=255, aspect="auto")
plt.title("Result of convolution with sobel kernel (edge detection filter)")
plt.show()In [22]:
import math
import autograd.numpy as np
from copy import deepcopy, copy
from autograd import grad
from typing import Callable
# global variables for index readability
input_index = 0
node_index = 1
bias_index = 1
input_channel_index = 1
feature_maps_index = 1
height_index = 2
width_index = 3
kernel_feature_maps_index = 1
kernel_input_channels_index = 0
class Layer:
def __init__(self, seed):
self.seed = seed
def _feedforward(self):
raise NotImplementedError
def _backpropagate(self):
raise NotImplementedError
def _reset_weights(self, previous_nodes):
raise NotImplementedErrorIn [23]:
class Convolution2DLayer(Layer):
def __init__(
self,
input_channels,
feature_maps,
kernel_height,
kernel_width,
v_stride,
h_stride,
pad,
act_func: Callable,
seed=None,
reset_weights_independently=True,
):
super().__init__(seed)
self.input_channels = input_channels
self.feature_maps = feature_maps
self.kernel_height = kernel_height
self.kernel_width = kernel_width
self.v_stride = v_stride
self.h_stride = h_stride
self.pad = pad
self.act_func = act_func
# such that the layer can be used on its own
# outside of the CNN module
if reset_weights_independently == True:
self._reset_weights_independently()
def _feedforward(self, X_batch):
# note that the shape of X_batch = [inputs, input_maps, img_height, img_width]
# pad the input batch
X_batch_padded = self._padding(X_batch)
# calculate height_index and width_index after stride
strided_height = int(np.ceil(X_batch.shape[height_index] / self.v_stride))
strided_width = int(np.ceil(X_batch.shape[width_index] / self.h_stride))
# create output array
output = np.ndarray(
(
X_batch.shape[input_index],
self.feature_maps,
strided_height,
strided_width,
)
)
# save input and output for backpropagation
self.X_batch_feedforward = X_batch
self.output_shape = output.shape
# checking for errors, no need to look here :)
self._check_for_errors()
# convolve input with kernel
for img in range(X_batch.shape[input_index]):
for chin in range(self.input_channels):
for fmap in range(self.feature_maps):
out_h = 0
for h in range(0, X_batch.shape[height_index], self.v_stride):
out_w = 0
for w in range(0, X_batch.shape[width_index], self.h_stride):
output[img, fmap, out_h, out_w] = np.sum(
X_batch_padded[
img,
chin,
h : h + self.kernel_height,
w : w + self.kernel_width,
]
* self.kernel[chin, fmap, :, :]
)
out_w += 1
out_h += 1
# Pay attention to the fact that we're not rotating the kernel by 180 degrees when filtering the image in
# the convolutional layer, as convolution in terms of Machine Learning is a procedure known as cross-correlation
# in image processing and signal processing
# return a
return self.act_func(output / (self.kernel_height))
def _backpropagate(self, delta_term_next):
# intiate matrices
delta_term = np.zeros((self.X_batch_feedforward.shape))
gradient_kernel = np.zeros((self.kernel.shape))
# pad input for convolution
X_batch_padded = self._padding(self.X_batch_feedforward)
# Since an activation function is used at the output of the convolution layer, its derivative
# has to be accounted for in the backpropagation -> as if ReLU was a layer on its own.
act_derivative = derivate(self.act_func)
delta_term_next = act_derivative(delta_term_next)
# fill in 0's for values removed by vertical stride in feedforward
if self.v_stride > 1:
v_ind = 1
for i in range(delta_term_next.shape[height_index]):
for j in range(self.v_stride - 1):
delta_term_next = np.insert(
delta_term_next, v_ind, 0, axis=height_index
)
v_ind += self.v_stride
# fill in 0's for values removed by horizontal stride in feedforward
if self.h_stride > 1:
h_ind = 1
for i in range(delta_term_next.shape[width_index]):
for k in range(self.h_stride - 1):
delta_term_next = np.insert(
delta_term_next, h_ind, 0, axis=width_index
)
h_ind += self.h_stride
# crops out 0-rows and 0-columns
delta_term_next = delta_term_next[
:,
:,
: self.X_batch_feedforward.shape[height_index],
: self.X_batch_feedforward.shape[width_index],
]
# the gradient received from the next layer also needs to be padded
delta_term_next = self._padding(delta_term_next)
# calculate delta term by convolving next delta term with kernel
for img in range(self.X_batch_feedforward.shape[input_index]):
for chin in range(self.input_channels):
for fmap in range(self.feature_maps):
for h in range(self.X_batch_feedforward.shape[height_index]):
for w in range(self.X_batch_feedforward.shape[width_index]):
delta_term[img, chin, h, w] = np.sum(
delta_term_next[
img,
fmap,
h : h + self.kernel_height,
w : w + self.kernel_width,
]
* np.rot90(np.rot90(self.kernel[chin, fmap, :, :]))
)
# calculate gradient for kernel for weight update
# also via convolution
for chin in range(self.input_channels):
for fmap in range(self.feature_maps):
for k_x in range(self.kernel_height):
for k_y in range(self.kernel_width):
gradient_kernel[chin, fmap, k_x, k_y] = np.sum(
X_batch_padded[
img,
chin,
h : h + self.kernel_height,
w : w + self.kernel_width,
]
* delta_term_next[
img,
fmap,
h : h + self.kernel_height,
w : w + self.kernel_width,
]
)
# all kernels are updated with weight gradient of kernel
self.kernel -= gradient_kernel
# return delta term
return delta_term
def _padding(self, X_batch, batch_type="image"):
# same padding for images
if self.pad == "same" and batch_type == "image":
padded_height = X_batch.shape[height_index] + (self.kernel_height // 2) * 2
padded_width = X_batch.shape[width_index] + (self.kernel_width // 2) * 2
half_kernel_height = self.kernel_height // 2
half_kernel_width = self.kernel_width // 2
# initialize padded array
X_batch_padded = np.ndarray(
(
X_batch.shape[input_index],
X_batch.shape[feature_maps_index],
padded_height,
padded_width,
)
)
# zero pad all images in X_batch
for img in range(X_batch.shape[input_index]):
padded_img = np.zeros(
(X_batch.shape[feature_maps_index], padded_height, padded_width)
)
padded_img[
:,
half_kernel_height : padded_height - half_kernel_height,
half_kernel_width : padded_width - half_kernel_width,
] = X_batch[img, :, :, :]
X_batch_padded[img, :, :, :] = padded_img[:, :, :]
return X_batch_padded
# same padding for gradients
elif self.pad == "same" and batch_type == "grad":
padded_height = X_batch.shape[height_index] + (self.kernel_height // 2) * 2
padded_width = X_batch.shape[width_index] + (self.kernel_width // 2) * 2
half_kernel_height = self.kernel_height // 2
half_kernel_width = self.kernel_width // 2
# initialize padded array
delta_term_padded = np.zeros(
(
X_batch.shape[input_index],
X_batch.shape[feature_maps_index],
padded_height,
padded_width,
)
)
# zero pad delta term
delta_term_padded[
:, :, : X_batch.shape[height_index], : X_batch.shape[width_index]
] = X_batch[:, :, :, :]
return delta_term_padded
else:
return X_batch
def _reset_weights_independently(self):
# sets seed to remove randomness inbetween runs
if self.seed is not None:
np.random.seed(self.seed)
# initializes kernel matrix
self.kernel = np.ndarray(
(
self.input_channels,
self.feature_maps,
self.kernel_height,
self.kernel_width,
)
)
# randomly initializes weights
for chin in range(self.kernel.shape[kernel_input_channels_index]):
for fmap in range(self.kernel.shape[kernel_feature_maps_index]):
self.kernel[chin, fmap, :, :] = np.random.rand(
self.kernel_height, self.kernel_width
)
def _reset_weights(self, previous_nodes):
# sets weights
self._reset_weights_independently()
# returns shape of output used for subsequent layer's weight initiation
strided_height = int(
np.ceil(previous_nodes.shape[height_index] / self.v_stride)
)
strided_width = int(np.ceil(previous_nodes.shape[width_index] / self.h_stride))
next_nodes = np.ones(
(
previous_nodes.shape[input_index],
self.feature_maps,
strided_height,
strided_width,
)
)
return next_nodes / self.kernel_height
def _check_for_errors(self):
if self.X_batch_feedforward.shape[input_channel_index] != self.input_channels:
raise AssertionError(
f"ERROR: Number of input channels in data ({self.X_batch_feedforward.shape[input_channel_index]}) is not equal to input channels in Convolution2DLayerOPT ({self.input_channels})! Please change the number of input channels of the Convolution2DLayer such that they are equal"
)In [24]:
import numpy as np
import imageio.v3 as imageio
import matplotlib.pyplot as plt
def plot_convolution_result(X, layer):
plt.imshow(X[0, 0, :, :], vmin=0, vmax=255, cmap="gray")
plt.title("Original image")
plt.colorbar()
plt.show()
conv_result = layer._feedforward(X)
plt.title("Result of convolutional layer")
plt.imshow(conv_result[0, 0, :, :], vmin=0, vmax=255, cmap="gray")
plt.colorbar()
plt.show()
# create layer
layer = Convolution2DLayer(
input_channels=3,
feature_maps=1,
kernel_height=4,
kernel_width=4,
v_stride=2,
h_stride=2,
pad="same",
act_func=identity,
seed=2023,
)
# read in image path, make data correct format
img_path = img_path = "data/IMG-2167.JPG"
image_of_cute_dog = imageio.imread(img_path)
image_shape = image_of_cute_dog.shape
image_of_cute_dog = image_of_cute_dog.reshape(1, image_shape[0], image_shape[1], image_shape[2])
image_of_cute_dog = image_of_cute_dog.transpose(0, 3, 1, 2)
# plot the result of the convolution
plot_convolution_result(image_of_cute_dog, layer)In [25]:
class Pooling2DLayer(Layer):
def __init__(
self,
kernel_height,
kernel_width,
v_stride,
h_stride,
pooling="max",
seed=None,
):
super().__init__(seed)
self.kernel_height = kernel_height
self.kernel_width = kernel_width
self.v_stride = v_stride
self.h_stride = h_stride
self.pooling = pooling
def _feedforward(self, X_batch):
# Saving the input for use in the backwardpass
self.X_batch_feedforward = X_batch
# check if user is silly
self._check_for_errors()
# Computing the size of the feature maps based on kernel size and the stride parameter
strided_height = (
X_batch.shape[height_index] - self.kernel_height
) // self.v_stride + 1
if X_batch.shape[height_index] == X_batch.shape[width_index]:
strided_width = strided_height
else:
strided_width = (
X_batch.shape[width_index] - self.kernel_width
) // self.h_stride + 1
# initialize output array
output = np.ndarray(
(
X_batch.shape[input_index],
X_batch.shape[feature_maps_index],
strided_height,
strided_width,
)
)
# select pooling action, either max or average pooling
if self.pooling == "max":
self.pooling_action = np.max
elif self.pooling == "average":
self.pooling_action = np.mean
# pool based on kernel size and stride
for img in range(output.shape[input_index]):
for fmap in range(output.shape[feature_maps_index]):
for h in range(strided_height):
for w in range(strided_width):
output[img, fmap, h, w] = self.pooling_action(
X_batch[
img,
fmap,
(h * self.v_stride) : (h * self.v_stride)
+ self.kernel_height,
(w * self.h_stride) : (w * self.h_stride)
+ self.kernel_width,
]
)
# output for feedforward in next layer
return output
def _backpropagate(self, delta_term_next):
# initiate delta term array
delta_term = np.zeros((self.X_batch_feedforward.shape))
for img in range(delta_term_next.shape[input_index]):
for fmap in range(delta_term_next.shape[feature_maps_index]):
for h in range(0, delta_term_next.shape[height_index], self.v_stride):
for w in range(
0, delta_term_next.shape[width_index], self.h_stride
):
# max pooling
if self.pooling == "max":
# get window
window = self.X_batch_feedforward[
img,
fmap,
h : h + self.kernel_height,
w : w + self.kernel_width,
]
# find max values indices in window
max_h, max_w = np.unravel_index(
window.argmax(), window.shape
)
# set values in new, upsampled delta term
delta_term[
img,
fmap,
(h + max_h),
(w + max_w),
] += delta_term_next[img, fmap, h, w]
# average pooling
if self.pooling == "average":
delta_term[
img,
fmap,
h : h + self.kernel_height,
w : w + self.kernel_width,
] = (
delta_term_next[img, fmap, h, w]
/ self.kernel_height
/ self.kernel_width
)
# returns input to backpropagation in previous layer
return delta_term
def _reset_weights(self, previous_nodes):
# calculate strided height, strided width
strided_height = (
previous_nodes.shape[height_index] - self.kernel_height
) // self.v_stride + 1
if previous_nodes.shape[height_index] == previous_nodes.shape[width_index]:
strided_width = strided_height
else:
strided_width = (
previous_nodes.shape[width_index] - self.kernel_width
) // self.h_stride + 1
# initiate output array
output = np.ones(
(
previous_nodes.shape[input_index],
previous_nodes.shape[feature_maps_index],
strided_height,
strided_width,
)
)
# returns output with shape used for reset weights in next layer
return output
def _check_for_errors(self):
# check if input is smaller than kernel size -> error
assert (
self.X_batch_feedforward.shape[width_index] >= self.kernel_width
), f"ERROR: Pooling kernel width_index ({self.kernel_width}) larger than data width_index ({self.X_batch_feedforward.input.shape[2]}), please lower the kernel width_index of the Pooling2DLayer"
assert (
self.X_batch_feedforward.shape[height_index] >= self.kernel_height
), f"ERROR: Pooling kernel height_index ({self.kernel_height}) larger than data height_index ({self.X_batch_feedforward.input.shape[3]}), please lower the kernel height_index of the Pooling2DLayer"In [26]:
class FlattenLayer(Layer):
def __init__(self, act_func=LRELU, seed=None):
super().__init__(seed)
self.act_func = act_func
def _feedforward(self, X_batch):
# save input for backpropagation
self.X_batch_feedforward_shape = X_batch.shape
# Remember, the data has the following shape: (I, FM, H, W, ) in the convolutional layers
# whilst the data has the shape (I, FM * H * W) in the fully connected layers
# I = Inputs, FM = Feature Maps, H = Height and W = Width.
X_batch = X_batch.reshape(
X_batch.shape[input_index],
X_batch.shape[feature_maps_index]
* X_batch.shape[height_index]
* X_batch.shape[width_index],
)
# add bias to a
self.z_matrix = X_batch
bias = np.ones((X_batch.shape[input_index], 1)) * 0.01
self.a_matrix = np.hstack([bias, X_batch])
# return a, the input to feedforward in next layer
return self.a_matrix
def _backpropagate(self, weights_next, delta_term_next):
activation_derivative = derivate(self.act_func)
# calculate delta term
delta_term = (
weights_next[bias_index:, :] @ delta_term_next.T
).T * activation_derivative(self.z_matrix)
# FlattenLayer does not update weights
# reshapes delta layer to convolutional layer data format [Input, Feature_Maps, Height, Width]
return delta_term.reshape(self.X_batch_feedforward_shape)
def _reset_weights(self, previous_nodes):
# note that the previous nodes to the FlattenLayer are from the convolutional layers
previous_nodes = previous_nodes.reshape(
previous_nodes.shape[input_index],
previous_nodes.shape[feature_maps_index]
* previous_nodes.shape[height_index]
* previous_nodes.shape[width_index],
)
# return shape used in reset_weights in next layer
return previous_nodes.shape[node_index]
def get_prev_a(self):
return self.a_matrixWarning:
Output truncated. This notebook contains too many cells to display efficiently.