{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"from random import random\nfrom numpy import load\nfrom numpy import zeros\nfrom numpy import ones\nfrom numpy import asarray\nfrom numpy.random import randint\nfrom keras.optimizers import Adam\nfrom keras.initializers import RandomNormal\nfrom keras.models import Model\nfrom tensorflow.keras import Input\nfrom keras.layers import Conv2D\nfrom keras.layers import Conv2DTranspose\nfrom keras.layers import LeakyReLU\nfrom keras.layers import Activation\nfrom keras.layers import Concatenate\nimport matplotlib.pyplot as plt\nfrom os import listdir\n#from keras.preprocessing.image import load_img\nfrom numpy import savez_compressed\nfrom keras import backend as K\nimport os\nimport numpy as np\nfrom PIL import Image\nimport pandas as pd\nimport tensorflow as tf\n\ntry:\n    tpu = tf.distribute.cluster_resolver.TPUClusterResolver()\n    print('Device:', tpu.master())\n    tf.config.experimental_connect_to_cluster(tpu)\n    tf.tpu.experimental.initialize_tpu_system(tpu)\n    strategy = tf.distribute.experimental.TPUStrategy(tpu)\nexcept:\n    strategy = tf.distribute.get_strategy()\nprint('Number of replicas:', strategy.num_replicas_in_sync)\n\nAUTOTUNE = tf.data.experimental.AUTOTUNE\n    \nprint(tf.__version__)\n\nwith strategy.scope():\n    normalized_pixel_values2 = pd.read_pickle('/kaggle/input/normalized1/normalized_pixel_values/npv2_data.pkl')\n\n    normalized_pixel_values1 = pd.read_pickle('/kaggle/input/normalized1/npv1_data.pkl/npv1_data.pkl')\n    \n    normalized_pixel_values3 = pd.read_pickle('/kaggle/input/normalized1/normalized_pixel_values/npv3_data.pkl')\n        \n    normalized_pixel_values4 = pd.read_pickle('/kaggle/input/normalized1/normalized_pixel_values/npv4_data.pkl')\n        \n\n        \nimport tensorflow as tf\nfrom tensorflow.keras.layers import GroupNormalization\n# define layer\nlayer = GroupNormalization(axis=-1, groups = -1)\nprint('import successful')\n\n\ndef discriminator_model(image_shape):\n        init = RandomNormal(stddev = 0.02) #weight initialization with a normal distirbution curve\n        input_image = Input(shape = image_shape)\n        d = Conv2D(64, (4,4), strides=(2,2), padding='same', kernel_initializer=init)(input_image) #64 - number of layers/output channels, (4,4) - size of filter, padding = \"same\" - ensures output has same dimensions as input after padding\n        d = LeakyReLU(alpha=0.2)(d)         \n\n        d = Conv2D(128, (4,4), strides=(2,2), padding='same', kernel_initializer=init)(d)\n        d = GroupNormalization(axis=-1, groups = 1)(d)\n        d = LeakyReLU(alpha=0.2)(d)\n\n        d = Conv2D(256, (4,4), strides=(2,2), padding='same', kernel_initializer=init)(d)\n        d = GroupNormalization(axis=-1, groups = 1)(d)\n        d = LeakyReLU(alpha=0.2)(d)\n\n        d = Conv2D(512, (4,4), strides=(2,2), padding='same', kernel_initializer=init)(d)\n        d = GroupNormalization(axis=-1, groups = 1)(d)\n        d = LeakyReLU(alpha=0.2)(d)\n\n        d = Conv2D(512, (4,4), padding='same', kernel_initializer=init)(d)\n        d = GroupNormalization(axis=-1, groups = 1)(d)\n        d = LeakyReLU(alpha=0.2)(d)\n\n        patch_out = Conv2D(1, (4,4), padding='same', kernel_initializer=init)(d)\n        model = Model(input_image, patch_out)\n\n        model.compile(loss='mse', optimizer=Adam(learning_rate=0.0002, beta_1=0.5), loss_weights=[0.5], metrics=['accuracy']) #lr - learning rate, beta_1 - influences moving average of past gradients, loss_weights - shows relative contribution of mse to the overall loss during training\n        print('discriminator model ready')\n        return model\n\ndef generator_model(image_shape):\n        init = RandomNormal(stddev = 0.02) #weight initialization with a normal distirbution curve\n        input_image = Input(shape = image_shape)\n        g = Conv2D(64, (7,7), padding = 'same', kernel_initializer = init)(input_image) #64 - number of filters, (7 x 7) - size of filter\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        g = Activation('relu')(g)\n\n        g =Conv2D(128, (3,3), strides = (2,2), padding = 'same', kernel_initializer = init)(g)\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        g = Activation('relu')(g)\n\n        g =Conv2D(256, (3,3), strides = (2,2), padding = 'same', kernel_initializer = init)(g)\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        g = Activation('relu')(g)\n\n        g = Conv2DTranspose(128, (3,3), strides = (2,2), padding = 'same', kernel_initializer = init)(g)\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        g = Activation('relu')(g)\n\n        g = Conv2DTranspose(64, (3,3), strides = (2,2), padding = 'same', kernel_initializer = init)(g)\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        g = Activation('relu')(g)\n\n        g = Conv2DTranspose(3, (7,7), padding = 'same', kernel_initializer = init)(g)\n        g = GroupNormalization(axis=-1, groups = 1)(g)\n        output_image = Activation('tanh')(g)\n        \n        model = Model(input_image, output_image)\n        print('generator model ready')\n        return model\n\ndef composite_model(g_model_1, d_model, g_model_2, image_shape):\n        g_model_1.trainable = True #we need to change the weights of the main generator\n        d_model.trainable = False\n        g_model_2.trainable = False\n\n        #discriminator element\n        input_gen = Input(shape = image_shape)\n        gen1_out = g_model_1(input_gen)\n        output_d = d_model(gen1_out)\n\n        #identity element\n        input_id = Input(shape = image_shape)\n        output_id = g_model_1(input_id)\n\n        #forward cycle\n        output_f = g_model_2(gen1_out)\n\n        #backward cycle\n        gen2_out = g_model_2(input_id)\n        output_b = g_model_1(gen2_out)\n\n        model = Model([input_gen, input_id], [output_d, output_id, output_f, output_b])\n        optimizer = Adam(learning_rate= 0.0002, beta_1 = 0.5)\n\n        model.compile(loss = ['mse', 'mae', 'mae', 'mae'], loss_weights = [1,5,10,10], optimizer = optimizer, metrics=['accuracy'])\n        print('composite model ready')\n        return model \n\n           \ndef generate_real_samples(dataset, n_samples, patch_shape):\n         # choose random instances\n        ix = randint(0, dataset.shape[0], n_samples)\n         # retrieve selected images\n        X = dataset[ix]\n         # generate 'real' class labels (1)\n        y = ones((n_samples, patch_shape, patch_shape, 1))\n        print('real sample model ready')\n        return X, y\n\n    \ndef generate_fake_samples(g_model, dataset, patch_shape):\n     # generate fake instance\n        X = g_model.predict(dataset)\n     # create 'fake' class labels (0)\n        y = zeros((len(X), patch_shape, patch_shape, 1))\n        print('fake sample model ready')\n        return X, y\n\nwith strategy.scope():\n    dataset = [np.concatenate((normalized_pixel_values1, normalized_pixel_values2)), np.concatenate((normalized_pixel_values3, normalized_pixel_values4))]\n    dataM, dataP = dataset\n    image_shape = (256, 256, 3)\n    g_model_MtoP = generator_model(image_shape)\n    g_model_PtoM = generator_model(image_shape)\n    d_model_M = discriminator_model(image_shape)\n    d_model_P = discriminator_model(image_shape)\n    c_model_MtoP = composite_model(g_model_MtoP, d_model_P, g_model_PtoM, image_shape)\n    c_model_PtoM = composite_model(g_model_PtoM, d_model_M, g_model_MtoP, image_shape)\nprint('ready')\nfrom keras.callbacks import ModelCheckpoint\nfrom keras.models import load_model\nwith strategy.scope():\n        # Define hyperparameters\n    num_epochs = 100 \n    batch_size = 2   \n    patch_shape = d_model_M.output_shape[1]   \n\n      \n    checkpoint_dir = '/kaggle/working/model_checkpoints'  \n\n        # Create the directory if it doesn't exist\n    import os\n    os.makedirs(checkpoint_dir, exist_ok=True)\n\n        # Define a file naming pattern for saved models\n    checkpoint_filepath = os.path.join(checkpoint_dir, 'model_epoch_{epoch:03d}.h5')\n\n        # Create a ModelCheckpoint callback to save models\n    model_checkpoint_callback = ModelCheckpoint(\n        filepath=checkpoint_filepath,\n        save_weights_only=True,  # Save only the weights, not the entire model\n        monitor='val_loss',     # Monitor a specific metric (e.g., validation loss)\n        save_best_only=False,   # Save all models or just the best one\n        verbose=1, # Provide progress updates\n\n        )\n\n        # Training loop\n    training_history = {'D_loss_realM': [], 'D_loss_realP': [], 'D_loss_fakeM': [], 'D_loss_fakeP': [], 'G_lossM': [], 'G_lossP': []}  \n    #Calclate number of steps for training\n    bat_per_epo = int(len(dataM) / batch_size)\n    n_steps = bat_per_epo * num_epochs\n\n    for epoch in range(n_steps):\n                \n        real_imagesM, real_labelsM = generate_real_samples(dataM, batch_size, patch_shape) #This line generates a batch of real images from the current batch in the dataset and assigns them real labels (usually 1, indicating real images).\n        real_imagesP, real_labelsP = generate_real_samples(dataP, batch_size, patch_shape)\n\n        fake_imagesM, fake_labelsM = generate_fake_samples(g_model_PtoM, real_imagesP, patch_shape) #: This line generates a batch of fake images by passing a batch from the dataset through the generator g_model_MtoP and assigns them fake labels (usually 0, indicating fake images).\n        fake_imagesP, fake_labelsP = generate_fake_samples(g_model_MtoP, real_imagesM, patch_shape)\n\n        g_lossP = c_model_PtoM.train_on_batch([real_imagesP, real_imagesM], [real_labelsM, real_imagesM, real_imagesP, real_imagesM])\n\n        d_loss_realM = d_model_M.train_on_batch(real_imagesM, real_labelsM)\n        d_loss_fakeM = d_model_M.train_on_batch(fake_imagesM, fake_labelsM)\n\n        g_lossM = c_model_MtoP.train_on_batch([real_imagesM, real_imagesP], [real_labelsP, real_imagesP, real_imagesM, real_imagesP])\n\n        d_loss_realP = d_model_P.train_on_batch(real_imagesP, real_labelsP)\n        d_loss_fakeP = d_model_P.train_on_batch(fake_imagesP, fake_labelsP)\n\n        # Print training progress\n        print(f\"Epoch {epoch+1}/{n_steps}, D Loss RealM: {d_loss_realM}, D Loss RealP: {d_loss_realP}, D Loss FakeM: {d_loss_fakeM}, D Loss FakeP: {d_loss_fakeP}, G LossM: {g_lossM}, G LossP: {g_lossP}\")\n        training_history['D_loss_realM'].append(d_loss_realM)\n        training_history['D_loss_realP'].append(d_loss_realP)\n        training_history['D_loss_fakeM'].append(d_loss_fakeM)\n        training_history['D_loss_fakeP'].append(d_loss_fakeP)\n        training_history['G_lossM'].append(g_lossM)\n        training_history['G_lossM'].append(g_lossP)\n\n\n\n    g_model_MtoP.save('/kaggle/working/generator_model_MtoP.h5')\n\n    g_model_PtoM.save('/kaggle/working/generator_model_PtoM.h5')\n  \n    d_model_M.save('/kaggle/working/discriminator_model_M.h5')\n\n    d_model_P.save('/kaggle/working/discriminator_model_P.h5')\n\n\n    import pickle\n    with open('/kaggle/working/training_history.pkl', 'wb') as history_file:\n        pickle.dump(training_history, history_file)\n\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\ninput_image = normalized_pixel_values3[:3]\nfake_images = g_model_PtoM.predict(input_image)\n\n# Display the first few generated images\nnum_images_to_display = 3\n\nfor i in range(num_images_to_display):\n    plt.figure(figsize=(8, 8))\n    # Ensure that pixel values are within the range [-1, 1] for displaying\n    clipped_image = np.clip(fake_images[i], -1, 1)\n    plt.imshow((clipped_image + 1) / 2)  # Normalize to [0, 1] for display\n    plt.axis('off')\n    plt.show()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from tensorflow import keras\n\n# Load the entire model\nloaded_model = keras.models.load_model('/kaggle/working/discriminator_model_M.h5')\n\n# Print model summary\nloaded_model.summary()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"loaded_model = keras.models.load_model('/kaggle/working/generator_model_PtoM.h5')\n\n# Print model summary\nloaded_model.summary()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import pickle\n\n# Load the pickled data from the file\nwith open('/kaggle/working/training_history.pkl', 'rb') as file:\n    training_history = pickle.load(file)\n\n# Now you can work with the loaded data\n# For example, you can print the contents of the loaded dictionary\nprint(training_history)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}