GAN(생성적 대립 신경망)

GAN의 원리는 간단하다.
유명한 골키퍼 앞에서 패널티킥 연습을 엄청 해서 골을 넣기 시작했다면, 피널티킥 슛팅하는 기술이 금새 늘어나는 원리이다.
생성모델(슈터)과 판별모델(골키퍼) 두 모델이 대립적인 과정을 통해 동시에 훈련된다.
위조지폐를 만들려는 ‘위조지폐범’이 지폐를 위조하지 못하게 하고, 또 위조지폐와 진폐를 잘 가려내고 싶은 ‘조폐공사’가 있다.
조폐공사는 위폐를 감별하는 기술을 지폐에 적용해야 한다.
반대로 위조지폐범은 감별하는 기술에 걸리지 않는 위조지폐를 만들려고 기술을 발전 시킨다.
서로의 정보를 참고해서 기술이 발달되기 때문에 지폐를 제작하는 기술이 점점 좋아지게 된다.
이를 ‘인공지능 화가’에 적용하면, Creator/Generator는 진짜 작품처럼 보이는 이미지를 생성하도록 배우게 되고, 동시에 Descriminator는 인공지능이 만든 가짜의 이미지와 진짜를 예술작품을 구별하게 되는 것을 배우게 되는 원리이다.

훈련과정동안 Generator는 점차 실제같은 이미지를 더 잘 생성하게 되고, Descriminator는 점차 진짜와 가짜를 더 잘 구별하게 된다.
이 과정은 Descriminator가 가짜 이미지에서 진짜 이미지를 더 이상 구별하지 못하게 될 때, 평형상태에 도달하게 된다.

위 이미지는 50epoch동안 훈련한 Generator가 생성해낸 연속된 이미지이다.
이미지들은 랜덤한 잡음으로부터 시작되어, 점차 시간이 지남에 따라 손으로 쓴 숫자를 닮아가게 된다.
MIT의 Intro to Deep Learning 수업

위 그림은 네트워크 구조이다.
autuencoder를 중심부터 바깥쪽으로 뒤집어 둔 개념이다.

import tensorflow as tf

import glob
import imageio
import matplotlib.pyplot as plt
import numpy as np
import os
import PIL
from tensorflow.keras import layers
import time

from IPython import display


# 데이터셋 로딩
# 생성자와 감별자 훈련하기 위해 MNIST 데이터셋 사용
# 생성자는 손글씨 숫자 데이터를 닮은 숫자를 생성
(train_images, train_labels), (_, _) = tf.keras.datasets.mnist.load_data()


train_images = train_images.reshape(train_images.shape[0], 28, 28, 1).astype('float32')
train_images = (train_images - 127.5) / 127.5 # 이미지를 [-1, 1]로 정규화


BUFFER_SIZE = 60000
BATCH_SIZE = 256


# 데이터 배치를 만들고 셔플
train_dataset = tf.data.Dataset.from_tensor_slices(train_images).shuffle(BUFFER_SIZE).batch(BATCH_SIZE)

DCGAN

Generator(생성자)

생성자는 시드값(seed; 랜덤한 잡음)으로부터 이미지를 생성하기 위해 Conv2D(업샘플링) 층을 이용한다.
처음 Dense 층은 이 시드값을 인풋으로 받는다.
그 다음 원하는 사이즈의 28 * 28 * 1의 이미지가 나오도록 업샘플링을 여러번 한다.
tanh를 사용하는 마지막 층을 제외한 나머지 각 층마다 활성 함수로 LeakyReLU를 사용한다.

def make_generator_model():
    model = tf.keras.Sequential()
    model.add(layers.Dense(7*7*256, use_bias=False, input_shape=(100,)))
    model.add(layers.BatchNormalization())
    model.add(layers.LeakyReLU())

    model.add(layers.Reshape((7, 7, 256)))
    assert model.output_shape == (None, 7, 7, 256) # 주목: 배치사이즈로 None이 주어진다.

    model.add(layers.Conv2DTranspose(128, (5, 5), strides=(1, 1), padding='same', use_bias=False))
    assert model.output_shape == (None, 7, 7, 128)
    model.add(layers.BatchNormalization())
    model.add(layers.LeakyReLU())

    model.add(layers.Conv2DTranspose(64, (5, 5), strides=(2, 2), padding='same', use_bias=False))
    assert model.output_shape == (None, 14, 14, 64)
    model.add(layers.BatchNormalization())
    model.add(layers.LeakyReLU())

    model.add(layers.Conv2DTranspose(1, (5, 5), strides=(2, 2), padding='same', use_bias=False, activation='tanh'))
    assert model.output_shape == (None, 28, 28, 1)

    return model


