{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":21755,"databundleVersionId":1475600,"sourceType":"competition"}],"dockerImageVersionId":30823,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# **Dependencies**","metadata":{}},{"cell_type":"code","source":"# basic packages\nimport numpy as np\nimport pandas as pd\nimport scipy\nimport pickle\nimport random\nimport os\n\n# loading in and transforming data\nimport torch\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision\nimport torchvision.datasets as datasets\nimport torchvision.transforms as transforms\nfrom torchvision.utils import make_grid\nimport PIL\nfrom PIL import Image\nimport imageio\n\n# visualizing data\nimport matplotlib.pyplot as plt\nimport matplotlib.image as mpimg","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-27T14:18:18.340750Z","execution_failed":"2024-12-27T14:18:56.608Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Paths et cetera\npath_save_up_to_5gb = '/kaggle/working/'\npath_monet = '../input/gan-getting-started/monet_jpg/'\npath_photo = '../input/gan-getting-started/photo_jpg/'\nimg_size = 256","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.608Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Seed**","metadata":{}},{"cell_type":"code","source":"def set_seed(seed):\n    random.seed(seed)\n    os.environ['PYTHONHASHSEED'] = str(seed)\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    torch.backends.cudnn.deterministic = True","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.608Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"set_seed(42)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.608Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Save and load**","metadata":{}},{"cell_type":"code","source":"def save_checkpoint(iteration, G_XtoY, G_YtoX, D_X, D_Y, checkpoint_dir='/kaggle/working/'):\n    \"\"\"\n    Saves the parameters of both generators and discriminators.\n    \"\"\"\n    #Path\n    G_XtoY_path = os.path.join(checkpoint_dir, 'G_XtoY.pkl')\n    G_YtoX_path = os.path.join(checkpoint_dir, 'G_YtoX.pkl')\n    D_X_path = os.path.join(checkpoint_dir, 'D_X.pkl')\n    D_Y_path = os.path.join(checkpoint_dir, 'D_Y.pkl')\n    #Saving\n    torch.save(G_XtoY.state_dict(), G_XtoY_path)\n    torch.save(G_YtoX.state_dict(), G_YtoX_path)\n    torch.save(D_X.state_dict(), D_X_path)\n    torch.save(D_Y.state_dict(), D_Y_path)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def load_checkpoint(checkpoint_path, map_location=None):\n    \"\"\"\n    Load checkoint\n    \"\"\"\n    #model.load_state_dict(torch.load(checkpoint_path))\n    checkpoint = torch.load(checkpoint_path, map_location=map_location)\n    print(' [*] Loading checkpoint from %s succeed!' % checkpoint_path)\n    return checkpoint","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def show_test(fixed_Y, fixed_X, G_YtoX, G_XtoY, mean_=0.5, std_=0.5):\n    \"\"\"\n    Shows results of generates based on test image input. \n    \"\"\"\n    #Identify correct device\n    device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n    \n    #Create fake pictures for both cycles\n    fake_X = G_YtoX(fixed_Y.to(device))\n    fake_Y = G_XtoY(fixed_X.to(device))\n    \n    #Generate grids\n    grid_x =  make_grid(fixed_X, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\n    grid_y =  make_grid(fixed_Y, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\n    grid_fake_x =  make_grid(fake_X, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\n    grid_fake_y =  make_grid(fake_Y, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\n    \n    #Normalize pictures to pixel range rom 0 to 255\n    X, fake_X = reverse_normalize(grid_x, mean_, std_), reverse_normalize(grid_fake_x, mean_, std_)\n    Y, fake_Y = reverse_normalize(grid_y, mean_, std_), reverse_normalize(grid_fake_y, mean_, std_)\n    \n    #Transformation from X -> Y\n    fig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, sharey=True, figsize=(20, 10))\n    ax1.imshow(X)\n    ax1.axis('off')\n    ax1.set_title('X')\n    ax2.imshow(fake_Y)\n    ax2.axis('off')\n    ax2.set_title('Fake Y  (Monet-esque)')\n    plt.show()","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Datasets & dataloaders**","metadata":{}},{"cell_type":"code","source":"class ImageDataset(Dataset):\n        \"\"\"\n        Custom dataset\n        \"\"\"\n        \n        def __init__(self, img_path, img_size=256, normalize=True):\n            self.img_path = img_path\n            \n            if normalize:\n                self.transform = transforms.Compose([\n                    transforms.Resize(img_size),\n                    transforms.ToTensor(),\n                    transforms.Normalize(mean=[0.5], std=[0.5])\n                ])\n            else:\n                self.transform = transforms.Compose([\n                    transforms.Resize(img_size),\n                    transforms.ToTensor()\n                ])\n            \n            #Dictionary entries\n            self.img_idx = dict()\n            for number_, img_ in enumerate(os.listdir(self.img_path)):\n                self.img_idx[number_] = img_\n                \n        def __len__(self):\n            #Length of dataset --> number of images\n            return len(self.img_idx)\n        \n        def __getitem__(self, idx):\n            img_path = os.path.join(self.img_path, self.img_idx[idx])\n            img = Image.open(img_path)\n            img = self.transform(img)\n            \n            return img","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def reverse_normalize(image, mean_=0.5, std_=0.5):\n    if torch.is_tensor(image):\n        image = image.detach().numpy()\n    un_normalized_img = image * std_ + mean_\n    un_normalized_img = un_normalized_img * 255\n    return np.uint8(un_normalized_img)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Create datasets\ndataset_monet = ImageDataset(path_monet, img_size=256, normalize=True) #monet\ndataset_photo = ImageDataset(path_photo, img_size=256, normalize=True) #photo","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Create test loaders\nbatch_size_test = 8\nbatch_size=16\ntest_dataloader_Y = DataLoader(dataset_monet, batch_size=batch_size_test, shuffle=False, num_workers=0, pin_memory=True)\ntest_dataloader_X = DataLoader(dataset_photo, batch_size=batch_size_test, shuffle=False, num_workers=0, pin_memory=True)\ndataloader_Y = DataLoader(dataset_monet, batch_size=batch_size, shuffle=True, num_workers=0, pin_memory=True)\ndataloader_X = DataLoader(dataset_photo, batch_size=batch_size, shuffle=True, num_workers=0, pin_memory=True)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Check dataset & dataloader results for test_dataloader_Y\ndataiter = iter(test_dataloader_Y)\nimages_normalized = next(dataiter)  # Corrected from dataiter.next() to next(dataiter)\ngrid_normalized = make_grid(images_normalized, nrow=4).permute(1, 2, 0).detach().numpy()\ngrid_original = reverse_normalize(grid_normalized)\nfig = plt.figure(figsize=(12, 8))\nplt.imshow(grid_original)\nplt.axis('off')\nplt.title('Monet')\nplt.show()\n\n# Check dataset & dataloader results for test_dataloader_X\ndataiter = iter(test_dataloader_X)\nimages_normalized = next(dataiter)  # Corrected from dataiter.next() to next(dataiter)\ngrid_normalized = make_grid(images_normalized, nrow=4).permute(1, 2, 0).detach().numpy()\ngrid_original = reverse_normalize(grid_normalized)\nfig = plt.figure(figsize=(12, 8))\nplt.imshow(grid_original)\nplt.axis('off')\nplt.title('Photo')\nplt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Model**","metadata":{}},{"cell_type":"markdown","source":"> # **Discriminator**","metadata":{}},{"cell_type":"code","source":"import torch.nn as nn\nimport torch.nn.functional as F\n\ndef conv(in_channels, out_channels, kernel_size, stride=2, padding=1, batch_norm=False, instance_norm=False):\n    \"\"\"\n    Creates a convolutional layer, with optional batch / instance normalization. \n    \"\"\"\n    \n    #Add layers\n    layers = []\n    conv_layer = nn.Conv2d(in_channels=in_channels, out_channels=out_channels, kernel_size=kernel_size, stride=stride, padding=padding, bias=False)\n    layers.append(conv_layer)\n    \n    #Batch normalization\n    if batch_norm:\n        layers.append(nn.BatchNorm2d(out_channels))\n    \n    #Instance normalization\n    if instance_norm:\n        layers.append(nn.InstanceNorm2d(out_channels))\n    return nn.Sequential(*layers)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.609Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class Discriminator(nn.Module):\n    \n    def __init__(self, conv_dim=64):\n        super(Discriminator, self).__init__()\n        \"\"\"\n        Input is RGB image (256x256x3) while output is a single value\n        \n        determine size = [(W−K+2P)/S]+1\n        W: input=256\n        K: kernel_size=4\n        P: padding=1\n        S: stride=2\n        \"\"\"\n        \n        #convolutional layers, increasing in depth\n        self.conv1 = conv(in_channels=3, out_channels=conv_dim, kernel_size=4) # (128, 128, 64)\n        self.conv2 = conv(in_channels=conv_dim, out_channels=conv_dim*2, kernel_size=4, instance_norm=True) # (64, 64, 128)\n        self.conv3 = conv(in_channels=conv_dim*2, out_channels=conv_dim*4, kernel_size=4, instance_norm=True) # (32, 32, 256)\n        self.conv4 = conv(in_channels=conv_dim*4, out_channels=conv_dim*8, kernel_size=4, instance_norm=True) # (16, 16, 512)\n        self.conv5 = conv(in_channels=conv_dim*8, out_channels=conv_dim*8, kernel_size=4, batch_norm=True) # (8, 8, 512)\n        \n        #final classification layer\n        self.conv6 = conv(conv_dim*8, out_channels=1, kernel_size=4, stride=1) # (8, 8, 1)\n    \n    def forward(self, x):\n        \n        #leaky relu applied to all conv layers but last\n        out = F.leaky_relu(self.conv1(x), negative_slope=0.2)\n        out = F.leaky_relu(self.conv2(out), negative_slope=0.2)\n        out = F.leaky_relu(self.conv3(out), negative_slope=0.2)\n        out = F.leaky_relu(self.conv4(out), negative_slope=0.2)\n#         out = F.leaky_relu(self.conv5(out), negative_slope=0.2)\n        \n        #classification layer (--> depending on the loss function we might want to use an activation function here, e.g. sigmoid)\n        out = self.conv6(out)\n        return out","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.610Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"> # **Generator**","metadata":{}},{"cell_type":"code","source":"class ResidualBlock(nn.Module):\n\n    def __init__(self, conv_dim):\n        super(ResidualBlock, self).__init__()\n        \"\"\"\n        Residual blocks help the model to effectively learn the transformation from one domain to another. \n        \"\"\"\n        self.conv1 = conv(in_channels=conv_dim, out_channels=conv_dim, kernel_size=3, stride=1, padding=1, instance_norm=True)\n        self.conv2 = conv(in_channels=conv_dim, out_channels=conv_dim, kernel_size=3, stride=1, padding=1, instance_norm=True)\n        \n    def forward(self, x):\n        out_1 = F.relu(self.conv1(x))\n        out_2 = x + self.conv2(out_1)\n        return out_2","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.610Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def deconv(in_channels, out_channels, kernel_size, stride=2, padding=1, batch_norm=False, instance_norm=False, dropout=False, dropout_ratio=0.5):\n    \"\"\"\n    Creates a transpose convolutional layer, with optional batch / instance normalization. Select either batch OR instance normalization. \n    \"\"\"\n    \n    #Add layers\n    layers = []\n    layers.append(nn.ConvTranspose2d(in_channels, out_channels, kernel_size, stride, padding, bias=False))\n    \n    #Batch normalization\n    if batch_norm:\n        layers.append(nn.BatchNorm2d(out_channels))\n    \n    #Instance normalization\n    if instance_norm:\n        layers.append(nn.InstanceNorm2d(out_channels))\n    \n    #Dropout\n    if dropout:\n        layers.append(nn.Dropout2d(dropout_ratio))\n    \n    return nn.Sequential(*layers)","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.610Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class CycleGenerator(nn.Module):\n    \n    def __init__(self, conv_dim=64, n_res_blocks=6):\n        super(CycleGenerator, self).__init__()\n        \"\"\"\n        Input is RGB image (256x256x3) while output is a single value\n        \n        determine size = [(W−K+2P)/S]+1\n        W: input=256\n        K: kernel_size=4\n        P: padding=1\n        S: stride=2\n        \"\"\"\n        \n        #Encoder layers\n        self.conv1 = conv(in_channels=3, out_channels=conv_dim, kernel_size=4) # (128, 128, 64)\n        self.conv2 = conv(in_channels=conv_dim, out_channels=conv_dim*2, kernel_size=4, instance_norm=True) # (64, 64, 128)\n        self.conv3 = conv(in_channels=conv_dim*2, out_channels=conv_dim*4, kernel_size=4, instance_norm=True) # (32, 32, 256)\n        \n        #Residual blocks (number depends on input parameter)\n        res_layers = []\n        for layer in range(n_res_blocks):\n            res_layers.append(ResidualBlock(conv_dim*4))\n        self.res_blocks = nn.Sequential(*res_layers)\n        \n        #Decoder layers\n        self.deconv4 = deconv(in_channels=conv_dim*4, out_channels=conv_dim*2, kernel_size=4, instance_norm=True) # (64, 64, 128)\n        self.deconv5 = deconv(in_channels=conv_dim*2, out_channels=conv_dim, kernel_size=4, instance_norm=True) # (128, 128, 64)\n        self.deconv6 = deconv(in_channels=conv_dim, out_channels=3, kernel_size=4, instance_norm=True) # (256, 256, 3)\n        \n    def forward(self, x):\n        \"\"\"\n        Given an image x, returns a transformed image.\n        \"\"\"\n        \n        #Encoder\n        out = F.leaky_relu(self.conv1(x), negative_slope=0.2) # (128, 128, 64)\n        out = F.leaky_relu(self.conv2(out), negative_slope=0.2) # (64, 64, 128)\n        out = F.leaky_relu(self.conv3(out), negative_slope=0.2) # (32, 32, 256)\n        \n        #Residual blocks\n        out = self.res_blocks(out)\n        \n        #Decoder\n        out = F.leaky_relu(self.deconv4(out), negative_slope=0.2) # (64, 64, 128)\n        out = F.leaky_relu(self.deconv5(out), negative_slope=0.2) # (128, 128, 64)\n        out = torch.tanh(self.deconv6(out)) # (256, 256, 3)\n        \n        return out","metadata":{"trusted":true,"execution":{"execution_failed":"2024-12-27T14:18:56.610Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from torch.nn import init\ndef weights_init_normal(m):\n    \"\"\"\n    Applies initial weights to certain layers in a model.\n    The weights are taken from a normal distribution with mean = 0, std dev = 0.02.\n    Param m: A module or layer in a network    \n    \"\"\"\n    #classname will be something like: `Conv`, `BatchNorm2d`, `Linear`, etc.\n    classname = m.__class__.__name__\n    \n    #normal distribution with given paramters\n    std_dev = 0.02\n    mean = 0.0\n    \n    # Initialize conv layer\n    if hasattr(m, 'weight') and (classname.find('Conv') != -1):\n        init.normal_(m.weight.data, mean, std_dev)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def build_model(g_conv_dim=64, d_conv_dim=64, n_res_blocks=6):\n    \"\"\"\n    Builds generators G_XtoY & G_YtoX and discriminators D_X & D_Y \n    \"\"\"\n    \n    #Generators\n    G_XtoY = CycleGenerator(conv_dim=g_conv_dim, n_res_blocks=n_res_blocks)\n    G_YtoX = CycleGenerator(conv_dim=g_conv_dim, n_res_blocks=n_res_blocks)\n    \n    #Discriminators\n    D_X = Discriminator(conv_dim=d_conv_dim) # Y-->X\n    D_Y = Discriminator(conv_dim=d_conv_dim) # X-->Y\n    \n    #Weight initialization\n    G_XtoY.apply(weights_init_normal)\n    G_YtoX.apply(weights_init_normal)\n    D_X.apply(weights_init_normal)\n    D_Y.apply(weights_init_normal)\n    \n    #Moves models to GPU, if available\n    if torch.cuda.is_available():\n        device = torch.device(\"cuda:0\")\n        G_XtoY.to(device)\n        G_YtoX.to(device)\n        D_X.to(device)\n        D_Y.to(device)\n        print('Models moved to GPU.')\n    else:\n        print('Only CPU available.')\n\n    return G_XtoY, G_YtoX, D_X, D_Y","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Function call\nG_XtoY, G_YtoX, D_X, D_Y = build_model()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Check model structure\ndef print_build(G_XtoY, G_YtoX, D_X, D_Y):\n    print(\"                     G_XtoY                    \")\n    print(\"-----------------------------------------------\")\n    print(G_XtoY)\n    print()\n\n    print(\"                     G_YtoX                    \")\n    print(\"-----------------------------------------------\")\n    print(G_YtoX)\n    print()\n\n    print(\"                      D_X                      \")\n    print(\"-----------------------------------------------\")\n    print(D_X)\n    print()\n\n    print(\"                      D_Y                      \")\n    print(\"-----------------------------------------------\")\n    print(D_Y)\n    print()\n    \nprint_build(G_XtoY, G_YtoX, D_X, D_Y)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"> # **Loss function**","metadata":{}},{"cell_type":"code","source":"def real_mse_loss(D_out, adverserial_weight=1):\n    #how close is the produced output from being \"real\"?\n    mse_loss = torch.mean((D_out-1)**2)*adverserial_weight\n    return mse_loss\n\ndef fake_mse_loss(D_out, adverserial_weight=1):\n    #how close is the produced output from being \"false\"?\n    mse_loss = torch.mean(D_out**2)*adverserial_weight\n    return mse_loss\n\ndef cycle_consistency_loss(real_img, reconstructed_img, lambda_weight=1):\n    reconstr_loss = torch.mean(torch.abs(real_img - reconstructed_img))\n    return lambda_weight*reconstr_loss \n\ndef identity_loss(real_img, generated_img, identity_weight=1):\n    ident_loss = torch.mean(torch.abs(real_img - generated_img))\n    return identity_weight*ident_loss","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Optimizers**","metadata":{}},{"cell_type":"code","source":"import torch.optim as optim\n\n#hyperparameter\nlr=0.0002 #0.0002\nbeta1=0.500 #exponential decay rate for the first moment estimates\nbeta2=0.999 #exponential decay rate for the second-moment estimates\ng_params = list(G_XtoY.parameters()) + list(G_YtoX.parameters())\n\n#Optimizers for generator and discriminator\ng_optimizer = optim.Adam(g_params, lr, [beta1, beta2])\nd_x_optimizer = optim.Adam(D_X.parameters(), lr, [beta1, beta2])\nd_y_optimizer = optim.Adam(D_Y.parameters(), lr, [beta1, beta2])","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Training**","metadata":{}},{"cell_type":"code","source":"def training_loop(dataloader_X, dataloader_Y, test_dataloader_X, test_dataloader_Y, n_epochs=1000):\n    \n    # Losses over time\n    losses = []\n    \n    # Additional weighting parameters (in reality only 2 are required as the third is kind of \"given relatively\" by the other two)\n    adverserial_weight = 0.5\n    lambda_weight = 10\n    identity_weight = 5\n    \n    # Get some fixed data from domains X and Y for sampling. Images are held constant throughout training and allow us to inspect the model's performance.\n    test_iter_X = iter(test_dataloader_X)\n    test_iter_Y = iter(test_dataloader_Y)\n    fixed_X = next(test_iter_X)  # Changed from test_iter_X.next() to next(test_iter_X)\n    fixed_Y = next(test_iter_Y)  # Changed from test_iter_Y.next() to next(test_iter_Y)\n    \n    # Batches per epoch\n    iter_X = iter(dataloader_X)\n    iter_Y = iter(dataloader_Y)\n    batches_per_epoch = min(len(dataloader_X), len(dataloader_Y))\n    \n    # Average loss over batches per epoch runs\n    d_total_loss_avg = 0.0\n    g_total_loss_avg = 0.0\n    \n    # Loop through epochs\n    for epoch in range(1, n_epochs+1):\n        \n        # Reset iterators for each epoch\n        if epoch % batches_per_epoch == 0:\n            iter_X = iter(dataloader_X)\n            iter_Y = iter(dataloader_Y)\n        \n        # Get images from domain X\n        images_X = next(iter_X)  # Changed from iter_X.next() to next(iter_X)\n        \n        # Get images from domain Y\n        images_Y = next(iter_Y)  # Changed from iter_Y.next() to next(iter_Y)\n        \n        # Move images to GPU if available (otherwise stay on CPU)\n        device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n        images_X = images_X.to(device)\n        images_Y = images_Y.to(device)\n        \n        # ============================================\n        #            TRAIN THE DISCRIMINATORS\n        # ============================================\n        \n        # --------------------------------------------\n        ## First: D_X, real and fake loss components\n        # --------------------------------------------\n        \n        # Train with real images\n        d_x_optimizer.zero_grad()\n        \n        # 1. Compute the discriminator losses on real images\n        out_x = D_X(images_X)\n        D_X_real_loss = real_mse_loss(out_x, adverserial_weight)\n        \n        # Train with fake images\n        # 2. Generate fake images that look like domain X based on real images in domain Y\n        fake_X = G_YtoX(images_Y)\n\n        # 3. Compute the fake loss for D_X\n        out_x = D_X(fake_X)\n        D_X_fake_loss = fake_mse_loss(out_x, adverserial_weight)\n        \n        # 4. Compute the total loss and perform backpropagation\n        d_x_loss = D_X_real_loss + D_X_fake_loss\n        d_x_loss.backward()\n        d_x_optimizer.step()\n        \n        # --------------------------------------------\n        ## Second: D_Y, real and fake loss components\n        # --------------------------------------------\n        \n        # Train with real images\n        d_y_optimizer.zero_grad()\n        \n        # 1. Compute the discriminator losses on real images\n        out_y = D_Y(images_Y)\n        D_Y_real_loss = real_mse_loss(out_y, adverserial_weight)\n        \n        # Train with fake images\n        # 2. Generate fake images that look like domain Y based on real images in domain X\n        fake_Y = G_XtoY(images_X)\n        \n        # 3. Compute the fake loss for D_Y\n        out_y = D_Y(fake_Y)\n        D_Y_fake_loss = fake_mse_loss(out_y, adverserial_weight)\n        \n        # 4. Compute the total loss and perform backprop\n        d_y_loss = D_Y_real_loss + D_Y_fake_loss\n        d_y_loss.backward()\n        d_y_optimizer.step()\n        \n        # 5. Compute total discriminator loss\n        d_total_loss = D_X_real_loss + D_X_fake_loss + D_Y_real_loss + D_Y_fake_loss\n        \n        # =========================================\n        #            TRAIN THE GENERATORS\n        # =========================================\n        \n        # --------------------------------------------\n        ## First: generate fake X images and reconstructed Y images\n        # --------------------------------------------\n        \n        # Back to the start\n        g_optimizer.zero_grad()\n        \n        # 1. Generate fake images that look like domain X based on real images in domain Y\n        fake_X = G_YtoX(images_Y)\n        \n        # 2. Compute the generator loss based on domain X\n        out_x = D_X(fake_X)\n        g_YtoX_loss = real_mse_loss(out_x, adverserial_weight)\n\n        # 3. Create a reconstructed y\n        reconstructed_Y = G_XtoY(fake_X)\n        \n        # 4. Compute the cycle consistency loss (the reconstruction loss)\n        reconstructed_y_loss = cycle_consistency_loss(images_Y, reconstructed_Y, lambda_weight=lambda_weight)\n        \n        # 5. Compute the identity loss from transformation Y-->X\n        identity_y_loss = identity_loss(images_Y, fake_X, identity_weight=identity_weight)\n        \n        # --------------------------------------------\n        ## Second: generate fake Y images and reconstructed X images\n        # --------------------------------------------\n        \n        # 1. Generate fake images that look like domain Y based on real images in domain X\n        fake_Y = G_XtoY(images_X)\n        \n        # 2. Compute the generator loss based on domain Y\n        out_y = D_Y(fake_Y)\n        g_XtoY_loss = real_mse_loss(out_y, adverserial_weight)\n        \n        # 3. Create a reconstructed x\n        reconstructed_X = G_YtoX(fake_Y)\n        \n        # 4. Compute the cycle consistency loss (the reconstruction loss)\n        reconstructed_x_loss = cycle_consistency_loss(images_X, reconstructed_X, lambda_weight=lambda_weight)\n        \n        # 5. Compute the identity loss from transformation X-->Y\n        identity_x_loss = identity_loss(images_X, fake_Y, identity_weight=identity_weight)\n        \n        # 6. Add up all generator and reconstructed losses and perform backprop\n        g_total_loss = g_YtoX_loss + g_XtoY_loss + reconstructed_y_loss + reconstructed_x_loss + identity_y_loss + identity_x_loss\n        g_total_loss.backward()\n        g_optimizer.step()\n        \n        # =========================================\n        # Admin \n        # =========================================\n        \n        # Average loss\n        d_total_loss_avg = d_total_loss_avg + d_total_loss / batches_per_epoch\n        g_total_loss_avg = g_total_loss_avg + g_total_loss / batches_per_epoch\n        \n        # Print log info\n        print_every = batches_per_epoch\n        if epoch % print_every == 0:\n            # Append real and fake discriminator losses and the generator loss\n            losses.append((d_total_loss_avg.item(), g_total_loss_avg.item()))\n            true_epoch_n = int(epoch / batches_per_epoch)\n            true_epoch_total = int(n_epochs / batches_per_epoch)\n            print(f'Epoch [{true_epoch_n:5d}/{true_epoch_total:5d}] | d_total_loss_avg: {d_total_loss_avg.item():6.4f} | g_total_loss: {g_total_loss_avg.item():6.4f}')\n        \n        # Show the generated samples\n        show_every = (batches_per_epoch * 10)\n        if epoch % show_every == 0:\n            # Set generators to eval mode for image generation\n            G_YtoX.eval()\n            G_XtoY.eval()\n            test_images = show_test(fixed_Y, fixed_X, G_YtoX, G_XtoY)\n            # Set generators to train mode to continue training\n            G_YtoX.train()\n            G_XtoY.train()\n        \n        # Reset average loss for each epoch\n        if epoch % batches_per_epoch == 0:\n            d_total_loss_avg = 0.0\n            g_total_loss_avg = 0.0\n    \n    return losses","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"batches_per_epoch = min(len(dataloader_X), len(dataloader_Y))\nepoch_true = 256","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"n_epochs = epoch_true * batches_per_epoch\nlosses = training_loop(dataloader_X, dataloader_Y, test_dataloader_X, test_dataloader_Y, n_epochs=n_epochs)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Plot loss functions over training\nfig, ax = plt.subplots(figsize=(12,8))\nlosses = np.array(losses)\nplt.plot(losses.T[0], label='Discriminators', alpha=0.5)\nplt.plot(losses.T[1], label='Generators', alpha=0.5)\nplt.title(\"Training Losses\")\nplt.legend()\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Get some fixed data from domains X and Y for sampling. These are images that are held constant throughout training, that allow us to inspect the model's performance.\ntest_iter_X = iter(test_dataloader_X)\ntest_iter_Y = iter(test_dataloader_Y)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Sampling\nfixed_X = next(test_iter_X)  # Corrected\nfixed_Y = next(test_iter_Y)  # Corrected\n\n# Evaluation\nG_XtoY.eval()\n\nmean_ = 0.5\nstd_ = 0.5\n\n# Identify correct device\ndevice = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n\n# Create fake pictures for both cycles\nfake_Y = G_XtoY(fixed_X.to(device))\n\n# Generate grids\ngrid_x = make_grid(fixed_X, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\ngrid_fake_y = make_grid(fake_Y, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\n\n# Normalize pictures to pixel range from 0 to 255\nX, fake_Y_ = reverse_normalize(grid_x, mean_, std_), reverse_normalize(grid_fake_y, mean_, std_)\n\n# Transformation from X -> Y\nfig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, sharey=True, figsize=(30, 20))\nax1.imshow(X)\nax1.axis('off')\nax1.set_title('X')\nax2.imshow(fake_Y_)\nax2.axis('off')\nax2.set_title('Fake Y, Monet-esque')\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Corrected Sampling\nfixed_Y = next(test_iter_Y)  # Fixed\nfixed_Y = next(test_iter_Y)  # Fixed\nfixed_Y = next(test_iter_Y)  # Fixed\n\n# Evaluation\nG_YtoX.eval()\n\nmean_ = 0.5\nstd_ = 0.5\n\n# Identify correct device\ndevice = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n    \n# Create fake pictures for both cycles\nfake_X = G_YtoX(fixed_Y.to(device))\n    \n# Generate grids\ngrid_y = make_grid(fixed_Y, nrow=4).permute(1, 2, 0).detach().cpu().numpy()\ngrid_fake_x = make_grid(fake_X, nrow=4).permute(1, 2, 0).detach().cpu().numpy()  \n    \n# Normalize pictures to pixel range from 0 to 255\nY, fake_X_ = reverse_normalize(grid_y, mean_, std_), reverse_normalize(grid_fake_x, mean_, std_)\n\n# Transformation from Y -> X\nfig, (ax1, ax2) = plt.subplots(2, 1, sharex=True, sharey=True, figsize=(30, 20))\nax1.imshow(Y)\nax1.axis('off')\nax1.set_title('Y')\nax2.imshow(fake_X_)\nax2.axis('off')\nax2.set_title('Fake X')\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **Submission**","metadata":{}},{"cell_type":"code","source":"# Submission\n!mkdir ../images","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Set model to evaluation\nG_XtoY.eval()\n\n# Get data loader for final transformation / submission\nsubmit_dataloader = DataLoader(dataset_photo, batch_size=1, shuffle=False, pin_memory=True)\ndataiter = iter(submit_dataloader)\n\n# Previous normalization chosen\nmean_ = 0.5 \nstd_ = 0.5\n\n# Loop through each picture\nfor image_idx in range(len(submit_dataloader)):\n    # Get base picture\n    fixed_X = next(dataiter)  # Corrected\n\n    # Identify correct device\n    device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n\n    # Create fake pictures (Monet-esque)\n    fake_Y = G_XtoY(fixed_X.to(device))\n    fake_Y = fake_Y.detach().cpu().numpy()\n    fake_Y = reverse_normalize(fake_Y, mean_, std_)\n    fake_Y = fake_Y[0].transpose(1, 2, 0)\n    fake_Y = np.uint8(fake_Y)\n    fake_Y = Image.fromarray(fake_Y)\n\n    # Save picture\n    fake_Y.save(f\"../images/{image_idx}.jpg\")\n\n# Back to training\nG_XtoY.train()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import shutil\nshutil.make_archive(\"/kaggle/working/images\", 'zip', \"/kaggle/images\")\n# shutil.rmtree('../images')","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}