{"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":"nvidiaTeslaT4","dataSources":[{"sourceId":104308,"databundleVersionId":12660285,"sourceType":"competition"},{"sourceId":12257263,"sourceType":"datasetVersion","datasetId":7713178}],"dockerImageVersionId":31040,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!cp -r /kaggle/input/kcc-dfc/* /kaggle/working/ ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-23T12:53:21.015693Z","iopub.execute_input":"2025-06-23T12:53:21.016242Z","iopub.status.idle":"2025-06-23T12:53:26.296074Z","shell.execute_reply.started":"2025-06-23T12:53:21.016209Z","shell.execute_reply":"2025-06-23T12:53:26.294991Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install -r requirements.txt","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-23T12:53:29.996670Z","iopub.execute_input":"2025-06-23T12:53:29.997316Z","iopub.status.idle":"2025-06-23T12:55:24.247345Z","shell.execute_reply.started":"2025-06-23T12:53:29.997285Z","shell.execute_reply":"2025-06-23T12:55:24.246489Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport glob\nimport json\nimport sys\n\nimport torch\nimport numpy as np\nimport pandas as pd\nfrom tqdm import tqdm\nfrom PIL import Image\nimport cv2\n\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as transforms\nfrom torchvision.transforms.functional import center_crop\n\n# feat RetinaFace\nfrom feat.utils.image_operations import convert_image_to_tensor, convert_color_vector_to_tensor\nfrom feat.face_detectors.Retinaface.Retinaface_model import (\n    RetinaFace as PyFeatRetina,\n    postprocess_retinaface\n)\n\nfrom model import DeepFakeDetector\n\n# =====================================================\n# 1) 상수 세팅\n# =====================================================\nDEVICE       = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nRETINA_DIR   = \"./models/retinaface\"\nADA_DIR      = \"./models/cvlface_adaface_ir50_ms1mv2\"\nFRAME_SIZE   = (224, 224)\nREC_INPUT_SZ = (112, 112)\n\nBATCH_SIZE   = 4\nNUM_FRAMES   = 16\nSTRATEGY     = 'consecutive'\nTHRESHOLD    = 0.5\n\n# =====================================================\n# 2) RetinaFace 모델 로딩\n# =====================================================\nwith open(os.path.join(RETINA_DIR, \"config.json\"), \"r\") as f:\n    retina_cfg = json.load(f)\nretina_model = PyFeatRetina(cfg=retina_cfg, phase=\"test\")\nckpt = torch.load(\n    os.path.join(RETINA_DIR, \"mobilenet0.25_Final.pth\"),\n    map_location=DEVICE,\n    weights_only=True\n)\nretina_model.load_state_dict(ckpt)\nretina_model.eval().to(DEVICE)\n\n_MEAN_VEC = np.array([123, 117, 104], dtype=np.float32)\n\ndef detect_faces(img: Image.Image, threshold=0.6):\n    img_np       = np.array(img)\n    frame_tensor = convert_image_to_tensor(img_np).to(DEVICE)\n    mean_tensor  = convert_color_vector_to_tensor(_MEAN_VEC).to(DEVICE)\n    x            = frame_tensor - mean_tensor\n\n    loc, conf, landm = retina_model(x)\n    out = postprocess_retinaface(loc, conf, landm, retina_cfg, x, device=DEVICE)\n\n    boxes_full = out[\"boxes\"]\n    scores_all = out[\"scores\"]\n    keep       = scores_all >= threshold\n    if keep.sum() == 0:\n        # 변경: device는 키워드 인자로 전달합니다.\n        return (\n            torch.empty((0, 4), device=DEVICE),\n            torch.empty((0,),    device=DEVICE),\n            torch.empty((0, 10), device=DEVICE)\n        )\n\n    bf = boxes_full[keep]\n    sf = scores_all[keep]\n    x1, y2, x2, y1 = bf[:,1], bf[:,2], bf[:,3], bf[:,4]\n    boxes = torch.stack([x1, y1, x2, y2], dim=1)\n    idx   = torch.argsort(sf, descending=True)\n    return boxes[idx], sf[idx], out[\"landmarks\"][keep][idx]\n\n# =====================================================\n# 3) AdaFace 모델 로딩\n# =====================================================\nsys.path.append(os.path.abspath(ADA_DIR))\nfrom wrapper import ModelConfig, CVLFaceRecognitionModel\n\ncfg = ModelConfig(model_path=os.path.join(ADA_DIR, \"model.safetensors\"))\nrec_model = CVLFaceRecognitionModel(cfg)\nrec_model.eval().to(DEVICE)\n\ndef extract_features(x: torch.Tensor) -> torch.Tensor:\n    m = rec_model.model\n    if hasattr(m, 'net'):\n        m = m.net\n    x = m.input_layer(x)\n    return m.body(x)\n\n# =====================================================\n# 4) Dataset 클래스\n# =====================================================\nclass ExternalVideoDataset(Dataset):\n    def __init__(self, video_files, num_frames=NUM_FRAMES, strategy=STRATEGY):\n        self.video_files = video_files\n        self.num_frames  = num_frames\n        self.strategy    = strategy\n        self.resize      = transforms.Resize(FRAME_SIZE)\n        self.to_tensor   = transforms.ToTensor()\n\n    def __len__(self):\n        return len(self.video_files)\n\n    def __getitem__(self, idx):\n        vp    = self.video_files[idx]\n        cap   = cv2.VideoCapture(vp)\n        total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n        if total <= 0:\n            raise RuntimeError(f\"Cannot read video: {vp}\")\n\n        if total < self.num_frames:\n            idxs = list(range(total)) + [total-1]*(self.num_frames-total)\n            idxs = idxs[:self.num_frames]\n        elif self.strategy == 'uniform':\n            idxs = [int(i * total / self.num_frames) for i in range(self.num_frames)]\n        else:\n            idxs = list(range(self.num_frames))\n\n        feats, imgs = [], []\n        for f in idxs:\n            cap.set(cv2.CAP_PROP_POS_FRAMES, f)\n            ret, frame = cap.read()\n            if not ret:\n                cap.set(cv2.CAP_PROP_POS_FRAMES, total-1)\n                ret, frame = cap.read()\n\n            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n            img       = Image.fromarray(frame_rgb)\n            img224    = self.resize(img)\n\n            boxes, _, _ = detect_faces(img224, threshold=THRESHOLD)\n            side        = min(FRAME_SIZE)\n            if boxes.shape[0] == 0:\n                crop = center_crop(img224, (side, side)).resize(REC_INPUT_SZ)\n            else:\n                x1, y1, x2, y2 = boxes[0].to(torch.int).tolist()\n                W, H = FRAME_SIZE\n                x1 = max(0, min(x1, W)); y1 = max(0, min(y1, H))\n                x2 = max(0, min(x2, W)); y2 = max(0, min(y2, H))\n                if (x2 - x1) <= 0 or (y2 - y1) <= 0:\n                    crop = center_crop(img224, (side, side)).resize(REC_INPUT_SZ)\n                else:\n                    crop = img224.crop((x1, y1, x2, y2)).resize(REC_INPUT_SZ)\n\n            xt = self.to_tensor(crop).unsqueeze(0).to(DEVICE)\n            with torch.no_grad():\n                fmap = extract_features(xt)\n            feats.append(fmap.squeeze(0).cpu())\n            imgs.append(self.to_tensor(img224))\n\n        cap.release()\n        return torch.stack(feats), torch.stack(imgs), os.path.basename(vp)\n\n# =====================================================\n# 5) 평가 함수 (threshold 고정)\n# =====================================================\ndef evaluate_model(model, loader, device):\n    model.eval()\n    names, preds = [], []\n    with torch.no_grad():\n        for feats, frames, ids in tqdm(loader):\n            feats, frames = feats.to(device), frames.to(device)\n            out = model(feats, frames).sigmoid().view(-1).cpu().numpy()\n            names.extend(ids)\n            preds.extend((out > THRESHOLD).astype(int).tolist())\n    return names, preds\n\n# =====================================================\n# 6) main: test_video_dir 만 외부에서 지정\n# =====================================================\ndef main(test_video_dir):\n    model = DeepFakeDetector().to(DEVICE)\n    model.load_state_dict(torch.load(\n        os.path.join(\"models/dfd_model.pt\"),\n        map_location=DEVICE\n    ))\n\n    vids   = sorted(glob.glob(os.path.join(test_video_dir, \"*.mp4\")))\n    ds     = ExternalVideoDataset(vids)\n    loader = DataLoader(ds, batch_size=BATCH_SIZE, shuffle=False, num_workers=0)\n\n    names, preds = evaluate_model(model, loader, DEVICE)\n    df = pd.DataFrame({\"ID\": names, \"label\": preds})\n    csv_path = \"result.csv\"\n    df.to_csv(csv_path, index=False)\n    print(f\"[INFO] Saved results to {csv_path}\")\n\n# 여기에 비디오(.mp4) 경로 지정하시면 됩니다\nTEST_VIDEO_DIR = \"/kaggle/input/Deepfake_Detection_and_Generation_Challenge_Blue_Team/test_sample_videos\"\nmain(TEST_VIDEO_DIR)\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-06-23T12:55:24.248812Z","iopub.execute_input":"2025-06-23T12:55:24.249064Z","iopub.status.idle":"2025-06-23T13:05:43.743272Z","shell.execute_reply.started":"2025-06-23T12:55:24.249042Z","shell.execute_reply":"2025-06-23T13:05:43.742625Z"}},"outputs":[],"execution_count":null}]}