0
$\begingroup$

I'm trying to implement DC GAN as they have described in the paper. Specifically, they mention the below points

  1. Use strided convolutions instead of pooling or upsampling layers.
  2. Use only one fully connected layer
  3. Use Batch Normalization: Directly applying batchnorm to all layers resulted in sample oscillation and model instability. This was avoided by not applying batchnorm to the generator output layer and the discriminator input layer.
  4. use ReLU for generator and Leaky ReLU for discriminator

I tried to implement a GAN for MNIST dataset. It is outputting garbage. I tried

  1. changing learning rate from 0.01 to 0.00001
  2. optimizer momentum as 0.5, 0.9
  3. Using BatchNormalization before and after activation layer
  4. BatchNormalization momentum as 0.5, 0.9, 0.99
  5. Training for upto 3,00,000 iterations

But nothing is working. I'm just getting garbage output. But I noticed two strange things

  1. Both generator and discriminator loss are going to 0, accuracy going to 1. How is this possible?
  2. If I remove all Batch Normalization layers from discriminator, the model starts working. Why? The paper suggests to use BatchNormalization, but it is working otherwise.

Any help, tips or suggestions is highly appreciated. Thanks!

Here is my full code:
MnistModel07.py

import numpy
from keras import Sequential
from keras.engine.saving import load_model
from keras.initializers import TruncatedNormal
from keras.layers import Activation, BatchNormalization, Conv2D, Conv2DTranspose, Dense, Flatten, LeakyReLU, Reshape
from keras.optimizers import Adam

from DcGanBaseModel import DcGanBaseModel


