手写数字识别mnist，手动完成，不利用框架

àì夳堔傛蜴生んèń 2022-04-24 15:58 155阅读 0赞

## 个人学习，仅供参考 ##

### 获取训练样本 ###

# -*- coding: utf-8 -*-
    import os
    import math
    import gzip
    import pickle
    import requests
    import numpy as np
    
    url_base = 'http://yann.lecun.com/exdb/mnist_learn/'
    key_file = {
        'train_img': 'train-images-idx3-ubyte.gz',
        'train_label': 'train-labels-idx1-ubyte.gz',
        'test_img': 't10k-images-idx3-ubyte.gz',
        'test_label': 't10k-labels-idx1-ubyte.gz',
    }
    
    dataset_dir = os.path.dirname(os.path.abspath(__file__))
    save_file = dataset_dir + '/mnist.pkl'
    
    train_num = 60000
    test_num = 10000
    img_size = 784
    
    
    def down_data():
        for value in key_file.values():
            file_name = dataset_dir + '/' + value
            if os.path.exists(file_name):
                continue
            context = requests.get(url_base + '/' + value)
            with open(file_name, 'wb') as f:
                f.write(context.content)
    
    
    def load_data(file_name):
        file_path = dataset_dir + '/' + file_name
        print('transform ' + file_name + 'to NumPy Array ')
    
        with gzip.open(file_path, 'rb') as f:
            # 将缓存中的数据解释为numpy数组，offset设置偏移量
            data = np.frombuffer(f.read(), np.uint8, offset=8)
            print("Well Done")
    
        return data
    
    
    def reload_data_pkl():
        data = {
            'train_img': load_data(key_file['train_img']),
            'train_label': load_data(key_file['train_label']),
            'test_img': load_data(key_file['test_img']),
            'test_label': load_data(key_file['test_label'])
        }
    
        with open('mnist.pkl', 'wb') as f:
            # 参数-1，选择最新版本的协议，对大型数据进行了存储优化，可以使用4代替
            pickle.dump(data, f, -1)
    
    
    def convert_to_one_hot(Y, C):
        """
    
        :param Y: 索引，Y的大小为6000*1，标记
        :param C: 对角矩阵的大小
        :return:Y：10*6000
        """
        # eye返回对角矩阵
        # 将标记矩阵Y返回
        Y = np.eye(C)[Y.reshape(-1)].T
        return Y
    
    
    def random_mini_batches(x, y, mini_batch_size=64, seed=0):
        np.random.seed(seed)
        m = x.shape[1]
        mini_batches = []
        permutation = list(np.random.permutation(m))
        shuffled_x = x[:, permutation]
        shuffled_y = y[:, permutation]
    
        num_complete_mini_batches = math.floor(m / mini_batch_size)
        for k in range(num_complete_mini_batches):
            mini_bath_x = shuffled_x[:, k * mini_batch_size:(k + 1) * mini_batch_size]
            mini_bath_y = shuffled_y[:, k * mini_batch_size:(k + 1) * mini_batch_size]
            mini_bath = (mini_bath_x, mini_bath_y)
    
            mini_batches.append(mini_bath)
        if m % mini_batch_size != 0:
            mini_batch_x = shuffled_x[:, num_complete_mini_batches * mini_batch_size: m]
            mini_batch_y = shuffled_y[:, num_complete_mini_batches * mini_batch_size: m]
            mini_batch = (mini_batch_x, mini_batch_y)
            mini_batches.append(mini_batch)
    
        return mini_batches
    
    
    def init_data():
        if not os.path.exists(save_file):
            reload_data_pkl()
        with open(save_file, 'rb') as f:
            dataset = pickle.load(f)
    
        # 归一化训练集
        train_x = dataset['train_img'].astype(np.float32).T / 255
        test_x = dataset['test_img'].astype(np.float32).T / 255
    
        train_y = convert_to_one_hot(dataset['train_label'], 10)
        test_y = convert_to_one_hot(dataset['test_label'], 10)
    
        return train_x, train_y, test_x, test_y

## 训练样本 ##