# 생성자를 이용해 이미지 생성
generator = make_generator_model()

noise = tf.random.normal([1, 100])
generated_image = generator(noise, training=False)

plt.imshow(generated_image[0, :, :, 0], cmap='gray')

Discriminator(감별자)

감별자는 합성곱 신경망(CNN) 기반의 이미지 분류기이다.

def make_discriminator_model():
    model = tf.keras.Sequential()
    model.add(layers.Conv2D(64, (5, 5), strides=(2, 2), padding='same',
                                     input_shape=[28, 28, 1]))
    model.add(layers.LeakyReLU())
    model.add(layers.Dropout(0.3))

    model.add(layers.Conv2D(128, (5, 5), strides=(2, 2), padding='same'))
    model.add(layers.LeakyReLU())
    model.add(layers.Dropout(0.3))

    model.add(layers.Flatten())
    model.add(layers.Dense(1))

    return model


# 감별자를 사용하여 생성된 이미지가 진짜인지 가짜인지 판별
# 진짜 이미지에는 양수값, 가짜 이미지에는 음수값 출력하도록 훈련
discriminator = make_discriminator_model()
decision = discriminator(generated_image)
print (decision)
'''
tf.Tensor([[-0.00101271]], shape=(1, 1), dtype=float32)
'''

손실함수와 옵티마이저 정의

# 엔트로피 손실함수 (cross entropy loss)를 계산하기 위해 헬퍼 (helper) 함수 반환
cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)

감별자 손실함수

감별자가 가짜 이미지에서 얼마나 진짜 이미지를 잘 판별하는지 수치화한다.
진짜 이미지에 대한 감별자의 예측과 1로 이루어진 행렬을 비교하고, 가짜 이미지에 대한 감별자의 예측과 0으로 이루어진 행렬을 비교한다.

def discriminator_loss(real_output, fake_output):
    real_loss = cross_entropy(tf.ones_like(real_output), real_output)
    fake_loss = cross_entropy(tf.zeros_like(fake_output), fake_output)
    total_loss = real_loss + fake_loss
    return total_loss

생성자 손실함수

생성자의 손실함수는 감별자를 얼마나 잘 속였는지에 대해 수치화한다.
직관적으로 생성자가 원활히 수행되고 있다면, 감별자는 가짜 이미지를 진짜로 분류할 것이다.
생성된 이미지에 대한 감별자의 결정을 1로 이루어진 행렬과 비교할 것이다.

def generator_loss(fake_output):
    return cross_entropy(tf.ones_like(fake_output), fake_output)


# 감별자와 생성자는 따로 훈련되기 때문에, 감별자와 생성자의 옵티마이저는 다르게 설정
generator_optimizer = tf.keras.optimizers.Adam(1e-4)
discriminator_optimizer = tf.keras.optimizers.Adam(1e-4)

체크포인트 저장

오랜 시간 진행되는 훈련이 방해되는 경우 유용하게 쓰일 수 있도록 학습 중간에 모델의 저장방법과 복구방법을 보여준다.

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)

훈련 루프 정의

훈련 루프는 생성자가 입력으로 랜덤 시드를 받는 것으로부터 시작된다.
그 시드값을 사용하여 이미지를 생성한다.
감별자를 사용하여 (훈련 세트에서 갖고 온)진짜 이미지와 (생성자가 생성해낸)가짜 이미지를 분류한다.
각 모델의 손실을 계산하고, gradients를 사용해 생성자와 감별자를 업데이트한다.

EPOCHS = 50
noise_dim = 100
num_examples_to_generate = 16

# 시드 시간이 지나도 재활용
# (GIF 애니메이션에서 진전 내용을 시각화하는데 쉽기 때문) 
seed = tf.random.normal([num_examples_to_generate, noise_dim])


