tensorboardで特徴量と再構成画像を可視化してみた。

以前に投稿したmnistをvariational auto encoderで教師なし学習し、潜在変数をクラスタリングして評価するのコードに下記を追記して、tensorboardで可視化した。

再構成画像が潜在変数空間内でどのようになっているのかを概観できる。

metadata.tsvをGUIで読み込まないとラベルが反映されない問題が出たが、tensorboardを起動するときに、./tflogではなく、./tflog/YYYY-MM-DDnnnnnnを指定することでmetadata.tsvが読み込まれるので注意
・・・できれば./tflogで読んで欲しいが、その場合はチェックポイントファイル自体も./tflogに書き出す必要がある模様。

projectorに指定する変数名はtensorboadに表示される変数名を指定すること

ラベルを可視化するためのメタデータを作成

# メタデータ作成・保存
import os
import numpy as np
import datetime
LOG_DIR = f"./tflog/{datetime.datetime.now().strftime('%Y-%m-%d-%H%M%S')}"
os.makedirs(os.path.join(LOG_DIR), exist_ok=True)
with open(os.path.join(LOG＿DIR, 'metadata.tsv'), 'w') as f:
    f.write("Index\tLabel\n")
    for index,label in enumerate(np.where(y_test)[1]):
        f.write("%d\t%d\n" % (index,label))

コールバックを作成し、学習中のデータを記録するようにする

tb_cb = tensorflow.keras.callbacks.TensorBoard(log_dir= LOG_DIR,
    histogram_freq=1)
cbks = [tb_cb]

history = vae.fit(x_train,
                epochs=20,
                batch_size=512,
                validation_data=(x_test, None),
                callbacks=cbks)

チェックポイントを作成し、Projectorに出力する(spriteイメージ付き)

result = vae.predict(x_test)

emb = result[1]

# embeddingsの作成
embedding_var = tensorflow.Variable(emb,  name='mnist_embedding')

CHECKPOINT_FILE = LOG_DIR + '/model.ckpt'
ckpt = tensorflow.train.Checkpoint(embeddings=embedding_var)
ckpt.save(CHECKPOINT_FILE)

from tensorboard.plugins import projector
# Projector設定
config = projector.ProjectorConfig()
embedding = config.embeddings.add()
embedding.tensor_name = "embeddings/.ATTRIBUTES/VARIABLE_VALUE"
# メタデータ(CSV)パス
embedding.metadata_path='metadata.tsv'

def create_sprite_image(images):
    """Returns a sprite image consisting of images passed as argument. Images should be count x width x height"""
    # MNISTは28ピクセル四方 
    img_h = images.shape[1]
    img_w = images.shape[2]

    # 画像数の平方根(切上)を計算(Sprite Imageの1辺の長さに使用)
    n_plots = int(np.ceil(np.sqrt(images.shape[0])))

    # 全要素0の配列作成
    spriteimage = np.ones((img_h * n_plots ,img_w * n_plots ))

    for i in range(n_plots):
        for j in range(n_plots):
            this_filter = i * n_plots + j

            # 画像がある限り実行(n_plotsが割り切れないためSprite Imageは少し余る)
            if this_filter < images.shape[0]:

                # Sprite Imageの所定の配列に画像を挿入
                spriteimage[i * img_h:(i + 1) * img_h, j * img_w:(j + 1) * img_w] = images[this_filter]

    return spriteimage

SPRITES="sprite.png"

# 画像をベクトルから配列に変換

#to_visualise = (1-np.reshape(x_test,(-1,28,28)))*255
#to_visualise = (1 - np.reshape(result[0],(-1,28,28)))*255
x = result[0]
xmean = x.mean(keepdims=True)
xstd  = np.std(x, keepdims=True)
to_visualise = (x-xmean)/xstd
to_visualise = to_visualise*100+128
to_visualise = 255-to_visualise
to_visualise = np.reshape(to_visualise,(-1,28,28))
# Sprite Image生成
sprite_image = create_sprite_image(to_visualise)

import matplotlib.pyplot as plt
# Sprite Image保存
plt.imsave(os.path.join(LOG_DIR,SPRITES), sprite_image,cmap='gray')

# Sprite Imageパスと設定
embedding.sprite.image_path=SPRITES
embedding.sprite.single_image_dim.extend([28,28])

# Projectorに出力
projector.visualize_embeddings(LOG_DIR, config)

全コードは下記

import tensorflow
# one-hot encodingを施すためのメソッド
from tensorflow.keras.utils import to_categorical

# 必要なライブラリのインポート
from tensorflow.keras.datasets import mnist
import numpy as np
#import pandas as pd
#import sklearn

from tensorflow.keras.layers import Lambda, Input, Dense
from tensorflow.keras.models import Model
from tensorflow.keras.losses import mse
from tensorflow.keras import backend as K

import datetime
LOG_DIR = f"./tflog/{datetime.datetime.now().strftime('%Y-%m-%d-%H%M%S')}"
tensorflow.debugging.experimental.enable_dump_debug_info(LOG_DIR, tensor_debug_mode="FULL_HEALTH", circular_buffer_size=-1)

from tensorflow.python.compiler.mlcompute import mlcompute
mlcompute.set_mlc_device(device_name = 'gpu')

#Kerasの関数でデータの読み込み。データをシャッフルして学習データと訓練データに分割
(x_train, y_train), (x_test, y_test) = mnist.load_data()

# 2次元データを数値に変換
x_train = x_train.reshape(60000, 784)
x_test = x_test.reshape(10000, 784)
# 型変換
x_train = x_train.astype('float32')
x_test = x_test.astype('float32')
# 255で割ったものを新たに変数とする
x_train /= 255
x_test /= 255

