 TensorFlow.orgで表示 |  GoogleColabで実行 |  GitHubでソースを表示 |  ノートブックをダウンロード |
このチュートリアルでは、Isola et al。による条件付き敵対的ネットワークを使用した画像から画像への変換で説明されているように、入力画像から出力画像へのマッピングを学習するpix2pixと呼ばれる条件付き生成敵対的ネットワーク（cGAN）を構築およびトレーニングする方法を示します。 （2017）。 pix2pixはアプリケーション固有ではありません。ラベルマップからの写真の合成、白黒画像からのカラー写真の生成、Googleマップの写真の空中画像への変換、スケッチの写真への変換など、さまざまなタスクに適用できます。
この例では、ネットワークは、プラハのチェコ工科大学の機械知覚センターが提供するCMPファサードデータベースを使用して、建物のファサードの画像を生成します。短くするために、pix2pixの作成者によって作成されたこのデータセットの前処理されたコピーを使用します。
pix2pix cGANでは、入力画像を条件付けて、対応する出力画像を生成します。 cGANは、条件付き生成的敵対的ネットで最初に提案されました（Mirza and Osindero、2014）
ネットワークのアーキテクチャには次のものが含まれます。
- U-Netベースのアーキテクチャを備えたジェネレータ。
- 畳み込みPatchGAN分類器（ pix2pixペーパーで提案されている）で表される弁別子。
各エポックは、単一のV100GPUで約15秒かかる場合があることに注意してください。
以下は、ファサードデータセット（80kステップ）で200エポックのトレーニング後にpix2pixcGANによって生成された出力の例です。
TensorFlowおよびその他のライブラリをインポートする
import tensorflow as tf
import os
import pathlib
import time
import datetime
from matplotlib import pyplot as plt
from IPython import display
データセットをロードする
CMPファサードデータベースデータ（30MB）をダウンロードします。追加のデータセットは、ここで同じ形式で利用できます。 Colabでは、ドロップダウンメニューから他のデータセットを選択できます。他のデータセットのいくつかはかなり大きいことに注意してください（ edges2handbagsは8GBです）。
dataset_name = "facades"
_URL = f'http://efrosgans.eecs.berkeley.edu/pix2pix/datasets/{dataset_name}.tar.gz'
path_to_zip = tf.keras.utils.get_file(
fname=f"{dataset_name}.tar.gz",
origin=_URL,
extract=True)
path_to_zip = pathlib.Path(path_to_zip)
PATH = path_to_zip.parent/dataset_name
Downloading data from http://efrosgans.eecs.berkeley.edu/pix2pix/datasets/facades.tar.gz 30171136/30168306 [==============================] - 19s 1us/step 30179328/30168306 [==============================] - 19s 1us/step
list(PATH.parent.iterdir())
[PosixPath('/home/kbuilder/.keras/datasets/facades.tar.gz'),
PosixPath('/home/kbuilder/.keras/datasets/YellowLabradorLooking_new.jpg'),
PosixPath('/home/kbuilder/.keras/datasets/facades'),
PosixPath('/home/kbuilder/.keras/datasets/mnist.npz')]
元の各画像のサイズは256 x 512で、2つの256 x 256画像が含まれています。
sample_image = tf.io.read_file(str(PATH / 'train/1.jpg'))
sample_image = tf.io.decode_jpeg(sample_image)
print(sample_image.shape)
(256, 512, 3)
plt.figure()
plt.imshow(sample_image)
<matplotlib.image.AxesImage at 0x7f35a3653c90>
実際の建物のファサード画像を建築ラベル画像から分離する必要があります。これらの画像はすべて256 x 256のサイズになります。
画像ファイルをロードし、2つの画像テンソルを出力する関数を定義します。
def load(image_file):
# Read and decode an image file to a uint8 tensor
image = tf.io.read_file(image_file)
image = tf.io.decode_jpeg(image)
# Split each image tensor into two tensors:
# - one with a real building facade image
# - one with an architecture label image
w = tf.shape(image)[1]
w = w // 2
input_image = image[:, w:, :]
real_image = image[:, :w, :]
# Convert both images to float32 tensors
input_image = tf.cast(input_image, tf.float32)
real_image = tf.cast(real_image, tf.float32)
return input_image, real_image
入力（建築ラベル画像）と実際の（建物の正面写真）画像のサンプルをプロットします。
inp, re = load(str(PATH / 'train/100.jpg'))
# Casting to int for matplotlib to display the images
plt.figure()
plt.imshow(inp / 255.0)
plt.figure()
plt.imshow(re / 255.0)
<matplotlib.image.AxesImage at 0x7f35981a4910>
pix2pixの論文で説明されているように、トレーニングセットを前処理するには、ランダムなジッターとミラーリングを適用する必要があります。
次のようないくつかの関数を定義します。
- 各
256 x 256画像のサイズをより大きな高さと幅（286 x 286）に変更します。 - ランダムにトリミングして
256 x 256に戻します。 - 画像を水平方向にランダムに反転します。つまり、左から右に反転します（ランダムミラーリング）。
- 画像を
[-1, 1]の範囲に正規化します。
# The facade training set consist of 400 images
BUFFER_SIZE = 400
# The batch size of 1 produced better results for the U-Net in the original pix2pix experiment
BATCH_SIZE = 1
# Each image is 256x256 in size
IMG_WIDTH = 256
IMG_HEIGHT = 256
def resize(input_image, real_image, height, width):
input_image = tf.image.resize(input_image, [height, width],
method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)
real_image = tf.image.resize(real_image, [height, width],
method=tf.image.ResizeMethod.NEAREST_NEIGHBOR)
return input_image, real_image
def random_crop(input_image, real_image):
stacked_image = tf.stack([input_image, real_image], axis=0)
cropped_image = tf.image.random_crop(
stacked_image, size=[2, IMG_HEIGHT, IMG_WIDTH, 3])
return cropped_image[0], cropped_image[1]
# Normalizing the images to [-1, 1]
def normalize(input_image, real_image):
input_image = (input_image / 127.5) - 1
real_image = (real_image / 127.5) - 1
return input_image, real_image
@tf.function()
def random_jitter(input_image, real_image):
# Resizing to 286x286
input_image, real_image = resize(input_image, real_image, 286, 286)
# Random cropping back to 256x256
input_image, real_image = random_crop(input_image, real_image)
if tf.random.uniform(()) > 0.5:
# Random mirroring
input_image = tf.image.flip_left_right(input_image)
real_image = tf.image.flip_left_right(real_image)
return input_image, real_image
前処理された出力の一部を検査できます。
plt.figure(figsize=(6, 6))
for i in range(4):
rj_inp, rj_re = random_jitter(inp, re)
plt.subplot(2, 2, i + 1)
plt.imshow(rj_inp / 255.0)
plt.axis('off')
plt.show()
ロードと前処理が機能することを確認したら、トレーニングセットとテストセットをロードして前処理するいくつかのヘルパー関数を定義しましょう。
def load_image_train(image_file):
input_image, real_image = load(image_file)
input_image, real_image = random_jitter(input_image, real_image)
input_image, real_image = normalize(input_image, real_image)
return input_image, real_image
def load_image_test(image_file):
input_image, real_image = load(image_file)
input_image, real_image = resize(input_image, real_image,
IMG_HEIGHT, IMG_WIDTH)
input_image, real_image = normalize(input_image, real_image)
return input_image, real_image
tf.dataを使用して入力パイプラインを構築します
train_dataset = tf.data.Dataset.list_files(str(PATH / 'train/*.jpg'))
train_dataset = train_dataset.map(load_image_train,
num_parallel_calls=tf.data.AUTOTUNE)
train_dataset = train_dataset.shuffle(BUFFER_SIZE)
train_dataset = train_dataset.batch(BATCH_SIZE)
try:
test_dataset = tf.data.Dataset.list_files(str(PATH / 'test/*.jpg'))
except tf.errors.InvalidArgumentError:
test_dataset = tf.data.Dataset.list_files(str(PATH / 'val/*.jpg'))
test_dataset = test_dataset.map(load_image_test)
test_dataset = test_dataset.batch(BATCH_SIZE)
ジェネレーターを構築する
pix2pix cGANのジェネレーターは、変更されたU-Netです。 U-Netは、エンコーダー（ダウンサンプラー）とデコーダー（アップサンプラー）で構成されます。 （詳細については、画像セグメンテーションチュートリアルおよびU-NetプロジェクトのWebサイトを参照してください。）
- エンコーダーの各ブロックは次のとおりです。畳み込み->バッチ正規化-> Leaky ReLU
- デコーダーの各ブロックは次のとおりです。転置畳み込み->バッチ正規化->ドロップアウト（最初の3ブロックに適用）-> ReLU
- エンコーダーとデコーダーの間にはスキップ接続があります（U-Netのように）。
ダウンサンプラー（エンコーダー）を定義します。
OUTPUT_CHANNELS = 3
def downsample(filters, size, apply_batchnorm=True):
initializer = tf.random_normal_initializer(0., 0.02)
result = tf.keras.Sequential()
result.add(
tf.keras.layers.Conv2D(filters, size, strides=2, padding='same',
kernel_initializer=initializer, use_bias=False))
if apply_batchnorm:
result.add(tf.keras.layers.BatchNormalization())
result.add(tf.keras.layers.LeakyReLU())
return result
down_model = downsample(3, 4)
down_result = down_model(tf.expand_dims(inp, 0))
print (down_result.shape)
(1, 128, 128, 3)
アップサンプラー（デコーダー）を定義します。
def upsample(filters, size, apply_dropout=False):
initializer = tf.random_normal_initializer(0., 0.02)
result = tf.keras.Sequential()
result.add(
tf.keras.layers.Conv2DTranspose(filters, size, strides=2,
padding='same',
kernel_initializer=initializer,
use_bias=False))
result.add(tf.keras.layers.BatchNormalization())
if apply_dropout:
result.add(tf.keras.layers.Dropout(0.5))
result.add(tf.keras.layers.ReLU())
return result
up_model = upsample(3, 4)
up_result = up_model(down_result)
print (up_result.shape)
(1, 256, 256, 3)
ダウンサンプラーとアップサンプラーを使用してジェネレーターを定義します。
def Generator():
inputs = tf.keras.layers.Input(shape=[256, 256, 3])
down_stack = [
downsample(64, 4, apply_batchnorm=False), # (batch_size, 128, 128, 64)
downsample(128, 4), # (batch_size, 64, 64, 128)
downsample(256, 4), # (batch_size, 32, 32, 256)
downsample(512, 4), # (batch_size, 16, 16, 512)
downsample(512, 4), # (batch_size, 8, 8, 512)
downsample(512, 4), # (batch_size, 4, 4, 512)
downsample(512, 4), # (batch_size, 2, 2, 512)
downsample(512, 4), # (batch_size, 1, 1, 512)
]
up_stack = [
upsample(512, 4, apply_dropout=True), # (batch_size, 2, 2, 1024)
upsample(512, 4, apply_dropout=True), # (batch_size, 4, 4, 1024)
upsample(512, 4, apply_dropout=True), # (batch_size, 8, 8, 1024)
upsample(512, 4), # (batch_size, 16, 16, 1024)
upsample(256, 4), # (batch_size, 32, 32, 512)
upsample(128, 4), # (batch_size, 64, 64, 256)
upsample(64, 4), # (batch_size, 128, 128, 128)
]
initializer = tf.random_normal_initializer(0., 0.02)
last = tf.keras.layers.Conv2DTranspose(OUTPUT_CHANNELS, 4,
strides=2,
padding='same',
kernel_initializer=initializer,
activation='tanh') # (batch_size, 256, 256, 3)
x = inputs
# Downsampling through the model
skips = []
for down in down_stack:
x = down(x)
skips.append(x)
skips = reversed(skips[:-1])
# Upsampling and establishing the skip connections
for up, skip in zip(up_stack, skips):
x = up(x)
x = tf.keras.layers.Concatenate()([x, skip])
x = last(x)
return tf.keras.Model(inputs=inputs, outputs=x)
ジェネレータモデルアーキテクチャを視覚化します。
generator = Generator()
tf.keras.utils.plot_model(generator, show_shapes=True, dpi=64)
ジェネレーターをテストします。
gen_output = generator(inp[tf.newaxis, ...], training=False)
plt.imshow(gen_output[0, ...])
Clipping input data to the valid range for imshow with RGB data ([0..1] for floats or [0..255] for integers). <matplotlib.image.AxesImage at 0x7f35cfd20610>
発電機の損失を定義する
GANはデータに適応する損失を学習しますが、cGANは、 pix2pixの論文で説明されているように、ネットワーク出力やターゲット画像とは異なる可能性のある構造にペナルティを課す構造化損失を学習します。
- ジェネレータ損失は、生成された画像と画像の配列のシグモイドクロスエントロピー損失です。
- pix2pixの論文では、生成された画像とターゲット画像の間のMAE（平均絶対誤差）であるL1損失についても言及されています。
- これにより、生成された画像をターゲット画像と構造的に類似させることができます。
- 総発電機損失を計算する式は、
gan_loss + LAMBDA * l1_lossです。ここで、LAMBDA = 100です。この値は、論文の著者によって決定されました。
LAMBDA = 100
loss_object = tf.keras.losses.BinaryCrossentropy(from_logits=True)
def generator_loss(disc_generated_output, gen_output, target):
gan_loss = loss_object(tf.ones_like(disc_generated_output), disc_generated_output)
# Mean absolute error
l1_loss = tf.reduce_mean(tf.abs(target - gen_output))
total_gen_loss = gan_loss + (LAMBDA * l1_loss)
return total_gen_loss, gan_loss, l1_loss
ジェネレータのトレーニング手順は次のとおりです。
弁別器を作成する
pix2pix cGANの識別子は、畳み込みPatchGAN分類器です。pix2pixの論文で説明されているように、各画像パッチが本物かどうかを分類しようとします。
- 弁別器の各ブロックは、畳み込み->バッチ正規化-> LeakyReLUです。
- 最後のレイヤーの後の出力の形状は
(batch_size, 30, 30, 1)です。 - 出力の各
30 x 30画像パッチは、入力画像の70 x 70部分を分類します。 - 弁別器は2つの入力を受け取ります。
- 入力画像とターゲット画像。実際に分類する必要があります。
- 入力画像と生成された画像（ジェネレータの出力）。これらは偽物として分類されます。
-
tf.concat([inp, tar], axis=-1)を使用して、これら2つの入力を連結します。
弁別子を定義しましょう：
def Discriminator():
initializer = tf.random_normal_initializer(0., 0.02)
inp = tf.keras.layers.Input(shape=[256, 256, 3], name='input_image')
tar = tf.keras.layers.Input(shape=[256, 256, 3], name='target_image')
x = tf.keras.layers.concatenate([inp, tar]) # (batch_size, 256, 256, channels*2)
down1 = downsample(64, 4, False)(x) # (batch_size, 128, 128, 64)
down2 = downsample(128, 4)(down1) # (batch_size, 64, 64, 128)
down3 = downsample(256, 4)(down2) # (batch_size, 32, 32, 256)
zero_pad1 = tf.keras.layers.ZeroPadding2D()(down3) # (batch_size, 34, 34, 256)
conv = tf.keras.layers.Conv2D(512, 4, strides=1,
kernel_initializer=initializer,
use_bias=False)(zero_pad1) # (batch_size, 31, 31, 512)
batchnorm1 = tf.keras.layers.BatchNormalization()(conv)
leaky_relu = tf.keras.layers.LeakyReLU()(batchnorm1)
zero_pad2 = tf.keras.layers.ZeroPadding2D()(leaky_relu) # (batch_size, 33, 33, 512)
last = tf.keras.layers.Conv2D(1, 4, strides=1,
kernel_initializer=initializer)(zero_pad2) # (batch_size, 30, 30, 1)
return tf.keras.Model(inputs=[inp, tar], outputs=last)
弁別器モデルのアーキテクチャを視覚化します。
discriminator = Discriminator()
tf.keras.utils.plot_model(discriminator, show_shapes=True, dpi=64)
弁別器をテストします。
disc_out = discriminator([inp[tf.newaxis, ...], gen_output], training=False)
plt.imshow(disc_out[0, ..., -1], vmin=-20, vmax=20, cmap='RdBu_r')
plt.colorbar()
<matplotlib.colorbar.Colorbar at 0x7f35cec82c50>
弁別器の損失を定義する
discriminator_loss関数は、実画像と生成された画像の2つの入力を受け取ります。-
real_lossは、実像とその配列のシグモイドクロスエントロピー損失です（これらは実像であるため） 。 -
generated_lossは、生成された画像とゼロの配列のシグモイドクロスエントロピー損失です（これらは偽の画像であるため） 。 -
total_lossは、real_lossとgenerated_lossの合計です。
def discriminator_loss(disc_real_output, disc_generated_output):
real_loss = loss_object(tf.ones_like(disc_real_output), disc_real_output)
generated_loss = loss_object(tf.zeros_like(disc_generated_output), disc_generated_output)
total_disc_loss = real_loss + generated_loss
return total_disc_loss
弁別器の訓練手順を以下に示します。
アーキテクチャとハイパーパラメータの詳細については、 pix2pixの論文を参照してください。
オプティマイザーとチェックポイントセーバーを定義する
generator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)
discriminator_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5)
checkpoint_dir = './training_checkpoints'
checkpoint_prefix = os.path.join(checkpoint_dir, "ckpt")
checkpoint = tf.train.Checkpoint(generator_optimizer=generator_optimizer,
discriminator_optimizer=discriminator_optimizer,
generator=generator,
discriminator=discriminator)
画像を生成する
トレーニング中にいくつかの画像をプロットする関数を記述します。
- テストセットからジェネレータに画像を渡します。
- 次に、ジェネレータは入力画像を出力に変換します。
- 最後のステップは、予測と出来上がりをプロットすることです！
def generate_images(model, test_input, tar):
prediction = model(test_input, training=True)
plt.figure(figsize=(15, 15))
display_list = [test_input[0], tar[0], prediction[0]]
title = ['Input Image', 'Ground Truth', 'Predicted Image']
for i in range(3):
plt.subplot(1, 3, i+1)
plt.title(title[i])
# Getting the pixel values in the [0, 1] range to plot.
plt.imshow(display_list[i] * 0.5 + 0.5)
plt.axis('off')
plt.show()
関数をテストします。
for example_input, example_target in test_dataset.take(1):
generate_images(generator, example_input, example_target)
トレーニング
- 各例について、入力は出力を生成します。
- 弁別器は、
input_imageと生成された画像を最初の入力として受け取ります。 2番目の入力はinput_imageとtarget_imageです。 - 次に、ジェネレータとディスクリミネータの損失を計算します。
- 次に、ジェネレーターとディスクリミネーターの両方の変数（入力）に関する損失の勾配を計算し、それらをオプティマイザーに適用します。
- 最後に、損失をTensorBoardに記録します。
log_dir="logs/"
summary_writer = tf.summary.create_file_writer(
log_dir + "fit/" + datetime.datetime.now().strftime("%Y%m%d-%H%M%S"))
@tf.function
def train_step(input_image, target, step):
with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:
gen_output = generator(input_image, training=True)
disc_real_output = discriminator([input_image, target], training=True)
disc_generated_output = discriminator([input_image, gen_output], training=True)
gen_total_loss, gen_gan_loss, gen_l1_loss = generator_loss(disc_generated_output, gen_output, target)
disc_loss = discriminator_loss(disc_real_output, disc_generated_output)
generator_gradients = gen_tape.gradient(gen_total_loss,
generator.trainable_variables)
discriminator_gradients = disc_tape.gradient(disc_loss,
discriminator.trainable_variables)
generator_optimizer.apply_gradients(zip(generator_gradients,
generator.trainable_variables))
discriminator_optimizer.apply_gradients(zip(discriminator_gradients,
discriminator.trainable_variables))
with summary_writer.as_default():
tf.summary.scalar('gen_total_loss', gen_total_loss, step=step//1000)
tf.summary.scalar('gen_gan_loss', gen_gan_loss, step=step//1000)
tf.summary.scalar('gen_l1_loss', gen_l1_loss, step=step//1000)
tf.summary.scalar('disc_loss', disc_loss, step=step//1000)
実際のトレーニングループ。このチュートリアルは複数のデータセットで実行でき、データセットのサイズは大きく異なるため、トレーニングループはエポックではなく段階的に機能するように設定されています。
- ステップ数を繰り返します。
- 10ステップごとにドット（
.）を出力します。 - 1kステップごと：表示をクリアし、
generate_imagesを実行して進行状況を表示します。 - 5kステップごと：チェックポイントを保存します。
def fit(train_ds, test_ds, steps):
example_input, example_target = next(iter(test_ds.take(1)))
start = time.time()
for step, (input_image, target) in train_ds.repeat().take(steps).enumerate():
if (step) % 1000 == 0:
display.clear_output(wait=True)
if step != 0:
print(f'Time taken for 1000 steps: {time.time()-start:.2f} sec\n')
start = time.time()
generate_images(generator, example_input, example_target)
print(f"Step: {step//1000}k")
train_step(input_image, target, step)
# Training step
if (step+1) % 10 == 0:
print('.', end='', flush=True)
# Save (checkpoint) the model every 5k steps
if (step + 1) % 5000 == 0:
checkpoint.save(file_prefix=checkpoint_prefix)
このトレーニングループは、トレーニングの進行状況を監視するためにTensorBoardで表示できるログを保存します。
ローカルマシンで作業する場合は、別のTensorBoardプロセスを起動します。ノートブックで作業する場合は、TensorBoardで監視するためのトレーニングを開始する前に、ビューアを起動してください。
ビューアを起動するには、以下をコードセルに貼り付けます。
%load_ext tensorboard
%tensorboard --logdir {log_dir}
最後に、トレーニングループを実行します。
fit(train_dataset, test_dataset, steps=40000)
Time taken for 1000 steps: 36.53 sec
Step: 39k ....................................................................................................
TensorBoardの結果を公開して共有する場合は、以下をコードセルにコピーして、ログをTensorBoard.devにアップロードできます。
tensorboard dev upload --logdir {log_dir}
このノートブックの前回の実行結果は、 TensorBoard.devで確認できます。
TensorBoard.devは、ML実験をホスト、追跡、およびすべての人と共有するための管理されたエクスペリエンスです。
<iframe>を使用してインラインで含めることもできます：
display.IFrame(
src="https://tensorboard.dev/experiment/lZ0C6FONROaUMfjYkVyJqw",
width="100%",
height="1000px")
単純な分類または回帰モデルと比較して、GAN（またはpix2pixのようなcGAN）をトレーニングする場合、ログの解釈はより微妙です。探すべきもの：
- ジェネレーターモデルもディスクリミネーターモデルも「勝った」ことを確認してください。
gen_gan_lossまたはdisc_lossのいずれかが非常に低くなった場合は、このモデルがもう一方を支配していることを示しており、結合されたモデルのトレーニングに成功していません。 - 値
log(2) = 0.69は、2の困惑を示しているため、これらの損失の適切な基準点です。弁別器は、平均して、2つのオプションについて等しく不確実です。 -
disc_lossの場合、0.69未満の値は、実際の画像と生成された画像の組み合わせセットで、ディスクリミネーターがランダムよりも優れていることを意味します。 -
gen_gan_lossの場合、0.69未満の値は、ジェネレーターがランダムよりもディスクリミネーターをだましていることを意味します。 - トレーニングが進むにつれて、
gen_l1_lossは下がるはずです。
最新のチェックポイントを復元し、ネットワークをテストします
ls {checkpoint_dir}
checkpoint ckpt-5.data-00000-of-00001 ckpt-1.data-00000-of-00001 ckpt-5.index ckpt-1.index ckpt-6.data-00000-of-00001 ckpt-2.data-00000-of-00001 ckpt-6.index ckpt-2.index ckpt-7.data-00000-of-00001 ckpt-3.data-00000-of-00001 ckpt-7.index ckpt-3.index ckpt-8.data-00000-of-00001 ckpt-4.data-00000-of-00001 ckpt-8.index ckpt-4.index
# Restoring the latest checkpoint in checkpoint_dir
checkpoint.restore(tf.train.latest_checkpoint(checkpoint_dir))
<tensorflow.python.training.tracking.util.CheckpointLoadStatus at 0x7f35cfd6b8d0>
テストセットを使用していくつかの画像を生成します
# Run the trained model on a few examples from the test set
for inp, tar in test_dataset.take(5):
generate_images(generator, inp, tar)