# 이 데코레이터는 함수를 "컴파일"한다.
@tf.function
def train_step(images):
    noise = tf.random.normal([BATCH_SIZE, noise_dim])

    with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:
      generated_images = generator(noise, training=True)

      real_output = discriminator(images, training=True)
      fake_output = discriminator(generated_images, training=True)

      gen_loss = generator_loss(fake_output)
      disc_loss = discriminator_loss(real_output, fake_output)

    gradients_of_generator = gen_tape.gradient(gen_loss, generator.trainable_variables)
    gradients_of_discriminator = disc_tape.gradient(disc_loss, discriminator.trainable_variables)

    generator_optimizer.apply_gradients(zip(gradients_of_generator, generator.trainable_variables))
    discriminator_optimizer.apply_gradients(zip(gradients_of_discriminator, discriminator.trainable_variables))


def train(dataset, epochs):
  for epoch in range(epochs):
    start = time.time()

    for image_batch in dataset:
      train_step(image_batch)

    # GIF를 위한 이미지 바로 생성
    display.clear_output(wait=True)
    generate_and_save_images(generator,
                             epoch + 1,
                             seed)

    # 15 에포크가 지날 때마다 모델 저장
    if (epoch + 1) % 15 == 0:
      checkpoint.save(file_prefix = checkpoint_prefix)
    
    # print (' 에포크 {} 에서 걸린 시간은 {} 초 입니다'.format(epoch +1, time.time()-start))
    print ('Time for epoch {} is {} sec'.format(epoch + 1, time.time()-start))

  # 마지막 에포크가 끝난 후 생성
  display.clear_output(wait=True)
  generate_and_save_images(generator,
                           epochs,
                           seed)


# 이미지 생성 및 저장
def generate_and_save_images(model, epoch, test_input):
  # `training`이 False로 맞춰진 것 주목
  # (배치정규화를 포함하여) 모든 층들이 추론 모드로 실행
  predictions = model(test_input, training=False)

  fig = plt.figure(figsize=(4,4))

  for i in range(predictions.shape[0]):
      plt.subplot(4, 4, i+1)
      plt.imshow(predictions[i, :, :, 0] * 127.5 + 127.5, cmap='gray')
      plt.axis('off')

  plt.savefig('image_at_epoch_{:04d}.png'.format(epoch))
  plt.show()

모델 훈련

train() 메소드를 생성자와 감별자를 동시에 훈련하기 위해 호출한다.
생성적 대립 신경망을 학습하는 것은 까다로울 수 있다.
생성자와 감별자가 서로 제압하지 않는 것이 중요하다.(학습률이 비슷하면 한쪽이 우세해진다.)
훈련 초반부에 생성된 이미지는 랜덤한 노이즈처럼 보인다.
훈련이 진행될수록, 생성된 숫자는 점차 진짜처럼 보일 것이다.
약 50epoch가 지난 후, MNIST 숫자와 닮은 이미지가 생성된다.

%%time
train(train_dataset, EPOCHS)
'''
CPU times: user 2min 2s, sys: 8.81 s, total: 2min 11s
Wall time: 5min 16s
'''

# 마지막 체크 포인트 복구
<tensorflow.python.training.tracking.util.CheckpointLoadStatus at 0x7f5f99ba7490>

GIF 생성

# 에포크 숫자를 사용하여 하나의 이미지 전시
def display_image(epoch_no):
  return PIL.Image.open('image_at_epoch_{:04d}.png'.format(epoch_no))


display_image(EPOCHS)

# imageio로 훈련 중에 저장된 이미지를 사용해 GIF 애니메이션을 만든다.
anim_file = 'dcgan.gif'

with imageio.get_writer(anim_file, mode='I') as writer:
  filenames = glob.glob('image*.png')
  filenames = sorted(filenames)
  last = -1
  for i,filename in enumerate(filenames):
    frame = 2*(i**0.5)
    if round(frame) > round(last):
      last = frame
    else:
      continue
    image = imageio.imread(filename)
    writer.append_data(image)
  image = imageio.imread(filename)
  writer.append_data(image)

import IPython
if IPython.version_info > (6,2,0,''):
  display.Image(filename=anim_file)


# 애니매이션 다운로드
try:
  from google.colab import files
except ImportError:
  pass
else:
  files.download(anim_file)

CycleGAN

