이 페이지는 Cloud Translation API를 통해 번역되었습니다.
Switch to English

Pix2Pix

TensorFlow.org에서보기 Google Colab에서 실행 GitHub에서 소스보기 노트북 다운로드

이 노트북은 조건부 적대 네트워크를 사용한 이미지-이미지 변환에 설명 된대로 조건부 GAN을 사용하여 이미지를 이미지로 변환하는 방법을 보여줍니다. 이 기술을 사용하여 흑백 사진에 색을 입히고 Google지도를 Google 어스로 변환하는 등의 작업을 수행 할 수 있습니다. 여기에서 건물 외관을 실제 건물로 변환합니다.

예를 들어, 프라하 에있는 체코 기술 대학기계 인식 센터 에서 유용하게 제공하는 CMP Facade Database를 사용할 것입니다. 예제를 짧게 유지하기 위해 위 논문 의 저자가 만든이 데이터 세트의 전처리 된 사본 을 사용합니다.

각 Epoch는 단일 V100 GPU에서 약 15 초가 걸립니다.

다음은 200 epoch 동안 모델을 학습 한 후 생성 된 출력입니다.

샘플 output_1샘플 output_2

TensorFlow 및 기타 라이브러리 가져 오기

import tensorflow as tf

import os
import time

from matplotlib import pyplot as plt
from IPython import display
pip install -q -U tensorboard
WARNING: You are using pip version 20.2.2; however, version 20.2.3 is available.
You should consider upgrading via the '/tmpfs/src/tf_docs_env/bin/python -m pip install --upgrade pip' command.

데이터 세트로드

여기 에서이 데이터 세트와 유사한 데이터 세트를 다운로드 할 수 있습니다. 백서 에서 언급했듯이 훈련 데이터 세트에 무작위 지 터링 및 미러링을 적용합니다.

  • 임의 지 터링에서 이미지 크기가 286 x 286 조정 된 다음 무작위로 256 x 256 잘립니다.
  • 랜덤 미러링에서는 이미지가 좌우로 무작위로 뒤집 힙니다.
_URL = 'https://people.eecs.berkeley.edu/~tinghuiz/projects/pix2pix/datasets/facades.tar.gz'

path_to_zip = tf.keras.utils.get_file('facades.tar.gz',
                                      origin=_URL,
                                      extract=True)

PATH = os.path.join(os.path.dirname(path_to_zip), 'facades/')
Downloading data from https://people.eecs.berkeley.edu/~tinghuiz/projects/pix2pix/datasets/facades.tar.gz
30171136/30168306 [==============================] - 2s 0us/step

BUFFER_SIZE = 400
BATCH_SIZE = 1
IMG_WIDTH = 256
IMG_HEIGHT = 256
def load(image_file):
  image = tf.io.read_file(image_file)
  image = tf.image.decode_jpeg(image)

  w = tf.shape(image)[1]

  w = w // 2
  real_image = image[:, :w, :]
  input_image = image[:, w:, :]

  input_image = tf.cast(input_image, tf.float32)
  real_image = tf.cast(real_image, tf.float32)

  return input_image, real_image
inp, re = load(PATH+'train/100.jpg')
# casting to int for matplotlib to show the image
plt.figure()
plt.imshow(inp/255.0)
plt.figure()
plt.imshow(re/255.0)
<matplotlib.image.AxesImage at 0x7f5576b28550>

png

png

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 286 x 286 x 3
  input_image, real_image = resize(input_image, real_image, 286, 286)

  # randomly cropping to 256 x 256 x 3
  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

아래 이미지에서 볼 수 있듯이 무작위 지 터링을 겪고 있음을 알 수 있습니다.

  1. 더 큰 높이와 너비로 이미지 크기 조정
  2. 대상 크기로 무작위로 자르기
  3. 이미지를 가로로 무작위로 뒤집기
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()

png

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

입력 파이프 라인

train_dataset = tf.data.Dataset.list_files(PATH+'train/*.jpg')
train_dataset = train_dataset.map(load_image_train,
                                  num_parallel_calls=tf.data.experimental.AUTOTUNE)
train_dataset = train_dataset.shuffle(BUFFER_SIZE)
train_dataset = train_dataset.batch(BATCH_SIZE)
test_dataset = tf.data.Dataset.list_files(PATH+'test/*.jpg')
test_dataset = test_dataset.map(load_image_test)
test_dataset = test_dataset.batch(BATCH_SIZE)

