TheAlgorithms-Python

1
"""
2
 - - - - - -- - - - - - - - - - - - - - - - - - - - - - -
3
Name - - CNN - Convolution Neural Network For Photo Recognizing
4
Goal - - Recognize Handing Writing Word Photo
5
Detail: Total 5 layers neural network
6
        * Convolution layer
7
        * Pooling layer
8
        * Input layer layer of BP
9
        * Hidden layer of BP
10
        * Output layer of BP
11
Author: Stephen Lee
12
Github: 245885195@qq.com
13
Date: 2017.9.20
14
- - - - - -- - - - - - - - - - - - - - - - - - - - - - -
15
"""
16

17
import pickle
18

19
import numpy as np
20
from matplotlib import pyplot as plt
21

22

23
class CNN:
24
    def __init__(
25
        self, conv1_get, size_p1, bp_num1, bp_num2, bp_num3, rate_w=0.2, rate_t=0.2
26
    ):
27
        """
28
        :param conv1_get: [a,c,d], size, number, step of convolution kernel
29
        :param size_p1: pooling size
30
        :param bp_num1: units number of flatten layer
31
        :param bp_num2: units number of hidden layer
32
        :param bp_num3: units number of output layer
33
        :param rate_w: rate of weight learning
34
        :param rate_t: rate of threshold learning
35
        """
36
        self.num_bp1 = bp_num1
37
        self.num_bp2 = bp_num2
38
        self.num_bp3 = bp_num3
39
        self.conv1 = conv1_get[:2]
40
        self.step_conv1 = conv1_get[2]
41
        self.size_pooling1 = size_p1
42
        self.rate_weight = rate_w
43
        self.rate_thre = rate_t
44
        rng = np.random.default_rng()
45
        self.w_conv1 = [
46
            np.asmatrix(-1 * rng.random((self.conv1[0], self.conv1[0])) + 0.5)
47
            for i in range(self.conv1[1])
48
        ]
49
        self.wkj = np.asmatrix(-1 * rng.random((self.num_bp3, self.num_bp2)) + 0.5)
50
        self.vji = np.asmatrix(-1 * rng.random((self.num_bp2, self.num_bp1)) + 0.5)
51
        self.thre_conv1 = -2 * rng.random(self.conv1[1]) + 1
52
        self.thre_bp2 = -2 * rng.random(self.num_bp2) + 1
53
        self.thre_bp3 = -2 * rng.random(self.num_bp3) + 1
54

55
    def save_model(self, save_path):
56
        # save model dict with pickle
57
        model_dic = {
58
            "num_bp1": self.num_bp1,
59
            "num_bp2": self.num_bp2,
60
            "num_bp3": self.num_bp3,
61
            "conv1": self.conv1,
62
            "step_conv1": self.step_conv1,
63
            "size_pooling1": self.size_pooling1,
64
            "rate_weight": self.rate_weight,
65
            "rate_thre": self.rate_thre,
66
            "w_conv1": self.w_conv1,
67
            "wkj": self.wkj,
68
            "vji": self.vji,
69
            "thre_conv1": self.thre_conv1,
70
            "thre_bp2": self.thre_bp2,
71
            "thre_bp3": self.thre_bp3,
72
        }
73
        with open(save_path, "wb") as f:
74
            pickle.dump(model_dic, f)
75

76
        print(f"Model saved: {save_path}")
77

78
    @classmethod
79
    def read_model(cls, model_path):
80
        # read saved model
81
        with open(model_path, "rb") as f:
82
            model_dic = pickle.load(f)  # noqa: S301
83

84
        conv_get = model_dic.get("conv1")
85
        conv_get.append(model_dic.get("step_conv1"))
86
        size_p1 = model_dic.get("size_pooling1")
87
        bp1 = model_dic.get("num_bp1")
88
        bp2 = model_dic.get("num_bp2")
89
        bp3 = model_dic.get("num_bp3")
90
        r_w = model_dic.get("rate_weight")