위에서 배운 DCGAN의 구조를 변경해서 새로운 아키텍쳐를 생성했다.
이번에는 쌍을 이루는 이미지를 볼 수 있는데, 굳이 쌍이 없더라도 학습과 테스트를 해볼 수 있도록 만들어진 것이 해당 논문의 핵심이었다.
페어링되지 않은 이미지 대 이미지 변환을 보여준다.
이 논문은 한 이미지 도메인의 특성(얼룩줄무늬)을 캡쳐하고 이러한 특성을 다른 이미지 도메인(말;)으로 변환할 수 있는 방법을 알아낼 수 있는 방법을 제안한다.
CycleGAN이라고도 알려진 Cycle-Consistent Adversarial Networks를 사용한 예시를 볼 수 있다.
- 그림 vs 사진
- 얼룩말 vs 말
- 여름 vs 겨울
- 화가 변환

CycleGAN은 주기 일관성 손실을 사용하여 쌍을 이루는 데이터 없이도 훈련을 가능케 한다.
소스 및 대상 도메인 간의 일대일 매핑 없이 한 도메인에서 다른 도메인으로 변환할 수 있다.
해상도 향상, 사진의 그림화, 스타일 변환 등과 같은 많은 흥미로운 작업을 수행할 수 있는 가능성을 열어준다.
필요한 것은 소스와 대상 데이터셋뿐이다.
과일의 종류를 바꾸는 예시

사진 속 계절을 변경하는 예시

실습

# 생성기와 판별자를 가져올 수 있는 tensorflow_examples 패키지 설치
import urllib.request
urllib.request.urlretrieve('https://ds-lecture-data.s3.ap-northeast-2.amazonaws.com/etc/note433/requirements.txt', 'GANrequirements.txt')
!ls


!pip install -r GANrequirements.txt


!pip install tfa-nightly

import os
import numpy as np
import matplotlib.pyplot as plt

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers

import tensorflow_addons as tfa
import tensorflow_datasets as tfds

tfds.disable_progress_bar()
autotune = tf.data.experimental.AUTOTUNE

Input Pipeline

말 이미지에서 얼룩말 이미지로 변환하는 모델을 학습시킨다.
훈련 데이터셋에 무작위 지터링하고, 미러링하여 학습에 이용되도록 설계되어 있다.
과적합을 방지하는 이미지 확대 기술 중 일부이다.
무작위 지터링에서 이미지는 286 * 286 크기가 조정된 다음 무작위로 256 * 256으로 잘린다.
랜덤 미러링에서는 이미지가 좌우로 무작위로 뒤집힌다.

# Load the horse-zebra dataset using tensorflow-datasets.
dataset, _ = tfds.load("cycle_gan/horse2zebra", with_info=True, as_supervised=True)
train_horses, train_zebras = dataset["trainA"], dataset["trainB"]
test_horses, test_zebras = dataset["testA"], dataset["testB"]

# Define the standard image size.
orig_img_size = (286, 286)
# Size of the random crops to be used during training.
input_img_size = (256, 256, 3)
# Weights initializer for the layers.
kernel_init = keras.initializers.RandomNormal(mean=0.0, stddev=0.02)
# Gamma initializer for instance normalization.
gamma_init = keras.initializers.RandomNormal(mean=0.0, stddev=0.02)

buffer_size = 256
batch_size = 1


def normalize_img(img):
    img = tf.cast(img, dtype=tf.float32)
    # Map values in the range [-1, 1]
    return (img / 127.5) - 1.0


def preprocess_train_image(img, label):
    # Random flip
    img = tf.image.random_flip_left_right(img)
    # Resize to the original size first
    img = tf.image.resize(img, [*orig_img_size])
    # Random crop to 256X256
    img = tf.image.random_crop(img, size=[*input_img_size])
    # Normalize the pixel values in the range [-1, 1]
    img = normalize_img(img)
    return img


def preprocess_test_image(img, label):
    # Only resizing and normalization for the test images.
    img = tf.image.resize(img, [input_img_size[0], input_img_size[1]])
    img = normalize_img(img)
    return img


# Apply the preprocessing operations to the training data
train_horses = (
    train_horses.map(preprocess_train_image, num_parallel_calls=autotune)
    .cache()
    .shuffle(buffer_size)
    .batch(batch_size)
)
train_zebras = (
    train_zebras.map(preprocess_train_image, num_parallel_calls=autotune)
    .cache()
    .shuffle(buffer_size)
    .batch(batch_size)
)

