{"cells":[{"metadata":{},"cell_type":"markdown","source":"# About\n\nA simple CNN binary classification model for predicting the presence of the Stereochemical Layer.  \n\nStereochemical Layer is represented by */t/m/s* sequential tags in InChI string. If a molecule contains one of the following bonds shown in the image below, Its InChI has to contain the */t* tag.\n\nHere, I trained a binary classifier using PyTorch.\n\nTo train the model, I created a balanced dataset that contains 50% stereo bonds. \n\n\n![](https://natefsi.weebly.com/uploads/5/4/8/7/54874551/944076.png?493)\n Image from: https://natefsi.weebly.com/\n"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom PIL import Image\n\nimport torch\nimport torchvision\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torch.utils.data import DataLoader, Dataset\nimport torchvision.transforms as transforms","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Config\nCFG = {\n    \"batch\": 20,\n    \"epoch\": 30 #Increase for better accuracy\n}","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Define dataset\nclass ImageDataset(Dataset):\n    def __init__(self, filenames, y_vals, size):\n        self.filenames = filenames\n        self.y_vals = y_vals\n        self.size = size\n        self.resize = transforms.Resize((size, size))\n        \n    def __len__(self):\n        return len(self.filenames)\n    \n    def __getitem__(self, idx):\n        \n        filename = self.filenames[idx]\n        image = np.array(Image.open(\"../input/bms-molecular-translation/\"+filename))\n        image = torch.FloatTensor(image)[None,:,:].repeat((3,1,1))/255.        \n        transforms.Normalize(mean=[0.485, 0.456, 0.406],std=[0.229, 0.224, 0.225])(image)\n            \n        X = self.resize(image)         \n        y = self.y_vals[idx]\n\n        return X.data, y","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Read data and split\nfeatured_data = pd.read_csv('../input/featured-train/featured_data.csv')\nfeatured_data_0 =featured_data[featured_data[\"/t\"]==0]\nfeatured_data_1 =featured_data[featured_data[\"/t\"]==1]\nfeatured_data.head()  ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Define data loaders (5000 for train) (2240 for test) (balance the classes)\nfeatured_data_train =  pd.concat([featured_data_0[:2500], featured_data_1[:2500]], axis=0) \nfeatured_data_test = pd.concat([featured_data_0[395000:396120], featured_data_1[395000:396120]], axis=0)   \n    \ntrain_data = ImageDataset(filenames=featured_data_train[\"path\"].tolist(),y_vals=featured_data_train[\"/t\"].tolist(),size=224)      \ntrain_loader = torch.utils.data.DataLoader(train_data, batch_size=CFG[\"batch\"], shuffle=True, num_workers=0)     \n\ntest_data = ImageDataset(filenames=featured_data_test[\"path\"].tolist(),y_vals=featured_data_test[\"/t\"].tolist(),size=224)      \ntest_loader = torch.utils.data.DataLoader(test_data, batch_size=CFG[\"batch\"], shuffle=True, num_workers=0)     ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Define Network\nclass Net1(nn.Module):\n    def __init__(self):\n        super(Net1, self).__init__()\n        # convolutional layer\n        self.conv1 = nn.Conv2d(3, 32, 3, padding=1)\n        self.conv2 = nn.Conv2d(32, 64, 3, padding=1)\n        self.conv3 = nn.Conv2d(64, 128, 3, padding=1)        \n        # max pooling layer\n        self.pool = nn.MaxPool2d(2, 2)      \n        #FC Network\n        self.fc1 = nn.Linear(128 * 28 * 28, 512)\n        self.fc2 = nn.Linear(512, 256)\n        self.fc3 = nn.Linear(256,32)\n        self.fc4 = nn.Linear(32,1)\n        #Dropout\n        self.dropout_fc = nn.Dropout(0.2)\n\n    def forward(self, x):\n        # add sequence of convolutional and max pooling layers\n        x = self.pool(F.leaky_relu(self.conv1(x)))\n        x = self.pool(F.leaky_relu(self.conv2(x)))\n        #x = self.dropout_cnn(x)\n        x = self.pool(F.leaky_relu(self.conv3(x)))\n        #Flatten\n        x = x.view(-1, 128 * 28 * 28)\n        #x = self.dropout_fc(x)\n        x = F.leaky_relu(self.fc1(x))\n        x = self.dropout_fc(x)\n        x = F.leaky_relu(self.fc2(x))\n        x = F.leaky_relu(self.fc3(x))\n        x = self.dropout_fc(x)\n        x = F.sigmoid(self.fc4(x))\n        return x","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Test function\ndef test_loop(dataloader, model, device, loss_fn):\n    \n    total_loss=0\n    total = 0\n    total_correct=0\n    with torch.no_grad():\n        for data in dataloader:\n            # images, labels = data\n            images, labels = data[0].to(device), data[1].to(device)\n            outputs = model(images)\n            total += 1\n            \n            #Get Loss\n            total_loss +=loss_fn(outputs, labels.resize(20,1).to(torch.float32)).item()\n            \n            #Count correct predictions\n            for i, predict in enumerate(outputs):\n                if (predict.item() >= 0.5) and (data[1][i].item()==1):\n                    total_correct += 1       \n                elif (predict.item() < 0.5) and (data[1][i].item()==0):\n                    total_correct += 1\n                # print(predict.item() ,\" - \", data[1][i].item())\n    \n    print('Test loss:', str(total_loss / total)[0:5], \"\\tAccuracy:\", str(total_correct / ((total * CFG[\"batch\"]) / 100))[0:5],)\n    return total_loss / total","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"torch.manual_seed(4)\nnet = Net1()\ndevice = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\nprint(device)\nnet.to(device)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Define loss function and optimizer\nimport torch.optim as optim\ncriterion = nn.BCELoss()\noptimizer = optim.SGD(net.parameters(), lr=0.03, momentum=0.9)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Training loop (for memory problems select new training data in each epoch)   \ntrain_loss=[]\ntest_loss=[]\n\n# for epoch in range(CFG[\"epoch\"]): \nfor epoch in range(CFG[\"epoch\"]):  # loop over the dataset multiple times\n    #Select the datapart and reload the data\n    start=(epoch)*2500\n    end=start+2500\n    featured_data_train =  pd.concat([featured_data_0[start:end], featured_data_1[start:end]], axis=0) \n    train_data = ImageDataset(filenames=featured_data_train[\"path\"].tolist(),y_vals=featured_data_train[\"/t\"].tolist(),size=224)      \n    train_loader = torch.utils.data.DataLoader(train_data, batch_size=CFG[\"batch\"], shuffle=True, num_workers=0)     \n    \n    \n    running_loss = 0.0\n    for i, data in enumerate(train_loader, 0):\n       \n        inputs, labels = data[0].to(device), data[1].to(device)\n        \n        # zero the parameter gradients\n        optimizer.zero_grad()\n\n        # forward + backward + optimize\n        outputs = net(inputs)\n        loss = criterion(outputs, labels.resize(20,1).to(torch.float32))\n        loss.backward()\n        optimizer.step()\n\n        # print statistics\n        running_loss += loss.item()\n        if i % 40 == 39:    # print every 10 mini-batches\n            print('[%d, %5d] loss: %.3f' %(epoch + 1, i + 1, running_loss / 40)) \n            train_loss.append(running_loss / 40)           \n            test_loss.append(test_loop(test_loader,net,device,criterion))\n            \n            running_loss = 0.0","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"net.eval()\ntest_loop(test_loader,net,device,criterion)\nnet.train()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Plot Loss\nx = np.linspace(1, len(test_loss), len(test_loss))\nplt.plot(x, test_loss, '-g', label='test loss')\nplt.plot(x, train_loss, ':b', label='train loss')\n# plt.axis('epoch')\nplt.legend();","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Plot Images as a batch grid \ndef imshow(img):\n    plt.figure(figsize=(20,20))\n    npimg = img.numpy()\n    plt.imshow(np.transpose(npimg, (1, 2, 0)))\n    plt.show()\n\ntorch.set_printoptions(precision=3)    \n#Get example batch predictions for comparison\ndataiter = iter(test_loader)\nimages, labels = dataiter.next()\noutputs = net(images.to(device)) #Iterated on 4 images above\n    \nimshow(torchvision.utils.make_grid(images, nrow=5))\nprint(\"Preds:\")\nprint(outputs.resize(4,5))\nprint(\"True values:\")\nprint(labels.resize(4,5))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Define Test set class (with rotation info)\nclass ImageTest(Dataset):\n    def __init__(self, filenames, rotated, size):\n        self.filenames = filenames\n        self.rotated = rotated\n        self.size = size\n        self.resize = transforms.Resize((size, size))        \n        \n    def __len__(self):\n        return len(self.filenames)\n    \n    def __getitem__(self, idx):\n        \n        filename = self.filenames[idx]\n        image = np.array(Image.open(\"../input/bms-molecular-translation/\"+filename))\n        image = torch.FloatTensor(image)[None,:,:].repeat((3,1,1))/255. \n        if self.rotated[idx]==1:\n            image = torch.rot90(image, k=1, dims=(1,2))\n        transforms.Normalize(mean=[0.485, 0.456, 0.406],std=[0.229, 0.224, 0.225])(image)\n        \n        X = self.resize(image)         \n\n        return X.data","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Test the original test data\nfeatured_test_data = pd.read_csv('../input/featured-train/featured_test_data_rot.csv')\nfeatured_test_sample=featured_test_data[:100]\nexternal_test_data = ImageTest(filenames=featured_test_sample[\"path\"].tolist(),rotated=featured_test_sample[\"rot_flag\"].tolist(),size=224)      \nexternal_test_loader = torch.utils.data.DataLoader(external_test_data, batch_size=CFG[\"batch\"], shuffle=False, num_workers=0) \n\ntorch.set_printoptions(precision=3)\ndataiter = iter(external_test_loader)\nimages = dataiter.next()\noutputs = net(images.to(device)) #Iterated on 4 images above\n    \nimshow(torchvision.utils.make_grid(images, nrow=5))\nprint(\"Preds:\")\nprint(outputs.resize(4,5))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Save model\n#PATH = './t_pred_net.pth'\n#torch.save(net.state_dict(), PATH)","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}