基于keras模拟深度残差网络(ResNet50)实现花贲识别

简单点说深度残差网络可以有效避免神经网络在层数过高之后引起的梯度消失问题，据说可以让神经网络层数达到上千层。之所以不能一直增加，肯定是层数过高后又必然有梯度爆炸的问题。随着神经元个数增加，其非线性拟合能力越来越强，既可以拟合更高维度事物。

网络结构与网上各种 讲解无异，损失函数因为花分位5类，所以这里选择多分类损失函数categorical_crossentropy。很多情况下分类任务选择交叉熵，回归任务选择MSE，这在机器学习里面基本是定式。之所以要这么做，主要原因是交叉熵作为损失函数求解是一个凸优化问题，很容易找到极值点，而MSE则很容易陷入局部最优解，从而大多数情况下效果不如交叉熵。

多分类问题往往在最后一层选择softmax分类函数。对于标签，需要做one_hot编码，这一点在代码中有体现。至于多分类为什么用交叉熵更合适，可以看这篇文章：https://zhuanlan.zhihu.com/p/35709485

激活函数选择LeakyReLU，就算量小，而且收敛比较快。

数据集下载地址：https://storage.googleapis.com/download.tensorflow.org/example_images/flower_photos.tgz

import os
import cv2
import glob
import numpy as np
import tensorflow as tf
import matplotlib.pyplot as plt

from PIL import Image
from tensorflow.keras.models import *
from tensorflow.keras.layers import *
from tensorflow.keras.utils import plot_model
#from keras import backend as K
#K.set_image_data_format('channels_last')

os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
tf.compat.v1.logging.set_verbosity(tf.compat.v1.logging.ERROR)
print('Tensorflow"{}'.format(tf.__version__))

imgSize = 128
classnum = {'daisy':0,'dandelion':1,'roses': 2, 'sunflowers': 3, 'tulips': 4}

class MakeTfDataset(object):

    def __init__(self):
        self.dataroot = ''
        self.X_data = []  #图片数据
        self.Y_data = []  #图片标签

    def LoadOneImage(self,addr,shape=(imgSize, imgSize)):
        img = cv2.imread(addr)
        img = cv2.resize(img,shape,interpolation=cv2.INTER_CUBIC)
        img = img.astype(np.float32)
        return img

    def LoadImageData(self):
        for i in classnum.keys():
            images = glob.glob(self.dataroot + '/' + str(i) + '/*.jpg')
            labels = int(classnum[i])
            print(labels,'\t\t',i)
            for img in images:
                img = self.LoadOneImage(img)
                self.X_data.append(img)
                self.Y_data.append(labels)

    def HandleTrainDataset(self):
        self.X_data = np.array(self.X_data)
        self.Y_data = np.array(self.Y_data)
        dx_train = tf.data.Dataset.from_tensor_slices(self.X_data)
        dy_train = tf.data.Dataset.from_tensor_slices(self.Y_data).map(lambda  z:tf.one_hot(z,len(classnum)))
        train_dataset = tf.data.Dataset.zip((dx_train,dy_train)).shuffle(50000).repeat().batch(256).prefetch(tf.data.experimental.AUTOTUNE)
        return train_dataset