# Apply the preprocessing operations to the test data
test_horses = (
    test_horses.map(preprocess_test_image, num_parallel_calls=autotune)
    .cache()
    .shuffle(buffer_size)
    .batch(batch_size)
)
test_zebras = (
    test_zebras.map(preprocess_test_image, num_parallel_calls=autotune)
    .cache()
    .shuffle(buffer_size)
    .batch(batch_size)
)


# dataset, metadata = tfds.load('cycle_gan/horse2zebra',
#                               with_info=True, as_supervised=True)
# dataset, metadata = tfds.load('cycle_gan/monet2photo', 
#                                with_info=True, as_supervised=True)


_, ax = plt.subplots(4, 2, figsize=(10, 15))
for i, samples in enumerate(zip(train_horses.take(4), train_zebras.take(4))):
    horse = (((samples[0][0] * 127.5) + 127.5).numpy()).astype(np.uint8)
    zebra = (((samples[1][0] * 127.5) + 127.5).numpy()).astype(np.uint8)
    ax[i, 0].imshow(horse)
    ax[i, 1].imshow(zebra)
plt.show()

# 빌더 모델
class ReflectionPadding2D(layers.Layer):
    """Implements Reflection Padding as a layer.

    Args:
        padding(tuple): Amount of padding for the
        spatial dimensions.

    Returns:
        A padded tensor with the same type as the input tensor.
    """

    def __init__(self, padding=(1, 1), **kwargs):
        self.padding = tuple(padding)
        super(ReflectionPadding2D, self).__init__(**kwargs)

    def call(self, input_tensor, mask=None):
        padding_width, padding_height = self.padding
        padding_tensor = [
            [0, 0],
            [padding_height, padding_height],
            [padding_width, padding_width],
            [0, 0],
        ]
        return tf.pad(input_tensor, padding_tensor, mode="REFLECT")


def residual_block(
    x,
    activation,
    kernel_initializer=kernel_init,
    kernel_size=(3, 3),
    strides=(1, 1),
    padding="valid",
    gamma_initializer=gamma_init,
    use_bias=False,
):
    dim = x.shape[-1]
    input_tensor = x

    x = ReflectionPadding2D()(input_tensor)
    x = layers.Conv2D(
        dim,
        kernel_size,
        strides=strides,
        kernel_initializer=kernel_initializer,
        padding=padding,
        use_bias=use_bias,
    )(x)
    x = tfa.layers.InstanceNormalization(gamma_initializer=gamma_initializer)(x)
    x = activation(x)

    x = ReflectionPadding2D()(x)
    x = layers.Conv2D(
        dim,
        kernel_size,
        strides=strides,
        kernel_initializer=kernel_initializer,
        padding=padding,
        use_bias=use_bias,
    )(x)
    x = tfa.layers.InstanceNormalization(gamma_initializer=gamma_initializer)(x)
    x = layers.add([input_tensor, x])
    return x


def downsample(
    x,
    filters,
    activation,
    kernel_initializer=kernel_init,
    kernel_size=(3, 3),
    strides=(2, 2),
    padding="same",
    gamma_initializer=gamma_init,
    use_bias=False,
):
    x = layers.Conv2D(
        filters,
        kernel_size,
        strides=strides,
        kernel_initializer=kernel_initializer,
        padding=padding,
        use_bias=use_bias,
    )(x)
    x = tfa.layers.InstanceNormalization(gamma_initializer=gamma_initializer)(x)
    if activation:
        x = activation(x)
    return x


def upsample(
    x,
    filters,
    activation,
    kernel_size=(3, 3),
    strides=(2, 2),
    padding="same",
    kernel_initializer=kernel_init,
    gamma_initializer=gamma_init,
    use_bias=False,
):
    x = layers.Conv2DTranspose(
        filters,
        kernel_size,
        strides=strides,
        padding=padding,
        kernel_initializer=kernel_initializer,
        use_bias=use_bias,
    )(x)
    x = tfa.layers.InstanceNormalization(gamma_initializer=gamma_initializer)(x)
    if activation:
        x = activation(x)
    return x


