conv1:32*32->6*28*28->relu:f(x)=max(0,x)
kenel:6*(1*5*5)
param:6*(1*5*5+1)
conn:6*(1*5*5+1)*28*28
pool2:6*28*28->6*14*14
kenel:6*(2*2)
param:6*(2*2+1)
conn:6*(2*2+1)*14*14
conv3:6*14*14->16*10*10->relu
kenel:16*(5*5)
param:6*(3*5*5+1)+6*(4*5*5+1)+3*(4*5*5+1)+1*(5*5*5+1)
conn:[6*(3*5*5+1)+6*(4*5*5+1)+3*(4*5*5+1)+1*(5*5*5+1)]*10*10
pool4:16*10*10->16*5*5
kenel:16*(2*2)
param:16*(2*2+1)
conn:16*(2*2+1)*5*5
fc5:16*5*5->120*1*1->relu
kenel:120*(16*5*5)
param:120*(16*5*5+1)
conn:120*(16*5*5+1)*1*1
fc6:120*1*1->84*1*1->relu
kenel:84*(120*1*1)
param:84*(120*1*1+1)
conn:84*(120*1*1+1)*1*1
fc7:84*1*1->10*1*1
kenel:10*(84*1*1)
param:10*(84*1*1+1)
conn:10*(84*1*1+1)*1*1
rule:src*a*a->dest*b*b
kenel:dest*(src*k*k)
param:dest*(src*k*k+1)
conn:dest*(src*k*k+1)*b*b
# -*- coding: utf-8 -*-
'''
Name：LeNet5.py
Author: JoyZhou
Date:20190721
'''
import numpy as np
from scipy.signal import convolve2d
from skimage.measure import block_reduce
import fetch_MNIST
class LeNet5(object):
def __init__(self, lr=0.1):
# learn rate:
self.lr = lr
# conv1
self.conv1 = xavier_init(6, 1, 5, 5)
# pool2
self.pool2 = [2, 2]
# conv3
self.conv3 = xavier_init(16, 6, 5, 5)
# pool4
self.pool4 = [2, 2]
# conn5
self.fc5 = xavier_init(256, 200, fc=True)
# conn6
self.fc6 = xavier_init(200, 10, fc=True)
def forward_prop(self, input_data):
self.l0 = np.expand_dims(input_data, axis=1) / 255 # 64*1*28*28
self.l1 = self.convolution(self.l0, self.conv1) # 64*1*28*28 6*1*5*5 -> 64*6*24*24
self.l2 = self.mean_pool(self.l1, self.pool2) # 64*6*24*24 -> 64*6*12*12
self.l3 = self.convolution(self.l2, self.conv3) # 64*6*12*12 16*6*5*5 -> 64*16*8*8
self.l4 = self.mean_pool(self.l3, self.pool4) # 64*16*8*8 -> 64*16*4*4
self.l5 = self.fully_connect(self.l4, self.fc5) # 64*256 256*200 -> 64*200
self.l6 = self.relu(self.l5) # 64*200
self.l7 = self.fully_connect(self.l6, self.fc6) # 64*200 200*10 -> 64*10
self.l8 = self.relu(self.l7) # 64*10
self.l9 = self.softmax(self.l8) # 64*10
return self.l9
def backward_prop(self, softmax_output, output_label):
l8_delta = (output_label - softmax_output) / softmax_output.shape[0] # 64*10
l7_delta = self.relu(self.l8, l8_delta, deriv=True) # 64*10
l6_delta, self.fc6 = self.fully_connect(self.l6, self.fc6, l7_delta, deriv=True) # 64*200 200*10 64*10 -> 64*200 200*10
l5_delta = self.relu(self.l6, l6_delta, deriv=True) # 64*200
l4_delta, self.fc5 = self.fully_connect(self.l4, self.fc5, l5_delta, deriv=True) # 64*16*4*4 256*200 64*200 -> 64*16*4*4 256*200
l3_delta = self.mean_pool(self.l3, self.pool4, l4_delta, deriv=True) # 64*16*8*8 2*2 64*16*4*4 -> 64*16*8*8
l2_delta, self.conv3 = self.convolution(self.l2, self.conv3, l3_delta, deriv=True) # (batch_sz, 6, 12, 12)
l1_delta = self.mean_pool(self.l1, self.pool2, l2_delta, deriv=True) # (batch_sz, 6, 24, 24)
l0_delta, self.conv1 = self.convolution(self.l0, self.conv1, l1_delta, deriv=True) # (batch_sz, 1, 28, 28)
def convolution(self, input_map, kernal, front_delta=None, deriv=False):
N, C, W, H = input_map.shape # 1:64*1*28*28
K_N, K_C, K_W, K_H = kernal.shape # 1:6*1*5*5
if deriv == False:
feature_map = np.zeros((N, K_N, W-K_W+1, H-K_H+1)) # 1:64*6*24*24
for img in range(N): # img 64
for kId in range(K_N): # kenel 6
for cId in range(C): # channel 1
feature_map[img][kId] += \
convolve2d(input_map[img][cId], kernal[kId,cId,:,:], mode='valid') # 28*28 5*5 -> 24*24
return feature_map
else :
# front->back (propagate loss)
back_delta = np.zeros((N, C, W, H))
kernal_gradient = np.zeros((K_N, K_C, K_W, K_H))
padded_front_delta = \
np.pad(front_delta, [(0,0), (0,0), (K_W-1, K_H-1), (K_W-1, K_H-1)], mode='constant', constant_values=0)
for imgId in range(N):
for cId in range(C):
for kId in range(K_N):
back_delta[imgId][cId] += \
convolve2d(padded_front_delta[imgId][kId], kernal[kId,cId,::-1,::-1], mode='valid')
kernal_gradient[kId][cId] += \
convolve2d(front_delta[imgId][kId], input_map[imgId,cId,::-1,::-1], mode='valid')
# update weights
kernal += self.lr * kernal_gradient
return back_delta, kernal
def mean_pool(self, input_map, pool, front_delta=None, deriv=False):
N, C, W, H = input_map.shape
P_W, P_H = tuple(pool)
if deriv == False:
# feature_map = np.zeros((N, C, W/P_W, H/P_H))
feature_map = block_reduce(input_map, tuple((1, 1, P_W, P_H)), func=np.mean)
return feature_map
else :
# front->back (propagate loss)
back_delta = np.zeros((N, C, W, H))
back_delta = front_delta.repeat(P_W, axis = 2).repeat(P_H, axis = 3)
back_delta /= (P_W * P_H)
return back_delta
def fully_connect(self, input_data, fc, front_delta=None, deriv=False):
N = input_data.shape[0]
if deriv == False:
output_data = np.dot(input_data.reshape(N, -1), fc)
return output_data
else :
# 1:64*10 * 10*200 -> 64*200 2:64*200 * 200*256 -> 64*256 -> 64*16*4*4
back_delta = np.dot(front_delta, fc.T).reshape(input_data.shape) # 误差*参数转置
# update weights 1:64*200 -> 200*64 * 64*10 -> 200*10 2:64*16*4*4->64*256->256*64 * 64*200 -> 256*200
fc += self.lr * np.dot(input_data.reshape(N, -1).T, front_delta) # 输入转置*误差
return back_delta, fc
def relu(self, x, front_delta=None, deriv=False):
if deriv == False:
return x * (x > 0)
else :
# propagate loss
back_delta = front_delta * 1. * (x > 0)
return back_delta
def softmax(self, x):
y = list()
for t in x:
e_t = np.exp(t - np.max(t))
y.append(e_t / e_t.sum())
return np.array(y)
def xavier_init(c1, c2, w=1, h=1, fc=False):
fan_1 = c2 * w * h
fan_2 = c1 * w * h
ratio = np.sqrt(6.0 / (fan_1 + fan_2))
params = ratio * (2*np.random.random((c1, c2, w, h)) - 1)
if fc == True:
params = params.reshape(c1, c2)
return params
def convertToOneHot(labels):
oneHotLabels = np.zeros((labels.size, labels.max()+1))
oneHotLabels[np.arange(labels.size), labels] = 1
return oneHotLabels
def shuffle_dataset(data, label):
N = data.shape[0]
index = np.random.permutation(N)
x = data[index, :, :]; y = label[index, :]
return x, y
if __name__ == '__main__':
train_imgs = fetch_MNIST.load_train_images() # 60,000*28*28
train_labs = fetch_MNIST.load_train_labels().astype(int) # 60,000*1
# data
data_size = train_imgs.shape[0] # 60000
# batch
batch_size = 64
# learning rate
lr = 0.01
# max iteration
max_iter = 50000;
# total
iter_mod = int(data_size/batch_size) # 937
# one-hot
train_labs = convertToOneHot(train_labs) # 60000*10
# forward train
train = LeNet5(lr)
for iters in range(max_iter):
# start index
start_idx = (iters % iter_mod) * batch_size # 0,64,128...
# shuffle the dataset
if start_idx == 0:
train_imgs, train_labs = shuffle_dataset(train_imgs, train_labs)
input_data = train_imgs[start_idx : start_idx + batch_size] # [0:64],[64:128],[128,192]...
output_label = train_labs[start_idx : start_idx + batch_size]# [0:64],[64:128],[128,192]...
output_train = train.forward_prop(input_data) # 64*28*28
if iters % 50 == 0:
# correct_number/batch_size
correct_list = [ int(np.argmax(output_train[i])==np.argmax(output_label[i])) for i in range(batch_size) ]
accuracy = float(np.array(correct_list).sum()) / batch_size
# calculate loss
correct_prob = [ output_train[i][np.argmax(output_label[i])] for i in range(batch_size) ]
# relu
for i in range(len(correct_prob)):
if correct_prob[i] > 0:
continue
else:
correct_prob[i] = 0
loss = -1.0 * np.sum(np.log(correct_prob))
print ("The %d iters result:" % iters)
print ("The accuracy is %f The loss is %f " % (accuracy, loss))
train.backward_prop(output_train, output_label)
# encoding: utf-8
'''
Name： fetch_MNIST.py
'''
import numpy as np
import struct
import matplotlib.pyplot as plt
#data_path = '/Users/didi/Desktop/python_workspace/Neural Network/data/'
# 训练集文件
train_images_idx3_ubyte_file = './data/train-images-idx3-ubyte'
# 训练集标签文件
train_labels_idx1_ubyte_file = './data/train-labels-idx1-ubyte'
# 测试集文件
test_images_idx3_ubyte_file = './data/t10k-images-idx3-ubyte'
# 测试集标签文件
test_labels_idx1_ubyte_file = './data/t10k-labels-idx1-ubyte'
def decode_idx3_ubyte(idx3_ubyte_file):
"""
解析idx3文件的通用函数
:param idx3_ubyte_file: idx3文件路径
:return: 数据集
"""
# 读取二进制数据
bin_data = open(idx3_ubyte_file, 'rb').read()
# 解析文件头信息，依次为魔数、图片数量、每张图片高、每张图片宽
offset = 0
fmt_header = '>iiii'
magic_number, num_images, num_rows, num_cols = struct.unpack_from(fmt_header, bin_data, offset)
print ('模数:%d, 图片数量: %d张, 图片大小: %d*%d' % (magic_number, num_images, num_rows, num_cols))
# 解析数据集
image_size = num_rows * num_cols
offset += struct.calcsize(fmt_header)
fmt_image = '>' + str(image_size) + 'B'
images = np.empty((num_images, num_rows, num_cols))
for i in range(num_images):
if (i + 1) % 10000 == 0:
print ('已解析 %d' % (i + 1) + '张')
images[i] = np.array(struct.unpack_from(fmt_image, bin_data, offset)).reshape((num_rows, num_cols))
offset += struct.calcsize(fmt_image)
return images
def decode_idx1_ubyte(idx1_ubyte_file):
"""
解析idx1文件的通用函数
:param idx1_ubyte_file: idx1文件路径
:return: 数据集
"""
# 读取二进制数据
bin_data = open(idx1_ubyte_file, 'rb').read()
# 解析文件头信息，依次为魔数和标签数
offset = 0
fmt_header = '>ii'
magic_number, num_images = struct.unpack_from(fmt_header, bin_data, offset)
print ('模数:%d, 图片数量: %d张' % (magic_number, num_images))
# 解析数据集
offset += struct.calcsize(fmt_header)
fmt_image = '>B'
labels = np.empty(num_images)
for i in range(num_images):
if (i + 1) % 10000 == 0:
print ('已解析 %d' % (i + 1) + '张')
labels[i] = struct.unpack_from(fmt_image, bin_data, offset)[0]
offset += struct.calcsize(fmt_image)
return labels
def load_train_images(idx_ubyte_file=train_images_idx3_ubyte_file):
"""
TRAINING SET IMAGE FILE (train-images-idx3-ubyte):
[offset] [type] [value] [description]
0000 32 bit integer 0x00000803(2051) magic number
0004 32 bit integer 60000 number of images
0008 32 bit integer 28 number of rows
0012 32 bit integer 28 number of columns
0016 unsigned byte ?? pixel
0017 unsigned byte ?? pixel
........
xxxx unsigned byte ?? pixel
Pixels are organized row-wise. Pixel values are 0 to 255. 0 means background (white), 255 means foreground (black).
:param idx_ubyte_file: idx文件路径
:return: n*row*col维np.array对象，n为图片数量
"""
return decode_idx3_ubyte(idx_ubyte_file)
def load_train_labels(idx_ubyte_file=train_labels_idx1_ubyte_file):
"""
TRAINING SET LABEL FILE (train-labels-idx1-ubyte):
[offset] [type] [value] [description]
0000 32 bit integer 0x00000801(2049) magic number (MSB first)
0004 32 bit integer 60000 number of items
0008 unsigned byte ?? label
0009 unsigned byte ?? label
........
xxxx unsigned byte ?? label
The labels values are 0 to 9.
:param idx_ubyte_file: idx文件路径
:return: n*1维np.array对象，n为图片数量
"""
return decode_idx1_ubyte(idx_ubyte_file)
def load_test_images(idx_ubyte_file=test_images_idx3_ubyte_file):
"""
TEST SET IMAGE FILE (t10k-images-idx3-ubyte):
[offset] [type] [value] [description]
0000 32 bit integer 0x00000803(2051) magic number
0004 32 bit integer 10000 number of images
0008 32 bit integer 28 number of rows
0012 32 bit integer 28 number of columns
0016 unsigned byte ?? pixel
0017 unsigned byte ?? pixel
........
xxxx unsigned byte ?? pixel
Pixels are organized row-wise. Pixel values are 0 to 255. 0 means background (white), 255 means foreground (black).
:param idx_ubyte_file: idx文件路径
:return: n*row*col维np.array对象，n为图片数量
"""
return decode_idx3_ubyte(idx_ubyte_file)
def load_test_labels(idx_ubyte_file=test_labels_idx1_ubyte_file):
"""
TEST SET LABEL FILE (t10k-labels-idx1-ubyte):
[offset] [type] [value] [description]
0000 32 bit integer 0x00000801(2049) magic number (MSB first)
0004 32 bit integer 10000 number of items
0008 unsigned byte ?? label
0009 unsigned byte ?? label
........
xxxx unsigned byte ?? label
The labels values are 0 to 9.
:param idx_ubyte_file: idx文件路径
:return: n*1维np.array对象，n为图片数量
"""
return decode_idx1_ubyte(idx_ubyte_file)
def run():
train_images = load_train_images() # (60000, 28, 28) 0~255
train_labels = load_train_labels() # (60000,) 1~10
# test_images = load_test_images()
# test_labels = load_test_labels()
print (type(train_images), train_images.shape)
print (type(train_labels), train_labels.shape)
# 查看前十个数据及其标签以读取是否正确
for i in range(10):
print (train_labels[i])
print (np.max(train_images), np.min(train_images))
plt.imshow(train_images[i], cmap='gray')
plt.show()
if __name__ == '__main__':
run()
