- 🤖 See full list of Machine Learning Experiments on GitHub
- ▶️ Interactive Demo: try this model and other machine learning experiments in action
In this experiment we will generate images of clothing using a Deep Convolutional Generative Adversarial Network (DCGAN). The code is written using the Keras Sequential API with a tf.GradientTape training loop. For training we will be using Fashion MNIST dataset.
A generative adversarial network (GAN) is a class of machine learning frameworks. Two neural networks contest with each other in a game. Two models are trained simultaneously by an adversarial process. A generator ("the artist") learns to create images that look real, while a discriminator ("the art critic") learns to tell real images apart from fakes.
Inspired by: Deep Convolutional Generative Adversarial Network tutorial.
import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np
import math
import datetime
import platform
import imageio
import PIL
import time
import os
import glob
import zipfile
from IPython import display
print('Python version:', platform.python_version())
print('Tensorflow version:', tf.__version__)
print('Keras version:', tf.keras.__version__)
# Checking the eager execution availability.
tf.executing_eagerly()
(x_train, y_train), (x_test, y_test) = tf.keras.datasets.fashion_mnist.load_data()
print('x_train.shape: ', x_train.shape)
print('y_train.shape: ', y_train.shape)
print()
print('x_test.shape: ', x_test.shape)
print('y_test.shape: ', y_test.shape)
# Since we don't need test examples we may concatenate both sets
x_train = np.concatenate((x_train, x_test), axis=0)
print('x_train.shape: ', x_train.shape)
TOTAL_EXAMPLES_NUM = x_train.shape[0]
print('TOTAL_EXAMPLES_NUM: ', TOTAL_EXAMPLES_NUM)
print('y_train[0] =', y_train[0])
Here are the map of classes for the dataset according to documentation:
Label | Class |
---|---|
0 | T-shirt/top |
1 | Trouser |
2 | Pullover |
3 | Dress |
4 | Coat |
5 | Sandal |
6 | Shirt |
7 | Sneaker |
8 | Bag |
9 | Ankle boot |
class_names = [
'T-shirt/top',
'Trouser',
'Pullover',
'Dress',
'Coat',
'Sandal',
'Shirt',
'Sneaker',
'Bag',
'Ankle boot'
]
print(x_train[0])
plt.figure(figsize=(2, 2))
plt.imshow(x_train[0], cmap=plt.cm.binary)
plt.show()
numbers_to_display = 25
num_cells = math.ceil(math.sqrt(numbers_to_display))
plt.figure(figsize=(8, 8))
for i in range(numbers_to_display):
plt.subplot(num_cells, num_cells, i+1)
plt.xticks([])
plt.yticks([])
plt.grid(False)
plt.imshow(x_train[i], cmap=plt.cm.binary)
plt.xlabel(class_names[y_train[i]])
plt.show()
x_train_reshaped = x_train.reshape(
x_train.shape[0],
x_train.shape[1],
x_train.shape[2],
1
).astype('float32')
print('x_train_reshaped.shape: ', x_train_reshaped.shape)
# Normalize image pixel values to [-1, 1] range
x_train_normalized = (x_train_reshaped - 127.5) / 127.5
print('Normalized data values:\n')
print(x_train_normalized[0,:,:,0])
SHUFFLE_BUFFER_SIZE = TOTAL_EXAMPLES_NUM
BATCH_SIZE = 1024
TRAINING_STEPS_PER_EPOCH = math.ceil(TOTAL_EXAMPLES_NUM / BATCH_SIZE)
print('BATCH_SIZE: ', BATCH_SIZE)
print('TRAINING_STEPS_PER_EPOCH: ', TRAINING_STEPS_PER_EPOCH)
train_dataset = tf.data.Dataset.from_tensor_slices(x_train_normalized) \
.shuffle(SHUFFLE_BUFFER_SIZE) \
.batch(BATCH_SIZE)
print(train_dataset)
def make_generator_model():
model = tf.keras.Sequential()
# Step 1.
model.add(tf.keras.layers.Dense(
units=7*7*256,
use_bias=False,
input_shape=(100,)
))
model.add(tf.keras.layers.BatchNormalization())
model.add(tf.keras.layers.LeakyReLU())
# Step 2.
model.add(tf.keras.layers.Reshape((7, 7, 256)))
assert model.output_shape == (None, 7, 7, 256) # None is a batch size.
# Step 3.
model.add(tf.keras.layers.Conv2DTranspose(
filters=128,
kernel_size=(5, 5),
strides=(1, 1),
padding='same',
use_bias=False
))
assert model.output_shape == (None, 7, 7, 128)
model.add(tf.keras.layers.BatchNormalization())
model.add(tf.keras.layers.LeakyReLU())
# Step 4.
model.add(tf.keras.layers.Conv2DTranspose(
filters=64,
kernel_size=(5, 5),
strides=(2, 2),
padding='same',
use_bias=False
))
assert model.output_shape == (None, 14, 14, 64)
model.add(tf.keras.layers.BatchNormalization())
model.add(tf.keras.layers.LeakyReLU())
# Step 5.
model.add(tf.keras.layers.Conv2DTranspose(
filters=1,
kernel_size=(5, 5),
strides=(2, 2),
padding='same',
use_bias=False,
activation='tanh'
))
assert model.output_shape == (None, 28, 28, 1)
return model
generator_model = make_generator_model()
generator_model.summary()
tf.keras.utils.plot_model(
generator_model,
show_shapes=True,
show_layer_names=True,
to_file='generator_model.png'
)
noise = tf.random.normal(shape=[1, 100])
print(noise.numpy())
generated_image = generator_model(noise, training=False)
print('generated_image.shape: ', generated_image.shape)
print(generated_image[0, :, :, 0].numpy())
plt.figure(figsize=(2, 2))
plt.imshow(generated_image[0, :, :, 0], cmap=plt.cm.binary)
The model will be trained to output positive values for real images, and negative values for fake images.
def make_discriminator_model():
model = tf.keras.Sequential()
# Step 1.
model.add(tf.keras.layers.Conv2D(
filters=64,
kernel_size=(5, 5),
strides=(2, 2),
padding='same',
input_shape=[28, 28, 1]
))
model.add(tf.keras.layers.LeakyReLU())
model.add(tf.keras.layers.Dropout(0.3))
# Step 2.
model.add(tf.keras.layers.Conv2D(
filters=128,
kernel_size=(5, 5),
strides=(2, 2),
padding='same'
))
model.add(tf.keras.layers.LeakyReLU())
model.add(tf.keras.layers.Dropout(0.3))
# Step 3.
model.add(tf.keras.layers.Flatten())
# Real vs Fake
model.add(tf.keras.layers.Dense(1))
return model
discriminator_model = make_discriminator_model()
discriminator_model.summary()
tf.keras.utils.plot_model(
discriminator_model,
show_shapes=True,
show_layer_names=True,
to_file='discriminator_model.png'
)
dicision = discriminator_model(generated_image)
print(dicision)
This method quantifies how well the discriminator is able to distinguish real images from fakes. It compares the discriminator's predictions on real images to an array of 1s, and the discriminator's predictions on fake (generated) images to an array of 0s.
def discriminator_loss(real_output, fake_output):
cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)
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
# Test discriminator loss function.
output_combinations = [
# REAL #FAKE
([-1.], [1.]),
([1.], [-1.]),
([1.], [0.]),
([10.], [-1.]),
]
for (real_output, fake_output) in output_combinations:
loss = discriminator_loss(real_output, fake_output).numpy()
print('Discriminator loss for:', real_output, fake_output)
print(' REAL output:', real_output)
print(' FAKE output:', fake_output)
print(' loss: ', loss)
print()
The generator's loss quantifies how well it was able to trick the discriminator. Intuitively, if the generator is performing well, the discriminator will classify the fake images as real (or 1). Here, we will compare the discriminators decisions on the generated images to an array of 1s.
def generator_loss(fake_output):
cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)
loss = cross_entropy(tf.ones_like(fake_output), fake_output)
return loss
# Test generator loss function.
print('Generator loss for >1: ', generator_loss([5.]).numpy())
print('Generator loss for =0: ', generator_loss([0.]).numpy())
generator_optimizer = tf.keras.optimizers.Adam(
learning_rate=0.0001
)
discriminator_optimizer = tf.keras.optimizers.Adam(
learning_rate=0.0001
)
checkpoint_dir = './tmp/ckpt'
checkpoint_prefix = os.path.join(checkpoint_dir, 'ckpt')
checkpoint = tf.train.Checkpoint(
generator_optimizer=generator_optimizer,
discriminator_optimizer=discriminator_optimizer,
generator_model=generator_model,
discriminator_model=discriminator_model
)
EPOCHS = 100
noise_dim = 100
num_examples_to_generate = 16
# We will reuse this seed overtime (so it's easier)
# to visualize progress in the animated GIF)
input_noise_seed = tf.random.normal([num_examples_to_generate, noise_dim])
The training loop begins with generator receiving a random seed as input. That seed is used to produce an image. The discriminator is then used to classify real images (drawn from the training set) and fakes images (produced by the generator). The loss is calculated for each of these models, and the gradients are used to update the generator and discriminator.
# This `tf.function` annotation causes the function to be "compiled".
# @tf.function
def train_step(real_images):
training_history = {
'discriminator': {
'loss': None
},
'generator': {
'loss': None
}
}
# Generate input noise.
noise_images = tf.random.normal([BATCH_SIZE, noise_dim])
with tf.GradientTape() as gen_tape, tf.GradientTape() as disc_tape:
# Generate fake images.
generated_images = generator_model(
noise_images,
training=True
)
# Detect fake and real images.
real_output = discriminator_model(
real_images,
training=True
)
fake_output = discriminator_model(
generated_images,
training=True
)
# Calculate losses.
gen_loss = generator_loss(fake_output)
disc_loss = discriminator_loss(
real_output,
fake_output
)
training_history['discriminator']['loss'] = disc_loss.numpy()
training_history['generator']['loss'] = gen_loss.numpy()
# Calculate gradients.
gradients_of_generator = gen_tape.gradient(
gen_loss,
generator_model.trainable_variables
)
gradients_of_discriminator = disc_tape.gradient(
disc_loss,
discriminator_model.trainable_variables
)
# Do gradient step.
generator_optimizer.apply_gradients(zip(
gradients_of_generator,
generator_model.trainable_variables
))
discriminator_optimizer.apply_gradients(zip(
gradients_of_discriminator,
discriminator_model.trainable_variables
))
return training_history
def train(dataset, epochs, start_epoch=0):
print('Start training...')
training_history = {
'discriminator': {
'loss': []
},
'generator': {
'loss': []
}
}
for epoch in range(epochs)[start_epoch:]:
print('Start epoch #{} ({} steps)...'.format(epoch + 1, TRAINING_STEPS_PER_EPOCH))
start = time.time()
step = 0
for image_batch in dataset:
step += 1
# display.clear_output(wait=True)
# show_progress(step, TRAINING_STEPS_PER_EPOCH)
# generate_and_save_images(
# generator_model,
# epoch + 1,
# input_noise_seed,
# save=False
# )
training_step_history = train_step(image_batch)
discriminator_step_loss = training_step_history['discriminator']['loss']
generator_step_loss = training_step_history['generator']['loss']
training_history['discriminator']['loss'].append(discriminator_step_loss)
training_history['generator']['loss'].append(generator_step_loss)
# Produce images for the GIF as we go.
display.clear_output(wait=True)
generate_and_save_images(
generator_model,
epoch + 1,
input_noise_seed
)
# Save the model every 10 epochs.
if (epoch + 1) % 10 == 0:
checkpoint.save(file_prefix=checkpoint_prefix)
print('Time for epoch #{} is {:.2f}s'.format(epoch + 1, time.time() - start))
print('Discriminator loss: {:.4f}'.format(discriminator_step_loss))
print('Generator loss: {:.4f}'.format(generator_step_loss))
return training_history
def show_progress(current_step, total_steps):
length_divider = 2
progress = math.floor(current_step * 100 / total_steps)
done_steps = progress
left_steps = 100 - done_steps
done_dots = ''.join(['◼︎' for step in range(math.floor(done_steps / length_divider))])
left_dors = ''.join(['・' for step in range(math.floor(left_steps / length_divider))])
print(f'{current_step}/{total_steps}: {done_dots}{left_dors}')
# Test progress function.
show_progress(15, 68)
IMAGES_PREVIEW_PATH = 'tmp/imgs/'
if not os.path.exists(IMAGES_PREVIEW_PATH):
os.makedirs(IMAGES_PREVIEW_PATH)
def generate_and_save_images(model, epoch, test_input, save=True):
# Notice `training` is set to False.
# This is so all layers run in inference mode (batchnorm).
predictions = model(test_input, training=False)
fig_dimension = int(math.sqrt(num_examples_to_generate))
plt.figure(figsize=(8, 8))
fig = plt.figure(figsize=(fig_dimension, fig_dimension))
for i in range(predictions.shape[0]):
plt.subplot(fig_dimension, fig_dimension, i+1)
plt.imshow(
predictions[i, :, :, 0] * 127.5 + 127.5,
cmap=plt.cm.binary
)
plt.axis('off')
if save:
plt.savefig('{}image_at_epoch_{:04d}.png'.format(IMAGES_PREVIEW_PATH, epoch))
plt.show()
if not 'training_history' in locals():
training_history = {
'discriminator': {
'loss': []
},
'generator': {
'loss': []
}
}
training_session_num = 1
start_epoch = training_session_num * EPOCHS
epochs_num = start_epoch + EPOCHS
training_history_current = train(
train_dataset,
epochs=epochs_num,
start_epoch=start_epoch
)
training_history['generator']['loss'] += training_history_current['generator']['loss']
training_history['discriminator']['loss'] += training_history_current['discriminator']['loss']
def render_training_history(training_history):
generator_loss = training_history['generator']['loss']
discriminator_loss = training_history['discriminator']['loss']
plt.figure(figsize=(14, 4))
plt.subplot(1, 2, 1)
plt.title('Generator Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.plot(generator_loss, label='Generator loss')
plt.legend()
plt.grid(linestyle='--', linewidth=1, alpha=0.5)
plt.subplot(1, 2, 2)
plt.title('Discriminator Loss')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.plot(discriminator_loss, label='Discriminator loss')
plt.legend()
plt.grid(linestyle='--', linewidth=1, alpha=0.5)
plt.show()
render_training_history(training_history)
checkpoint.restore(tf.train.latest_checkpoint(checkpoint_dir))
generator_model.save('generator_model.h5', save_format='h5')
discriminator_model.save('discriminator_model.h5', save_format='h5')
# Restore models from files if needed.
# generator_model.load_weights('./generator_model.h5')
# discriminator_model.load_weights('./discriminator_model.h5')
def zip_image_previews():
images_previews_path = IMAGES_PREVIEW_PATH
images_previews_zip_name = 'images_previews.zip'
zipped_files_num = 0
with zipfile.ZipFile(images_previews_zip_name, mode='w') as zip_obj:
for folder_name, subfolders, filenames in os.walk(images_previews_path):
for filename in filenames:
zipped_files_num += 1
file_path = os.path.join(folder_name, filename)
zip_obj.write(file_path, os.path.basename(file_path))
print('Zipped {} files to '.format(zipped_files_num), images_previews_zip_name)
zip_image_previews()
test_examples_num = 10
noise_images = tf.random.normal([test_examples_num, 100])
generated_images = generator_model(noise_images, training=False)
for example_num in range(test_examples_num):
plt.figure(figsize=(3, 3))
plt.subplot(1, 2, 1)
plt.imshow(np.reshape(noise_images[example_num], (10, 10)), cmap=plt.cm.binary)
plt.subplot(1, 2, 2)
plt.imshow(generated_images[example_num, :, :, 0], cmap=plt.cm.binary)
# Display a single image using the epoch number
def display_image(epoch_no):
return PIL.Image.open('{}image_at_epoch_{:04d}.png'.format(IMAGES_PREVIEW_PATH, epoch_no))
display_image(EPOCHS)
anim_file = 'clothes_generation_dcgan.gif'
with imageio.get_writer(anim_file, mode='I') as writer:
filenames = glob.glob(IMAGES_PREVIEW_PATH + '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)
display.Image(filename=anim_file)
To use this model on the web we need to convert it into the format that will be understandable by tensorflowjs. To do so we may use tfjs-converter as following:
tensorflowjs_converter --input_format keras \
./experiments/clothes_generation_dcgan/generator_model.h5 \
./demos/public/models/clothes_generation_dcgan
You find this experiment in the Demo app and play around with it right in you browser to see how the model performs in real life.