机器学习CV代码练习（二）之训练模型大致流程+鸢尾花图像分类

Myth丶恋晨 2022-10-18 12:56 147阅读 0赞

### CV视频代码练习 ###

*  1. 机器学习代码大体流程（以自动驾驶部分代码为例）
 *  2. 鸢尾花图像分类（包含：加载数据、类变为数值变为One-hot、定义网络、K折交叉验证、保存模型、加载模型用其预测）
 *   *  预处理Y：str变为数值变为One-hot
     *  预处理X
     *  定义模型
     *  K折交叉验证
     *  保存模型
     *  加载模型，预测结果

# 1. 机器学习代码大体流程（以自动驾驶部分代码为例） #

包括：

1.  定义网络模型Model
2.  图像预处理image\_transformation
3.  产生批处理数据batch\_generator
4.  加载数据，分割为Train和Val
5.  想要保留中间过程：Callback
6.  训练模型(过程：epoch、batch\_size、per\_epoch：将数据整体全部训练epoch遍，其中每一遍内有per\_epoch次，每次训练batch\_size数据)
7.  [使用Tensorboard查看训练过程][Tensorboard]
8.  保存模型

# -*- coding: utf-8 -*-
    
    import numpy as np
    from keras.optimizers import SGD, Adam
    from keras.layers.core import Dense, Dropout, Activation
    from keras.layers import Conv2D, MaxPooling2D, Flatten, PReLU
    from keras.models import Sequential, Model
    from keras import backend as K
    from keras.regularizers import l2
    import os.path
    import csv
    import cv2
    import glob
    import pickle
    from sklearn.utils import shuffle
    from sklearn.model_selection import train_test_split
    import json
    from keras import callbacks
    import math
    from matplotlib import pyplot
    
    SEED = 13
    
    def get_model(shape):
        ''' 预测方向盘角度: 以图像为输入, 预测方向盘的转动角度 shape: 输入图像的尺寸, 例如(128, 128, 3) '''
    
        model = Sequential()
        
        model.add(Conv2D(8, (5, 5), strides=(1, 1), padding="valid", activation='relu', input_shape=shape))
        model.add(MaxPooling2D(pool_size=(2,2)))
        
        model.add(Flatten())
        
        model.add(Dense(128, activation='relu'))
        model.add(Dense(1, activation='linear'))
    
        sgd = SGD(lr=0.01)
        model.compile(optimizer=sgd, loss='mean_squared_error')
        return model
    
    
    def image_transformation(img_address, degree, data_dir):
        img = cv2.imread(data_dir + img_address)
        img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    
        return (img, degree)
    
    
    def batch_generator(x, y, batch_size, shape, training=True, data_dir='data/', monitor=True, yieldXY=True, discard_rate=0.95):
        """ 产生批处理的数据的generator x: 文件路径list y: 方向盘的角度 training: 值为True时产生训练数据 值为True时产生validation数据 batch_size: 批处理大小 shape: 输入图像的尺寸(高, 宽, 通道) data_dir: 数据目录, 包含一个IMG文件夹 monitor: 保存一个batch的样本为 'X_batch_sample.npy‘ 和'y_bag.npy’ yieldXY: 为True时, 返回(X, Y) 为False时, 只返回 X only discard_rate: 随机丢弃角度为零的训练数据的概率 """
        
        if training:
            y_bag = []
            x, y = shuffle(x, y)
            new_x = x
            new_y = y
        else:
            new_x = x
            new_y = y
        
        offset = 0
        while True: 
            X = np.empty((batch_size, *shape))
            Y = np.empty((batch_size, 1))
    
            for example in range(batch_size):
                img_address, img_steering = new_x[example + offset], new_y[example + offset]
                
                if training:
                    img, img_steering = image_transformation(img_address, img_steering, data_dir)
                else:
                    img = cv2.imread(data_dir + img_address)
                    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
                
                X[example,:,:,:] = cv2.resize(img[80:140, 0:320], (shape[0], shape[1]) ) / 255 - 0.5#图像0-255之间，变成归一化
                #第几个；高度；宽度；第几个通道
                Y[example] = img_steering
                if training:
                    y_bag.append(img_steering)
                
                ''' 到达原来数据的结尾, 从头开始 '''
                if (example + 1) + offset > len(new_y) - 1:
                    x, y = shuffle(x, y)
                    new_x = x
                    new_y = y
                    offset = 0
            if yieldXY:
                yield (X, Y)
            else:
                yield X
    
            offset = offset + batch_size
            if training:
                np.save('y_bag.npy', np.array(y_bag) )
                np.save('Xbatch_sample.npy', X ) 
    
    
    if __name__ == '__main__':
    
        data_path = 'data/'
        with open(data_path + 'driving_log.csv', 'r') as csvfile:
            file_reader = csv.reader(csvfile, delimiter=',')
            log = []
            for row in file_reader:
                log.append(row)
    
        log = np.array( log )
        # 去掉文件第一行
        log = log[1:,:] 
        
        # 判断图像文件数量是否等于csv日志文件中记录的数量
        ls_imgs = glob.glob(data_path+ 'IMG/*.jpg')#ls_imgs 所有文件名
        assert len(ls_imgs) == len(log)*3, 'number of images does not match'
    
        # 使用20%的数据作为测试数据
        validation_ratio = 0.2
        shape = (128, 128, 3)# 输入尺寸
        batch_size =32#每批次32个
        nb_epoch = 2#将所有数据跑2遍
    
        x_ = log[:, 0] 
        y_ = log[:, 3].astype(float)#角度
        x_, y_ = shuffle(x_, y_)
        X_train, X_val, y_train, y_val = train_test_split(x_, y_, test_size=validation_ratio, random_state=SEED)#拆分数据
    
        print('batch size: {}'.format(batch_size))
        print('Train set size: {} | Validation set size: {}'.format(len(X_train), len(X_val)))
            
        samples_per_epoch = batch_size #每个epoch中跑多少
        # 使得validation数据量大小为batch_size的整数陪
        nb_val_samples = len(y_val) - len(y_val)%batch_size
        model = get_model(shape)#自己定义网络结构的方法
        print(model.summary())#打印网络结构，及参数数量
    #想要保留中间过程——callback。传进函数，保存信息。
        # 根据validation loss保存最优模型
        save_best = callbacks.ModelCheckpoint('best_model.h5', monitor='val_loss', verbose=1, 
                                             save_best_only=True, mode='min')#训练网络希望每次训练过程中保存好的结果。存的文件名等信息。
    
        # 如果训练持续没有validation loss的提升, 提前结束训练 
        early_stop = callbacks.EarlyStopping(monitor='val_loss', min_delta=0, patience=15, 
                                             verbose=0, mode='auto')#选择epoch=10000，到100时候就保持不变了，需要提前停止。看15步内没有变化。
        tbCallBack = callbacks.TensorBoard(log_dir='./Graph',write_graph=True,write_image=True)
        callbacks_list = [early_stop, save_best,tbCallBack ]# 三个callback放在列表里
    
        history = model.fit_generator(batch_generator(X_train, y_train, batch_size, shape, training=True),
                                      steps_per_epoch = samples_per_epoch,
                                      validation_steps = nb_val_samples // batch_size,
                                      validation_data = batch_generator(X_val, y_val, batch_size, shape, 
                                                                      training=False, monitor=False),
                                      epochs=nb_epoch, verbose=1, callbacks=callbacks_list)# (batch_generator产生训练数据，
    
    #存history文件
        with open('./trainHistoryDict.p', 'wb') as file_pi:
            pickle.dump(history.history, file_pi)#保存history的属性history
    
    #画图，将图片存起来
        pyplot.plot(history.history['loss'])
        pyplot.plot(history.history['val_loss'])
        pyplot.title('model train vs validation loss')
        pyplot.ylabel('loss')
        pyplot.xlabel('epoch')
        pyplot.legend(['train', 'validation'], loc='upper right')
        pyplot.savefig('train_val_loss.jpg')
    
        # 保存模型
        with open('model.json', 'w') as f:
                f.write( model.to_json() )
        model.save('model.h5')
        print('Done!')

