wehub-resource-sync
734 lines
27 KiB
Python
734 lines
27 KiB
Python
|
|
from __future__ import print_function, division
|
|
import math
|
|
import numpy as np
|
|
import copy
|
|
from mlfromscratch.deep_learning.activation_functions import Sigmoid, ReLU, SoftPlus, LeakyReLU
|
|
from mlfromscratch.deep_learning.activation_functions import TanH, ELU, SELU, Softmax
|
|
|
|
|
|
class Layer(object):
|
|
|
|
def set_input_shape(self, shape):
|
|
""" Sets the shape that the layer expects of the input in the forward
|
|
pass method """
|
|
self.input_shape = shape
|
|
|
|
def layer_name(self):
|
|
""" The name of the layer. Used in model summary. """
|
|
return self.__class__.__name__
|
|
|
|
def parameters(self):
|
|
""" The number of trainable parameters used by the layer """
|
|
return 0
|
|
|
|
def forward_pass(self, X, training):
|
|
""" Propogates the signal forward in the network """
|
|
raise NotImplementedError()
|
|
|
|
def backward_pass(self, accum_grad):
|
|
""" Propogates the accumulated gradient backwards in the network.
|
|
If the has trainable weights then these weights are also tuned in this method.
|
|
As input (accum_grad) it receives the gradient with respect to the output of the layer and
|
|
returns the gradient with respect to the output of the previous layer. """
|
|
raise NotImplementedError()
|
|
|
|
def output_shape(self):
|
|
""" The shape of the output produced by forward_pass """
|
|
raise NotImplementedError()
|
|
|
|
|
|
class Dense(Layer):
|
|
"""A fully-connected NN layer.
|
|
Parameters:
|
|
-----------
|
|
n_units: int
|
|
The number of neurons in the layer.
|
|
input_shape: tuple
|
|
The expected input shape of the layer. For dense layers a single digit specifying
|
|
the number of features of the input. Must be specified if it is the first layer in
|
|
the network.
|
|
"""
|
|
def __init__(self, n_units, input_shape=None):
|
|
self.layer_input = None
|
|
self.input_shape = input_shape
|
|
self.n_units = n_units
|
|
self.trainable = True
|
|
self.W = None
|
|
self.w0 = None
|
|
|
|
def initialize(self, optimizer):
|
|
# Initialize the weights
|
|
limit = 1 / math.sqrt(self.input_shape[0])
|
|
self.W = np.random.uniform(-limit, limit, (self.input_shape[0], self.n_units))
|
|
self.w0 = np.zeros((1, self.n_units))
|
|
# Weight optimizers
|
|
self.W_opt = copy.copy(optimizer)
|
|
self.w0_opt = copy.copy(optimizer)
|
|
|
|
def parameters(self):
|
|
return np.prod(self.W.shape) + np.prod(self.w0.shape)
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.layer_input = X
|
|
return X.dot(self.W) + self.w0
|
|
|
|
def backward_pass(self, accum_grad):
|
|
# Save weights used during forwards pass
|
|
W = self.W
|
|
|
|
if self.trainable:
|
|
# Calculate gradient w.r.t layer weights
|
|
grad_w = self.layer_input.T.dot(accum_grad)
|
|
grad_w0 = np.sum(accum_grad, axis=0, keepdims=True)
|
|
|
|
# Update the layer weights
|
|
self.W = self.W_opt.update(self.W, grad_w)
|
|
self.w0 = self.w0_opt.update(self.w0, grad_w0)
|
|
|
|
# Return accumulated gradient for next layer
|
|
# Calculated based on the weights used during the forward pass
|
|
accum_grad = accum_grad.dot(W.T)
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
return (self.n_units, )
|
|
|
|
|
|
class RNN(Layer):
|
|
"""A Vanilla Fully-Connected Recurrent Neural Network layer.
|
|
|
|
Parameters:
|
|
-----------
|
|
n_units: int
|
|
The number of hidden states in the layer.
|
|
activation: string
|
|
The name of the activation function which will be applied to the output of each state.
|
|
bptt_trunc: int
|
|
Decides how many time steps the gradient should be propagated backwards through states
|
|
given the loss gradient for time step t.
|
|
input_shape: tuple
|
|
The expected input shape of the layer. For dense layers a single digit specifying
|
|
the number of features of the input. Must be specified if it is the first layer in
|
|
the network.
|
|
|
|
Reference:
|
|
http://www.wildml.com/2015/09/recurrent-neural-networks-tutorial-part-2-implementing-a-language-model-rnn-with-python-numpy-and-theano/
|
|
"""
|
|
def __init__(self, n_units, activation='tanh', bptt_trunc=5, input_shape=None):
|
|
self.input_shape = input_shape
|
|
self.n_units = n_units
|
|
self.activation = activation_functions[activation]()
|
|
self.trainable = True
|
|
self.bptt_trunc = bptt_trunc
|
|
self.W = None # Weight of the previous state
|
|
self.V = None # Weight of the output
|
|
self.U = None # Weight of the input
|
|
|
|
def initialize(self, optimizer):
|
|
timesteps, input_dim = self.input_shape
|
|
# Initialize the weights
|
|
limit = 1 / math.sqrt(input_dim)
|
|
self.U = np.random.uniform(-limit, limit, (self.n_units, input_dim))
|
|
limit = 1 / math.sqrt(self.n_units)
|
|
self.V = np.random.uniform(-limit, limit, (input_dim, self.n_units))
|
|
self.W = np.random.uniform(-limit, limit, (self.n_units, self.n_units))
|
|
# Weight optimizers
|
|
self.U_opt = copy.copy(optimizer)
|
|
self.V_opt = copy.copy(optimizer)
|
|
self.W_opt = copy.copy(optimizer)
|
|
|
|
def parameters(self):
|
|
return np.prod(self.W.shape) + np.prod(self.U.shape) + np.prod(self.V.shape)
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.layer_input = X
|
|
batch_size, timesteps, input_dim = X.shape
|
|
|
|
# Save these values for use in backprop.
|
|
self.state_input = np.zeros((batch_size, timesteps, self.n_units))
|
|
self.states = np.zeros((batch_size, timesteps+1, self.n_units))
|
|
self.outputs = np.zeros((batch_size, timesteps, input_dim))
|
|
|
|
# Set last time step to zero for calculation of the state_input at time step zero
|
|
self.states[:, -1] = np.zeros((batch_size, self.n_units))
|
|
for t in range(timesteps):
|
|
# Input to state_t is the current input and output of previous states
|
|
self.state_input[:, t] = X[:, t].dot(self.U.T) + self.states[:, t-1].dot(self.W.T)
|
|
self.states[:, t] = self.activation(self.state_input[:, t])
|
|
self.outputs[:, t] = self.states[:, t].dot(self.V.T)
|
|
|
|
return self.outputs
|
|
|
|
def backward_pass(self, accum_grad):
|
|
_, timesteps, _ = accum_grad.shape
|
|
|
|
# Variables where we save the accumulated gradient w.r.t each parameter
|
|
grad_U = np.zeros_like(self.U)
|
|
grad_V = np.zeros_like(self.V)
|
|
grad_W = np.zeros_like(self.W)
|
|
# The gradient w.r.t the layer input.
|
|
# Will be passed on to the previous layer in the network
|
|
accum_grad_next = np.zeros_like(accum_grad)
|
|
|
|
# Back Propagation Through Time
|
|
for t in reversed(range(timesteps)):
|
|
# Update gradient w.r.t V at time step t
|
|
grad_V += accum_grad[:, t].T.dot(self.states[:, t])
|
|
# Calculate the gradient w.r.t the state input
|
|
grad_wrt_state = accum_grad[:, t].dot(self.V) * self.activation.gradient(self.state_input[:, t])
|
|
# Gradient w.r.t the layer input
|
|
accum_grad_next[:, t] = grad_wrt_state.dot(self.U)
|
|
# Update gradient w.r.t W and U by backprop. from time step t for at most
|
|
# self.bptt_trunc number of time steps
|
|
for t_ in reversed(np.arange(max(0, t - self.bptt_trunc), t+1)):
|
|
grad_U += grad_wrt_state.T.dot(self.layer_input[:, t_])
|
|
grad_W += grad_wrt_state.T.dot(self.states[:, t_-1])
|
|
# Calculate gradient w.r.t previous state
|
|
grad_wrt_state = grad_wrt_state.dot(self.W) * self.activation.gradient(self.state_input[:, t_-1])
|
|
|
|
# Update weights
|
|
self.U = self.U_opt.update(self.U, grad_U)
|
|
self.V = self.V_opt.update(self.V, grad_V)
|
|
self.W = self.W_opt.update(self.W, grad_W)
|
|
|
|
return accum_grad_next
|
|
|
|
def output_shape(self):
|
|
return self.input_shape
|
|
|
|
class Conv2D(Layer):
|
|
"""A 2D Convolution Layer.
|
|
|
|
Parameters:
|
|
-----------
|
|
n_filters: int
|
|
The number of filters that will convolve over the input matrix. The number of channels
|
|
of the output shape.
|
|
filter_shape: tuple
|
|
A tuple (filter_height, filter_width).
|
|
input_shape: tuple
|
|
The shape of the expected input of the layer. (batch_size, channels, height, width)
|
|
Only needs to be specified for first layer in the network.
|
|
padding: string
|
|
Either 'same' or 'valid'. 'same' results in padding being added so that the output height and width
|
|
matches the input height and width. For 'valid' no padding is added.
|
|
stride: int
|
|
The stride length of the filters during the convolution over the input.
|
|
"""
|
|
def __init__(self, n_filters, filter_shape, input_shape=None, padding='same', stride=1):
|
|
self.n_filters = n_filters
|
|
self.filter_shape = filter_shape
|
|
self.padding = padding
|
|
self.stride = stride
|
|
self.input_shape = input_shape
|
|
self.trainable = True
|
|
|
|
def initialize(self, optimizer):
|
|
# Initialize the weights
|
|
filter_height, filter_width = self.filter_shape
|
|
channels = self.input_shape[0]
|
|
limit = 1 / math.sqrt(np.prod(self.filter_shape))
|
|
self.W = np.random.uniform(-limit, limit, size=(self.n_filters, channels, filter_height, filter_width))
|
|
self.w0 = np.zeros((self.n_filters, 1))
|
|
# Weight optimizers
|
|
self.W_opt = copy.copy(optimizer)
|
|
self.w0_opt = copy.copy(optimizer)
|
|
|
|
def parameters(self):
|
|
return np.prod(self.W.shape) + np.prod(self.w0.shape)
|
|
|
|
def forward_pass(self, X, training=True):
|
|
batch_size, channels, height, width = X.shape
|
|
self.layer_input = X
|
|
# Turn image shape into column shape
|
|
# (enables dot product between input and weights)
|
|
self.X_col = image_to_column(X, self.filter_shape, stride=self.stride, output_shape=self.padding)
|
|
# Turn weights into column shape
|
|
self.W_col = self.W.reshape((self.n_filters, -1))
|
|
# Calculate output
|
|
output = self.W_col.dot(self.X_col) + self.w0
|
|
# Reshape into (n_filters, out_height, out_width, batch_size)
|
|
output = output.reshape(self.output_shape() + (batch_size, ))
|
|
# Redistribute axises so that batch size comes first
|
|
return output.transpose(3,0,1,2)
|
|
|
|
def backward_pass(self, accum_grad):
|
|
# Reshape accumulated gradient into column shape
|
|
accum_grad = accum_grad.transpose(1, 2, 3, 0).reshape(self.n_filters, -1)
|
|
|
|
if self.trainable:
|
|
# Take dot product between column shaped accum. gradient and column shape
|
|
# layer input to determine the gradient at the layer with respect to layer weights
|
|
grad_w = accum_grad.dot(self.X_col.T).reshape(self.W.shape)
|
|
# The gradient with respect to bias terms is the sum similarly to in Dense layer
|
|
grad_w0 = np.sum(accum_grad, axis=1, keepdims=True)
|
|
|
|
# Update the layers weights
|
|
self.W = self.W_opt.update(self.W, grad_w)
|
|
self.w0 = self.w0_opt.update(self.w0, grad_w0)
|
|
|
|
# Recalculate the gradient which will be propogated back to prev. layer
|
|
accum_grad = self.W_col.T.dot(accum_grad)
|
|
# Reshape from column shape to image shape
|
|
accum_grad = column_to_image(accum_grad,
|
|
self.layer_input.shape,
|
|
self.filter_shape,
|
|
stride=self.stride,
|
|
output_shape=self.padding)
|
|
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
channels, height, width = self.input_shape
|
|
pad_h, pad_w = determine_padding(self.filter_shape, output_shape=self.padding)
|
|
output_height = (height + np.sum(pad_h) - self.filter_shape[0]) / self.stride + 1
|
|
output_width = (width + np.sum(pad_w) - self.filter_shape[1]) / self.stride + 1
|
|
return self.n_filters, int(output_height), int(output_width)
|
|
|
|
|
|
class BatchNormalization(Layer):
|
|
"""Batch normalization.
|
|
"""
|
|
def __init__(self, momentum=0.99):
|
|
self.momentum = momentum
|
|
self.trainable = True
|
|
self.eps = 0.01
|
|
self.running_mean = None
|
|
self.running_var = None
|
|
|
|
def initialize(self, optimizer):
|
|
# Initialize the parameters
|
|
self.gamma = np.ones(self.input_shape)
|
|
self.beta = np.zeros(self.input_shape)
|
|
# parameter optimizers
|
|
self.gamma_opt = copy.copy(optimizer)
|
|
self.beta_opt = copy.copy(optimizer)
|
|
|
|
def parameters(self):
|
|
return np.prod(self.gamma.shape) + np.prod(self.beta.shape)
|
|
|
|
def forward_pass(self, X, training=True):
|
|
|
|
# Initialize running mean and variance if first run
|
|
if self.running_mean is None:
|
|
self.running_mean = np.mean(X, axis=0)
|
|
self.running_var = np.var(X, axis=0)
|
|
|
|
if training and self.trainable:
|
|
mean = np.mean(X, axis=0)
|
|
var = np.var(X, axis=0)
|
|
self.running_mean = self.momentum * self.running_mean + (1 - self.momentum) * mean
|
|
self.running_var = self.momentum * self.running_var + (1 - self.momentum) * var
|
|
else:
|
|
mean = self.running_mean
|
|
var = self.running_var
|
|
|
|
# Statistics saved for backward pass
|
|
self.X_centered = X - mean
|
|
self.stddev_inv = 1 / np.sqrt(var + self.eps)
|
|
|
|
X_norm = self.X_centered * self.stddev_inv
|
|
output = self.gamma * X_norm + self.beta
|
|
|
|
return output
|
|
|
|
def backward_pass(self, accum_grad):
|
|
|
|
# Save parameters used during the forward pass
|
|
gamma = self.gamma
|
|
|
|
# If the layer is trainable the parameters are updated
|
|
if self.trainable:
|
|
X_norm = self.X_centered * self.stddev_inv
|
|
grad_gamma = np.sum(accum_grad * X_norm, axis=0)
|
|
grad_beta = np.sum(accum_grad, axis=0)
|
|
|
|
self.gamma = self.gamma_opt.update(self.gamma, grad_gamma)
|
|
self.beta = self.beta_opt.update(self.beta, grad_beta)
|
|
|
|
batch_size = accum_grad.shape[0]
|
|
|
|
# The gradient of the loss with respect to the layer inputs (use weights and statistics from forward pass)
|
|
accum_grad = (1 / batch_size) * gamma * self.stddev_inv * (
|
|
batch_size * accum_grad
|
|
- np.sum(accum_grad, axis=0)
|
|
- self.X_centered * self.stddev_inv**2 * np.sum(accum_grad * self.X_centered, axis=0)
|
|
)
|
|
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
return self.input_shape
|
|
|
|
|
|
class PoolingLayer(Layer):
|
|
"""A parent class of MaxPooling2D and AveragePooling2D
|
|
"""
|
|
def __init__(self, pool_shape=(2, 2), stride=1, padding=0):
|
|
self.pool_shape = pool_shape
|
|
self.stride = stride
|
|
self.padding = padding
|
|
self.trainable = True
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.layer_input = X
|
|
|
|
batch_size, channels, height, width = X.shape
|
|
|
|
_, out_height, out_width = self.output_shape()
|
|
|
|
X = X.reshape(batch_size*channels, 1, height, width)
|
|
X_col = image_to_column(X, self.pool_shape, self.stride, self.padding)
|
|
|
|
# MaxPool or AveragePool specific method
|
|
output = self._pool_forward(X_col)
|
|
|
|
output = output.reshape(out_height, out_width, batch_size, channels)
|
|
output = output.transpose(2, 3, 0, 1)
|
|
|
|
return output
|
|
|
|
def backward_pass(self, accum_grad):
|
|
batch_size, _, _, _ = accum_grad.shape
|
|
channels, height, width = self.input_shape
|
|
accum_grad = accum_grad.transpose(2, 3, 0, 1).ravel()
|
|
|
|
# MaxPool or AveragePool specific method
|
|
accum_grad_col = self._pool_backward(accum_grad)
|
|
|
|
accum_grad = column_to_image(accum_grad_col, (batch_size * channels, 1, height, width), self.pool_shape, self.stride, 0)
|
|
accum_grad = accum_grad.reshape((batch_size,) + self.input_shape)
|
|
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
channels, height, width = self.input_shape
|
|
out_height = (height - self.pool_shape[0]) / self.stride + 1
|
|
out_width = (width - self.pool_shape[1]) / self.stride + 1
|
|
assert out_height % 1 == 0
|
|
assert out_width % 1 == 0
|
|
return channels, int(out_height), int(out_width)
|
|
|
|
|
|
class MaxPooling2D(PoolingLayer):
|
|
def _pool_forward(self, X_col):
|
|
arg_max = np.argmax(X_col, axis=0).flatten()
|
|
output = X_col[arg_max, range(arg_max.size)]
|
|
self.cache = arg_max
|
|
return output
|
|
|
|
def _pool_backward(self, accum_grad):
|
|
accum_grad_col = np.zeros((np.prod(self.pool_shape), accum_grad.size))
|
|
arg_max = self.cache
|
|
accum_grad_col[arg_max, range(accum_grad.size)] = accum_grad
|
|
return accum_grad_col
|
|
|
|
class AveragePooling2D(PoolingLayer):
|
|
def _pool_forward(self, X_col):
|
|
output = np.mean(X_col, axis=0)
|
|
return output
|
|
|
|
def _pool_backward(self, accum_grad):
|
|
accum_grad_col = np.zeros((np.prod(self.pool_shape), accum_grad.size))
|
|
accum_grad_col[:, range(accum_grad.size)] = 1. / accum_grad_col.shape[0] * accum_grad
|
|
return accum_grad_col
|
|
|
|
|
|
class ConstantPadding2D(Layer):
|
|
"""Adds rows and columns of constant values to the input.
|
|
Expects the input to be of shape (batch_size, channels, height, width)
|
|
|
|
Parameters:
|
|
-----------
|
|
padding: tuple
|
|
The amount of padding along the height and width dimension of the input.
|
|
If (pad_h, pad_w) the same symmetric padding is applied along height and width dimension.
|
|
If ((pad_h0, pad_h1), (pad_w0, pad_w1)) the specified padding is added to beginning and end of
|
|
the height and width dimension.
|
|
padding_value: int or tuple
|
|
The value the is added as padding.
|
|
"""
|
|
def __init__(self, padding, padding_value=0):
|
|
self.padding = padding
|
|
self.trainable = True
|
|
if not isinstance(padding[0], tuple):
|
|
self.padding = ((padding[0], padding[0]), padding[1])
|
|
if not isinstance(padding[1], tuple):
|
|
self.padding = (self.padding[0], (padding[1], padding[1]))
|
|
self.padding_value = padding_value
|
|
|
|
def forward_pass(self, X, training=True):
|
|
output = np.pad(X,
|
|
pad_width=((0,0), (0,0), self.padding[0], self.padding[1]),
|
|
mode="constant",
|
|
constant_values=self.padding_value)
|
|
return output
|
|
|
|
def backward_pass(self, accum_grad):
|
|
pad_top, pad_left = self.padding[0][0], self.padding[1][0]
|
|
height, width = self.input_shape[1], self.input_shape[2]
|
|
accum_grad = accum_grad[:, :, pad_top:pad_top+height, pad_left:pad_left+width]
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
new_height = self.input_shape[1] + np.sum(self.padding[0])
|
|
new_width = self.input_shape[2] + np.sum(self.padding[1])
|
|
return (self.input_shape[0], new_height, new_width)
|
|
|
|
|
|
class ZeroPadding2D(ConstantPadding2D):
|
|
"""Adds rows and columns of zero values to the input.
|
|
Expects the input to be of shape (batch_size, channels, height, width)
|
|
|
|
Parameters:
|
|
-----------
|
|
padding: tuple
|
|
The amount of padding along the height and width dimension of the input.
|
|
If (pad_h, pad_w) the same symmetric padding is applied along height and width dimension.
|
|
If ((pad_h0, pad_h1), (pad_w0, pad_w1)) the specified padding is added to beginning and end of
|
|
the height and width dimension.
|
|
"""
|
|
def __init__(self, padding):
|
|
self.padding = padding
|
|
if isinstance(padding[0], int):
|
|
self.padding = ((padding[0], padding[0]), padding[1])
|
|
if isinstance(padding[1], int):
|
|
self.padding = (self.padding[0], (padding[1], padding[1]))
|
|
self.padding_value = 0
|
|
|
|
|
|
class Flatten(Layer):
|
|
""" Turns a multidimensional matrix into two-dimensional """
|
|
def __init__(self, input_shape=None):
|
|
self.prev_shape = None
|
|
self.trainable = True
|
|
self.input_shape = input_shape
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.prev_shape = X.shape
|
|
return X.reshape((X.shape[0], -1))
|
|
|
|
def backward_pass(self, accum_grad):
|
|
return accum_grad.reshape(self.prev_shape)
|
|
|
|
def output_shape(self):
|
|
return (np.prod(self.input_shape),)
|
|
|
|
|
|
class UpSampling2D(Layer):
|
|
""" Nearest neighbor up sampling of the input. Repeats the rows and
|
|
columns of the data by size[0] and size[1] respectively.
|
|
|
|
Parameters:
|
|
-----------
|
|
size: tuple
|
|
(size_y, size_x) - The number of times each axis will be repeated.
|
|
"""
|
|
def __init__(self, size=(2,2), input_shape=None):
|
|
self.prev_shape = None
|
|
self.trainable = True
|
|
self.size = size
|
|
self.input_shape = input_shape
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.prev_shape = X.shape
|
|
# Repeat each axis as specified by size
|
|
X_new = X.repeat(self.size[0], axis=2).repeat(self.size[1], axis=3)
|
|
return X_new
|
|
|
|
def backward_pass(self, accum_grad):
|
|
# Down sample input to previous shape
|
|
accum_grad = accum_grad[:, :, ::self.size[0], ::self.size[1]]
|
|
return accum_grad
|
|
|
|
def output_shape(self):
|
|
channels, height, width = self.input_shape
|
|
return channels, self.size[0] * height, self.size[1] * width
|
|
|
|
|
|
class Reshape(Layer):
|
|
""" Reshapes the input tensor into specified shape
|
|
|
|
Parameters:
|
|
-----------
|
|
shape: tuple
|
|
The shape which the input shall be reshaped to.
|
|
"""
|
|
def __init__(self, shape, input_shape=None):
|
|
self.prev_shape = None
|
|
self.trainable = True
|
|
self.shape = shape
|
|
self.input_shape = input_shape
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.prev_shape = X.shape
|
|
return X.reshape((X.shape[0], ) + self.shape)
|
|
|
|
def backward_pass(self, accum_grad):
|
|
return accum_grad.reshape(self.prev_shape)
|
|
|
|
def output_shape(self):
|
|
return self.shape
|
|
|
|
|
|
class Dropout(Layer):
|
|
"""A layer that randomly sets a fraction p of the output units of the previous layer
|
|
to zero.
|
|
|
|
Parameters:
|
|
-----------
|
|
p: float
|
|
The probability that unit x is set to zero.
|
|
"""
|
|
def __init__(self, p=0.2):
|
|
self.p = p
|
|
self._mask = None
|
|
self.input_shape = None
|
|
self.n_units = None
|
|
self.pass_through = True
|
|
self.trainable = True
|
|
|
|
def forward_pass(self, X, training=True):
|
|
c = (1 - self.p)
|
|
if training:
|
|
self._mask = np.random.uniform(size=X.shape) > self.p
|
|
c = self._mask
|
|
return X * c
|
|
|
|
def backward_pass(self, accum_grad):
|
|
return accum_grad * self._mask
|
|
|
|
def output_shape(self):
|
|
return self.input_shape
|
|
|
|
activation_functions = {
|
|
'relu': ReLU,
|
|
'sigmoid': Sigmoid,
|
|
'selu': SELU,
|
|
'elu': ELU,
|
|
'softmax': Softmax,
|
|
'leaky_relu': LeakyReLU,
|
|
'tanh': TanH,
|
|
'softplus': SoftPlus
|
|
}
|
|
|
|
class Activation(Layer):
|
|
"""A layer that applies an activation operation to the input.
|
|
|
|
Parameters:
|
|
-----------
|
|
name: string
|
|
The name of the activation function that will be used.
|
|
"""
|
|
|
|
def __init__(self, name):
|
|
self.activation_name = name
|
|
self.activation_func = activation_functions[name]()
|
|
self.trainable = True
|
|
|
|
def layer_name(self):
|
|
return "Activation (%s)" % (self.activation_func.__class__.__name__)
|
|
|
|
def forward_pass(self, X, training=True):
|
|
self.layer_input = X
|
|
return self.activation_func(X)
|
|
|
|
def backward_pass(self, accum_grad):
|
|
return accum_grad * self.activation_func.gradient(self.layer_input)
|
|
|
|
def output_shape(self):
|
|
return self.input_shape
|
|
|
|
|
|
# Method which calculates the padding based on the specified output shape and the
|
|
# shape of the filters
|
|
def determine_padding(filter_shape, output_shape="same"):
|
|
|
|
# No padding
|
|
if output_shape == "valid":
|
|
return (0, 0), (0, 0)
|
|
# Pad so that the output shape is the same as input shape (given that stride=1)
|
|
elif output_shape == "same":
|
|
filter_height, filter_width = filter_shape
|
|
|
|
# Derived from:
|
|
# output_height = (height + pad_h - filter_height) / stride + 1
|
|
# In this case output_height = height and stride = 1. This gives the
|
|
# expression for the padding below.
|
|
pad_h1 = int(math.floor((filter_height - 1)/2))
|
|
pad_h2 = int(math.ceil((filter_height - 1)/2))
|
|
pad_w1 = int(math.floor((filter_width - 1)/2))
|
|
pad_w2 = int(math.ceil((filter_width - 1)/2))
|
|
|
|
return (pad_h1, pad_h2), (pad_w1, pad_w2)
|
|
|
|
|
|
# Reference: CS231n Stanford
|
|
def get_im2col_indices(images_shape, filter_shape, padding, stride=1):
|
|
# First figure out what the size of the output should be
|
|
batch_size, channels, height, width = images_shape
|
|
filter_height, filter_width = filter_shape
|
|
pad_h, pad_w = padding
|
|
out_height = int((height + np.sum(pad_h) - filter_height) / stride + 1)
|
|
out_width = int((width + np.sum(pad_w) - filter_width) / stride + 1)
|
|
|
|
i0 = np.repeat(np.arange(filter_height), filter_width)
|
|
i0 = np.tile(i0, channels)
|
|
i1 = stride * np.repeat(np.arange(out_height), out_width)
|
|
j0 = np.tile(np.arange(filter_width), filter_height * channels)
|
|
j1 = stride * np.tile(np.arange(out_width), out_height)
|
|
i = i0.reshape(-1, 1) + i1.reshape(1, -1)
|
|
j = j0.reshape(-1, 1) + j1.reshape(1, -1)
|
|
|
|
k = np.repeat(np.arange(channels), filter_height * filter_width).reshape(-1, 1)
|
|
|
|
return (k, i, j)
|
|
|
|
|
|
# Method which turns the image shaped input to column shape.
|
|
# Used during the forward pass.
|
|
# Reference: CS231n Stanford
|
|
def image_to_column(images, filter_shape, stride, output_shape='same'):
|
|
filter_height, filter_width = filter_shape
|
|
|
|
pad_h, pad_w = determine_padding(filter_shape, output_shape)
|
|
|
|
# Add padding to the image
|
|
images_padded = np.pad(images, ((0, 0), (0, 0), pad_h, pad_w), mode='constant')
|
|
|
|
# Calculate the indices where the dot products are to be applied between weights
|
|
# and the image
|
|
k, i, j = get_im2col_indices(images.shape, filter_shape, (pad_h, pad_w), stride)
|
|
|
|
# Get content from image at those indices
|
|
cols = images_padded[:, k, i, j]
|
|
channels = images.shape[1]
|
|
# Reshape content into column shape
|
|
cols = cols.transpose(1, 2, 0).reshape(filter_height * filter_width * channels, -1)
|
|
return cols
|
|
|
|
|
|
|
|
# Method which turns the column shaped input to image shape.
|
|
# Used during the backward pass.
|
|
# Reference: CS231n Stanford
|
|
def column_to_image(cols, images_shape, filter_shape, stride, output_shape='same'):
|
|
batch_size, channels, height, width = images_shape
|
|
pad_h, pad_w = determine_padding(filter_shape, output_shape)
|
|
height_padded = height + np.sum(pad_h)
|
|
width_padded = width + np.sum(pad_w)
|
|
images_padded = np.zeros((batch_size, channels, height_padded, width_padded))
|
|
|
|
# Calculate the indices where the dot products are applied between weights
|
|
# and the image
|
|
k, i, j = get_im2col_indices(images_shape, filter_shape, (pad_h, pad_w), stride)
|
|
|
|
cols = cols.reshape(channels * np.prod(filter_shape), -1, batch_size)
|
|
cols = cols.transpose(2, 0, 1)
|
|
# Add column content to the images at the indices
|
|
np.add.at(images_padded, (slice(None), k, i, j), cols)
|
|
|
|
# Return image without padding
|
|
return images_padded[:, :, pad_h[0]:height+pad_h[0], pad_w[0]:width+pad_w[0]]
|