手势识别：使用EfficientNet模型迁移、VGG16模型迁移

分手后的思念是犯贱 2023-07-17 15:37 4阅读 0赞

# ![20191009191333910.png][] #

# [日萌社][Link 1] #

[人工智能AI：Keras PyTorch MXNet TensorFlow PaddlePaddle 深度学习实战（不定时更新）][AI_Keras PyTorch MXNet TensorFlow PaddlePaddle]

--------------------

### EfficientNet中的每个模型要求的输入形状大小 ###

每个网络要求的输入形状大小：   
    	EfficientNetB0 - (224, 224, 3)
    	EfficientNetB1 - (240, 240, 3)
    	EfficientNetB2 - (260, 260, 3)
    	EfficientNetB3 - (300, 300, 3)
    	EfficientNetB4 - (380, 380, 3)
    	EfficientNetB5 - (456, 456, 3)
    	EfficientNetB6 - (528, 528, 3)
    	EfficientNetB7 - (600, 600, 3)

--------------------

EfficientNet模型迁移的使用注意事项：
    	1.因为该模型的源码是在tensorflow 1.x的版本，并非是tensorflow 2.0的版本，因此在tensorflow 2.0环境中使用的话，
    	  需要用到tf.compat.v1.disable_eager_execution()，表示关闭默认的eager模式，但要注意的是，如果关闭默认的eager模式了的话，
    	  那么同时还使用tf.keras.callbacks.TensorBoard的话会报错，tf.keras.callbacks.ModelCheckpoint不会报错，
    	  那么解决的方式要么此时不使用TensorBoard，或者不关闭默认的eager模式。
    	2.layers.py中的class Swish中的call()函数返回值的修改建议。
    		如果使用了tf.compat.v1.disable_eager_execution()之后，报错No registered 'swish_f32' OpKernel for GPU devices compatible with node的话，
    		把 layers.py中的class Swish中的call()函数返回值 return tf.nn.swish(inputs) 修改为 return inputs * tf.math.sigmoid(inputs) 即可解决，
    		实际上底层是 tf.nn.swish(x) 封装了 x* tf.math.sigmoid(x)，不使用tf.nn.swish之后，即可也把tf.compat.v1.disable_eager_execution()给注释掉，
    		即不需要关闭默认的eager模式了，那么此时也可以正常同时使用TensorBoard。

--------------------

# **使用EfficientNet模型迁移 第一个版本** #

