手寫數字識別簡化版——0和1的二分類問題

一、數據集解析

1. 數據集格式介紹

該數據集可以在Yann LeCun的官網上查看。官網鏈接：手寫數字識別數據集。他這個數據集保存形式比較特殊，四個文件（訓練集、測試集的圖片和標簽）都是以IDX文件格式保存的。IDX文件格式是各種數值類型的向量和多維矩陣的簡單格式。

以官網的train-images.idx3-ubyte為例來說明IDX格式。

• offset是用16進制數表示的，代表偏移量，也就是在該文件中的存放地址。文件按字節存儲，比如第一行起始地址為0000H，存儲類型為32bit的整數，占4個字節，所以第二行的起始地址就是0004H。
• 圖片的存儲是在description中，該文件前面四行分別是magic number、number of images、number of rows、number of columns，后三個的value分別是60000、28、28，告訴了我們訓練集中圖片的數量與尺寸信息。
• 后面的每一行的pixel表示一張圖片中一個像素點的大小，因為像素的范圍是0-255的（0是白色、255是黑色），也就是\(2^8\)，占8bit，1個字節，也就是占一行，offset＋0001H。於是我們也可以推算出：
  • 0016H~0799H空間內存放了第一張圖片的信息，因為一張圖片是28*28，占784行；
• train-images.idx3-ubyte整個文件大小為16+784*60000=47040016B≈45937.5KB

2. 數據集處理

為了直觀看到圖片，我們需要將pixel數據可視化。而且我對IDX文件不是很了解，更傾向於對圖片數據進行處理，於是在網上找了下面的程序，用以轉化。

2.1 訓練集處理

將四個數據集文件放在同一個目錄下，運行下面的代碼，可以生成mnist_train的文件夾，里面有0-9個子文件，每個子文件都有對應的圖片。

理解好上面介紹的文件格式和unpack_from緩存流的用法，就理解了data_file_size要改成什么值以及為什么要改變值。

import numpy as np
import struct
 
from PIL import Image
import os
 
data_file = 'train-images.idx3-ubyte'
# It's 47040016B, but we should set to 47040000B
data_file_size = 47040016
data_file_size = str(data_file_size - 16) + 'B'
 
data_buf = open(data_file, 'rb').read()
 
magic, numImages, numRows, numColumns = struct.unpack_from(
    '>IIII', data_buf, 0)
datas = struct.unpack_from(
    '>' + data_file_size, data_buf, struct.calcsize('>IIII'))
datas = np.array(datas).astype(np.uint8).reshape(
    numImages, 1, numRows, numColumns)
 
label_file = 'train-labels.idx1-ubyte'
 
# It's 60008B, but we should set to 60000B
label_file_size = 60008
label_file_size = str(label_file_size - 8) + 'B'
 
label_buf = open(label_file, 'rb').read()
 
magic, numLabels = struct.unpack_from('>II', label_buf, 0)
labels = struct.unpack_from(
    '>' + label_file_size, label_buf, struct.calcsize('>II'))
labels = np.array(labels).astype(np.int64)
 
datas_root = 'mnist_train'
if not os.path.exists(datas_root):
    os.mkdir(datas_root)
 
for i in range(10):
    file_name = datas_root + os.sep + str(i)
    if not os.path.exists(file_name):
        os.mkdir(file_name)
 
for ii in range(numLabels):
    img = Image.fromarray(datas[ii, 0, 0:28, 0:28])
    label = labels[ii]
    file_name = datas_root + os.sep + str(label) + os.sep + \
        'mnist_train_' + str(ii) + '.png'
    img.save(file_name)

2.2 測試集處理

與訓練集的代碼差不多，改了改文件名字和大小而已。

import numpy as np
import struct
 
from PIL import Image
import os
 
data_file = 't10k-images.idx3-ubyte'
# It's 7840016B, but we should set to 7840000B
data_file_size = 7840016
data_file_size = str(data_file_size - 16) + 'B'
 
data_buf = open(data_file, 'rb').read()
 
magic, numImages, numRows, numColumns = struct.unpack_from(
    '>IIII', data_buf, 0)
datas = struct.unpack_from(
    '>' + data_file_size, data_buf, struct.calcsize('>IIII'))
datas = np.array(datas).astype(np.uint8).reshape(
    numImages, 1, numRows, numColumns)
 
