{"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":10066714,"sourceType":"datasetVersion","datasetId":6194564}],"dockerImageVersionId":30787,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"pip install torch torchvision facenet-pytorch opencv-python albumentations pandas","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:47:27.567616Z","iopub.execute_input":"2024-12-02T13:47:27.567974Z","iopub.status.idle":"2024-12-02T13:49:43.849874Z","shell.execute_reply.started":"2024-12-02T13:47:27.567935Z","shell.execute_reply":"2024-12-02T13:49:43.848880Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install torchsummary","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:49:43.851419Z","iopub.execute_input":"2024-12-02T13:49:43.851716Z","iopub.status.idle":"2024-12-02T13:49:52.475701Z","shell.execute_reply.started":"2024-12-02T13:49:43.851689Z","shell.execute_reply":"2024-12-02T13:49:52.474539Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"pip install expecttest","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:49:52.477212Z","iopub.execute_input":"2024-12-02T13:49:52.477623Z","iopub.status.idle":"2024-12-02T13:50:00.670849Z","shell.execute_reply.started":"2024-12-02T13:49:52.477582Z","shell.execute_reply":"2024-12-02T13:50:00.669723Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import json\nimport pandas as pd\n\njson_file = '/kaggle/input/traindataset0/dfdc_train_part_0/dfdc_train_part_0/metadata.json'\ndf = pd.read_json(json_file)\ndf = df.T\n\ndf.head()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:50:00.673630Z","iopub.execute_input":"2024-12-02T13:50:00.673926Z","iopub.status.idle":"2024-12-02T13:50:01.385878Z","shell.execute_reply.started":"2024-12-02T13:50:00.673898Z","shell.execute_reply":"2024-12-02T13:50:01.385022Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import json\nimport os\n\n# 메타데이터 파일 경로 리스트\nmetadata_paths = [\n    '/kaggle/input/traindataset0/dfdc_train_part_0/dfdc_train_part_0/metadata.json', \n    '/kaggle/input/traindataset0/dfdc_train_part_01/dfdc_train_part_1/metadata.json'\n]\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-12-02T13:50:01.387172Z","iopub.execute_input":"2024-12-02T13:50:01.387850Z","iopub.status.idle":"2024-12-02T13:50:01.407357Z","shell.execute_reply.started":"2024-12-02T13:50:01.387805Z","shell.execute_reply":"2024-12-02T13:50:01.406606Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport json\nimport random\nimport pandas as pd\n\n# 데이터셋 경로 설정\nfolders = [\n    '/kaggle/input/traindataset0/dfdc_train_part_0/dfdc_train_part_0'\n    #, '/kaggle/input/traindataset0/dfdc_train_part_01/dfdc_train_part_1'\n]\n\n# 메타데이터 파일 경로 설정\nmetadata_paths = ['/kaggle/input/traindataset0/dfdc_train_part_0/dfdc_train_part_0/metadata.json']\n                  #, '/kaggle/input/traindataset0/dfdc_train_part_01/dfdc_train_part_1/metadata.json']\n\n# 모든 메타데이터 로드\nmetadata_json = {}  # 초기화\nfor metadata_path in metadata_paths:\n    with open(metadata_path, \"r\") as f:\n        metadata_json.update(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, min(len(fake_videos), int(len(test_real) * 2)))  # REAL의 2배로 샘플링\n\n# 테스트용 비디오 제거 후 나머지로 훈련 데이터셋 구성\ntrain_real = list(set(real_videos) - set(test_real))\ntrain_fake = list(set(fake_videos) - set(test_fake))\ntrain_fake = random.sample(train_fake, min(len(train_fake), len(train_real) * 2))  # REAL의 2배\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# 테스트 데이터프레임 생성\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# 테스트 데이터 저장\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\"Test data saved to: {test_save_path}\")\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(test_data, \"테스트\")\n\n# 교집합 확인 함수\ndef check_overlap(train_df, test_df):\n    overlap = set(train_df[\"video_name\"]).intersection(set(test_df[\"video_name\"]))\n    if overlap:\n        print(f\"훈련과 테스트 데이터셋이 겹친 비디오 파일들: {overlap}\")\n    else:\n        print(\"훈련과 테스트 데이터셋은 겹치지 않습니다.\")\n\ncheck_overlap(train_data, test_data)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:55:06.428320Z","iopub.execute_input":"2024-12-02T13:55:06.428877Z","iopub.status.idle":"2024-12-02T13:55:06.489532Z","shell.execute_reply.started":"2024-12-02T13:55:06.428810Z","shell.execute_reply":"2024-12-02T13:55:06.488526Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport cv2\nimport torch\nfrom torch.utils.data import Dataset, DataLoader, WeightedRandomSampler\nfrom torchvision import transforms\nimport albumentations as A \nfrom facenet_pytorch import MTCNN\nimport pandas as pd\nfrom collections import Counter\n\nimport warnings\nwarnings.filterwarnings(\"ignore\", category=FutureWarning)\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# 데이터 증강 클래스\nclass DataAugmentation:\n    def __init__(self, target_size=(224, 224)):\n        self.target_size = target_size\n        self.transform = A.Compose([\n            A.HorizontalFlip(p=0.5),  # 랜덤 수평 뒤집기\n            A.VerticalFlip(p=0.2),   # 랜덤 수직 뒤집기\n            A.RandomBrightnessContrast(p=0.5),  # 밝기 및 대비 조정\n            A.Rotate(limit=45, p=0.5),  # 랜덤 회전 (최대 45도)\n            A.ShiftScaleRotate(shift_limit=0.1, scale_limit=0.1, rotate_limit=15, p=0.5),  # 이동, 스케일, 회전\n            A.RandomGamma(p=0.2),  # 감마 변환\n            A.GaussNoise(p=0.3),   # 가우시안 노이즈 추가\n            A.MotionBlur(blur_limit=5, p=0.3),  # 모션 블러 추가\n            A.ColorJitter(p=0.2),  # 색상 변화\n            A.CLAHE(clip_limit=2.0, p=0.3),  # 히스토그램 평활화\n            A.ElasticTransform(alpha=120, sigma=120 * 0.05, p=0.2),  # 탄성 변형, alpha_affine 제거\n            A.Blur(blur_limit=3, p=0.1),  # 블러\n            A.Sharpen(alpha=(0.2, 0.5), lightness=(0.5, 1.0), p=0.3),  # 선명화\n            A.RandomResizedCrop(height=224, width=224, scale=(0.8, 1.0), ratio=(0.75, 1.33), p=0.3),  # 랜덤 크롭 후 리사이즈\n            A.CoarseDropout(max_holes=8, max_height=16, max_width=16, min_holes=1, min_height=4, min_width=4, \n                            fill_value=0, p=0.5)  # 랜덤 마스킹\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\n# 비디오 데이터셋 클래스\nclass VideoFrameDataset(Dataset):\n    def __init__(self, video_dir, df, preprocess, albumentations_transform=None, num_frames=10, target_size=(224, 224), augment=False):\n        self.video_dir = video_dir\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.frames_and_labels = self._load_frames_and_labels()\n\n    def _load_frames_and_labels(self):\n        frames_and_labels = []\n        for _, row in self.df.iterrows():\n            video_file = row['video_name']\n            label = row.get('label', -1)  # 테스트 데이터는 라벨 없이 처리\n\n            video_path = os.path.join(self.video_dir, video_file)\n            if not os.path.exists(video_path):\n                print(f\"Warning: {video_file} does not exist. 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                    if self.albumentations_transform is not None:\n                        face = self.albumentations_transform(face)\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# 오버샘플링을 위한 샘플링 가중치 생성 함수\ndef create_balanced_sampler(dataset):\n    labels = [label for _, label in dataset.frames_and_labels]  # 모든 레이블 추출\n    label_counts = Counter(labels)  # 레이블별 개수\n\n    # 각 클래스의 목표 샘플 수를 최대 클래스 수로 맞춤\n    max_count = max(label_counts.values())  # 가장 많은 클래스 샘플 수\n\n    # 클래스별 가중치 계산 (1:1 비율로 샘플링)\n    class_weights = {label: max_count / count for label, count in label_counts.items()}\n    sample_weights = [class_weights[label] for label in labels]\n\n    # WeightedRandomSampler 생성\n    sampler = WeightedRandomSampler(sample_weights, num_samples=max_count * len(label_counts), replacement=True)\n    return sampler\n\n\n\n# CSV 파일 로드\ntrain_data = pd.read_csv('/kaggle/working/train_data.csv')\ntest_data = pd.read_csv('/kaggle/working/test_data.csv')\n\n# 데이터셋 및 로더 생성\npreprocess = MTCNNPreprocess()\nalbumentations_transform = DataAugmentation()\n\n# 훈련 데이터셋 (모든 데이터 증강 포함)\ntrain_dataset = VideoFrameDataset(\n    video_dir='/kaggle/input/traindataset0',  # 콤마 추가\n    df=train_data,\n    preprocess=preprocess,\n    albumentations_transform=albumentations_transform,\n    augment=True\n)\n\n\n# 테스트 데이터셋 (증강 미포함)\ntest_dataset = VideoFrameDataset(\n    video_dir= '/kaggle/input/deepfake-detection-challenge/test_videos',\n    df=test_data,\n    preprocess=preprocess,\n    albumentations_transform=None,\n    augment=False\n)\n\n# 균형 샘플러 생성\ntrain_sampler = create_balanced_sampler(train_dataset)\n\n# DataLoader 생성\ntrain_loader = DataLoader(train_dataset, batch_size=16, sampler=train_sampler)\ntest_loader = DataLoader(test_dataset, batch_size=16, shuffle=False)\n\n# 데이터 확인\nprint(f\"훈련 데이터로더 샘플 크기: {len(train_loader)}\")\nprint(f\"테스트 데이터로더 샘플 크기: {len(test_loader)}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:55:08.435758Z","iopub.execute_input":"2024-12-02T13:55:08.436058Z","iopub.status.idle":"2024-12-02T14:18:58.382879Z","shell.execute_reply.started":"2024-12-02T13:55:08.436033Z","shell.execute_reply":"2024-12-02T14:18:58.381955Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for inputs, labels in train_loader:\n    print(inputs.shape)  # 배치 크기 확인\n    break","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:44.183071Z","iopub.execute_input":"2024-12-02T14:25:44.183930Z","iopub.status.idle":"2024-12-02T14:25:44.205401Z","shell.execute_reply.started":"2024-12-02T14:25:44.183895Z","shell.execute_reply":"2024-12-02T14:25:44.204422Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"for inputs, labels in train_loader:\n    print(labels.shape)  # 레이블의 차원\n    print(labels[:10])  # 레이블 일부 출력\n    break\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:44.380536Z","iopub.execute_input":"2024-12-02T14:25:44.381283Z","iopub.status.idle":"2024-12-02T14:25:44.396792Z","shell.execute_reply.started":"2024-12-02T14:25:44.381254Z","shell.execute_reply":"2024-12-02T14:25:44.395878Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 데이터 증강을 테스트\nfrom matplotlib import pyplot as plt\nfor inputs, labels in train_loader:\n    plt.imshow(inputs[0].permute(1, 2, 0))  # 첫 번째 이미지를 출력\n    plt.show()\n    break\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:44.540799Z","iopub.execute_input":"2024-12-02T14:25:44.541334Z","iopub.status.idle":"2024-12-02T14:25:44.807301Z","shell.execute_reply.started":"2024-12-02T14:25:44.541303Z","shell.execute_reply":"2024-12-02T14:25:44.806496Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 데이터 로더에서 첫 번째 배치를 확인\nfor inputs, labels in train_loader:\n    print(inputs.shape, labels.shape)  # 입력 데이터와 레이블의 크기 확인\n    break\n    ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:44.809064Z","iopub.execute_input":"2024-12-02T14:25:44.809682Z","iopub.status.idle":"2024-12-02T14:25:44.826468Z","shell.execute_reply.started":"2024-12-02T14:25:44.809642Z","shell.execute_reply":"2024-12-02T14:25:44.825419Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"print(type(inputs))  # torch.Tensor여야 합니다.","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:45.649299Z","iopub.execute_input":"2024-12-02T14:25:45.650092Z","iopub.status.idle":"2024-12-02T14:25:45.654163Z","shell.execute_reply.started":"2024-12-02T14:25:45.650059Z","shell.execute_reply":"2024-12-02T14:25:45.653304Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# 클래스별 샘플 수 확인\nlabels = [label for _, label in train_dataset.frames_and_labels]\nlabel_counts = Counter(labels)\n\n# 클래스별 가중치와 샘플 수 확인\nclass_weights = {label: len(labels) / count for label, count in label_counts.items()}\nprint(f\"클래스별 샘플 수: {label_counts}\")\nprint(f\"클래스별 샘플링 가중치: {class_weights}\")\n\n# 오버샘플링 정도 확인\ntotal_samples = len(labels)\nover_sampled_counts = {label: int(class_weights[label] * label_counts[label]) for label in label_counts}\nprint(f\"각 클래스에 대해 오버샘플링 된 샘플 수: {over_sampled_counts}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:54.799333Z","iopub.execute_input":"2024-12-02T14:25:54.800177Z","iopub.status.idle":"2024-12-02T14:25:54.806514Z","shell.execute_reply.started":"2024-12-02T14:25:54.800141Z","shell.execute_reply":"2024-12-02T14:25:54.805629Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\n# 하나의 배치 가져오기\ndef show_batch_images_and_labels(loader, num_images=5):\n    # 첫 번째 배치에서 이미지와 레이블 가져오기\n    data_iter = iter(loader)\n    images, labels = next(data_iter)\n\n    # 이미지 몇 개를 시각화하기\n    fig, axes = plt.subplots(1, num_images, figsize=(15, 15))\n    for i in range(num_images):\n        ax = axes[i]\n        img = images[i].permute(1, 2, 0).numpy()  # Tensor에서 numpy 배열로 변환 (C x H x W -> H x W x C)\n        ax.imshow(img)\n        ax.axis('off')\n        ax.set_title(f\"Label: {labels[i].item()}\")\n\n    # 라벨 값 출력\n    print(\"배치 라벨 값:\", labels[:num_images].tolist())\n\n    plt.show()\n\n# 훈련 데이터의 첫 번째 배치 이미지와 라벨 값 시각화\nshow_batch_images_and_labels(train_loader, num_images=5)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:56.374662Z","iopub.execute_input":"2024-12-02T14:25:56.375017Z","iopub.status.idle":"2024-12-02T14:25:56.864274Z","shell.execute_reply.started":"2024-12-02T14:25:56.374987Z","shell.execute_reply":"2024-12-02T14:25:56.863386Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from collections import Counter\n\n# 모든 라벨을 리스트로 저장\nall_labels = []\nfor _, labels in train_loader:\n    all_labels.extend(labels.numpy())\n\n# 라벨 분포 출력\nlabel_counts = Counter(all_labels)\nprint(\"Label distribution:\", label_counts)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:25:58.486088Z","iopub.execute_input":"2024-12-02T14:25:58.486457Z","iopub.status.idle":"2024-12-02T14:25:59.043119Z","shell.execute_reply.started":"2024-12-02T14:25:58.486413Z","shell.execute_reply":"2024-12-02T14:25:59.042231Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"batch_size = 32\ntrain_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=1)\ntest_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=1)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:26:00.011957Z","iopub.execute_input":"2024-12-02T14:26:00.012272Z","iopub.status.idle":"2024-12-02T14:26:00.017552Z","shell.execute_reply.started":"2024-12-02T14:26:00.012246Z","shell.execute_reply":"2024-12-02T14:26:00.016482Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport timm\n\n# EfficientNet-B0 모델 (num_classes=1로 설정)\nclass EfficientNetB0(nn.Module):\n    def __init__(self, num_classes=1, dropout_prob=0.3):\n        super(EfficientNetB0, self).__init__()\n        self.model = timm.create_model('efficientnet_b0', pretrained=True, num_classes=num_classes)\n        in_features = self.model.get_classifier().in_features\n        self.model.classifier = nn.Sequential(\n            nn.Dropout(dropout_prob),\n            nn.Linear(in_features, num_classes)  # 출력 크기 1로 설정\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\n# ConvNeXt Tiny 모델 (num_classes=1로 설정)\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, num_classes=num_classes)\n        in_features = self.model.get_classifier().in_features\n        self.model.classifier = nn.Sequential(\n            nn.Dropout(dropout_prob),\n            nn.Linear(in_features, num_classes)  # 출력 크기 1로 설정\n        )\n\n    def forward(self, x):\n        return self.model(x)\n\n# 앙상블 모델\nclass EnsembleModel(nn.Module):\n    def __init__(self, model1, model2):\n        super(EnsembleModel, self).__init__()\n        self.model1 = model1\n        self.model2 = model2\n\n    def forward(self, x):\n        output1 = self.model1(x)\n        output2 = self.model2(x)\n        output = (output1 + output2) / 2  # 평균 앙상블\n        return output  # BCEWithLogitsLoss는 로짓을 그대로 사용\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)\n\n# 모델 요약 출력\nfrom torchsummary import summary\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nensemble_model = ensemble_model.to(device)\n\nprint(\"\\nEnsemble Model Summary:\")\nsummary(ensemble_model, input_size=(3, 224, 224), device=device.type)\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:26:01.346434Z","iopub.execute_input":"2024-12-02T14:26:01.346784Z","iopub.status.idle":"2024-12-02T14:26:04.081918Z","shell.execute_reply.started":"2024-12-02T14:26:01.346757Z","shell.execute_reply":"2024-12-02T14:26:04.080986Z"}},"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-12-02T14:26:04.084071Z","iopub.execute_input":"2024-12-02T14:26:04.084758Z","iopub.status.idle":"2024-12-02T14:26:04.093035Z","shell.execute_reply.started":"2024-12-02T14:26:04.084714Z","shell.execute_reply":"2024-12-02T14:26:04.092175Z"}},"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\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    mean = torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1).to(device)\n    std = torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1).to(device)\n\n    for i, (inputs, labels) in enumerate(dataloader):\n        inputs = inputs.to(device)\n        labels = labels.to(device).unsqueeze(1).float()\n\n        if i == 0:\n            for k in range(min(5, inputs.size(0))):\n                img = (inputs[k] * std + mean).clamp(0, 1).cpu()\n                plt.imshow(img.permute(1, 2, 0).numpy())\n                plt.show()\n\n        optimizer.zero_grad()\n        outputs = model(inputs)\n        loss = criterion(outputs, labels)\n        loss.backward()\n        optimizer.step()\n\n        preds = (torch.sigmoid(outputs) >= 0.5).float()\n        running_loss += loss.item() * inputs.size(0)\n        correct += preds.eq(labels.view(-1, 1)).sum().item()\n        total += labels.size(0)\n\n        if (i + 1) % log_interval == 0:\n            print(f\"Batch {i+1}/{len(dataloader)} - Loss: {loss.item():.4f}\")\n\n    return running_loss / len(dataloader.dataset), 100.0 * correct / total\n\n\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).unsqueeze(1).float()\n            outputs = model(inputs)\n            loss = criterion(outputs, labels)\n\n            # 이진화\n            preds = (torch.sigmoid(outputs) >= 0.5).float()  # Sigmoid 후 이진화\n            running_loss += loss.item() * inputs.size(0)\n            correct += preds.eq(labels).sum().item()\n            total += labels.size(0)\n\n    return running_loss / len(dataloader.dataset), 100 * correct / total\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T13:50:01.438692Z","iopub.status.idle":"2024-12-02T13:50:01.438957Z","shell.execute_reply.started":"2024-12-02T13:50:01.438824Z","shell.execute_reply":"2024-12-02T13:50:01.438837Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import copy\nimport torch\nimport torch.nn.functional as F\nimport numpy as np\nimport matplotlib.pyplot as plt\n\n# LogLoss 계산 함수\ndef calculate_log_loss(predictions, labels):\n    epsilon = 1e-12  # 안전 클램핑\n    predictions = torch.sigmoid(predictions)  # 로짓 -> 확률 변환\n    predictions = torch.clamp(predictions, epsilon, 1. - epsilon)  # 클램핑\n\n    log_loss = -torch.mean(labels * torch.log(predictions) + (1 - labels) * torch.log(1 - predictions))\n    \n    # NaN 체크\n    if torch.isnan(log_loss).any():\n        print(f\"LogLoss NaN Debug - Predictions: {predictions}, Labels: {labels}\")\n        return torch.tensor(float('nan')).to(labels.device)  # NaN 반환\n    \n    return log_loss\n\n\n# 학습 함수 (LogLoss 포함)\ndef train_one_epoch_with_logloss(model, dataloader, criterion, optimizer, device, log_interval=100):\n    model.train()  # 모델을 학습 모드로 설정\n    running_loss = 0.0\n    correct = 0\n    total = 0\n    batch_losses = []  # 배치별 손실 저장\n    log_losses = []  # LogLoss 값을 위한 리스트 추가\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        outputs = model(inputs)\n\n        # BCEWithLogitsLoss 계산\n        loss = criterion(outputs, labels)  # 손실 계산\n        loss.backward()\n        optimizer.step()\n\n        # LogLoss 계산\n        log_loss = calculate_log_loss(outputs, labels)\n        if torch.isnan(log_loss):\n            print(\"NaN detected in LogLoss calculation! Skipping this batch.\")\n            continue  # NaN 감지 시 해당 배치 생략\n\n        log_losses.append(log_loss.item())  # 배치별 LogLoss 저장\n\n        # 손실 및 정확도 계산\n        running_loss += loss.item() * inputs.size(0)\n        batch_losses.append(loss.item())  # 배치 손실 추가\n        preds_prob = torch.sigmoid(outputs)  # 시그모이드 적용하여 확률로 변환\n        preds_prob = torch.clamp(preds_prob, 0.0, 1.0)  # 예측 확률 클램핑\n        preds = (preds_prob >= 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}, LogLoss: {log_loss.item():.4f}')\n\n    epoch_loss = running_loss / len(dataloader.dataset)\n    epoch_accuracy = 100 * correct / total\n    average_log_loss = np.mean(log_losses)  # 전체 에포크에 대한 평균 LogLoss 계산\n    return epoch_loss, epoch_accuracy, batch_losses, average_log_loss\n\n\n# 테스트 함수 (LogLoss 포함)\ndef test_model_with_logloss(model, dataloader, criterion, device):\n    model.eval()  # 모델을 평가 모드로 설정\n    predictions = []\n    true_labels = []\n    test_losses = []  # 테스트 손실 저장\n    log_losses = []  # 테스트 LogLoss 값 저장\n\n    with torch.no_grad():\n        for inputs, labels in dataloader:\n            # 데이터 GPU로 이동\n            inputs, labels = inputs.to(device), labels.to(device)\n            labels = labels.unsqueeze(1).float()  # 테스트 데이터도 이진 분류에 맞게 변경\n\n            outputs = model(inputs)\n            preds_prob = torch.sigmoid(outputs)  # 시그모이드 적용 후 확률로 변환\n            preds_prob = torch.clamp(preds_prob, 0.0, 1.0)  # 예측 확률 클램핑\n            preds = (preds_prob >= 0.5).float()  # 0 또는 1로 변환\n            predictions.append(preds)\n            true_labels.append(labels)\n\n            # 손실 계산\n            loss = criterion(outputs, labels)\n            test_losses.append(loss.item())\n\n            # LogLoss 계산\n            log_loss = calculate_log_loss(outputs, labels)\n            if not torch.isnan(log_loss):\n                log_losses.append(log_loss.item())\n\n    predictions = torch.cat(predictions, dim=0).cpu().numpy()\n    true_labels = torch.cat(true_labels, dim=0).cpu().numpy()\n    avg_test_loss = np.mean(test_losses)\n    avg_log_loss = np.mean(log_losses) if log_losses else float('nan')  # LogLoss가 비어있을 경우 처리\n    return true_labels, predictions, avg_test_loss, avg_log_loss\n\n\n# 학습 및 테스트 설정\nnum_epochs = 20\ntrain_losses = []  # 에포크별 손실 저장\ntrain_accuracies = []  # 에포크별 정확도 저장\ntrain_log_losses = []  # 에포크별 LogLoss 저장\nbest_model_wts = copy.deepcopy(ensemble_model.state_dict())  # 초기 모델 가중치 저장\nbest_acc = 0.0  # 최적 정확도 기준 설정\n\nfor epoch in range(num_epochs):\n    print(f\"Epoch {epoch + 1}/{num_epochs}\")\n\n    # Train 데이터셋 학습\n    train_loss, train_acc, _, avg_log_loss = train_one_epoch_with_logloss(\n        ensemble_model, train_loader, criterion, optimizer, device\n    )\n    train_losses.append(train_loss)\n    train_accuracies.append(train_acc)\n    train_log_losses.append(avg_log_loss)\n    print(f\"Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.4f}, Train LogLoss: {avg_log_loss:.4f}\")\n\n    # 최적 모델 가중치 업데이트 (정확도가 이전보다 향상되었을 때)\n    if train_acc > best_acc:\n        best_acc = train_acc\n        best_model_wts = copy.deepcopy(ensemble_model.state_dict())  # 최적 가중치 저장\n\n# 최적 모델 로드\nensemble_model.load_state_dict(best_model_wts)\n\n# 테스트 데이터셋 평가\ntest_true_labels, test_predictions, test_loss, test_log_loss = test_model_with_logloss(\n    ensemble_model, test_loader, criterion, device\n)\n\n# 정확도 계산\ntest_true_labels = np.array(test_true_labels).astype(int)\ntest_predictions = np.array(test_predictions).astype(int)\ntest_acc = np.mean(test_true_labels == test_predictions) * 100\nprint(f\"Test Loss: {test_loss:.4f}, Test Accuracy: {test_acc:.4f}%, Test LogLoss: {test_log_loss:.4f}\")\n\n# 손실 및 정확도 시각화\ndef plot_metrics(train_losses, train_accuracies, test_loss, test_acc, train_log_losses, test_log_loss):\n    epochs = range(1, len(train_losses) + 1)\n\n    # 손실 시각화\n    plt.figure(figsize=(12, 6))\n    plt.subplot(1, 2, 1)\n    plt.plot(epochs, train_losses, 'bo-', label='Train Loss')\n    plt.scatter([len(epochs)], [test_loss], color='red', label='Test Loss (Final)')\n    plt.title('Loss Over Epochs')\n    plt.xlabel('Epochs')\n    plt.ylabel('Loss')\n    plt.legend()\n\n    # 정확도 시각화\n    plt.subplot(1, 2, 2)\n    plt.plot(epochs, train_accuracies, 'go-', label='Train Accuracy')\n    plt.scatter([len(epochs)], [test_acc], color='red', label='Test Accuracy (Final)')\n    plt.title('Accuracy Over Epochs')\n    plt.xlabel('Epochs')\n    plt.ylabel('Accuracy')\n    plt.legend()\n\n    # LogLoss 시각화\n    plt.figure(figsize=(12, 6))\n    plt.plot(epochs, train_log_losses, 'bo-', label='Train LogLoss')\n    plt.scatter([len(epochs)], [test_log_loss], color='red', label='Test LogLoss (Final)')\n    plt.title('LogLoss Over Epochs')\n    plt.xlabel('Epochs')\n    plt.ylabel('LogLoss')\n    plt.legend()\n\n    plt.tight_layout()\n    plt.show()\n\n# 결과 시각화 호출\nplot_metrics(train_losses, train_accuracies, test_loss, test_acc, train_log_losses, test_log_loss)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-02T14:26:17.677355Z","iopub.execute_input":"2024-12-02T14:26:17.677720Z","iopub.status.idle":"2024-12-02T14:39:57.105053Z","shell.execute_reply.started":"2024-12-02T14:26:17.677692Z","shell.execute_reply":"2024-12-02T14:39:57.103924Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}