{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.14","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":16880,"databundleVersionId":858837,"sourceType":"competition"},{"sourceId":10011479,"sourceType":"datasetVersion","datasetId":6163488}],"dockerImageVersionId":30787,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"pip install scipy<1.14","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:05:23.275845Z","iopub.execute_input":"2024-11-27T07:05:23.276599Z","iopub.status.idle":"2024-11-27T07:05:24.307838Z","shell.execute_reply.started":"2024-11-27T07:05:23.276566Z","shell.execute_reply":"2024-11-27T07:05:24.306876Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install -U albumentations","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:10:39.290745Z","iopub.execute_input":"2024-11-27T07:10:39.291437Z","iopub.status.idle":"2024-11-27T07:10:50.406082Z","shell.execute_reply.started":"2024-11-27T07:10:39.291402Z","shell.execute_reply":"2024-11-27T07:10:50.405034Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install facenet-pytorch","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:11:18.757832Z","iopub.execute_input":"2024-11-27T07:11:18.758558Z","iopub.status.idle":"2024-11-27T07:13:41.768770Z","shell.execute_reply.started":"2024-11-27T07:11:18.758523Z","shell.execute_reply":"2024-11-27T07:13:41.767767Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import json\nimport pandas as pd\n\njson_file = '/kaggle/input/deepfake-detection-challenge/train_sample_videos/metadata.json'\ndf = pd.read_json(json_file)\ndf = df.T\n\ndf.head()","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:17:50.965136Z","iopub.execute_input":"2024-11-27T07:17:50.965994Z","iopub.status.idle":"2024-11-27T07:17:52.308243Z","shell.execute_reply.started":"2024-11-27T07:17:50.965958Z","shell.execute_reply":"2024-11-27T07:17:52.307124Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import json\nimport os\n\n# 메타데이터 파일 경로 리스트\nmetadata_paths = ['/kaggle/input/deepfake-detection-challenge/train_sample_videos/metadata.json']\n\n# 메타데이터 병합\nmetadata = {}\nfor path in metadata_paths:\n    if os.path.exists(path):  # 파일 존재 여부 확인\n        with open(path, \"r\") as f:\n            data = json.load(f)\n            metadata.update(data)  # 기존 metadata에 새 데이터를 병합\n    else:\n        print(f\"Warning: Metadata file not found at {path}\")\n\n# 데이터 확인\nprint(f\"Total metadata entries: {len(metadata)}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:18:08.709461Z","iopub.execute_input":"2024-11-27T07:18:08.710104Z","iopub.status.idle":"2024-11-27T07:18:08.717591Z","shell.execute_reply.started":"2024-11-27T07:18:08.710069Z","shell.execute_reply":"2024-11-27T07:18:08.716831Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport json\nimport random\nimport pandas as pd\n\n# 데이터셋 경로 설정\nfolders = [\n    '/kaggle/input/deepfake-detection-challenge/train_sample_videos'\n]\n\n# 메타데이터 파일 경로 설정\nmetadata_path = '/kaggle/input/deepfake-detection-challenge/train_sample_videos/metadata.json'\n\n# 메타데이터 로드\nwith open(metadata_path, \"r\") as f:\n    metadata_json = json.load(f)\n\n# 모든 비디오 파일과 라벨 수집\nall_videos = []\nfor folder in folders:\n    for file_name in os.listdir(folder):\n        if file_name.endswith(\".mp4\") and file_name in metadata_json:  # .mp4 확장자 및 메타데이터 확인\n            label = metadata_json[file_name][\"label\"]  # 메타데이터에서 라벨 가져오기\n            all_videos.append({\n                \"video_name\": os.path.join(folder, file_name),  # 전체 경로 저장\n                \"label\": 0 if label == \"REAL\" else 1  # 0: REAL, 1: FAKE\n            })\n\n# Pandas DataFrame 생성\nmetadata_df = pd.DataFrame(all_videos)\n\n# REAL과 FAKE 비디오 분리\nreal_videos = metadata_df[metadata_df[\"label\"] == 0][\"video_name\"].tolist()\nfake_videos = metadata_df[metadata_df[\"label\"] == 1][\"video_name\"].tolist()\n\n# 테스트용 비디오 샘플링 (20%)\ntest_real = random.sample(real_videos, int(len(real_videos) * 0.2))\ntest_fake = random.sample(fake_videos, int(len(fake_videos) * 0.2))\n\n# 테스트용 비디오 제거 후 나머지로 훈련/검증 데이터셋 구성\nremaining_real = list(set(real_videos) - set(test_real))\nremaining_fake = list(set(fake_videos) - set(test_fake))\n\n# 훈련용 데이터와 검증용 데이터는 80%와 20%로 나누기\nval_real = random.sample(remaining_real, int(len(remaining_real) * 0.2))\nval_fake = random.sample(remaining_fake, int(len(remaining_fake) * 0.2))\n\n# 훈련용 데이터에서 검증용 데이터를 제거\ntrain_real = list(set(remaining_real) - set(val_real))\ntrain_fake = list(set(remaining_fake) - set(val_fake))\n\n# 훈련 데이터프레임 생성\ntrain_data = pd.DataFrame({\n    \"video_name\": train_real + train_fake,\n    \"label\": [0] * len(train_real) + [1] * len(train_fake)  # 0: REAL, 1: FAKE\n})\n\n# 검증 데이터프레임 생성\nval_data = pd.DataFrame({\n    \"video_name\": val_real + val_fake,\n    \"label\": [0] * len(val_real) + [1] * len(val_fake)  # 0: REAL, 1: FAKE\n})\n\n# 테스트 데이터프레임 생성\ntest_data = pd.DataFrame({\n    \"video_name\": test_real + test_fake,\n    \"label\": [0] * len(test_real) + [1] * len(test_fake)  # 0: REAL, 1: FAKE\n})\n\n# 훈련 데이터 저장\ntrain_save_path = \"/kaggle/working/train_data.csv\"\ntrain_data.to_csv(train_save_path, index=False)\n\n# 검증 데이터 저장\nval_save_path = \"/kaggle/working/validation_data.csv\"\nval_data.to_csv(val_save_path, index=False)\n\n# 테스트 데이터 저장\ntest_save_path = \"/kaggle/working/test_data.csv\"\ntest_data.to_csv(test_save_path, index=False)\n\n# 확인을 위해 저장된 파일 경로 출력\nprint(f\"Train data saved to: {train_save_path}\")\nprint(f\"Validation data saved to: {val_save_path}\")\nprint(f\"Test data saved to: {test_save_path}\")\n\n# 훈련, 검증, 테스트 데이터셋 간의 모든 경우의 수에서 겹치는 데이터 확인\ndef check_overlap(train_df, val_df, test_df):\n    # 각 데이터셋 간 교집합 확인\n    overlap_train_val = set(train_df[\"video_name\"]).intersection(set(val_df[\"video_name\"]))\n    overlap_train_test = set(train_df[\"video_name\"]).intersection(set(test_df[\"video_name\"]))\n    overlap_val_test = set(val_df[\"video_name\"]).intersection(set(test_df[\"video_name\"]))\n    overlap_all = set(train_df[\"video_name\"]).intersection(set(val_df[\"video_name\"])).intersection(set(test_df[\"video_name\"]))\n\n    # 결과 출력\n    if overlap_train_val:\n        print(f\"훈련과 검증 데이터셋이 겹친 비디오 파일들: {overlap_train_val}\")\n    else:\n        print(\"훈련과 검증 데이터셋은 겹치지 않습니다.\")\n\n    if overlap_train_test:\n        print(f\"훈련과 테스트 데이터셋이 겹친 비디오 파일들: {overlap_train_test}\")\n    else:\n        print(\"훈련과 테스트 데이터셋은 겹치지 않습니다.\")\n\n    if overlap_val_test:\n        print(f\"검증과 테스트 데이터셋이 겹친 비디오 파일들: {overlap_val_test}\")\n    else:\n        print(\"검증과 테스트 데이터셋은 겹치지 않습니다.\")\n\n    if overlap_all:\n        print(f\"훈련, 검증, 테스트 데이터셋 모두에서 겹친 비디오 파일들: {overlap_all}\")\n    else:\n        print(\"훈련, 검증, 테스트 데이터셋 모두에서 겹치는 데이터가 없습니다.\")\n\n\n# 각 데이터셋에서 라벨별 데이터 개수 출력 함수\ndef print_label_counts(dataset, name):\n    label_counts = dataset['label'].value_counts().to_dict()\n    real_count = label_counts.get(0, 0)  # REAL(0)의 개수\n    fake_count = label_counts.get(1, 0)  # FAKE(1)의 개수\n    print(f\"{name} 데이터셋 - REAL(0): {real_count}, FAKE(1): {fake_count}\")\n\n# 각 데이터셋에서 라벨별 데이터 개수 출력\nprint_label_counts(train_data, \"훈련\")\nprint_label_counts(val_data, \"검증\")\nprint_label_counts(test_data, \"테스트\")\n\n# 기존 교집합 확인 함수 호출\ncheck_overlap(train_data, val_data, test_data)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:18:20.204634Z","iopub.execute_input":"2024-11-27T07:18:20.204972Z","iopub.status.idle":"2024-11-27T07:18:20.258279Z","shell.execute_reply.started":"2024-11-27T07:18:20.204941Z","shell.execute_reply":"2024-11-27T07:18:20.257481Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(f\"Number of REAL videos: {len(real_videos)}\")\nprint(f\"Number of FAKE videos: {len(fake_videos)}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:18:23.723960Z","iopub.execute_input":"2024-11-27T07:18:23.724700Z","iopub.status.idle":"2024-11-27T07:18:23.733736Z","shell.execute_reply.started":"2024-11-27T07:18:23.724648Z","shell.execute_reply":"2024-11-27T07:18:23.732435Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport cv2\nimport torch\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision import transforms\nimport albumentations as A\nfrom facenet_pytorch import MTCNN\nimport pandas as pd\n\n# 얼굴 인식 및 전처리 함수\nclass MTCNNPreprocess:\n    def __init__(self):\n        self.detector = MTCNN(keep_all=False, device='cuda' if torch.cuda.is_available() else 'cpu')\n\n    def __call__(self, frame):\n        faces = self.detector.detect(frame)[0]  # 얼굴 좌표만 가져오기\n        if faces is None or len(faces) == 0:\n            return None  # 얼굴 검출 실패\n\n        faces_list = []\n        for face in faces:\n            x1, y1, x2, y2 = map(int, face)\n            h, w, _ = frame.shape\n            x1, y1 = max(0, x1), max(0, y1)\n            x2, y2 = min(w, x2), min(h, y2)  # 이미지 크기 초과 방지\n            face_region = frame[y1:y2, x1:x2]\n\n            if face_region.size == 0:  # face가 비어있으면 None 반환\n                continue\n\n            faces_list.append(face_region)\n\n        return faces_list  # 여러 얼굴을 리스트로 반환\n\n# 데이터 증강 (albumentations)\nclass DataAugmentation:\n    def __init__(self, target_size=(224, 224)):\n        self.target_size = target_size\n        self.transform = A.Compose([\n            A.HorizontalFlip(),\n            A.RandomBrightnessContrast(),\n            A.Rotate(limit=45),\n            A.RandomScale(scale_limit=0.1),\n        ])\n\n    def __call__(self, frame):\n        augmented = self.transform(image=frame)[\"image\"]  # 증강\n        resized = cv2.resize(augmented, self.target_size)  # 리사이즈\n        return resized\n\nclass VideoFrameDataset(Dataset):\n    def __init__(self, video_dirs, df, preprocess, albumentations_transform=None, num_frames=10, target_size=(224, 224), augment=False):\n        self.video_dirs = video_dirs\n        self.df = df\n        self.preprocess = preprocess\n        self.albumentations_transform = albumentations_transform if augment else None  # 데이터 증강이 필요한 경우만 적용\n        self.num_frames = num_frames\n        self.target_size = target_size  # 목표 크기 설정\n        self.video_files = df['video_name'].tolist()\n        self.frames_and_labels = self._load_frames_and_labels()\n\n    def _find_video_path(self, video_file):\n        for video_dir in self.video_dirs:\n            video_path = os.path.join(video_dir, video_file)\n            if os.path.exists(video_path):\n                return video_path\n        return None\n\n    def _load_frames_and_labels(self):\n        frames_and_labels = []\n        for video_file in self.video_files:\n            video_path = self._find_video_path(video_file)\n            if video_path is None:\n                print(f\"Warning: {video_file} does not exist in specified directories. Skipping.\")\n                continue\n\n            cap = cv2.VideoCapture(video_path)\n            total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n            frame_interval = max(total_frames // self.num_frames, 1)\n            valid_frames = []\n\n            for i in range(self.num_frames):\n                cap.set(cv2.CAP_PROP_POS_FRAMES, i * frame_interval)\n                success, frame = cap.read()\n                if not success:\n                    continue\n\n                faces = self.preprocess(frame)\n                if faces is None:  # 얼굴 검출 실패\n                    continue\n\n                for face in faces:\n                    # 라벨에 따라서만 데이터 증강을 적용\n                    label_name = self.df.loc[self.df['video_name'] == video_file, 'label'].values[0]\n                    label = 1 if label_name == 'FAKE' else 0\n\n                    # 진짜 데이터에만 증강 적용 (훈련 데이터셋에서만)\n                    if self.albumentations_transform is not None and label == 0:  # REAL 데이터일 때만 증강\n                        face = self.albumentations_transform(face)  # 데이터 증강\n\n                    # 리사이징을 여기서 적용\n                    face_resized = cv2.resize(face, self.target_size)  # 목표 크기로 리사이징\n                    valid_frames.append((face_resized, label))\n\n            cap.release()\n\n            if valid_frames:\n                frames_and_labels.extend(valid_frames)\n\n        return frames_and_labels\n\n    def __len__(self):\n        return len(self.frames_and_labels)\n\n    def __getitem__(self, idx):\n        frame, label = self.frames_and_labels[idx]\n        frame_tensor = transforms.ToTensor()(frame)  # 이미지를 텐서로 변환\n        return frame_tensor, label\n\n\npreprocess = MTCNNPreprocess()\nalbumentations_transform = DataAugmentation()\n\n# 데이터셋 설정\n# 훈련 데이터셋에만 데이터 증강 적용\ntrain_dataset = VideoFrameDataset(folders, train_data, preprocess, albumentations_transform, augment=True)\nval_dataset = VideoFrameDataset(folders, val_data, preprocess, albumentations_transform, augment=False)\ntest_dataset = VideoFrameDataset(folders, test_data, preprocess, albumentations_transform, augment=False)\n\n# DataLoader 설정\ntrain_loader = DataLoader(train_dataset, batch_size=16, shuffle=True)\nval_loader = DataLoader(val_dataset, batch_size=16, shuffle=False)\ntest_loader = DataLoader(test_dataset, batch_size=16, shuffle=False)\n\n# 데이터 로더가 잘 준비되었는지 확인\nprint(f\"훈련 데이터로더 샘플 크기: {len(train_loader)}\")\nprint(f\"검증 데이터로더 샘플 크기: {len(val_loader)}\")\nprint(f\"테스트 데이터로더 샘플 크기: {len(test_loader)}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:18:37.279886Z","iopub.execute_input":"2024-11-27T07:18:37.280204Z","iopub.status.idle":"2024-11-27T07:52:02.750179Z","shell.execute_reply.started":"2024-11-27T07:18:37.280177Z","shell.execute_reply":"2024-11-27T07:52:02.749071Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport numpy as np\n\ndef visualize_preprocessed_data(dataset, num_samples=5):\n    \"\"\"\n    Dataset에서 샘플 이미지를 랜덤으로 선택해 전처리 결과를 시각화합니다.\n\n    Args:\n        dataset (Dataset): Dataset 객체\n        num_samples (int): 시각화할 샘플 이미지 수\n    \"\"\"\n    num_samples = min(num_samples, len(dataset))  # 데이터셋 크기를 초과하지 않도록 제한\n    indices = np.random.choice(len(dataset), num_samples, replace=False)\n\n    plt.figure(figsize=(15, 5))\n    for i, idx in enumerate(indices):\n        frame, label = dataset[idx]\n        frame = frame.permute(1, 2, 0).numpy()  # Tensor에서 이미지 형식으로 변환 (C, H, W → H, W, C)\n        frame = (frame * 255).astype(np.uint8)  # Normalize를 복원하여 0-255로 변환\n        label_text = \"FAKE\" if label == 1 else \"REAL\"\n\n        plt.subplot(1, num_samples, i + 1)\n        plt.imshow(frame)\n        plt.title(f\"Label: {label_text}\")\n        plt.axis(\"off\")\n    plt.tight_layout()\n    plt.show()\n\n# 전처리 결과 시각화\nvisualize_preprocessed_data(train_dataset, num_samples=5)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:15.216773Z","iopub.execute_input":"2024-11-27T07:53:15.217357Z","iopub.status.idle":"2024-11-27T07:53:15.951163Z","shell.execute_reply.started":"2024-11-27T07:53:15.217322Z","shell.execute_reply":"2024-11-27T07:53:15.950343Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"batch_size = 16\ntrain_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=1)\nval_pre_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=1)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:19.678412Z","iopub.execute_input":"2024-11-27T07:53:19.679097Z","iopub.status.idle":"2024-11-27T07:53:19.683162Z","shell.execute_reply.started":"2024-11-27T07:53:19.679062Z","shell.execute_reply":"2024-11-27T07:53:19.682341Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# train_loader에서 첫 번째 배치 데이터 확인\nfor batch_idx, (data, labels) in enumerate(val_pre_loader):\n    print(f\"Batch {batch_idx}: Data Shape: {data.shape}, Labels: {labels}\")\n    if batch_idx == 5:  # 첫 번째 배치에서만 확인\n        break\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:22.437622Z","iopub.execute_input":"2024-11-27T07:53:22.438330Z","iopub.status.idle":"2024-11-27T07:53:22.666770Z","shell.execute_reply.started":"2024-11-27T07:53:22.438300Z","shell.execute_reply":"2024-11-27T07:53:22.665697Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"'''\ndata_iter = iter(val_loader)\ninputs, labels = next(data_iter)\nprint(\"Inputs:\", inputs)\nprint(\"Labels:\", labels)\n'''","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T02:54:47.445923Z","iopub.execute_input":"2024-11-27T02:54:47.446268Z","iopub.status.idle":"2024-11-27T02:54:47.485105Z","shell.execute_reply.started":"2024-11-27T02:54:47.446238Z","shell.execute_reply":"2024-11-27T02:54:47.484256Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install torchsummary","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:33.735940Z","iopub.execute_input":"2024-11-27T07:53:33.736343Z","iopub.status.idle":"2024-11-27T07:53:42.000168Z","shell.execute_reply.started":"2024-11-27T07:53:33.736290Z","shell.execute_reply":"2024-11-27T07:53:41.999070Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install efficientnet-pytorch","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:42.002370Z","iopub.execute_input":"2024-11-27T07:53:42.002789Z","iopub.status.idle":"2024-11-27T07:53:52.173062Z","shell.execute_reply.started":"2024-11-27T07:53:42.002747Z","shell.execute_reply":"2024-11-27T07:53:52.171980Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport timm  # ConvNeXt 모델 로드용 timm 라이브러리\nfrom efficientnet_pytorch import EfficientNet\n\n# EfficientNet-B0 모델\nclass EfficientNetB0(nn.Module):\n    def __init__(self, num_classes=1, dropout_prob=0.3):\n        super(EfficientNetB0, self).__init__()\n        self.model = EfficientNet.from_pretrained('efficientnet-b0')\n        self.model._fc = nn.Sequential(\n            nn.Dropout(dropout_prob),\n            nn.Linear(self.model._fc.in_features, num_classes)\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\n# ConvNeXt Tiny 모델\nclass ConvNeXtTiny(nn.Module):\n    def __init__(self, num_classes=1, dropout_prob=0.3):\n        super(ConvNeXtTiny, self).__init__()\n        self.model = timm.create_model('convnext_tiny', pretrained=True)\n        in_features = self.model.get_classifier().in_features  # Classifier 입력 차원 확인\n        self.model.classifier = nn.Sequential(\n            nn.Dropout(dropout_prob),\n            nn.Linear(in_features, num_classes)\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\n\n# 앙상블 모델 (Dropout 추가)\nclass EnsembleModel(nn.Module):\n    def __init__(self, model1, model2, dropout_prob=0.2):\n        super(EnsembleModel, self).__init__()\n        self.model1 = model1\n        self.model2 = model2\n        self.dropout = nn.Dropout(dropout_prob)  # Dropout 레이어 추가\n\n    def forward(self, x):\n        # 두 모델의 예측값을 얻음\n        output1 = self.model1(x)\n        output2 = self.model2(x)\n\n        # 두 출력값의 평균 계산\n        output = (output1 + output2) / 2\n\n        # Dropout 적용\n        output = self.dropout(output)  # Dropout을 통해 일부 뉴런을 무작위로 끔\n\n        # Sigmoid 활성화 적용\n        output = torch.sigmoid(output)\n\n        # 평균을 내서 배치 단위로 차원을 맞춤 (배치 크기, 1)\n        output = output.mean(dim=1, keepdim=True)  # 각 배치의 평균 값 사용\n\n        return output\n\n# 모델 인스턴스 생성\neffnet_model = EfficientNetB0(num_classes=1, dropout_prob=0.3)\nconvnext_model = ConvNeXtTiny(num_classes=1, dropout_prob=0.3)\nensemble_model = EnsembleModel(effnet_model, convnext_model, dropout_prob=0.2)\n\n# 모델을 GPU로 이동\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nensemble_model = ensemble_model.to(device)\n\n# 모델 요약 출력\nfrom torchsummary import summary\nsummary(ensemble_model, input_size=(3, 224, 224), device='cuda' if torch.cuda.is_available() else 'cpu')\n\n# 학습용 DataLoader 예시\nfrom torch.utils.data import DataLoader, TensorDataset\n\n# 예제용 데이터셋 (임의의 데이터 생성)\ntrain_dataset = TensorDataset(\n    torch.randn(100, 3, 224, 224),  # 100개의 224x224 RGB 이미지\n    torch.randint(0, 2, (100, 1)).float()  # 이진 레이블 (0 또는 1)\n)\nbatch_size = 32  # 배치 크기 설정\ntrain_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4)\n\n# 배치 데이터 확인\nfor frames, labels in train_loader:\n    print(f\"배치 프레임의 크기: {frames.shape}\")  # 예상: [32, 3, 224, 224]\n    print(f\"배치 레이블의 크기: {labels.shape}\")  # 예상: [32, 1]\n    break\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:53:56.854336Z","iopub.execute_input":"2024-11-27T07:53:56.855181Z","iopub.status.idle":"2024-11-27T07:54:00.714070Z","shell.execute_reply.started":"2024-11-27T07:53:56.855144Z","shell.execute_reply":"2024-11-27T07:54:00.712906Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch.nn as nn\nimport torch.optim as optim\n\n# 모델 정의 (예: 앙상블 모델 사용)\nmodel = ensemble_model  # 이미 정의된 ensemble_model을 사용\n\n# BCEWithLogitsLoss를 사용\ncriterion = nn.BCEWithLogitsLoss()  # BCEWithLogitsLoss로 변경\n\n# 옵티마이저 설정: Adam + L2 정규화\noptimizer = optim.Adam(model.parameters(), lr=1e-4, weight_decay=1e-5)  # L2 정규화 추가\n\n# 손실 함수와 옵티마이저 확인\nprint(\"Loss Function:\", criterion)\nprint(\"Optimizer:\", optimizer)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:54:08.279213Z","iopub.execute_input":"2024-11-27T07:54:08.279572Z","iopub.status.idle":"2024-11-27T07:54:08.289206Z","shell.execute_reply.started":"2024-11-27T07:54:08.279534Z","shell.execute_reply":"2024-11-27T07:54:08.288357Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn.functional as F\nfrom torchvision import transforms\nimport matplotlib.pyplot as plt\n\n# Random Erasing 데이터 증강 정의\nrandom_erasing = transforms.RandomErasing(p=0.5, scale=(0.02, 0.2), ratio=(0.3, 3.3), value=0)\n\n# 훈련 루프에서 데이터 증강을 적용하는 함수\ndef train_one_epoch(model, dataloader, criterion, optimizer, device, log_interval=100):\n    model.train()  # 모델을 학습 모드로 설정\n    running_loss = 0.0  # 누적 손실 초기화\n    correct = 0  # 맞은 예측 수 초기화\n    total = 0  # 총 예측 수 초기화\n\n    # 데이터로더에서 배치별로 데이터를 가져옴\n    for i, (inputs, labels) in enumerate(dataloader):\n        inputs = inputs.to(device)  # 입력 데이터를 장치(GPU 또는 CPU)로 이동\n        labels = labels.to(device).unsqueeze(1).float()  # 라벨을 (batch_size, 2) 형식에 맞춰 변환\n\n        # 첫 번째 배치의 몇 개의 샘플을 시각화하여 증강 확인\n        if i == 0:  # 첫 번째 배치에만 샘플을 출력\n            print(\"Random Erasing이 적용된 샘플들을 확인합니다:\")\n            for k in range(min(5, inputs.size(0))):  # 최대 5개의 샘플을 출력\n                plt.imshow(inputs[k].cpu().permute(1, 2, 0).numpy())  # 채널 순서를 변경하여 이미지 출력\n                plt.show()  # 이미지를 화면에 표시\n\n        # 입력 이미지에 랜덤 지우기 증강 적용\n        for j in range(inputs.size(0)):\n            inputs[j] = random_erasing(inputs[j])  # 각 이미지를 랜덤 지우기 증강 적용\n\n        optimizer.zero_grad()  # 옵티마이저의 경사도 초기화\n        outputs = model(inputs)  # 모델을 통해 예측 수행\n        loss = criterion(outputs, labels)  # 손실 계산\n        loss.backward()  # 역전파를 통해 경사도 계산\n        optimizer.step()  # 옵티마이저로 가중치 업데이트\n\n        # 예측 결과를 이진 분류에 맞게 0과 1로 변환\n        outputs[outputs >= 0.5] = 1\n        outputs[outputs < 0.5] = 0\n\n        # 누적 손실 계산\n        running_loss += loss.item() * inputs.size(0)\n        # 맞춘 예측 수를 합산\n        correct += outputs.eq(labels).int().sum()\n\n    # 평균 손실과 정확도를 반환\n    return running_loss / len(dataloader), 100 * correct / len(dataloader.dataset)\n\n# 평가 함수는 Mixup이 적용되지 않으므로 기존 코드와 동일하게 유지\ndef evaluate(model, dataloader, criterion, device):\n    model.eval()  # 모델을 평가 모드로 설정\n    running_loss = 0.0  # 누적 손실 초기화\n    correct = 0  # 맞은 예측 수 초기화\n    total = 0  # 총 예측 수 초기화\n\n    with torch.no_grad():  # 평가에서는 경사도를 계산하지 않음\n        for inputs, labels in dataloader:\n            inputs, labels = inputs.to(device), labels.to(device)\n            outputs = model(inputs)\n            labels = labels.unsqueeze(1).float()\n            loss = criterion(outputs, labels)\n\n            # 예측 결과를 이진 분류에 맞게 0과 1로 변환\n            outputs[outputs >= 0.5] = 1\n            outputs[outputs < 0.5] = 0\n\n            # 누적 손실 계산\n            running_loss += loss.item() * inputs.size(0)\n            # 맞춘 예측 수를 합산\n            correct += outputs.eq(labels).int().sum()\n\n    # 평균 손실과 정확도를 반환\n    return running_loss / len(dataloader), 100 * correct / len(dataloader.dataset)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:54:11.879970Z","iopub.execute_input":"2024-11-27T07:54:11.880311Z","iopub.status.idle":"2024-11-27T07:54:11.891843Z","shell.execute_reply.started":"2024-11-27T07:54:11.880280Z","shell.execute_reply":"2024-11-27T07:54:11.891002Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import copy\nimport torch\nfrom torch.utils.data import DataLoader\nimport torch.nn as nn\nimport torch.optim as optim\nimport numpy as np\nimport pandas as pd\n\n# 모델을 GPU로 이동\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nensemble_model = ensemble_model.to(device)  # 이미 정의된 앙상블 모델\n\n# 손실 함수 및 옵티마이저 설정\ncriterion = nn.BCEWithLogitsLoss()\noptimizer = optim.Adam(ensemble_model.parameters(), lr=1e-4, weight_decay=1e-3)\n\n\n# Train과 Validation DataLoader 설정\nbatch_size = 32  # 배치 크기 설정\ntrain_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4)  # num_workers는 데이터 로딩 속도를 최적화\nval_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=4)\n\n# Test DataLoader 설정\ntest_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=4)\n\n# Train 함수 정의\ndef train_one_epoch(model, dataloader, criterion, optimizer, device, log_interval=100):\n    model.train()  # 모델을 학습 모드로 설정\n    running_loss = 0.0\n    correct = 0\n    total = 0\n\n    for batch_idx, (inputs, labels) in enumerate(dataloader):\n        # 데이터도 GPU로 이동\n        inputs, labels = inputs.to(device), labels.to(device)\n        labels = labels.unsqueeze(1).float()  # 이진 분류이므로 labels의 형태를 (batch_size, 1)로 맞춤\n\n        optimizer.zero_grad()\n\n        # 레이블 차원 맞추기 (필요 시)\n        labels = labels.squeeze(-1)\n\n        # 모델에 입력 전달\n        outputs = model(inputs)\n        loss = criterion(outputs, labels)  # 손실 계산\n        loss.backward()\n        optimizer.step()\n\n        # 손실 및 정확도 계산\n        running_loss += loss.item() * inputs.size(0)\n        preds = (outputs >= 0.5).float()  # 시그모이드 적용 후 예측값을 0 또는 1로 변환\n        correct += (preds == labels).sum().item()\n        total += labels.size(0)\n\n        # 로그 간격마다 출력\n        if (batch_idx + 1) % log_interval == 0:\n            print(f'Batch [{batch_idx + 1}/{len(dataloader)}], Loss: {loss.item():.4f}')\n\n    epoch_loss = running_loss / len(dataloader.dataset)\n    epoch_accuracy = 100 * correct / total\n    return epoch_loss, epoch_accuracy\n\n# 검증 함수 정의\ndef evaluate(model, dataloader, device):\n    model.eval()  # 모델을 평가 모드로 설정\n    all_predictions = []\n    all_true_labels = []\n\n    with torch.no_grad():\n        for batch_idx, (inputs, labels) in enumerate(dataloader):\n            # 데이터 GPU로 이동\n            inputs, labels = inputs.to(device), labels.to(device)\n            outputs = model(inputs)\n            preds = (outputs >= 0.5).float().cpu().numpy()  # 시그모이드 적용 후 예측값을 0 또는 1로 변환\n\n            # 데이터를 리스트에 저장\n            all_predictions.extend(preds.flatten())\n            all_true_labels.extend(labels.cpu().numpy())\n\n            # 배치별로 DataFrame으로 출력\n            df = pd.DataFrame({\n                \"True Labels\": labels.cpu().numpy(),\n                \"Predictions\": preds.flatten()\n            })\n            #print(f\"Batch {batch_idx + 1} Results:\\n{df.to_string(index=False)}\\n\")\n\n    return all_true_labels, all_predictions\n\n\n# 학습 및 검증 설정\nnum_epochs = 20\nbest_val_acc = 0.0\nbest_model_wts = copy.deepcopy(ensemble_model.state_dict())  # 초기 모델 가중치 저장\n\nfor epoch in range(num_epochs):\n    print(f\"Epoch {epoch + 1}/{num_epochs}\")\n\n    # Train 데이터셋 학습 (라벨이 있음)\n    train_loss, train_acc = train_one_epoch(ensemble_model, train_loader, criterion, optimizer, device)\n    print(f\"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}\")\n\n    # Validation 데이터셋 평가 (라벨 있음)\n    val_true_labels, val_predictions = evaluate(ensemble_model, val_loader, device)\n\n    # 정확도 계산 (0 또는 1로 변환 후 백분율 계산)\n    val_true_labels = np.array(val_true_labels).astype(int)  # true_labels를 int로 변환\n    val_predictions = np.array(val_predictions).astype(int)  # predictions을 int로 변환\n\n    # 정확도 계산\n    val_acc = np.mean(val_true_labels == val_predictions) * 100  # 비교 후 백분율로 변환\n    print(f\"Validation Accuracy: {val_acc:.4f}%\")\n\n    # 모델의 성능 개선 여부 확인\n    if val_acc > best_val_acc:\n        best_val_acc = val_acc\n        best_model_wts = copy.deepcopy(ensemble_model.state_dict())  # 가장 성능 좋은 가중치 저장\n\n# 가장 성능 좋은 가중치로 모델 업데이트\nensemble_model.load_state_dict(best_model_wts)\n\n# 테스트 함수 정의\ndef test_model(model, dataloader, device):\n    model.eval()  # 모델을 평가 모드로 설정\n    predictions = []\n    true_labels = []\n\n    with torch.no_grad():\n        for batch in dataloader:\n            # 데이터 GPU로 이동\n            inputs, labels = batch\n            inputs = inputs.to(device)\n            labels = labels.to(device)\n            outputs = model(inputs)\n            preds = (outputs >= 0.5).float().cpu().numpy()  # 시그모이드 적용 후 예측값을 0 또는 1로 변환\n\n            predictions.extend(preds)\n            true_labels.extend(labels.cpu().numpy())\n\n    return true_labels, predictions\n\n# 테스트 데이터셋 평가 (라벨 있음)\ntest_true_labels, test_predictions = test_model(ensemble_model, test_loader, device)\n\n# 정확도 계산 (0 또는 1로 변환 후 백분율 계산)\ntest_true_labels = np.array(test_true_labels).astype(int)  # true_labels를 int로 변환\ntest_predictions = np.array(test_predictions).astype(int)  # predictions을 int로 변환\n\n# 정확도 계산\ntest_acc = np.mean(test_true_labels == test_predictions) * 100  # 비교 후 백분율로 변환\nprint(f\"Test Accuracy: {test_acc:.4f}%\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-11-27T07:55:55.856072Z","iopub.execute_input":"2024-11-27T07:55:55.856461Z","iopub.status.idle":"2024-11-27T07:58:39.607318Z","shell.execute_reply.started":"2024-11-27T07:55:55.856430Z","shell.execute_reply":"2024-11-27T07:58:39.606156Z"},"_kg_hide-output":true,"_kg_hide-input":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}