如何自己训练一个热狗识别模型 | 《阿里云机器学习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。大家也可以拿这个模型测试一下自己身边长的像是热狗但是又不是热狗的东西，看看到底能得多少分～