{"cells":[{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load in \n\n# import numpy as np # linear algebra\n# import pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the \"../input/\" directory.\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\n# import os\n# for dirname, _, filenames in os.walk('/kaggle/input/'):\n#     for filename in filenames:\n#         print(os.path.join(dirname, filename))\n\n# Any results you write to the current directory are saved as output.","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"import json\nimport os\nimport numpy as np\nimport pandas as pd\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torchvision.transforms.functional as t_F\nimport torchvision.models as models\nimport torchvision.transforms as transforms\nimport torch.utils.data as data\nimport torchvision\nfrom torch.autograd import Variable\nfrom torch.utils.data import Dataset\nimport cv2\n\nimport matplotlib.pyplot as plt\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.preprocessing import OneHotEncoder, LabelEncoder\nfrom sklearn.metrics import accuracy_score","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# set path\ndata_path = \"/kaggle/input/deepfake-detection-challenge/test_videos\"\nsave_model_path = \"/kaggle/input/single-frame/\"\nmeta_data = \"metadata.json\"\n\nres_size = 224        # ResNet image size\n\n# training parameters\nk = 2             # number of target category\nepochs = 30        # training epochs\nbatch_size = 32\nlearning_rate = 1e-3\nlog_interval = 10   # interval for displaying training info","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class FrameDataset(Dataset):\n    \"\"\"Dataset Class for Loading Video\"\"\"\n\n    def __init__(self, files, labels, num_frames, transform=None, test=False):\n        \"\"\"\n        \"\"\"\n        self.files = files\n        self.labels  = labels\n        self.num_frames = num_frames\n        self.max_num_frames = 60\n        self.transform = transform\n        self.test = test\n        self.frame_no = num_frames\n        self.face_cascade = cv2.CascadeClassifier('/kaggle/input/single-frame/haarcascade_frontalface_default.xml')\n\n    def face_detect(self, frame):\n        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)\n        # Resize frame of video to 1/4 size for faster face detection processing\n        small_frame = cv2.resize(gray, (0, 0), fx=0.25, fy=0.25)\n        # Detect the faces\n        faces = self.face_cascade.detectMultiScale(small_frame, 1.1, 4)\n        return faces\n\n\n    def __len__(self):\n        return len(self.files)\n\n\n    def readVideo(self, videoFile):\n\n        # Load the cascade\n\n        # Open the video file\n        cap = cv2.VideoCapture(videoFile)\n        # cap.set(1, self.frame_no)\n        # nFrames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n        # frames = torch.FloatTensor(self.channels, self.timeDepth, self.xSize, self.ySize)\n\n        attempts = 0\n        while attempts < self.max_num_frames:\n            ret, frame = cap.read()\n            attempts += 1\n            if ret:\n                last_good_frame = frame\n                try:\n                    faces = self.face_detect(frame)\n                    # Face detected\n                    if len(faces) > 0:\n                        # Get the face, if more than two, use the whole frame\n                        if len(faces) > 1:\n                            break\n                        x, y, w, h = faces[0] * 4\n                        face_img = frame[y: y + h, x: x + w]\n                        frame = torch.from_numpy(face_img)\n                        # HWC2CHW\n                        frame = frame.permute(2, 0, 1)\n                        if self.transform is not None:\n                            frame = t_F.to_pil_image(frame)\n                            frame = self.transform(frame)\n                            cap.release()\n                            return frame\n                except:\n                    print(\"Face detection error\")\n            else:\n                break\n\n        frame = torch.from_numpy(last_good_frame)\n        # HWC2CHW\n        frame = frame.permute(2, 0, 1)\n        if self.transform is not None:\n            frame = t_F.to_pil_image(frame)\n            frame = self.transform(frame)\n        cap.release()\n        return frame\n\n    def __getitem__(self, index):\n\n        file = self.files[index]\n        X = self.readVideo(file)\n        if self.test:\n            y = self.labels[index]\n        else:\n            y = torch.LongTensor([self.labels[index]])  # (labels) LongTensor are for int64 instead of FloatTensor\n\n        return X, y","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def test(model, device, test_loader):\n    # set model as testing mode\n    output_file = 'submission.csv'\n    if os.path.exists(output_file):\n        os.remove(output_file)      \n    cnn_encoder= model\n    cnn_encoder.eval()\n\n    results = {}\n    with torch.no_grad():\n        for X, y in test_loader:\n            # distribute data to device\n            X = X.to(device)\n            # y = y.to(device).view(-1, )\n            output = cnn_encoder(X)\n            output_prob = F.softmax(output, dim=1)\n            for i, item in enumerate(output_prob):\n                file_name = y[i].split('/')[-1]\n                #file_name = y[i] \n                prob = output_prob[i][1].item()\n                results[file_name] = prob\n                \n    df =  pd.DataFrame([results.keys(), results.values()]).T\n    df.columns = ['filename', 'label']\n    df.fillna(0.5)\n    df.to_csv(output_file, sep=',', index=False)\n    print(\"Finished prediction!!!\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_X(data_folder, valid=False):\n    X = []\n    y = []\n    videos = os.listdir(data_folder)\n    if valid:\n         with open(os.path.join(data_folder, meta_data)) as json_file:\n            label_data = json.load(json_file)\n    for v in videos:\n        if v.endswith('mp4'):\n            X.append(os.path.join(data_folder, v))\n            if valid:\n                if label_data[v]['label'] == 'FAKE':\n                    y.append(1)\n                else:\n                    y.append(0)\n    return X, y","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Detect devices\nuse_cuda = torch.cuda.is_available()                   # check if GPU exists\ndevice = torch.device(\"cuda\" if use_cuda else \"cpu\")   # use CPU or GPU\n\n# Data loading parameters\nparams = {'batch_size': batch_size, 'shuffle': True, 'pin_memory': True} if use_cuda else {}\n\ntest_X, test_y = get_X(data_path)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"transform = transforms.Compose([transforms.Resize([res_size, res_size]),\n                                transforms.ToTensor(),\n                                transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])])\n\n# selected_frames = np.arange(begin_frame, end_frame, skip_frame).tolist()\nnum_frames = 60\n\ntest_set = FrameDataset(test_X, test_X, num_frames, transform=transform, test=True)\ntest_loader = data.DataLoader(test_set, **params)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Create model\nmodel_ft = models.resnet18()\nnum_ftrs = model_ft.fc.in_features\nmodel_ft.fc = nn.Linear(num_ftrs, 2)\nmodel_ft = model_ft.to(device)\n\n# Load model\nencoder_model_path = os.path.join(save_model_path, 'cnn_encoder_epoch1.pth')\nmodel_ft.load_state_dict(torch.load(encoder_model_path))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Predict\ntest(model_ft, device, test_loader)","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}