label_file = 't10k-labels.idx1-ubyte'
 
# It's 10008B, but we should set to 10000B
label_file_size = 10008
label_file_size = str(label_file_size - 8) + 'B'
 
label_buf = open(label_file, 'rb').read()
 
magic, numLabels = struct.unpack_from('>II', label_buf, 0)
labels = struct.unpack_from(
    '>' + label_file_size, label_buf, struct.calcsize('>II'))
labels = np.array(labels).astype(np.int64)
 
datas_root = 'mnist_test'
if not os.path.exists(datas_root):
    os.mkdir(datas_root)
 
for i in range(10):
    file_name = datas_root + os.sep + str(i)
    if not os.path.exists(file_name):
        os.mkdir(file_name)
 
for ii in range(numLabels):
    img = Image.fromarray(datas[ii, 0, 0:28, 0:28])
    label = labels[ii]
    file_name = datas_root + os.sep + str(label) + os.sep + \
        'mnist_test_' + str(ii) + '.png'
    img.save(file_name)

二、接口load設計

對於我這個階段，模型什么的往往都只是套用即可，而data_loader是實現過程中的難點，也是最需要編程的地方，小白應該訓練的基本功。

• 函數data_load就是將圖片和對應的標簽以矩陣的形式存放，再分別加入列表中。如果是訓練就要將列表隨機shuffle，如果是測試我覺得就不用了。
• yield的生成器這一塊需要好好理解，菜鳥教程上面寫的很詳細了。釋放出去時應該是要將列表轉換成array，釋放的列表的大小基本都是Batchsize，除了最后一批次
• 因為本案例只是做了0和1的二分類，所以就沒有將標簽為2-9的文件存進去，因此直接列了0和1的讀取，沒有把它寫出一個函數，代碼看上去稍微冗雜了點。

def data_load(mode='train'):
    data_list = []
    # 分別讀取
    if mode == 'train':
        dir = 'mnist_train\\0'
        for filename in os.listdir(dir):
            img_path = dir + '\\' + filename 
            img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
            img = np.reshape(img, [1, IMG_ROWS, IMG_COLS]).astype('float32')
            label = np.array([0]).astype('int64')
            data_list.append((img, label))

        dir = 'mnist_train\\1'
        for filename in os.listdir(dir):
            img_path = dir + '\\' + filename 
            img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
            img = np.reshape(img, [1, IMG_ROWS, IMG_COLS]).astype('float32')
            label = np.array([1]).astype('int64')
            data_list.append((img, label))
    
        # 打亂
        random.shuffle(data_list)
    
    if mode == 'eval':
        dir = 'mnist_test\\0'
        for filename in os.listdir(dir):
            img_path = dir + '\\' + filename 
            img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
            img = np.reshape(img, [1, IMG_ROWS, IMG_COLS]).astype('float32')
            label = np.array([0]).astype('int64')
            data_list.append((img, label))

        dir = 'mnist_test\\1'
        for filename in os.listdir(dir):
            img_path = dir + '\\' + filename 
            img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)
            img = np.reshape(img, [1, IMG_ROWS, IMG_COLS]).astype('float32')
            label = np.array([1]).astype('int64')
            data_list.append((img, label))

    imgs_list = []
    labels_list = []
    for data in data_list:
        imgs_list.append(data[0])
        labels_list.append(data[1])
        if len(imgs_list) == BATCHSIZE:
            yield np.array(imgs_list), np.array(labels_list)
            imgs_list = []
            labels_list = []
            
    if len(imgs_list) > 0:
        yield np.array(imgs_list), np.array(labels_list)
    
    return data_load

可以先來測試一下：

# 聲明數據讀取函數，從訓練集中讀取數據
train_loader = data_load
# 以迭代的形式讀取數據
for batch_id, data in enumerate(train_loader()):
    image_data, label_data = data
    if batch_id == 0:
        # 打印數據shape和類型
        print("打印第一個batch數據的維度，以及數據的類型:")
        print("圖像維度: {}, 標簽維度: {}, 圖像數據類型: {}, 標簽數據類型: {}".format(image_data.shape, label_data.shape, type(image_data), type(label_data)))
        
        break
# 打印第一個batch數據的維度，以及數據的類型:
# 圖像維度: (100, 1, 28, 28), 標簽維度: (100, 1), 圖像數據類型: <class 'numpy.ndarray'>, 標簽數據類型: <class 'numpy.ndarray'>

