{"cells":[{"metadata":{},"cell_type":"markdown","source":"The notebook is inspired by the paper - [Grad-CAM: Visual Explainations from Deep Networks via Gradient based localizations](https://arxiv.org/pdf/1610.02391.pdf) \n\nThe intution behind the algorithm is that the CNN model has used some pixels in the image to decide upon the class for the model. This can be interpreted as some encoded features that activated the final activation mapping. A visual inspection can also help us in analyzing the wrongly classified samples! \n\nThe influence on a model can be described as the gradient. So the algorithm finds the most dominant logit w.r.t to the latest activation mapping. \n\nIn this notebook, I will implement the grad-CAM algorithm for the ResNext50 model. I have commented most of the code, so it should be easy to read through. \n\nDisclaimer: \n\nThis is my attempt at recreating Grad-CAM for ResNext, the implementation may not be correct. So please bear with me. Also if you can give me some inputs, it would be great!\n\n\n\n\n\nReferences: \n\n- A very straight forward blog by Ian Pointer - [Class Activation Mapping in PyTorch](https://snappishproductions.com/blog/2018/01/03/class-activation-mapping-in-pytorch.html.html)\n- [Implementing Grad-CAM in PyTorch](https://medium.com/@stepanulyanin/implementing-grad-cam-in-pytorch-ea0937c31e82)\n- A big THANKS! to @HumanAnalog for his training & inference notebooks!\n\n"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"# Loading the required libraries\nimport os, sys, time\nimport cv2\nimport random\nimport numpy as np\nimport pandas as pd\nimport skimage.transform\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom torchvision import models, transforms\nfrom torch.autograd import Variable\nfrom torch import topk\n\n%matplotlib inline\nimport matplotlib.pyplot as plt\nfrom PIL import Image\nfrom matplotlib.pyplot import imshow\nfrom tqdm.notebook import tqdm\n","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"# Read the test videos \ntest_dir = \"/kaggle/input/deepfake-detection-challenge/test_videos/\"\n\ntest_videos = sorted([x for x in os.listdir(test_dir) if x[-4:] == \".mp4\"])\nlen(test_videos)\n\n#Load the test labels -- this was created by mapping the metadata file with the test file names\ntest_labels = pd.read_csv('/kaggle/input/sample-face-crop/test_video_labels.csv')\ntest_labels.head()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"The total number of fakes == number of real videos in the *public test set*"},{"metadata":{"trusted":true},"cell_type":"code","source":"test_labels['label'].value_counts()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Confirm the env variables\nprint(\"PyTorch version:\", torch.__version__)\nprint(\"CUDA version:\", torch.version.cuda)\nprint(\"cuDNN version:\", torch.backends.cudnn.version())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Check if GPU is available\ngpu = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\ngpu","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Attach the required libraries to the system path\n# This have a few helper functions & path to the pre-trained model\n\nimport sys\nsys.path.insert(0, \"/kaggle/input/blazeface-pytorch\")\nsys.path.insert(0, \"/kaggle/input/deepfakes-inference-demo\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Initalize blazeface \n\nfrom blazeface import BlazeFace\nfacedet = BlazeFace().to(gpu)\nfacedet.load_weights(\"/kaggle/input/blazeface-pytorch/blazeface.pth\")\nfacedet.load_anchors(\"/kaggle/input/blazeface-pytorch/anchors.npy\")\n_ = facedet.train(False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from helpers.read_video_1 import VideoReader\nfrom helpers.face_extract_1 import FaceExtractor\n\nframes_per_video = 16\n\nvideo_reader = VideoReader()\nvideo_read_fn = lambda x: video_reader.read_frames(x, num_frames=frames_per_video)\nface_extractor = FaceExtractor(video_read_fn, facedet)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"input_size = 224 # Define the input size of the image\n\n# Define the normalizing functions with ImageNet parameters \nfrom torchvision.transforms import Normalize\n\nmean = [0.485, 0.456, 0.406]\nstd = [0.229, 0.224, 0.225]\nnormalize_transform = Normalize(mean, std)\n\n# Define some helper functions for re-sizing image & making them into perfect squares\ndef isotropically_resize_image(img, size, resample=cv2.INTER_AREA):\n    h, w = img.shape[:2]\n    if w > h:\n        h = h * size // w\n        w = size\n    else:\n        w = w * size // h\n        h = size\n\n    resized = cv2.resize(img, (w, h), interpolation=resample)\n    return resized\n\n\ndef make_square_image(img):\n    h, w = img.shape[:2]\n    size = max(h, w)\n    t = 0\n    b = size - h\n    l = 0\n    r = size - w\n    return cv2.copyMakeBorder(img, t, b, l, r, cv2.BORDER_CONSTANT, value=0)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Define the ResNext Model with the given blocks & load the checkpoint\n\nimport torch.nn as nn\nimport torchvision.models as models\n\nclass MyResNeXt(models.resnet.ResNet):\n    def __init__(self, training=True):\n        super(MyResNeXt, self).__init__(block=models.resnet.Bottleneck,\n                                        layers=[3, 4, 6, 3], \n                                        groups=32, \n                                        width_per_group=4)\n        \n        self.fc = nn.Linear(2048, 1)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#Load the checkpoint & update the model for prediction\ncheckpoint = torch.load(\"/kaggle/input/deepfakes-inference-demo/resnext.pth\", map_location=gpu)\n\nmodel = MyResNeXt().to(gpu)\nmodel.load_state_dict(checkpoint)\n_ = model.eval()\n\ndel checkpoint","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"#model.layer4 <<- You can use this to inspect the 4th layer or any other layers","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"sample_video= random.choice(test_videos) # Select a random test video \nvideo_path = os.path.join(test_dir, sample_video)\nprint(\"Selected Video: \", sample_video)\n\ny = test_labels[test_labels['processedVideo'] == sample_video]['label'].values[0]\nprint(\"True Value: \",y)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"batch_size = 16 # Extract faces from 16 frames in the video\nfaces = face_extractor.process_video(video_path)\nprint(\"No. of frames extracted: \", len(faces))\nprint(\"Keys in the extracted info: \", faces[0].keys())\ntry:\n    print(\"Shape of extracted face_crop: \", faces[0]['faces'][0].shape) # multiple faces can be captured. In this set only a single face is detected\n    print(\"Scores of the face crop: \", faces[0]['scores'][0])\nexcept:\n    print(\"=====================================\")\n    print(\"No faces detected! Please run again.\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Only look at one face per frame. This removes multiple faces from each frame, keeping only the best face\nface_extractor.keep_only_best_face(faces)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def preprocessTensor(sample_face):\n    \"\"\"\n    param imageArray: face crop passed to the function\n    return imageTensor(x), resized_face: The processed tensor and resized_face\n    \n    The following activities are performed:\n    1# resizing the image, via zero padding\n    2# make the crop into a square\n    3# Convert to a tensor and load to the gpu\n    4# Permutate the axis so that the channel is at zero index\n    5# Normalize & add a new dimension in zero index\n    \"\"\"\n    resized_face = isotropically_resize_image(sample_face, input_size)\n    resized_face = make_square_image(resized_face)\n    \n    x = torch.tensor(resized_face, device=gpu).float()\n    x = x.permute(( 2, 0, 1))\n    x = normalize_transform(x / 255.)\n    x = x.unsqueeze(0)\n    \n    return x, resized_face","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Resize to the model's required input size.\n# We keep the aspect ratio intact and add zero --???? This could be a problem! \n# padding if necessary.  \n\nsample_face = faces[0]['faces'][0]\nx, resized_face = preprocessTensor(sample_face)\nprint(x.shape)\n\n# Make a prediction.\nwith torch.no_grad():\n    y_pred = model(x)\n    y_pred = torch.sigmoid(y_pred.squeeze())\n\nprint(\"Prediction: \", y_pred)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Resize to the model's required input size.\n# We keep the aspect ratio intact and add zero --???? This could be a problem! \n# padding if necessary.                    \nsample_face = faces[0]['faces'][0]\nresized_face = isotropically_resize_image(sample_face, input_size)\nresized_face = make_square_image(resized_face)\nresized_face.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"x = torch.tensor(resized_face, device=gpu).float()\nprint(x.shape)\n# Preprocess the images.\nx = x.permute(( 2, 0, 1))\nx = normalize_transform(x / 255.)\nx = x.unsqueeze(0)\nprint(x.shape)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Make a prediction.\nwith torch.no_grad():\n    y_pred = model(x)\n    y_pred = torch.sigmoid(y_pred.squeeze())\n\nprint(\"Prediction: \", y_pred)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Since we need the gradients we should predict **w/o** the \n> torch.no_grad()\n\noption"},{"metadata":{},"cell_type":"markdown","source":"NOW for computing the class activations you require the gradients. So the above method is modified by wrapping the prediction image in a Variable. It also provides a backward method to perform backpropagation"},{"metadata":{},"cell_type":"markdown","source":"But first we will use the following (copy+paste from Jeremy Howard's [fast.ai](https://course.fast.ai/)) code sample.\n\nIt allows you to attach a hook to any model.....or for the part ...any layer in the model. \n\nThe hook can then save the activation features as an instance variable. Note that if this hook is not *hooked* the activation features are lost as soon as backward gradient is calculated"},{"metadata":{"trusted":true},"cell_type":"code","source":"class SaveFeatures(): \n    features=None\n    def __init__(self, m): self.hook = m.register_forward_hook(self.hook_fn)  # attach the hook to the specified layer\n    def hook_fn(self, module, input, output): self.features = ((output.cpu()).data).numpy() # copy the activation features as an instance variable\n    def remove(self): self.hook.remove()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"final_layer = model._modules.get('layer4') # Grab the final layer of the model\nactivated_features = SaveFeatures(final_layer) # attach the call back hook to the final layer of the model","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"prediction_var = Variable(x.cuda(), requires_grad=True) # Squeeze the  variable to add an additional dimension & then\nprediction_var.shape                                    # wrap it in a Variable which stores the grad_training weights","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"y_pred = model(prediction_var)\ny_pred = torch.sigmoid(y_pred.squeeze())\n\nprint(\"Prediction: \", y_pred)\npred_probabilities = F.softmax(y_pred).data.squeeze() # Pass the predictions through a softmax layer to convert into probabilities for each class\nprint(\"Predicted Class: \", pred_probabilities)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"In the next function we calculate the class activation mapping. This is where the meat of the notebook lies. *Some changes in the code here* "},{"metadata":{"trusted":true},"cell_type":"code","source":"def getCAM(feature_conv, weight_fc, class_idx):\n    _, nc, h, w = feature_conv.shape\n    cam = weight_fc.dot(feature_conv.reshape((nc, h*w)))\n    cam = cam.reshape(h, w)\n    cam = cam - np.min(cam)\n    cam_img = cam / np.max(cam)\n    return [cam_img]\n\n#overlay = getCAM(activated_features.features, weight_softmax, 0. )","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"weight_softmax_params = list(model._modules.get('fc').parameters()) # This gives a list of weights for the fully connected layers \nprint(len(weight_softmax_params))\nprint(weight_softmax_params[0].shape, weight_softmax_params[1].shape) # weghts for the last two layers\nweight_softmax = weight_softmax_params[0].cpu().data.numpy() # what does this do ??\nprint(weight_softmax.shape)\n#activated_features.remove()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"cam_img = getCAM(activated_features.features, weight_softmax, pred_probabilities )\nprint(cam_img[0].shape)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"So let's first plot the heatmap "},{"metadata":{"trusted":true},"cell_type":"code","source":"imshow(cam_img[0], alpha=0.5, cmap='jet')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"fig, ax = plt.subplots(1,2, figsize=(10,10))\n\nax[0].imshow(resized_face)\nax[0].set_title(\"Video: \" + sample_video + \"Actual: \" + y )\nax[1].imshow(resized_face)\nax[1].imshow(skimage.transform.resize(cam_img[0], (resized_face.shape[0],resized_face.shape[1] )), alpha=0.25, cmap='jet')\ny_pred = str(y_pred.cpu().data.numpy())\nax[1].set_title(y_pred)\nfig.tight_layout()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"I will run the prediction loop on the entire set of test_videos. This will use a single frame to make the prediction. \nThis will identify videos where there is a big difference between the predicted values & the actual values (loss values)\nThis can then be used to analyze the wrongly classified images \n\nRunning this loop should take around 18-20mins\n\nThe face detector fails for some videos and a 0.5 probability is given to it:\n- bzvzpwrabw.mp4\n- coujjnypba.mp4 .....\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"ypred =list()\n\nfor idx, row in tqdm(test_labels.iterrows(), total=len(test_labels)):\n    video_path = os.path.join(test_dir, row['processedVideo']) # Join together the path to the video file (test_video)\n    batch_size = 1 # Extract faces from 01 frames in the video\n    \n    faces = face_extractor.process_video(video_path)  # xtract the face crops from the frames\n    face_extractor.keep_only_best_face(faces) # Is this required ??\n    \n    try:\n        sample_face = faces[0]['faces'][0]\n        \n        x,_ = preprocessTensor(sample_face)\n        \n        with torch.no_grad():\n            y_pred = model(x)\n            y_pred = torch.sigmoid(y_pred.squeeze())\n        \n        ypred.append(y_pred.cpu().data.numpy())\n        \n    except:\n        print(\"Error in processing file: \", row['processedVideo'])\n        ypred.append(0.5)\n        continue\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"test_labels['y_pred'] = ypred\n\ntest_labels['label_10'] = test_labels['label'].apply(lambda x: 1. if x == 'FAKE' else 0.) # Converting the labels to an ordinal value \ntest_labels['diff'] = abs(test_labels['label_10'] - test_labels['y_pred']) # Taking the naive absolute difference of the prediction to the actual ordinal value\ntest_labels.sort_values(by=['diff'], ascending=False, inplace=True) # Sorting the dataframe on difference gives us the samples with the most digression\ntest_labels.reset_index(drop=True, inplace=True)\ntest_labels.head() # LEts check the samples with the highest loss","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"test_labels['diff'].describe()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"test_labels[test_labels['label_10'] == 0].head()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Store the predictions. Since I don't want to keep running the above loop \ntest_labels.to_csv('test_labels.csv')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"idx = 16\nsample_video = test_labels.iloc[idx]['processedVideo']\nvideo_path = os.path.join(test_dir, sample_video)\ny = test_labels.iloc[idx]['label']\nprint(sample_video)\n\nbatch_size = 1 # Extract faces from 16 frames in the video\nfaces = face_extractor.process_video(video_path)\ntry:\n    sample_face = faces[0]['faces'][0]\n    x, resized_face = preprocessTensor(sample_face)\n    \n    final_layer = model._modules.get('layer4')\n    activated_features = SaveFeatures(final_layer)\n    \n    prediction_var = Variable(x.cuda(), requires_grad=True) \n    y_pred = model(prediction_var)\n    y_pred = torch.sigmoid(y_pred.squeeze())\n    \n    pred_probabilities = F.softmax(y_pred).data.squeeze()\n    weight_softmax_params = list(model._modules.get('fc').parameters())\n    \n    weight_softmax = weight_softmax_params[0].cpu().data.numpy() \n    \n    cam_img = getCAM(activated_features.features, weight_softmax, pred_probabilities )\n    \n    fig, ax = plt.subplots(1,2, figsize=(10,10))\n\n    ax[0].imshow(resized_face)\n    ax[0].set_title(\"Video: \" + sample_video + \" Actual: \" + y )\n    ax[1].imshow(resized_face)\n    ax[1].imshow(skimage.transform.resize(cam_img[0], (resized_face.shape[0],resized_face.shape[1] )), alpha=0.25, cmap='jet')\n    y_pred = \"PredProb:\" + str(y_pred.cpu().data.numpy()) + \" DIFF: \" + str(test_labels.iloc[idx]['diff'])\n    ax[1].set_title(y_pred)\n    fig.tight_layout()\nexcept:\n    print(\"Error processing file: \", sample_video)\n    ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"","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}