{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":104308,"databundleVersionId":12660285,"sourceType":"competition"},{"sourceId":12324539,"sourceType":"datasetVersion","datasetId":7688715}],"dockerImageVersionId":31040,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"#!pip install image-quality","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-30T06:10:59.996126Z","iopub.execute_input":"2025-06-30T06:10:59.996363Z","iopub.status.idle":"2025-06-30T06:11:00.000162Z","shell.execute_reply.started":"2025-06-30T06:10:59.996340Z","shell.execute_reply":"2025-06-30T06:10:59.999474Z"},"jupyter":{"source_hidden":true}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#import torch modules\n\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torch.optim as optim\nfrom torch.nn import DataParallel\n#from torch.utils.tensorboard import SummaryWriter\nimport torch.utils.model_zoo as model_zoo\nfrom torch.nn import init\nimport torch.nn.parallel\nimport torch.backends.cudnn as cudnn\nimport torch.utils.data\nimport torch.optim as optim","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:00.000909Z","iopub.execute_input":"2025-06-30T06:11:00.001124Z","iopub.status.idle":"2025-06-30T06:11:01.735858Z","shell.execute_reply.started":"2025-06-30T06:11:00.001101Z","shell.execute_reply":"2025-06-30T06:11:01.735075Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#loss\n\nclass CrossEntropyLoss(nn.Module):\n    def __init__(self):\n        super().__init__()\n        self.loss_fn = nn.CrossEntropyLoss()\n\n    def forward(self, inputs, targets):\n        # Compute the cross-entropy loss\n        loss = self.loss_fn(inputs, targets)\n\n        return loss","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:01.736747Z","iopub.execute_input":"2025-06-30T06:11:01.737142Z","iopub.status.idle":"2025-06-30T06:11:01.741792Z","shell.execute_reply.started":"2025-06-30T06:11:01.737115Z","shell.execute_reply":"2025-06-30T06:11:01.741061Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#dlib face detector\n\nimport os\nimport sys\nimport time\nimport cv2\nimport dlib\nimport yaml\nimport logging\nimport datetime\nimport glob\nimport concurrent.futures\nimport numpy as np\nfrom tqdm import tqdm\nfrom pathlib import Path\nfrom imutils import face_utils\nfrom skimage import transform as trans\n\ndef get_keypts(image, face, predictor, face_detector):\n    # detect the facial landmarks for the selected face\n    shape = predictor(image, face)\n    \n    # select the key points for the eyes, nose, and mouth\n    leye = np.array([shape.part(37).x, shape.part(37).y]).reshape(-1, 2)\n    reye = np.array([shape.part(44).x, shape.part(44).y]).reshape(-1, 2)\n    nose = np.array([shape.part(30).x, shape.part(30).y]).reshape(-1, 2)\n    lmouth = np.array([shape.part(49).x, shape.part(49).y]).reshape(-1, 2)\n    rmouth = np.array([shape.part(55).x, shape.part(55).y]).reshape(-1, 2)\n    \n    pts = np.concatenate([leye, reye, nose, lmouth, rmouth], axis=0)\n\n    return pts\n\n\ndef extract_aligned_face_dlib(face_detector, predictor, image, res=256, mask=None):\n    def img_align_crop(img, landmark=None, outsize=None, scale=1.3, mask=None):\n        \"\"\" \n        align and crop the face according to the given bbox and landmarks\n        landmark: 5 key points\n        \"\"\"\n\n        M = None\n        target_size = [112, 112]\n        dst = np.array([\n            [30.2946, 51.6963],\n            [65.5318, 51.5014],\n            [48.0252, 71.7366],\n            [33.5493, 92.3655],\n            [62.7299, 92.2041]], dtype=np.float32)\n\n        if target_size[1] == 112:\n            dst[:, 0] += 8.0\n\n        dst[:, 0] = dst[:, 0] * outsize[0] / target_size[0]\n        dst[:, 1] = dst[:, 1] * outsize[1] / target_size[1]\n\n        target_size = outsize\n\n        margin_rate = scale - 1\n        x_margin = target_size[0] * margin_rate / 2.\n        y_margin = target_size[1] * margin_rate / 2.\n\n        # move\n        dst[:, 0] += x_margin\n        dst[:, 1] += y_margin\n\n        # resize\n        dst[:, 0] *= target_size[0] / (target_size[0] + 2 * x_margin)\n        dst[:, 1] *= target_size[1] / (target_size[1] + 2 * y_margin)\n\n        src = landmark.astype(np.float32)\n\n        # use skimage tranformation\n        tform = trans.SimilarityTransform()\n        tform.estimate(src, dst)\n        M = tform.params[0:2, :]\n\n        # M: use opencv\n        # M = cv2.getAffineTransform(src[[0,1,2],:],dst[[0,1,2],:])\n\n        img = cv2.warpAffine(img, M, (target_size[1], target_size[0]))\n\n        if outsize is not None:\n            img = cv2.resize(img, (outsize[1], outsize[0]))\n        \n        if mask is not None:\n            mask = cv2.warpAffine(mask, M, (target_size[1], target_size[0]))\n            mask = cv2.resize(mask, (outsize[1], outsize[0]))\n            return img, mask\n        else:\n            return img, None\n\n    # Image size\n    height, width = image.shape[:2]\n\n    # Convert to rgb\n    rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n\n    # Detect with dlib\n    faces = face_detector(rgb, 1)\n    if len(faces):\n        # For now only take the biggest face\n        face = max(faces, key=lambda rect: rect.width() * rect.height())\n        \n        # Get the landmarks/parts for the face in box d only with the five key points\n        landmarks = get_keypts(rgb, face, predictor, face_detector)\n\n        # Align and crop the face\n        cropped_face, mask_face = img_align_crop(rgb, landmarks, outsize=(res, res), mask=mask)\n        #cropped_face = cv2.cvtColor(cropped_face, cv2.COLOR_RGB2BGR)\n        \n        # Extract the all landmarks from the aligned face\n        face_align = face_detector(cropped_face, 1)\n        if len(face_align) == 0:\n            return None, None, None\n        landmark = predictor(cropped_face, face_align[0])\n        landmark = face_utils.shape_to_np(landmark)\n\n        return cropped_face, landmark, mask_face\n    \n    else:\n        return None, None, None\n\ndef extract_face(frame):\n    face_detector = dlib.get_frontal_face_detector()\n    predictor_path = '/kaggle/input/dfc-xception/shape_predictor_81_face_landmarks.dat'\n    face_predictor = dlib.shape_predictor(predictor_path)\n\n    frame_org = frame\n    frame_mask = None\n\n    # For now only take the biggest face\n\n    cropped_face, landmarks, _ = extract_aligned_face_dlib(face_detector, \n                                                        face_predictor, \n                                                        frame_org, \n                                                        mask=frame_mask)\n\n    if cropped_face is None:\n        cropped_face = cv2.resize(frame, (256,256))\n\n    return cropped_face\n","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:01.742497Z","iopub.execute_input":"2025-06-30T06:11:01.743199Z","iopub.status.idle":"2025-06-30T06:11:01.848037Z","shell.execute_reply.started":"2025-06-30T06:11:01.743176Z","shell.execute_reply":"2025-06-30T06:11:01.847494Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Xception Backbone\n\nimport os\nimport argparse\nimport logging\n\nimport math\n\nfrom typing import Union\n\nlogger = logging.getLogger(__name__)\n\n\n\nclass SeparableConv2d(nn.Module):\n    def __init__(self, in_channels, out_channels, kernel_size=1, stride=1, padding=0, dilation=1, bias=False):\n        super(SeparableConv2d, self).__init__()\n\n        self.conv1 = nn.Conv2d(in_channels, in_channels, kernel_size,\n                               stride, padding, dilation, groups=in_channels, bias=bias)\n        self.pointwise = nn.Conv2d(\n            in_channels, out_channels, 1, 1, 0, 1, 1, bias=bias)\n\n    def forward(self, x):\n        x = self.conv1(x)\n        x = self.pointwise(x)\n        return x\n\n\nclass Block(nn.Module):\n    def __init__(self, in_filters, out_filters, reps, strides=1, start_with_relu=True, grow_first=True):\n        super(Block, self).__init__()\n\n        if out_filters != in_filters or strides != 1:\n            self.skip = nn.Conv2d(in_filters, out_filters,\n                                  1, stride=strides, bias=False)\n            self.skipbn = nn.BatchNorm2d(out_filters)\n        else:\n            self.skip = None\n\n        self.relu = nn.ReLU(inplace=True)\n        rep = []\n\n        filters = in_filters\n        if grow_first:   # whether the number of filters grows first\n            rep.append(self.relu)\n            rep.append(SeparableConv2d(in_filters, out_filters,\n                                       3, stride=1, padding=1, bias=False))\n            rep.append(nn.BatchNorm2d(out_filters))\n            filters = out_filters\n\n        for i in range(reps-1):\n            rep.append(self.relu)\n            rep.append(SeparableConv2d(filters, filters,\n                                       3, stride=1, padding=1, bias=False))\n            rep.append(nn.BatchNorm2d(filters))\n\n        if not grow_first:\n            rep.append(self.relu)\n            rep.append(SeparableConv2d(in_filters, out_filters,\n                                       3, stride=1, padding=1, bias=False))\n            rep.append(nn.BatchNorm2d(out_filters))\n\n        if not start_with_relu:\n            rep = rep[1:]\n        else:\n            rep[0] = nn.ReLU(inplace=False)\n\n        if strides != 1:\n            rep.append(nn.MaxPool2d(3, strides, 1))\n        self.rep = nn.Sequential(*rep)\n\n    def forward(self, inp):\n        x = self.rep(inp)\n\n        if self.skip is not None:\n            skip = self.skip(inp)\n            skip = self.skipbn(skip)\n        else:\n            skip = inp\n\n        x += skip\n        return x\n\ndef add_gaussian_noise(ins, mean=0, stddev=0.2):\n    noise = ins.data.new(ins.size()).normal_(mean, stddev)\n    return ins + noise\n\n\nclass Xception(nn.Module):\n    \"\"\"\n    Xception optimized for the ImageNet dataset, as specified in\n    https://arxiv.org/pdf/1610.02357.pdf\n    \"\"\"\n\n    def __init__(self, xception_config):\n        \"\"\" Constructor\n        Args:\n            xception_config: configuration file with the dict format\n        \"\"\"\n        super(Xception, self).__init__()\n        self.num_classes = xception_config[\"num_classes\"]\n        self.mode = xception_config[\"mode\"]\n        inc = xception_config[\"inc\"]\n        dropout = xception_config[\"dropout\"]\n\n        # Entry flow\n        self.conv1 = nn.Conv2d(inc, 32, 3, 2, 0, bias=False)\n        \n        self.bn1 = nn.BatchNorm2d(32)\n        self.relu = nn.ReLU(inplace=True)\n\n        self.conv2 = nn.Conv2d(32, 64, 3, bias=False)\n        self.bn2 = nn.BatchNorm2d(64)\n        # do relu here\n\n        self.block1 = Block(\n            64, 128, 2, 2, start_with_relu=False, grow_first=True)\n        self.block2 = Block(\n            128, 256, 2, 2, start_with_relu=True, grow_first=True)\n        self.block3 = Block(\n            256, 728, 2, 2, start_with_relu=True, grow_first=True)\n\n        # middle flow\n        self.block4 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block5 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block6 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block7 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n\n        self.block8 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block9 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block10 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n        self.block11 = Block(\n            728, 728, 3, 1, start_with_relu=True, grow_first=True)\n\n        # Exit flow\n        self.block12 = Block(\n            728, 1024, 2, 2, start_with_relu=True, grow_first=False)\n\n        self.conv3 = SeparableConv2d(1024, 1536, 3, 1, 1)\n        self.bn3 = nn.BatchNorm2d(1536)\n\n        # do relu here\n        self.conv4 = SeparableConv2d(1536, 2048, 3, 1, 1)\n        self.bn4 = nn.BatchNorm2d(2048)\n        # used for iid\n        final_channel = 2048\n        if self.mode == 'adjust_channel_iid':\n            final_channel = 512\n            self.mode = 'adjust_channel'\n        self.last_linear = nn.Linear(final_channel, self.num_classes)\n        if dropout:\n            self.last_linear = nn.Sequential(\n                nn.Dropout(p=dropout),\n                nn.Linear(final_channel, self.num_classes)\n            )\n\n        self.adjust_channel = nn.Sequential(\n            nn.Conv2d(2048, 512, 1, 1),\n            nn.BatchNorm2d(512),\n            nn.ReLU(inplace=False),\n        )\n           \n    def fea_part1_0(self, x):\n        x = self.conv1(x)\n        x = self.bn1(x)\n        x = self.relu(x)  \n\n        return x\n\n    def fea_part1_1(self, x):  \n        \n        x = self.conv2(x)\n        x = self.bn2(x)\n        x = self.relu(x) \n\n        return x\n    \n    def fea_part1(self, x):\n        x = self.conv1(x)\n        x = self.bn1(x)\n        x = self.relu(x)     \n        \n        x = self.conv2(x)\n        x = self.bn2(x)\n        x = self.relu(x) \n\n        return x\n    \n    def fea_part2(self, x):\n        x = self.block1(x)\n        x = self.block2(x)\n        x = self.block3(x)\n\n        return x\n\n    def fea_part3(self, x):\n        if self.mode == \"shallow_xception\":\n            return x\n        else:\n            x = self.block4(x)\n            x = self.block5(x)\n            x = self.block6(x)\n            x = self.block7(x)\n        return x\n\n    def fea_part4(self, x):\n        if self.mode == \"shallow_xception\":\n            x = self.block12(x)\n        else:\n            x = self.block8(x)\n            x = self.block9(x)\n            x = self.block10(x)\n            x = self.block11(x)\n            x = self.block12(x)\n        return x\n\n    def fea_part5(self, x):\n        x = self.conv3(x)\n        x = self.bn3(x)\n        x = self.relu(x)\n\n        x = self.conv4(x)\n        x = self.bn4(x)\n\n        return x\n     \n    def features(self, input):\n        x = self.fea_part1(input)    \n\n        x = self.fea_part2(x)\n        x = self.fea_part3(x)\n        x = self.fea_part4(x)\n\n        x = self.fea_part5(x)\n\n        if self.mode == 'adjust_channel':\n            x = self.adjust_channel(x)\n        \n        return x\n\n    def classifier(self, features,id_feat=None):\n        # for iid\n        if self.mode == 'adjust_channel':\n            x = features\n        else:\n            x = self.relu(features)\n\n        if len(x.shape) == 4:\n            x = F.adaptive_avg_pool2d(x, (1, 1))\n            x = x.view(x.size(0), -1)\n        self.last_emb = x\n        # for iid\n        if id_feat!=None:\n            out = self.last_linear(x-id_feat)\n        else:\n            out = self.last_linear(x)\n        return out\n\n    def forward(self, input):\n        x = self.features(input)\n        out = self.classifier(x)\n        return out, x","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:01.848771Z","iopub.execute_input":"2025-06-30T06:11:01.848968Z","iopub.status.idle":"2025-06-30T06:11:01.872025Z","shell.execute_reply.started":"2025-06-30T06:11:01.848951Z","shell.execute_reply":"2025-06-30T06:11:01.871296Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Xception Detector\nimport os\nimport datetime\nimport logging\nimport numpy as np\nfrom sklearn import metrics\nfrom typing import Union\nfrom collections import defaultdict\nimport abc\n\n\nlogger = logging.getLogger(__name__)\n\nclass XceptionDetector(nn.Module, metaclass=abc.ABCMeta):\n    def __init__(self, config):\n        super().__init__()\n        self.config = config\n        self.backbone = self.build_backbone(config)\n        self.loss_func = self.build_loss(config)\n        self.prob, self.label = [], []\n        self.video_names = []\n        self.correct, self.total = 0, 0\n        \n    def build_backbone(self, config):\n        # prepare the backbone\n        model_config = config['backbone_config']\n        backbone = Xception(model_config)\n        # if donot load the pretrained weights, fail to get good results\n        state_dict = torch.load(config['pretrained'])\n        for name, weights in state_dict.items():\n            if 'pointwise' in name:\n                state_dict[name] = weights.unsqueeze(-1).unsqueeze(-1)\n        state_dict = {k:v for k, v in state_dict.items() if 'fc' not in k}\n        backbone.load_state_dict(state_dict, False)\n        logger.info('Load pretrained model successfully!')\n        return backbone\n    \n    def build_loss(self, config):\n        return CrossEntropyLoss()\n    \n    def features(self, data_dict: dict) -> torch.tensor:\n        return self.backbone.features(data_dict['image']) #32,3,256,256\n\n    def classifier(self, features: torch.tensor) -> torch.tensor:\n        return self.backbone.classifier(features)\n    \n    def get_losses(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        loss = self.loss_func(pred, label)\n        overall_loss = loss\n        loss_dict = {'overall': overall_loss, 'cls': loss,}\n        return loss_dict\n    \n    def get_train_metrics(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        # compute metrics for batch data\n        auc, eer, acc, ap = calculate_metrics_for_train(label.detach(), pred.detach())\n        metric_batch_dict = {'acc': acc, 'auc': auc, 'eer': eer, 'ap': ap}\n        # we dont compute the video-level metrics for training\n        self.video_names = []\n        return metric_batch_dict\n\n    def forward(self, data_dict: dict, inference=False) -> dict:\n        # get the features by backbone\n        features = self.features(data_dict)\n        # get the prediction by classifier\n        pred = self.classifier(features)\n        # get the probability of the pred\n        prob = torch.softmax(pred, dim=1)[:, 1]\n        # build the prediction dict for each output\n        pred_dict = {'cls': pred, 'prob': prob, 'feat': features}\n        return pred_dict","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:01.874001Z","iopub.execute_input":"2025-06-30T06:11:01.874179Z","iopub.status.idle":"2025-06-30T06:11:02.390817Z","shell.execute_reply.started":"2025-06-30T06:11:01.874164Z","shell.execute_reply":"2025-06-30T06:11:02.390254Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#SPSL Detector\n\nimport os\nimport datetime\nimport logging\nimport numpy as np\nfrom sklearn import metrics\nfrom typing import Union\nfrom collections import defaultdict\nimport random\n\nlogger = logging.getLogger(__name__)\n\nclass SpslDetector(nn.Module, metaclass=abc.ABCMeta):\n    def __init__(self, config):\n        super().__init__()\n        self.config = config\n        self.backbone = self.build_backbone(config)\n        self.loss_func = self.build_loss(config)\n\n    def build_backbone(self, config):\n        # prepare the backbone\n        model_config = config['backbone_config']\n        backbone = Xception(model_config)\n\n        # To get a good performance, use the ImageNet-pretrained Xception model\n        state_dict = torch.load(config['pretrained'])\n        for name, weights in state_dict.items():\n            if 'pointwise' in name:\n                state_dict[name] = weights.unsqueeze(-1).unsqueeze(-1)\n        state_dict = {k:v for k, v in state_dict.items() if 'fc' not in k}\n\n        # remove conv1 from state_dict\n        conv1_data = state_dict.pop('conv1.weight')\n\n        backbone.load_state_dict(state_dict, False)\n        logger.info('Load pretrained model from {}'.format(config['pretrained']))\n\n        # copy on conv1\n        # let new conv1 use old param to balance the network\n        backbone.conv1 = nn.Conv2d(4, 32, 3, 2, 0, bias=False)\n        avg_conv1_data = conv1_data.mean(dim=1, keepdim=True)  # average across the RGB channels\n        backbone.conv1.weight.data = avg_conv1_data.repeat(1, 4, 1, 1)  # repeat the averaged weights across the 4 new channels\n        logger.info('Copy conv1 from pretrained model')\n        return backbone\n\n    def build_loss(self, config):\n        return CrossEntropyLoss()\n    \n    def features(self, data_dict: dict, phase_fea) -> torch.tensor:\n        features = torch.cat((data_dict['image'], phase_fea), dim=1)\n        return self.backbone.features(features)\n\n    def classifier(self, features: torch.tensor) -> torch.tensor:\n        return self.backbone.classifier(features)\n    \n    def get_losses(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        loss = self.loss_func(pred, label)\n        loss_dict = {'overall': loss}\n        return loss_dict\n\n    def get_train_metrics(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        # compute metrics for batch data\n        auc, eer, acc, ap = calculate_metrics_for_train(label.detach(), pred.detach())\n        metric_batch_dict = {'acc': acc, 'auc': auc, 'eer': eer, 'ap': ap}\n        # we dont compute the video-level metrics for training\n        self.video_names = []\n        return metric_batch_dict\n    \n    def forward(self, data_dict: dict, inference=False) -> dict:\n        # get the phase features\n        phase_fea = self.phase_without_amplitude(data_dict['image'])\n        # bp\n        features = self.features(data_dict, phase_fea)\n        # get the prediction by classifier\n        pred = self.classifier(features)\n        # get the probability of the pred\n        prob = torch.softmax(pred, dim=1)[:, 1]\n        # build the prediction dict for each output\n        pred_dict = {'cls': pred, 'prob': prob, 'feat': features}\n\n        return pred_dict\n\n    def phase_without_amplitude(self, img):\n        # Convert to grayscale\n        gray_img = torch.mean(img, dim=1, keepdim=True) # shape: (batch_size, 1, 256, 256)\n        # Compute the DFT of the input signal\n        X = torch.fft.fftn(gray_img,dim=(-1,-2))\n        #X = torch.fft.fftn(img)\n        # Extract the phase information from the DFT\n        phase_spectrum = torch.angle(X)\n        # Create a new complex spectrum with the phase information and zero magnitude\n        reconstructed_X = torch.exp(1j * phase_spectrum)\n        # Use the IDFT to obtain the reconstructed signal\n        reconstructed_x = torch.real(torch.fft.ifftn(reconstructed_X,dim=(-1,-2)))\n        # reconstructed_x = torch.real(torch.fft.ifftn(reconstructed_X))\n        return reconstructed_x","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:02.391516Z","iopub.execute_input":"2025-06-30T06:11:02.391907Z","iopub.status.idle":"2025-06-30T06:11:02.404965Z","shell.execute_reply.started":"2025-06-30T06:11:02.391882Z","shell.execute_reply":"2025-06-30T06:11:02.404272Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#f3net Detector\n\nimport os\nimport datetime\nimport logging\nimport numpy as np\nfrom sklearn import metrics\nfrom typing import Union\nfrom collections import defaultdict\n\nlogger = logging.getLogger(__name__)\n\nclass F3netDetector(nn.Module, metaclass=abc.ABCMeta):\n    def __init__(self, config):\n        super().__init__()\n        self.config = config\n        self.backbone = self.build_backbone(config)\n        self.loss_func = self.build_loss(config)\n        # modules only use in FAD\n        img_size = config['resolution']\n        self.FAD_head = FAD_Head(img_size)\n\n    def build_backbone(self, config):\n        # prepare the backbone\n        model_config = config['backbone_config']\n        backbone = Xception(model_config)\n        \n        # To get a good performance, use the ImageNet-pretrained Xception model\n        state_dict = torch.load(config['pretrained'])\n        for name, weights in state_dict.items():\n            if 'pointwise' in name:\n                state_dict[name] = weights.unsqueeze(-1).unsqueeze(-1)\n        state_dict = {k:v for k, v in state_dict.items() if 'fc' not in k}\n        conv1_data = state_dict['conv1.weight'].data\n        backbone.load_state_dict(state_dict, False)\n        logger.info('Load pretrained model from {}'.format(config['pretrained']))\n\n        # copy on conv1\n        # let new conv1 use old param to balance the network\n        backbone.conv1 = nn.Conv2d(12, 32, 3, 2, 0, bias=False)\n        for i in range(4):\n           backbone.conv1.weight.data[:, i*3:(i+1)*3, :, :] = conv1_data / 4.0\n        logger.info('Copy conv1 from pretrained model')\n        return backbone\n    \n    def build_loss(self, config):\n        return CrossEntropyLoss()\n    \n    def features(self, data_dict: dict) -> torch.tensor:\n        fea_FAD = self.FAD_head(data_dict['image']) # [B, 12, 256, 256]\n        return self.backbone.features(fea_FAD)\n\n    def classifier(self, features: torch.tensor) -> torch.tensor:\n        return self.backbone.classifier(features)\n    \n    def get_losses(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        loss = self.loss_func(pred, label)\n        loss_dict = {'overall': loss}\n        return loss_dict\n\n    def get_train_metrics(self, data_dict: dict, pred_dict: dict) -> dict:\n        label = data_dict['label']\n        pred = pred_dict['cls']\n        # compute metrics for batch data\n        auc, eer, acc, ap = calculate_metrics_for_train(label.detach(), pred.detach())\n        metric_batch_dict = {'acc': acc, 'auc': auc, 'eer': eer, 'ap': ap}\n        return metric_batch_dict\n\n    def forward(self, data_dict: dict, inference=False) -> dict:\n        # get the features by backbone\n        features = self.features(data_dict)\n        # get the prediction by classifier\n        pred = self.classifier(features)\n        # get the probability of the pred\n        prob = torch.softmax(pred, dim=1)[:, 1]\n        # build the prediction dict for each output\n        pred_dict = {'cls': pred, 'prob': prob, 'feat': features}\n\n        return pred_dict\n\n\n# ===================================== other modules for F3Net # =====================================\n\n\n# Filter Module\nclass Filter(nn.Module):\n    def __init__(self, size, band_start, band_end, use_learnable=True, norm=False):\n        super(Filter, self).__init__()\n        self.use_learnable = use_learnable\n\n        self.base = nn.Parameter(torch.tensor(generate_filter(band_start, band_end, size)), requires_grad=False)\n        if self.use_learnable:\n            self.learnable = nn.Parameter(torch.randn(size, size), requires_grad=True)\n            self.learnable.data.normal_(0., 0.1)\n\n        self.norm = norm\n        if norm:\n            self.ft_num = nn.Parameter(torch.sum(torch.tensor(generate_filter(band_start, band_end, size))), requires_grad=False)\n\n\n    def forward(self, x):\n        if self.use_learnable:\n            filt = self.base + norm_sigma(self.learnable)\n        else:\n            filt = self.base\n\n        if self.norm:\n            y = x * filt / self.ft_num\n        else:\n            y = x * filt\n        return y\n\n\n# FAD Module\nclass FAD_Head(nn.Module):\n    def __init__(self, size):\n        super(FAD_Head, self).__init__()\n\n        # init DCT matrix\n        self._DCT_all = nn.Parameter(torch.tensor(DCT_mat(size)).float(), requires_grad=False)\n        self._DCT_all_T = nn.Parameter(torch.transpose(torch.tensor(DCT_mat(size)).float(), 0, 1), requires_grad=False)\n\n        # define base filters and learnable\n        # 0 - 1/16 || 1/16 - 1/8 || 1/8 - 1\n        low_filter = Filter(size, 0, size // 2.82)\n        middle_filter = Filter(size, size // 2.82, size // 2)\n        high_filter = Filter(size, size // 2, size * 2)\n        all_filter = Filter(size, 0, size * 2)\n\n        self.filters = nn.ModuleList([low_filter, middle_filter, high_filter, all_filter])\n\n    def forward(self, x):\n        # DCT\n        x_freq = self._DCT_all @ x @ self._DCT_all_T    # [N, 3, 299, 299]\n\n        # 4 kernel\n        y_list = []\n        for i in range(4):\n            x_pass = self.filters[i](x_freq)  # [N, 3, 299, 299]\n            y = self._DCT_all_T @ x_pass @ self._DCT_all    # [N, 3, 299, 299]\n            y_list.append(y)\n        out = torch.cat(y_list, dim=1)    # [N, 12, 299, 299]\n        return out\n\n# utils\ndef DCT_mat(size):\n    m = [[ (np.sqrt(1./size) if i == 0 else np.sqrt(2./size)) * np.cos((j + 0.5) * np.pi * i / size) for j in range(size)] for i in range(size)]\n    return m\n\ndef generate_filter(start, end, size):\n    return [[0. if i + j > end or i + j < start else 1. for j in range(size)] for i in range(size)]\n\ndef norm_sigma(x):\n    return 2. * torch.sigmoid(x) - 1.","metadata":{"trusted":true,"jupyter":{"source_hidden":true},"execution":{"iopub.status.busy":"2025-06-30T06:11:02.405682Z","iopub.execute_input":"2025-06-30T06:11:02.405908Z","iopub.status.idle":"2025-06-30T06:11:02.431540Z","shell.execute_reply.started":"2025-06-30T06:11:02.405883Z","shell.execute_reply":"2025-06-30T06:11:02.431013Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#Inference\nimport numpy as np\n\nfrom PIL import Image\n\n\n\"\"\"\neval pretained model.\n\"\"\"\nimport os\nimport numpy as np\nfrom os.path import join\nimport cv2\nimport random\nimport datetime\nimport time\nimport yaml\nimport pickle\nfrom tqdm import tqdm\nfrom copy import deepcopy\nfrom PIL import Image as pil_image\nimport argparse\n\n\ndef init_seed(config):\n    if config['manualSeed'] is None:\n        config['manualSeed'] = random.randint(1, 10000)\n    random.seed(config['manualSeed'])\n    torch.manual_seed(config['manualSeed'])\n    if config['cuda']:\n        torch.cuda.manual_seed_all(config['manualSeed'])\n\n@torch.no_grad()\ndef inference(model, data_dict):\n    predictions = model(data_dict, inference=True)\n    return predictions\n    \ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(device)\n\n\ndef load_image(image):\n        # Open the image file\n        # Convert the image to a numpy array\n        image_data_rgb = np.array(image)\n\n        # Convert from (height, width, 3) to (3, height, width)\n        image_data = np.transpose(image_data_rgb, (2, 0, 1))\n\n        # Rescale the values back to the range (-1, 1)\n        image_data = np.round(image_data, 4)\n\n        image_data = image_data.astype(np.float32)\n        image_data = (image_data / 127.5) - 1.0  # Reverse the normalization\n    \n\n        # Convert to a PyTorch tensor\n        tensor_image = torch.tensor(image_data).cuda()  # Assuming you want to keep it on the GPU\n        return tensor_image\n\ndef init_model(detector_path=\"/kaggle/input/dfc-xception/spsl.yaml\", \n          weights_path=\"/kaggle/input/dfc-xception/spsl_c23_e30.pth\",\n              detector=SpslDetector):\n    # init seed\n    \n    with open(detector_path, 'r') as f:\n        config = yaml.safe_load(f)\n    with open('/kaggle/input/dfc-xception/test_config.yaml', 'r') as f:\n        config2 = yaml.safe_load(f)\n    config.update(config2)\n    if 'label_dict' in config:\n        config2['label_dict']=config['label_dict']\n    init_seed(config)\n    # set cudnn benchmark if needed\n    if config['cudnn']:\n        cudnn.benchmark = True\n\n    config['pretrained'] = \"/kaggle/input/dfc-xception/xception-b5690688.pth\"\n\n    #model = XceptionDetector(config).to(device)\n    model = detector(config).to(device)\n    \n    epoch = 0\n    if weights_path:\n        ckpt = torch.load(weights_path, map_location=device)\n        ckpt = {k.replace('module.', ''): v for k, v in ckpt.items()}  \n        \n        model.load_state_dict(ckpt, strict=True)\n        print('===> Load checkpoint done!')\n    model.eval()\n\n    return model\n\n\ndef infer(model, image):\n        '''\n        label\n        image a tensor of shape torch.Size([batch_num,3, 256, 256])\n        tensor([0, 1, 1, 1, 1, 1, 0, 1, 1, 0, 1, 1, 1, 1, 0, 1, 1, 1, 1, 0, 1, 1, 1, 1,\n                0, 1, 1, 0, 1, 1, 0, 0], device='cuda:0')\n        landmark\n        None\n        mask\n        None'\n        '''\n        tensor_image_cuda = load_image(image)\n        # Add a new dimension at the beginning to make it [1, 3, 256, 256]\n        tensor_image_cuda = tensor_image_cuda.unsqueeze(0)\n\n        data_dict = {}\n        data_dict['image'] = tensor_image_cuda   \n        data_dict['landmark'] = None\n        data_dict['mask'] = None\n        pseudo_label = 1 \n        data_dict['label'] = torch.tensor([pseudo_label], device=device)\n\n        predictions = inference(model, data_dict)\n        return predictions","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-30T06:11:02.432215Z","iopub.execute_input":"2025-06-30T06:11:02.432457Z","iopub.status.idle":"2025-06-30T06:11:02.480654Z","shell.execute_reply.started":"2025-06-30T06:11:02.432439Z","shell.execute_reply":"2025-06-30T06:11:02.479998Z"},"jupyter":{"source_hidden":true}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#main code\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\n\nimport cv2\nimport skimage\n#import imquality.brisque as brisque\n\ndef extract_frames(video_path, max_frames=4):\n    cap = cv2.VideoCapture(video_path)\n    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n\n    # 샘플링할 프레임 인덱스 계산\n    if total_frames <= max_frames:\n        indices = list(range(total_frames))\n    else:\n        indices = np.linspace(0, total_frames - 1, max_frames, dtype=int)\n\n    frames = []\n    current_idx = 0\n    frame_idx_set = set(indices)\n\n    while cap.isOpened():\n        try:\n            ret, frame = cap.read()\n        except Exception as e:\n            print(e)\n        if not ret:\n            break\n        if current_idx in frame_idx_set:\n            frames.append(frame)\n            if len(frames) == max_frames:\n                break\n        current_idx += 1\n\n    cap.release()\n    return np.array(frames)\n\n    \nimport os\n\nprint(\"model loading start\")\n\nmodel_c23 = init_model(detector_path=\"/kaggle/input/dfc-xception/spsl.yaml\", \n          weights_path=\"/kaggle/input/dfc-xception/spsl_c23.pth\",\n                      detector=SpslDetector)\n\nmodel_c40 = init_model(detector_path=\"/kaggle/input/dfc-xception/f3net.yaml\", \n          weights_path=\"/kaggle/input/dfc-xception/f3net_c40.pth\",\n                      detector=F3netDetector)\n\nthreshold = 100\n\ndef apply_c40(img, quality=40):\n    # img: numpy array (H, W, 3), BGR format\n    encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), quality]\n    result, encimg = cv2.imencode('.jpg', img, encode_param)\n    if not result:\n        raise Exception(\"Could not encode image!\")\n    # Decode it back\n    decimg = cv2.imdecode(encimg, 1)\n    return decimg\n\n\ndef infer_file(path):\n    frames = extract_frames(path)\n            \n    p_c23 = 0\n    p_c40 = 0\n    p_c40_2 = 0\n    bs = 0\n    \n    nsample = len(frames)\n            \n    for frame in frames:\n        face = extract_face(frame)\n        try:\n            pred = infer(model_c23, face)\n            prob = pred['prob'].item()\n            p_c23 += prob\n\n            pred = infer(model_c40, face)\n            prob = pred['prob'].item()\n            p_c40 += prob           \n            \n            pred = infer(model_c40, apply_c40(face))\n            prob = pred['prob'].item()\n            p_c40_2 += prob\n        except Exception as e:\n            print(e, end=\" \")\n            nsample -= 1\n\n    if nsample >= 1:\n        p_c23 /= nsample\n        p_c40 /= nsample\n        p_c40_2 /= nsample\n    \n    return p_c23, p_c40, p_c40_2\n\n\ndef find_thres(name, y_true, y_score, threshold):\n    from sklearn.metrics import roc_curve,accuracy_score, f1_score, precision_recall_curve\n    print(name +\" threshold\")\n    \n    precisions, recalls, thresholds = precision_recall_curve(y_true, y_score)\n    f1_scores = 2 * (precisions * recalls) / (precisions + recalls + 1e-8)\n    best_idx = np.argmax(f1_scores)\n    \n    thres = thresholds[best_idx]\n\n    video_preds = [1 if p >= threshold else 0 for p in y_score]\n\n    # 성능 지표\n    acc = accuracy_score(y_true, video_preds)\n    f1 = f1_score(y_true, video_preds)\n\n    print(threshold, \" thres: acc: \", acc, \" f1: \", f1)\n\n    video_preds = [1 if p >= thres else 0 for p in y_score]\n\n    # 성능 지표\n    acc = accuracy_score(y_true, video_preds)\n    f1 = f1_score(y_true, video_preds)\n\n    print(thres, \" thres: acc: \", acc, \" f1: \", f1)\n\nyt = []\nys_c23 = []\nys_c40 = []\nys_c40_2 = []\nys_m = []\nys_m_2 = []\nthreshold = 0.2573552876710892\n\ndef infer_folder(path = '/kaggle/input/dfc-xception', resfile = \"submission.csv\"):\n    print(\"processing start\")\n    video_names = []\n    labels = []\n    \n    for dirname, _, filenames in os.walk(path):\n        print(dirname)\n        ans=1\n        if \"0_\" in dirname:\n            ans = 0\n            \n        cnt = 0\n        print('[',end='')\n        for filename in filenames:\n\n            f = os.path.join(dirname, filename)\n            \n            if not \".mp4\" in f and not \".avi\" in f:\n                continue\n            else:\n                print(\"=\",end='')\n                \n            video_names.append(filename)\n            pc23, pc40, pc40_2= infer_file(f)\n            \n            res = 0\n            if (pc23 + pc40) / 2 > threshold:\n                res = 1\n            \n            labels.append(res)\n\n            yt.append(ans)\n            ys_c23.append(pc23)\n            ys_c40.append(pc40)\n            ys_c40_2.append(pc40_2)\n            ys_m.append((pc23+pc40)/2)\n            ys_m_2.append((pc23+pc40_2)/2)\n\n            \n        print(']')\n    \n    print('done')\n    df = pd.DataFrame({\n        'ID': video_names,\n        'label': labels\n    })\n    print(df)\n    df.to_csv(resfile, index=False)\n    print(\"df to csv done\")\n    \n\ninfer_folder(\"/kaggle/input/dfc-xception/1\", \"submission_1.csv\")\ninfer_folder(\"/kaggle/input/dfc-xception/2\", \"submission_2.csv\")\ninfer_folder(\"/kaggle/input/dfc-xception/3\", \"submission_3.csv\")\n'''\nfind_thres(\"c23\", yt, ys_c23, threshold)\nfind_thres(\"c40\", yt, ys_c40, threshold)\nfind_thres(\"c40_2\", yt, ys_c40_2,threshold)\nfind_thres(\"mean\", yt, ys_m, threshold)\nfind_thres(\"mean_2\", yt, ys_m_2, threshold)\n'''\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-06-30T06:11:02.481451Z","iopub.execute_input":"2025-06-30T06:11:02.482083Z","iopub.status.idle":"2025-06-30T06:27:12.125303Z","shell.execute_reply.started":"2025-06-30T06:11:02.482060Z","shell.execute_reply":"2025-06-30T06:27:12.124676Z"},"jupyter":{"source_hidden":true}},"outputs":[],"execution_count":null}]}