91
        r_t = model_dic.get("rate_thre")
92
        # create model instance
93
        conv_ins = CNN(conv_get, size_p1, bp1, bp2, bp3, r_w, r_t)
94
        # modify model parameter
95
        conv_ins.w_conv1 = model_dic.get("w_conv1")
96
        conv_ins.wkj = model_dic.get("wkj")
97
        conv_ins.vji = model_dic.get("vji")
98
        conv_ins.thre_conv1 = model_dic.get("thre_conv1")
99
        conv_ins.thre_bp2 = model_dic.get("thre_bp2")
100
        conv_ins.thre_bp3 = model_dic.get("thre_bp3")
101
        return conv_ins
102

103
    def sig(self, x):
104
        return 1 / (1 + np.exp(-1 * x))
105

106
    def do_round(self, x):
107
        return round(x, 3)
108

109
    def convolute(self, data, convs, w_convs, thre_convs, conv_step):
110
        # convolution process
111
        size_conv = convs[0]
112
        num_conv = convs[1]
113
        size_data = np.shape(data)[0]
114
        # get the data slice of original image data, data_focus
115
        data_focus = []
116
        for i_focus in range(0, size_data - size_conv + 1, conv_step):
117
            for j_focus in range(0, size_data - size_conv + 1, conv_step):
118
                focus = data[
119
                    i_focus : i_focus + size_conv, j_focus : j_focus + size_conv
120
                ]
121
                data_focus.append(focus)
122
        # calculate the feature map of every single kernel, and saved as list of matrix
123
        data_featuremap = []
124
        size_feature_map = int((size_data - size_conv) / conv_step + 1)
125
        for i_map in range(num_conv):
126
            featuremap = []
127
            for i_focus in range(len(data_focus)):
128
                net_focus = (
129
                    np.sum(np.multiply(data_focus[i_focus], w_convs[i_map]))
130
                    - thre_convs[i_map]
131
                )
132
                featuremap.append(self.sig(net_focus))
133
            featuremap = np.asmatrix(featuremap).reshape(
134
                size_feature_map, size_feature_map
135
            )
136
            data_featuremap.append(featuremap)
137

138
        # expanding the data slice to One dimenssion
139
        focus1_list = []
140
        for each_focus in data_focus:
141
            focus1_list.extend(self.Expand_Mat(each_focus))
142
        focus_list = np.asarray(focus1_list)
143
        return focus_list, data_featuremap
144

145
    def pooling(self, featuremaps, size_pooling, pooling_type="average_pool"):
146
        # pooling process
147
        size_map = len(featuremaps[0])
148
        size_pooled = int(size_map / size_pooling)
149
        featuremap_pooled = []
150
        for i_map in range(len(featuremaps)):
151
            feature_map = featuremaps[i_map]
152
            map_pooled = []
153
            for i_focus in range(0, size_map, size_pooling):
154
                for j_focus in range(0, size_map, size_pooling):
155
                    focus = feature_map[
156
                        i_focus : i_focus + size_pooling,
157
                        j_focus : j_focus + size_pooling,
158
                    ]
159
                    if pooling_type == "average_pool":
160
                        # average pooling
161
                        map_pooled.append(np.average(focus))
162
                    elif pooling_type == "max_pooling":
163
                        # max pooling
164
                        map_pooled.append(np.max(focus))
165
            map_pooled = np.asmatrix(map_pooled).reshape(size_pooled, size_pooled)
166
            featuremap_pooled.append(map_pooled)
167
        return featuremap_pooled
168

169
    def _expand(self, data):
170
        # expanding three dimension data to one dimension list
171
        data_expanded = []
172
        for i in range(len(data)):
173
            shapes = np.shape(data[i])
174
            data_listed = data[i].reshape(1, shapes[0] * shapes[1])
175
            data_listed = data_listed.getA().tolist()[0]
176
            data_expanded.extend(data_listed)
177
        data_expanded = np.asarray(data_expanded)
178
        return data_expanded
179