class Net(object):

    def __init__(self):
        self.mDataset = MakeTfDataset()
        self.saver_root = './weights/'
        self.BuildModel()

    # 恒等模块——identity_block
    def identity_block(self, X, f, filters, stage, block,activation="LeakyReLU"):

        conv_name_base = "res" + str(stage) + block + "_branch"
        bn_name_base = "bn" + str(stage) + block + "_branch"

        # 过滤器
        F1, F2, F3 = filters

        # 保存输入值,后面将需要添加回主路径
        X_shortcut = X

        # 主路径第一部分
        X = Conv2D(filters=F1, kernel_size=(1, 1), strides=(1, 1), padding="valid", name=conv_name_base + "2a",)(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2a")(X)
        X = Activation(activation)(X)

        # 主路径第二部分
        X = Conv2D(filters=F2, kernel_size=(f, f), strides=(1, 1), padding="same",name=conv_name_base + "2b")(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2b")(X)
        X = Activation(activation)(X)

        # 主路径第三部分
        X = Conv2D(filters=F3, kernel_size=(1, 1), strides=(1, 1), padding="valid",name=conv_name_base + "2c")(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2c")(X)

        # 主路径最后部分,为主路径添加shortcut并通过relu激活
        X = add([X, X_shortcut])
        X = Activation(activation)(X)
        return X

    def convolutional_block(self,X, f, filters, stage, block,s=2,activation="LeakyReLU"):

        conv_name_base = "res" + str(stage) + block + "_branch"
        bn_name_base = "bn" + str(stage) + block + "_branch"

        # 过滤器
        F1, F2, F3 = filters

        # 保存输入值,后面将需要添加回主路径
        X_shortcut = X

        # 主路径第一部分
        X = Conv2D(filters=F1, kernel_size=(1, 1), strides=(s, s), padding="valid")(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2a")(X)
        X = Activation(activation)(X)

        # 主路径第二部分
        X = Conv2D(filters=F2, kernel_size=(f, f), strides=(1, 1), padding="same",name=conv_name_base + "2b")(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2b")(X)
        X = Activation(activation)(X)

        # 主路径第三部分
        X = Conv2D(filters=F3, kernel_size=(1, 1), strides=(1, 1), padding="valid",name=conv_name_base + "2c")(X)
        X = BatchNormalization(axis=3, name=bn_name_base + "2c")(X)

        # shortcut路径
        X_shortcut = Conv2D(filters=F3, kernel_size=(1, 1), strides=(s, s), padding="valid",name=conv_name_base + "1")(X_shortcut)
        X_shortcut = BatchNormalization(axis=3, name=bn_name_base + "1")(X_shortcut)

        # 主路径最后部分,为主路径添加shortcut并通过relu激活
        X = add([X, X_shortcut])
        X = Activation(activation)(X)
        return X

    def BuildModel(self,input_shape=(imgSize, imgSize, 3), classes=5):

        X_input = Input(input_shape)

        # Zero-Padding
        X = ZeroPadding2D((3, 3))(X_input)

        # Stage 1
        X = Conv2D(64, kernel_size=(7, 7), strides=(2, 2), name="conv1")(X)
        X = BatchNormalization(axis=3, name="bn_conv1")(X)
        X = Activation("LeakyReLU")(X)
        X = MaxPooling2D(pool_size=(3, 3), strides=(2, 2))(X)

        # Stage 2
        X = self.convolutional_block(X, f=3, filters=[64, 64, 256], stage=2, block="a", s=1)
        X = self.identity_block(X, f=3, filters=[64, 64, 256], stage=2, block="b")
        X = self.identity_block(X, f=3, filters=[64, 64, 256], stage=2, block="c")

        # Stage 3
        X = self.convolutional_block(X, f=3, filters=[128, 128, 512], stage=3, block="a", s=2)
        X = self.identity_block(X, f=3, filters=[128, 128, 512], stage=3, block="b")
        X = self.identity_block(X, f=3, filters=[128, 128, 512], stage=3, block="c")
        X = self.identity_block(X, f=3, filters=[128, 128, 512], stage=3, block="d")

        # Stage 4
        X = self.convolutional_block(X, f=3, filters=[256, 256, 1024], stage=4, block="a", s=2)
        X = self.identity_block(X, f=3, filters=[256, 256, 1024], stage=4, block="b")
        X = self.identity_block(X, f=3, filters=[256, 256, 1024], stage=4, block="c")
        X = self.identity_block(X, f=3, filters=[256, 256, 1024], stage=4, block="d")
        X = self.identity_block(X, f=3, filters=[256, 256, 1024], stage=4, block="e")
        X = self.identity_block(X, f=3, filters=[256, 256, 1024], stage=4, block="f")

        # Stage 5
        X = self.convolutional_block(X, f=3, filters=[512, 512, 2048], stage=5, block="a", s=2)
        X = self.identity_block(X, f=3, filters=[256, 256, 2048], stage=5, block="b")
        X = self.identity_block(X, f=3, filters=[256, 256, 2048], stage=5, block="c")

        # 最后阶段
        # 平均池化
        X = AveragePooling2D(padding="same")(X)
        # 输出层
        X = Flatten()(X)  # 展平
        X = Dense(classes, activation="softmax", name="fc" + str(classes))(X)

        self.model = Model(inputs=X_input, outputs=X, name="MyResNet50")
        self.model.compile(optimizer=tf.compat.v1.train.AdamOptimizer(1e-4),loss='categorical_crossentropy',metrics=['categorical_accuracy'])

    def LoadWeightModel(self):
        if tf.train.latest_checkpoint(self.saver_root) != None:
            # 加载模型的权值
            self.model.load_weights(self.saver_root + 'my_model')

    def train(self,epochs=None,saver=False,steps_per_epoch=100):
        self.mDataset.LoadImageData()
        self.model.fit(self.mDataset.HandleTrainDataset(),epochs=epochs,steps_per_epoch=steps_per_epoch)

        if saver:
            self.model.save_weights(self.saver_root + 'my_model')

if __name__ == '__main__':

    N = Net()
    train = True
    if train:
        N.mDataset.dataroot = 'F:/data/flower_photos'
        N.train(30, True)
        plot_model(N.model, to_file='model.png', show_shapes=True)
    else:
        N.LoadWeightModel()
        num = [0, 0, 0, 0, 0]
        photo = cv2.imread('C:/daisy4.jpg')
        img = cv2.resize(photo, (imgSize, imgSize), interpolation=cv2.INTER_CUBIC)
        img = Image.fromarray(img)  # 载入到内存中
        img = np.reshape(img, (-1, imgSize, imgSize, 3))  # 加一个维度
        pred = N.model.predict(img)  # 预测
        print(pred)
        # 返回最大概率值
        classes = pred[0].tolist().index(max(pred[0]))
        classType= ['daisy', 'dandelion', 'roses', 'sunflowers', 'tulips']
        cv2.putText(photo, classType[classes], (10, 30), cv2.FONT_ITALIC, 0.5, (0, 0, 255), 2)
        cv2.imshow("photo", photo)
        cv2.waitKey(0)