keras_efficientnetをファインチューニングし、Grad-CAMを適用してみる

.
├ dataset
│  ├ test
│  │  ├ homura
│  │  │  ├ homura_01.jpg
│  │  │  └ xxx.jpg
│  │  ├ kyoko
│  │  └ ...
│  └ train
│     ├ homura
│     │  ├ homura_01.jpg
│     │  └ xxx.jpg
│     ├ kyoko
│     └ ...
├ image_val
│  ├ homura_val_01.jpg
│  └ xxx.jpg
├ finetune.py
└ gradcam.py

$ pip install keras-efficientnet 

import glob
import os

import matplotlib.pyplot as plt
import numpy as np
from keras.callbacks import EarlyStopping, ReduceLROnPlateau, TensorBoard
from keras.layers import BatchNormalization, Dense, Dropout, GlobalAveragePooling2D
from keras.models import Model
from keras.optimizers import Adam
from keras.utils.np_utils import to_categorical
from keras_efficientnets import EfficientNetB4
from PIL import Image
from sklearn.model_selection import train_test_split
from tensorflow.keras.preprocessing.image import ImageDataGenerator

# クラス取得
classes = os.listdir("./dataset/train")

# クラス数
num_classes = len(classes)

# 画像サイズ
image_size = 380
input_shape = (image_size, image_size, 3)

# lossの設定
if num_classes > 2:
    loss_type = "categorical_crossentropy"
else:
    loss_type = "binary_crossentropy"


# 画像を取得し、配列に変換
def im2array(path):
    X = []
    y = []
    class_num = 0

    for class_name in classes:
        if class_num == num_classes:
            break

        imgfiles = glob.glob("{0}/{1}/*".format(path, class_name))
        for imgfile in imgfiles:
            # 画像読み込み
            image = Image.open(imgfile)
            # RGB変換
            image = image.convert("RGB")
            # リサイズ
            image = image.resize((image_size, image_size))
            # 画像から配列に変換
            data = np.asarray(image)
            X.append(data)
            y.append(classes.index(class_name))
        class_num += 1

    X = np.array(X)
    y = np.array(y)

    return X, y


# trainデータ取得
X_train, y_train = im2array("./dataset/train")

# testデータ取得
X_test, y_test = im2array("./dataset/test")

# データ型の変換
X_train = X_train.astype("float32")
X_test = X_test.astype("float32")

# 正規化
X_train /= 255
X_test /= 255

# one-hot 変換
y_train = to_categorical(y_train, num_classes=num_classes)
y_test = to_categorical(y_test, num_classes=num_classes)

# trainデータからvalidデータを分割
X_train, X_valid, y_train, y_valid = train_test_split(
    X_train, y_train, random_state=0, stratify=y_train, test_size=0.2
)

# data augmentation
datagen = ImageDataGenerator(
    featurewise_center=False,
    samplewise_center=False,
    featurewise_std_normalization=False,
    samplewise_std_normalization=False,
    zca_whitening=False,
    rotation_range=20,
    width_shift_range=0.2,
    height_shift_range=0.2,
    horizontal_flip=True,
    vertical_flip=False,
)

# EarlyStopping
early_stopping = EarlyStopping(
    monitor="val_loss", patience=10, min_delta=0, mode="auto", verbose=1
)

# reduce learning rate
reduce_lr = ReduceLROnPlateau(
    monitor="val_loss",
    factor=0.1,
    patience=10,
    mode="auto",
    epsilon=0.0001,
    cooldown=0,
    min_lr=0,
    verbose=1,
)


# モデル学習
def model_fit():
    hist = model.fit_generator(
        datagen.flow(X_train, y_train, batch_size=4),
        steps_per_epoch=X_train.shape[0] // 4,
        epochs=25,
        validation_data=(X_valid, y_valid),
        callbacks=[early_stopping, reduce_lr],
        shuffle=True,
        verbose=1,
    )
    return hist


# モデル保存
def model_save():
    model_dir = "./model"
    if os.path.exists(model_dir) == False:
        os.mkdir(model_dir)
    model_path = os.path.join(model_dir, "finetune_model")
    if not os.path.exists(model_path):
        os.makedirs(model_path)
    model.save(model_path)


# 学習曲線をプロット
def learning_plot(title):
    plt.figure(figsize=(18, 6))

    # accuracy
    plt.subplot(1, 2, 1)
    plt.plot(hist.history["accuracy"], label="accuracy", marker="o")
    plt.plot(hist.history["val_accuracy"], label="val_accuracy", marker="o")
    plt.ylabel("accuracy")
    plt.xlabel("epoch")
    plt.title(title)
    plt.legend(loc="best")
    plt.grid(color="gray", alpha=0.2)

    # loss
    plt.subplot(1, 2, 2)
    plt.plot(hist.history["loss"], label="loss", marker="o")
    plt.plot(hist.history["val_loss"], label="val_loss", marker="o")
    plt.ylabel("loss")
    plt.xlabel("epoch")
    plt.title(title)
    plt.legend(loc="best")
    plt.grid(color="gray", alpha=0.2)

    plt.show()