def get_resnet_generator(
    filters=64,
    num_downsampling_blocks=2,
    num_residual_blocks=9,
    num_upsample_blocks=2,
    gamma_initializer=gamma_init,
    name=None,
):
    img_input = layers.Input(shape=input_img_size, name=name + "_img_input")
    x = ReflectionPadding2D(padding=(3, 3))(img_input)
    x = layers.Conv2D(filters, (7, 7), kernel_initializer=kernel_init, use_bias=False)(
        x
    )
    x = tfa.layers.InstanceNormalization(gamma_initializer=gamma_initializer)(x)
    x = layers.Activation("relu")(x)

    # Downsampling
    for _ in range(num_downsampling_blocks):
        filters *= 2
        x = downsample(x, filters=filters, activation=layers.Activation("relu"))

    # Residual blocks
    for _ in range(num_residual_blocks):
        x = residual_block(x, activation=layers.Activation("relu"))

    # Upsampling
    for _ in range(num_upsample_blocks):
        filters //= 2
        x = upsample(x, filters, activation=layers.Activation("relu"))

    # Final block
    x = ReflectionPadding2D(padding=(3, 3))(x)
    x = layers.Conv2D(3, (7, 7), padding="valid")(x)
    x = layers.Activation("tanh")(x)

    model = keras.models.Model(img_input, x, name=name)
    return model


def get_discriminator(
    filters=64, kernel_initializer=kernel_init, num_downsampling=3, name=None
):
    img_input = layers.Input(shape=input_img_size, name=name + "_img_input")
    x = layers.Conv2D(
        filters,
        (4, 4),
        strides=(2, 2),
        padding="same",
        kernel_initializer=kernel_initializer,
    )(img_input)
    x = layers.LeakyReLU(0.2)(x)

    num_filters = filters
    for num_downsample_block in range(3):
        num_filters *= 2
        if num_downsample_block < 2:
            x = downsample(
                x,
                filters=num_filters,
                activation=layers.LeakyReLU(0.2),
                kernel_size=(4, 4),
                strides=(2, 2),
            )
        else:
            x = downsample(
                x,
                filters=num_filters,
                activation=layers.LeakyReLU(0.2),
                kernel_size=(4, 4),
                strides=(1, 1),
            )

    x = layers.Conv2D(
        1, (4, 4), strides=(1, 1), padding="same", kernel_initializer=kernel_initializer
    )(x)

    model = keras.models.Model(inputs=img_input, outputs=x, name=name)
    return model


# Get the generators
gen_G = get_resnet_generator(name="generator_G")
gen_F = get_resnet_generator(name="generator_F")

# Get the discriminators
disc_X = get_discriminator(name="discriminator_X")
disc_Y = get_discriminator(name="discriminator_Y")