三、模型選用

這一塊直接用經典的卷積神經網絡就行了，選用paddle2.2作為深度學習的框架。值得注意的是，模型選用與data_load是密切聯系的，因為希望用卷積神經網絡，所以采用NCWH的格式， 即上面的(100, 1, 28, 28)。如果不采用卷積神經網絡，而用BP網絡，格式就應該是(100, 1, 784)。注意這里的N就是Batchsize，我自己開始設定的是100。
至於為什么前向計算的地方又加了一個if，這個是用在后面測試的時候；當然也可以選擇去掉這塊，放在測試那塊。
至於網絡中的參數，直接代卷積公式、池化公式就好了。
下面是一張圖片的尺寸變化：

• conv1后的尺寸：(28-5+2*2)/1+1=28
• pool1以后的尺寸：28/2=14
• conv2后的尺寸(14-5+2*2)/1+1=14
• pool2以后的尺寸：14/2=7
• reshape后，全鏈接前的維度 7*7*20=980

class MNIST(paddle.nn.Layer):
    def __init__(self):
        super(MNIST, self).__init__()
         
        self.conv1 = nn.Conv2D(in_channels=1, out_channels=20, kernel_size=5, stride=1, padding=2)
        self.max_pool1 = nn.MaxPool2D(kernel_size=2, stride=2)
        self.conv2 = nn.Conv2D(in_channels=20, out_channels=20, kernel_size=5, stride=1, padding=2)
        self.max_pool2 = nn.MaxPool2D(kernel_size=2, stride=2)
        self.fc = nn.Linear(in_features=980, out_features=10)

    def forward(self, inputs, label=None):
        x = self.conv1(inputs)
        x = F.relu(x)
        x = self.max_pool1(x)
        x = self.conv2(x)
        x = F.relu(x)
        x = self.max_pool2(x)
        x = paddle.reshape(x, [x.shape[0], -1])
        x = self.fc(x)
        if label is not None:
            acc = paddle.metric.accuracy(input=x, label=label)
            return x, acc
        else:
            return x

四、開始訓練

訓練的部分就不是很復雜了，我直接搬的網上的代碼，最多改改設定的事情。

def train(model):
    model.train()
    train_loader = data_load

    opt = paddle.optimizer.SGD(learning_rate=0.001, parameters=model.parameters())
    
    EPOCH_NUM = 10
    for epoch_id in range(EPOCH_NUM):
        for batch_id, data in enumerate(train_loader()):
            images, labels = data
            images = paddle.to_tensor(images)
            labels = paddle.to_tensor(labels)
            predicts = model(images)

            loss = F.cross_entropy(predicts, labels)
            avg_loss = paddle.mean(loss)
            
            if batch_id % 100 == 0:
                print("epoch: {}, batch: {}, loss is: {}".format(epoch_id, batch_id, avg_loss.numpy()))
            
            avg_loss.backward()
            opt.step()
            opt.clear_grad()
   
    paddle.save(model.state_dict(), 'mnist2.pdparams')
    
model = MNIST()
#啟動訓練過程
train(model)

五、測試評價

def evaluation(model):
    print('start evaluation .......')
    # 定義預測過程
    params_file_path = 'mnist2.pdparams'
    # 加載模型參數
    param_dict = paddle.load(params_file_path)
    model.load_dict(param_dict)

    model.eval()
    eval_loader = data_load

    acc_set = []
    avg_loss_set = []
    for batch_id, data in enumerate(eval_loader('eval')):
        images, labels = data
        images = paddle.to_tensor(images)
        labels = paddle.to_tensor(labels)
        predicts, acc = model(images, labels)
        loss = F.cross_entropy(input=predicts, label=labels)
        avg_loss = paddle.mean(loss)
        acc_set.append(float(acc.numpy()))
        avg_loss_set.append(float(avg_loss.numpy()))
    
    #計算多個batch的平均損失和准確率
    acc_val_mean = np.array(acc_set).mean()
    avg_loss_val_mean = np.array(avg_loss_set).mean()

    print('loss={}, acc={}'.format(avg_loss_val_mean, acc_val_mean))

model = MNIST()
evaluation(model)
# start evaluation .......
# loss=0.05132369852472607, acc=0.9950000047683716

想要可視化，可以改進四、五兩節的函數，return返回值，繪圖。