# モデル評価
def model_evaluate():
    score = model.evaluate(X_test, y_test, verbose=1)
    print("evaluate loss: {[0]:.4f}".format(score))
    print("evaluate acc: {[1]:.1%}".format(score))


# xception
base_model = EfficientNetB4(
    include_top=False, weights="imagenet", input_shape=input_shape
)

# 全結合層の新規構築
x = GlobalAveragePooling2D()(base_model.output)
x = BatchNormalization()(x)
# 間にDropoutを追加すると性能が上がる場合がある
x = Dropout(0.5)(x)
output = Dense(num_classes, activation="softmax", name="last_output")(x)

# ネットワーク定義
model = Model(inputs=base_model.input, outputs=output, name="model")
print("{}層".format(len(model.layers)))

# 109層までfreeze
for layer in base_model.layers[:109]:
    layer.trainable = False

# モデルのcompile
model.compile(
    optimizer=Adam(learning_rate=0.0001), loss=loss_type, metrics=["accuracy"]
)

hist = model_fit()

learning_plot("EfficientNetB4")

model_evaluate()

model_save()

evaluate loss: 0.0379
evaluate acc: 97.3%

import glob
import os

import cv2
import matplotlib.pyplot as plt
import numpy as np
import tensorflow as tf
from keras.models import Model, load_model
from keras_efficientnets.efficientnet import preprocess_input
from PIL import Image
from tensorflow import keras
from tensorflow.keras.preprocessing import image as kimage

# 画像サイズ
image_size = 380


def apply_gradcam(model, img, layer_name):
    original_img = kimage.img_to_array(img)
    img = img.resize((image_size, image_size))
    img_array = kimage.img_to_array(img)
    img_array = np.expand_dims(img_array, axis=0)
    img_array = preprocess_input(img_array)

    # レイヤーを取得
    target_layer = model.get_layer(layer_name)

    # 予測クラスの取得
    prediction = model.predict(img_array)
    prediction_idx = np.argmax(prediction)
    print("prediction:", prediction)
    print("prediction_idx:", prediction_idx)

    # 予測されたTopのクラスのGradientを計算
    with tf.GradientTape() as tape:
        gradient_model = Model([model.inputs], [target_layer.output, model.output])
        conv2d_out, prediction = gradient_model(img_array)
        # Prediction Lossの取得
        loss = prediction[:, prediction_idx]

    # conv_outputに対する損失の勾配を取得
    gradients = tape.gradient(loss, conv2d_out)

    # Shape [1 x H x W x CHANNEL] -> [H x W x CHANNEL]から出力を取得
    output = conv2d_out[0]

    # 勾配を空間的に平均化する
    weights = tf.reduce_mean(gradients[0], axis=(1, 2))

    activation_map = np.zeros(output.shape[0:2], dtype=np.float32)

    # layerごとにweightを乗算する
    for idx, weight in enumerate(weights):
        activation_map += weight * output[:, :, idx]

    # 元画像のサイズにリサイズを行う
    activation_map = cv2.resize(
        activation_map.numpy(), (original_img.shape[1], original_img.shape[0])
    )

    # 負の数をなくす
    activation_map = np.maximum(activation_map, 0)

    max_val = activation_map.max()
    min_val = activation_map.min()

    # Activation Mapを0 - 255に変換
    if max_val - min_val == 0:
        activation_map = np.zeros_like(activation_map)
    else:
        activation_map = (activation_map - min_val) / (max_val - min_val)

    activation_map = np.uint8(255 * activation_map)

    # heatmapに変換
    heatmap = cv2.applyColorMap(activation_map, cv2.COLORMAP_JET)

    original_img = np.uint8(
        (original_img - original_img.min())
        / (original_img.max() - original_img.min())
        * 255
    )

    cvt_heatmap = cv2.cvtColor(heatmap, cv2.COLOR_BGR2RGB)

    # heatmapと元画像を重ねる
    result = np.uint8(original_img * 0.5 + cvt_heatmap * (1 - 0.5))

    return result


def run_on_image(model, layer_name, image_path, output_dir):
    img = Image.open(image_path).convert("RGB")

    # heatmapと元画像が重なった画像を取得
    result = apply_gradcam(model, img, layer_name)

    # 結果を保存
    result_path = os.path.join(
        output_dir, f"{os.path.basename(image_path)}_gradcam.jpg"
    )
    plt.rcParams["figure.dpi"] = 100
    plt.imsave(result_path, result)


if __name__ == "__main__":
    input_dir = "image_val"
    output_dir = "image_result"

    model = load_model("./model/finetune_model", compile=False)

    # GradCAMの対象レイヤー（最後のConv2Dレイヤー）を取得
    last_conv_layer = next(
        x for x in model.layers[::-1] if isinstance(x, keras.layers.Conv2D)
    )
    target_layer_name = last_conv_layer.name

    if not os.path.exists(output_dir):
        os.mkdir(output_dir)

    image_paths = glob.glob(f"{input_dir}/*.jpg")

    for image_path in image_paths:
        print(f"Processing {image_path}")
        run_on_image(model, target_layer_name, image_path, output_dir)