import numpy as np
from tensorflow.keras.models import Sequential, Model
from tensorflow.keras.layers import Dense, Dropout, Flatten
from tensorflow.keras.layers import Conv2D, MaxPooling2D, BatchNormalization
from tensorflow.keras.layers import Input, concatenate
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.regularizers import l2
from tensorflow.keras.callbacks import EarlyStopping
from sklearn.model_selection import train_test_split
from PIL import Image

batch_size = 128 # 批量大小
num_classes = 26 # 分類數(shù)目
epochs = 50 # 訓(xùn)練輪數(shù)

# 加載數(shù)據(jù)集
def load_dataset(path):
    with np.load(path) as data:
        X_train = data['X_train']
        y_train = data['y_train']
        X_test = data['X_test']
        y_test = data['y_test']
    return (X_train, y_train), (X_test, y_test)
# 加載數(shù)據(jù)集并進(jìn)行歸一化處理
def preprocess_data(X_train, y_train, X_test, y_test):
    # 將圖像矩陣歸一化到0-1之間
    X_train = X_train.astype('float32') / 255.
    X_test = X_test.astype('float32') / 255.
    # 將標(biāo)簽矩陣轉(zhuǎn)換為one-hot編碼
    y_train = keras.utils.to_categorical(y_train, num_classes)
    y_test = keras.utils.to_categorical(y_test, num_classes)
    return X_train, y_train, X_test, y_test
# 加載訓(xùn)練和測(cè)試數(shù)據(jù)
(X_train_val, y_train_val), (X_test, y_test) = load_dataset('/data/emnist/mnist.npz')
# 劃分訓(xùn)練集和驗(yàn)證集
X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val,
                                                  test_size=0.2, random_state=42)
# 對(duì)數(shù)據(jù)進(jìn)行歸一化處理
X_train, y_train, X_val, y_val = preprocess_data(X_train, y_train, X_val, y_val)
X_test, y_test = preprocess_data(X_test, y_test, [], [])

# 定義dense_block函數(shù)
def dense_block(x, blocks, growth_rate):
    for i in range(blocks):
        x1 = BatchNormalization()(x)
        x1 = Conv2D(growth_rate * 4, (1, 1), padding='same', activation='relu',
                    kernel_initializer='he_normal')(x1)
        x1 = BatchNormalization()(x1)
        x1 = Conv2D(growth_rate, (3, 3), padding='same', activation='relu',
                    kernel_initializer='he_normal')(x1)
        x = concatenate([x, x1])
    return x
# 定義transition_layer函數(shù)
def transition_layer(x, reduction):
    x = BatchNormalization()(x)
    x = Conv2D(int(x.shape.as_list()[-1] * reduction), (1, 1), activation='relu',
                kernel_initializer='he_normal')(x)
    x = MaxPooling2D((2, 2), strides=(2, 2))(x)
    return x
# 構(gòu)建DenseNet網(wǎng)絡(luò)
def DenseNet(input_shape, num_classes, dense_blocks=3, dense_layers=-1,
             growth_rate=12, reduction=0.5, dropout_rate=0.0, weight_decay=1e-4):
    # 指定初始通道數(shù)和塊數(shù)
    depth = dense_blocks * dense_layers + 2
    in_channels = 2 * growth_rate
    inputs = Input(shape=input_shape)
    # 第一層卷積
    x = Conv2D(in_channels, (3, 3), padding='same', use_bias=False,
               kernel_initializer='he_normal')(inputs)
    # 堆疊密集塊和過(guò)渡層
    for i in range(dense_blocks):
        x = dense_block(x, dense_layers, growth_rate)
        in_channels += growth_rate * dense_layers
        if i != dense_blocks - 1:
            x = transition_layer(x, reduction)
    # 全局平均池化
    x = BatchNormalization()(x)
    x = Activation('relu')(x)
    x = GlobalAveragePooling2D()(x)
    # 輸出層
    outputs = Dense(num_classes, activation='softmax',
                     kernel_initializer='he_normal')(x)
    # 定義模型
    model = Model(inputs=inputs, outputs=outputs, name='DenseNet')
    return model
# 構(gòu)建DenseNet121網(wǎng)絡(luò)
model = DenseNet(input_shape=(28, 28, 1), num_classes=num_classes, dense_blocks=3,
           dense_layers=4, growth_rate=12, reduction=0.5, dropout_rate=0.0,
           weight_decay=1e-4)
# 指定優(yōu)化器、損失函數(shù)和評(píng)價(jià)指標(biāo)
opt = Adam(lr=0.001, beta_1=0.9, beta_2=0.999, epsilon=1e-08)
model.compile(optimizer=opt, loss='categorical_crossentropy', metrics=['accuracy'])
# 輸出模型概況
model.summary()

# 定義早停策略
earlystop = EarlyStopping(monitor='val_loss', min_delta=0.0001, patience=5,
                          verbose=1, mode='auto', restore_best_weights=True)
# 訓(xùn)練模型
history = model.fit(X_train, y_train, batch_size=batch_size, epochs=epochs,
          verbose=1, validation_data=(X_val, y_val), callbacks=[earlystop])

# 對(duì)模型進(jìn)行評(píng)估
score = model.evaluate(X_test, y_test, verbose=0)
# 計(jì)算各項(xiàng)指標(biāo)
test_accuracy = score[1]
print('Test accuracy:', test_accuracy)
# 保存模型
model.save('densenet121.h5')