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 # 訓練輪數(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ù)集并進行歸一化處理
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.
    # 將標簽矩陣轉(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
# 加載訓練和測試數(shù)據(jù)
(X_train_val, y_train_val), (X_test, y_test) = load_dataset('/data/emnist/mnist.npz')
# 劃分訓練集和驗證集
X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val,
                                                  test_size=0.2, random_state=42)
# 對數(shù)據(jù)進行歸一化處理
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)
    # 堆疊密集塊和過渡層
    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ù)和評價指標
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)
# 訓練模型
history = model.fit(X_train, y_train, batch_size=batch_size, epochs=epochs,
          verbose=1, validation_data=(X_val, y_val), callbacks=[earlystop])

# 對模型進行評估
score = model.evaluate(X_test, y_test, verbose=0)
# 計算各項指標
test_accuracy = score[1]
print('Test accuracy:', test_accuracy)
# 保存模型
model.save('densenet121.h5')