{"cells":[{"metadata":{},"cell_type":"markdown","source":"# Triplet Training\nTraining is done in 2 phases / models\n- Phase 1 trains a CNN to output embeddings that have high pairwise distance between fake and real images in comparison to distance between fake and fake.\n\n- Phase 2 adds a linear layer to above CNN to do binary classification.\n\nNote: Triplet training did seem to be better than Siamese training, however the final classifier was nowhere nearly as effective as simple binary classification. I could only do a few experiments before giving up.\n\n\n**References**\n\nhttps://discuss.pytorch.org/t/triplet-loss-in-pytorch/30634/3\n\nhttps://gombru.github.io/2019/04/03/ranking_loss/\n"},{"metadata":{"trusted":true},"cell_type":"code","source":"import numpy as np\nimport pandas as pd\nimport os\nimport sys\nimport cv2\nimport glob\nimport fastai\nimport PIL\nimport torch\nfrom functools import partial\nfrom fastai import *\nfrom fastai.vision import *\nfrom fastai.callbacks import *\nfrom fastai.basics import *\nfrom fastai.vision import learner\n\nfrom tqdm import tqdm\n\n\ntqdm.pandas()\n\nINPUT_PATH = \"../input/deepfake-detection-challenge\"\nVERBOSE = True\nEPS = 1e-5\nRUN_NOTEBOOK=False\nFACES_PATH = 'faces'\n\nos.makedirs(FACES_PATH, exist_ok=True)\nfrom dfdc_face_extractor import *\nfrom dfdc_fastai_reusables import *","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":true},"cell_type":"code","source":"pair_df = pd.read_csv(f'../input/fakereal-pairs-in-dfdc-test-videos/dfdc_test_video_pairs.csv')\n\nextractor = DFDCVideoFaceExtractor(backend='CV2')\npair_df = pair_df[:20]\nfor index, row in tqdm(pair_df.iterrows(), total=len(pair_df)):\n    video_filename = row[\"filename\"]\n    basename, _ = basename_and_ext(video_filename)\n    video_path=f'{INPUT_PATH}/test_videos/{video_filename}'\n    # extract_faces_with_cv2(video_path, basename, seq_length=10,stride=1, output_path=\"cv2_faces\")\n    extractor.extract_faces(video_path, seq_length=10,stride=1, faces_path=\"faces\", margin=1)\n\n    video_filename = row[\"original\"]\n    basename, _ = basename_and_ext(video_filename)\n    video_path=f'{INPUT_PATH}/test_videos/{video_filename}'\n    # extract_faces_with_cv2(video_path, basename, seq_length=10,stride=1, output_path=\"cv2_faces\")\n    extractor.extract_faces(video_path, seq_length=10,stride=1, faces_path=\"faces\", margin=1)\n\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class TripletImageList(DeepFakeImageList):\n    resize_option = 2\n       # 0 - No custom resize, resizing to be done with fastai transform \n       # 1 - keep original size, center and crop if too big or pad and reflect if too small\n       # 2 - center and size to fit with same aspect ratio and reflect the border\n\n    @classmethod\n    def from_df(cls, df,**kwargs):\n        return cls(items=range(len(df)),inner_df=df, **kwargs)\n\n    def get_image(self, pth):\n        im = PIL.Image.open(pth)\n        if self.resize_option == 1:\n            im = crop_pad(im)\n        elif self.resize_option == 2:\n            im = size_to_fit(im)\n        return im\n    \n\n    def get(self, i):\n        row = self.inner_df.iloc[i]\n        fake=row['fake']\n        fake = mp4_to_glob(fake)\n        original=row['original']\n        original = mp4_to_glob(original)\n        # Randomly selects one face per video\n        fake_files = glob.glob(fake)\n        fake1 = random.choice(fake_files)\n        \n        # Commented out as this notebook is using small amount of demo data\n        # fake2_files = [f for f in fake_files if f != fake1 ]\n        # fake2 = random.choice(fake2_files)\n        fake2 = random.choice(fake_files)\n\n        original_files = glob.glob(original)\n        original = random.choice(original_files)\n        \n        fake1 = self.get_image(fake1)\n        fake2 = self.get_image(fake2)\n        original = self.get_image(original)\n        im = concat(fake1, fake2, original)\n        im = to_fastai(im)\n        return im","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def get_triplet_data(bs=4, faces_path='faces', tfms=[[],[]] ):\n    unlike_df = pd.read_csv(f'../input/fakereal-pairs-in-dfdc-test-videos/dfdc_test_video_pairs.csv')\n    files = os.listdir(faces_path)\n    videos = [jpg_to_mp4name(f) for f in files]\n    \n    unlike_df = unlike_df[unlike_df['filename'].isin(videos)].copy()\n    unlike_df = unlike_df[unlike_df['original'].isin(videos)].copy()\n    unlike_df.rename(columns={'filename':'fake'}, inplace=True)\n    unlike_df['is_valid']=False\n    unlike_df.reset_index(inplace=True)\n    unlike_df['is_valid'].iloc[:4] = True\n    unlike_df.drop(columns=['index'], inplace=True)\n    unlike_df = unlike_df.sample(frac=1)\n    databunch = TripletImageList.from_df(unlike_df)\\\n                     .split_from_df(col='is_valid')\\\n                     .label_empty()\\\n                     .transform(tfms)\\\n                    .databunch(bs=bs).normalize(imagenet_stats)\n    \n    return databunch","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"tfms = get_dfdc_transforms()\ndata = get_triplet_data(tfms=tfms)\ndata.show_batch()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class TripletLoss(nn.Module):\n    \"Loss designed to increase difference of pairwise distance between fake-fake(anchor-positive) and fake-real(anchor-negative)\"\n    def __init__(self, margin=1.):\n        super(TripletLoss, self).__init__()\n        self.margin = margin\n        \n    def triplet_loss(self, a,p,n,  size_average=True):\n        d = nn.PairwiseDistance(p=2)\n        distance = d(a, p) - d(a, n) + self.margin \n        loss = torch.mean(torch.max(distance, torch.zeros_like(distance))) \n        return loss\n\n    def forward(self, triple_out, target, size_average=True):\n        a, p, n = triple_out[0], triple_out[1], triple_out[2]\n        return self.triplet_loss(a,p,n)\n    \nclass DapDan(Callback):\n    \"Reports difference of pairwise distance between fake-fake(anchor-positive) and fake-real(anchor-negative)\"\n    def on_epoch_begin(self, **kwargs):\n        self.distances = Tensor([])\n        \n    def on_batch_end(self, last_output:Tensor, last_target:Tensor, **kwargs):\n        a, p, n = last_output[0], last_output[1], last_output[2]\n        d = nn.PairwiseDistance(p=2)\n        distance = d(a, p) - d(a, n)\n        self.distances = torch.cat((self.distances, distance.squeeze(-1).cpu()))\n    \n    def on_epoch_end(self, last_metrics, **kwargs):\n        return add_metrics(last_metrics, self.distances.mean())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"class TripleNet(nn.Module):\n    # Passes 2 fakes and 1 real through the same encoder\n    def __init__(self, arch=models.resnet50, lin_ftrs=[256], emb_sz=128,ps=0., bn_final=True):\n        super(TripleNet, self).__init__()\n        self.arch, self.emb_sz = arch, emb_sz\n        self.lin_ftrs, self.ps, self.bn_final = lin_ftrs, ps, bn_final\n        self.body = learner.create_body(self.arch, True, learner.cnn_config(self.arch)['cut'])\n        self.head = learner.create_head(num_features_model(self.body)*2, self.emb_sz, self.lin_ftrs, self.ps,self.bn_final)\n        self.cnn = nn.Sequential(self.body, self.head)\n                                  \n    def trivide(self, triplet):\n        n = triplet.shape[-1] // 3\n        return torch.split(triplet, n, dim=-1)\n        \n    def forward(self, triplet):\n        fake0, fake1, original = self.trivide(triplet)\n        a = self.cnn(fake0)\n        p = self.cnn(fake1)\n        n = self.cnn(original)\n        return a, p, n\n\n    def get_embedding(self, x):\n        return self.cnn(x)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Training Phase 1\nmodel_dir = 'models/dfdc-triplet'\nos.makedirs(model_dir, exist_ok=True)\ndata = get_triplet_data()\nmodel = TripleNet()\n\nloss_func = TripletLoss()\ntriplet_1_learn = Learner(data,\n                model,\n                loss_func=loss_func,\n                metrics=[DapDan()],\n                model_dir=model_dir)\n\ntriplet_1_learn.fit(1)\n","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# Training Phase 2\ntriplet_cnn = triplet_1_learn.model.cnn\nhead = bn_drop_lin(128,1,True,p=0.25)\ntriplet_2_net = nn.Sequential(triplet_cnn, *head)\n\n\nmodel_dir = 'models/dfdc-triplet-2'\nos.makedirs(model_dir, exist_ok=True)\ndata = get_deepfakeimagelist_data()\n\n\ntriplet_2_learn = Learner(data,\n                model=triplet_2_net,\n                loss_func=BCEWithLogitsFlat(),\n                metrics=[DFDCAUROC(),RealLoss(),FakeLoss()],\n                model_dir=model_dir)\ntriplet_2_learn.split( lambda m: m[1])\ntriplet_2_learn.unfreeze()\ntriplet_2_learn.freeze_to(1)\ntriplet_2_learn.fit(1)","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":4}