생성기 구축

  • 제너레이터의 아키텍처는 수정 된 U-Net입니다.
  • 인코더의 각 블록은 (Conv-> Batchnorm-> Leaky ReLU)입니다.
  • 디코더의 각 블록은 (Transposed Conv-> Batchnorm-> Dropout (처음 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), # (bs, 128, 128, 64)
    downsample(128, 4), # (bs, 64, 64, 128)
    downsample(256, 4), # (bs, 32, 32, 256)
    downsample(512, 4), # (bs, 16, 16, 512)
    downsample(512, 4), # (bs, 8, 8, 512)
    downsample(512, 4), # (bs, 4, 4, 512)
    downsample(512, 4), # (bs, 2, 2, 512)
    downsample(512, 4), # (bs, 1, 1, 512)
  ]

  up_stack = [
    upsample(512, 4, apply_dropout=True), # (bs, 2, 2, 1024)
    upsample(512, 4, apply_dropout=True), # (bs, 4, 4, 1024)
    upsample(512, 4, apply_dropout=True), # (bs, 8, 8, 1024)
    upsample(512, 4), # (bs, 16, 16, 1024)
    upsample(256, 4), # (bs, 32, 32, 512)
    upsample(128, 4), # (bs, 64, 64, 256)
    upsample(64, 4), # (bs, 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') # (bs, 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)

png

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 0x7f54c85167b8>

png

  • 발전기 손실
    • 생성 된 이미지와 이미지 배열의 시그 모이 드 교차 엔트로피 손실 입니다 .
    • 논문 에는 생성 된 이미지와 대상 이미지 사이의 MAE (평균 절대 오차) 인 L1 손실도 포함되어 있습니다.
    • 이렇게하면 생성 된 이미지가 대상 이미지와 구조적으로 유사 해집니다.
    • 총 발전기 손실을 계산하는 공식 = gan_loss + LAMBDA * l1_loss, 여기서 LAMBDA = 100.이 값은 논문 저자가 결정했습니다.

생성기에 대한 교육 절차는 다음과 같습니다.

LAMBDA = 100
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

발전기 업데이트 이미지

판별 자 구축

  • 판별자는 PatchGAN입니다.
  • 식별기의 각 블록은 (Conv-> BatchNorm-> Leaky ReLU)입니다.
  • 마지막 레이어 이후의 출력 모양은 (batch_size, 30, 30, 1)입니다.
  • 출력의 각 30x30 패치는 입력 이미지의 70x70 부분을 분류합니다 (이러한 아키텍처를 PatchGAN이라고 함).
  • 판별 기는 2 개의 입력을받습니다.
    • 실제로 분류해야하는 입력 이미지와 대상 이미지.
    • 입력 이미지와 생성 된 이미지 (생성기의 출력)를 가짜로 분류해야합니다.
    • 이 두 입력을 코드에서 함께 연결합니다 ( tf.concat([inp, tar], axis=-1) ).
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]) # (bs, 256, 256, channels*2)

  down1 = downsample(64, 4, False)(x) # (bs, 128, 128, 64)
  down2 = downsample(128, 4)(down1) # (bs, 64, 64, 128)
  down3 = downsample(256, 4)(down2) # (bs, 32, 32, 256)

  zero_pad1 = tf.keras.layers.ZeroPadding2D()(down3) # (bs, 34, 34, 256)
  conv = tf.keras.layers.Conv2D(512, 4, strides=1,
                                kernel_initializer=initializer,
                                use_bias=False)(zero_pad1) # (bs, 31, 31, 512)

  batchnorm1 = tf.keras.layers.BatchNormalization()(conv)

  leaky_relu = tf.keras.layers.LeakyReLU()(batchnorm1)

  zero_pad2 = tf.keras.layers.ZeroPadding2D()(leaky_relu) # (bs, 33, 33, 512)

  last = tf.keras.layers.Conv2D(1, 4, strides=1,
                                kernel_initializer=initializer)(zero_pad2) # (bs, 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)

png

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 0x7f54c83a3fd0>

png

판별 자 손실

  • 판별 기 손실 함수는 2 개의 입력을받습니다. 실제 이미지, 생성 된 이미지
  • real_loss는 실제 이미지이미지배열의 시그 모이 드 교차 엔트로피 손실입니다 ( 실제 이미지 이므로).
  • generated_loss는 생성 된 이미지0배열의 시그 모이 드 교차 엔트로피 손실입니다 (가짜 이미지이기 때문에)
  • 그러면 total_loss는 real_loss와 generated_loss의 합입니다.
loss_object = tf.keras.losses.BinaryCrossentropy(from_logits=True)
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