import multiprocessing
    import numpy as np
    import tensorflow as tf
    from tensorflow.keras.preprocessing.image import ImageDataGenerator
    from tensorflow.keras.callbacks import TensorBoard, Callback
    from tensorflow.keras.layers import Dense, GlobalAveragePooling2D
    from tensorflow.keras.models import Model
    from tensorflow.keras.optimizers import Adam, RMSprop
    from efficientnet import model as EfficientNet
    import os
    from sklearn.model_selection import train_test_split
    from tensorflow.keras.applications.vgg16 import VGG16, preprocess_input
    
    
    # 注意关闭默认的eager模式
    # tf.compat.v1.disable_eager_execution()
    
    num_epochs = 10
    batch_size = 10
    learning_rate = 0.001
    Dataset_dir = './Sign-Language-Digits-Dataset-master/Dataset/'
    Dataset_test_dir = './Sign-Language-Digits-Dataset-master/Examples/'
    
    Dataset_dir_list = []
    Dataset_test_dir_list = []
    lable_list = []
    train_img_fileNames = []
    train_img_lables = []
    test_img_fileNames = []
    test_img_lables = []
    input_size = 100
    optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)
    losses = tf.keras.losses.sparse_categorical_crossentropy
    metrics = [tf.keras.metrics.sparse_categorical_accuracy]
    
    
    """
    metrics 是 sparse_categorical_crossentropy，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_sparse_categorical_crossentropy
    metrics 是 acc，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_acc
    metrics 是 accuracy，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_accuracy
    
    """
    
    def decode_and_resize(filename, label):
        image_string = tf.io.read_file(filename)
        image_decoded = tf.image.decode_jpeg(image_string)
        image_resized = tf.image.resize(image_decoded, [100, 100]) / 255.0
        return image_resized, label
    
    def VGG_224x244_decode_and_resize(filename, label):
        image_string = tf.io.read_file(filename)
        image_decoded = tf.image.decode_jpeg(image_string)
        image_resized = tf.image.resize(image_decoded, [224, 224]) / 255.0
        return image_resized, label
    
    def EfficientNet_300x300_decode_and_resize(filename, label):
        image_string = tf.io.read_file(filename)
        image_decoded = tf.image.decode_jpeg(image_string)
        image_resized = tf.image.resize(image_decoded, [300, 300]) / 255.0
        return image_resized, label
    
    def get_model():
        # inputs = tf.keras.Input(shape=(100, 100, 3))
        # conv1 = tf.keras.layers.Conv2D(
        #     filters=32,  # 卷积核数量
        #     kernel_size=[3, 3],  # 卷积核大小
        #     padding='same',  # 领填充方式
        #     activation=tf.nn.relu,  # 激活函数
        #     kernel_regularizer=tf.keras.regularizers.l2(0.001)
        # )(inputs)
        # pool1 = tf.keras.layers.MaxPool2D(pool_size=[2, 2], strides=2)(conv1)
        # conv2 = tf.keras.layers.Conv2D(
        #     filters=64,  # 卷积核数量
        #     kernel_size=[3, 3],  # 卷积核大小
        #     padding='same',  # 领填充方式
        #     activation=tf.nn.relu,  # 激活函数
        #     kernel_regularizer=tf.keras.regularizers.l2(0.001)
        # )(pool1)
        # pool2 = tf.keras.layers.MaxPool2D(pool_size=[2, 2], strides=2)(conv2)
        # flatten = tf.keras.layers.Flatten()(pool2)
        # dense1 = tf.keras.layers.Dense(units=1024, activation=tf.nn.relu,
        #                                kernel_regularizer=tf.keras.regularizers.l2(0.001))(flatten)
        # # dropout = tf.keras.layers.Dropout(rate=0.4)(dense1)
        # dense2 = tf.keras.layers.Dense(units=10)(dense1)
        # outputs = tf.keras.layers.Softmax()(dense2)
        # model = tf.keras.Model(inputs=inputs, outputs=outputs)
        # ================================================================
    
        model = tf.keras.Sequential([
            tf.keras.layers.Conv2D(32, 3, activation='relu', input_shape=(100, 100, 3)),
            tf.keras.layers.MaxPooling2D(),
            tf.keras.layers.Conv2D(64, 3, activation='relu'),
            tf.keras.layers.MaxPooling2D(),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(1024, activation='relu'),
            tf.keras.layers.Dense(10, activation='softmax')
        ])
        return model
    
    def train_model_1(model):
        # lable = os.listdir(Dataset_dir)
        # print(lable) #['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        lable = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        for i in lable:
            temp_list = []
            for filename in os.listdir(Dataset_dir + i):
                temp_list.append(Dataset_dir + str(i) + "/" + filename)
            Dataset_dir_list.append(temp_list)
            lable_list.append(np.full(len(temp_list), int(i)))  # np.full(5,10)
    
        # print(len(lable_list)) #10
        # print(len(Dataset_dir_list)) #10
    
        # ================================================================
    
        for filename in os.listdir(Dataset_test_dir):
            Dataset_test_dir_list.append(Dataset_test_dir + filename)
    
        Dataset_dir_lists = tf.constant([f for filenames in Dataset_dir_list for f in filenames])
        lable_lists = tf.constant([int(s) for lables in lable_list for s in lables])
        Dataset_dir_lists_test = tf.constant([f for f in Dataset_test_dir_list])
        lable_lists_test = tf.constant([int(s) for s in lable])
        # ================================================================
    
        train_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists, lable_lists))
        train_dataset = train_dataset.map(decode_and_resize)
        train_dataset = train_dataset.batch(batch_size)
        train_dataset = train_dataset.shuffle(buffer_size=5000)
        train_dataset = train_dataset.prefetch(tf.data.experimental.AUTOTUNE)
    
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable_lists_test))
        test_dataset = test_dataset.map(decode_and_resize)
        test_dataset = test_dataset.batch(batch_size)
        # ================================================================
    
        #还要预先建立 ./graph/train/plugins/profile 的目录
        #要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        #http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True, write_images=True)
        #预先建立./ckpt目录
        check = tf.keras.callbacks.ModelCheckpoint('./ckpt/weights_{epoch:02d}-{val_loss:.2f}.h5',
                                                   monitor='val_loss',
                                                   save_best_only=True,
                                                   save_weights_only=False,
                                                   mode='auto',
                                                   period=1)
    
        # ================================================================
    
        model.compile(loss=losses, optimizer=optimizer, metrics=metrics)
        model.fit(train_dataset, validation_data=test_dataset, epochs=num_epochs, callbacks=[check, tensorboard]) #
    
        y_pred = model.predict(test_dataset)
        # 定义评估函数
        sparse_categorical_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
        # 定义测试数据集一共批次的大小
        sparse_categorical_accuracy.update_state(y_true=lable_lists_test, y_pred=y_pred)
        print("测试准确率: %f" % sparse_categorical_accuracy.result())
    
    def train_model_2(model):
        # lable = os.listdir(Dataset_dir)
        # print(lable) #['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        lable = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        for i in lable:
            temp_list = []
            for filename in os.listdir(Dataset_dir+i):
                temp_list.append(Dataset_dir+str(i)+"/"+filename)
            Dataset_dir_list.append(temp_list)
            lable_list.append(np.full(len(temp_list),int(i))) #np.full(5,10)
    
        print(len(lable_list)) #10
        print(len(Dataset_dir_list)) #10
        # ================================================================
    
        for i in lable:
            x_train, x_test, y_train, y_test = train_test_split(Dataset_dir_list[int(i)], lable_list[int(i)], test_size=0.1,
                                                                random_state=22)
            train_img_fileNames.append(x_train)
            train_img_lables.append(y_train)
            test_img_fileNames.append(x_test)
            test_img_lables.append(y_test)
    
        print(len(train_img_fileNames)) #10
        print(len(train_img_lables)) #10
        print(len(test_img_fileNames)) #10
        print(len(test_img_lables)) #10
    
        Dataset_dir_lists = [f for train_img in train_img_fileNames for f in train_img]
        lable_lists = [f for train_lable in train_img_lables for f in train_lable]
        Dataset_dir_lists_test = [f for test_img in test_img_fileNames for f in test_img]
        lable_lists_test = [f for test_lables in test_img_lables for f in test_lables]
    
        print(len(Dataset_dir_lists)) #1852
        print(len(lable_lists)) #1852
        print(len(Dataset_dir_lists_test)) #210
        print(len(lable_lists_test)) #210
        # ================================================================
    
        train_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists, lable_lists))
        train_dataset = train_dataset.map(decode_and_resize)
        train_dataset = train_dataset.batch(batch_size)
        train_dataset = train_dataset.shuffle(buffer_size=5000)
        train_dataset = train_dataset.prefetch(tf.data.experimental.AUTOTUNE)
    
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable_lists_test))
        test_dataset = test_dataset.map(decode_and_resize)
        test_dataset = test_dataset.batch(batch_size)
    
        # ================================================================
    
        #还要预先建立 ./graph/train/plugins/profile 的目录
        #要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        #http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True, write_images=True)
        #预先建立./ckpt目录
        check = tf.keras.callbacks.ModelCheckpoint('./ckpt/weights_{epoch:02d}-{val_loss:.2f}.h5',
                                                   monitor='val_loss',
                                                   save_best_only=True,
                                                   save_weights_only=False,
                                                   mode='auto',
                                                   period=1)
    
        # ================================================================
    
        model.compile(loss=losses, optimizer=optimizer, metrics=metrics)
        model.fit(train_dataset, validation_data=test_dataset, epochs=num_epochs, callbacks=[check, tensorboard]) #
    
        y_pred = model.predict(test_dataset)
        # 定义评估函数
        sparse_categorical_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
        # 定义测试数据集一共批次的大小
        sparse_categorical_accuracy.update_state(y_true=lable_lists_test, y_pred=y_pred)
        print("测试准确率: %f" % sparse_categorical_accuracy.result())
    
    def train_model_3(model):
        train_datagen = ImageDataGenerator(
            rescale=1. / 255,
            shear_range=0.2,
            zoom_range=0.2,
            horizontal_flip=True)
    
        train_generator = train_datagen.flow_from_directory(
            './Sign-Language-Digits-Dataset-master/Dataset/',
            target_size=(100, 100),
            batch_size=10,
            class_mode='binary')
    
        for filename in os.listdir(Dataset_test_dir):
            Dataset_test_dir_list.append(Dataset_test_dir + filename)
    
        Dataset_dir_lists_test = [f for f in Dataset_test_dir_list]
        lable = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
    
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable))
        test_dataset = test_dataset.map(decode_and_resize)
        test_dataset = test_dataset.batch(batch_size)
    
        # ================================================================
    
        #还要预先建立 ./graph/train/plugins/profile 的目录
        #要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        #http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True, write_images=True)
        modelckpt = tf.keras.callbacks.ModelCheckpoint('./ckpt/transfer_{epoch:02d}-{val_loss:.2f}.h5',
                                                       monitor='val_loss',
                                                       save_best_only=True,
                                                       save_weights_only=False,
                                                       mode='auto',
                                                       period=1)
    
        model.compile(loss=losses, optimizer=optimizer, metrics=metrics)
        model.fit_generator(
            train_generator,
            epochs=10,
            validation_steps=800, validation_data=train_generator,
            callbacks=[modelckpt, tensorboard])
    
        y_pred = model.predict(test_dataset)
        # 定义评估函数
        sparse_categorical_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
        # 定义测试数据集一共批次的大小
        sparse_categorical_accuracy.update_state(y_true=lable, y_pred=y_pred)
        print("测试准确率: %f" % sparse_categorical_accuracy.result())
    
    def train_model_4():
        train_datagen = ImageDataGenerator(
            rescale=1. / 255,
            shear_range=0.2,
            zoom_range=0.2,
            horizontal_flip=True)
    
        train_generator = train_datagen.flow_from_directory(
            './Sign-Language-Digits-Dataset-master/Dataset/',
            target_size=(224, 224),
            batch_size=10,
            class_mode='binary')
    
        for filename in os.listdir(Dataset_test_dir):
            Dataset_test_dir_list.append(Dataset_test_dir + filename)
    
        Dataset_dir_lists_test = [f for f in Dataset_test_dir_list]
        lable = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
    
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable))
        test_dataset = test_dataset.map(VGG_224x244_decode_and_resize)
        test_dataset = test_dataset.batch(10)
    
        # ================================================================
    
        base_model = VGG16(weights='imagenet', include_top=False)
        #修改VGG的模型，在VGG.outputs[0]输出的为[None, ?, ?, 512]--->全局平均池化-->两个全连接层1024, 10
        # 1、获取VGG模型的输出，不包含原有模型的top结构
        x = base_model.outputs[0]
    
        # 2、在VGG的输出之后定义自己的模型
        x = tf.keras.layers.GlobalAveragePooling2D()(x)
    
        # 两个全连接层
        x = tf.keras.layers.Dense(1024, activation=tf.nn.relu)(x)
        y_predict = tf.keras.layers.Dense(10, activation=tf.nn.softmax)(x)
    
        #冻结VGG的前面卷积结构，不参与训练
        # 循环获取base_model当中的层
        for layer in base_model.layers:
            layer.trainable = False
    
        # 3、使用Model封装新的模型返回
        transfer_model = tf.keras.models.Model(inputs=base_model.inputs, outputs=y_predict)
    
        # ================================================================
    
        #还要预先建立 ./graph/train/plugins/profile 的目录
        #要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        #http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True, write_images=True)
        modelckpt = tf.keras.callbacks.ModelCheckpoint('./ckpt/transfer_{epoch:02d}-{val_loss:.2f}.h5',
                                                       monitor='val_loss',
                                                       save_best_only=True,
                                                       save_weights_only=False,
                                                       mode='auto',
                                                       period=1)
    
        transfer_model.compile(loss=losses, optimizer=optimizer, metrics=metrics)
        transfer_model.summary()
        transfer_model.fit_generator(
            train_generator,
            steps_per_epoch=207,
            epochs=2,
            validation_data=train_generator,
            callbacks=[modelckpt, tensorboard])
    
        y_pred = transfer_model.predict(test_dataset)
        # 定义评估函数
        sparse_categorical_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
        # 定义测试数据集一共批次的大小
        sparse_categorical_accuracy.update_state(y_true=lable, y_pred=y_pred)
        print(sparse_categorical_accuracy.result())
    
    def train_model_5():
        # lable = os.listdir(Dataset_dir)
        # print(lable) #['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        lable = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
    
        for i in lable:
            temp_list = []
            for filename in os.listdir(Dataset_dir+i):
                temp_list.append(Dataset_dir+str(i)+"/"+filename)
            Dataset_dir_list.append(temp_list)
            lable_list.append(np.full(len(temp_list),int(i))) #np.full(5,10)
    
        print(len(lable_list)) #10
        print(len(Dataset_dir_list)) #10
    
        # ================================================================
    
        for filename in os.listdir(Dataset_test_dir):
            Dataset_test_dir_list.append(Dataset_test_dir+filename)
    
        train_dataset_len = len(Dataset_dir_list)
        print(train_dataset_len)
    
        Dataset_dir_lists = tf.constant([f for filenames in Dataset_dir_list for f in filenames])
        lable_lists = tf.constant([int(s) for lables in lable_list for s in lables])
        Dataset_dir_lists_test = tf.constant([f for f in Dataset_test_dir_list])
        lable_lists_test = tf.constant([int(s) for s in lable])
    
        # ================================================================
        """
        1.使用repeat() 解决如下报错：
            WARNING:tensorflow:Your input ran out of data; interrupting training. 
            Make sure that your dataset or generator can generate at least `steps_per_epoch * epochs` 
            batches (in this case, 414 batches). You may need to use the repeat() function 
            when building your dataset.
            警告：tensorflow：输入的数据用完；中断训练。请确保您的数据集或生成器至少可以生成
            “每个steps_per_epoch*epoch”批（在本例中为414 batches）。在构建数据集时，可能需要使用repeat()函数。
        
        2.例子：batch、repeat、steps_per_epoch、epochs的使用
                1.先用 batch(批次大小) 然后才用 repeat(重复次数)，
                  比如：repeat(2)重复数据集2次，即复制一份数据集，最终即有两份数据集。
                2.steps_per_epoch 即表示 一个epoch 里面遍历批量数据的次数，即遍历多少个批量数据完成一个epoch。
                3.应保证要输入到模型的数据集(包括训练集/验证集)的批量个数都均为steps_per_epoch*epoch。
     
        3.使用顺序：from_tensor_slices -> map -> shuffle -> batch -> repeat -> prefetch
        """
        train_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists, lable_lists))
        train_dataset = train_dataset.map(EfficientNet_300x300_decode_and_resize, num_parallel_calls=tf.data.experimental.AUTOTUNE)
        train_dataset = train_dataset.shuffle(buffer_size=5000)
        train_dataset = train_dataset.batch(batch_size)
        epochs_repeats = 10
        train_dataset = train_dataset.repeat(epochs_repeats)
        train_dataset = train_dataset.prefetch(tf.data.experimental.AUTOTUNE)
    
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable_lists_test))
        test_dataset = test_dataset.map(EfficientNet_300x300_decode_and_resize, num_parallel_calls=tf.data.experimental.AUTOTUNE)
        test_dataset = test_dataset.batch(batch_size)
        test_dataset = test_dataset.repeat(epochs_repeats)
    
        # ================================================================
    
        #还要预先建立 ./graph/train/plugins/profile 的目录
        #要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        #http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True, write_images=True)
        #预先建立./ckpt目录
        check = tf.keras.callbacks.ModelCheckpoint('./ckpt/weights_{epoch:02d}-{val_loss:.2f}.h5',
                                                   monitor='val_loss',
                                                   save_best_only=True,
                                                   save_weights_only=False,
                                                   mode='auto',
                                                   period=1)
        # ================================================================
    
        num_classes = 10
    
        base_model = EfficientNet.EfficientNetB3(include_top=False, input_shape=(300, 300, 3), classes=num_classes)
        x = base_model.output
        x = GlobalAveragePooling2D(name='avg_pool')(x)
        predictions = Dense(num_classes, activation='softmax')(x)
    
        model = Model(inputs=base_model.input, outputs=predictions)
        model.compile(loss=losses, optimizer=optimizer, metrics=metrics)
        model.summary()
    
        model.fit(
            train_dataset,
            # 一个epoch需要多少步
            #Train on 207 steps, validate on 1 steps
            steps_per_epoch=207,
            epochs=epochs_repeats,
            callbacks=[check, tensorboard],
            validation_data=test_dataset
        )
        return None
    
    def predict_model_5():
        for filename in os.listdir(Dataset_test_dir):
            Dataset_test_dir_list.append(Dataset_test_dir+filename)
        lable = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
        lable_list = []
        epochs_repeats = 10
        for _ in range(epochs_repeats):
            lable_list.extend(lable)
        Dataset_dir_lists_test = tf.constant([f for f in Dataset_test_dir_list])
        lable_lists_test = tf.constant([int(s) for s in lable])
        test_dataset = tf.data.Dataset.from_tensor_slices((Dataset_dir_lists_test, lable_lists_test))
        test_dataset = test_dataset.map(EfficientNet_300x300_decode_and_resize, num_parallel_calls=tf.data.experimental.AUTOTUNE)
        test_dataset = test_dataset.batch(batch_size)
        test_dataset = test_dataset.repeat(epochs_repeats)
    
        num_classes = 10
        base_model = EfficientNet.EfficientNetB3(include_top=False, input_shape=(300, 300, 3), classes=num_classes)
        x = base_model.output
        x = GlobalAveragePooling2D(name='avg_pool')(x)
        predictions = Dense(num_classes, activation='softmax')(x)
        model = Model(inputs=base_model.input, outputs=predictions)
        model.load_weights("./ckpt/weights_09-0.00.h5")
    
        y_predict = model.predict(test_dataset)
        sparse_categorical_accuracy = tf.keras.metrics.SparseCategoricalAccuracy()
        sparse_categorical_accuracy.update_state(y_true=lable_list, y_pred=y_predict)
        print("测试集的准确率为: %f" % sparse_categorical_accuracy.result())
        return
    
    if __name__ == '__main__':
        # model = get_model()
    
        #常规的用 examples文件夹中的图片来测试
        # train_model_1(model)
    
        #使用 sklearn的 train_test_split 来进行数据集分割
        # train_model_2(model)
    
        #使用 ImageDataGenerator 来直接读取各个类别文件中的图片
        # train_model_3(model)
    
        #使用 vgg16 进行迁移学习
        # train_model_4()
    
        """
        1.tf.nn.swish(x) 等同于把 x * tf.sigmoid(beta * x) 封装了。
          如果使用了tf.nn.swish(x) 则需要同时使用tf.compat.v1.disable_eager_execution()。
          如果使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)的话，则可以不使用tf.compat.v1.disable_eager_execution()。
        2.但注意此处可能环境问题使用tf.nn.swish(x)的话会报错，所以此处使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)
          报错信息如下：
            tensorflow/core/grappler/utils/graph_view.cc:830] No registered 'swish_f32' OpKernel for GPU devices compatible with node
           {
       {node swish_75/swish_f32}}  Registered:  <no registered kernels>
        3.使用 inputs * tf.math.sigmoid(inputs) 之后 就不用  tf.compat.v1.disable_eager_execution() 了，
          TensorBoard  那就也可以正常使用了，因为此处同时使用 tf.compat.v1.disable_eager_execution() 和 TensorBoard
          的时候，TensorBoard会报错：tensorflow.python.eager.core._FallbackException: This function does not handle the case of the path where all inputs are not already EagerTensors.
          
        """
        # EfficientNet 必须注意关闭默认的eager模式
        # tf.compat.v1.disable_eager_execution()
    
        #使用 EfficientNet模型 进行迁移学习
        train_model_5()
        # 使用 EfficientNet模型 进行 预测
        # predict_model_5()

