Project_4_Git.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, we navigate all directories within 'Parking_Lot_Vehicle_Images_UFPR04' to find and copy all images into a newly created directory named 'all_images'."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import shutil\n",
    "\n",
    "curr = os.getcwd()\n",
    "newDir = os.path.join(curr,'all_images')\n",
    "#print(newDir)\n",
    "\n",
    "if not os.path.exists(newDir):\n",
    "    os.makedirs(newDir)\n",
    "    \n",
    "imgPath = os.path.join(curr,'Parking_Lot_Vehicle_Images_UFPR04')\n",
    "#Need to set dirUFPR04 to filepath of UFPR04 folder\n",
    "dirUFPR04 = os.path.join(imgPath,'UFPR04')\n",
    "#print(dirUFPR04)   \n",
    "\n",
    "\n",
    "for mainDir in os.listdir(dirUFPR04):\n",
    "    weatherDir = os.path.join(dirUFPR04, mainDir)\n",
    "    for dateDir in os.listdir(weatherDir):\n",
    "        occDir = os.path.join(weatherDir, dateDir)\n",
    "        for dir in os.listdir(occDir):\n",
    "            imageDir = os.path.join(occDir, dir)\n",
    "            for file in os.listdir(imageDir):\n",
    "                if file.endswith('.jpg'):\n",
    "                    fullpath = os.path.join(imageDir, file)\n",
    "                    shutil.copy(fullpath, newDir)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, we initialize a custom pytorch Dataset class named 'VehicleDataset' for our images."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import torch\n",
    "from skimage import io, transform\n",
    "from PIL import Image\n",
    "import natsort\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from torchvision import transforms, utils\n",
    "\n",
    "class VehicleDataset(Dataset):\n",
    "    def __init__(self, main_dir, transform=None):\n",
    "        self.main_dir = main_dir\n",
    "        self.transform = transform\n",
    "        all_imgs = os.listdir(main_dir)\n",
    "        self.total_imgs = natsort.natsorted(all_imgs)\n",
    "\n",
    "    def __len__(self):\n",
    "        return len(self.total_imgs)\n",
    "\n",
    "    def __getitem__(self, idx):\n",
    "        img_loc = os.path.join(self.main_dir, self.total_imgs[idx])\n",
    "        image = Image.open(img_loc).convert(\"RGB\")\n",
    "        #image = io.imread(img_loc)\n",
    "        tensor_image = self.transform(image)\n",
    "        return tensor_image"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We then initialize the torchvision transforms object that will resize the image data to 50x50, transform it to tensors, and normalize it between -1 and 1. We apply the transformations to the VehicleDatset object, 'dataset', and pass it into the torch DataLoader class with batch size 100 and shuffle set to True."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import torch\n",
    "from torch import nn, optim\n",
    "from torch.autograd.variable import Variable\n",
    "#from torchvision import transforms\n",
    "#from PIL import Image\n",
    "\n",
    "#import import_ipynb\n",
    "#from CustomDataset import VehicleDataset\n",
    "\n",
    "from utils import Logger # this is used to visualize generator data and print status logs during training\n",
    "\n",
    "trsfm = transforms.Compose([\n",
    "        transforms.Resize((50,50),interpolation=Image.NEAREST),\n",
    "        transforms.ToTensor(),\n",
    "        transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))]) \n",
    "\n",
    "\n",
    "dataset = VehicleDataset(main_dir=newDir, transform=trsfm)\n",
    "data_loader = torch.utils.data.DataLoader(dataset, batch_size=100, shuffle=True)\n",
    "\n",
    "# check if cuda is available and use it in the '.to(device)' methods used later if it is. If not, use cpu in the method.\n",
    "use_cuda = torch.cuda.is_available()\n",
    "device = torch.device(\"cuda\" if use_cuda else \"cpu\")\n",
    "\n",
    "#number of batches\n",
    "num_batches = len(data_loader)\n",
    "print('number of batches:', num_batches)\n",
    "\n",
    "from itertools import islice\n",
    "\n",
    "# verifying contents of data_loader is as expected\n",
    "print('Using DataLoader to show data: ')\n",
    "for i, img in enumerate(islice(data_loader, 5)):\n",
    "    print('Batch:', i, ',Size:', img.size(), '\\nTensor: ', img[0,:,0:3,0:3])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Defining the Discriminator neural network."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class DiscriminatorNet(torch.nn.Module):\n",
    "    \"\"\"\n",
    "    A three hidden-layer discriminative neural network\n",
    "    \"\"\"\n",
    "    def __init__(self):\n",
    "        super(DiscriminatorNet, self).__init__()\n",
    "        n_features = 7500\n",
    "        n_out = 1\n",
    "        \n",
    "        self.hidden0 = nn.Sequential( \n",
    "            nn.Linear(n_features, 1024),\n",
    "            nn.LeakyReLU(0.2),\n",
    "            nn.Dropout(0.3)\n",
    "        )\n",
    "        self.hidden1 = nn.Sequential(\n",
    "            nn.Linear(1024, 512),\n",
    "            nn.LeakyReLU(0.2),\n",
    "            nn.Dropout(0.3)\n",
    "        )\n",
    "        self.hidden2 = nn.Sequential(\n",
    "            nn.Linear(512, 256),\n",
    "            nn.LeakyReLU(0.2),\n",
    "            nn.Dropout(0.3)\n",
    "        )\n",
    "        self.out = nn.Sequential(\n",
    "            torch.nn.Linear(256, n_out),\n",
    "            torch.nn.Sigmoid()\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.hidden0(x)\n",
    "        x = self.hidden1(x)\n",
    "        x = self.hidden2(x)\n",
    "        x = self.out(x)\n",
    "        return x\n",
    "    \n",
    "discriminator = DiscriminatorNet()\n",
    "discriminator.to(device)\n",
    "\n",
    "print(discriminator)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Methods for transforming 4d image tensors to flattened image tensors and vice versa."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def images_to_tensors(images):\n",
    "    return images.view(images.size(0), 7500)\n",
    "\n",
    "def tensors_to_images(tensors):\n",
    "    return tensors.view(tensors.size(0), 3, 50, 50)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Defining the Generator neural network."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class GeneratorNet(torch.nn.Module):\n",
    "    \"\"\"\n",
    "    A three hidden-layer generative neural network\n",
    "    \"\"\"\n",
    "    def __init__(self):\n",
    "        super(GeneratorNet, self).__init__()\n",
    "        n_features = 1000\n",
    "        n_out = 7500\n",
    "        \n",
    "        self.hidden0 = nn.Sequential(\n",
    "            nn.Linear(n_features, 256),\n",
    "            nn.LeakyReLU(0.2)\n",
    "        )\n",
    "        self.hidden1 = nn.Sequential(            \n",
    "            nn.Linear(256, 512),\n",
    "            nn.LeakyReLU(0.2)\n",
    "        )\n",
    "        self.hidden2 = nn.Sequential(\n",
    "            nn.Linear(512, 1024),\n",
    "            nn.LeakyReLU(0.2)\n",
    "        )\n",
    "        \n",
    "        self.out = nn.Sequential(\n",
    "            nn.Linear(1024, n_out),\n",
    "            nn.Tanh()\n",
    "        )\n",
    "\n",
    "    def forward(self, x):\n",
    "        x = self.hidden0(x)\n",
    "        x = self.hidden1(x)\n",
    "        x = self.hidden2(x)\n",
    "        x = self.out(x)\n",
    "        return x\n",
    "    \n",
    "generator = GeneratorNet()\n",
    "generator.to(device)\n",
    "\n",
    "# Loads pretrained generator weights\n",
    "# #path = os.path.join(os.getcwd(), 'data', 'models')\n",
    "\n",
    "# if os.path.exists(path):\n",
    "#     GeneratorNet.load_state_dict(torch.load(path))\n",
    "\n",
    "print(generator)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Method for creating random noise corresponding to shape (batch size, number of inputs to generator)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def noise(size):\n",
    "    '''\n",
    "    Generates a 1-d vector of gaussian sampled random values\n",
    "    '''\n",
    "    n = Variable(torch.randn(size, 1000))\n",
    "    return n.to(device)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Initializing adam optimizers for discriminator and generator networks with learning rate of 0.0002 and Binary Cross Entropy loss function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "d_optimizer = optim.Adam(discriminator.parameters(), lr=0.0002)\n",
    "g_optimizer = optim.Adam(generator.parameters(), lr=0.0002)\n",
    "\n",
    "loss = nn.BCELoss()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Methods for creating targets of ones for real data and targets of zeroes for fake generator output."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ones_target(size):\n",
    "    '''\n",
    "    Tensor containing ones, with shape = size\n",
    "    '''\n",
    "    data = Variable(torch.ones(size, 1))\n",
    "    return data.to(device)\n",
    "\n",
    "def zeros_target(size):\n",
    "    '''\n",
    "    Tensor containing zeros, with shape = size\n",
    "    '''\n",
    "    data = Variable(torch.zeros(size, 1))\n",
    "    return data.to(device)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Function for training Discriminator network."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def train_discriminator(optimizer, real_data, fake_data):\n",
    "    N = real_data.size(0)\n",
    "    # Reset gradients\n",
    "    optimizer.zero_grad()\n",
    "    \n",
    "    # 1.1 Train on Real Data\n",
    "    prediction_real = discriminator(real_data)\n",
    "    # Calculate error and backpropagate\n",
    "    error_real = loss(prediction_real, ones_target(N) )\n",
    "    error_real.backward()\n",
    "\n",
    "    # 1.2 Train on Fake Data\n",
    "    prediction_fake = discriminator(fake_data)\n",
    "    # Calculate error and backpropagate\n",
    "    error_fake = loss(prediction_fake, zeros_target(N))\n",
    "    error_fake.backward()\n",
    "    \n",
    "    # 1.3 Update weights with gradients\n",
    "    optimizer.step()\n",
    "    \n",
    "    # Return error and predictions for real and fake inputs\n",
    "    return error_real + error_fake, prediction_real, prediction_fake"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Function for training Generator network."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def train_generator(optimizer, fake_data):\n",
    "    N = fake_data.size(0)\n",
    "    # Reset gradients\n",
    "    optimizer.zero_grad()\n",
    "    # Sample noise and generate fake data\n",
    "    prediction = discriminator(fake_data)\n",
    "    # Calculate error and backpropagate\n",
    "    error = loss(prediction, ones_target(N))\n",
    "    error.backward()\n",
    "    # Update weights with gradients\n",
    "    optimizer.step()\n",
    "    # Return error\n",
    "    return error"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Creating 16 samples of noise data to visualize 16 generated images using tensorboard logger."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_test_samples = 16\n",
    "test_noise = noise(num_test_samples)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, training the GAN and displaying progress using Logger functions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create logger instance\n",
    "logger = Logger(model_name='VGAN', data_name='Parked Vehicles')\n",
    "\n",
    "# Total number of epochs to train\n",
    "num_epochs = 200\n",
    "\n",
    "g_losses = []\n",
    "d_losses = []\n",
    "\n",
    "for epoch in range(num_epochs):\n",
    "    \n",
    "    for n_batch, (real_batch) in enumerate(data_loader):\n",
    "        N = real_batch.size(0)\n",
    "        # 1. Train Discriminator\n",
    "        real_data = Variable(images_to_tensors(real_batch)).to(device)\n",
    "        # Generate fake data and detach \n",
    "        # (so gradients are not calculated for generator)\n",
    "        fake_data = generator(noise(N)).detach()\n",
    "        # Train D\n",
    "        d_error, d_pred_real, d_pred_fake = \\\n",
    "              train_discriminator(d_optimizer, real_data, fake_data)\n",
    "\n",
    "        # 2. Train Generator\n",
    "        # Generate fake data\n",
    "        fake_data = generator(noise(N))\n",
    "        # Train G\n",
    "        g_error = train_generator(g_optimizer, fake_data)\n",
    "        # Log batch error\n",
    "        logger.log(d_error, g_error, epoch, n_batch, num_batches)\n",
    "        \n",
    "        # Display Progress every few batches\n",
    "        if (n_batch) % 50 == 0: \n",
    "            test_images = tensors_to_images(generator(test_noise)).cpu().detach()\n",
    "            test_images = test_images.data\n",
    "            logger.log_images(\n",
    "                test_images, num_test_samples, \n",
    "                epoch, n_batch, num_batches\n",
    "            );\n",
    "            # Display status Logs\n",
    "            logger.display_status(\n",
    "                epoch, num_epochs, n_batch, num_batches,\n",
    "                d_error, g_error, d_pred_real, d_pred_fake\n",
    "            )\n",
    "            # record generator and discriminator losses every 50 batch for plotting \n",
    "            g_losses.append(g_error/n_batch)\n",
    "            d_losses.append(d_error/n_batch)\n",
    "        \n",
    "        # saves generator weights every 50 epochs\n",
    "#         if(epoch > 49):\n",
    "#             path = os.path.join(os.getcwd(), 'data', 'models')\n",
    "#             logger._make_dir(path)\n",
    "#             if (epoch) % 50 == 0:\n",
    "#                 torch.save(GeneratorNet.state_dict(),\n",
    "#                            '%s\\G_epoch_%s' % (path,epoch))\n",
    "        \n",
    "print('\\nTraining Finished')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Plotting Generator and Discriminator losses per epoch."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from matplotlib import pyplot as plt\n",
    "\n",
    "plt.plot(g_losses, label='Generator_Losses')\n",
    "plt.plot(d_losses, label='Discriminator Losses')\n",
    "plt.legend()\n",
    "plt.savefig('loss.png')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}