如何自己訓練一個熱狗識别模型 | 《阿裡雲機器學習PAI-DSW入門指南》

點選即可參與機器學習PAI-DSW動手實驗室 點選可下載下傳完整電子書《阿裡雲機器學習PAI-DSW入門指南》

美劇《矽谷》大家想必都沒怎麼看過，大家可能都不知道人工智能識别熱狗曾是矽谷最賺錢的技術之一。HBO 曾釋出了官方的 Not Hotdog 應用，支援 iOS 和 Android 平台，據說是用 TensorFlow、Keras 和 React Native 打造的，但是源碼沒有公開。

我們今天要做都就是這部美劇裡面第四季裡面讓楊建成為百萬富翁的模型：熱狗識别模型。這一次，就讓阿裡雲的資料科學老司機帶你一起，利用機器學習pai平台訓練自己的熱狗識别模型，打破技術封鎖。讓你出任CEO,迎娶白富美/高富帥,走上人生巅峰。

工欲善其事，必先利其器。沒有好的工具就想要訓練出好的模型簡直是天方夜譚。

大家進入DSW環境之後，就可以上傳訓練代碼以及資料了。

我們先從

這裡

下載下傳我為各位精心準備好的代碼和訓練資料集壓縮包。下載下傳到本地之後，點選上傳這個按鈕 就可以把你的檔案上傳上來了。

上傳成功後，我們打開Terminal 進入這個路徑，然後輸入

$ unzip ./not_hotdog.zip # 解壓整個檔案
$ cd not_hotdog.zip
$ unzip seefood.zip # 解壓訓練資料集           

然後就會看到我們的檔案夾已經乖乖躺在我們的左側的資料總管裡邊兒了。

接下來就是我們的硬核部分，我們直接把代碼放上來。我們直接運作就可以拉。

#!/usr/bin/env python
# coding: utf-8

# # Import dependencies 導入依賴

# In[1]:


import numpy as np
import pandas as pd
import os

import tensorflow as tf
rand_state = 42 # 順便定義一個随機種子 
tf.set_random_seed(rand_state)
np.random.seed(rand_state)

from skimage import exposure
import cv2
import glob
import time
import matplotlib.pyplot as plt
from keras.utils.vis_utils import plot_model


# # 圖像預處理的函數們

# In[2]:


def rotateImage(img, angle):
    '''
    img:三通道的圖檔
    angle:随機角度
    
    本功能是樣本增強功能，對圖檔樣本進行随機的旋轉縮放
    
    return:傳回一個變換後的圖檔
    
    '''
    
    (rows, cols, ch) = img.shape   # 得到源圖檔尺寸
    
    #第一個參數旋轉中心，第二個參數旋轉角度，第三個參數：縮放比例
    M = cv2.getRotationMatrix2D((cols/2,rows/2), angle, 1)
    
    return cv2.warpAffine(img, M, (cols,rows))  # 圖像進行上面操作後生成的圖像
    
    
def loadBlurImg(path, imgSize):
    '''
    path：圖檔路徑，字元串
    imgsize:圖檔的尺寸,二進制組，元素都是int
    '''
    img = cv2.imread(path)  # 讀取圖檔資料
    angle = np.random.randint(0, 360)  # 生成0，360之間生成随機數，離散均勻随機，整形
    img = rotateImage(img, angle)   # 圖檔随機旋轉，縮放
    img = cv2.blur(img,(5,5))       # 每5*5的尺寸進行均值模糊
    img = cv2.resize(img, imgSize)  # 圖檔按照尺寸縮放   
    return img


def loadImgClass(classPath, classLable, classSize, imgSize):
    '''
    classPath:傳入圖檔的路徑，list集合
    classLable:圖檔的類别，數值int
    classSize:樣本數量
    imgsize:圖檔的尺寸,二進制組，元素都是int
    
    return:傳回classSize個樣本及标簽
    
    本函數從樣本位址中生成classSize個資料，樣本是經過旋轉，縮放等變換的，圖檔規格是imgsize
    
    '''
    x = []
    y = []
    
    for path in classPath:
        img = loadBlurImg(path, imgSize)   # 加載位址中的圖檔并進行樣本增強，生成imgsize大的圖檔    
        x.append(img)
        y.append(classLable)
        
    while len(x) < classSize:
        randIdx = np.random.randint(0, len(classPath))
        img = loadBlurImg(classPath[randIdx], imgSize)
        x.append(img)
        y.append(classLable)
        
    return x, y

