{"cells":[{"metadata":{},"cell_type":"markdown","source":"## Imports and function definitions","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"!pip install efficientnet-pytorch\n\nfrom tensorflow.keras.preprocessing import image\nimport requests\nfrom skimage.segmentation import slic\nimport numpy as np\nimport shap\nimport os\nfrom PIL import Image\nimport imageio\nfrom tqdm import tqdm\nimport time\n\nimport torch\nfrom torch.utils.model_zoo import load_url\nimport matplotlib.pyplot as plt\nfrom scipy.special import expit\n\nimport sys\nsys.path.append('../input/icpr2020')\n\nfrom blazeface import FaceExtractor, BlazeFace, VideoReader\nfrom architectures import fornet,weights\nfrom isplutils import utils","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Utility functions","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# numpy image sequence from directory of images\ndef npSeqFromDir(directory, targetSize=None, normalize=True, frameLimit=np.infty):\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\nimport io\ndef 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\ndef fig2arrayRGB():\n    return np.array(fig2pil())[:,:,:3]\n\nfrom IPython.display import Image as IImg, display\ndef saveAndDisplayGIF(sequence, outputName=\"sequence.gif\", FPS=5, displayOnNotebook=True):\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(IImg(\"../working/\"+outputName))\n\ndef 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(IImg(outputPath))\n    \ndef saveAverageSequence(sequence, outputName=\"avg_sequence.png\", displayOnNotebook=False):\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)\n    if (displayOnNotebook):\n        display(avg_img)\n    \n# to suppress SHAP's warnings on numerical precision\nfrom contextlib import contextmanager\nimport sys, os\n\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            \n@contextmanager\ndef suppress_stdout():\n    with open(os.devnull, \"w\") as devnull:\n        old_stdout = sys.stdout\n        sys.stdout = devnull\n        try:  \n            yield\n        finally:\n            sys.stdout = old_stdout","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Plotting functions","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"from matplotlib.colors import LinearSegmentedColormap\n# Fill every segment with the color relative to the shapley value\ndef 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\ndef getExplanationFigure(img, imageTrueClass, prediction, shap_values, segments_slic, fakeClassValue):\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    for l in np.linspace(1, 0, 100):\n        colors.append((245/255, 39/255, 87/255, l))\n    for l in np.linspace(0,1,100):\n        colors.append((24/255, 196/255, 93/255, l))\n    cm = LinearSegmentedColormap.from_list(\"shap\", colors)\n    \n    # reverse shap values so that red is fake\n    if (fakeClassValue==0):\n        adjusted_shap_values = shap_values\n    else:\n        adjusted_shap_values = np.negative(shap_values)\n\n    # set first image (original image)\n    fig, axes = plt.subplots(nrows=1, ncols=2, figsize=(6,4))\n    axes[0].imshow(img)\n    axes[0].set_title(\"class: {}, pred: {:.3f}\".format(imageTrueClass, prediction))\n    axes[0].axis('off')\n    \n    #max_val = np.max([np.max(np.abs(adjusted_shap_values[0][:-1])) for i in range(len(shap_values))])\n    max_val = 0.3\n    # set second image (segmented and colored image)\n    m = fill_segmentation(adjusted_shap_values[0], segments_slic)\n    axes[1].set_title(\"green: real - red: fake\")\n    grayImage = np.mean(img, axis=2).astype(\"uint8\")\n    grayImage = np.dstack((grayImage,grayImage,grayImage))\n    axes[1].imshow(grayImage, alpha=0.15)\n    im = axes[1].imshow(m, cmap=cm, vmin=-max_val, vmax=max_val)\n    axes[1].axis('off')\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":{},"cell_type":"markdown","source":"## ICRP model (definitions and config)","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"def icrpGetFaceCroppedImages(images):\n    faceList = []\n    for image in images:\n        faceImages = 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\ndef icrpGetFaceCroppedVideo(video):\n    INPUT_SIZE = 224\n    faceList = face_extractor.process_video(video)\n    faceList = [np.array(frame['faces'][0]) for frame in faceList if len(frame['faces'])]\n    sequence = np.zeros((len(faceList), INPUT_SIZE, 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((INPUT_SIZE,INPUT_SIZE)))\n    return sequence\n\ndef icprPredictFaceImages(net, images):\n    faces_t = torch.stack( [ transf(image=img)['image'] for img in images] )\n    with torch.no_grad():\n        faces_pred = torch.sigmoid(net(faces_t.to(device))).cpu().numpy().flatten()\n    return faces_pred\n\ndef icprPredictImages(net, images):\n    faceList = icrpGetFaceCroppedImages(images)\n    faces_pred = icprPredictFaceImages(net, faceList)\n    return faces_pred","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"\"\"\"\nChoose an architecture between\n- EfficientNetB4\n- EfficientNetB4ST\n- EfficientNetAutoAttB4\n- EfficientNetAutoAttB4ST\n- Xception\n\"\"\"\nnet_model = 'EfficientNetAutoAttB4'\n\n\"\"\"\nChoose a training dataset between\n- DFDC\n- FFPP\n\"\"\"\ntrain_db = 'DFDC'\n\ndevice = torch.device('cuda:0') if torch.cuda.is_available() else torch.device('cpu')\n\nmodel_url = weights.weight_url['{:s}_{:s}'.format(net_model,train_db)]\nnet = getattr(fornet,net_model)().eval().to(device)\nnet.load_state_dict(load_url(model_url,map_location=device,check_hash=True))\n\n# output values for the classifier\nREAL_CLASS_VAL = 0\nFAKE_CLASS_VAL = 1","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"face_policy = 'scale'\nFACE_SIZE = 224\nFRAMES_PER_VIDEO = 15\n\ntransf = utils.get_transformer(face_policy, FACE_SIZE, net.get_normalizer(), train=False)\n\nfacedet = BlazeFace().to(device)\nfacedet.load_weights(\"../input/icpr2020/blazeface/blazeface.pth\")\nfacedet.load_anchors(\"../input/icpr2020/blazeface/anchors.npy\")\nvideoreader = VideoReader(verbose=False)\nvideo_read_fn = lambda x: videoreader.read_frames(x, num_frames=FRAMES_PER_VIDEO)\nface_extractor = FaceExtractor(video_read_fn=video_read_fn,facedet=facedet)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## SHAP (definitions and config)","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# a function that depends on a binary mask representing if an image region is hidden\ndef mask_sequence(mask_pattern, segmentation, sequence, background=None):\n    # mask_pattern: an array having length 'nSegments'\n    if background is None:\n        background = sequence.mean((0,1,2))\n    \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    return out","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"def f_icpr(maskingPatterns, media : np.array, segments_slic) -> np.array:\n\n    hideProgressBar = (maskingPatterns.shape[0] <= 1)\n\n    predictions = []\n\n    # if it's an image\n    if (len(media.shape)==3):\n        avg = media.mean((0,1))\n        for maskingPattern in maskingPatterns:\n            masked_image = mask_image(maskingPattern, segments_slic, media, avg)\n            preds = icprPredictFaceImages(net, [media])\n            predictions.append(preds)\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            masked_sequence = mask_sequence(maskingPattern, segments_slic, media, avg)\n            frames_preds = icprPredictFaceImages(net, masked_sequence)\n            video_pred = np.mean(np.array(frames_preds) > 0.5)\n            predictions.append(np.array(video_pred, ndmin=1))\n\n    return np.array(predictions, ndmin=2)\n\n\ndef explain(model, media, runTimes, nSegments=50, segCompactness=10) -> tuple:\n\n    # segment the image so we don't have to explain every pixel\n\n    # if it's an image\n    if (len(media.shape)==3):\n        spatial_smooth = 1       # variance of gaussian smooth across adiacent pixels\n        segments_slic = slic(media, n_segments=nSegments, compactness=segCompactness, sigma=1)\n\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        temporal_smooth = 0.5    # variance of gaussian smooth across conseutive frame\n        spatial_smooth = 1       # variance of gaussian smooth across adiacent pixels\n        s = (temporal_smooth, spatial_smooth, spatial_smooth)\n        segments_slic = slic(media, n_segments=nSegments, compactness=segCompactness, sigma=s)\n    \n    else:\n        print(\"Media shape not recognized:\", media.shape)\n        return\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        return f_icpr(z, media, segments_slic)\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=runTimes, l1_reg=\"aic\")\n    \n    return shap_values, segments_slic","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Starting point","execution_count":null},{"metadata":{"trusted":true},"cell_type":"code","source":"# Configuration\nSHAP_SAMPLES = 500\nN_SEGMENTS = 200\nVIDEOS_TO_ANALYZE = 20","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# load videos\n\nimport json\n\ndirectory = \"../input/deepfake-detection-challenge/train_sample_videos/\"\nwith open(os.path.join(directory, \"metadata.json\")) as f:\n    metadata = json.load(f)\n\nvidList = []\nfor vidname in sorted(os.listdir(directory))[:VIDEOS_TO_ANALYZE]:\n    sequence_value = FAKE_CLASS_VAL if metadata[vidname][\"label\"]==\"FAKE\" else REAL_CLASS_VAL\n    vidList.append((os.path.join(directory, vidname), sequence_value))\n    \nprint(\"Total videos:\", len(vidList))","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"for i, (vid, class_value) in enumerate(vidList):\n    \n    print(f\"--- Analizing video: {i+1}/{len(vidList)} ---\")\n    \n    imgClassDesc = str(class_value) + (' (fake)' if class_value==FAKE_CLASS_VAL else ' (real)')\n    \n    start_time = time.time()\n    faceSequence = icrpGetFaceCroppedVideo(vid)\n    print(\"Face cropping: %s seconds\" % (time.time() - start_time))\n    \n    start_time = time.time()\n    with suppress_stderr():\n        shap_values, segments_slic = explain(net, faceSequence, SHAP_SAMPLES, nSegments=N_SEGMENTS)\n    print(\"Shap time for %d samples: %s seconds\" % (SHAP_SAMPLES, time.time() - start_time))\n    \n    start_time = time.time()\n    preds = icprPredictFaceImages(net, faceSequence)\n    fig_sequence = []\n    for idx,img in enumerate(faceSequence):\n        fig = getExplanationFigure(img, imgClassDesc, np.mean(preds), shap_values, segments_slic[idx], FAKE_CLASS_VAL)\n        figarray = fig2arrayRGB()\n        fig_sequence.append(figarray)\n        plt.close('all')\n    # Convert the list of frames into a 4D numpy array (frame, width, height, color)\n    fig_sequence = np.asarray(fig_sequence)\n    print(\"Create animation for %d frames: %s seconds\" % (fig_sequence.shape[0], time.time() - start_time))\n    \n    start_time = time.time()\n    saveAndDisplayGIF(fig_sequence, outputName=f\"dfdc_{i}_shap2D.gif\")\n    print(\"Save and show animation: %s seconds\" % (time.time() - start_time))\n    \n    start_time = time.time()\n    saveAverageSequence(fig_sequence, outputName=f\"dfdc_{i}_shap2D_avg.png\")\n    print(\"Show average image: %s seconds\" % (time.time() - start_time))\nprint(\" \")","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":1}