自定義卷積網路完成分類。影象預處理（直方圖均衡化增加對比度）。

使用資料：德國交通訊號識別，其中train/test dataset的Images and annotations及test dataset的Extended annotations including class ids。

實驗結果

資料及程式碼組織結構：

訓練過程與結果：

程式碼

"""
@file: tranfficSignRec.py
@time: 2018/10/26
"""
import pandas as pd
import numpy as np
from skimage import
 
 io, color, exposure, transform
import glob
import h5py
from keras.models import Sequential, model_from_json
from keras.layers.core import Dense, Dropout, Activation, Flatten
from keras.layers.convolutional import Conv2D
from keras.layers.pooling import MaxPooling2D
from keras.optimizers import SGD
from
 
 keras.callbacks import LearningRateScheduler, ModelCheckpoint
from matplotlib import pyplot as plt
import os
from pathlib import PurePath
import warnings

warnings.filterwarnings('ignore')  # 忽略警告

NUM_CLASSES = 43  # 43種交通標誌
IMG_SIZE = 48  # 影象大小歸一化為48

batch_size = 32 # 訓練的引數
nb_epoch = 10
lr = 0.01
 



# 影象直方圖均衡化（調整對比度）、取中心、resize
def preprocess_img(img):
    hsv = color.rgb2hsv(img)
    hsv[:, :, 2] = exposure.equalize_hist(hsv[:, :, 2])
    img = color.hsv2rgb(hsv)

    ms = min(img.shape[:2])
    xx = (img.shape[0] - ms) // 2
    yy = (img.shape[1] - ms) // 2
    img = img[xx:xx + ms, yy:yy + ms, :]

    img = transform.resize(img, (IMG_SIZE, IMG_SIZE))

    return img


# 根據路徑（圖片上層目錄）得到標籤
def get_class(img_path):
    return int(PurePath(img_path).parts[- 2])


def readfile():
    # 讀取所有圖片、標籤（onehot），存放至h5py檔案
    try:
        with h5py.File('X.h5') as hf:
            X, Y = hf['imgs'][:], hf['labels'][:]

        print("Loaded images from X.h5")

    except BaseException:
        print("Error in reading X.h5. Processing all images...")
        root_dir = r'../data/GTSRB/Final_Training/Images'
        imgs = []
        labels = []

        all_img_paths = glob.glob(os.path.join(root_dir,'*/*.ppm'))  # 提取所有ppm檔案完整路徑

        np.random.shuffle(all_img_paths)  # 打散
        for img_path in all_img_paths:
            try:
                img = preprocess_img(io.imread(img_path))
                label = get_class(img_path)
                imgs.append(img)
                labels.append(label)

                if len(imgs) % 1000 == 0:
                    print("Processed %d/%d" %(len(imgs), len(all_img_paths)))
            except BaseException:
                print('missed', img_path)
                pass
        X = np.array(imgs, dtype='float32')
        # labels陣列轉onehot
        Y = np.eye(len(labels), NUM_CLASSES, dtype=np.uint8)[labels]

        # 可以加速載入與處理
        with h5py.File('X.h5', 'w') as hf:
            hf.create_dataset('imgs', data=X)
            hf.create_dataset('labels', data=Y)
    return X, Y


def cnn_model():

    model = Sequential()

    model.add(Conv2D(32,(3,3),padding='same',activation='relu',input_shape=(IMG_SIZE,IMG_SIZE,3)))
    model.add(Conv2D(32, (3, 3), activation='relu'))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Dropout(0.2))

    model.add(Conv2D(64, (3, 3), padding='same', activation='relu'))
    model.add(Conv2D(64, (3, 3), padding='same', activation='relu'))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Dropout(0.2))

    model.add(Conv2D(128, (3, 3), padding='same', activation='relu'))
    model.add(Conv2D(128, (3, 3), padding='same', activation='relu'))
    model.add(MaxPooling2D(pool_size=(2, 2)))
    model.add(Dropout(0.2))

    model.add(Flatten())
    model.add(Dense(512, activation='relu'))
    model.add(Dropout(0.5))
    model.add(Dense(NUM_CLASSES, activation='softmax'))

    sgd = SGD(lr=lr, decay=1e-6, momentum=0.9, nesterov=True)
    model.compile(
        loss='categorical_crossentropy',
        optimizer=sgd,
        metrics=['accuracy'])

    return model

# 每10個epoch學習率遞減0.1倍
def lr_schedule(epoch):

    return lr * (0.1 ** int(epoch / 10))


def train():
    model = cnn_model()
    print(model.summary())
    X, Y = readfile()
    history = model.fit(X,Y,
        batch_size=batch_size,
        epochs=nb_epoch,
        validation_split=0.2,
        shuffle=True,
        verbose=2,
        callbacks=[
            LearningRateScheduler(lr_schedule),
            # ModelCheckpoint('model.h5',save_best_only=True)
        ])
    # 視覺化訓練曲線（train,val）
    plt.figure(figsize=(8, 4))
    plt.subplot(1, 2, 1)
    plot_train_history(history, 'loss', 'val_loss')
    plt.subplot(1, 2, 2)
    plot_train_history(history, 'acc', 'val_acc')
    plt.show()
    return model

def plot_train_history(history, train_metrics, val_metrics):
    plt.plot(history.history.get(train_metrics), '-o')
    plt.plot(history.history.get(val_metrics), '-o')
    plt.ylabel(train_metrics)
    plt.xlabel('Epochs')
    plt.legend(['train', 'validation'])


# 在測試集上測試
def test(model):
    test = pd.read_csv('../data/GTSRB/GT-final_test.csv',sep=';')
    X_test = []
    y_test = []
    for file_name, class_id in zip(list(test['Filename']), list(test['ClassId'])):
        img_path = os.path.join('../data/GTSRB/Final_Test/Images/', file_name)
        X_test.append(preprocess_img(io.imread(img_path)))
        y_test.append(class_id)
    X_test = np.array(X_test)
    y_test = np.array(y_test)

    print("X_test.shape: ", X_test.shape)
    print("y_test.shape: ", y_test.shape)

    y_pred = model.predict_classes(X_test) # 返回預測值
    acc = np.sum(y_pred == y_test) / np.size(y_pred)
    print("Test accuracy = {} ".format(acc))


if __name__ == '__main__':
    model=train()
    test(model)

參考：https://github.com/erhwenkuo/deep-learning-with-keras-notebooks