Verification_RNCAPSNET.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from __future__ import print_function, division\n",
    "import torch\n",
    "import torch.nn as nn\n",
    "import torch.optim as optim\n",
    "from torch.optim import lr_scheduler\n",
    "from torch.optim import Adam\n",
    "import torch.nn.functional as F\n",
    "\n",
    "import csv\n",
    "from skimage import io\n",
    "\n",
    "from PIL import Image\n",
    "import pandas as pd\n",
    "\n",
    "import numpy as np\n",
    "import torchvision\n",
    "from torchvision import datasets, models, transforms\n",
    "from torch.utils.data import Dataset, DataLoader\n",
    "from torch.autograd import Variable\n",
    "from torchsummary import summary\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "import matplotlib as mpl\n",
    "import time\n",
    "import os\n",
    "import copy\n",
    "\n",
    "import import_ipynb\n",
    "import ResNetCaps_E\n",
    "import AT_T_triplet_generator\n",
    "import LFW_triplet_generator\n",
    "\n",
    "verbose = False\n",
    "USE_CUDA = True\n",
    "device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
    "LFW_use = True\n",
    "ATET_use = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "dataset_transform = transforms.Compose([\n",
    "    transforms.Resize((224,224)),\n",
    "    transforms.ToTensor(),        \n",
    "    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.247, 0.243, 0.261))\n",
    "])\n",
    "batch_size = 100\n",
    "\n",
    "if ATET_use:\n",
    "    folder = \"/home/rita/JupyterProjects/EYE-SEA/DataSets/Verification/ATeT_faces/orl_faces/\"\n",
    "    triplet_generator = AT_T_triplet_generator.AT_T_TripletGenerator(folder, transform = dataset_transform,hold_out_positive=True)\n",
    "    dataLoader_generator = torch.utils.data.DataLoader(triplet_generator,batch_size=batch_size, shuffle=True)\n",
    "    triplet_generator_test = AT_T_triplet_generator.AT_T_TripletGenerator(folder,train = False, transform = dataset_transform,hold_out_positive=True)\n",
    "    dataLoader_generator_test = torch.utils.data.DataLoader(triplet_generator_test,batch_size=batch_size, shuffle=True)\n",
    "if LFW_use:\n",
    "    folder = \"/home/rita/JupyterProjects/EYE-SEA/DataSets/Verification/lfw\"\n",
    "    triplet_generator = LFW_triplet_generator.LFW_TripletGenerator(folder, transform = dataset_transform,hold_out_positive = True)\n",
    "    dataLoader_generator = torch.utils.data.DataLoader(triplet_generator,batch_size=batch_size, shuffle=True)\n",
    "    triplet_generator_test = LFW_triplet_generator.LFW_TripletGenerator(folder,train = False, transform = dataset_transform,hold_out_positive=True)\n",
    "    dataLoader_generator_test = torch.utils.data.DataLoader(triplet_generator_test,batch_size=batch_size, shuffle=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#TripletLoss layer \n",
    "class TripletLossLayer(torch.nn.Module):\n",
    "    def __init__(self,alpha):\n",
    "        super(TripletLossLayer, self).__init__()\n",
    "        self.ALPHA = alpha\n",
    "        \n",
    "    def triplet_loss(self,a,p,n):\n",
    "        p_l2 = a-p\n",
    "        p_dist = torch.sum(torch.mul(p_l2,p_l2))\n",
    "        #print(\"p_dist {}\".format(p_dist))\n",
    "        n_l2 = a-n\n",
    "        n_dist = torch.sum(torch.mul(n_l2,n_l2))\n",
    "        #print(\"n_dist {}\".format(n_dist))\n",
    "        zero=torch.zeros([1,1], dtype=torch.float32, device=device)\n",
    "        \n",
    "        return [torch.max(p_dist-n_dist+self.ALPHA,zero),p_dist,n_dist]\n",
    "    \n",
    "    def forward(self,a,p,n):\n",
    "        loss, p_dist, n_dist = self.triplet_loss(a,p,n)\n",
    "        self.loss = loss\n",
    "        return loss, p_dist, n_dist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ResNetCaps_E.ResNetCaps_E()\n",
    "model = model.to(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()),lr = 0.001)\n",
    "criterion = TripletLossLayer(0.2)\n",
    "n_epochs = 10\n",
    "\n",
    "p_dist_list_l = []\n",
    "n_dist_list_l = []\n",
    "\n",
    "p_dist_list_b = []\n",
    "n_dist_list_b = []\n",
    "loss_list_b = []\n",
    "\n",
    "for epoch in range(n_epochs): \n",
    "    print('epoch {}:{}'.format(epoch+1, n_epochs)) \n",
    "    model.train()\n",
    "    loss_collect = 0\n",
    "    p_dist_collect = 0\n",
    "    n_dist_collect = 0\n",
    "    p_dist_list = []\n",
    "    n_dist_list = []\n",
    "    for batch_id, (in_a, in_p, in_n)  in enumerate(dataLoader_generator):\n",
    "    #in_a, in_p, in_n = next(iter(dataLoader_generator))\n",
    "        in_a = in_a.to(device)\n",
    "        in_p = in_p.to(device)\n",
    "        in_n = in_n.to(device) \n",
    "        #Compute embeddings for anchor, positive, and negative images\n",
    "\n",
    "        emb_a = model(in_a)\n",
    "        emb_p = model(in_p)\n",
    "        emb_n = model(in_n)\n",
    "\n",
    "        optimizer.zero_grad()\n",
    "\n",
    "        loss,p_dist,n_dist = criterion(emb_a,emb_p,emb_n)\n",
    "        loss_collect +=loss\n",
    "        p_dist_collect +=  p_dist\n",
    "        n_dist_collect +=  n_dist\n",
    "\n",
    "                   \n",
    "        if batch_id % 100 == 0:   \n",
    "            print(\"p_dist {} n_dist {}\".format(p_dist,n_dist))\n",
    "            print(\"loss per batch {}\".format(loss))\n",
    "            p_dist_list.append(p_dist)\n",
    "            n_dist_list.append(n_dist)\n",
    "\n",
    "        loss.backward()\n",
    "        optimizer.step()\n",
    "        \n",
    "    \n",
    "    p_dist_list_l.append(p_dist_list)\n",
    "    n_dist_list_l.append(n_dist_list)    \n",
    "\n",
    "             \n",
    "    p_dist_list_b.append(p_dist_collect/batch_id)\n",
    "    n_dist_list_b.append(n_dist_collect/batch_id)\n",
    "    loss_list_b.append(loss_collect/batch_id)\n",
    "        \n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cmap_positive = mpl.cm.summer\n",
    "cmap_negative = mpl.cm.autumn\n",
    "plt.figure(figsize=(10,10))\n",
    "for it_l in range(0,n_epochs):\n",
    "    plt.plot(np.arange(1,len(p_dist_list_l[it_l])+1), p_dist_list_l[it_l],'+',color=cmap_positive(it_l / float(n_epochs)))\n",
    "    plt.plot(np.arange(1,len(n_dist_list_l[it_l])+1), n_dist_list_l[it_l],'*',color=cmap_negative(it_l / float(n_epochs)))\n",
    "\n",
    "plt.xlabel('batchs')\n",
    "plt.ylabel('p_dist,n_dist')\n",
    "plt.title('Training phase')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = np.arange(1,n_epochs+1)\n",
    "plt.plot(epochs, p_dist_list_b, color='g')\n",
    "plt.plot(epochs, n_dist_list_b, color='orange')\n",
    "plt.xlabel('epochs')\n",
    "plt.ylabel('p_dist,n_dist')\n",
    "plt.title('Training phase')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.plot(epochs, loss_list_b, color='pink')\n",
    "plt.xlabel('epochs')\n",
    "plt.ylabel('loss')\n",
    "plt.title('Training phase')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "batch_size = 4\n",
    "triplet_generator_test = AT_T_triplet_generator.AT_T_TripletGenerator(folder,train = False, transform = dataset_transform,hold_out_positive=True)\n",
    "dataLoader_generator_test = torch.utils.data.DataLoader(triplet_generator_test,batch_size=batch_size, shuffle=True)\n",
    "size_test = len(dataLoader_generator_test)\n",
    "#Test \n",
    "p_dist_list_Test =[]\n",
    "n_dist_list_Test =[]\n",
    "for i,(in_a,in_p,in_n) in enumerate(dataLoader_generator_test):\n",
    " \n",
    "\n",
    "    for i in range(0,len(in_a)):\n",
    "        in_a_s = in_a[i,:,:,:]\n",
    "        in_p_s = in_p[i,:,:,:]\n",
    "        in_n_s = in_n[i,:,:,:]\n",
    "\n",
    "        #in_a, in_p, in_n = next(iter(dataLoader_generator))\n",
    "        in_a_s = in_a_s.to(device)\n",
    "        in_p_s = in_p_s.to(device)\n",
    "        in_n_s = in_n_s.to(device) \n",
    "        #Compute embeddings for anchor, positive, and negative images\n",
    "\n",
    "        emb_a = model(in_a_s.unsqueeze(0))\n",
    "        emb_p = model(in_p_s.unsqueeze(0))\n",
    "        emb_n = model(in_n_s.unsqueeze(0))\n",
    "\n",
    "        loss,p_dist,n_dist = criterion(emb_a,emb_p,emb_n)\n",
    "        p_dist_list_Test.append(p_dist)\n",
    "        n_dist_list_Test.append(n_dist)\n",
    "        \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(sum(p_dist_list_Test)/len(p_dist_list_Test))\n",
    "print(sum(n_dist_list_Test)/len(n_dist_list_Test))\n",
    "\n",
    "plt.plot(range(0,len(p_dist_list_Test)), p_dist_list_Test, color='pink')\n",
    "#plt.plot(range(0,len(n_dist_list)), n_dist_list, color='g')\n",
    "plt.xlabel('elements')\n",
    "plt.ylabel('p_dist_value')\n",
    "plt.title('test')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#plt.plot(range(0,len(p_dist_list)), p_dist_list, color='pink')\n",
    "plt.plot(range(0,len(n_dist_list_Test)), n_dist_list_Test, color='g')\n",
    "plt.xlabel('elements')\n",
    "plt.ylabel('n_dist_value')\n",
    "plt.title('test')\n",
    "plt.show() \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "threshold = 0.3\n",
    "positives_True =  0\n",
    "for i in p_dist_list_Test:\n",
    "    if i < threshold: positives_True += 1 \n",
    "negatives_True =  0\n",
    "for i in n_dist_list_Test:\n",
    "    if i > threshold: negatives_True += 1     \n",
    "\n",
    "print(\"TA({}) {} out of P_same {}\".format(threshold,positives_True,len(p_dist_list_Test)))\n",
    "print(\"FA({}) {} out of P_diff {}\".format(threshold,negatives_True,len(n_dist_list_Test)))\n",
    "print(\"VAL({}) (TP/P)  {}\".format(threshold,(positives_True/len(p_dist_list_Test))))\n",
    "print(\"FAR({}) (TN/N)  {}\".format(threshold,(negatives_True/len(n_dist_list_Test))))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Test only one batch\n",
    "\n",
    "in_a,in_p,in_n = next(iter(dataLoader_generator_test))\n",
    "\n",
    "p_dist_list =[]\n",
    "n_dist_list =[]\n",
    "\n",
    "for i in range(0,batch_size):\n",
    "    in_a_s = in_a[i,:,:,:]\n",
    "    in_p_s = in_p[i,:,:,:]\n",
    "    in_n_s = in_n[i,:,:,:]\n",
    "    \n",
    "    #in_a, in_p, in_n = next(iter(dataLoader_generator))\n",
    "    in_a_s = in_a_s.to(device)\n",
    "    in_p_s = in_p_s.to(device)\n",
    "    in_n_s = in_n_s.to(device) \n",
    "    #Compute embeddings for anchor, positive, and negative images\n",
    "\n",
    "    emb_a = model(in_a_s.unsqueeze(0))\n",
    "    emb_p = model(in_p_s.unsqueeze(0))\n",
    "    emb_n = model(in_n_s.unsqueeze(0))\n",
    "\n",
    "    loss,p_dist,n_dist = criterion(emb_a,emb_p,emb_n)\n",
    "    p_dist_list.append(p_dist)\n",
    "    n_dist_list.append(n_dist)\n",
    "\n",
    "fig=plt.figure(figsize=(8, 8))\n",
    "\n",
    "columns = 3\n",
    "rows = batch_size\n",
    "in_a_np = in_a.cpu().numpy()\n",
    "in_p_np = in_p.cpu().numpy()\n",
    "in_n_np = in_n.cpu().numpy()\n",
    "j = 1\n",
    "for i in range(1, rows +1):\n",
    "    fig.add_subplot(rows, columns, j)\n",
    "    j+=1\n",
    "    #anchor\n",
    "    img_a = in_a_np[i-1,:,:,:]\n",
    "    plt.imshow(np.transpose(img_a,[1,2,0]))   \n",
    "    plt.axis('off') \n",
    "    fig.add_subplot(rows, columns, j)\n",
    "    j+=1\n",
    "    #positive\n",
    "    img_p = in_p_np[i-1,:,:,:]\n",
    "    plt.imshow(np.transpose(img_p,[1,2,0]))\n",
    "    plt.title(str(p_dist_list[i-1].detach().cpu().numpy()))\n",
    "    plt.axis('off') \n",
    "    fig.add_subplot(rows, columns, j)\n",
    "    j+=1\n",
    "    #positive\n",
    "    img_n = in_n_np[i-1,:,:,:]\n",
    "    plt.imshow(np.transpose(img_n,[1,2,0]))\n",
    "    plt.title(str(n_dist_list[i-1].detach().cpu().numpy()))        \n",
    "    plt.axis('off') \n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# # Try with single images"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def __loadfile(data_file):\n",
    "\n",
    "    image = io.imread(data_file)\n",
    "    if len(image.shape)<3:\n",
    "        image = np.stack((image,)*3, axis=-1)\n",
    "    return image\n",
    "\n",
    "def triplet_generator(transform):\n",
    "    i_a = Image.fromarray(__loadfile(\"/home/rita/JupyterProjects/EYE-SEA/Verification_RNCAPS/Images_examples/anchor.JPG\"))\n",
    "    b_a = transform(i_a)\n",
    "    i_p = Image.fromarray(__loadfile(\"/home/rita/JupyterProjects/EYE-SEA/Verification_RNCAPS/Images_examples/positive.jpg\"))\n",
    "    b_p = transform(i_p)  \n",
    "    i_n = Image.fromarray(__loadfile(\"/home/rita/JupyterProjects/EYE-SEA/Verification_RNCAPS/Images_examples/negative.JPG\"))\n",
    "    b_n = transform(i_n)\n",
    "    return b_a, b_p, b_n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#INPUT: anchor, positive, and negative\n",
    "\n",
    "dataset_transform = transforms.Compose([\n",
    "    transforms.Resize((224,224)),\n",
    "    transforms.ToTensor(),        \n",
    "    transforms.Normalize((0.4914, 0.4822, 0.4465), (0.247, 0.243, 0.261))\n",
    "])\n",
    "in_a, in_p, in_n = triplet_generator(dataset_transform)\n",
    "#in_a, in_p, in_n = Variable(in_a), Variable(in_p),Variable(in_n)\n",
    "in_a = in_a.to(device)\n",
    "in_p = in_p.to(device)\n",
    "in_n = in_n.to(device)     \n",
    "in_a = in_a.unsqueeze(0)\n",
    "in_p = in_p.unsqueeze(0)\n",
    "in_n = in_n.unsqueeze(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ResNetCaps_E.ResNetCaps_E()\n",
    "model = model.to(device)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "summary(model,(3,224,224))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()),lr = 0.001)\n",
    "criterion = TripletLossLayer(0.2)\n",
    "n_epochs = 100\n",
    "\n",
    "p_dist_list = []\n",
    "n_dist_list = []\n",
    "loss_list = []\n",
    "\n",
    "for epoch in range(n_epochs): \n",
    "    model.train()\n",
    "    print('epoch {}:{}'.format(epoch+1, n_epochs)) \n",
    "    emb_a = model(in_a)\n",
    "    emb_p = model(in_p)\n",
    "    emb_n = model(in_n)\n",
    "    \n",
    "    optimizer.zero_grad()\n",
    "    \n",
    "    loss,p_dist,n_dist = criterion(emb_a,emb_p,emb_n)\n",
    "    p_dist_list.append(p_dist)\n",
    "    n_dist_list.append(n_dist)\n",
    "    loss_list.append(loss)\n",
    "    print(\"loss {}\".format(loss))\n",
    "    loss.backward()\n",
    "    optimizer.step()\n",
    "    \n",
    "   "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "emb_a = model(in_a)\n",
    "emb_p = model(in_p)\n",
    "emb_n = model(in_n)\n",
    "\n",
    "print(emb_a)\n",
    "print(emb_p)\n",
    "print(emb_n)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = np.arange(1,n_epochs+1)\n",
    "plt.plot(epochs, p_dist_list, color='g')\n",
    "plt.plot(epochs, n_dist_list, color='orange')\n",
    "plt.xlabel('epochs')\n",
    "plt.ylabel('p_dist,n_dist')\n",
    "plt.title('Training phase')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.plot(epochs, loss_list, color='pink')\n",
    "plt.xlabel('epochs')\n",
    "plt.ylabel('loss')\n",
    "plt.title('Training phase')\n",
    "plt.show() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}