--------------------

# **使用EfficientNet模型迁移 第二个版本**  #

from day05.homework.efficientnet import model as EfficientNet
    import tensorflow as tf
    from tensorflow.keras.layers import Dense, GlobalAveragePooling2D
    from tensorflow.keras.models import Model
    from day05.homework.data_gen import data_from_sequence
    from day05.homework.utils.lr_scheduler import WarmUpCosineDecayScheduler
    
    
    num_epochs = 10
    batch_size = 10
    learning_rate = 0.001
    Dataset_dir = './Sign-Language-Digits-Dataset-master/Dataset/'
    Dataset_test_dir = './Sign-Language-Digits-Dataset-master/Examples/'
    num_classes = 10
    input_size = 300
    
    Dataset_dir_list = []
    Dataset_test_dir_list = []
    lable_list = []
    train_img_fileNames = []
    train_img_lables = []
    test_img_fileNames = []
    test_img_lables = []
    lr=1e-3
    optimizer = tf.keras.optimizers.Adam(lr) #0.001
    sparse_categorical_crossentropy = tf.keras.losses.sparse_categorical_crossentropy
    #['accuracy']/['acc'] 可以代替 [tf.keras.metrics.sparse_categorical_accuracy]，
    # 因为['accuracy']/['acc']底层都会默认适配为[tf.keras.metrics.sparse_categorical_accuracy]
    sparse_categorical_accuracy = [tf.keras.metrics.sparse_categorical_accuracy]
    
    
    """
    metrics 是 sparse_categorical_crossentropy，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_sparse_categorical_crossentropy
    metrics 是 acc，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_acc
    metrics 是 accuracy，则ModelCheckpoint中的第一个路劲字符串信息中和monitor参数中都需要用val_accuracy
    #['accuracy']/['acc'] 可以代替 [tf.keras.metrics.sparse_categorical_accuracy]，
    #因为['accuracy']/['acc']底层都会默认适配为[tf.keras.metrics.sparse_categorical_accuracy]
    """
    
    
    def train_model():
        # 1、读取sequence数据
        train_sequence, validation_sequence = data_from_sequence(Dataset_dir, batch_size, num_classes, input_size)
    
        # ================================================================
    
        # 还要预先建立 ./graph/train/plugins/profile 的目录
        # 要在graph的同级目录下 使用命令 tensorboard --logdir=graph 或者 tensorboard --logdir='./graph'
        # http://localhost:6006/ 访问
        tensorboard = tf.keras.callbacks.TensorBoard(log_dir='./graph', histogram_freq=1, write_graph=True,
                                                     write_images=True)
        # 预先建立./ckpt目录
        check = tf.keras.callbacks.ModelCheckpoint('./ckpt/weights_{epoch:02d}-{val_loss:.2f}.h5',
                                                   monitor='val_loss',
                                                   save_best_only=True,
                                                   save_weights_only=False,
                                                   mode='auto',
                                                   period=1)
        # ================================================================
    
        base_model = EfficientNet.EfficientNetB3(include_top=False, input_shape=(300, 300, 3), classes=num_classes)
        x = base_model.output
        x = GlobalAveragePooling2D(name='avg_pool')(x)
        predictions = Dense(num_classes, activation='softmax')(x)
    
        model = Model(inputs=base_model.input, outputs=predictions)
    
        # model.compile(loss=sparse_categorical_crossentropy, optimizer=optimizer, metrics=sparse_categorical_accuracy)
        model.compile(loss='categorical_crossentropy', optimizer=optimizer, metrics= ['accuracy'])
    
        model.summary()
        # ================================================================
    
        # 余弦退回warmup
        # 得到总样本数
        sample_count = len(train_sequence)
    
        # 第二阶段学习率以及总步数
        learning_rate_base = lr
        max_epochs = 10
        total_steps = int(max_epochs * sample_count) / batch_size
        # 计算第一阶段的步数需要多少 warmup_steps
        warmup_epoch = 5
        warmup_steps = int(warmup_epoch * sample_count) / batch_size
    
        print("sample_count:",sample_count) #185
        print("total_steps:",total_steps) #185.0
        print("warmup_steps:",warmup_steps) #92.5
    
        warm_lr = WarmUpCosineDecayScheduler(learning_rate_base=learning_rate_base,
                                             total_steps=total_steps,
                                             warmup_learning_rate=0,
                                             warmup_steps=warmup_steps,
                                             hold_base_rate_steps=0)
        # ================================================================
        """
        要求训练样本数/标签数和验证样本/标签数数都必须是可以整除batch_size的，
        否则在mixup中X = X1 * X_l + X2 * (1 - X_l)代码再进行矩阵广播运算的时候会报错，因为形状不一致，
        报错信息：ValueError: operands could not be broadcast together with shapes (不满batch_size大小,300,300,3) (batch_size,1,1,1)。
        正因为如此，所以要求训练样本数和验证样本数都必须是可以整除batch_size的，因为在一个epoch中遍历到最后一个step时，
        剩余的训练样本数不足以整除batch size，那么就会报错。
        """
        """
        因为手势识别的数据集过小，因此不要使用warm_lr(warmup以及余弦退火学习)、数据增强(随机擦处/ImageDataGenerator)、mixup，
        这样loss才能快速降低，准确率才能上升。
        """
        # 4、训练步骤
        model.fit_generator(
            train_sequence,
            steps_per_epoch=int(sample_count / batch_size),  # 一个epoch需要多少步 ， 1epoch sample_out 140000多样本， 140000 / 16 = 步数
            epochs=max_epochs,
            verbose=1,
            callbacks=[check, tensorboard],
            # callbacks=[check, tensorboard, warm_lr],
            validation_data=validation_sequence,
            # max_queue_size=10,
            # workers=int(multiprocessing.cpu_count() * 0.7),
            # use_multiprocessing=True,
            # shuffle=True
        )
        return None
    
    
    
    if __name__ == '__main__':
        """
        1.tf.nn.swish(x) 等同于把 x * tf.sigmoid(beta * x) 封装了。
          如果使用了tf.nn.swish(x) 则需要同时使用tf.compat.v1.disable_eager_execution()。
          如果使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)的话，则可以不使用tf.compat.v1.disable_eager_execution()。
        2.但注意此处可能环境问题使用tf.nn.swish(x)的话会报错，所以此处使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)
          报错信息如下：
            tensorflow/core/grappler/utils/graph_view.cc:830] No registered 'swish_f32' OpKernel for GPU devices compatible with node
           {
       {node swish_75/swish_f32}}  Registered:  <no registered kernels>
        3.使用 inputs * tf.math.sigmoid(inputs) 之后 就不用  tf.compat.v1.disable_eager_execution() 了，
          TensorBoard  那就也可以正常使用了，因为此处同时使用 tf.compat.v1.disable_eager_execution() 和 TensorBoard
          的时候，TensorBoard会报错：tensorflow.python.eager.core._FallbackException: This function does not handle the case of the path where all inputs are not already EagerTensors.
    
        """
        # EfficientNet 必须注意关闭默认的eager模式
        # tf.compat.v1.disable_eager_execution()
    
        # 使用 EfficientNet模型 进行迁移学习
        train_model()