# クラス数は10
num_classes = 10
feature_dim = 10

y_train = y_train.astype('int32')
y_test = y_test.astype('int32')
labels = y_test
# one-hot encoding
y_train = to_categorical(y_train, num_classes)
y_test =  to_categorical(y_test, num_classes)

original_dim = x_train.shape[1]

input_shape = (original_dim, )
latent_dim = feature_dim

# Reparametrization Trick 
def sampling(args):
    z_mean, z_logvar = args
    batch = K.shape(z_mean)[0]
    dim = K.int_shape(z_mean)[1]
    epsilon = K.random_normal(shape=(batch, dim), seed = 5) # ε
    return z_mean + K.exp(0.5 * z_logvar) * epsilon

# VAEモデル構築
inputs = Input(shape=input_shape)
x1 = Dense(256, activation='relu')(inputs)  
x2 = Dense(64, activation='relu')(x1) 
z_mean = Dense(latent_dim)(x2)
z_logvar = Dense(latent_dim)(x2)
z = Lambda(sampling, output_shape=(latent_dim,))([z_mean, z_logvar])
encoder = Model(inputs, [z_mean, z_logvar, z], name='encoder')
encoder.summary()

latent_inputs = Input(shape=(latent_dim,))
x3 = Dense(64, activation='relu')(latent_inputs)  
x4 = Dense(256, activation='relu')(x3)  
outputs = Dense(original_dim, activation='sigmoid')(x4)
decoder = Model(latent_inputs, outputs, name='decoder')
decoder.summary()

z_output = encoder(inputs)[2]
outputs = [decoder(z_output),z_output]
vae = Model(inputs, outputs, name='variational_autoencoder')

# 損失関数
# Kullback-Leibler Loss
kl_loss = 1 + z_logvar - K.square(z_mean) - K.exp(z_logvar)
kl_loss = K.sum(kl_loss, axis=-1)
kl_loss *= -0.5
# Reconstruction Loss
reconstruction_loss = mse(inputs, outputs[0])
reconstruction_loss *= original_dim

vae_loss = K.mean(reconstruction_loss + kl_loss)
vae.add_loss(vae_loss)
vae.compile(optimizer='adam')
vae.summary()

# メタデータ作成・保存
import os
import numpy as np
import datetime
LOG_DIR = f"./tflog/{datetime.datetime.now().strftime('%Y-%m-%d-%H%M%S')}"
os.makedirs(os.path.join(LOG_DIR), exist_ok=True)
with open(os.path.join(LOG＿DIR, 'metadata.tsv'), 'w') as f:
    f.write("Index\tLabel\n")
    for index,label in enumerate(np.where(y_test)[1]):
        f.write("%d\t%d\n" % (index,label))

tb_cb = tensorflow.keras.callbacks.TensorBoard(log_dir= LOG_DIR,
    histogram_freq=1)
cbks = [tb_cb]

history = vae.fit(x_train,
                epochs=20,
                batch_size=512,
                validation_data=(x_test, None),
                callbacks=cbks)

result = vae.predict(x_test)

emb = result[1]

# embeddingsの作成
embedding_var = tensorflow.Variable(emb,  name='mnist_embedding')

CHECKPOINT_FILE = LOG_DIR + '/model.ckpt'
ckpt = tensorflow.train.Checkpoint(embeddings=embedding_var)
ckpt.save(CHECKPOINT_FILE)

from tensorboard.plugins import projector
# Projector設定
config = projector.ProjectorConfig()
embedding = config.embeddings.add()
embedding.tensor_name = "embeddings/.ATTRIBUTES/VARIABLE_VALUE"
# メタデータ(CSV)パス
embedding.metadata_path='metadata.tsv'

def create_sprite_image(images):
    """Returns a sprite image consisting of images passed as argument. Images should be count x width x height"""
    # MNISTは28ピクセル四方 
    img_h = images.shape[1]
    img_w = images.shape[2]

    # 画像数の平方根(切上)を計算(Sprite Imageの1辺の長さに使用)
    n_plots = int(np.ceil(np.sqrt(images.shape[0])))

    # 全要素0の配列作成
    spriteimage = np.ones((img_h * n_plots ,img_w * n_plots ))

    for i in range(n_plots):
        for j in range(n_plots):
            this_filter = i * n_plots + j

            # 画像がある限り実行(n_plotsが割り切れないためSprite Imageは少し余る)
            if this_filter < images.shape[0]:

                # Sprite Imageの所定の配列に画像を挿入
                spriteimage[i * img_h:(i + 1) * img_h, j * img_w:(j + 1) * img_w] = images[this_filter]

    return spriteimage

SPRITES="sprite.png"

# 画像をベクトルから配列に変換

#to_visualise = (1-np.reshape(x_test,(-1,28,28)))*255
#to_visualise = (1 - np.reshape(result[0],(-1,28,28)))*255
x = result[0]
xmean = x.mean(keepdims=True)
xstd  = np.std(x, keepdims=True)
to_visualise = (x-xmean)/xstd
to_visualise = to_visualise*100+128
to_visualise = 255-to_visualise
to_visualise = np.reshape(to_visualise,(-1,28,28))
# Sprite Image生成
sprite_image = create_sprite_image(to_visualise)

import matplotlib.pyplot as plt
# Sprite Image保存
plt.imsave(os.path.join(LOG_DIR,SPRITES), sprite_image,cmap='gray')

# Sprite Imageパスと設定
embedding.sprite.image_path=SPRITES
embedding.sprite.single_image_dim.extend([28,28])

# Projectorに出力
projector.visualize_embeddings(LOG_DIR, config)