class CycleGan(keras.Model):
    def __init__(
        self,
        generator_G,
        generator_F,
        discriminator_X,
        discriminator_Y,
        lambda_cycle=10.0,
        lambda_identity=0.5,
    ):
        super(CycleGan, self).__init__()
        self.gen_G = generator_G
        self.gen_F = generator_F
        self.disc_X = discriminator_X
        self.disc_Y = discriminator_Y
        self.lambda_cycle = lambda_cycle
        self.lambda_identity = lambda_identity

    def compile(
        self,
        gen_G_optimizer,
        gen_F_optimizer,
        disc_X_optimizer,
        disc_Y_optimizer,
        gen_loss_fn,
        disc_loss_fn,
    ):
        super(CycleGan, self).compile()
        self.gen_G_optimizer = gen_G_optimizer
        self.gen_F_optimizer = gen_F_optimizer
        self.disc_X_optimizer = disc_X_optimizer
        self.disc_Y_optimizer = disc_Y_optimizer
        self.generator_loss_fn = gen_loss_fn
        self.discriminator_loss_fn = disc_loss_fn
        self.cycle_loss_fn = keras.losses.MeanAbsoluteError()
        self.identity_loss_fn = keras.losses.MeanAbsoluteError()

    def train_step(self, batch_data):
        # x is Horse and y is zebra
        real_x, real_y = batch_data

        # For CycleGAN, we need to calculate different
        # kinds of losses for the generators and discriminators.
        # We will perform the following steps here:
        #
        # 1. Pass real images through the generators and get the generated images
        # 2. Pass the generated images back to the generators to check if we
        #    we can predict the original image from the generated image.
        # 3. Do an identity mapping of the real images using the generators.
        # 4. Pass the generated images in 1) to the corresponding discriminators.
        # 5. Calculate the generators total loss (adverserial + cycle + identity)
        # 6. Calculate the discriminators loss
        # 7. Update the weights of the generators
        # 8. Update the weights of the discriminators
        # 9. Return the losses in a dictionary

        with tf.GradientTape(persistent=True) as tape:
            # Horse to fake zebra
            fake_y = self.gen_G(real_x, training=True)
            # Zebra to fake horse -> y2x
            fake_x = self.gen_F(real_y, training=True)

            # Cycle (Horse to fake zebra to fake horse): x -> y -> x
            cycled_x = self.gen_F(fake_y, training=True)
            # Cycle (Zebra to fake horse to fake zebra) y -> x -> y
            cycled_y = self.gen_G(fake_x, training=True)

            # Identity mapping
            same_x = self.gen_F(real_x, training=True)
            same_y = self.gen_G(real_y, training=True)

            # Discriminator output
            disc_real_x = self.disc_X(real_x, training=True)
            disc_fake_x = self.disc_X(fake_x, training=True)

            disc_real_y = self.disc_Y(real_y, training=True)
            disc_fake_y = self.disc_Y(fake_y, training=True)

            # Generator adverserial loss
            gen_G_loss = self.generator_loss_fn(disc_fake_y)
            gen_F_loss = self.generator_loss_fn(disc_fake_x)

            # Generator cycle loss
            cycle_loss_G = self.cycle_loss_fn(real_y, cycled_y) * self.lambda_cycle
            cycle_loss_F = self.cycle_loss_fn(real_x, cycled_x) * self.lambda_cycle

            # Generator identity loss
            id_loss_G = (
                self.identity_loss_fn(real_y, same_y)
                * self.lambda_cycle
                * self.lambda_identity
            )
            id_loss_F = (
                self.identity_loss_fn(real_x, same_x)
                * self.lambda_cycle
                * self.lambda_identity
            )

            # Total generator loss
            total_loss_G = gen_G_loss + cycle_loss_G + id_loss_G
            total_loss_F = gen_F_loss + cycle_loss_F + id_loss_F

            # Discriminator loss
            disc_X_loss = self.discriminator_loss_fn(disc_real_x, disc_fake_x)
            disc_Y_loss = self.discriminator_loss_fn(disc_real_y, disc_fake_y)

        # Get the gradients for the generators
        grads_G = tape.gradient(total_loss_G, self.gen_G.trainable_variables)
        grads_F = tape.gradient(total_loss_F, self.gen_F.trainable_variables)

        # Get the gradients for the discriminators
        disc_X_grads = tape.gradient(disc_X_loss, self.disc_X.trainable_variables)
        disc_Y_grads = tape.gradient(disc_Y_loss, self.disc_Y.trainable_variables)

        # Update the weights of the generators
        self.gen_G_optimizer.apply_gradients(
            zip(grads_G, self.gen_G.trainable_variables)
        )
        self.gen_F_optimizer.apply_gradients(
            zip(grads_F, self.gen_F.trainable_variables)
        )

        # Update the weights of the discriminators
        self.disc_X_optimizer.apply_gradients(
            zip(disc_X_grads, self.disc_X.trainable_variables)
        )
        self.disc_Y_optimizer.apply_gradients(
            zip(disc_Y_grads, self.disc_Y.trainable_variables)
        )

        return {
            "G_loss": total_loss_G,
            "F_loss": total_loss_F,
            "D_X_loss": disc_X_loss,
            "D_Y_loss": disc_Y_loss,
        }


class GANMonitor(keras.callbacks.Callback):
    """A callback to generate and save images after each epoch"""

    def __init__(self, num_img=5):
        self.num_img = num_img

    def on_epoch_end(self, epoch, logs=None):
        _, ax = plt.subplots(5, 2, figsize=(12, 12))
        for i, img in enumerate(test_horses.take(self.num_img)):
            prediction = self.model.gen_G(img)[0].numpy()
            prediction = (prediction * 127.5 + 127.5).astype(np.uint8)
            img = (img[0] * 127.5 + 127.5).numpy().astype(np.uint8)

            ax[i, 0].imshow(img)
            ax[i, 1].imshow(prediction)
            ax[i, 0].set_title("Input image")
            ax[i, 1].set_title("Translated image")
            ax[i, 0].axis("off")
            ax[i, 1].axis("off")

            prediction = keras.preprocessing.image.array_to_img(prediction)
            prediction.save(
                "generated_img_{i}_{epoch}.png".format(i=i, epoch=epoch + 1)
            )
        plt.show()
        plt.close()


