{"cells":[{"metadata":{"trusted":true},"cell_type":"code","source":"import os\nfrom glob import glob\nimport random\nfrom tqdm import tqdm","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"\nclasses = glob('../input/acc-german-traffic-sign-classification/GTSRB_Challenge/train/*')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"samples_in_classes = []\nfor c in classes:\n    samples_in_classes.append(len(glob(os.path.join(c,'*'))))\nprint(samples_in_classes)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"sum(samples_in_classes)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"random.choice(classes)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"img1_names_list = []\nimg2_names_list = []\nsimilarity_label = []\n\nfor c in tqdm(classes):\n    class_label = c.split('/')[-1]\n    print(class_label)\n    for file in glob(os.path.join(c,'*')):\n        img1_names_list.append(file)\n        random_selected_class_label = random.choice(classes)\n        while random_selected_class_label == class_label:\n            random_selected_class_label = random.choice(classes)\n        img2_names_list.append(random.choice(glob(os.path.join(random_selected_class_label,'*'))))\n        similarity_label.append(0)\n        img1_names_list.append(file)\n        img2_names_list.append(random.choice(glob(os.path.join(c,'*'))))\n        similarity_label.append(1)\n\n    ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"len(img1_names_list)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"len(img2_names_list)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"len(similarity_label)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(img1_names_list[:5])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(img2_names_list[:5])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(similarity_label[:5])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"len(classes)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Import all the necessary Library \nimport torchvision\nimport torch.utils.data as utils\nfrom torchvision import datasets\nimport torchvision.transforms as transforms\nfrom torch.utils.data import DataLoader,Dataset\nfrom torch.autograd import Variable\nimport matplotlib.pyplot as plt\nimport torchvision.utils\nimport numpy as np\nimport time\nimport copy\nfrom torch.optim import lr_scheduler\nimport os\nfrom PIL import Image\nimport torch\nfrom torch.autograd import Variable\nimport PIL.ImageOps    \nimport torch.nn as nn\nfrom torch import optim\nimport torch.nn.functional as F\nimport pandas as pd \n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def imshow(img,text=None,should_save=False):\n    npimg = img.numpy()\n    plt.axis(\"off\")\n    if text:\n        plt.text(75, 8, text, style='italic',fontweight='bold',\n            bbox={'facecolor':'white', 'alpha':0.8, 'pad':10})\n    plt.imshow(np.transpose(npimg, (1, 2, 0)))\n    plt.show()    \n\ndef show_plot(iteration,loss):\n    plt.plot(iteration,loss)\n    plt.show()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class SiameseNetworkDataset():\n    \n    def __init__(self,transform=None):\n        # used to prepare the labels and images path\n        #self.training_df=pd.read_csv(training_csv)\n        #self.training_df.columns =[\"image1\",\"image2\",\"label\"]\n        #self.training_dir = training_dir    \n        self.transform = transform\n        self.img1_list = img1_names_list\n        self.img2_list = img2_names_list\n        self.similarity_label = similarity_label\n\n    def __getitem__(self,index):\n        \n        # getting the image path\n        #image1_path=os.path.join(self.training_dir,self.training_df.iat[index,0])\n        #image2_path=os.path.join(self.training_dir,self.training_df.iat[index,1])\n        image1_path = self.img1_list[index]\n        image2_path = self.img2_list[index]\n        \n        # Loading the image\n        img0 = Image.open(image1_path)\n        img1 = Image.open(image2_path)\n        #img0 = img0.convert(\"L\")\n        #img1 = img1.convert(\"L\")\n        \n        # Apply image transformations\n        if self.transform is not None:\n            img0 = self.transform(img0)\n            img1 = self.transform(img1)\n        \n        return img0, img1 , self.similarity_label[index]\n    \n    def __len__(self):\n        return len(self.similarity_label)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Load the the dataset from raw image folders\nsiamese_dataset = SiameseNetworkDataset(\n                                        transform=transforms.Compose([transforms.Resize((105,105)),\n                                                                      transforms.ToTensor()\n                                                                      ])\n                                       )","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Viewing the sample of images and to check whether its loading properly\nvis_dataloader = DataLoader(siamese_dataset,\n                        shuffle=True,\n                        batch_size=8)\ndataiter = iter(vis_dataloader)\n\n\nexample_batch = next(dataiter)\nconcatenated = torch.cat((example_batch[0],example_batch[1]),0)\nimshow(torchvision.utils.make_grid(concatenated))\nprint(example_batch[2].numpy())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class SiameseNetwork(nn.Module):\n    def __init__(self):\n        super(SiameseNetwork, self).__init__()\n        \n        # Setting up the Sequential of CNN Layers\n        self.cnn1 = nn.Sequential(\n            \n            nn.Conv2d(3, 96, kernel_size=11,stride=1),\n            nn.ReLU(inplace=True),\n            nn.LocalResponseNorm(5,alpha=0.0001,beta=0.75,k=2),\n            nn.MaxPool2d(3, stride=2),\n            \n            nn.Conv2d(96, 256, kernel_size=5,stride=1,padding=2),\n            nn.ReLU(inplace=True),\n            nn.LocalResponseNorm(5,alpha=0.0001,beta=0.75,k=2),\n            nn.MaxPool2d(3, stride=2),\n            nn.Dropout2d(p=0.3),\n\n            nn.Conv2d(256,384 , kernel_size=3,stride=1,padding=1),\n            nn.ReLU(inplace=True),\n            nn.Conv2d(384,256 , kernel_size=3,stride=1,padding=1),\n            nn.ReLU(inplace=True),\n            nn.MaxPool2d(3, stride=2),\n            nn.Dropout2d(p=0.3),\n\n        )\n        \n        # Defining the fully connected layers\n        self.fc1 = nn.Sequential(\n            nn.Linear(30976, 1024),\n            nn.ReLU(inplace=True),\n            nn.Dropout2d(p=0.5),\n            \n            nn.Linear(1024, 128),\n            nn.ReLU(inplace=True),\n            \n            nn.Linear(128,2))\n        \n  \n  \n    def forward_once(self, x):\n        # Forward pass \n        output = self.cnn1(x)\n        output = output.view(output.size()[0], -1)\n        output = self.fc1(output)\n        return output\n\n    def forward(self, input1, input2):\n        # forward pass of input 1\n        output1 = self.forward_once(input1)\n        # forward pass of input 2\n        output2 = self.forward_once(input2)\n        return output1, output2\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class ContrastiveLoss(torch.nn.Module):\n    \"\"\"\n    Contrastive loss function.\n    Based on: http://yann.lecun.com/exdb/publis/pdf/hadsell-chopra-lecun-06.pdf\n    \"\"\"\n\n    def __init__(self, margin=2.0):\n        super(ContrastiveLoss, self).__init__()\n        self.margin = margin\n\n    def forward(self, output1, output2, label):\n        euclidean_distance = F.pairwise_distance(output1, output2)\n        loss_contrastive = torch.mean((1-label) * torch.pow(euclidean_distance, 2) +\n                                      (label) * torch.pow(torch.clamp(self.margin - euclidean_distance, min=0.0), 2))\n\n\n        return loss_contrastive","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Load the dataset as pytorch tensors using dataloader\ntrain_dataloader = DataLoader(siamese_dataset,\n                        shuffle=True,\n                        num_workers=8,\n                        batch_size=32)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Check whether you have GPU is loaded or not\nif torch.cuda.is_available():\n    print('Yes')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Declare Siamese Network\nnet = SiameseNetwork().cuda()\n# Decalre Loss Function\ncriterion = ContrastiveLoss()\n# Declare Optimizer\noptimizer = optim.RMSprop(net.parameters(), lr=1e-4, alpha=0.99, eps=1e-8, weight_decay=0.0005, momentum=0.9)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def train(epochs):\n    counter = []\n    loss_history = [] \n    iteration_number= 0\n    \n    for epoch in range(0,epochs):\n        print(f\"Epoch number {epoch}\\n\")\n        for i, data in enumerate(train_dataloader):\n            img0, img1 , label = data\n            img0, img1 , label = img0.cuda(), img1.cuda() , label.cuda()\n            optimizer.zero_grad()\n            output1,output2 = net(img0,img1)\n            loss_contrastive = criterion(output1,output2,label)\n            loss_contrastive.backward()\n            optimizer.step()\n            if i % 100 == 0 :\n                print(\"Current loss {}\\n\".format(loss_contrastive.item()))\n                iteration_number +=10\n                counter.append(iteration_number)\n                loss_history.append(loss_contrastive.item())\n    return net","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n# Train the model\nmodel = train(2)\ntorch.save(model.state_dict(), \"model.pt\")\nprint(\"Model Saved Successfully\")","execution_count":null,"outputs":[]}],"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":4,"nbformat_minor":4}