# 2. 鸢尾花图像分类（包含：加载数据、类变为数值变为One-hot、定义网络、K折交叉验证、保存模型、加载模型用其预测） #

鸢尾花Iris数据集链接：链接：[https://pan.baidu.com/s/1M1BpoKLLoAOFMBjw8i5H1w][https_pan.baidu.com_s_1M1BpoKLLoAOFMBjw8i5H1w]  
提取码：63an

#!/usr/bin/env python
    # coding: utf-8
    
    # In[1]:
    import pandas as pd 
    import numpy as np
    from keras import utils as np_utils# One-hot编码
    # In[2]:
    seed = 20
    np.random.seed(seed)

## 预处理Y：str变为数值变为One-hot ##

# In[3]:
    # 预处理数据Y 有两种方法：
    # 法一：预处理数据Y（通过sklearn的预处理方法将字符串数值编码；在调用keras中的utils方法将数值编码变为One-hot编码）
    data = pd.read_csv('Iris.csv')
    
    from sklearn.preprocessing import LabelEncoder# 数值编码
    from keras import utils as np_utils# One-hot编码
    
    Y = data.iloc[:,5]
    # 转换成数值编码
    encoder = LabelEncoder()
    Y_encoded = encoder.fit_transform(Y)
    print(Y_encoded)
    # 转换成One-hot编码
    Y_onehot = np_utils.to_categorical(Y_encoded)
    print("Y_onehot前五个值\n",Y_onehot[0:5])
    # In[4]:
    # # 法二：预处理数据Y：利用pandas中的loc方法将字符串转换成数值编码，再使用最后一列，利用keras中的utils进行one-hot编码
    # data = pd.read_csv('Iris.csv')
    
    # # Id SepalLengthCm SepalWidthCm PetalLengthCm PetalWidthCm Species
    # # 选取Y：Species将Species的三种分类分别编为数值
    # data.loc[data["Species"]=="Iris-setosa", "Species"] = 0
    # data.loc[data["Species"]=="Iris-versicolor", "Species"] = 1
    # data.loc[data["Species"]=="Iris-virginica", "Species"] = 2
    # #使用最后一列
    # Y = data.iloc[:,5]
    # # One-hot编码
    # from keras import utils as np_utils
    # Y_onehot = np_utils.to_categorical(Y)
    # print("Y_onehot前五个值\n",Y_onehot[0:5])

## 预处理X ##

# In[5]:
    # 预处理数据X 有两种方法：
    # 法一：选取X的部分：删除最后一列
    # X = df.drop(["Species"], axis=1)
    # print(X)
    # In[6]:
    # 法二：选取X的部分：选取前几列
    X = data.iloc[:,[1,2,3,4]].values
    print(X)
    # In[7]:
    # print(X)
    # print(Y_onehot)

## 定义模型 ##

# In[8]:
    from keras.models import Sequential#模型架构
    from keras.layers import Dense# Dense层
    from keras import optimizers#优化器
    # 代码提示快捷键：按住Shift+双击Tab
    
    def f1_score(y_true, y_pred): 
    
        # Count positive samples. 
        c1 = K.sum(K.round(K.clip(y_true * y_pred, 0, 1))) 
        c2 = K.sum(K.round(K.clip(y_pred, 0, 1))) 
        c3 = K.sum(K.round(K.clip(y_true, 0, 1))) 
        # If there are no true samples, fix the F1 score at 0. 
        if c3 == 0: 
            return 0 
        # How many selected items are relevant? 
        precision = c1/c2 
        # How many relevant items are selected? 
        recall = c1/c3 
        # Calculate f1_score 
        f1_score = 2 * (precision * recall)/(precision + recall) 
        return f1_score 
    
    def dense_model():
        model = Sequential()
        model.add(Dense(units=7,input_dim=4,kernel_initializer='random_uniform',activation="tanh"))
        model.add(Dense(3,activation="softmax"))
        model.summary()
    # model.compile(loss="mean_squared_error",optimizer="sgd",metrics=["accuracy","f1_score"])
        model.compile(loss="mean_squared_error",optimizer="sgd",metrics=["accuracy"])
        return model
    # 注意在其他部分加载模型时：
    # from keras import models 
    # model = models.load_model(model_path, custom_objects= {'f1_score': f1_score})

## K折交叉验证 ##

K折交叉验证——将所有数据分成K份，每次训练（k-1）份，其余一份作为验证。  
交叉验证目的：得到更可靠的模型。（消除因为训练集选择不好，而导致模型不好的情况）  
如何在scikit-learn模型中使用Keras？ ——通过用 KerasClassifier 或 KerasRegressor 类包装Keras模型，可将其用于scikit-learn。  
要使用这些包装，必须定义一个函数create\_model(model函数）创建并返回Keras，  
然后当构建 KerasClassifier 类时，把该函数create\_model(model函数）传递给 build\_fn 参数。  
KerasClassifier类 的构建器可以采取默认参数，并将其被传递给 model.fit() 的调用函数，比如 epochs数目和批尺寸（batch size)；  
也可以使用新的参数，使之能够传递给自定义的create\_model()函数。  
这些新的参数，也必须由使用默认参数的 create\_model() 函数的签名定义。

# In[9]:
    # K折交叉验证——将所有数据分成K份，每次训练（k-1）份，其余一份作为验证。
    # 交叉验证目的：得到更可靠的模型。（消除因为训练集选择不好，而导致模型不好的情况）
    # 如何在scikit-learn模型中使用Keras？ ——通过用 KerasClassifier 或 KerasRegressor 类包装Keras模型，可将其用于scikit-learn。
    # 要使用这些包装，必须定义一个函数create_model(model函数）创建并返回Keras，
    # 然后当构建 KerasClassifier 类时，把该函数create_model(model函数）传递给 build_fn 参数。
    # KerasClassifier类 的构建器可以采取默认参数，并将其被传递给 model.fit() 的调用函数，比如 epochs数目和批尺寸（batch size)；
    # 也可以使用新的参数，使之能够传递给自定义的create_model()函数。
    # 这些新的参数，也必须由使用默认参数的 create_model() 函数的签名定义。
    # def create_model(dropout_rate=0.0):
    # ...
    # return model 
    # model = KerasClassifier(build_fn=create_model, nb_epoch=10，dropout_rate=0.2)
    # 获取交叉验证结果的类预测、类概率估计、平均精度的方法：
    # pred = estimator.predict(X_test)#返回给定测试数据的类预测。
    # pred1=estimator.predict_proba(X_test)#返回给定类概率估计。
    # pred3=estimator.score(X_test,Y_test)#返回给定测试数据和标签的平均精度。
    
    #模型评估：
    # 直接比对真实值与预测值
    # y_predict=estimator.predict(x_test)
    # y_test==y_predict
    # 计算准确率
    # accuracy=estimator.score(x_test,y_test)
    from keras.wrappers.scikit_learn import KerasClassifier# KerasClassifier将create_model(model函数）
    from sklearn.model_selection import cross_val_score #性能评估方法，选择不同的性能度量函数
    from sklearn.model_selection import KFold #k-fold:实现了分层交叉切分
    #KerasClassifier包装Keras模型
    estimator = KerasClassifier(build_fn=dense_model,epochs=20,batch_size=1,verbose=1)
    # 包装后的estimator用于Sklearn
    kfold = KFold(n_splits=10,shuffle=True,random_state=seed)
    
    # 性能评估方法
    result = cross_val_score(estimator,X,Y_onehot,cv=kfold)
    print("Accuracy of cross validation,mean %.2f, std %.2f"%(result.mean(),result.std()))

## 保存模型 ##

# In[10]:
    from keras.models import model_from_json# 保存模型
    # 实例化一个estimator，计算生产一个机器学习模型
    estimator.fit(X,Y_onehot)
    # 保存模型
    model_json = estimator.model.to_json()#保存网络结构
    with open('model.json','w') as json_file:
        json_file.write(model_json)
    estimator.model.save_weights('model.h5')#保存网络权重
    print("saved model to disk")

## 加载模型，预测结果 ##

# In[11]:
    from keras.models import model_from_json
    #加载已经保存的模型，用其进行预测
    json_file = open('model.json','r')
    loaded_model_json = json_file.read()
    json_file.close()
    
    loaded_model = model_from_json(loaded_model_json)
    loaded_model.load_weights("model.h5")
    print("load model from disk")
    #模型评估：
    # 直接比对真实值与预测值
    # y_predict=estimator.predict(x_test)
    # y_test==y_predict
    # 计算准确率
    # accuracy=estimator.score(x_test,y_test)
    
    predicted = loaded_model.predict(X)
    print("predicted probability："+str(predicted))
    
    predicted_label = loaded_model.predict_classes(X)
    print("predicted label："+str(predicted_label))

[Tensorboard]: https://blog.csdn.net/qq_37486501/article/details/118598891
[https_pan.baidu.com_s_1M1BpoKLLoAOFMBjw8i5H1w]: https://pan.baidu.com/s/1M1BpoKLLoAOFMBjw8i5H1w