# Loss function for evaluating adversarial loss
adv_loss_fn = keras.losses.MeanSquaredError()

# Define the loss function for the generators
def generator_loss_fn(fake):
    fake_loss = adv_loss_fn(tf.ones_like(fake), fake)
    return fake_loss


# Define the loss function for the discriminators
def discriminator_loss_fn(real, fake):
    real_loss = adv_loss_fn(tf.ones_like(real), real)
    fake_loss = adv_loss_fn(tf.zeros_like(fake), fake)
    return (real_loss + fake_loss) * 0.5


# Create cycle gan model
cycle_gan_model = CycleGan(
    generator_G=gen_G, generator_F=gen_F, discriminator_X=disc_X, discriminator_Y=disc_Y
)

# Compile the model
cycle_gan_model.compile(
    gen_G_optimizer=keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5),
    gen_F_optimizer=keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5),
    disc_X_optimizer=keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5),
    disc_Y_optimizer=keras.optimizers.Adam(learning_rate=2e-4, beta_1=0.5),
    gen_loss_fn=generator_loss_fn,
    disc_loss_fn=discriminator_loss_fn,
)
# Callbacks
plotter = GANMonitor()

# Train the model for just one epoch only using 100 images
cycle_gan_model.fit(
    tf.data.Dataset.zip((train_horses.take(100), train_zebras.take(100))), # for 100 samples
    epochs=1,
    callbacks=plotter,
)

!curl -LO https://github.com/AakashKumarNain/CycleGAN_TF2/releases/download/v1.0/saved_checkpoints.zip
!unzip -qq saved_checkpoints.zip


# Load the checkpoints
weight_file = "./saved_checkpoints/cyclegan_checkpoints.090"
cycle_gan_model.load_weights(weight_file).expect_partial()
print("Weights loaded successfully")

_, ax = plt.subplots(4, 2, figsize=(10, 15))
for i, img in enumerate(test_horses.take(4)):
    prediction = cycle_gan_model.gen_G(img, training=False)[0].numpy()
    prediction = (prediction * 127.5 + 127.5).astype(np.uint8)
    img = (img[0] * 127.5 + 127.5).numpy().astype(np.uint8)

    ax[i, 0].imshow(img)
    ax[i, 1].imshow(prediction)
    ax[i, 0].set_title("Input image")
    ax[i, 0].set_title("Input image")
    ax[i, 1].set_title("Translated image")
    ax[i, 0].axis("off")
    ax[i, 1].axis("off")

    prediction = keras.preprocessing.image.array_to_img(prediction)
    prediction.save("predicted_img_{i}.png".format(i=i))
plt.tight_layout()
plt.show()

Review

다른 종류의 GAN
- Least Squares GAN : LSGAN
- Semi-Supervised GAN : SSGAN, Class 분류와 동시에 진위여부를 확인하는 GAN
- Auxiliary Classifier GAN : Class, noise를 이용하여 만들어내는 GAN
- Stack GAN : Text를 이용해서 이미지를 생성하는 GAN
- Cycle GAN : 그림을 사진으로, 사진을 그림으로, 말을 얼룩말로 얼룩말을 말로!!
- Disco GAN : Cycle GAN과 유사, 가방에 신발 디자인을 반영하는 Cross Modality GAN
- Design GAN : 새로운 디자인 티셔츠를 만들기
- Style GAN : 1024x1024 고화질의 이미지를 만들어내는 GAN
- Adapitve instance normalizaton : AdaIN - Style transfer

[Deep Learning] GAN

GAN(생성적 대립 신경망)

DCGAN

Generator(생성자)

Discriminator(감별자)

손실함수와 옵티마이저 정의

감별자 손실함수

생성자 손실함수

체크포인트 저장

훈련 루프 정의

모델 훈련

GIF 생성

CycleGAN

실습

Input Pipeline

Review

Reference