class MnistModel07(DcGanBaseModel):
    def __init__(self, verbose: bool = False):
        super().__init__(verbose)
        self.generator_model = None
        self.discriminator_model = None
        self.concatenated_model = None
        self.verbose = verbose

    def build_models(self):
        self.generator_model = self.build_generator_model()
        self.discriminator_model = self.build_discriminator_model()
        self.concatenated_model = self.build_concatenated_model()
        self.print_model_summary()

    def build_generator_model(self):
        if self.generator_model:
            return self.generator_model

        generator_model = Sequential()
        generator_model.add(Dense(7 * 7 * 512, input_dim=100,
                                  kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        generator_model.add(Activation('relu'))
        generator_model.add(BatchNormalization(momentum=0.9))
        generator_model.add(Reshape((7, 7, 512)))

        generator_model.add(Conv2DTranspose(256, 3, strides=2, padding='same',
                                            kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        generator_model.add(Activation('relu'))
        generator_model.add(BatchNormalization(momentum=0.9))

        generator_model.add(Conv2DTranspose(128, 3, strides=2, padding='same',
                                            kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        generator_model.add(Activation('relu'))
        generator_model.add(BatchNormalization(momentum=0.9))

        generator_model.add(Conv2D(1, 3, padding='same',
                                   kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        generator_model.add(Activation('tanh'))

        return generator_model

    def build_discriminator_model(self):
        if self.discriminator_model:
            return self.discriminator_model

        discriminator_model = Sequential()
        discriminator_model.add(Conv2D(128, 3, strides=2, input_shape=(28, 28, 1), padding='same',
                                       kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        discriminator_model.add(LeakyReLU(alpha=0.2))

        discriminator_model.add(Conv2D(256, 3, strides=2, padding='same',
                                       kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        discriminator_model.add(LeakyReLU(alpha=0.2))
        discriminator_model.add(BatchNormalization(momentum=0.9))

        discriminator_model.add(Flatten())
        discriminator_model.add(Dense(1, kernel_initializer=TruncatedNormal(mean=0.0, stddev=0.02)))
        discriminator_model.add(Activation('sigmoid'))

        return discriminator_model

    def build_concatenated_model(self):
        if self.concatenated_model:
            return self.concatenated_model

        concatenated_model = Sequential()
        concatenated_model.add(self.generator_model)
        concatenated_model.add(self.discriminator_model)

        return concatenated_model

    def print_model_summary(self):
        self.verbose_log(self.generator_model.summary())
        self.verbose_log(self.discriminator_model.summary())
        self.verbose_log(self.concatenated_model.summary())

    def build_dc_gan(self):
        """
        Binary Cross-Entropy Loss is used for both Generator and Discriminator
        Discriminator: loss = -log(D(x)) when x is real image and loss=-log(1-D(x)) when x is fake image
        Optimizer minimizes this loss. This is equivalent to maximize over D(x) as specified in original GAN paper
        Generator: loss = -log(D(G(z))
        Optimizer minimizes this loss. This is the second loss function defined in paper, not the one in min-max
                definition
        Since while training Generator we are not minimizing log(1-D(G(z))), the analytical results we derived won't
                hold for generator part.
        Ideally, Discriminator loss = -ln(0.5); Generator loss = -ln(0.5) = 0.693

        metrics = accuracy: binary_accuracy is used
        https://github.com/keras-team/keras/blob/d8b226f26b35348d934edb1213061993e7e5a1fa/keras/engine/training.py#L651
        https://github.com/keras-team/keras/blob/c2e36f369b411ad1d0a40ac096fe35f73b9dffd3/keras/metrics.py#L6
        Binary_accuracy: Average of correct predictions
        Discriminator: Ideally, discriminator should be completely confused i.e. accuracy=0.5
        Generator: Ideally, Generator should be able to fool discriminator. So, accuracy=1. But, since Discriminator
                    is confused, it randomly flags some images as fake. So, accuracy=0.5
        """
        self.build_models()

        self.discriminator_model.trainable = True
        optimizer = Adam(lr=0.0002, beta_1=0.5, beta_2=0.999, decay=0)
        self.discriminator_model.compile(loss='binary_crossentropy', optimizer=optimizer, metrics=['accuracy'])
        self.discriminator_model.trainable = False
        optimizer = Adam(lr=0.0002, beta_1=0.5, beta_2=0.999, decay=0)
        self.concatenated_model.compile(loss='binary_crossentropy', optimizer=optimizer, metrics=['accuracy'])

    def train_on_batch(self, images_real: numpy.ndarray):
        # Generator output has tanh activation whose range is [-1,1]
        images_real = (images_real.astype('float32') * 2 / 255) - 1

        # Generate Fake Images
        batch_size = images_real.shape[0]
        noise = numpy.random.uniform(-1.0, 1.0, size=[batch_size, 100])
        images_fake = self.generator_model.predict(noise)

        # Train discriminator on both real and fake images
        x = numpy.concatenate((images_real, images_fake), axis=0)
        y = numpy.ones([2 * batch_size, 1])
        y[batch_size:, :] = 0
        d_loss = self.discriminator_model.train_on_batch(x, y)

        # Train generator i.e. concatenated model
        # Note that in concatenated model, training of discriminator weights is disabled
        noise = numpy.random.uniform(-1.0, 1.0, size=[batch_size, 100])
        y = numpy.ones([batch_size, 1])
        g_loss = self.concatenated_model.train_on_batch(noise, y)

        return g_loss, d_loss

    def generate_images(self, num_images=1, noise=None) -> numpy.ndarray:
        if noise is None:
            noise = numpy.random.uniform(-1, 1, size=[num_images, 100])
        # Generator output has tanh activation whose range is [-1,1]
        images = (self.generator_model.predict(noise) + 1) * 255 / 2
        images = numpy.round(images).astype('uint8')
        return images

    def save_generator_model(self, save_path):
        self.generator_model.save(save_path)

    def save_generator_model_data(self, json_path, weights_path):
        with open(json_path, 'w') as json_file:
            json_file.write(self.generator_model.to_json())
        self.generator_model.save_weights(weights_path)

    def load_generator_model(self, model_path):
        self.generator_model = load_model(model_path)

    def load_generator_model_weights(self, weights_path):
        self.generator_model.load_weights(weights_path)

    def save_discriminator_model(self, save_path):
        self.discriminator_model.save(save_path)

    def save_discriminator_model_data(self, json_path, weights_path):
        with open(json_path, 'w') as json_file:
            json_file.write(self.discriminator_model.to_json())
        self.discriminator_model.save_weights(weights_path)

    def load_discriminator_model(self, model_path):
        self.discriminator_model = load_model(model_path)

    def load_discriminator_model_weights(self, weights_path):
        self.discriminator_model.load_weights(weights_path)

    def save_concatenated_model(self, save_path):
        self.concatenated_model.save(save_path)

    def save_concatenated_model_data(self, json_path, weights_path):
        with open(json_path, 'w') as json_file:
            json_file.write(self.concatenated_model.to_json())
        self.concatenated_model.save_weights(weights_path)

    def load_concatenated_model(self, model_path):
        self.concatenated_model = load_model(model_path)

    def load_concatenated_model_weights(self, weights_path):
        self.concatenated_model.load_weights(weights_path)

MnistTrainer.py

import datetime
import os
import time

import numpy
from keras.datasets import mnist
from matplotlib import pyplot as plt

from evaluation.EvaluationMetricsWrapper import ClassifierData, Evaluator
from utils import CommonUtils, GraphPlotter
from utils.CommonUtils import check_output_dir


class MnistTrainer:
    def __init__(self, model, classifier_data: ClassifierData, verbose=False):
        self.x_train = self.get_train_data()
        self.dc_gan = model(verbose=verbose)
        self.dc_gan.build_dc_gan()
        self.evaluator = Evaluator(classifier_data, num_classes=10) if classifier_data is not None else None
        self.verbose = verbose

    @staticmethod
    def get_train_data():
        (x_train, y_train), _ = mnist.load_data()
        x_train = x_train.reshape(x_train.shape[0], x_train.shape[1], x_train.shape[2], 1)
        return x_train

    def train(self, train_steps, batch_size, loss_log_interval, save_interval, output_folder_path=None):
        self.verbose_log('Training begins: ' + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        if output_folder_path is not None:
            CommonUtils.check_output_dir(output_folder_path)
            loss_file_path = os.path.join(output_folder_path, 'TrainLosses.csv')
            self.initialize_loss_file(loss_file_path)
            self.sample_real_images(output_folder_path)
            if self.evaluator is not None:
                metrics_filepath = os.path.join(output_folder_path, 'Evaluation/EvaluationMetrics.csv')
                self.initialize_metrics_file(metrics_filepath)

        for i in range(train_steps):
            # Get real (Dataset) Images
            images_real = self.x_train[numpy.random.randint(0, self.x_train.shape[0], size=batch_size), :, :, :]
            g_loss, d_loss = self.dc_gan.train_on_batch(images_real)

            if output_folder_path is not None:
                # Save train losses,  models, generate sample images
                if (i + 1) % loss_log_interval == 0:
                    # noinspection PyUnboundLocalVariable
                    self.append_losses(loss_file_path, i + 1, g_loss, d_loss)
                if (i + 1) % save_interval == 0:
                    self.save_models(output_folder_path, i + 1)
                    self.generate_images(output_folder_path, i + 1)
                    if self.evaluator is not None:
                        # noinspection PyUnboundLocalVariable
                        self.append_metrics(metrics_filepath, i + 1)

        if output_folder_path is not None:
            # Plot the loss functions and accuracy
            graph_file_path = os.path.join(output_folder_path, 'LossAccuracyPlot.png')
            GraphPlotter.plot_loss_and_accuracy(loss_file_path, graph_file_path)
            if self.evaluator is not None:
                metrics_graph_path = os.path.join(output_folder_path, 'Evaluation/EvaluationMetrics.png')
                GraphPlotter.plot_evaluation_metrics(metrics_filepath, metrics_graph_path)

        self.verbose_log('Training ends: ' + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

    @staticmethod
    def initialize_loss_file(loss_file_path):
        line = 'Iteration No, Generator Loss, Generator Accuracy, Discriminator Loss, Discriminator Accuracy, Time\n'
        with open(loss_file_path, 'w') as loss_file:
            loss_file.write(line)

    def append_losses(self, loss_file_path, iteration_no, g_loss, d_loss):
        line = '{0:05},{1:2.4f},{2:0.4f},{3:2.4f},{4:0.4f},{5}\n' \
            .format(iteration_no, g_loss[0], g_loss[1], d_loss[0], d_loss[1],
                    datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
        with open(loss_file_path, 'a') as loss_file:
            loss_file.write(line)
        self.verbose_log(line)

    def save_models(self, output_folder_path, iteration_no):
        models_save_dir = os.path.join(output_folder_path, 'TrainedModels')
        if not os.path.exists(models_save_dir):
            os.makedirs(models_save_dir)

        self.dc_gan.save_generator_model(
            os.path.join(models_save_dir, 'generator_model_{0}.h5'.format(iteration_no)))
        self.dc_gan.save_generator_model_data(
            os.path.join(models_save_dir, 'generator_model_arch_{0}.json'.format(iteration_no)),
            os.path.join(models_save_dir, 'generator_model_weights_{0}.h5'.format(iteration_no))
        )

        self.dc_gan.save_discriminator_model(
            os.path.join(models_save_dir, 'discriminator_model_{0}.h5'.format(iteration_no)))
        self.dc_gan.save_discriminator_model_data(
            os.path.join(models_save_dir, 'discriminator_model_arch_{0}.json'.format(iteration_no)),
            os.path.join(models_save_dir, 'discriminator_model_weights_{0}.h5'.format(iteration_no))
        )

        self.dc_gan.save_concatenated_model(
            os.path.join(models_save_dir, 'concatenated_model_{0}.h5'.format(iteration_no)))
        self.dc_gan.save_concatenated_model_data(
            os.path.join(models_save_dir, 'concatenated_model_arch_{0}.json'.format(iteration_no)),
            os.path.join(models_save_dir, 'concatenated_model_weights_{0}.h5'.format(iteration_no))
        )

    def sample_real_images(self, output_folder_path):
        filepath = os.path.join(output_folder_path, 'MNIST_Sample_Real_Images.png')
        i = numpy.random.randint(0, self.x_train.shape[0], 16)
        images = self.x_train[i, :, :, :]
        plt.figure(figsize=(10, 10))
        for i in range(16):
            plt.subplot(4, 4, i + 1)
            image = images[i, :, :, :]
            image = numpy.reshape(image, [28, 28])
            plt.imshow(image, cmap='gray')
            plt.axis('off')
        plt.tight_layout()
        plt.savefig(filepath)
        plt.close('all')

    def generate_images(self, output_folder_path, iteration_no, noise=None):
        gen_images_dir = os.path.join(output_folder_path, 'Generated_Images')
        if not os.path.exists(gen_images_dir):
            os.makedirs(gen_images_dir)
        filepath = os.path.join(gen_images_dir, 'MNIST_Gen_Image{0}.png'.format(iteration_no))
        images = self.dc_gan.generate_images(16, noise)
        plt.figure(figsize=(10, 10))
        for i in range(16):
            plt.subplot(4, 4, i + 1)
            image = images[i, :, :, :]
            image = numpy.reshape(image, [28, 28])
            plt.imshow(image, cmap='gray')
            plt.axis('off')
        plt.tight_layout()
        plt.savefig(filepath)
        plt.close('all')

    def initialize_metrics_file(self, filepath: str):
        check_output_dir(os.path.split(filepath)[0])
        with open(filepath, 'w') as metrics_file:
            metrics_file.write('Iteration No,' + ','.join(self.evaluator.get_metrics_names()) + '\n')

    def append_metrics(self, filepath: str, iteration_no):
        metrics = self.evaluator.evaluate(self.dc_gan)
        with open(filepath, 'a') as metrics_file:
            metrics_file.write(str(iteration_no) + ',' + ','.join(map(str, metrics)) + '\n')

    def verbose_log(self, log_line):
        if self.verbose:
            print(log_line)


def main():
    """
    Execute in src directory
    """
    from mnist.MnistModel05 import MnistModel05

    train_steps = 10000
    batch_size = 128
    loss_log_interval = 10
    save_interval = 100
    output_folder_path = '../Runs/Run01'

    classifier_name = 'MnistClassifier06'
    classifier_filepath = '../../../../DiscriminativeModels/01_MNIST_Classification/src/MnistClassifierModel06.py'
    classifier_json_path = \
        '../../../../DiscriminativeModels/01_MNIST_Classification/Runs/MnistClassifier06/Run01/TrainedModels' \
        '/MNIST_Model_Arch_30.json'
    classifier_weights_path = \
        '../../../../DiscriminativeModels/01_MNIST_Classification/Runs/MnistClassifier06/Run01/TrainedModels' \
        '/MNIST_Model_Weights_30.h5'
    classifier_data = ClassifierData(classifier_name, classifier_filepath, classifier_json_path,
                                     classifier_weights_path)

    mnist_trainer = MnistTrainer(model=MnistModel05, classifier_data=classifier_data, verbose=True)
    mnist_trainer.train(train_steps, batch_size, loss_log_interval, save_interval, output_folder_path)
    del mnist_trainer.dc_gan
    return


if __name__ == '__main__':
    start_time = time.time()
    print('Program Started at {0}'.format(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))))
    try:
        main()
    except Exception as e:
        print(e)
    end_time = time.time()
    print('Program Ended at {0}'.format(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))))
    print('Total Execution Time: {0}s'.format(datetime.timedelta(seconds=end_time - start_time)))
$\endgroup$
0
$\begingroup$

Golden Rule: In Keras, if using Batch Normalization layer, train the discriminator on real and fake images separately. Don't combine them.


I was able to solve it by changing the discriminator training code as follows:

d_loss = self.discriminator_model.train_on_batch(images_real, numpy.ones((batch_size, 1)))
d_loss = self.discriminator_model.train_on_batch(images_fake, numpy.zeros((batch_size, 1)))

With this change, the issue of generator and discriminator accuracy being at 1 was also solved. I guess combining real and fake images in a single batch causes some problem with the Batch Normalization in Keras. That was the problem. Why that causing the problem, I have no idea.

$\endgroup$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.