def loadData(img_size, classSize, hotdogs, notHotdogs):    
    '''
    img_size:要傳回圖檔的大小,int
    classSize:正例，負例樣本數量,int
    hotsdogs,notHotdogs:正例，負例樣本位址，都是個list
    
    return；傳回訓練樣本及對應的标簽
    
    本函數讀取資料并傳回樣本及标簽
    '''
    
    imgSize = (img_size, img_size)     # 要輸入圖檔的尺寸
    xHotdog, yHotdog = loadImgClass(hotdogs, 0, classSize, imgSize)   # 生成正樣本，classSize個
    xNotHotdog, yNotHotdog = loadImgClass(notHotdogs, 1, classSize, imgSize)  # 生成負樣本，classSize個
    print("There are", len(xHotdog), "hotdog images")
    print("There are", len(xNotHotdog), "not hotdog images")
    
    X = np.array(xHotdog + xNotHotdog)      
    y = np.array(yHotdog + yNotHotdog)
    
    return X, y

def toGray(images):
    
    '''
    樣本灰階轉換，生成後的圖檔是一個通道的
    '''
    # rgb2gray converts RGB values to grayscale values by forming a weighted sum of the R, G, and B components:
    # 0.2989 * R + 0.5870 * G + 0.1140 * B 
    # source: https://www.mathworks.com/help/matlab/ref/rgb2gray.html
    
    images = 0.2989*images[:,:,:,0] + 0.5870*images[:,:,:,1] + 0.1140*images[:,:,:,2]
    return images

def normalizeImages(images):
    '''
    images:1個通道的圖像
    return：圖像像素經過比例縮放，直方圖均衡後的圖像
    '''
    # use Histogram equalization to get a better range
    # source http://scikit-image.org/docs/dev/api/skimage.exposure.html#skimage.exposure.equalize_hist
    images = (images / 255.).astype(np.float32)  # rgb像素是0-255之間，縮放至0-1的範圍
    
    for i in range(images.shape[0]):
        images[i] = exposure.equalize_hist(images[i])   # 直方圖均衡之後的圖像數組
    
    images = images.reshape(images.shape + (1,))   #  二維擴成三維
    return images

def preprocessData(images):
    '''
    images:三通道的image
    return:傳回一通道，且數值經過比例縮放的圖檔（除以255，使之數值範圍集中在0-1之間）
    '''
    grayImages = toGray(images)
    return normalizeImages(grayImages)


# # 我們需要對圖像做一些騷操作 畢竟500張圖檔還是太少了

# In[3]:


from keras.utils.np_utils import to_categorical
from sklearn.model_selection import train_test_split

size = 32
classSize = 20000


# In[7]:


# 導入資料
hotdogs = glob.glob('./train/hot_dog/**/*.jpg', recursive=True)
notHotdogs = glob.glob('./train/not_hot_dog/**/*.jpg', recursive=True)


# In[12]:


dd = (20000,20000)
print(dd)


# In[14]:


# 騷操作一波 
scaled_X, y = loadData(size, classSize, hotdogs, notHotdogs)
scaled_X = preprocessData(scaled_X)


# In[15]:


y = to_categorical(y)    # 目标變量獨熱


n_classes=2
print("y shape", y.shape)
X_train, X_test, y_train, y_test = train_test_split(
    scaled_X, 
    y, 
    test_size=0.2, 
    random_state=rand_state
)    # 資料按照訓練集0.8的比例分割

print("train shape X", X_train.shape)
print("train shape y", y_train.shape)
print("Test shape X:", X_test.shape)
print("Test shape y: ", y_test.shape)

inputShape = (size, size, 1)


# In[8]:


def plot_history(history):
    loss_list = [s for s in history.history.keys() if 'loss' in s and 'val' not in s]
    val_loss_list = [s for s in history.history.keys() if 'loss' in s and 'val' in s]
    acc_list = [s for s in history.history.keys() if 'acc' in s and 'val' not in s]
    val_acc_list = [s for s in history.history.keys() if 'acc' in s and 'val' in s]
    
    if len(loss_list) == 0:
        print('Loss is missing in history')
        return 
    
    ## As loss always exists
    epochs = range(1,len(history.history[loss_list[0]]) + 1)
    
    ## Loss
    plt.figure(1)
    for l in loss_list:
        plt.plot(epochs, history.history[l], 'b', label='Training loss (' + str(str(format(history.history[l][-1],'.5f'))+')'))
    for l in val_loss_list:
        plt.plot(epochs, history.history[l], 'g', label='Validation loss (' + str(str(format(history.history[l][-1],'.5f'))+')'))
    
    plt.title('Loss')
    plt.xlabel('Epochs')
    plt.ylabel('Loss')
    plt.legend()
    
    ## Accuracy
    plt.figure(2)
    for l in acc_list:
        plt.plot(epochs, history.history[l], 'b', label='Training accuracy (' + str(format(history.history[l][-1],'.5f'))+')')
    for l in val_acc_list:    
        plt.plot(epochs, history.history[l], 'g', label='Validation accuracy (' + str(format(history.history[l][-1],'.5f'))+')')

    plt.title('Accuracy')
    plt.xlabel('Epochs')
    plt.ylabel('Accuracy')
    plt.legend()
    plt.show()


