{"cells":[{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import os\nimport numpy as np\nfrom skimage.segmentation import slic\nfrom tqdm import tqdm","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","collapsed":true,"_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":false},"cell_type":"code","source":"from time import time\n\nclass Timer:\n def __init__(self, title):\n self.title = title\n def __enter__(self):\n self.start_time = time()\n def __exit__(self, exc_type, exc_val, exc_tb):\n print(\"{} time: {:.3f} s\".format(self.title, time() - self.start_time))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import json\n\nclass VideoLoader:\n \n DFDC_metadataPath = \"../input/deepfake-detection-challenge/train_sample_videos/metadata.json\"\n DFDC_trainVideoDir = \"../input/deepfake-detection-challenge/train_sample_videos/\"\n DFDC_testVideoDir = \"../input/deepfake-detection-challenge/test_videos/\"\n MDFD_testVideoDir = \"../input/mesonet-dataset-sfw/validation/\"\n \n def npSeqFromDir(directory : str, targetSize:tuple=None, normalize=True, frameLimit=np.infty) -> np.ndarray:\n \n # take only the files for which the extension represents an image\n imageList = [img for img in sorted(os.listdir(directory)) if img[-4:].lower() in [\".jpg\",\".jpeg\",\".png\"] ]\n\n if (not targetSize):\n img = Image.open(os.path.join(directory, imageList[0]))\n targetSize = (img.width, img.height)\n\n sequence = np.zeros(( min(len(imageList),frameLimit), targetSize[0], targetSize[1], 3 ), dtype=\"uint8\")\n\n for idx,filename in enumerate(imageList):\n # check the list length\n if (idx >= frameLimit):\n break\n # load the image with PIL\n img = Image.open(os.path.join(directory, filename))\n # resize if necessary\n imgSize = (img.width, img.height)\n if (imgSize != targetSize):\n img = img.resize(targetSize)\n # add the image to the list\n sequence[idx,:,:,:] = np.array(img)\n\n if (normalize):\n # return a numpy 4D array with values in [0,1]\n return sequence / 255.\n else:\n # return a numpy 4D array with values in [0,255]\n return sequence\n \n def loadFilenamesDFDC(videoCount=10, fakeClassValue=1, realClassValue=0):\n # Load metadata file containing labels for videos (\"REAL\" or \"FAKE\")\n with open(VideoLoader.DFDC_metadataPath) as f:\n metadata = json.load(f)\n \n videosInDirectory = [ vid for vid in sorted(os.listdir(VideoLoader.DFDC_trainVideoDir)) if vid[-4:].lower() == \".mp4\" ]\n vidList = []\n for vidname in videosInDirectory[ : min(videoCount, len(videosInDirectory))]:\n sequence_value = fakeClassValue if metadata[vidname][\"label\"]==\"FAKE\" else realClassValue\n vidList.append((vidname, sequence_value))\n \n print(\"Total video names extracted from DFDC:\", len(vidList))\n return vidList\n \n def loadDirnamesMDFD(label=\"df\", videoCount=10, fakeClassValue=1, realClassValue=0):\n # label: \"real\" or \"df\"\n videosInDirectory = sorted(os.listdir(os.path.join(VideoLoader.MDFD_testVideoDir, label)))\n vidList = []\n for vidname in videosInDirectory[ : min(videoCount, len(videosInDirectory))]:\n sequence_value = fakeClassValue if label==\"df\" else realClassValue\n vidList.append((vidname, sequence_value))\n print(\"Total video names extracted from MDFD:\", len(vidList))\n return vidList","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class Segmenter:\n \n def __init__(self, mode=\"color\", segmentsNumber=100, segCompactness=20):\n # mode : \"color\", \"grid\"\n self.mode = mode\n self.segmentsNumber = segmentsNumber\n self.segCompactness = segCompactness\n \n def segment(self, media : np.ndarray) -> np.ndarray:\n \n if (self.mode == \"color\"):\n # if it's an image\n if (len(media.shape)==3):\n return slic(media, n_segments=self.segmentsNumber, compactness=self.segCompactness, sigma=3)\n # if it's a sequence\n elif (len(media.shape)==4):\n # sigma parameters is the size of the gaussian filter that pre-smooth the data\n # I defined the sigma as a triplet where the dimensions represent (time, image_x, image_y)\n return slic(media, n_segments=self.segmentsNumber, compactness=self.segCompactness, sigma=(0.5,3,3))\n else:\n raise Exception(f\"Media shape not recognized: {media.shape}\")\n \n elif (self.mode == \"grid\"):\n pass\n ","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import shap\n\n# if GPU is available, import the numpy accelerated module\nfrom torch import cuda\nGPU = cuda.is_available()\nif (GPU):\n try:\n import cupy as cp\n except ImportError:\n ! pip install cupy-cuda101\n import cupy as cp\n print(\"Using GPU acceleration for Numpy\")\n\n\nfrom contextlib import contextmanager\nimport sys, os\n@contextmanager\ndef suppress_stderr():\n with open(os.devnull, \"w\") as devnull:\n old_stderr = sys.stderr\n sys.stderr = devnull\n try: \n yield\n finally:\n sys.stderr = old_stderr\n\nclass Explainer:\n \n def __init__(self, classifier, trackTime=False):\n # modelName : \"mesonet\" or \"icpr\"\n self.classifier = classifier\n self.model = classifier.getModel()\n self.classifierName = classifier.NAME\n self.trackTime = trackTime\n \n def normalizePredictions(self, p):\n # the predictions are normalized so that FAKE = -1 and REAL = +1\n a = self.classifier.FAKE_CLASS_VAL\n b = self.classifier.REAL_CLASS_VAL\n return 2 * (p - (a+b)/2) / (b-a)\n \n def explain(self, media : np.ndarray, segmentation : np.ndarray, nSegments, shapSamples : int) -> tuple:\n \n # in order to use the GPU and increase performance, numpy arrays are converted at the beginning to\n # cupy arrays\n if (GPU):\n cp_media = cp.asarray(media)\n cp_segmentation = cp.asarray(segmentation)\n else:\n cp_media = media\n cp_segmentation = segmentation\n \n # prediction function to pass to SHAP\n def f(z):\n # z: feature vectors from the point of view of shap\n # from our point of view they are binary vectors defining which segments are active in the image.\n p = self.predictSamples(z, cp_media, cp_segmentation)\n return self.normalizePredictions(p)\n \n # use Kernel SHAP to explain the network's predictions\n background_data = np.zeros((1, nSegments)) # https://shap.readthedocs.io/en/latest/#shap.KernelExplainer\n samples_features = np.ones(nSegments) # A vector of features on which to explain the model’s output.\n explainer = shap.KernelExplainer(f, background_data)\n shap_values = explainer.shap_values(samples_features, nsamples=shapSamples, l1_reg=\"aic\")\n \n return shap_values[0], explainer.expected_value[0]\n\n def predictSamples(self, maskingPatterns, media, segments_slic):\n hideProgressBar = (maskingPatterns.shape[0] <= 1 or self.trackTime)\n\n predictions = []\n\n mask_time = 0\n pred_time = 0\n\n # if it's an image\n if (len(media.shape)==3):\n avg = media.mean((0,1))\n \n # create batches of masking patterns (for performance reasons)\n batchSize = 50\n batches = []\n i = 0\n while (i < maskingPatterns.shape[0]):\n j = i+batchSize if (i+batchSize < maskingPatterns.shape[0]) else maskingPatterns.shape[0]\n batches.append(maskingPatterns[i:j, :])\n i += batchSize\n \n for batch in batches:\n \n # create masked images for this batch\n start_mask_time = time()\n masked_images_batch = []\n for maskingPattern in batch:\n masked_images_batch.append(Explainer.mask_image(maskingPattern, segments_slic, media, avg))\n mask_time += (time() - start_mask_time)\n \n # predict masked images for this batch\n start_pred_time = time()\n if (self.classifierName == \"mesonet\"):\n preds = self.model.predict(np.array(masked_image, ndmin=4))[0]\n elif (self.classifierName == \"icpr\"):\n preds = self.classifier.predictFaceImages(masked_images_batch)\n pred_time += (time() - start_pred_time)\n \n # concatenate this predictions with previous batch predictions\n predictions += list(preds)\n \n if (self.trackTime and maskingPatterns.shape[0]>1):\n print(\"--- Masking: %s seconds ---\" % (mask_time))\n print(\"--- Predicting: %s seconds ---\" % (pred_time))\n\n # if it's a sequence\n elif (len(media.shape)==4):\n avg = media.mean((0,1,2))\n for idx, maskingPattern in tqdm(enumerate(maskingPatterns), disable=hideProgressBar):\n\n start_mask_time = time()\n masked_sequence = Explainer.mask_sequence(maskingPattern, segments_slic, media, avg)\n mask_time += (time() - start_mask_time)\n\n start_pred_time = time()\n if (self.classifierName == \"mesonet\"):\n frames_preds = model.predict(np.array(masked_sequence, ndmin=4))\n elif (self.classifierName == \"icpr\"):\n frames_preds = self.classifier.predictFaceImages(masked_sequence)\n video_pred = np.mean(frames_preds)\n pred_time += (time() - start_pred_time)\n\n predictions.append(video_pred)\n\n if (self.trackTime and maskingPatterns.shape[0]>1):\n print(\"--- Masking: %s seconds ---\" % (mask_time))\n print(\"--- Predicting: %s seconds ---\" % (pred_time))\n \n return np.array(predictions, ndmin=2)\n # Predictions should be a numpy array like\n # [[0.6044281 ] [0.6797433 ] [0.5042769 ] ... ]\n # or\n # [[0.49638838 0.99638945 0.42781693 ... ]]\n \n # a function that depends on a binary mask representing if an image region is hidden\n def mask_image(mask_pattern, segmentation, image, background=None) -> np.ndarray:\n # mask_pattern : an array having length 'nSegments'\n # segmentation : a 3D cupy or numpy array containing the segmentId of every pixel\n # image : a 3D cupy or numpy array containing the image\n if background is None:\n background = image.mean((0,1))\n \n if (GPU):\n out = cp.zeros((image.shape[0], image.shape[1], image.shape[2]))\n else:\n out = np.zeros((image.shape[0], image.shape[1], image.shape[2]))\n out[:,:,:] = image\n\n for j,segm_state in enumerate(mask_pattern):\n if (segm_state == 0):\n out[segmentation==j, :] = background\n if (GPU):\n return cp.asnumpy(out)\n else:\n return out\n\n # a function that depends on a binary mask representing if an image region is hidden\n def mask_sequence(mask_pattern, segmentation, sequence, background=None) -> np.ndarray:\n # mask_pattern : an array having length 'nSegments'\n # segmentation : a 4D cupy or numpy array containing the segmentId of every pixel\n # image : a 4D cupy or numpy array containing the image sequence\n if background is None:\n background = sequence.mean((0,1,2))\n \n if (GPU):\n out = cp.zeros((sequence.shape[0], sequence.shape[1], sequence.shape[2], sequence.shape[3]))\n else:\n out = np.zeros((sequence.shape[0], sequence.shape[1], sequence.shape[2], sequence.shape[3]))\n out[:,:,:,:] = sequence\n\n for j,segm_state in enumerate(mask_pattern):\n if (segm_state == 0):\n out[segmentation==j, :] = background\n if (GPU):\n return cp.asnumpy(out)\n else:\n return out\n \n \n def getExplanationFigure(self, img, imageTrueClass, prediction, shap_values, segments_slic):\n # shap values : a numpy array of length N_SEGMENTS\n \n from matplotlib.colors import LinearSegmentedColormap\n # Fill every segment with the color relative to the shapley value\n def fill_segmentation(values, segmentation):\n out = np.zeros(segmentation.shape)\n for i in range(len(values)):\n out[segmentation == i] = values[i]\n return out\n\n def subplotWithTitle(ax, img, title=\"\", alpha=1):\n ax.imshow(img)\n ax.set_title(title)\n ax.axis('off')\n\n if (img.dtype != \"uint8\"):\n print(\"getExplanationFigure(): 'img' numpy array must be of type 'uint8'\")\n return None\n\n # make a color map\n colors = []\n fc = (.96, .15, .34) # fake color (RGB): red\n rc = (.09, .77, .36) # real color (RGB): green\n for alpha in np.linspace(1, 0, 100):\n colors.append((fc[0], fc[1], fc[2], alpha))\n for alpha in np.linspace(0, 1, 100):\n colors.append((rc[0], rc[1], rc[2], alpha))\n cm = LinearSegmentedColormap.from_list(\"shap\", colors)\n\n fig, axes = plt.subplots(nrows=1, ncols=2, figsize=(6,4))\n \n # set up first image (original image)\n subplotWithTitle(axes[0], img, title = \"class: {}, pred: {:.3f}\".format(\n self.normalizePredictions(imageTrueClass), self.normalizePredictions(prediction)))\n \n # set up second image (gray image)\n grayImage = np.mean(img, axis=2).astype(\"uint8\")\n grayImage = np.dstack((grayImage,grayImage,grayImage))\n subplotWithTitle(axes[1], grayImage, alpha=0.15, title=\"green: real - red: fake\")\n \n # set up segments color overlay\n m = fill_segmentation(shap_values, segments_slic)\n max_val = np.max( [ np.max(np.abs(shap_values[:-1])) for i in range(len(shap_values)) ] )\n \n im = axes[1].imshow(m, cmap=cm, vmin=-max_val, vmax=max_val)\n \n # horizontal bar\n cb = fig.colorbar(im, ax=axes.ravel().tolist(), label=\"SHAP value\", orientation=\"horizontal\", aspect=60)\n cb.outline.set_visible(False)\n\n return fig","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import io\nimport imageio\nfrom PIL import Image\nfrom IPython.display import display, Image as InteractiveImage\nimport matplotlib.pyplot as plt\n\nclass FigureManager:\n \n def fig2pil():\n buf = io.BytesIO()\n plt.savefig(buf, format='png')\n buf.seek(0)\n im = Image.open(buf)\n im.show()\n buf.close()\n return im\n\n def fig2arrayRGB():\n return np.array(FigureManager.fig2pil())[:,:,:3]\n\n def saveAndDisplayGIF(sequence, outputName=\"sequence.gif\", FPS=5, displayOnNotebook=True):\n outputPath = f\"../working/{outputName}\"\n with imageio.get_writer(outputName, mode='I', fps=FPS) as writer:\n for frame in sequence:\n writer.append_data(frame)\n if (displayOnNotebook):\n display(InteractiveImage(outputPath))\n\n def saveAndDisplayImages(images, outputPrefix=\"output\", displayOnNotebook=True):\n for i,image in enumerate(images):\n outputPath = f\"../working/{outputPrefix}_{i}.jpg\"\n with imageio.get_writer(outputPath, mode='i') as writer:\n writer.append_data(image)\n if (displayOnNotebook):\n display(Image.open(outputPath))\n\n def saveAverageSequence(sequence, outputName=\"avg_sequence.png\"):\n avg_array = np.mean(np.array(sequence), axis=0)\n # Round values in array and cast as 8-bit integer\n avg_array = np.array(np.round(avg_array), dtype=np.uint8)\n avg_img = Image.fromarray(avg_array, mode=\"RGB\")\n avg_img.save(outputName)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"import torch\nfrom torch.utils.model_zoo import load_url\nfrom scipy.special import expit\n\n! pip install efficientnet-pytorch\n\nimport sys\nsys.path.append('../input/icpr2020')\nfrom blazeface import FaceExtractor, BlazeFace, VideoReader\nfrom architectures import fornet,weights\nfrom isplutils import utils\n\nclass ICPR:\n \n NAME = \"icpr\"\n INPUT_SIZE = 224\n # output values for the classifier\n REAL_CLASS_VAL = 0\n FAKE_CLASS_VAL = 1\n \n def __init__(self, frames_per_video=10):\n # Choose an architecture between:\n # \"EfficientNetB4\", \"EfficientNetB4ST\", \"EfficientNetAutoAttB4\", \"EfficientNetAutoAttB4ST\", \"Xception\"\n net_model = \"EfficientNetAutoAttB4\"\n\n # Choose a training dataset between:\n # DFDC, FFPP\n train_db = \"DFDC\"\n\n self.device = torch.device('cuda:0') if torch.cuda.is_available() else torch.device('cpu')\n face_policy = 'scale'\n\n model_url = weights.weight_url['{:s}_{:s}'.format(net_model,train_db)]\n self.net = getattr(fornet,net_model)().eval().to(self.device)\n self.net.load_state_dict(load_url(model_url, map_location=self.device, check_hash=True))\n \n self.transf = utils.get_transformer(face_policy, self.INPUT_SIZE, self.net.get_normalizer(), train=False)\n\n facedet = BlazeFace().to(self.device)\n facedet.load_weights(\"../input/icpr2020/blazeface/blazeface.pth\")\n facedet.load_anchors(\"../input/icpr2020/blazeface/anchors.npy\")\n videoreader = VideoReader(verbose=False)\n video_read_fn = lambda x: videoreader.read_frames(x, num_frames=frames_per_video)\n self.face_extractor = FaceExtractor(video_read_fn=video_read_fn, facedet=facedet)\n \n def getModel(self):\n return self.net\n \n def getFaceCroppedImages(self, images):\n faceList = []\n for image in images:\n faceImages = self.face_extractor.process_image(img=image)\n # take the face with the highest confidence score found by BlazeFace\n if (faceImages['faces']):\n faceList.append(faceImages['faces'][0])\n return np.array(faceList)\n\n def getFaceCroppedVideo(self, videoPath):\n faceList = self.face_extractor.process_video(videoPath)\n faceList = [np.array(frame['faces'][0]) for frame in faceList if len(frame['faces'])]\n sequence = np.zeros((len(faceList), self.INPUT_SIZE, self.INPUT_SIZE, 3), dtype=\"uint8\")\n for idx,face in enumerate(faceList):\n # resize the image\n sequence[idx,:,:,:] = np.array(Image.fromarray(face).resize((self.INPUT_SIZE, self.INPUT_SIZE)))\n return sequence\n\n def predictFaceImages(self, images):\n faces_t = torch.stack( [ self.transf(image=img)['image'] for img in images] )\n with torch.no_grad():\n raw_preds = self.net(faces_t.to(self.device))\n faces_pred = torch.sigmoid(raw_preds).cpu().numpy().flatten()\n return faces_pred\n\n def predictImages(self, images):\n faceList = self.getFaceCroppedImages(images)\n faces_pred = self.predictFaceImages(faceList)\n return faces_pred","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class Pipeline:\n def __init__(self, classifier, seg : Segmenter, expl : Explainer, segmentationDim : str, explanationMode : str,\n nSegments : int, shapSamples : int):\n # segmentationDim : a string among [\"2D\", \"3D\"]\n # explanationMode : a string among [\"frame\", \"video\"]\n self.classifier = classifier\n self.seg = seg\n self.expl = expl\n self.segmentationDim = segmentationDim\n self.explanationMode = explanationMode\n self.nSegments = nSegments\n self.shapSamples = shapSamples\n \n def start(self, imageSequence : np.ndarray):\n if (not isinstance(imageSequence, np.ndarray) or len(imageSequence.shape)!=4):\n raise Exception(\"imageSequence must be a 4-dimensional Numpy array.\")\n \n # SEGMENT IMAGE OR VIDEO\n \n with Timer(\"Segmenting\"):\n segmentation = self._segment(imageSequence)\n \n # COMPUTE SHAP VALUES\n \n with Timer(\"Computing SHAP values\"):\n shap_values = self._explain(imageSequence, segmentation)\n \n # COMPUTE CLASSIFIER PREDICTIONS FOR FRAMES OR VIDEO\n \n with Timer(\"Classifier original predictions\"):\n if (self.explanationMode == \"video\"):\n videoPred = self._predictSequence(imageSequence)\n #print(\"Normalized pred.: {:.3f} | Sum of shap values: {:.3f}\".format(\n # expl.normalizePredictions(videoPred), np.sum(shap_values)))\n framesPred = np.ones(imageSequence.shape[0]) * videoPred\n framesShapValues = np.tile(shap_values, (imageSequence.shape[0],1) )\n elif (self.explanationMode == \"frame\"):\n framesPred = np.zeros(imageSequence.shape[0])\n for i in range(imageSequence.shape[0]):\n framesPred[i] = self._predictSequence([imageSequence[i]])\n framesShapValues = shap_values\n \n # SHOW FIGURES\n \n with Timer(\"Showing and saving figures\"):\n fig_sequence = []\n for i in range(imageSequence.shape[0]):\n fig = self.expl.getExplanationFigure(imageSequence[i], vidClass, framesPred[i], framesShapValues[i], segmentation[i])\n fig_sequence.append(FigureManager.fig2arrayRGB())\n plt.close('all')\n\n # Convert the list of frames into a 4D numpy array (frame, width, height, color)\n fig_sequence = np.array(fig_sequence)\n\n FigureManager.saveAndDisplayGIF(fig_sequence, outputName=f\"{vidName}_shap2D.gif\")\n FigureManager.saveAverageSequence(fig_sequence, outputName=f\"{vidName}_shap2D_avg.png\")\n \n def _segment(self, imageSequence):\n if (self.segmentationDim == \"3D\"):\n segmentation = self.seg.segment(imageSequence)\n elif (self.segmentationDim == \"2D\"):\n segmentation = np.zeros((imageSequence.shape[0], imageSequence.shape[1], imageSequence.shape[2]))\n for i,frame in enumerate(imageSequence):\n segmentation[i,:,:] = self.seg.segment(frame)\n return segmentation\n \n def _explain(self, imageSequence, segmentation):\n if (self.explanationMode == \"video\"):\n with suppress_stderr():\n shap_values, expected_value = self.expl.explain(imageSequence, segmentation, self.nSegments, self.shapSamples)\n #print(\"Prediction on all-masked image | expected value: {:.3f}\".format(expected_value))\n \n elif (self.explanationMode == \"frame\"):\n shap_values = np.zeros((imageSequence.shape[0], N_SEGMENTS))\n for i in range(imageSequence.shape[0]):\n print(f\"\\rAnalizing frame: {i+1}/{imageSequence.shape[0]}\",\n end=\"\" if (i+1)