{"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_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install /kaggle/input/facenetpytorch/facenet_pytorch-2.1.1-py3-none-any.whl","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nimport os\nimport gc\nimport json\nimport cv2\nimport torch\nimport torch.nn as nn\nfrom torchvision import models, transforms\nimport facenet_pytorch\nfrom PIL import Image","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"len(os.listdir('/kaggle/input/deepfake-detection-challenge/test_videos'))","metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"to_predict = os.listdir('/kaggle/input/deepfake-detection-challenge/test_videos')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_video(fn):\n    path = f'/kaggle/input/deepfake-detection-challenge/test_videos/{fn}'\n    vid = cv2.VideoCapture(path)\n    return vid","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_n_every_second(vid,n):\n    frames = []\n    inds = []\n    i = 0\n    fps = vid.get(cv2.CAP_PROP_FPS)\n    n = int(fps/n)\n    while True:\n        ret, frame = vid.read()\n        if ret:\n            if i%n==0:\n                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n                frame = Image.fromarray(frame)\n                frames.append(frame)\n                inds.append(i)\n            i += 1\n        else:\n            break\n    return inds, frames","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"device = 'cuda' if torch.cuda.is_available() else 'cpu'\n\ndetector = facenet_pytorch.MTCNN(margin=14,keep_all=True,device=device)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"device","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"metadata = pd.DataFrame({'filename':to_predict,'boxes':None})","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"max_frames = 22\nfor i,(_,row) in enumerate(metadata.iterrows()):\n    try:\n        fn = row.filename\n        if i%40==0:\n            print(i)\n        vid = get_video(fn)\n        inds, frames = get_n_every_second(vid,2)\n        if not len(frames):\n            print(f'found bad file {fn}')\n            continue\n        boxes = []\n        for u in range(0, len(frames), max_frames):\n            try:\n                bbb, probs = detector.detect(frames[u:u+max_frames])\n                for i in range(len(bbb)):\n                    boxes.append(bbb[i])\n            except Exception as ex:\n                print('error',ex,fn)\n\n        if len(boxes)==0:\n            print(f'no faces found')\n        to_save = {}\n        for ind, box in zip(inds,boxes):\n            to_save[ind] = box.tolist() if box is not None else None\n        metadata.loc[metadata.filename==fn,'boxes'] = json.dumps(to_save)\n    except:\n        pass\n    ","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_n_every_second2(vid,n):\n    frames = []\n    inds = []\n    i = 0\n    fps = vid.get(cv2.CAP_PROP_FPS)\n    n = int(fps/n)\n    while True:\n        ret, frame = vid.read()\n        if ret:\n            if i%n==0:\n                frames.append(frame)\n                inds.append(i)\n            i += 1\n        else:\n            break\n    return inds, frames\n\ndef pad_image(img,height,width):\n    h, w = img.shape[:2]\n    t = 0\n    b = height - h\n    l = 0\n    r = width - w\n    return cv2.copyMakeBorder(img, t, b, l, r, cv2.BORDER_CONSTANT, value=0)\n\ndef resize_and_pad(img, height, width, resample=cv2.INTER_AREA):\n    target_aspect_ratio = height/width\n    im_h, im_w, _ = img.shape\n    im_aspect_aspect_ratio = im_h/im_w\n    if im_aspect_aspect_ratio>target_aspect_ratio:\n        target_height = height\n        target_width = int(im_w * target_height/im_h)\n    else:\n        target_width = width\n        target_height = int(im_h * target_width/im_w)\n    resized = cv2.resize(img, (target_width, target_height), interpolation=resample)\n    return pad_image(resized,height,width)\n\ndef get_image_chunks_face(fn, boxes, height, width, transform, num_chunks):\n    chunks = []\n    boxes = json.loads(boxes)\n    vid = get_video(fn)\n    for ind, frame in zip(*get_n_every_second2(vid,2)):\n        if not boxes[str(ind)]:\n            continue\n        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n        frame_height, frame_width, _ = frame.shape\n        for ind_2, box in enumerate(boxes[str(ind)]):\n            left, top, right, bottom = box\n            left -= 20\n            right += 20\n            bottom += 20\n            top -= 20\n            left = int(max(left,0))\n            right = int(min(right,frame_width))\n            top = int(max(top,0))\n            bottom = int(min(bottom,frame_height))\n            if right-left<90 or bottom-top<110: ## min size of a face w/ padding\n                continue\n            crop = resize_and_pad(frame[top:bottom,left:right],height,width)\n            chunks.append(transform(crop))\n    \n    print(f'num chunks found: {len(chunks)}')\n    if not chunks:\n        print('no faces found, adding noise images')\n    if len(chunks)<num_chunks:\n        chunks.extend([chunks[-1] for _ in range(num_chunks-len(chunks))])\n    elif len(chunks)>num_chunks:\n        chunks = chunks[:num_chunks]\n    return torch.stack(chunks)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"normalize = transforms.Normalize(mean=[0.485, 0.456, 0.406],std=[0.229, 0.224, 0.225])\ntransform = transforms.Compose([transforms.ToPILImage(),transforms.Resize((300,225)),\n                                transforms.ToTensor(),normalize])\n\nfrom torchvision.models.resnet import resnet50\n\nclass ImageDetector(nn.Module):\n    def __init__(self,dropout_p=0):\n        super().__init__()\n        self.model = resnet50(pretrained=False)\n        self.model.fc = nn.Sequential(nn.Dropout(p=dropout_p),nn.Linear(2048,1))\n\n\n    def forward(self,x):\n        return self.model(x)\n\nclass Learner(nn.Module):\n    def __init__(self,dropout_p=0):\n\n        super().__init__()\n\n        self.image_detector = ImageDetector(dropout_p)\n        self.bias = nn.Parameter(torch.tensor(0.,requires_grad=True))\n\n    def forward(self,x):\n        ## x is: Batch x Chunk x Channel x Height x Width\n\n        ## transform to (Batch * Chunk) x Channel x Height x Width\n        x_size = x.size()\n        size = (x_size[0]*x_size[1],x_size[2],x_size[3],x_size[4])\n\n        ## outputs should come back (Batch * Chunk) x 1\n        out = self.image_detector(x.view(size))\n\n        ## transform outputs to be size Batch using max operator\n        out = out.view(x_size[0],x_size[1]).max(1).values\n        \n        return out\n\ndevice = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\nmodel = Learner(.5)\nmodel.to(device)\n\nstate_dict = torch.load('/kaggle/input/milearningimage1/33.pt',map_location=device)\nmodel.load_state_dict(state_dict)\n\nmodel.eval();","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def predict_one(row, model, device, clamp_level=1.6):\n    with torch.set_grad_enabled(False):\n        chunks = get_image_chunks_face(row.filename,row.boxes,400,300,transform,24)\n        chunks = chunks.view(1,*chunks.size()).to(device)\n        res = model(chunks)\n        res = torch.clamp(res,-clamp_level,clamp_level)\n        score = res.cpu().item()\n        return 1/(1+np.exp(-1*score))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"fn_to_prob = {}\nfor i, (_,row) in enumerate(metadata.iterrows()):\n    if i%40==0:\n        print(i)\n    if not row.boxes:\n        print(f'no boxes found: {row.filename}')\n        fn_to_prob[row.filename] = .5\n        continue\n    try:\n        fn_to_prob[row.filename] = predict_one(row, model, device)\n    except Exception as ex:\n        print('error',ex,row.filename)\n        fn_to_prob[row.filename] = .5","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"res = pd.DataFrame({'filename':to_predict,\n                    'label':[fn_to_prob[fn] for fn in to_predict]})\n\nprint(res.sample(n=30).head(30))\nprint(res.shape)\n\nres.to_csv('submission.csv',index=False)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"res.label.describe()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}