【入門】ディープラーニング（オートエンコーダ）を用いた異常検知入門【Python】

import os
from sklearn.metrics import roc_auc_score, roc_curve
import matplotlib.pyplot as plt
import numpy as np
import copy

import tensorflow as tf
from sklearn.metrics import accuracy_score, precision_score, recall_score
from sklearn.model_selection import train_test_split
from tensorflow.keras import datasets, layers, losses, optimizers
from tensorflow.keras.models import Model

(x_train, y_train), (x_test, y_test) = datasets.mnist.load_data()
x_train = x_train / 255
x_test = x_test / 255

normal_idx = 1
abnormal_idx = 0

x_normal_train = x_train[np.where(y_train==normal_idx)]

x_normal_test = x_test[np.where(y_test==normal_idx)]
x_abnormal_test = x_test[np.where(y_test==abnormal_idx)]

latent_dim = 32

# エンコーダ
inputs = layers.Input(shape=(28, 28, 1))
x = layers.Conv2D(128, 3,  strides=1, padding="same",activation="relu")(inputs)
x = layers.Conv2D(64, 3,  strides=1, padding="same", activation="relu")(x)
x = layers.Flatten()(x)
x = layers.Dense(1024, activation="relu")(x)
x = layers.Dense(64, activation="relu")(x)
x = layers.Dense(latent_dim)(x)

# デコーダ
x = layers.Dense(64, activation="relu")(x)
x = layers.Dense(1024, activation="relu")(x)
x = layers.Dense(7 * 7 * 64, activation="relu")(x)

x = layers.Reshape((7, 7, 64))(x)
x = layers.Conv2DTranspose(64, 3, activation="relu", strides=2, padding="same")(x)
x = layers.Conv2DTranspose(128, 3, activation="relu", strides=2, padding="same")(x)

outputs = layers.Conv2DTranspose(1, 3, strides=1,activation="sigmoid", padding="same")(x)


# モデルを作る
model = Model(inputs, outputs)
print(model.summary())
model.compile(optimizer=optimizers.Adam(), loss="binary_crossentropy")

x_normal_train = np.expand_dims(x_normal_train, axis=-1)
x_train, x_val = train_test_split(x_normal_train,train_size=0.9)

MODEL_DIR = "/content/drive/MyDrive/"

checkpoint_filepath = os.path.join(MODEL_DIR,'checkpoint')
model_checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(
    filepath=checkpoint_filepath,
    save_weights_only=True,
    monitor='val_loss',
    mode='min',
    verbose=1,
    save_best_only=True)

history = model.fit(x_train,x_train, epochs=30, shuffle=True, validation_data=(x_val, x_val), callbacks=[model_checkpoint_callback])
plt.plot(np.arange(len(history.history["loss"])),history.history["loss"], label="loss")
plt.plot(np.arange(len(history.history["val_loss"])),history.history["val_loss"], label="val_loss")
plt.legend()

model.load_weights(checkpoint_filepath)

def plot_input_output(input_imgs):
    plt.figure(figsize=(10,10))
    plt_idx = 1
    for i in range(36):
        if plt_idx > 36:
            break
        plt.subplot(6,6,plt_idx)
        plt.xticks([])
        plt.yticks([])
        plt.grid(False)
        plt.imshow(np.squeeze(input_imgs[i]), cmap="gray", vmin=0, vmax=1)
        plt.xlabel("input_{}".format(i))

        plt.subplot(6,6,plt_idx+1)
        plt.xticks([])
        plt.yticks([])
        plt.grid(False)
        
        output = model.predict(tf.expand_dims(input_imgs[i],axis=0))
        plt.imshow(np.squeeze(output), cmap="gray", vmin=0, vmax=1)
        plt.xlabel("output_{}".format(i))
        plt_idx+=2
    plt.show()

plot_input_output(x_normal_test)

plot_input_output(x_abnormal_test)

import seaborn as sns
sns.set(style='white', context='notebook', palette='deep')

def get_errors(input_imgs):
    output_img = model.predict(input_imgs)
    output_imgs = np.squeeze(output_img)
    print(output_imgs.shape)
    print(input_imgs.shape)
    sub_imgs = np.abs(input_imgs-output_imgs)
    errors = np.sum(sub_imgs, axis=(1,2))
    return errors

x_normal_errors = get_errors(x_normal_test) 
x_abnormal_errors = get_errors(x_abnormal_test) 
ax=sns.distplot(x_normal_errors,bins=20, label="normal")
sns.distplot(x_abnormal_errors,ax=ax,bins=20, label="abnormal")
ax.set_xlabel("error")
plt.legend()