180
    def _expand_mat(self, data_mat):
181
        # expanding matrix to one dimension list
182
        data_mat = np.asarray(data_mat)
183
        shapes = np.shape(data_mat)
184
        data_expanded = data_mat.reshape(1, shapes[0] * shapes[1])
185
        return data_expanded
186

187
    def _calculate_gradient_from_pool(
188
        self, out_map, pd_pool, num_map, size_map, size_pooling
189
    ):
190
        """
191
        calculate the gradient from the data slice of pool layer
192
        pd_pool: list of matrix
193
        out_map: the shape of data slice(size_map*size_map)
194
        return: pd_all: list of matrix, [num, size_map, size_map]
195
        """
196
        pd_all = []
197
        i_pool = 0
198
        for i_map in range(num_map):
199
            pd_conv1 = np.ones((size_map, size_map))
200
            for i in range(0, size_map, size_pooling):
201
                for j in range(0, size_map, size_pooling):
202
                    pd_conv1[i : i + size_pooling, j : j + size_pooling] = pd_pool[
203
                        i_pool
204
                    ]
205
                    i_pool = i_pool + 1
206
            pd_conv2 = np.multiply(
207
                pd_conv1, np.multiply(out_map[i_map], (1 - out_map[i_map]))
208
            )
209
            pd_all.append(pd_conv2)
210
        return pd_all
211

212
    def train(
213
        self, patterns, datas_train, datas_teach, n_repeat, error_accuracy, draw_e=bool
214
    ):
215
        # model training
216
        print("----------------------Start Training-------------------------")
217
        print((" - - Shape: Train_Data  ", np.shape(datas_train)))
218
        print((" - - Shape: Teach_Data  ", np.shape(datas_teach)))
219
        rp = 0
220
        all_mse = []
221
        mse = 10000
222
        while rp < n_repeat and mse >= error_accuracy:
223
            error_count = 0
224
            print(f"-------------Learning Time {rp}--------------")
225
            for p in range(len(datas_train)):
226
                # print('------------Learning Image: %d--------------'%p)
227
                data_train = np.asmatrix(datas_train[p])
228
                data_teach = np.asarray(datas_teach[p])
229
                data_focus1, data_conved1 = self.convolute(
230
                    data_train,
231
                    self.conv1,
232
                    self.w_conv1,
233
                    self.thre_conv1,
234
                    conv_step=self.step_conv1,
235
                )
236
                data_pooled1 = self.pooling(data_conved1, self.size_pooling1)
237
                shape_featuremap1 = np.shape(data_conved1)
238
                """
239
                print('  -----original shape   ', np.shape(data_train))
240
                print('  ---- after convolution  ',np.shape(data_conv1))
241
                print('  -----after pooling  ',np.shape(data_pooled1))
242
               """
243
                data_bp_input = self._expand(data_pooled1)
244
                bp_out1 = data_bp_input
245

246
                bp_net_j = np.dot(bp_out1, self.vji.T) - self.thre_bp2
247
                bp_out2 = self.sig(bp_net_j)
248
                bp_net_k = np.dot(bp_out2, self.wkj.T) - self.thre_bp3
249
                bp_out3 = self.sig(bp_net_k)
250

251
                # --------------Model Leaning ------------------------
252
                # calculate error and gradient---------------
253
                pd_k_all = np.multiply(
254
                    (data_teach - bp_out3), np.multiply(bp_out3, (1 - bp_out3))
255
                )
256
                pd_j_all = np.multiply(
257
                    np.dot(pd_k_all, self.wkj), np.multiply(bp_out2, (1 - bp_out2))
258
                )
259
                pd_i_all = np.dot(pd_j_all, self.vji)
260

261
                pd_conv1_pooled = pd_i_all / (self.size_pooling1 * self.size_pooling1)
262
                pd_conv1_pooled = pd_conv1_pooled.T.getA().tolist()