# -*- coding: utf-8 -*-
    
    import os
    import sys
    import math
    import numpy as np
    import matplotlib.pyplot as plt
    
    sys.path.append(os.pardir)
    from prepare_data import random_mini_batches, init_data
    
    
    # x: 784*6000, 784*1000
    # y: 10*6000, 10*1000
    # W1: 25*784
    # b1: 25*1
    # W2: 12*25
    # b2: 12*1
    # W3: 10*12
    # b3: 10*1
    
    
    def softmax(z):
        # 实现softmax分类器输出转化
        z_exp = np.exp(z)
        z_sum_column = z_exp.sum(axis=0).T
        return z_exp / z_sum_column
    
    
    def relu(x):
        return np.maximum(x, 0)
    
    
    def init_parameters(layers_dim):
        """
        784, 25, 12, 10
        :return:
        """
        m = len(layers_dim)
        parameters = {}
        for i in range(m - 1):
            parameters['W' + str(i + 1)] = np.random.randn(layers_dim[i + 1], layers_dim[i]) * np.sqrt(
                2 / layers_dim[i])
            parameters['b' + str(i + 1)] = np.zeros((layers_dim[i + 1], 1))
    
        return parameters
    
    
    def init_adam(parameters):
        """
        优化方法adam算法
        :param parameters:
        :return:
        """
        length = len(parameters) // 2
        v = {}
        s = {}
        for i in range(length):
            v["dW" + str(i + 1)] = np.zeros(parameters['W' + str(i + 1)].shape)
            v["db" + str(i + 1)] = np.zeros(parameters['b' + str(i + 1)].shape)
            s["dW" + str(i + 1)] = np.zeros(parameters['b' + str(i + 1)].shape)
            s["db" + str(i + 1)] = np.zeros(parameters['b' + str(i + 1)].shape)
    
        return v, s
    
    
    def update_parameters_with_adam(parameters, grads, v, s, t, learning_rate=0.01, beta1=0.9, beta2=0.999, epsilon=1e-8):
        """
        更新参数，adam算法
        :param parameters:
        :param grads:
        :param v: adam算法初始值
        :param s: adam算法初始值
        :param t: 随着迭代次数的增加，主键减小adam的作用
        :param learning_rate: 学习率
        :param beta1: momentum算法参数，一般取默认值
        :param beta2: RMSprop算法参数，一般去默认值
        :param epsilon:
        :return:
        """
        length = len(parameters) // 2
    
        v_corrected = {}
        s_corrected = {}
        for i in range(length):
            v["dW" + str(i + 1)] = beta1 * v["dW" + str(i + 1)] + (1 - beta1) * grads['dW' + str(i + 1)]
            v["db" + str(i + 1)] = beta1 * v["db" + str(i + 1)] + (1 - beta1) * grads['db' + str(i + 1)]
    
            v_corrected["dW" + str(i + 1)] = v["dW" + str(i + 1)] / (1 - math.pow(beta1, t))
            v_corrected["db" + str(i + 1)] = v["db" + str(i + 1)] / (1 - math.pow(beta1, t))
    
            s["dW" + str(i + 1)] = beta2 * s["dW" + str(i + 1)] + (1 - beta2) * (grads['dW' + str(i + 1)] ** 2)
            s["db" + str(i + 1)] = beta2 * s["db" + str(i + 1)] + (1 - beta2) * (grads['db' + str(i + 1)] ** 2)
    
            s_corrected["dW" + str(i + 1)] = s["dW" + str(i + 1)] / (1 - math.pow(beta2, t))
            s_corrected["db" + str(i + 1)] = s["db" + str(i + 1)] / (1 - math.pow(beta2, t))
    
            parameters["W" + str(i + 1)] = parameters["W" + str(i + 1)] - learning_rate * v_corrected[
                "dW" + str(i + 1)] / (np.sqrt(s_corrected["dW" + str(i + 1)]) + epsilon)
            parameters["b" + str(i + 1)] = parameters["b" + str(i + 1)] - learning_rate * v_corrected[
                "db" + str(i + 1)] / (np.sqrt(s_corrected["db" + str(i + 1)]) + epsilon)
    
        return parameters, v, s
    
    
    def compute_cost(a3, a):
        """
        cost
        :param a3: 预测值
        :param a: 实际值
        :return:
        """
        m = a3.shape[1]
        a3 = np.multiply(a3, a).sum(axis=0)
        loss = -np.log(a3)
        cost = loss.sum() / m
    
        return cost
    
    
    def forward_propagation(x, mini_batch_size, parameters):
        """
        前向
        :param x:
        :param parameters:
        :return:
        """
        W1 = parameters['W1']
        b1 = parameters['b1']
        W2 = parameters['W2']
        b2 = parameters['b2']
        W3 = parameters['W3']
        b3 = parameters['b3']
    
        Z1 = np.dot(W1, x) + b1
        A1 = relu(Z1)
        Z2 = np.dot(W2, A1) + b2
        A2 = relu(Z2)
        Z3 = np.dot(W3, A2) + b3
        A3 = softmax(Z3)
    
        return A3, (Z1, A1, Z2, A2, Z3, A3)
    
    
    def backward_propagation(x, y, cache, parameters):
        """
        反向传播
        :param x:
        :param y:
        :param cache:
        :param parameters:
        :return:
        """
        W2 = parameters['W2']
        W3 = parameters['W3']
    
        (Z1, A1, Z2, A2, Z3, A3) = cache
    
        dz3 = A3 - y
        dW3 = np.dot(dz3, A2.T)
        db3 = np.sum(dz3, axis=1, keepdims=True)
    
        da2 = np.dot(W3.T, dz3)
        dz2 = np.multiply(da2, np.int64(A2 > 0))
        dW2 = np.dot(dz2, A1.T)
        db2 = np.sum(dz2, axis=1, keepdims=True)
    
        da1 = np.dot(W2.T, dz2)
        dz1 = np.multiply(da1, np.int64(A1 > 0))
        dW1 = np.dot(dz1, x.T)
        db1 = np.sum(dz1, axis=1, keepdims=True)
    
        gradients = {"dz3": dz3, "dW3": dW3, "db3": db3,
                     "da2": da2, "dz2": dz2, "dW2": dW2, "db2": db2,
                     "da1": da1, "dz1": dz1, "dW1": dW1, "db1": db1}
    
        return gradients
    
    
    def model(X, Y, layers_dims, learning_rate=0.0007, mini_batch_size=64, beta1=0.9, beta2=0.999,
              epsilon=1e-8, num_epochs=10000, print_cost=True):
        """
        一般过程
        """
        parameters = init_parameters(layers_dims)
        v, s = init_adam(parameters)
        t = 0
        seed = 0
        costs = []
        for i in range(num_epochs):
            seed += 1
            mini_batches = random_mini_batches(X, Y, mini_batch_size, seed)
            for mini_bath in mini_batches:
                t += 1
                mini_batch_x, mini_batch_y = mini_bath
                A3, cache = forward_propagation(mini_batch_x, parameters)
                cost = compute_cost(A3, mini_batch_y)
                gradients = backward_propagation(mini_batch_x, mini_batch_y, cache, parameters)
                parameters, v, s = update_parameters_with_adam(parameters, gradients, v, s, t, learning_rate, beta1, beta2,
                                                               epsilon)
            if print_cost and i % 1000 == 0:
                print("Cost after epoch %i: %f" % (i, cost))
            if print_cost and i % 100 == 0:
                costs.append(cost)
    
        plt.plot(costs)
        plt.xlabel('epochs (per 100)')
        plt.ylabel('cost')
        plt.title("Learning rate = " + str(learning_rate))
    
        return parameters
    
    
    def predict(x, y, parameters):
        """
        预测结果
        :param x:
        :param y_one: 一维标记，便于计算
        :param parameters:
        :return:
        """
        m = x.shape[1]
        n = x.shape[0]
        a3, _ = forward_propagation(x, parameters)
        predict_result = np.where(a3 > 0.5, 1, 0)
    
        n = np.mean((y != predict_result)) * m * n / 2
        accuracy = n / m
    
        return accuracy
    
    
    if __name__ == '__main__':
        train_x, train_y, test_x, test_y= init_data()
        layers = [train_x.shape[0], 25, 12, train_y.shape[0]]
        parameters_ = model(train_x, train_y, layers_dims=layers)
        accuracy_ = predict(test_x, test_y, parameters_)