판별 자에 대한 교육 절차는 다음과 같습니다.

아키텍처 및 하이퍼 파라미터에 대한 자세한 내용은 문서를 참조하십시오.

판별 기 업데이트 이미지

최적화 도구 및 체크 포인트 보호기 정의

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 between [0, 1] to plot it.
    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)

png

훈련

  • 각 예제 입력에 대해 출력을 생성합니다.
  • 판별 기는 input_image와 생성 된 이미지를 첫 번째 입력으로받습니다. 두 번째 입력은 input_image 및 target_image입니다.
  • 다음으로 생성기와 판별 기 손실을 계산합니다.
  • 그런 다음 생성기 및 판별 변수 (입력)에 대한 손실의 기울기를 계산하고이를 최적화기에 적용합니다.
  • 그런 다음 손실을 TensorBoard에 기록합니다.
EPOCHS = 150
import datetime
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, epoch):
  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=epoch)
    tf.summary.scalar('gen_gan_loss', gen_gan_loss, step=epoch)
    tf.summary.scalar('gen_l1_loss', gen_l1_loss, step=epoch)
    tf.summary.scalar('disc_loss', disc_loss, step=epoch)

실제 훈련 루프 :

  • Epoch 수를 반복합니다.
  • 각 epoch에서 디스플레이를 지우고 generate_images 를 실행 generate_images 진행 상황을 표시합니다.
  • 각 에포크에서 학습 데이터 세트를 반복하여 '.'를 인쇄합니다. 각 예에 대해.
  • 20 epoch마다 체크 포인트를 저장합니다.
def fit(train_ds, epochs, test_ds):
  for epoch in range(epochs):
    start = time.time()

    display.clear_output(wait=True)

    for example_input, example_target in test_ds.take(1):
      generate_images(generator, example_input, example_target)
    print("Epoch: ", epoch)

    # Train
    for n, (input_image, target) in train_ds.enumerate():
      print('.', end='')
      if (n+1) % 100 == 0:
        print()
      train_step(input_image, target, epoch)
    print()

    # saving (checkpoint) the model every 20 epochs
    if (epoch + 1) % 20 == 0:
      checkpoint.save(file_prefix = checkpoint_prefix)

    print ('Time taken for epoch {} is {} sec\n'.format(epoch + 1,
                                                        time.time()-start))
  checkpoint.save(file_prefix = checkpoint_prefix)

이 훈련 루프는 TensorBoard에서 쉽게 볼 수있는 로그를 저장하여 훈련 진행 상황을 모니터링합니다. 로컬에서 작업하면 별도의 텐서 보드 프로세스를 시작합니다. 노트북에서 TensorBoard로 모니터링하려면 교육을 시작하기 전에 뷰어를 시작하는 것이 가장 쉽습니다.

뷰어를 시작하려면 다음을 코드 셀에 붙여 넣으십시오.

%load_ext tensorboard
%tensorboard --logdir {log_dir}

이제 훈련 루프를 실행하십시오.

fit(train_dataset, EPOCHS, test_dataset)

png

Epoch:  149
....................................................................................................
....................................................................................................
....................................................................................................
....................................................................................................

Time taken for epoch 150 is 16.14578342437744 sec


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의 로그를 해석하는 것은 단순한 분류 또는 회귀 모델보다 더 미묘합니다. 찾아야 할 사항 ::

  • 두 모델 모두 "승리"하지 않았는지 확인하십시오. 어느 경우 gen_gan_loss 또는 disc_loss 매우 낮은 얻는다는이 모델이 다른 사람을 지배하고, 성공적으로 결합 된 모델을 훈련하지 않는 것을 나타내는 인디케이터 (indicator)이다.
  • log(2) = 0.69 값은 2의 난이도를 나타내므로 이러한 손실에 대한 좋은 참조 포인트입니다. 판별자는 두 옵션에 대해 평균적으로 동일하게 불확실합니다.
  • disc_loss 의 경우 0.69 미만의 값은 실제 + 생성 된 이미지의 결합 된 세트에서 판별 disc_loss 무작위보다 더 잘하고 있음을 의미합니다.
  • gen_gan_loss 의 경우 0.69 미만의 값은 제너레이터 i가 구분자를 폴딩 할 때 무작위보다 더 잘 수행함을 의미합니다.
  • 훈련이 진행됨에 따라 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 0x7f4fce701160>

테스트 데이터 세트를 사용하여 생성

# Run the trained model on a few examples from the test dataset
for inp, tar in test_dataset.take(5):
  generate_images(generator, inp, tar)

png

png

png

png

png