--------------------

# data\_gen文件夹 processing\_data.py #

import math
    import os
    import random
    import numpy as np
    from PIL import Image
    from tensorflow.keras.preprocessing.image import ImageDataGenerator
    from tensorflow.keras.utils import to_categorical, Sequence
    from sklearn.model_selection import train_test_split
    
    from data_gen.random_eraser import get_random_eraser
    
    
    class GarbageDataSequence(Sequence):
        """垃圾分类数据流，每次batch返回batch_size大小数据
        model.fit_generator使用
        """
        def __init__(self, img_paths, labels, batch_size, img_size, use_aug):
            print("20:",np.array(img_paths).shape) #batch_size
            print("21:",np.array(labels).shape) #[batch_size, 10]
    
            # 1、获取训练特征与目标值的合并结果 np.hstack( [batch_size, 1],  [batch_size, 40]) 结果为 [batch_size, 41]
            self.x_y = np.hstack((np.array(img_paths).reshape(len(img_paths), 1), np.array(labels)))
            print("22:",self.x_y.shape)
            print("23:",self.x_y)
    
    
            self.batch_size = batch_size
            self.img_size = img_size  # (300, 300)
            self.use_aug = use_aug
            self.alpha = 0.2
            # 随机擦出方法
            self.eraser = get_random_eraser(s_h=0.3, pixel_level=True)
    
        def __len__(self):
            return math.ceil(len(self.x_y) / self.batch_size)
    
        @staticmethod
        def center_img(img, size=None, fill_value=255):
            """改变图片尺寸到300x300，并且做填充使得图像处于中间位置
            """
            h, w = img.shape[:2]
            if size is None:
                size = max(h, w)
            shape = (size, size) + img.shape[2:]
            background = np.full(shape, fill_value, np.uint8)
            center_x = (size - w) // 2
            center_y = (size - h) // 2
            background[center_y:center_y + h, center_x:center_x + w] = img
            return background
    
        def preprocess_img(self, img_path):
            """处理每张图片，大小， 数据增强
            :param img_path:
            :return:
            """
            # 1、读取图片对应内容，做形状，内容处理, (h, w)
            img = Image.open(img_path)
            # [180, 200, 3]
            scale = self.img_size[0] / max(img.size[:2])
            img = img.resize((int(img.size[0] * scale), int(img.size[1] * scale)))
            img = img.convert('RGB')
            img = np.array(img)
    
            # 2、数据增强：如果是训练集进行数据增强操作
            # if self.use_aug:
            #
            #     # 1、随机擦处
            #     img = self.eraser(img)
            #
            #     # 2、翻转
            #     datagen = ImageDataGenerator(
            #         width_shift_range=0.05,
            #         height_shift_range=0.05,
            #         horizontal_flip=True,
            #         vertical_flip=True,
            #     )
            #     img = datagen.random_transform(img)
    
            # 4、处理一下形状 【300， 300， 3】
            # 改变到[300, 300] 建议不要进行裁剪操作，变形操作，保留数据增强之后的效果，填充到300x300
            img = self.center_img(img, self.img_size[0])
            return img
    
        def __getitem__(self, idx):
    
            # 1、获取当前批次idx对应的特征值和目标值
            batch_x = self.x_y[idx * self.batch_size: self.batch_size * (idx + 1), 0]
            batch_y = self.x_y[idx * self.batch_size: self.batch_size * (idx + 1), 1:]
            # print("batch_x:",batch_x)
            # print("batch_y:",batch_y)
    
            batch_x = np.array([self.preprocess_img(img_path) for img_path in batch_x])
            batch_y = np.array(batch_y).astype(np.float32)
    
            # 2、mixup
            # batch_x, batch_y = self.mixup(batch_x, batch_y)
    
            # 3、归一化处理
            batch_x = self.preprocess_input(batch_x)
    
            return batch_x, batch_y
    
        def on_epoch_end(self):
            np.random.shuffle(self.x_y)
    
        def mixup(self, batch_x, batch_y):
            """
            数据混合mixup
            :param batch_x: 要mixup的batch_X
            :param batch_y: 要mixup的batch_y
            :return: mixup后的数据
            """
            size = self.batch_size
            l = np.random.beta(self.alpha, self.alpha, size)
    
            X_l = l.reshape(size, 1, 1, 1)
            y_l = l.reshape(size, 1)
    
            X1 = batch_x
            Y1 = batch_y
            X2 = batch_x[::-1]
            Y2 = batch_y[::-1]
    
            X = X1 * X_l + X2 * (1 - X_l)
            Y = Y1 * y_l + Y2 * (1 - y_l)
    
            return X, Y
    
        def preprocess_input(self, x):
            """归一化处理样本特征值
            :param x:
            :return:
            """
            assert x.ndim in (3, 4)
            assert x.shape[-1] == 3
    
            MEAN_RGB = [0.485 * 255, 0.456 * 255, 0.406 * 255]
            STDDEV_RGB = [0.229 * 255, 0.224 * 255, 0.225 * 255]
    
            x = x - np.array(MEAN_RGB)
            x = x / np.array(STDDEV_RGB)
    
            return x
    
    
    def smooth_labels(y, smooth_factor=0.1):
    
        assert len(y.shape) == 2
        if 0 <= smooth_factor <= 1:
            y *= 1 - smooth_factor
            y += smooth_factor / y.shape[1]
        else:
            raise Exception(
                'Invalid label smoothing factor: ' + str(smooth_factor))
        return y
    
    
    def data_from_sequence(train_data_dir, batch_size, num_classes, input_size):
        """读取本地数据到sequence
        :param train_data_dir: 训练数据目录
        :param batch_size: 批次大小
        :param num_classes: 总类别书40
        :param input_size: 输入图片大小(300, 300)
        :return:
        """
        # 1、读取txt文件，打乱文件顺序, .jpg, .txt
        # label_files = [os.path.join(train_data_dir, filename) for filename
        #                in os.listdir(train_data_dir) if filename.endswith('.txt')]
        # random.shuffle(label_files)
    
        # # 2、解析txt文件当中 特征值以及目标值（标签）
        # img_paths = []
        # labels = []
        #
        # for index, file_path in enumerate(label_files):
        #     with open(file_path, 'r') as f:
        #         line = f.readline()
        #
        #     line_split = line.strip().split(', ')
        #     # line '*.jpg, 0'
        #     if len(line_split) != 2:
        #         print("% 文件格式出错", (file_path))
        #         continue
        #
        #     img_name = line_split[0]
        #     label = int(line_split[1])
        #
        #     # 最后保存到所有的列表当中
        #     img_paths.append(os.path.join(train_data_dir, img_name))
        #     labels.append(label)
        # print(img_paths, labels)
    
    
        print("batch_size:",batch_size)
        print("num_classes:",num_classes)
        print("input_size:",input_size)
    
    
        lable = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9']
        lable_list = []
        Dataset_dir_list = []
    
        for i in lable:
            temp_list = []
            for filename in os.listdir(train_data_dir + i):
                temp_list.append(train_data_dir + str(i) + "/" + filename)
            Dataset_dir_list.append(temp_list)
            lable_list.append(np.full(len(temp_list), int(i)))  # np.full(5,10)
    
        print(len(lable_list))  # 10
        print(len(Dataset_dir_list))  # 10
    
        train_img_fileNames = []
        train_img_lables = []
        test_img_fileNames = []
        test_img_lables = []
    
        for i in lable:
            # 3、目标标签类别ont_hot编码转换， 平滑处理
            labels = to_categorical(lable_list[int(i)], num_classes)
            print("labels:", labels.shape)
    
            labels = smooth_labels(labels)
            print("smooth_labels:", labels.shape)
    
            """
            必须使用expand_dims在第一维，即每个路径字符串都封装多一层作为一个数组，那么第一维每个数组中都只有一个路径字符串。
            这样做才能配合二维的labels在train_test_split函数中进行切割使用
            """
            img = np.expand_dims(Dataset_dir_list[int(i)], axis=1)
            print("img.shape:",img.shape)
            # print("img:",img)
    
            x_train, x_test, y_train, y_test = train_test_split(img, labels, test_size=0.1, random_state=22)
    
            # x_train, x_test, y_train, y_test = train_test_split(Dataset_dir_list[int(i)], lable_list[int(i)], test_size=0.1, random_state=22)
    
            train_img_fileNames.append(x_train)
            train_img_lables.append(y_train)
            test_img_fileNames.append(x_test)
            test_img_lables.append(y_test)
    
        print("1:",len(train_img_fileNames)) #10
        print("2:",len(train_img_lables)) #10
        print("3:",len(test_img_fileNames)) #10
        print("4:",len(test_img_lables)) #10
    
        # print("5:",train_img_fileNames[0]) # 整个是 三维数组，取出来的[0] 是二维数组
        # print("6:",train_img_lables[0]) #整个是 三维数组，取出来的[0] 是二维数组
        # print("7:",test_img_fileNames[0]) #整个是 三维数组，取出来的[0] 是二维数组
        # print("8:",test_img_lables[0]) #整个是 三维数组，取出来的[0] 是二维数组
    
        Dataset_dir_lists = [f1 for train_img in train_img_fileNames for f in train_img for f1 in f]
        lable_lists = [f for train_lable in train_img_lables for f in train_lable]
        Dataset_dir_lists_test = [f1 for test_img in test_img_fileNames for f in test_img for f1 in f]
        lable_lists_test = [f for test_lables in test_img_lables for f in test_lables]
    
        print("9:",len(Dataset_dir_lists)) #1852
        print("10:",len(lable_lists))#1852
        print("11:",len(Dataset_dir_lists_test))#210
        print("12:",len(lable_lists_test)) #210
    
        print("13:",Dataset_dir_lists[0])
        print("14:",lable_lists[0])
        print("15:",Dataset_dir_lists_test[0])
        print("16:",lable_lists_test[0])
    
        # # 3、目标标签类别ont_hot编码转换， 平滑处理
        # labels = to_categorical(lable_list, num_classes)
        # labels = smooth_labels(labels)
    
        # 分割训练集合验证集合
        # train_img_paths, validation_img_paths, train_labels, validation_labels \
        #     = train_test_split(Dataset_dir_list, labels, test_size=0.15, random_state=0)
        # print(validation_img_paths)
        # print(train_labels, validation_labels)
        print("总共样本数： %d , 训练样本数： %d, 验证样本数： %d" % (len(Dataset_dir_lists)+len(Dataset_dir_lists_test),
                                                    len(Dataset_dir_lists), len(Dataset_dir_lists_test)))
    
        """
        要求训练样本数/标签数和验证样本/标签数数都必须是可以整除batch_size的，
        否则在mixup中X = X1 * X_l + X2 * (1 - X_l)代码再进行矩阵广播运算的时候会报错，因为形状不一致，
        报错信息：ValueError: operands could not be broadcast together with shapes (不满batch_size大小,300,300,3) (batch_size,1,1,1)。
        正因为如此，所以要求训练样本数和验证样本数都必须是可以整除batch_size的，因为在一个epoch中遍历到最后一个step时，
        剩余的训练样本数不足以整除batch size，那么就会报错。
        """
        i = -2
        j = -2
        while 1:
            if (len(Dataset_dir_lists) % batch_size == 0):
                break
            else:
                Dataset_dir_lists = Dataset_dir_lists[:i]
                lable_lists = lable_lists[:i]
                i -= 1
    
        while 1:
            if (len(Dataset_dir_lists_test) % batch_size == 0):
                break
            else:
                Dataset_dir_lists_test = Dataset_dir_lists[:j]
                lable_lists_test = lable_lists_test[:j]
                j -= 1
    
        print("处理后的总共样本数： %d , 处理后的训练样本数： %d, 处理后的验证样本数： %d, 处理后的训练样本标签数： %d, 处理后的验证样本标签数： %d" %
              (len(Dataset_dir_lists)+len(Dataset_dir_lists_test), len(Dataset_dir_lists), len(Dataset_dir_lists_test)
               , len(lable_lists), len(lable_lists_test)
               ))
    
        # 4、Sequence调用测试
        train_sequence = GarbageDataSequence(Dataset_dir_lists, lable_lists, batch_size, [input_size, input_size], use_aug=True)
        validation_sequence = GarbageDataSequence(Dataset_dir_lists_test, lable_lists_test, batch_size, [input_size, input_size], use_aug=False)
    
        return train_sequence, validation_sequence
    
    
    # if __name__ == '__main__':
    #     train_data_dir = "../data/garbage_classify/train_data"
    #     batch_size = 32
    #
    #     train_sequence, validation_sequence = data_from_sequence(train_data_dir, batch_size, num_classes=40, input_size=300)
    #
    #     for i in range(100):
    #         print("第 %d 批次数据" % i)
    #         batch_x, batch_y = train_sequence.__getitem__(i)
    #         print(batch_x, batch_y)