# # 重點來了：構模組化型就是這兒了

# In[9]:


import keras
from keras.models import Sequential
from keras.callbacks import EarlyStopping, ModelCheckpoint
from keras.layers import Conv2D, MaxPooling2D, Dense, Dropout, Flatten
from keras.layers.normalization import BatchNormalization


model = Sequential()
model.add(Conv2D(32, kernel_size=(3, 3),
                 activation='relu',
                 kernel_initializer='he_normal',
                 input_shape=inputShape))   # 卷積
model.add(MaxPooling2D((2, 2)))             # 池化
model.add(Dropout(0.25))                    # 随機失活
model.add(Conv2D(64, (3, 3), activation='relu'))
model.add(MaxPooling2D(pool_size=(2, 2)))
model.add(Dropout(0.25))
model.add(Conv2D(128, (3, 3), activation='relu'))
model.add(Dropout(0.4))
model.add(Flatten())                       # 展成一維
model.add(Dense(128, activation='relu'))   # 全連接配接
model.add(Dropout(0.3))
model.add(Dense(2, activation='softmax'))

model.compile(loss=keras.losses.binary_crossentropy,
              optimizer=keras.optimizers.Adam(lr=1e-4),
              metrics=['accuracy'])

start = time.time()

model.summary()
# Set callback functions to early stop training and save the best model so far
callbacks = [
    EarlyStopping(
        monitor='val_loss', 
        patience=3
    ),
    ModelCheckpoint(
        filepath='model.h5', 
        monitor='val_acc', 
        save_best_only=True
    )
]

history = model.fit(
    X_train, 
    y_train,
    batch_size=32,
    epochs=100, 
    callbacks=callbacks,
    verbose=0,
    validation_data=(X_test, y_test)
)

end = time.time()
print('Execution time: ', end-start)

plot_history(history)
            

訓練完成之後，我們可以簡單的測試一下我們模型的準确率。下面這段代碼就可以幫我們做到這一點。

hotdogs = glob.glob('./test/hot_dog/**/*.jpg', recursive=True) 
notHotdogs = glob.glob('./test/not_hot_dog/**/*.jpg', recursive=True)

scaled_X_test, y_test = loadData(size, 250, hotdogs, notHotdogs)
scaled_X_test = preprocessData(scaled_X_test)

#get the predictions for the test data
predicted_classes = model.predict_classes(scaled_X_test)

# setup the true classes: just 250 hotdogs followed by 250 not hotdogs
y_true = np.concatenate((np.zeros((250,)), np.ones((250,))))
from sklearn.metrics import classification_report
print(classification_report(y_true, predicted_classes, target_names=['hotdog', 'not hotdog']))           

這樣我們就可以看到我們模型的比較重要的一些評估結果了，比如準确率什麼的。

但是我們既然辛辛苦苦訓練了，我們就要好好把玩一下這個模型。我們可以直接用下面這段代碼來預測一個圖檔裡面是不是有熱狗。在這之前需要我們先建立一個名叫foo的檔案夾，并把你想要測試的圖檔放進去。

from PIL import Image
import numpy as np
from skimage import transform


from IPython.display import Image as ipy_Image
from IPython.display import display

# 定義一個加載圖檔的函數，使我們的圖檔變成np array
def load(filename):
   np_image = Image.open(filename)
   np_image = np.array(np_image).astype('float32')/255
   np_image = transform.resize(np_image, (32, 32, 1))
   np_image = np.expand_dims(np_image, axis=0)
   return np_image

import os
from os.path import join

image_dir = './foo'
os.listdir(image_dir)
img_paths = [join(image_dir,filename) for filename in os.listdir(image_dir)]

index_number = 0

image = load(img_paths[index_number])
score = model.predict(image)
result = model.predict_classes(image)
print(score[0][0], result)
display(ipy_Image(img_paths[index_number]))           

比如我們這裡上傳一張直播中網友們發來的圖檔，這張圖檔在直播的時候成功騙過了模型，得分最高。

我們運作一下 就可以看到結果了。

我們可以看到這個圖檔完美騙過了我們的模型，幾乎達到了1。大家也可以拿這個模型測試一下自己身邊長的像是熱狗但是又不是熱狗的東西，看看到底能得多少分～