{"cells":[{"metadata":{},"cell_type":"markdown","source":"This notebook demonstrates the extraction of noise patterns in each video and whether they can be a useful feature\nin deep fake classification."},{"metadata":{"trusted":true},"cell_type":"code","source":"!pip install -q /kaggle/input/pip-mtcnn/mtcnn-0.1.0-py3-none-any.whl","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nimport os\nfrom time import time\nimport cv2\nfrom tqdm.notebook import tqdm\nfrom mtcnn import MTCNN\nimport matplotlib\nimport matplotlib.pyplot as plt\nimport matplotlib.patches as patches\nimport tensorflow as tf\nfrom multiprocess import Pool\nfrom itertools import repeat\nfrom sklearn.model_selection import StratifiedKFold\nfrom sklearn.metrics import log_loss\nimport lightgbm as lgb","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"INPUT_PATH = '/kaggle/input/deepfake-detection-challenge/train_sample_videos/'\nmetadata = pd.read_json(os.path.join(INPUT_PATH, 'metadata.json')).T\nEXTRACT_NOISE = True\nWINDOW = 224\nFACE_CONFIDENCE = .8\nFRAMES_PER_VIDEO = 1\nNOISE_DEPTH = 1","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"The main functions are presented here: For a subset of frames in each video, select a random crop and extract the noise pattern."},{"metadata":{"trusted":true},"cell_type":"code","source":"def face_fft(img):\n    img1 = img - cv2.GaussianBlur(img, (3,3), 0)\n    imgs1 = np.sum(img1, axis=2)\n    sf = np.stack([\n         np.fft.fftshift(np.fft.fft2( imgs1 )),\n         np.fft.fftshift(np.fft.fft2( img1[:,:,0] - img1[:,:,1] )),\n         np.fft.fftshift(np.fft.fft2( img1[:,:,1] - img1[:,:,2] )),\n         np.fft.fftshift(np.fft.fft2( img1[:,:,2] - img1[:,:,0] )) ], axis=-1)\n    return np.abs(sf).astype(np.float16)\n\ndef pattern_ftt(img):\n    if np.isnan(img).any():\n        return np.nan\n    else:\n        img1 = img - cv2.GaussianBlur(img, (3,3), 0)\n        imgs1 = np.sum(img1, axis=2)\n        if NOISE_DEPTH == 1:\n            sf = np.fft.fftshift(np.fft.fft2(imgs1))\n            eps = np.max(sf) * 1e-2\n            s1 = np.log(sf + eps) - np.log(eps) \n            sf = (s1 * 255 / np.max(s1))\n            sf = np.abs(sf)\n        else:\n            sf = np.stack([\n                 np.fft.fftshift(np.fft.fft2( imgs1 )),\n                 np.fft.fftshift(np.fft.fft2( img1[:,:,0] - img1[:,:,1] )),\n                 np.fft.fftshift(np.fft.fft2( img1[:,:,1] - img1[:,:,2] )),\n                 np.fft.fftshift(np.fft.fft2( img1[:,:,2] - img1[:,:,0] ))],\n                 axis=-1)\n            sf = np.abs(sf)\n            nchans = sf.shape[2]\n            for c in range(nchans):\n                eps = np.max(sf[:,:,c]) * 1e-2\n                s1 = np.log(sf[:,:,c] + eps) - np.log(eps) \n                sf[:, :, c] = (s1 * 255 / np.max(s1))\n        return sf.astype(np.float16)\n\ndef check_dims(img, W):\n    d1, d2 = img.shape[:2]\n    if d1 == W and d2 == W:\n        return True\n    else:\n        return False\n    \ndef extract_faces(fn, detector):\n    KPS = 1\n    video_path = os.path.join(INPUT_PATH, fn)\n    try:\n        vidcap = cv2.VideoCapture(video_path)\n        fps = round(vidcap.get(cv2.CAP_PROP_FPS))\n        hop = round(fps / KPS)\n        retval, image = vidcap.read()\n    except:\n        return np.nan\n    if not vidcap.isOpened():\n        return np.nan\n    count = 0\n    i = 0\n    W = WINDOW\n    extracted_image = []\n    while retval:\n        if count % hop == 0:\n            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n            detected_faces = detector.detect_faces(image)\n            for this_face in range(len(detected_faces)):\n                if detected_faces[this_face]['confidence'] > FACE_CONFIDENCE:\n                    bbox = detected_faces[this_face]['box']\n                    crop_center = [int(bbox[0] + .5*bbox[2]),\n                                   int(bbox[1] + .5*bbox[3])]\n                    fixed_xy = [int(crop_center[0] - .5*W),\n                                int(crop_center[1] - .5*W)]\n                    image = image[fixed_xy[1]:fixed_xy[1]+W,\n                                  fixed_xy[0]:fixed_xy[0]+W, :]\n                    image = image / 255.0\n                    if check_dims(image, W) and image.max() > 0:\n                        extracted_image.append(image)\n                        i += 1\n                        if i >= FRAMES_PER_VIDEO:\n                            break\n                        \n        if i >= FRAMES_PER_VIDEO:\n            # Break outer loop\n            break\n        try:\n            retval, image = vidcap.read()\n        except:\n            return np.nan\n        count += 1\n        if count >= fps*60:\n            break\n    extracted_image = np.array(extracted_image)\n    if (extracted_image.size == 0):\n        return np.nan\n    if FRAMES_PER_VIDEO == 1:\n        extracted_image = extracted_image.reshape(W,W,3)\n    return extracted_image\n    \ndef plot_model_features(s):\n    nchans = s.shape[2]\n    nrows = (nchans + 3) // 4\n    _, ax = plt.subplots(nrows, 4, figsize=(16, 4 * nrows))\n    ax = ax.flatten()\n\n    for c in range(nchans):\n        eps = np.max(s[:,:,c]) * 1e-2\n        s1 = np.log(s[:,:,c] + eps) - np.log(eps) \n        img = (s1 * 255 / np.max(s1)).astype(np.uint8)\n        ax[c].imshow(cv2.equalizeHist(img))\n        ax[c].grid(False)\n        ax[c].xaxis.set_visible(False)\n        ax[c].yaxis.set_visible(False)\n    for ax1 in ax[nchans:]:\n        ax1.axis('off')\n\ndef build_df(metadata_df, return_patterns=True):\n    detector = MTCNN()\n    df = pd.DataFrame({'filename': metadata.index.values,\n                        'label': metadata.label.values,\n                      })\n    df['binary_label'] = 0\n    df.loc[df.label == 'FAKE', 'binary_label'] = 1\n    tqdm.pandas()\n    patterns = list(tqdm(map(extract_faces, df.filename.values, repeat(detector)),\n                         total=df.filename.shape[0], desc='Face Extraction'))\n    if return_patterns:\n        with Pool() as pool:\n            patterns = list(tqdm(pool.imap(pattern_ftt, patterns),\n                                 total=len(patterns), desc='Pattern Extraction'))\n    df['pattern'] = patterns\n    return df\n\ndef build_test_df(metadata_df, return_patterns=True):\n    detector = MTCNN()\n    df = pd.DataFrame({'filename': metadata_df.filename.values})\n    tqdm.pandas()\n    patterns = list(tqdm(map(extract_faces, df.filename.values, repeat(detector)),\n                         total=df.filename.shape[0], desc='Face Extraction'))\n    if return_patterns:\n        with Pool() as pool:\n            patterns = list(tqdm(pool.imap(pattern_ftt, patterns),\n                                 total=len(patterns), desc='Pattern Extraction'))\n    df['pattern'] = patterns\n    return df\n\ndef plot_frames(fn, label):\n    KPS = 1\n    video_path = os.path.join(INPUT_PATH, fn)\n    vidcap = cv2.VideoCapture(video_path)\n    fps = round(vidcap.get(cv2.CAP_PROP_FPS))\n    hop = round(fps / KPS)\n    detector = MTCNN()\n    retval, image = vidcap.read()\n    fig, axes = plt.subplots(1, 1, figsize=(8, 5))\n    count = 0\n    i = 0\n    W = WINDOW\n    all_frames = []\n    while retval:\n        if count % hop == 0:\n            image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n            detected_faces = detector.detect_faces(image)\n            axes.imshow(image)\n            if len(detected_faces):\n                for this_face in range(len(detected_faces)):\n                    if detected_faces[this_face]['confidence'] > FACE_CONFIDENCE:\n                        bbox = detected_faces[0]['box']\n                        crop_center = [int(bbox[0] + .5*bbox[2]),\n                                       int(bbox[1] + .5*bbox[3])]\n                        fixed_xy = [int(crop_center[0] - .5*W),\n                                    int(crop_center[1] - .5*W), W, W]\n                        image = image[fixed_xy[1]:fixed_xy[1]+W,\n                                      fixed_xy[0]:fixed_xy[0]+W, :]\n                        image = image / 255.0\n                        if check_dims(image, W) and image.max() > 0:\n                            image = face_fft(image)\n                            all_frames.append(image)\n                            rect = patches.Rectangle((fixed_xy[0],\n                                                      fixed_xy[1]),\n                                                     fixed_xy[2],\n                                                     fixed_xy[3],\n                                                     fill=False, linewidth=3.)\n                            axes.add_patch(rect)\n                            axes.set_title(f'Filename: {fn} - Label: {label}',\n                                             color='black')\n                            axes.xaxis.set_visible(False)\n                            axes.yaxis.set_visible(False)\n                            i += 1\n                            if i >= FRAMES_PER_VIDEO:\n                                break\n                            \n        if i >= FRAMES_PER_VIDEO:\n            break\n        retval, image = vidcap.read()\n        count += 1\n        if count >= fps*60:\n            break\n    all_frames = np.array(all_frames)\n    if (all_frames.size == 0):\n        print(f'No Pattern found')\n    else:\n        all_frames = np.mean(all_frames, axis=0)\n        plot_model_features(all_frames)\n    plt.tight_layout()\n    plt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"A demonstration of this practice is presented below:"},{"metadata":{"trusted":true},"cell_type":"code","source":"FILES = metadata.index\nLABELS = metadata.label\nrnd_file = np.random.randint(0, FILES.shape[0], 1)\nplot_frames(FILES[rnd_file[0]], LABELS[rnd_file[0]])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Train the model to classify each image based on its noise pattern."},{"metadata":{"trusted":true},"cell_type":"code","source":"noise_pattern_df = build_df(metadata, return_patterns=EXTRACT_NOISE)\ns0 = noise_pattern_df.shape[0]\nnoise_pattern_df = noise_pattern_df[~noise_pattern_df.pattern.apply(\n    lambda x: pd.isna(np.ravel(x)).any())].copy()\nprint(f'{s0-noise_pattern_df.shape[0]} videos have been dropped.')\n\nX = np.stack(noise_pattern_df['pattern'].values) / 255.0\nX = X.reshape(X.shape[0], -1)\ny = noise_pattern_df['binary_label'].values\n\npos = np.where(y==1)[0].shape[0]\nneg = np.where(y==0)[0].shape[0]\nprint(f'Number of Fake videos: {pos} - Number of Real videos: {neg}')\nprint('Training model ...')\n\n# Parameters were tuned offline with Bayesian optimization\nbest_params = {\n        'num_leaves': int(12.373597389205074),\n        'max_bin': 63,\n        'min_data_in_leaf': int(15.182532994098365),\n        'learning_rate': 0.04770828591430053,\n        'min_sum_hessian_in_leaf': 0.00266281112712854,\n        'bagging_fraction': 0.28258080526211365,\n        'bagging_freq': int(5.0310417355831465),\n        'feature_fraction': 0.458867976391893,\n        'lambda_l1': 1.4680707418683974,\n        'lambda_l2': 1.4388766929317436,\n        'min_gain_to_split': 0.21162811600005904,\n        'max_depth': int(3.232403494443565),\n        'save_binary': True, \n        'seed': 1337,\n        'feature_fraction_seed': 1337,\n        'bagging_seed': 1337,\n        'drop_seed': 1337,\n        'data_random_seed': 1337,\n        'objective': 'binary',\n        'boosting_type': 'gbdt',\n        'verbose': 1,\n        'metric': 'binary_logloss',\n        'is_unbalance': True,\n        'boost_from_average': False,   \n\n    }\n\nskf = StratifiedKFold(\n    n_splits=5,\n    shuffle=True,\n    random_state=0)\n\npredictions = np.zeros(X.shape[0])\nfor i, (train_index, test_index) in enumerate(skf.split(X, y)):\n        X_train_fold, X_test_fold = X[train_index], X[test_index]\n        y_train_fold, y_test_fold = y[train_index], y[test_index]\n    \n        xg_train = lgb.Dataset(X_train_fold,\n                               label=y_train_fold)\n        xg_valid = lgb.Dataset(X_test_fold,\n                               y_test_fold)\n        num_round = 5000\n        clf = lgb.train(best_params, xg_train,\n                        num_round, valid_sets=[xg_valid],\n                        verbose_eval=0, early_stopping_rounds = 50)\n        predictions[test_index] = clf.predict(X_test_fold,\n                                              num_iteration=clf.best_iteration)\n        score = log_loss(y, predictions)\nprint('Model Training complete.')\nprint(f'CV score: {score:.6f}')","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Plot the train set distributions regarding predictions and labels"},{"metadata":{"trusted":true},"cell_type":"code","source":"pred_df = pd.DataFrame({'predictions': predictions,\n                        'labels': y})\ny_pred = []\ny_pred.append(pred_df[pred_df.labels == 1].predictions.values)\ny_pred.append(pred_df[pred_df.labels == 0].predictions.values)\ny_label = pred_df.groupby('labels').size()\nwith plt.style.context('seaborn'):\n    fig, ax = plt.subplots(1, 2, figsize=(12, 6))\n    ax[0].hist(y_pred, stacked=True, bins=10)\n    ax[0].set_title('Distribution of predictions')\n    _ = y_label.plot(kind='bar', ax=ax[1],\n                     color=['#55a868', '#4c72b0'],\n                     title='Distribution of Labels')\n    plt.tight_layout()\n    plt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"~~Colorblind test~~ Plot the predictions based on their label"},{"metadata":{"trusted":true},"cell_type":"code","source":"with plt.style.context('seaborn'):\n    plt.scatter(range(len(predictions)), predictions, c=y,\n                cmap=matplotlib.colors.ListedColormap(['#55a868', '#4c72b0']))\n    plt.plot([], 'o', c='#55a868', label='Real Videos')\n    plt.plot([], 'o', c='#4c72b0', label='Fake Videos')\n    plt.axhline(.5, linestyle=':', c='tab:red')\n    plt.xlabel('Sample index')\n    plt.ylabel('Prediction %')\n    plt.title('Train set Predictions')\n    plt.tight_layout()\n    plt.legend(loc='lower left')\n    plt.show()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Predict on the test data and submit the results."},{"metadata":{"trusted":true},"cell_type":"code","source":"# Extract noise patterns from validation dataset\nsub_path = '/kaggle/input/deepfake-detection-challenge'\nsub_metadata = pd.read_csv(os.path.join(sub_path, 'sample_submission.csv'))\nINPUT_PATH = os.path.join(sub_path, 'test_videos')\nsubmission_data = build_test_df(sub_metadata,\n                                return_patterns=EXTRACT_NOISE)\nsubmission_subset = submission_data[~submission_data.pattern.apply(\n    lambda x: pd.isna(np.ravel(x)).any())].copy()\nprint(f'{sub_metadata.shape[0]-submission_subset.shape[0]} videos have been dropped.')\n\n# Predict the videos\ntest_data = np.stack(submission_subset['pattern'].values) / 255.0\ntest_data = test_data.reshape(test_data.shape[0], -1)\npredictions = clf.predict(test_data, num_iteration=clf.best_iteration)\n\n# Submit the predictions\nsubmission_subset['label'] = predictions\nsubmission_subset = submission_subset[['filename', 'label']]\nsubmissions = pd.DataFrame({'filename': submission_data.filename.values})\nsubmissions = submissions.merge(submission_subset, how='left', on='filename')\nsubmissions.label.fillna(.5, inplace=True)\nsubmissions.to_csv('submission.csv', index=False)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Plot the distribution of sample predictions"},{"metadata":{"trusted":true},"cell_type":"code","source":"with plt.style.context('seaborn'):\n    plt.scatter(range(len(submissions.label.values)),\n                submissions.label.values,\n                )\n    plt.axhline(.5, linestyle=':', c='tab:red')\n    plt.xlabel('Sample index')\n    plt.ylabel('Prediction %')\n    plt.title('Sample Predictions')\n    plt.tight_layout()\n    plt.legend(loc='lower left')\n    plt.show()","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":1}