Convolution in 2D

Code (using only numpy)

class ConvolutionLayer:
  def __init__(self, in_channels, filter_size, numfilters, stride):
    # Method to initialize a Convolution Layer
    # Parameters
    # in_channels - list of 3 elements denoting size of input for convolution layer
    # filter_size - list of 2 elements denoting size of kernel weights for convolution layer
    # numfilters  - number of feature maps (denoting output depth)
    # stride    - stride to used during convolution forward pass
    self.in_depth, self.in_row, self.in_col = in_channels
    self.filter_row, self.filter_col = filter_size
    self.stride = stride

    self.out_depth = numfilters
    self.out_row = int((self.in_row - self.filter_row)/self.stride + 1)
    self.out_col = int((self.in_col - self.filter_col)/self.stride + 1)

    # Stores the outgoing summation of weights * feautres
    self.data = None

    # Initializes the Weights and Biases using a Normal Distribution with Mean 0 and Standard Deviation 0.1
    self.weights = np.random.normal(0,0.1, (self.out_depth, self.in_depth, self.filter_row, self.filter_col))
    self.biases = np.random.normal(0,0.1,self.out_depth)


    ## This can be done in self.forwardpass() also but placing it here is more time-efficient.
    ## k, i, j are indices of X [(depth, row, col)] accessed in row-major order of each convolution stacked in the order of row-major convolutions
    ## (Slightly difficult to describe, but very intuitive!)

    # Here i finally is a (in_depth * filter_row * filter_col) x (out_row * out_col)
    i_filter_indices = np.tile(np.repeat(np.arange(self.filter_row), self.filter_col), self.in_depth).reshape(-1, 1).astype(int)
    i_convolve_indices = self.stride * np.repeat(np.arange(self.out_row), self.out_col).reshape(1, -1).astype(int)
    self.i = i_filter_indices + i_convolve_indices # Using broadcasting to our advantage

    # Similarly for j. Here also we get a (in_depth * filter_row * filter_col) x (out_row * out_col)
    j_filter_indices = np.tile(np.arange(self.filter_col), self.filter_row * self.in_depth).reshape(-1, 1).astype(int)
    j_convolve_indices = self.stride * np.tile(np.arange(self.out_col), self.out_row).reshape(1, -1).astype(int)
    self.j = j_filter_indices + j_convolve_indices

    # For k, we save memory by using broadcasting to our advantage
    # Here k is a (in_depth * filter_row * filter_col) x 1 matrix, which is OK since broadcasting while using k, i, j together will give proper full k matrix (alteast for intuition)
    # This could be done here, since k has same columns if written as a full matrix like i, j on a paper.
    self.k = np.repeat(np.arange(self.in_depth), self.filter_row * self.filter_col).reshape(-1, 1).astype(int)


  def forwardpass(self, X):
    # print('Forward CN ',self.weights.shape)
    # Input
    # X : Activations from previous layer/input
    # Output
    # activations : Activations after one forward pass through this layer
    n = X.shape[0]  # batch size
    # INPUT activation matrix      :[n X self.in_channels[0] X self.in_channels[1] X self.in_channels[2]]
    # OUTPUT activation matrix    :[n X self.outputsize[0] X self.outputsize[1] X self.numfilters]

    ## The basic idea I'm doing here is that since weights are shared, each input value interacts with multiple weights to possibly contribute to different output activations
    ## So, we duplicate these X entries (Yes! indeed requires quite higher memory), in a very specific order so that convolution boils down to X_flattened and W_flattened interacting exactly same as a linear layer
    ## After working on flattened space, we reshape and go back properly to our actual depth x row x col spaces.

    X_flattened = X[:, self.k, self.i, self.j].transpose(1, 2, 0).reshape(self.in_depth * self.filter_row * self.filter_col, -1)
    W_flattened = self.weights.reshape(self.out_depth, -1)
    out_activation = sigmoid(W_flattened @ X_flattened + self.biases.reshape(-1, 1))

    # We'll go back to original shapes now!
    out_activation = out_activation.reshape(self.out_depth, self.out_row, self.out_col, n).transpose(3, 0, 1, 2)
    self.data = (out_activation, X_flattened)
    return out_activation

  def backwardpass(self, lr, activation_prev, delta):
    # Input
    # lr : learning rate of the neural network
    # activation_prev : Activations from previous layer
    # delta : del_Error/ del_activation_curr
    # Output
    # new_delta : del_Error/ del_activation_prev

    # Update self.weights and self.biases for this layer by backpropagation
    n = activation_prev.shape[0] # batch size

    ## Just similar ot forward propagation, working in flattened space and going back to depth x row x col space. This time getting the new_delta is quite tricky.
    ## We use the np.add.at() function which we use in such a way that same (k, i, j) tuples that contributed more than once, their corresponding contributing gradients gets added up.

    # X_flattened = activation_prev[:, self.k, self.i, self.j].transpose(1, 2, 0).reshape(self.in_depth * self.filter_row * self.filter_col, -1)
    out_activation, X_flattened = self.data # X_flattened can also be obtained as above, but storing it in self.data for reuse is more time efficient.
    del_in = delta * out_activation * (1 - out_activation)

    del_in_flattened = del_in.transpose(1, 2, 3, 0).reshape(self.out_depth, -1)
    W_flattened = self.weights.reshape(self.out_depth, -1)
    new_delta_flattened = W_flattened.T @ del_in_flattened
    new_delta_flattened = new_delta_flattened.reshape(self.in_depth * self.filter_row * self.filter_col, -1, n).transpose(2, 0, 1)

    # We'll go back to original shapes now!
    new_delta = np.zeros((n, self.in_depth, self.in_row, self.in_col), dtype=new_delta_flattened.dtype) # Ensure same type so that np.add.at() doesn't cause any type troubles.
    np.add.at(new_delta, (slice(None), self.k, self.i, self.j), new_delta_flattened) # Described above

    self.biases -= lr * np.sum(del_in, axis=(0, 2, 3))
    self.weights -= lr * (del_in_flattened @ X_flattened.T).reshape(self.weights.shape)
    return new_delta