--------------------

# data\_gen文件夹 random\_eraser.py #

import numpy as np
    import tensorflow as tf
    
    
    def get_random_eraser(p=0.5, s_l=0.02, s_h=0.4, r_1=0.3, r_2=1/0.3, v_l=0, v_h=255, pixel_level=False):
        def eraser(input_img):
            img_h, img_w, img_c = input_img.shape
            p_1 = np.random.rand()
    
            if p_1 > p:
                return input_img
    
            while True:
                s = np.random.uniform(s_l, s_h) * img_h * img_w
                r = np.random.uniform(r_1, r_2)
                w = int(np.sqrt(s / r))
                h = int(np.sqrt(s * r))
                left = np.random.randint(0, img_w)
                top = np.random.randint(0, img_h)
    
                if left + w <= img_w and top + h <= img_h:
                    break
    
            if pixel_level:
                c = np.random.uniform(v_l, v_h, (h, w, img_c))
            else:
                c = np.random.uniform(v_l, v_h)
    
            input_img[top:top + h, left:left + w, :] = c
    
            return input_img
    
        return eraser

--------------------

# **utils文件夹 lr\_scheduler.py** #

import numpy as np
    import tensorflow as tf
    from tensorflow.keras import backend as K
    from tensorflow.keras.models import Sequential
    from tensorflow.keras.layers import Dense
    import os
    os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"
    
    
    def cosine_decay_with_warmup(global_step,
                                 learning_rate_base,
                                 total_steps,
                                 warmup_learning_rate=0.0,
                                 warmup_steps=0,
                                 hold_base_rate_steps=0):
        """
        每批次带有warmup余弦退火学习率计算
        :param global_step: 当前到达的步数
        :param learning_rate_base: warmup之后的基础学习率
        :param total_steps: 总需要批次数
        :param warmup_learning_rate: warmup开始的学习率
        :param warmup_steps:warmup学习率 步数
        :param hold_base_rate_steps: 预留总步数和warmup步数间隔
        :return:
        """
        if total_steps < warmup_steps:
            raise ValueError("总步数要大于wamup步数")
        # 1、余弦退火学习率计算
        learning_rate = 0.5 * learning_rate_base * (1 + np.cos(
            np.pi * (global_step - warmup_steps - hold_base_rate_steps) / float(total_steps - warmup_steps - hold_base_rate_steps)
        ))
    
        # 2、warmup之后的学习率计算
        # 预留步数阶段
        # 如果预留大于0，判断目前步数是否 > warmup步数+预留步数，是的话返回刚才上面计算的学习率，不是的话使用warmup之后的基础学习率
        learning_rate = np.where(global_step > warmup_steps + hold_base_rate_steps, learning_rate, learning_rate_base)
        # 3、warmup学习率计算，并判断大小
        # 第一个阶段的学习率计算
        if warmup_steps > 0:
            if learning_rate_base < warmup_learning_rate:
                raise ValueError("第二阶段学习率要大于第一阶段学习率")
    
            slope = (learning_rate_base - warmup_learning_rate) / warmup_steps
            warmup_rate = slope * global_step + warmup_learning_rate
            learning_rate = np.where(global_step < warmup_steps, warmup_rate, learning_rate)
        # 4、如果最后当前到达的步数大于总步数，则归0，否则返回当前的计算出来的学习率（可能是warmup学习率也可能是余弦衰减结果）
    
        return np.where(global_step > total_steps, 0.0, learning_rate)
    
    
    class WarmUpCosineDecayScheduler(tf.keras.callbacks.Callback):
        """带有warnup的余弦退火学习率实现
        """
        def __init__(self,
                     learning_rate_base,
                     total_steps,
                     global_step_init=0,
                     warmup_learning_rate=0.0,
                     warmup_steps=0,
                     hold_base_rate_steps=0,
                     verbose=0):
    
            super(WarmUpCosineDecayScheduler, self).__init__()
            self.learning_rate_base = learning_rate_base
            self.total_steps = total_steps
            self.global_step = global_step_init
            self.warmup_learning_rate = warmup_learning_rate
            self.warmup_steps = warmup_steps
            self.hold_base_rate_steps = hold_base_rate_steps
            # 是否在每次训练结束打印学习率
            self.verbose = verbose
            # 记录所有批次下来的每次准确的学习率，可以用于打印显示
            self.learning_rates = []
    
        def on_batch_end(self, batch, logs=None):
    
            # 记录当前训练到走到第几步数
            self.global_step = self.global_step + 1
            # 记录下所有每次的学习到列表，要统计画图可以使用
    
            lr = K.get_value(self.model.optimizer.lr)
            self.learning_rates.append(lr)
    
        def on_batch_begin(self, batch, logs=None):
    
            # 计算这批次开始的学习率 lr
            lr = cosine_decay_with_warmup(global_step=self.global_step,
                                          learning_rate_base=self.learning_rate_base,
                                          total_steps=self.total_steps,
                                          warmup_learning_rate=self.warmup_learning_rate,
                                          warmup_steps=self.warmup_steps,
                                          hold_base_rate_steps=self.hold_base_rate_steps)
    
            # 设置模型的学习率为lr
            K.set_value(self.model.optimizer.lr, lr)
    
            if self.verbose > 0:
                print('\n批次数 %05d: 设置学习率为'
                      ' %s.' % (self.global_step + 1, lr))
    
    
    if __name__ == '__main__':
        # 1、创建模型
        model = Sequential()
        model.add(Dense(32, activation='relu', input_dim=100))
        model.add(Dense(10, activation='softmax'))
        model.compile(optimizer='rmsprop',
                      loss='categorical_crossentropy',
                      metrics=['accuracy'])
    
        # 2、参数设置
        sample_count = 1000  # 样本数
        epochs = 4  # 总迭代次数
        warmup_epoch = 3  # warmup 迭代次数
        batch_size = 16  # 批次大小
        learning_rate_base = 0.0001  # warmup后的初始学习率
        total_steps = int(epochs * sample_count / batch_size)  # 总迭代批次步数
        warmup_steps = int(warmup_epoch * sample_count / batch_size)  # warmup总批次数
    
        # 3、创建测试数据
        data = np.random.random((sample_count, 100))
        labels = np.random.randint(10, size=(sample_count, 1))
        # 转换目标类别
        one_hot_labels = tf.keras.utils.to_categorical(labels, num_classes=10)
    
        # 5、创建余弦warmup调度器
        warm_up_lr = WarmUpCosineDecayScheduler(learning_rate_base=learning_rate_base,
                                                total_steps=total_steps,
                                                warmup_learning_rate=4e-06,  # warmup开始学习率
                                                warmup_steps=warmup_steps,
                                                hold_base_rate_steps=0,
                                                )
    
        # 训练模型
        model.fit(data, one_hot_labels, epochs=epochs, batch_size=batch_size, verbose=0, callbacks=[warm_up_lr])
    
        print(warm_up_lr.learning_rates)