263
                pd_conv1_all = self._calculate_gradient_from_pool(
264
                    data_conved1,
265
                    pd_conv1_pooled,
266
                    shape_featuremap1[0],
267
                    shape_featuremap1[1],
268
                    self.size_pooling1,
269
                )
270
                # weight and threshold learning process---------
271
                # convolution layer
272
                for k_conv in range(self.conv1[1]):
273
                    pd_conv_list = self._expand_mat(pd_conv1_all[k_conv])
274
                    delta_w = self.rate_weight * np.dot(pd_conv_list, data_focus1)
275

276
                    self.w_conv1[k_conv] = self.w_conv1[k_conv] + delta_w.reshape(
277
                        (self.conv1[0], self.conv1[0])
278
                    )
279

280
                    self.thre_conv1[k_conv] = (
281
                        self.thre_conv1[k_conv]
282
                        - np.sum(pd_conv1_all[k_conv]) * self.rate_thre
283
                    )
284
                # all connected layer
285
                self.wkj = self.wkj + pd_k_all.T * bp_out2 * self.rate_weight
286
                self.vji = self.vji + pd_j_all.T * bp_out1 * self.rate_weight
287
                self.thre_bp3 = self.thre_bp3 - pd_k_all * self.rate_thre
288
                self.thre_bp2 = self.thre_bp2 - pd_j_all * self.rate_thre
289
                # calculate the sum error of all single image
290
                errors = np.sum(abs(data_teach - bp_out3))
291
                error_count += errors
292
                # print('   ----Teach      ',data_teach)
293
                # print('   ----BP_output  ',bp_out3)
294
            rp = rp + 1
295
            mse = error_count / patterns
296
            all_mse.append(mse)
297

298
        def draw_error():
299
            yplot = [error_accuracy for i in range(int(n_repeat * 1.2))]
300
            plt.plot(all_mse, "+-")
301
            plt.plot(yplot, "r--")
302
            plt.xlabel("Learning Times")
303
            plt.ylabel("All_mse")
304
            plt.grid(True, alpha=0.5)
305
            plt.show()
306

307
        print("------------------Training Complished---------------------")
308
        print((" - - Training epoch: ", rp, f"     - - Mse: {mse:.6f}"))
309
        if draw_e:
310
            draw_error()
311
        return mse
312

313
    def predict(self, datas_test):
314
        # model predict
315
        produce_out = []
316
        print("-------------------Start Testing-------------------------")
317
        print((" - - Shape: Test_Data  ", np.shape(datas_test)))
318
        for p in range(len(datas_test)):
319
            data_test = np.asmatrix(datas_test[p])
320
            data_focus1, data_conved1 = self.convolute(
321
                data_test,
322
                self.conv1,
323
                self.w_conv1,
324
                self.thre_conv1,
325
                conv_step=self.step_conv1,
326
            )
327
            data_pooled1 = self.pooling(data_conved1, self.size_pooling1)
328
            data_bp_input = self._expand(data_pooled1)
329

330
            bp_out1 = data_bp_input
331
            bp_net_j = bp_out1 * self.vji.T - self.thre_bp2
332
            bp_out2 = self.sig(bp_net_j)
333
            bp_net_k = bp_out2 * self.wkj.T - self.thre_bp3
334
            bp_out3 = self.sig(bp_net_k)
335
            produce_out.extend(bp_out3.getA().tolist())
336
        res = [list(map(self.do_round, each)) for each in produce_out]
337
        return np.asarray(res)
338

339
    def convolution(self, data):
340
        # return the data of image after convoluting process so we can check it out
341
        data_test = np.asmatrix(data)
342
        data_focus1, data_conved1 = self.convolute(
343
            data_test,
344
            self.conv1,
345
            self.w_conv1,
346
            self.thre_conv1,
347
            conv_step=self.step_conv1,
348
        )
349
        data_pooled1 = self.pooling(data_conved1, self.size_pooling1)
350

351
        return data_conved1, data_pooled1
352

353

354
if __name__ == "__main__":
355
    """
356
    I will put the example on other file
357
    """
358