![]() | ![]() | ![]() | ![]() |
このチュートリアルでは、基本、画像のノイズ除去、異常検出の3つの例を使用してオートエンコーダを紹介します。
オートエンコーダは、入力を出力にコピーするようにトレーニングされた特殊なタイプのニューラルネットワークです。たとえば、手書きの数字の画像が与えられた場合、オートエンコーダは最初に画像を低次元の潜在表現にエンコードし、次に潜在表現をデコードして画像に戻します。オートエンコーダは、再構成エラーを最小限に抑えながらデータを圧縮することを学習します。
オートエンコーダの詳細については、から第14章を読んでみて下さいディープラーニングイアン・グッドフェロー、ヨシュア・ベンジオ、およびアーロンCourvilleで。
TensorFlowおよびその他のライブラリをインポートします
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
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 layers, losses
from tensorflow.keras.datasets import fashion_mnist
from tensorflow.keras.models import Model
データセットをロードする
まず、FashonMNISTデータセットを使用して基本的なオートエンコーダーをトレーニングします。このデータセットの各画像は28x28ピクセルです。
(x_train, _), (x_test, _) = fashion_mnist.load_data()
x_train = x_train.astype('float32') / 255.
x_test = x_test.astype('float32') / 255.
print (x_train.shape)
print (x_test.shape)
Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/train-labels-idx1-ubyte.gz 32768/29515 [=================================] - 0s 0us/step Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/train-images-idx3-ubyte.gz 26427392/26421880 [==============================] - 1s 0us/step Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/t10k-labels-idx1-ubyte.gz 8192/5148 [===============================================] - 0s 0us/step Downloading data from https://storage.googleapis.com/tensorflow/tf-keras-datasets/t10k-images-idx3-ubyte.gz 4423680/4422102 [==============================] - 0s 0us/step (60000, 28, 28) (10000, 28, 28)
最初の例:基本的なオートエンコーダ
画像を64次元の潜在ベクトルに圧縮するencoder
と、潜在空間から元の画像を再構築するdecoder
2つの高密度レイヤーを使用してオートencoder
を定義します。
モデルを定義するには、 Keras Model SubclassingAPIを使用します。
latent_dim = 64
class Autoencoder(Model):
def __init__(self, latent_dim):
super(Autoencoder, self).__init__()
self.latent_dim = latent_dim
self.encoder = tf.keras.Sequential([
layers.Flatten(),
layers.Dense(latent_dim, activation='relu'),
])
self.decoder = tf.keras.Sequential([
layers.Dense(784, activation='sigmoid'),
layers.Reshape((28, 28))
])
def call(self, x):
encoded = self.encoder(x)
decoded = self.decoder(encoded)
return decoded
autoencoder = Autoencoder(latent_dim)
autoencoder.compile(optimizer='adam', loss=losses.MeanSquaredError())
入力とターゲットの両方としてx_train
を使用してモデルをx_train
します。 encoder
はデータセットを784次元から潜在空間に圧縮することを学習し、 decoder
は元の画像を再構築することを学習します。 。
autoencoder.fit(x_train, x_train,
epochs=10,
shuffle=True,
validation_data=(x_test, x_test))
Epoch 1/10 1875/1875 [==============================] - 4s 2ms/step - loss: 0.0400 - val_loss: 0.0131 Epoch 2/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0123 - val_loss: 0.0106 Epoch 3/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0104 - val_loss: 0.0098 Epoch 4/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0096 - val_loss: 0.0094 Epoch 5/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0094 - val_loss: 0.0093 Epoch 6/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0092 - val_loss: 0.0091 Epoch 7/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0091 - val_loss: 0.0092 Epoch 8/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0090 - val_loss: 0.0090 Epoch 9/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0089 - val_loss: 0.0090 Epoch 10/10 1875/1875 [==============================] - 3s 2ms/step - loss: 0.0089 - val_loss: 0.0089 <tensorflow.python.keras.callbacks.History at 0x7f966d534588>
モデルがトレーニングされたので、テストセットから画像をエンコードおよびデコードしてモデルをテストしましょう。
encoded_imgs = autoencoder.encoder(x_test).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()
n = 10
plt.figure(figsize=(20, 4))
for i in range(n):
# display original
ax = plt.subplot(2, n, i + 1)
plt.imshow(x_test[i])
plt.title("original")
plt.gray()
ax.get_xaxis().set_visible(False)
ax.get_yaxis().set_visible(False)
# display reconstruction
ax = plt.subplot(2, n, i + 1 + n)
plt.imshow(decoded_imgs[i])
plt.title("reconstructed")
plt.gray()
ax.get_xaxis().set_visible(False)
ax.get_yaxis().set_visible(False)
plt.show()
2番目の例:画像のノイズ除去
オートエンコーダは、画像からノイズを除去するようにトレーニングすることもできます。次のセクションでは、各画像にランダムノイズを適用して、ノイズの多いバージョンのFashionMNISTデータセットを作成します。次に、ノイズの多い画像を入力として使用し、元の画像をターゲットとして使用して、オートエンコーダーをトレーニングします。
データセットを再インポートして、以前に行った変更を省略しましょう。
(x_train, _), (x_test, _) = fashion_mnist.load_data()
x_train = x_train.astype('float32') / 255.
x_test = x_test.astype('float32') / 255.
x_train = x_train[..., tf.newaxis]
x_test = x_test[..., tf.newaxis]
print(x_train.shape)
(60000, 28, 28, 1)
画像にランダムノイズを追加する
noise_factor = 0.2
x_train_noisy = x_train + noise_factor * tf.random.normal(shape=x_train.shape)
x_test_noisy = x_test + noise_factor * tf.random.normal(shape=x_test.shape)
x_train_noisy = tf.clip_by_value(x_train_noisy, clip_value_min=0., clip_value_max=1.)
x_test_noisy = tf.clip_by_value(x_test_noisy, clip_value_min=0., clip_value_max=1.)
ノイズの多い画像をプロットします。
n = 10
plt.figure(figsize=(20, 2))
for i in range(n):
ax = plt.subplot(1, n, i + 1)
plt.title("original + noise")
plt.imshow(tf.squeeze(x_test_noisy[i]))
plt.gray()
plt.show()
畳み込みオートエンコーダーを定義する
この例では、 encoder
Conv2Dレイヤーと、 decoder
Conv2DTransposeレイヤーを使用して畳み込みオートencoder
をトレーニングします。
class Denoise(Model):
def __init__(self):
super(Denoise, self).__init__()
self.encoder = tf.keras.Sequential([
layers.Input(shape=(28, 28, 1)),
layers.Conv2D(16, (3, 3), activation='relu', padding='same', strides=2),
layers.Conv2D(8, (3, 3), activation='relu', padding='same', strides=2)])
self.decoder = tf.keras.Sequential([
layers.Conv2DTranspose(8, kernel_size=3, strides=2, activation='relu', padding='same'),
layers.Conv2DTranspose(16, kernel_size=3, strides=2, activation='relu', padding='same'),
layers.Conv2D(1, kernel_size=(3, 3), activation='sigmoid', padding='same')])
def call(self, x):
encoded = self.encoder(x)
decoded = self.decoder(encoded)
return decoded
autoencoder = Denoise()
autoencoder.compile(optimizer='adam', loss=losses.MeanSquaredError())
autoencoder.fit(x_train_noisy, x_train,
epochs=10,
shuffle=True,
validation_data=(x_test_noisy, x_test))
Epoch 1/10 1875/1875 [==============================] - 8s 3ms/step - loss: 0.0336 - val_loss: 0.0098 Epoch 2/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0094 - val_loss: 0.0087 Epoch 3/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0084 - val_loss: 0.0079 Epoch 4/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0077 - val_loss: 0.0074 Epoch 5/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0072 - val_loss: 0.0071 Epoch 6/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0071 - val_loss: 0.0071 Epoch 7/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0070 - val_loss: 0.0070 Epoch 8/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0069 - val_loss: 0.0070 Epoch 9/10 1875/1875 [==============================] - 6s 3ms/step - loss: 0.0069 - val_loss: 0.0069 Epoch 10/10 1875/1875 [==============================] - 5s 3ms/step - loss: 0.0068 - val_loss: 0.0068 <tensorflow.python.keras.callbacks.History at 0x7f96703253c8>
エンコーダーの概要を見てみましょう。画像が28x28から7x7にダウンサンプリングされていることに注目してください。
autoencoder.encoder.summary()
Model: "sequential_2" _________________________________________________________________ Layer (type) Output Shape Param # ================================================================= conv2d (Conv2D) (None, 14, 14, 16) 160 _________________________________________________________________ conv2d_1 (Conv2D) (None, 7, 7, 8) 1160 ================================================================= Total params: 1,320 Trainable params: 1,320 Non-trainable params: 0 _________________________________________________________________
デコーダーは画像を7x7から28x28にアップサンプリングします。
autoencoder.decoder.summary()
Model: "sequential_3" _________________________________________________________________ Layer (type) Output Shape Param # ================================================================= conv2d_transpose (Conv2DTran (None, 14, 14, 8) 584 _________________________________________________________________ conv2d_transpose_1 (Conv2DTr (None, 28, 28, 16) 1168 _________________________________________________________________ conv2d_2 (Conv2D) (None, 28, 28, 1) 145 ================================================================= Total params: 1,897 Trainable params: 1,897 Non-trainable params: 0 _________________________________________________________________
オートエンコーダによって生成されたノイズの多い画像とノイズ除去された画像の両方をプロットします。
encoded_imgs = autoencoder.encoder(x_test).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()
n = 10
plt.figure(figsize=(20, 4))
for i in range(n):
# display original + noise
ax = plt.subplot(2, n, i + 1)
plt.title("original + noise")
plt.imshow(tf.squeeze(x_test_noisy[i]))
plt.gray()
ax.get_xaxis().set_visible(False)
ax.get_yaxis().set_visible(False)
# display reconstruction
bx = plt.subplot(2, n, i + n + 1)
plt.title("reconstructed")
plt.imshow(tf.squeeze(decoded_imgs[i]))
plt.gray()
bx.get_xaxis().set_visible(False)
bx.get_yaxis().set_visible(False)
plt.show()
3番目の例:異常検出
概要概要
この例では、 ECG5000データセットの異常を検出するようにオートエンコーダーをトレーニングします。このデータセットには、それぞれ140のデータポイントを持つ5,000の心電図が含まれています。データセットの簡略化されたバージョンを使用します。各例には、 0
(異常なリズムに対応)または1
(正常なリズムに対応)のいずれかのラベルが付けられています。あなたは異常なリズムを特定することに興味があります。
オートエンコーダーを使用して異常をどのように検出しますか?オートエンコーダは、再構築エラーを最小限に抑えるようにトレーニングされていることを思い出してください。通常のリズムでのみオートエンコーダーをトレーニングし、それを使用してすべてのデータを再構築します。私たちの仮説は、異常なリズムはより高い再構成エラーを持つだろうということです。次に、再構成エラーが固定しきい値を超えた場合、リズムを異常として分類します。
ECGデータをロードする
使用するデータセットは、 timeseriesclassification.comのデータセットに基づいています。
# Download the dataset
dataframe = pd.read_csv('http://storage.googleapis.com/download.tensorflow.org/data/ecg.csv', header=None)
raw_data = dataframe.values
dataframe.head()
# The last element contains the labels
labels = raw_data[:, -1]
# The other data points are the electrocadriogram data
data = raw_data[:, 0:-1]
train_data, test_data, train_labels, test_labels = train_test_split(
data, labels, test_size=0.2, random_state=21
)
データを[0,1]
正規化します。
min_val = tf.reduce_min(train_data)
max_val = tf.reduce_max(train_data)
train_data = (train_data - min_val) / (max_val - min_val)
test_data = (test_data - min_val) / (max_val - min_val)
train_data = tf.cast(train_data, tf.float32)
test_data = tf.cast(test_data, tf.float32)
このデータセットで1
とラベル付けされている通常のリズムのみを使用してオートエンコーダーをトレーニングします。正常なリズムを異常なリズムから分離します。
train_labels = train_labels.astype(bool)
test_labels = test_labels.astype(bool)
normal_train_data = train_data[train_labels]
normal_test_data = test_data[test_labels]
anomalous_train_data = train_data[~train_labels]
anomalous_test_data = test_data[~test_labels]
通常のECGをプロットします。
plt.grid()
plt.plot(np.arange(140), normal_train_data[0])
plt.title("A Normal ECG")
plt.show()
異常なECGをプロットします。
plt.grid()
plt.plot(np.arange(140), anomalous_train_data[0])
plt.title("An Anomalous ECG")
plt.show()
モデルを構築する
class AnomalyDetector(Model):
def __init__(self):
super(AnomalyDetector, self).__init__()
self.encoder = tf.keras.Sequential([
layers.Dense(32, activation="relu"),
layers.Dense(16, activation="relu"),
layers.Dense(8, activation="relu")])
self.decoder = tf.keras.Sequential([
layers.Dense(16, activation="relu"),
layers.Dense(32, activation="relu"),
layers.Dense(140, activation="sigmoid")])
def call(self, x):
encoded = self.encoder(x)
decoded = self.decoder(encoded)
return decoded
autoencoder = AnomalyDetector()
autoencoder.compile(optimizer='adam', loss='mae')
オートエンコーダは通常のECGのみを使用してトレーニングされますが、完全なテストセットを使用して評価されることに注意してください。
history = autoencoder.fit(normal_train_data, normal_train_data,
epochs=20,
batch_size=512,
validation_data=(test_data, test_data),
shuffle=True)
Epoch 1/20 5/5 [==============================] - 1s 40ms/step - loss: 0.0579 - val_loss: 0.0531 Epoch 2/20 5/5 [==============================] - 0s 12ms/step - loss: 0.0555 - val_loss: 0.0515 Epoch 3/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0524 - val_loss: 0.0501 Epoch 4/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0490 - val_loss: 0.0481 Epoch 5/20 5/5 [==============================] - 0s 10ms/step - loss: 0.0452 - val_loss: 0.0460 Epoch 6/20 5/5 [==============================] - 0s 10ms/step - loss: 0.0413 - val_loss: 0.0445 Epoch 7/20 5/5 [==============================] - 0s 10ms/step - loss: 0.0375 - val_loss: 0.0425 Epoch 8/20 5/5 [==============================] - 0s 12ms/step - loss: 0.0343 - val_loss: 0.0412 Epoch 9/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0320 - val_loss: 0.0399 Epoch 10/20 5/5 [==============================] - 0s 13ms/step - loss: 0.0302 - val_loss: 0.0392 Epoch 11/20 5/5 [==============================] - 0s 12ms/step - loss: 0.0285 - val_loss: 0.0382 Epoch 12/20 5/5 [==============================] - 0s 12ms/step - loss: 0.0275 - val_loss: 0.0379 Epoch 13/20 5/5 [==============================] - 0s 13ms/step - loss: 0.0268 - val_loss: 0.0373 Epoch 14/20 5/5 [==============================] - 0s 13ms/step - loss: 0.0259 - val_loss: 0.0367 Epoch 15/20 5/5 [==============================] - 0s 13ms/step - loss: 0.0254 - val_loss: 0.0364 Epoch 16/20 5/5 [==============================] - 0s 12ms/step - loss: 0.0245 - val_loss: 0.0357 Epoch 17/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0238 - val_loss: 0.0351 Epoch 18/20 5/5 [==============================] - 0s 55ms/step - loss: 0.0231 - val_loss: 0.0349 Epoch 19/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0227 - val_loss: 0.0340 Epoch 20/20 5/5 [==============================] - 0s 11ms/step - loss: 0.0224 - val_loss: 0.0338
plt.plot(history.history["loss"], label="Training Loss")
plt.plot(history.history["val_loss"], label="Validation Loss")
plt.legend()
<matplotlib.legend.Legend at 0x7f96601235c0>
再構成エラーが通常のトレーニング例からの1標準偏差より大きい場合、すぐにECGを異常として分類します。まず、トレーニングセットからの通常のECG、オートエンコーダによってエンコードおよびデコードされた後の再構成、および再構成エラーをプロットしてみましょう。
encoded_imgs = autoencoder.encoder(normal_test_data).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()
plt.plot(normal_test_data[0], 'b')
plt.plot(decoded_imgs[0], 'r')
plt.fill_between(np.arange(140), decoded_imgs[0], normal_test_data[0], color='lightcoral')
plt.legend(labels=["Input", "Reconstruction", "Error"])
plt.show()
同様のプロットを作成します。今回は異常なテスト例です。
encoded_imgs = autoencoder.encoder(anomalous_test_data).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()
plt.plot(anomalous_test_data[0], 'b')
plt.plot(decoded_imgs[0], 'r')
plt.fill_between(np.arange(140), decoded_imgs[0], anomalous_test_data[0], color='lightcoral')
plt.legend(labels=["Input", "Reconstruction", "Error"])
plt.show()
異常を検出する
再構成損失が固定しきい値より大きいかどうかを計算することにより、異常を検出します。このチュートリアルでは、トレーニングセットからの正規例の平均平均誤差を計算し、再構成誤差がトレーニングセットからの1標準偏差よりも大きい場合、将来の例を異常として分類します。
トレーニングセットからの通常のECGに再構成エラーをプロットします
reconstructions = autoencoder.predict(normal_train_data)
train_loss = tf.keras.losses.mae(reconstructions, normal_train_data)
plt.hist(train_loss, bins=50)
plt.xlabel("Train loss")
plt.ylabel("No of examples")
plt.show()
平均より1標準偏差上のしきい値を選択します。
threshold = np.mean(train_loss) + np.std(train_loss)
print("Threshold: ", threshold)
Threshold: 0.033693388
テストセットの異常な例の再構成エラーを調べると、ほとんどの場合、しきい値よりも再構成エラーが大きいことがわかります。しきい値を変更することで、分類器の適合率と再現率を調整できます。
reconstructions = autoencoder.predict(anomalous_test_data)
test_loss = tf.keras.losses.mae(reconstructions, anomalous_test_data)
plt.hist(test_loss, bins=50)
plt.xlabel("Test loss")
plt.ylabel("No of examples")
plt.show()
再構成エラーがしきい値よりも大きい場合は、ECGを異常として分類します。
def predict(model, data, threshold):
reconstructions = model(data)
loss = tf.keras.losses.mae(reconstructions, data)
return tf.math.less(loss, threshold)
def print_stats(predictions, labels):
print("Accuracy = {}".format(accuracy_score(labels, preds)))
print("Precision = {}".format(precision_score(labels, preds)))
print("Recall = {}".format(recall_score(labels, preds)))
preds = predict(autoencoder, test_data, threshold)
print_stats(preds, test_labels)
Accuracy = 0.943 Precision = 0.9921722113502935 Recall = 0.9053571428571429
次のステップ
オートエンコーダーを使用した異常検出の詳細については、VictorDibiaがTensorFlow.jsを使用して構築したこの優れたインタラクティブな例をご覧ください。実際のユースケースでは、 エアバスがTensorFlowを使用してISSテレメトリデータの異常を検出する方法を学ぶことができます。基本の詳細については、FrançoisCholletによるこのブログ投稿を読むことを検討してください。詳細については、第14章からチェックアウトディープラーニングイアン・グッドフェローによって、ヨシュア・ベンジオ、およびアーロンCourville。