#  #

--------------------

# **EfficientNet模型B0~B7 源码** #

# **initializers.py** #

import numpy as np
    import tensorflow as tf
    import keras.backend as K
    
    from keras.initializers import Initializer
    from keras.utils.generic_utils import get_custom_objects
    
    
    class EfficientConv2DKernelInitializer(Initializer):
        """Initialization for convolutional kernels.
        The main difference with tf.variance_scaling_initializer is that
        tf.variance_scaling_initializer uses a truncated normal with an uncorrected
        standard deviation, whereas here we use a normal distribution. Similarly,
        tf.contrib.layers.variance_scaling_initializer uses a truncated normal with
        a corrected standard deviation.
        Args:
          shape: shape of variable
          dtype: dtype of variable
          partition_info: unused
        Returns:
          an initialization for the variable
        """
    
        def __call__(self, shape, dtype=K.floatx(), **kwargs):
            kernel_height, kernel_width, _, out_filters = shape
            fan_out = int(kernel_height * kernel_width * out_filters)
            return tf.random_normal(
                shape, mean=0.0, stddev=np.sqrt(2.0 / fan_out), dtype=dtype)
    
    
    class EfficientDenseKernelInitializer(Initializer):
        """Initialization for dense kernels.
        This initialization is equal to
          tf.variance_scaling_initializer(scale=1.0/3.0, mode='fan_out',
                                          distribution='uniform').
        It is written out explicitly here for clarity.
        Args:
          shape: shape of variable
          dtype: dtype of variable
        Returns:
          an initialization for the variable
        """
    
        def __call__(self, shape, dtype=K.floatx(), **kwargs):
            """Initialization for dense kernels.
            This initialization is equal to
              tf.variance_scaling_initializer(scale=1.0/3.0, mode='fan_out',
                                              distribution='uniform').
            It is written out explicitly here for clarity.
            Args:
              shape: shape of variable
              dtype: dtype of variable
            Returns:
              an initialization for the variable
            """
            init_range = 1.0 / np.sqrt(shape[1])
            return tf.random_uniform(shape, -init_range, init_range, dtype=dtype)
    
    
    conv_kernel_initializer = EfficientConv2DKernelInitializer()
    dense_kernel_initializer = EfficientDenseKernelInitializer()
    
    
    get_custom_objects().update({
        'EfficientDenseKernelInitializer': EfficientDenseKernelInitializer,
        'EfficientConv2DKernelInitializer': EfficientConv2DKernelInitializer,
    })

#  #

# **layers.py** #

# """   import tensorflow as tf   import numpy as np #

# x=np.array(\[\[1.,8.,7.\],\[10.,14.,3.\],\[1.,2.,4.\]\])   tf.math.sigmoid(x)       <tf.Tensor: id=22, shape=(3, 3), dtype=float64, numpy=       array(\[\[0.73105858, 0.99966465, 0.99908895\],              \[0.9999546 , 0.99999917, 0.95257413\],              \[0.73105858, 0.88079708, 0.98201379\]\])> #

# tf.compat.v1.disable\_eager\_execution()   x=np.array(\[\[1.,8.,7.\],\[10.,14.,3.\],\[1.,2.,4.\]\])        tf.nn.swish(x)   \----------------------------------------------------------------   1.tf.nn.swish(x) 等同于把 x \* tf.sigmoid(beta \* x) 封装了。     如果使用了tf.nn.swish(x) 则需要同时使用tf.compat.v1.disable\_eager\_execution()。     如果使用x \* tf.sigmoid(beta \* x)来代替tf.nn.swish(x)的话，则可以不使用tf.compat.v1.disable\_eager\_execution()。   2.但注意此处可能环境问题使用tf.nn.swish(x)的话会报错，所以此处使用x \* tf.sigmoid(beta \* x)来代替tf.nn.swish(x)     报错信息如下：       tensorflow/core/grappler/utils/graph\_view.cc:830\] No registered 'swish\_f32' OpKernel for GPU devices compatible with node      \{ \{node swish\_75/swish\_f32\}\}  Registered:  <no registered kernels>   """ #

import tensorflow as tf
    import tensorflow.keras.backend as K
    import tensorflow.keras.layers as KL
    from tensorflow.keras.utils import get_custom_objects
    
    """
    import tensorflow as tf
    import numpy as np
    
    x=np.array([[1.,8.,7.],[10.,14.,3.],[1.,2.,4.]])
    tf.math.sigmoid(x)
        <tf.Tensor: id=22, shape=(3, 3), dtype=float64, numpy=
        array([[0.73105858, 0.99966465, 0.99908895],
               [0.9999546 , 0.99999917, 0.95257413],
               [0.73105858, 0.88079708, 0.98201379]])>
    
    tf.compat.v1.disable_eager_execution()
    x=np.array([[1.,8.,7.],[10.,14.,3.],[1.,2.,4.]])	 
    tf.nn.swish(x)
    ----------------------------------------------------------------
    1.tf.nn.swish(x) 等同于把 x * tf.sigmoid(beta * x) 封装了。
      如果使用了tf.nn.swish(x) 则需要同时使用tf.compat.v1.disable_eager_execution()。
      如果使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)的话，则可以不使用tf.compat.v1.disable_eager_execution()。
    2.但注意此处可能环境问题使用tf.nn.swish(x)的话会报错，所以此处使用x * tf.sigmoid(beta * x)来代替tf.nn.swish(x)
      报错信息如下：
        tensorflow/core/grappler/utils/graph_view.cc:830] No registered 'swish_f32' OpKernel for GPU devices compatible with node
       {
       {node swish_75/swish_f32}}  Registered:  <no registered kernels>
    """
    class Swish(KL.Layer):
        def __init__(self, **kwargs):
            super().__init__(**kwargs)
        def call(self, inputs, **kwargs):
            # return tf.nn.swish(inputs)
            return inputs * tf.math.sigmoid(inputs)
    
    
    class DropConnect(KL.Layer):
    
        def __init__(self, drop_connect_rate=0., **kwargs):
            super().__init__(**kwargs)
            self.drop_connect_rate = drop_connect_rate
    
        def call(self, inputs, training=None):
    
            def drop_connect():
                keep_prob = 1.0 - self.drop_connect_rate
    
                # Compute drop_connect tensor
                batch_size = tf.shape(inputs)[0]
                random_tensor = keep_prob
                random_tensor += tf.random.uniform([batch_size, 1, 1, 1], dtype=inputs.dtype)
                binary_tensor = tf.floor(random_tensor)
                output = tf.math.divide(inputs, keep_prob) * binary_tensor
                return output
    
            return K.in_train_phase(drop_connect, inputs, training=training)
    
        def get_config(self):
            config = super().get_config()
            config['drop_connect_rate'] = self.drop_connect_rate
            return config
    
    
    get_custom_objects().update({
        'DropConnect': DropConnect,
        'Swish': Swish,
    })

#  #

# **model.py** #

