深度学习之BP神经网络识别手写数字（五）

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import struct
from bp import *
from datetime import datetime
import matplotlib.pyplot as plt

# 数据加载器基类
class Loader(object):
    def __init__(self, path, count):
        '''
        初始化加载器
        path: 数据文件路径
        count: 文件中的样本个数
        '''
        self.path = path
        self.count = count
    def get_file_content(self):
        '''
        读取文件内容
        '''
        f = open(self.path, 'rb')
        content = f.read()
        f.close()
        return content
    def to_int(self, byte):
        '''
        将unsigned byte字符转换为整数
        '''
        return struct.unpack('B', byte)[0]
# 图像数据加载器
class ImageLoader(Loader):
    def get_picture(self, content, index):
        '''
        内部函数，从文件中获取图像
        '''
        ##从偏移量位置开始读取有效数据
        start = index * 28 * 28 + 16

        picture = []
        for i in range(28):
            picture.append([])
            for j in range(28): 
                picture[i].append(
                    self.to_int(content[start + i * 28 + j]))
        #picture 结构 二位数组 28*28
        return picture
    def get_one_sample(self, picture):
        '''
        内部函数，将图像转化为样本的输入向量
        '''
        sample = []
        for i in range(28):
            for j in range(28):
                sample.append(picture[i][j])
        #将样本
        return sample
    def load(self):
        '''
        加载数据文件，获得全部样本的输入向量
        '''
        #读取所有的图片样本
        content = self.get_file_content()
        data_set = []
        for index in range(self.count):
            data_set.append(
                self.get_one_sample(
                    self.get_picture(content, index)))
        return data_set
# 标签数据加载器
class LabelLoader(Loader):
    def load(self):
        '''
        加载数据文件，获得全部样本的标签向量
        '''
        content = self.get_file_content()
        labels = []
        for index in range(self.count):
            labels.append(self.norm(content[index + 8]))
        return labels
    def norm(self, label):
        '''
        内部函数，将一个值转换为10维标签向量
        '''
        label_vec = []
        label_value = self.to_int(label)
        for i in range(10):
            if i == label_value:
                label_vec.append(0.9)
            else:
                label_vec.append(0.1)
        return label_vec

def get_training_data_set():
    '''
    获得训练数据集
    '''
    image_loader = ImageLoader('train-images.idx3-ubyte', 60000)
    label_loader = LabelLoader('train-labels.idx1-ubyte', 60000)
    return image_loader.load(), label_loader.load()

def get_test_data_set():
    '''
    获得测试数据集
    '''
    image_loader = ImageLoader('t10k-images.idx3-ubyte', 10000)
    label_loader = LabelLoader('t10k-labels.idx1-ubyte', 10000)
    return image_loader.load(), label_loader.load()

# coding=utf-8
import numpy as np
import getImage as gim
#全连接神经网络层类
class BPLayer(object):
    def __init__(self, input_size, output_size, activator):
        '''
        input_siez:本层输入向量维度
        output_size:本层输出向量维度
        activator:本层激活函数
        '''
        self.input_size = input_size;
        self.output_size = output_size;
        self.activator = activator;
        #权值数组(范围-0.1~0.1)
        self.W = (np.random.rand(output_size, input_size)-0.5)*2;
        #偏执项
        self.B = np.zeros((output_size, 1));
        #输出向量
        self.output = np.zeros((output_size, 1));
        return;

    def forward(self, input_array):
        '''
        向前运算
        '''
        self.input = input_array;
        self.output = self.activator.forward(np.dot(self.W, self.input)+self.B);
        return;

    def backward(self, detal_array):
        '''
        向后运算
        '''
        self.detal = self.activator.backward(self.input)*np.dot(self.W.T, detal_array);
        self.W_grad = np.dot(detal_array, self.input.T);
        self.B_grad = detal_array;
        return;

    def update(self, learning_rate):
        '''
        更新权重
        '''
        self.W +=learning_rate*self.W_grad;
        self.B +=learning_rate*self.B_grad;
        return;

#激活函数类
class SigmoidActivator(object):
    def forward(self, x):
        return 1/(1+np.exp(-x));

    def backward(self, x):
        return x*(1-x);

#BP神经网络类
class BPNetWork(object):
    def __init__(self, layers):
        self.layers = [];
        for i in range(len(layers)-1):
            self.layers.append(BPLayer(layers[i], layers[i+1], SigmoidActivator()));

    def predict(self, sample):
        '''
        预测实现
        '''
        output = sample;
        for layer in self.layers:
            layer.forward(output);
            output = layer.output;
        return output;

    def train(self, labels, data_set, rate, epoch):
        '''
        训练网络
        '''
        for i in range(epoch):
            for d in range(len(data_set)):
                #按照矩阵乘的结构具状数据 W [300行*784列]  input[1行*784列]
                self.train_one_sample(np.array([labels[d]]).T, np.array([data_set[d]]).T, rate);
##              self.train_one_sample(labels[d], data_set[d], rate);
        return;

    def train_one_sample(self, label, date, rate):
        self.predict(date);
        self.calc_gradient(label);
        self.update_w(rate);
        return;

    def calc_gradient(self, label):
        detal = self.layers[-1].activator.backward(self.layers[-1].output)*(label - self.layers[-1].output);
        for layer in self.layers[::-1]:
            layer.backward(detal);
            detal = layer.detal;
        return;

    def update_w(self, rate):
        for layer in self.layers:
            layer.update(rate);
        return;

def get_result(vec):
    max_value_index = 0;
    max_value = 0;
    for i in range(len(vec)):
        if vec[i] > max_value:
            max_value = vec[i];
            max_value_index = i;

    return max_value_index;

def evaluate(network, test_data_set, test_labels):
    error = 0;
    total = len(test_data_set);
    for index in range(total):
        label = get_result(test_labels[index]);
        predict = get_result(network.predict(np.array([test_data_set[index]]).T));
        if label != predict:
            error += 1;
    return float(error)/float(total);

def  train_and_evaluate():
    last_error_ratio = 1.0;
    epoch = 0;
    x_train,y_train = gim.get_training_data_set();
    x_test,y_test = gim.get_test_data_set();
    
    layers=[784,300,10];
    bpNet = BPNetWork(layers);

    while True:
        epoch += 1;

        bpNet.train(y_train, x_train, 0.3, 1);
        print 'epoch %d finished' % (epoch);

        if epoch % 3 == 0:
            error_ratio = evaluate(bpNet, x_test, y_test);
            print 'after epoch %d , error ratio is %f' % (epoch, error_ratio);

            if error_ratio > last_error_ratio:
                break;
            else:
                last_error_ratio = error_ratio;
    


if __name__ == '__main__':
    train_and_evaluate();