Upload mainmodel (1).py

mainmodel (1).py

@@ -0,0 +1,332 @@
+# -*- coding: utf-8 -*-
+"""MainModel.ipynb
+
+Automatically generated by Colaboratory.
+
+Original file is located at
+    https://colab.research.google.com/drive/1-4IJzCALsOTWJG0nqR1_thQSGLh-Zfm9
+
+## Adversarial AutoEncoder
+
+#### Generative model to learn to reproduce the input and check the output is generated from a input in reconstruction process or generated creatively. The idea is combination of GAN (Generative Adversarial Networks) and AE (AutoEncoder) to produce a super  generative model.
+
+## Initialization
+
+We mounted google drive to access dataset from there.
+"""
+
+from google.colab import drive
+drive.mount('/content/drive')
+
+"""Import useful libraries."""
+
+import numpy as np
+import tensorflow.keras as ke
+import tensorflow as tf
+from pathlib import Path
+import time
+
+"""Check GPU availability"""
+
+print("Num GPUs Available: ", len(tf.config.list_physical_devices('GPU')))
+
+"""## Preparing Data For Train
+
+Read dataset from google drive and load it to `dataset` variable and preprocessing data to prepare them to learn the model with.
+"""
+
+batchsize = 64
+
+directory = '/content/drive/MyDrive/MainModel/train'
+
+dataset = tf.keras.utils.image_dataset_from_directory(
+    directory,
+    labels=None,
+    batch_size=batchsize,
+    image_size=(224, 224)
+)
+
+dataset = dataset.unbatch()
+dataset = dataset.batch(batchsize, drop_remainder=True)
+
+"""## Make Model
+
+Set latent vector size to use in the model.
+"""
+
+latent_vector_size = 256
+
+"""Make the builder function of encoder part of model."""
+
+def build_model_enc():
+    model = ke.models.Sequential()
+    model.add(ke.layers.Conv2D(64,  (5, 5), strides=(2, 2), activation="relu", padding='same', input_shape=(224, 224, 3)))
+    model.add(ke.layers.Conv2D(128, (5, 5), strides=(2, 2), activation="relu", padding='same'))
+    model.add(ke.layers.Conv2D(256, (5, 5), strides=(2, 2), activation="relu", padding='same'))
+    model.add(ke.layers.Conv2D(512, (5, 5), strides=(2, 2), activation="relu", padding='same'))
+    model.add(ke.layers.Conv2D(512, (5, 5), strides=(2, 2), activation="relu", padding='same'))
+    model.add(ke.layers.Flatten())
+    model.add(ke.layers.Dense(512, activation="relu"))
+    model.add(ke.layers.Dense(latent_vector_size, activation="relu"))
+    return model
+
+"""Make the builder function of decoder part of model."""
+
+def build_model_dec():
+    model = ke.models.Sequential()
+    model.add(ke.layers.Dense(512, input_shape=(latent_vector_size,)))
+    model.add(ke.layers.Dense(25088))
+    model.add(ke.layers.Reshape((7, 7, 512)))
+    model.add(ke.layers.Conv2D(512, (5, 5), activation="relu", padding="same"))
+    model.add(ke.layers.UpSampling2D())
+    model.add(ke.layers.Conv2D(512, (5, 5), activation="relu", padding="same"))
+    model.add(ke.layers.UpSampling2D())
+    model.add(ke.layers.Conv2D(256, (5, 5), activation="relu", padding="same"))
+    model.add(ke.layers.UpSampling2D())
+    model.add(ke.layers.Conv2D(128, (5, 5), activation="relu", padding="same"))
+    model.add(ke.layers.UpSampling2D())
+    model.add(ke.layers.Conv2D(64, (5, 5), activation="relu", padding="same"))
+    model.add(ke.layers.UpSampling2D())
+    model.add(ke.layers.Conv2D(3, (5, 5), activation="relu", padding="same"))
+    return model
+
+"""Make the builder function of discriminator part of model."""
+
+def build_model_disc():
+    model = ke.models.Sequential()
+    model.add(ke.layers.Dense(512, activation="relu", input_shape=(latent_vector_size,)))
+    model.add(ke.layers.Dense(256, activation="relu"))
+    model.add(ke.layers.Dense(1, activation="sigmoid"))
+    return model
+
+"""Make the builder function of the all parts of model."""
+
+def build_model_aae():
+    model_encoder = build_model_enc()
+    model_decoder = build_model_dec()
+    model_discriminator = build_model_disc()
+    model_autoencoder = ke.models.Sequential()
+    model_autoencoder.add(model_encoder)
+    model_autoencoder.add(model_decoder)
+    model_encoder__generator__discriminator = ke.models.Sequential()
+    model_encoder__generator__discriminator.add(model_encoder)
+    model_encoder__generator__discriminator.add(model_discriminator)
+    return model_encoder, model_decoder, model_discriminator, model_autoencoder, model_encoder__generator__discriminator
+
+"""Build the model, compile it and then check parts of model.
+
+Then show the model structure.
+"""
+
+model_enc, model_dec, model_disc, model_ae, model_enc_disc = build_model_aae()
+model_enc.summary()
+model_dec.summary()
+model_disc.summary()
+model_ae.summary()
+model_enc_disc.summary()
+model_disc.compile(optimizer=ke.optimizers.Adam(lr=1e-4), loss="binary_crossentropy", metrics=[tf.keras.metrics.BinaryAccuracy()])
+model_enc_disc.compile(optimizer=ke.optimizers.Adam(lr=1e-4), loss="binary_crossentropy", metrics=[tf.keras.metrics.BinaryAccuracy()])
+model_ae.compile(optimizer=ke.optimizers.Adam(lr=1e-3), loss=ke.losses.MeanSquaredError())
+
+"""## Trainable function
+
+Because our model has some complexity to train and has 2 seperate parts (decoder and discriminator) connected to same part (encoder) which makes 2 paths to train the model and optimize the model (autoencoder and adversarial), we should lock some parts of model for traing other parts and change this option a lot. So we defined a function to set a layer and all it's inside layers trainable or untrainable.
+"""
+
+def settrainable(model, toset):
+    for layer in model.layers:
+        layer.trainable = toset
+    model.trainable = toset
+
+"""## Inception Score"""
+
+import math
+from keras.applications.inception_v3 import InceptionV3
+from keras.applications.inception_v3 import preprocess_input
+
+inception_model = InceptionV3()
+
+
+def inception_score(images, n_split=10, eps=1E-16):
+  images = tf.image.resize(images, (299, 299))
+  from tensorflow.python.ops.numpy_ops import np_config
+  np_config.enable_numpy_behavior()
+  processed = images.astype('float32')
+  processed = preprocess_input(images)
+  yhat = inception_model.predict(processed)
+  scores = list()
+  n_part = math.floor(images.shape[0] / n_split)
+  for i in range(n_split):
+    ix_start, ix_end = i * n_part, i * n_part + n_part
+    p_yx = yhat[ix_start:ix_end]
+    p_y = np.expand_dims(p_yx.mean(axis=0), 0)
+    kl_d = p_yx * (np.log(p_yx + eps) - np.log(p_y + eps))
+    sum_kl_d = kl_d.sum(axis=1)
+    avg_kl_d = np.mean(sum_kl_d)
+    is_score = np.exp(avg_kl_d)
+    scores.append(is_score)
+  is_avg, is_std = np.mean(scores), np.std(scores)
+  return is_avg, is_std
+
+"""## Train the model
+
+In this part model is trained based on the model defined and inputs loaded.
+
+The default batch size is seted to 64 and epoch numbers to 20. In training each batch of each epoch at the first we train the autoencoder, then discriminator and then encoder-discriminator (adversarial part).
+"""
+
+epochs = 20
+
+datasetnumpy = dataset.map(lambda x : (x, x))
+datasetnumpy2 = dataset.map(lambda x : (x, np.ones((batchsize, 1))))
+
+enc_path =      '/content/drive/MyDrive/MainModel/model_enc'
+dec_path =      '/content/drive/MyDrive/MainModel/model_dec'
+disc_path =     '/content/drive/MyDrive/MainModel/model_disc'
+ae_path =       '/content/drive/MyDrive/MainModel/model_ae'
+enc_disc_path = '/content/drive/MyDrive/MainModel/model_enc_disc'
+log_path =      '/content/drive/MyDrive/MainModel/log.txt'
+
+initial_epoch = -1
+
+my_file = Path(log_path)
+if my_file.is_file():
+  with open(log_path, 'r') as file:
+    initial_epoch = int(file.readline())
+    print(f'log found. epoch: {initial_epoch}')
+  model_enc = ke.models.load_model(enc_path)
+  model_dec = ke.models.load_model(dec_path)
+  model_disc = ke.models.load_model(disc_path)
+  model_ae = ke.models.load_model(ae_path)
+  model_enc_disc = ke.models.load_model(enc_disc_path)
+else:
+  print('log not found.')
+
+for epochnumber in range(epochs):
+  if epochnumber < initial_epoch:
+    continue
+  batchcounter = 0
+  batch_size = 0
+  for batchtensor in dataset:
+    batchcounter = batchcounter + 1;
+    settrainable(model_ae, True)
+    settrainable(model_enc, True)
+    settrainable(model_dec, True)
+    batch = batchtensor.numpy()
+    batch_size = batch.shape[0]
+    model_ae.train_on_batch(batch, batch)
+    settrainable(model_disc, True)
+    batchpred = model_enc.predict(batch)
+    fakepred = np.random.standard_normal((batch_size, latent_vector_size))
+    discbatch_x = np.concatenate([batchpred, fakepred])
+    discbatch_y = np.concatenate([np.zeros(batch_size), np.ones(batch_size)])
+    model_disc.train_on_batch(discbatch_x, discbatch_y)
+    settrainable(model_enc_disc, True)
+    settrainable(model_enc, True)
+    settrainable(model_disc, False)
+    model_enc_disc.train_on_batch(batch, np.ones(batch_size))
+    if batchcounter % 100 == 0:
+      model_enc.save(enc_path)
+      model_dec.save(dec_path)
+      model_disc.save(disc_path)
+      model_ae.save(ae_path)
+      model_enc_disc.save(enc_disc_path)
+      with open(log_path, 'w') as file:
+        file.write(f'{epochnumber}\n{batchcounter}')
+    elif batchcounter % 20 == 0:
+      print(batchcounter)
+  with open(log_path, 'w') as file:
+    file.write(f'{epochnumber}')
+  ae_evaluation = model_ae.evaluate(datasetnumpy, batch_size=batchsize)
+  adversarial_evaluation = model_enc_disc.evaluate(datasetnumpy2, batch_size=batchsize)
+  print('Reconstruction Loss:', ae_evaluation)
+  print('Adverserial Loss:', adversarial_evaluation[0])
+  print('Adverserial Accuracy:', adversarial_evaluation[1])
+  if batchcounter >= 1000:
+    num_batch = 100
+  else:
+    num_batch = 10
+  fakepred = np.random.standard_normal((batch_size * num_batch, latent_vector_size))
+  generated_data = model_dec.predict(fakepred)
+  inc_score = inception_score(generated_data)
+  print(f'Inception score for decoder is: {inc_score}')
+
+import os
+try:
+  my_file = Path(log_path)
+  if my_file.is_file():
+    os.remove(log_path)
+except:
+  print('Exception occured.')
+
+ae_evaluation = model_ae.evaluate(datasetnumpy, batch_size=batchsize)
+adversarial_evaluation = model_enc_disc.evaluate(datasetnumpy2, batch_size=batchsize)
+print('Reconstruction Loss:', ae_evaluation)
+print('Adverserial Loss:', adversarial_evaluation[0])
+print('Adverserial Accuracy:', adversarial_evaluation[1])
+if batchcounter >= 1000:
+  num_batch = 100
+else:
+  num_batch = 10
+fakepred = np.random.standard_normal((batch_size * num_batch, latent_vector_size))
+generated_data = model_dec.predict(fakepred)
+inc_score = inception_score(generated_data)
+print(f'Inception score for decoder is: {inc_score}')
+model_enc.save(enc_path)
+model_dec.save(dec_path)
+model_disc.save(disc_path)
+model_ae.save(ae_path)
+model_enc_disc.save(enc_disc_path)
+
+"""## Preparing Data For Evaluation | Test
+
+Read dataset from google drive and load it to `dataset` variable and preprocessing data to prepare them to learn the model with.
+"""
+
+batchsize = 64
+
+directory = '/content/drive/MyDrive/MainModel/test'
+
+dataset = tf.keras.utils.image_dataset_from_directory(
+    directory,
+    labels=None,
+    batch_size=batchsize,
+    image_size=(224, 224)
+)
+
+dataset = dataset.unbatch()
+dataset = dataset.batch(batchsize, drop_remainder=True)
+
+"""## Evaluation | Test"""
+
+enc_path =      '/content/drive/MyDrive/MainModel/model_enc'
+dec_path =      '/content/drive/MyDrive/MainModel/model_dec'
+disc_path =     '/content/drive/MyDrive/MainModel/model_disc'
+ae_path =       '/content/drive/MyDrive/MainModel/model_ae'
+enc_disc_path = '/content/drive/MyDrive/MainModel/model_enc_disc'
+log_path =      '/content/drive/MyDrive/MainModel/log.txt'
+
+model_enc = ke.models.load_model(enc_path)
+model_dec = ke.models.load_model(dec_path)
+model_disc = ke.models.load_model(disc_path)
+model_ae = ke.models.load_model(ae_path)
+model_enc_disc = ke.models.load_model(enc_disc_path)
+
+with open(log_path, 'r') as file:
+  initial_epoch = int(file.readline())
+  print(f'log found. Epochs model trained: {initial_epoch}')
+
+datasetnumpy = dataset.map(lambda x : (x, x))
+datasetnumpy2 = dataset.map(lambda x : (x, np.ones((batchsize, 1))))
+ae_evaluation = model_ae.evaluate(datasetnumpy, batch_size=batchsize)
+adversarial_evaluation = model_enc_disc.evaluate(datasetnumpy2, batch_size=batchsize)
+print('Reconstruction Loss:', ae_evaluation)
+print('Adverserial Accuracy(positive):', adversarial_evaluation[1])
+fakepred = np.random.standard_normal((batchsize, latent_vector_size))
+fake_pred = model_disc.predict(fakepred)
+fake_disc_score = fake_pred.round() == 0
+fake_disc_score = fake_disc_score.sum()
+print('Adverserial Accuracy(negative):', fake_disc_score / batchsize)
+generated_data = model_dec.predict(fakepred)
+inc_score = inception_score(generated_data)
+print(f'Inception score for decoder is: {inc_score}')