# Copyright 2019 The TensorFlow Authors. All Rights Reserved.
    #
    # Licensed under the Apache License, Version 2.0 (the "License");
    # you may not use this file except in compliance with the License.
    # You may obtain a copy of the License at
    #
    #     http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.
    # ==============================================================================
    """Contains definitions for EfficientNet model.
    [1] Mingxing Tan, Quoc V. Le
      EfficientNet: Rethinking Model Scaling for Convolutional Neural Networks.
      ICML'19, https://arxiv.org/abs/1905.11946
    """
    
    from __future__ import absolute_import
    from __future__ import division
    from __future__ import print_function
    
    import collections
    import math
    import numpy as np
    import six
    from six.moves import xrange  # pylint: disable=redefined-builtin
    import tensorflow as tf
    
    import tensorflow.keras.backend as K
    import tensorflow.keras.models as KM
    import tensorflow.keras.layers as KL
    from tensorflow.keras.utils import get_file
    from tensorflow.keras.initializers import Initializer
    from .layers import Swish, DropConnect
    from .params import get_model_params, IMAGENET_WEIGHTS
    from .initializers import conv_kernel_initializer, dense_kernel_initializer
    
    
    __all__ = ['EfficientNet', 'EfficientNetB0', 'EfficientNetB1', 'EfficientNetB2', 'EfficientNetB3',
               'EfficientNetB4', 'EfficientNetB5', 'EfficientNetB6', 'EfficientNetB7']
    
    class ConvKernalInitializer(Initializer):
        def __call__(self, shape, dtype=K.floatx(), partition_info=None):
            """Initialization for convolutional kernels.
            The main difference with tf.variance_scaling_initializer is that
            tf.variance_scaling_initializer uses a truncated normal with an uncorrected
            standard deviation, whereas here we use a normal distribution. Similarly,
            tf.contrib.layers.variance_scaling_initializer uses a truncated normal with
            a corrected standard deviation.
            Args:
            shape: shape of variable
            dtype: dtype of variable
            partition_info: unused
            Returns:
            an initialization for the variable
            """
            del partition_info
            kernel_height, kernel_width, _, out_filters = shape
            fan_out = int(kernel_height * kernel_width * out_filters)
            return tf.random.normal(
                shape, mean=0.0, stddev=np.sqrt(2.0 / fan_out), dtype=dtype)
    
    
    class DenseKernalInitializer(Initializer):
        def __call__(self, shape, dtype=K.floatx(), partition_info=None):
            """Initialization for dense kernels.
            This initialization is equal to
            tf.variance_scaling_initializer(scale=1.0/3.0, mode='fan_out',
                                            distribution='uniform').
            It is written out explicitly here for clarity.
            Args:
            shape: shape of variable
            dtype: dtype of variable
            partition_info: unused
            Returns:
            an initialization for the variable
            """
            del partition_info
            init_range = 1.0 / np.sqrt(shape[1])
            return tf.random_uniform(shape, -init_range, init_range, dtype=dtype)
    
    
    def round_filters(filters, global_params):
        """Round number of filters based on depth multiplier."""
        orig_f = filters
        multiplier = global_params.width_coefficient
        divisor = global_params.depth_divisor
        min_depth = global_params.min_depth
        if not multiplier:
            return filters
    
        filters *= multiplier
        min_depth = min_depth or divisor
        new_filters = max(min_depth, int(filters + divisor / 2) // divisor * divisor)
        # Make sure that round down does not go down by more than 10%.
        if new_filters < 0.9 * filters:
            new_filters += divisor
        # print('round_filter input={} output={}'.format(orig_f, new_filters))
        return int(new_filters)
    
    
    def round_repeats(repeats, global_params):
        """Round number of filters based on depth multiplier."""
        multiplier = global_params.depth_coefficient
        if not multiplier:
            return repeats
        return int(math.ceil(multiplier * repeats))
    
    
    def SEBlock(block_args, global_params):
        num_reduced_filters = max(
            1, int(block_args.input_filters * block_args.se_ratio))
        filters = block_args.input_filters * block_args.expand_ratio
        if global_params.data_format == 'channels_first':
            channel_axis = 1
            spatial_dims = [2, 3]
        else:
            channel_axis = -1
            spatial_dims = [1, 2]
    
        def block(inputs):
            x = inputs
            x = KL.Lambda(lambda a: K.mean(a, axis=spatial_dims, keepdims=True))(x)
            x = KL.Conv2D(
                num_reduced_filters,
                kernel_size=[1, 1],
                strides=[1, 1],
                kernel_initializer=ConvKernalInitializer(),
                padding='same',
                use_bias=True
            )(x)
            x = Swish()(x)
            # Excite
            x = KL.Conv2D(
                filters,
                kernel_size=[1, 1],
                strides=[1, 1],
                kernel_initializer=ConvKernalInitializer(),
                padding='same',
                use_bias=True
            )(x)
            x = KL.Activation('sigmoid')(x)
            out = KL.Multiply()([x, inputs])
            return out
    
        return block
    
    
    def MBConvBlock(block_args, global_params, drop_connect_rate=None):
        batch_norm_momentum = global_params.batch_norm_momentum
        batch_norm_epsilon = global_params.batch_norm_epsilon
    
        if global_params.data_format == 'channels_first':
            channel_axis = 1
            spatial_dims = [2, 3]
        else:
            channel_axis = -1
            spatial_dims = [1, 2]
    
        has_se = (block_args.se_ratio is not None) and (
                block_args.se_ratio > 0) and (block_args.se_ratio <= 1)
    
        filters = block_args.input_filters * block_args.expand_ratio
        kernel_size = block_args.kernel_size
    
        def block(inputs):
    
            if block_args.expand_ratio != 1:
                x = KL.Conv2D(
                    filters,
                    kernel_size=[1, 1],
                    strides=[1, 1],
                    kernel_initializer=ConvKernalInitializer(),
                    padding='same',
                    use_bias=False
                )(inputs)
                x = KL.BatchNormalization(
                    axis=channel_axis,
                    momentum=batch_norm_momentum,
                    epsilon=batch_norm_epsilon
                )(x)
                x = Swish()(x)
            else:
                x = inputs
    
            x = KL.DepthwiseConv2D(
                [kernel_size, kernel_size],
                strides=block_args.strides,
                depthwise_initializer=ConvKernalInitializer(),
                padding='same',
                use_bias=False
            )(x)
            x = KL.BatchNormalization(
                axis=channel_axis,
                momentum=batch_norm_momentum,
                epsilon=batch_norm_epsilon
            )(x)
            x = Swish()(x)
    
            if has_se:
                x = SEBlock(block_args, global_params)(x)
    
            # output phase
    
            x = KL.Conv2D(
                block_args.output_filters,
                kernel_size=[1, 1],
                strides=[1, 1],
                kernel_initializer=ConvKernalInitializer(),
                padding='same',
                use_bias=False
            )(x)
            x = KL.BatchNormalization(
                axis=channel_axis,
                momentum=batch_norm_momentum,
                epsilon=batch_norm_epsilon
            )(x)
    
            if block_args.id_skip:
                if all(
                        s == 1 for s in block_args.strides
                ) and block_args.input_filters == block_args.output_filters:
                    # only apply drop_connect if skip presents.
                    if drop_connect_rate:
                        x = DropConnect(drop_connect_rate)(x)
                    x = KL.Add()([x, inputs])
            return x
    
        return block
    
    
    def EfficientNet(input_shape, block_args_list, global_params, include_top=True, pooling=None):
        batch_norm_momentum = global_params.batch_norm_momentum
        batch_norm_epsilon = global_params.batch_norm_epsilon
        if global_params.data_format == 'channels_first':
            channel_axis = 1
        else:
            channel_axis = -1
    
        # Stem part
        inputs = KL.Input(shape=input_shape)
        x = inputs
        x = KL.Conv2D(
            filters=round_filters(32, global_params),
            kernel_size=[3, 3],
            strides=[2, 2],
            kernel_initializer=ConvKernalInitializer(),
            padding='same',
            use_bias=False
        )(x)
        x = KL.BatchNormalization(
            axis=channel_axis,
            momentum=batch_norm_momentum,
            epsilon=batch_norm_epsilon
        )(x)
        x = Swish()(x)
    
        # Blocks part
        block_idx = 1
        n_blocks = sum([block_args.num_repeat for block_args in block_args_list])
        drop_rate = global_params.drop_connect_rate or 0
        drop_rate_dx = drop_rate / n_blocks
    
        for block_args in block_args_list:
            assert block_args.num_repeat > 0
            # Update block input and output filters based on depth multiplier.
            block_args = block_args._replace(
                input_filters=round_filters(block_args.input_filters, global_params),
                output_filters=round_filters(block_args.output_filters, global_params),
                num_repeat=round_repeats(block_args.num_repeat, global_params)
            )
    
            # The first block needs to take care of stride and filter size increase.
            x = MBConvBlock(block_args, global_params,
                            drop_connect_rate=drop_rate_dx * block_idx)(x)
            block_idx += 1
    
            if block_args.num_repeat > 1:
                block_args = block_args._replace(input_filters=block_args.output_filters, strides=[1, 1])
    
            for _ in xrange(block_args.num_repeat - 1):
                x = MBConvBlock(block_args, global_params,
                                drop_connect_rate=drop_rate_dx * block_idx)(x)
                block_idx += 1
    
        # Head part
        x = KL.Conv2D(
            filters=round_filters(1280, global_params),
            kernel_size=[1, 1],
            strides=[1, 1],
            kernel_initializer=ConvKernalInitializer(),
            padding='same',
            use_bias=False
        )(x)
        x = KL.BatchNormalization(
            axis=channel_axis,
            momentum=batch_norm_momentum,
            epsilon=batch_norm_epsilon
        )(x)
        x = Swish()(x)
    
        if include_top:
            x = KL.GlobalAveragePooling2D(data_format=global_params.data_format)(x)
            if global_params.dropout_rate > 0:
                x = KL.Dropout(global_params.dropout_rate)(x)
            x = KL.Dense(global_params.num_classes, kernel_initializer=DenseKernalInitializer())(x)
            x = KL.Activation('softmax')(x)
        else:
            if pooling == 'avg':
                x = KL.GlobalAveragePooling2D(data_format=global_params.data_format)(x)
            elif pooling == 'max':
                x = KL.GlobalMaxPooling2D(data_format=global_params.data_format)(x)
    
        outputs = x
        model = KM.Model(inputs, outputs)
    
        return model
    
    
    def _get_model_by_name(model_name, input_shape=None, include_top=True, weights=None, classes=1000, pooling=None):
        """Re-Implementation of EfficientNet for Keras
        Reference:
            https://arxiv.org/abs/1807.11626
        Args:
            input_shape: optional, if ``None`` default_input_shape is used
                EfficientNetB0 - (224, 224, 3)
                EfficientNetB1 - (240, 240, 3)
                EfficientNetB2 - (260, 260, 3)
                EfficientNetB3 - (300, 300, 3)
                EfficientNetB4 - (380, 380, 3)
                EfficientNetB5 - (456, 456, 3)
                EfficientNetB6 - (528, 528, 3)
                EfficientNetB7 - (600, 600, 3)
            include_top: whether to include the fully-connected
                layer at the top of the network.
            weights: one of `None` (random initialization),
                  'imagenet' (pre-training on ImageNet).
            classes: optional number of classes to classify images
                into, only to be specified if `include_top` is True, and
                if no `weights` argument is specified.
            pooling: optional [None, 'avg', 'max'], if ``include_top=False``
                add global pooling on top of the network
                - avg: GlobalAveragePooling2D
                - max: GlobalMaxPooling2D
        Returns:
            A Keras model instance.
        """
        if weights not in {None, 'imagenet'}:
            raise ValueError('Parameter `weights` should be one of [None, "imagenet"]')
    
        if weights == 'imagenet' and model_name not in IMAGENET_WEIGHTS:
            raise ValueError('There are not pretrained weights for {} model.'.format(model_name))
    
        if weights == 'imagenet' and include_top and classes != 1000:
            raise ValueError('If using `weights` and `include_top`'
                             ' `classes` should be 1000')
    
        block_agrs_list, global_params, default_input_shape = get_model_params(
            model_name, override_params={'num_classes': classes}
        )
    
        if input_shape is None:
            input_shape = (default_input_shape, default_input_shape, 3)
    
        model = EfficientNet(input_shape, block_agrs_list, global_params, include_top=include_top, pooling=pooling)
        model._name = model_name
    
        if weights:
            if not include_top:
                weights_name = model_name + '-notop'
            else:
                weights_name = model_name
            weights = IMAGENET_WEIGHTS[weights_name]
            weights_path = get_file(
                weights['name'],
                weights['url'],
                cache_subdir='models',
                md5_hash=weights['md5'],
            )
            model.load_weights(weights_path)
    
        return model
    
    
    def EfficientNetB0(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b0', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB1(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b1', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB2(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b2', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB3(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b3', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB4(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b4', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB5(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b5', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB6(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b6', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    def EfficientNetB7(include_top=True, input_shape=None, weights=None, classes=1000, pooling=None):
        return _get_model_by_name('efficientnet-b7', include_top=include_top, input_shape=input_shape,
                                  weights=weights, classes=classes, pooling=pooling)
    
    
    EfficientNetB0.__doc__ = _get_model_by_name.__doc__
    EfficientNetB1.__doc__ = _get_model_by_name.__doc__
    EfficientNetB2.__doc__ = _get_model_by_name.__doc__
    EfficientNetB3.__doc__ = _get_model_by_name.__doc__
    EfficientNetB4.__doc__ = _get_model_by_name.__doc__
    EfficientNetB5.__doc__ = _get_model_by_name.__doc__
    EfficientNetB6.__doc__ = _get_model_by_name.__doc__
    EfficientNetB7.__doc__ = _get_model_by_name.__doc__

#  #

# params.py #

import os
    import re
    import collections
    
    
    IMAGENET_WEIGHTS = {
    
        'efficientnet-b0': {
            'name': 'efficientnet-b0_imagenet_1000.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b0_imagenet_1000.h5',
            'md5': 'bca04d16b1b8a7c607b1152fe9261af7',
        },
    
        'efficientnet-b0-notop': {
            'name': 'efficientnet-b0_imagenet_1000_notop.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b0_imagenet_1000_notop.h5',
            'md5': '45d2f3b6330c2401ef66da3961cad769',
        },
    
        'efficientnet-b1': {
            'name': 'efficientnet-b1_imagenet_1000.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b1_imagenet_1000.h5',
            'md5': 'bd4a2b82f6f6bada74fc754553c464fc',
        },
    
        'efficientnet-b1-notop': {
            'name': 'efficientnet-b1_imagenet_1000_notop.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b1_imagenet_1000_notop.h5',
            'md5': '884aed586c2d8ca8dd15a605ec42f564',
        },
    
        'efficientnet-b2': {
            'name': 'efficientnet-b2_imagenet_1000.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b2_imagenet_1000.h5',
            'md5': '45b28b26f15958bac270ab527a376999',
        },
    
        'efficientnet-b2-notop': {
            'name': 'efficientnet-b2_imagenet_1000_notop.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b2_imagenet_1000_notop.h5',
            'md5': '42fb9f2d9243d461d62b4555d3a53b7b',
        },
    
        'efficientnet-b3': {
            'name': 'efficientnet-b3_imagenet_1000.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b3_imagenet_1000.h5',
            'md5': 'decd2c8a23971734f9d3f6b4053bf424',
        },
    
        'efficientnet-b3-notop': {
            'name': 'efficientnet-b3_imagenet_1000_notop.h5',
            'url': 'https://github.com/qubvel/efficientnet/releases/download/v0.0.1/efficientnet-b3_imagenet_1000_notop.h5',
            'md5': '1f7d9a8c2469d2e3d3b97680d45df1e1',
        },
    
    }
    
    
    GlobalParams = collections.namedtuple('GlobalParams', [
        'batch_norm_momentum', 'batch_norm_epsilon', 'dropout_rate', 'data_format',
        'num_classes', 'width_coefficient', 'depth_coefficient',
        'depth_divisor', 'min_depth', 'drop_connect_rate',
    ])
    GlobalParams.__new__.__defaults__ = (None,) * len(GlobalParams._fields)
    
    BlockArgs = collections.namedtuple('BlockArgs', [
        'kernel_size', 'num_repeat', 'input_filters', 'output_filters',
        'expand_ratio', 'id_skip', 'strides', 'se_ratio'
    ])
    # defaults will be a public argument for namedtuple in Python 3.7
    # https://docs.python.org/3/library/collections.html#collections.namedtuple
    BlockArgs.__new__.__defaults__ = (None,) * len(BlockArgs._fields)
    
    
    def efficientnet_params(model_name):
      """Get efficientnet params based on model name."""
      params_dict = {
          # (width_coefficient, depth_coefficient, resolution, dropout_rate)
          'efficientnet-b0': (1.0, 1.0, 224, 0.2),
          'efficientnet-b1': (1.0, 1.1, 240, 0.2),
          'efficientnet-b2': (1.1, 1.2, 260, 0.3),
          'efficientnet-b3': (1.2, 1.4, 300, 0.3),
          'efficientnet-b4': (1.4, 1.8, 380, 0.4),
          'efficientnet-b5': (1.6, 2.2, 456, 0.4),
          'efficientnet-b6': (1.8, 2.6, 528, 0.5),
          'efficientnet-b7': (2.0, 3.1, 600, 0.5),
      }
      return params_dict[model_name]
    
    
    class BlockDecoder(object):
      """Block Decoder for readability."""
    
      def _decode_block_string(self, block_string):
        """Gets a block through a string notation of arguments."""
        assert isinstance(block_string, str)
        ops = block_string.split('_')
        options = {}
        for op in ops:
          splits = re.split(r'(\d.*)', op)
          if len(splits) >= 2:
            key, value = splits[:2]
            options[key] = value
    
        if 's' not in options or len(options['s']) != 2:
          raise ValueError('Strides options should be a pair of integers.')
    
        return BlockArgs(
            kernel_size=int(options['k']),
            num_repeat=int(options['r']),
            input_filters=int(options['i']),
            output_filters=int(options['o']),
            expand_ratio=int(options['e']),
            id_skip=('noskip' not in block_string),
            se_ratio=float(options['se']) if 'se' in options else None,
            strides=[int(options['s'][0]), int(options['s'][1])])
    
      def _encode_block_string(self, block):
        """Encodes a block to a string."""
        args = [
            'r%d' % block.num_repeat,
            'k%d' % block.kernel_size,
            's%d%d' % (block.strides[0], block.strides[1]),
            'e%s' % block.expand_ratio,
            'i%d' % block.input_filters,
            'o%d' % block.output_filters
        ]
        if block.se_ratio > 0 and block.se_ratio <= 1:
          args.append('se%s' % block.se_ratio)
        if block.id_skip is False:
          args.append('noskip')
        return '_'.join(args)
    
      def decode(self, string_list):
        """Decodes a list of string notations to specify blocks inside the network.
        Args:
          string_list: a list of strings, each string is a notation of block.
        Returns:
          A list of namedtuples to represent blocks arguments.
        """
        assert isinstance(string_list, list)
        blocks_args = []
        for block_string in string_list:
          blocks_args.append(self._decode_block_string(block_string))
        return blocks_args
    
      def encode(self, blocks_args):
        """Encodes a list of Blocks to a list of strings.
        Args:
          blocks_args: A list of namedtuples to represent blocks arguments.
        Returns:
          a list of strings, each string is a notation of block.
        """
        block_strings = []
        for block in blocks_args:
          block_strings.append(self._encode_block_string(block))
        return block_strings
    
    
    def efficientnet(width_coefficient=None,
                     depth_coefficient=None,
                     dropout_rate=0.2,
                     drop_connect_rate=0.2):
      """Creates a efficientnet model."""
      blocks_args = [
          'r1_k3_s11_e1_i32_o16_se0.25', 'r2_k3_s22_e6_i16_o24_se0.25',
          'r2_k5_s22_e6_i24_o40_se0.25', 'r3_k3_s22_e6_i40_o80_se0.25',
          'r3_k5_s11_e6_i80_o112_se0.25', 'r4_k5_s22_e6_i112_o192_se0.25',
          'r1_k3_s11_e6_i192_o320_se0.25',
      ]
      global_params = GlobalParams(
          batch_norm_momentum=0.99,
          batch_norm_epsilon=1e-3,
          dropout_rate=dropout_rate,
          drop_connect_rate=drop_connect_rate,
          data_format='channels_last',
          num_classes=1000,
          width_coefficient=width_coefficient,
          depth_coefficient=depth_coefficient,
          depth_divisor=8,
          min_depth=None)
      decoder = BlockDecoder()
      return decoder.decode(blocks_args), global_params
    
    
    def get_model_params(model_name, override_params=None):
      """Get the block args and global params for a given model."""
      if model_name.startswith('efficientnet'):
        width_coefficient, depth_coefficient, input_shape, dropout_rate = (
            efficientnet_params(model_name))
        blocks_args, global_params = efficientnet(
            width_coefficient, depth_coefficient, dropout_rate)
      else:
        raise NotImplementedError('model name is not pre-defined: %s' % model_name)
    
      if override_params:
        # ValueError will be raised here if override_params has fields not included
        # in global_params.
        global_params = global_params._replace(**override_params)
    
      #print('global_params= %s', global_params)
      #print('blocks_args= %s', blocks_args)
      return blocks_args, global_params, input_shape

#  #

# preprocessing.py #

import numpy as np
    from skimage.transform import resize
    
    MEAN_RGB = [0.485 * 255, 0.456 * 255, 0.406 * 255]
    STDDEV_RGB = [0.229 * 255, 0.224 * 255, 0.225 * 255]
    
    MAP_INTERPOLATION_TO_ORDER = {
        'nearest': 0,
        'bilinear': 1,
        'biquadratic': 2,
        'bicubic': 3,
    }
    
    
    def center_crop_and_resize(image, image_size, crop_padding=32, interpolation='bicubic'):
        assert image.ndim in {2, 3}
        assert interpolation in MAP_INTERPOLATION_TO_ORDER.keys()
    
        h, w = image.shape[:2]
    
        padded_center_crop_size = int((image_size / (image_size + crop_padding)) * min(h, w))
        offset_height = ((h - padded_center_crop_size) + 1) // 2
        offset_width = ((w - padded_center_crop_size) + 1) // 2
    
        image_crop = image[offset_height:padded_center_crop_size + offset_height,
                           offset_width:padded_center_crop_size + offset_width]
        resized_image = resize(
            image_crop,
            (image_size, image_size),
            order=MAP_INTERPOLATION_TO_ORDER[interpolation],
            preserve_range=True,
        )
    
        return resized_image
    
    
    def preprocess_input(x):
        assert x.ndim in (3, 4)
        assert x.shape[-1] == 3
    
        x = x - np.array(MEAN_RGB)
        x = x / np.array(STDDEV_RGB)
    
        return x

#  #

[20191009191333910.png]: /images/20230528/54fc86512b6c4b41bf20c253129df0c4.png
[Link 1]: http://www.rimengshe.com/
[AI_Keras PyTorch MXNet TensorFlow PaddlePaddle]: https://blog.csdn.net/zimiao552147572/article/details/88867161