{"cells":[{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"import os\nimport sys\nimport numpy as np\nimport pandas as pd\nimport cv2\nfrom keras import layers, Model","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"!pip install ../input/efnet6weights/efficientnet-1.1.0-py3-none-any.whl\nimport efficientnet.keras as efn","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Taken from here: https://github.com/1adrianb/face-alignment\n# Thanks to Adrian Bulat (https://github.com/1adrianb)\n# Licensed under BSD 3-Clause, which allows modification, distribution, commercial and private use\n# Imported as package because pip installation requires internet to download weights for detector\n# I've modified the code to remove all stuff related to landmarks extraction, so I've used only s3fd detection module\nsys.path.insert(0, \"../input\")\nimport s3fdfacedetector\n\nfd = s3fdfacedetector.S3FDFaceDetector()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Big thanks to Human Analog\n# Taken from here: https://www.kaggle.com/humananalog/deepfakes-inference-demo\nclass VideoReader:\n    \"\"\"Helper class for reading one or more frames from a video file.\"\"\"\n\n    def __init__(self, verbose=True):\n        \"\"\"Creates a new VideoReader.\n\n        Arguments:\n            verbose: whether to print warnings and error messages\n            insets: amount to inset the image by, as a percentage of \n                (width, height). This lets you \"zoom in\" to an image \n                to remove unimportant content around the borders. \n                Useful for face detection, which may not work if the \n                faces are too small.\n        \"\"\"\n        self.verbose = verbose\n\n    def read_frames_at_indices(self, path, frame_idxs):\n        \"\"\"Reads frames from a video and puts them into a NumPy array.\n\n        Arguments:\n            path: the video file\n            frame_idxs: a list of frame indices. Important: should be\n                sorted from low-to-high! If an index appears multiple\n                times, the frame is still read only once.\n\n        Returns:\n            - a NumPy array of shape (num_frames, height, width, 3)\n            - a list of the frame indices that were read\n\n        Reading stops if loading a frame fails, in which case the first\n        dimension returned may actually be less than num_frames.\n\n        Returns None if an exception is thrown for any reason, or if no\n        frames were read.\n        \"\"\"\n        assert len(frame_idxs) > 0\n        capture = cv2.VideoCapture(path)\n        result = self._read_frames_at_indices(path, capture, frame_idxs)\n        capture.release()\n        return result\n\n    def _read_frames_at_indices(self, path, capture, frame_idxs):\n        try:\n            frames = []\n            idxs_read = []\n            for frame_idx in range(frame_idxs[0], frame_idxs[-1] + 1):\n                ret = capture.grab()\n                if not ret:\n                    if self.verbose:\n                        print(\"Error grabbing frame %d from movie %s\" % (frame_idx, path))\n                    break\n\n                current = len(idxs_read)\n                if frame_idx == frame_idxs[current]:\n                    ret, frame = capture.retrieve()\n                    if not ret or frame is None:\n                        if self.verbose:\n                            print(\"Error retrieving frame %d from movie %s\" % (frame_idx, path))\n                        break\n\n                    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n                    frames.append(frame)\n                    idxs_read.append(frame_idx)\n\n            if len(frames) > 0:\n                return np.stack(frames), idxs_read\n            if self.verbose:\n                print(\"No frames read from movie %s\" % path)\n            return None\n        except:\n            if self.verbose:\n                print(\"Exception while reading movie %s\" % path)\n            return None\n        \nvideo_reader = VideoReader()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"test_dir = \"/kaggle/input/deepfake-detection-challenge/test_videos/\"\ntest_videos = sorted([x for x in os.listdir(test_dir)])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"weights = [\"efnet_6_2020-03-26_14-08-45_epoch_06\",\n           \"efnet_6_2020-03-27_21-28-54_epoch_06\",\n           \"efnet_6_2020-03-29_21-28-21_epoch_06\",\n           \"efnet_6_2020-03-29_21-29-50_epoch_06\",\n           \"efnet_6_2020-03-30_00-26-00_epoch_06\"\n          ]","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def efnet(backbone='6', input_shape=(224,224,3), num_classes=1):\n    \n    DENSE_KERNEL_INITIALIZER = {\n        'class_name': 'VarianceScaling',\n        'config': {\n            'scale': 1. / 3.,\n            'mode': 'fan_out',\n            'distribution': 'uniform'\n        }\n    }\n    \n    if backbone == '0':\n        EFNet = efn.EfficientNetB0\n    elif backbone == '1':\n        EFNet = efn.EfficientNetB1\n    elif backbone == '2':\n        EFNet = efn.EfficientNetB2\n    elif backbone == '3':\n        EFNet = efn.EfficientNetB3\n    elif backbone == '4':\n        EFNet = efn.EfficientNetB4\n    elif backbone == '5':\n        EFNet = efn.EfficientNetB5\n    elif backbone == '6':\n        EFNet = efn.EfficientNetB6\n    elif backbone == '7':\n        EFNet = efn.EfficientNetB7\n\n    base_model = EFNet(input_shape=input_shape, weights=None, include_top=False, classes=num_classes)\n    x = layers.GlobalAveragePooling2D()(base_model.output)\n    x = layers.Dropout(0.2)(x)\n    output = layers.Dense(num_classes, activation='sigmoid', kernel_initializer=DENSE_KERNEL_INITIALIZER)(x)\n        \n    return Model(inputs=[base_model.input], outputs=[output])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"models = []\n\nfor i in range(len(weights)):\n    models.append(efnet())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"for model, weight in zip(models, weights):\n    model.load_weights('/kaggle/input/efnet6weights/' + weight + '.hdf5')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"faces_batch_size = 55\n\nframes_per_video = 55\n\nface_size = 224\n\nclip = 1e-15\n\nws = [1/(len(models))]*len(models)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"out_probs = []\n\nfor video in test_videos:\n    \n    fake_probs = []\n    faces_batch = []\n    \n    try:\n        capture = cv2.VideoCapture(f'{test_dir+video}')\n        frame_count = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))\n        frame_idxs = np.linspace(0, frame_count - 1, frames_per_video, endpoint=True, dtype=np.int)\n        result = video_reader.read_frames_at_indices(f'{test_dir+video}', frame_idxs=frame_idxs)\n        \n        size_multiplier = 2 if min(result[0][0][:,:,0].shape) < 1080 else 4\n        resize_size = tuple(int(ti/size_multiplier) for ti in result[0][0][:,:,0].shape[::-1])\n\n        smalls = min(result[0][0][:,:,0].shape)//20\n        multiplier = 1080/min(result[0][0][:,:,0].shape)\n\n        for original_image in result[0]:\n\n            resized_image = cv2.resize(original_image, resize_size)\n            predictions = fd.get_detections(resized_image)\n\n            try:\n                prediction = predictions[0]\n            except:\n                print(\"Unable to find face\")\n            else:\n                for prediction in predictions:\n                        \n                    left, right, top, bottom = size_multiplier*int(prediction[0]), size_multiplier*int(prediction[2]), size_multiplier*int(prediction[1]), size_multiplier*int(prediction[3])\n                    \n                    if left < 0 or right < 0 or top < 0 or bottom < 0 or left > original_image.shape[1] or right > original_image.shape[1] or top > original_image.shape[0] or bottom > original_image.shape[0]:\n                        continue\n\n                    top += int((bottom - top)/4)\n                    left += int((right - left)/12)\n                    right -= int((right - left)/12)\n                    \n                    if (right-left < smalls) or (bottom - top < smalls):\n                        continue\n                        \n                    face_max_side = max(bottom - top, right-left)\n\n                    if face_max_side <= 64/multiplier:\n                        face_offset = 8/multiplier\n                    elif face_max_side <= 128/multiplier:\n                        face_offset = 12/multiplier\n                    elif face_max_side <= 192/multiplier:\n                        face_offset = 16/multiplier\n                    elif face_max_side <= 256/multiplier:\n                        face_offset = 20/multiplier\n                    else:\n                        face_offset = 24/multiplier\n\n                    face_offset = int(face_offset)\n\n                    face_image = cv2.resize(original_image[max(top-face_offset, 0):min(bottom+face_offset, resize_size[1]*size_multiplier), max(left-face_offset, 0):min(right+face_offset, resize_size[0]*size_multiplier)], (face_size, face_size))\n                    faces_batch.append(face_image)\n                    \n        while len(faces_batch) > 0:\n\n            batch_size = min(len(faces_batch), faces_batch_size)\n            x = np.zeros((batch_size, face_size, face_size, 3))\n\n            for b in range(batch_size):\n                x[b] = faces_batch[b]\n\n            del(faces_batch[:batch_size])\n\n            x = efn.preprocess_input(x)\n            \n            ys = []\n            for model in models:\n                ys.append(model.predict(x))\n\n            for (y0, y1, y2, y3, y4) in zip(ys[0], ys[1], ys[2], ys[3], ys[4]):\n                fake_probs.append(ws[0]*y0[-1] + ws[1]*y1[-1] + ws[2]*y2[-1] + ws[3]*y3[-1] + ws[4]*y4[-1])\n\n    except Exception as e:\n        print(\"Prediction error on video %s: %s\" % (video, str(e)))\n        \n    if len(fake_probs) == 0:\n        out_prob = 0.5\n    else:\n        out_prob = np.median(fake_probs).clip(clip, 1-clip)\n        \n    out_probs.append(out_prob)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"submission_df = pd.DataFrame({\"filename\": test_videos, \"label\": out_probs})\nsubmission_df.to_csv(\"submission.csv\", index=False)","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}