MLProjects/cnn.py at master · tushar-c/MLProjects · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
""" A convolutional neural network on the MNIST dataset. Note that the terminology and definitions are taken from the following
    paper, link: https://pdfs.semanticscholar.org/5d79/11c93ddcb34cac088d99bd0cae9124e5dcd1.pdf """


import conv_utils
import numpy as np


def forward_conv_pass(input_, conv_layers, kernels, biases, poolings, kernel_strides=1, pool_stride=2):
    curr_input = input_
    convolution_cache = []
    activation_cache = []
    pooling_cache = []
    for c in range(conv_layers):
        convolution = conv_utils.conv2d(curr_input, kernels[c], biases[c], kernel_strides)
        activated_convolution = conv_utils.stable_sigmoid(convolution)
        pooling = conv_utils.avg_pooling(activated_convolution, poolings[c], pool_stride)
        convolution_cache.append(convolution)
        activation_cache.append(activated_convolution)
        pooling_cache.append(pooling)
        curr_input = pooling
    return [convolution_cache, activation_cache, pooling_cache]


def backward_conv_pass(input_, preactivated_output, feature, label, conv_layers, kernels, weights, biases,
                       fc_layer,pooling_cache, convolution_cache, final_output_shape, observed_probs,
                       kernel_stride=1):

    kernel_gradients, biases_gradients, output_layer_weights_gradients, output_layer_biases_gradients = [], [], [], []
    output_layer_weights = weights
    output_layer_biases = biases
    error = (input_ - label) * conv_utils.grad_softmax(preactivated_output)
    delta_output_layer_weights = np.matmul(error, fc_layer.T)
    delta_output_layer_biases = error
    output_layer_weights_gradients.append((output_layer_weights, delta_output_layer_weights))
    output_layer_biases_gradients.append((output_layer_biases, delta_output_layer_biases))
    delta_fc_layer = np.matmul(delta_output_layer_weights.T, error)
    delta_final_pool = delta_fc_layer.reshape(final_output_shape, final_output_shape)
    delta_final_conv = conv_utils.upscale(delta_final_pool, convolution_cache[-1])
    delta_final_conv_sigma = delta_final_conv * conv_utils.sigmoid_gradient(convolution_cache[-1])
    if conv_layers > 1:
        delta_final_kernel = conv_utils.conv2d(np.rot90(pooling_cache[-2], 2), delta_final_conv_sigma,
                                               biases[-1], kernel_stride)
        delta_final_conv_sigma = conv_utils.stability_check(delta_final_conv_sigma)
        delta_final_kernel = conv_utils.stability_check(delta_final_kernel)
        delta_final_bias = np.sum(delta_final_conv_sigma)
        kernel_gradients.append((kernels[-1], delta_final_kernel))
        biases_gradients.append((biases[-1], delta_final_bias))
        curr_delta_conv_sigma = delta_final_conv_sigma
        pooling_cache.append(feature)
        for j in range(conv_layers - 2, -1, -1):
            delta_conv_sigma = curr_delta_conv_sigma
            delta_curr_pool = conv_utils.conv2d(delta_conv_sigma, np.rot90(kernels[j + 1], 2), biases[j + 1], kernel_stride)
            delta_curr_conv = conv_utils.upscale(delta_curr_pool, convolution_cache[j])
            delta_curr_conv = conv_utils.stability_check(delta_curr_conv)
            curr_delta_conv_sigma = delta_curr_conv * conv_utils.sigmoid_gradient(convolution_cache[j])
            delta_curr_kernel = conv_utils.conv2d(np.rot90(pooling_cache[j - 1], 2), curr_delta_conv_sigma, 0, kernel_stride)
            delta_curr_kernel = conv_utils.stability_check(delta_curr_kernel)
            delta_curr_bias = np.sum(curr_delta_conv_sigma)
            kernel_gradients.append((kernels[j], delta_curr_kernel))
            biases_gradients.append((biases[j], delta_curr_bias))
    else:
        delta_final_kernel = conv_utils.conv2d(np.rot90(feature, 2), delta_final_conv_sigma,
                                               biases[-1], kernel_stride)
        delta_final_conv_sigma = conv_utils.stability_check(delta_final_conv_sigma)
        delta_final_kernel = conv_utils.stability_check(delta_final_kernel)
        delta_final_bias = np.sum(delta_final_conv_sigma)
        kernel_gradients.append((kernels[-1], delta_final_kernel))
        biases_gradients.append((biases[-1], delta_final_bias))

    return kernel_gradients, biases_gradients, output_layer_weights_gradients, output_layer_biases_gradients


def train(features, labels, conv_layers, kernels, biases, poolings, eta, observed_probs,
          input_shape=28, output_classes=10, epochs=1000, sample_size=100):

    print('training cnn with {} layer(s) for {} epochs with learning rate {} on {} samples'.format(conv_layers, epochs,
                                                                                                   eta, len(features)))
    final_output_shape = conv_utils.infer_output_layer_shape(input_shape, conv_layers, kernels, poolings,
                                                            kernel_strides=1, pool_stride=2)
    output_layer_weights = conv_utils.output_layer_weights_biases(output_classes, final_output_shape)[0]
    output_layer_biases = conv_utils.output_layer_weights_biases(output_classes, final_output_shape)[1]
    for e in range(epochs):
        mse = 0
        correct = 0
        for d in range(len(features)):
            if d % 2000 == 0 and d != 0:
                print('epoch {} / {}, sample {} / {}'.format(e + 1, epochs, d, len(features)))
            forward_pass = forward_conv_pass(features[d], conv_layers, kernels, biases, poolings)
            convolution_cache, activation_cache, pooling_cache = forward_pass[0], forward_pass[1], forward_pass[2]
            final_pooling_layer = pooling_cache[-1]
            fc_layer = np.ravel(final_pooling_layer).reshape(final_pooling_layer.shape[0] * final_pooling_layer.shape[1], 1)
            preactivated_output = np.matmul(output_layer_weights, fc_layer) + output_layer_biases
            final_output = conv_utils.stable_softmax(preactivated_output)
            mse += conv_utils.mse(conv_utils.softmax_prediction(final_output), labels[d])
            if conv_utils.check_prediction(final_output, labels[d]) == 0:
                correct += 1
            backward_pass = backward_conv_pass(final_output, preactivated_output, features[d], labels[d],
                                               conv_layers, kernels, output_layer_weights, output_layer_biases,
                                                fc_layer, pooling_cache, convolution_cache, final_output_shape,
                                               observed_probs)

            kernels, biases, output_layer_weights, output_layer_biases = conv_utils.update_params(kernels, biases,
                                                                                                  backward_pass, eta)
        print('epoch {} / {}, epoch accuracy = {} %, correct predictions = {} out of {}'\
              .format(e + 1, epochs,(correct * 100) / len(features), correct, len(features)))
    return kernels, biases, output_layer_weights, output_layer_biases


# here sliced indicates how many data points we want from the whole dataset. Default value will use the whole dataset.
mnist_train_data = conv_utils.get_mnist_data(sliced=100)
train_features = mnist_train_data[0]
train_labels = mnist_train_data[1]
output_classes = 10
conv_layers = 2
eta = 0.504
epochs = 50
observed_probs = conv_utils.get_empirical_probs(output_classes, train_labels)
kernels = conv_utils.init_kernels(conv_layers, shape=2)
biases = conv_utils.init_biases(conv_layers)
poolings = conv_utils.init_poolings(conv_layers)
train_cnn = train(train_features, train_labels, conv_layers, kernels, biases, poolings, eta, observed_probs,
                  output_classes=output_classes, epochs=epochs)