{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.6.6","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":16880,"databundleVersionId":858837,"sourceType":"competition"},{"sourceId":8661913,"sourceType":"datasetVersion","datasetId":5189844},{"sourceId":8664205,"sourceType":"datasetVersion","datasetId":5191601},{"sourceId":8679034,"sourceType":"datasetVersion","datasetId":5202615},{"sourceId":225000801,"sourceType":"kernelVersion"},{"sourceId":234796630,"sourceType":"kernelVersion"}],"dockerImageVersionId":29845,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# mtcnn installation\n!pip install /kaggle/input/mtcnn1/mtcnn-0.1.0-py3-none-any.whl -q\n!pip install /kaggle/input/facenet/facenet_pytorch-2.2.9-py3-none-any.whl -q\n!pip install mtcnn -q","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:18.400304Z","iopub.execute_input":"2025-04-29T03:19:18.400832Z","iopub.status.idle":"2025-04-29T03:19:34.488415Z","shell.execute_reply.started":"2025-04-29T03:19:18.400564Z","shell.execute_reply":"2025-04-29T03:19:34.487631Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Hyperparameters Configuration","metadata":{}},{"cell_type":"code","source":"import os\nfrom enum import Enum\n\nclass ModelType(Enum):\n    RESNET50 = \"resnet50\"\n    EFFICIENTNET = \"efficientnet\"\n    CUSTOM_CNN = \"custom_cnn\"\n\n# Dataset config\nDATASET_PATH = \"/kaggle/input/deepfake-detection-challenge/train_sample_videos\"\nTRAIN_RATIO = 0.8\nFRAME_SAMPLE_RATE = 1\n\n# Face detection config\nFACE_DETECTION_SIZE = (224, 224)\nMIN_FACE_CONFIDENCE = 0.8\n\n# EVA config\nEVA_LEVELS = 4\nEVA_AMPLIFICATION_FACTOR = 10\nEVA_FREQUENCY_MIN = 0.05\nEVA_FREQUENCY_MAX = 0.4\n\n# Training config\nBATCH_SIZE = 32\nEPOCHS = 200\nLEARNING_RATE = 1e-4\nFEATURE_DIM = 512\n\n# Paths\nMODEL_SAVE_PATH = \"saved_models\"\nVISUALIZATION_PATH = \"visualizations\"","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:34.490738Z","iopub.execute_input":"2025-04-29T03:19:34.491053Z","iopub.status.idle":"2025-04-29T03:19:34.497487Z","shell.execute_reply.started":"2025-04-29T03:19:34.490996Z","shell.execute_reply":"2025-04-29T03:19:34.496768Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Extract Faces from videos","metadata":{}},{"cell_type":"code","source":"import os\nimport cv2\nimport numpy as np\nfrom tqdm import tqdm\nfrom mtcnn import MTCNN # face cropping model\nfrom matplotlib import pyplot as plt\n\nimport extract_faces as ef\n# dir(ef)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:34.498848Z","iopub.execute_input":"2025-04-29T03:19:34.499127Z","iopub.status.idle":"2025-04-29T03:19:39.799059Z","shell.execute_reply.started":"2025-04-29T03:19:34.499076Z","shell.execute_reply":"2025-04-29T03:19:39.798447Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"df = ef.create_dataframe('/kaggle/input/deepfake-detection-challenge/train_sample_videos/metadata.json')\nfor i in range(len(df)):\n    df['video_name'][i] = os.path.join(DATASET_PATH, df['video_name'][i])","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:39.800412Z","iopub.execute_input":"2025-04-29T03:19:39.800724Z","iopub.status.idle":"2025-04-29T03:19:39.873203Z","shell.execute_reply.started":"2025-04-29T03:19:39.800668Z","shell.execute_reply":"2025-04-29T03:19:39.872685Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"final_df = df","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:39.875290Z","iopub.execute_input":"2025-04-29T03:19:39.875488Z","iopub.status.idle":"2025-04-29T03:19:39.878724Z","shell.execute_reply.started":"2025-04-29T03:19:39.875442Z","shell.execute_reply":"2025-04-29T03:19:39.877896Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"dataset_dir = \"dataset\"\nreal_dir = \"real\"\nfake_dir = \"fake\"\n\ntry:\n    os.makedirs(os.path.join(dataset_dir, real_dir))\n    os.makedirs(os.path.join(dataset_dir, fake_dir))\nexcept FileExistsError:\n    print(f\"One or more directory with the given name already exists.\")\nexcept Exception as e:\n    print(f\"An error occured: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:39.881095Z","iopub.execute_input":"2025-04-29T03:19:39.881381Z","iopub.status.idle":"2025-04-29T03:19:39.893094Z","shell.execute_reply.started":"2025-04-29T03:19:39.881329Z","shell.execute_reply":"2025-04-29T03:19:39.892227Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# VideoCapture(var), take path as var to open a video\nimport cv2\nfrom matplotlib import pyplot as plt\n\ncap = cv2.VideoCapture(df.iloc[0][0])\nwhile cap.isOpened():\n    ret, frame = cap.read()\n    if not ret: break\n    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n    plt.imshow(frame)\n    plt.show()\n    break\ncap.release()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:39.894112Z","iopub.execute_input":"2025-04-29T03:19:39.894336Z","iopub.status.idle":"2025-04-29T03:19:40.497932Z","shell.execute_reply.started":"2025-04-29T03:19:39.894291Z","shell.execute_reply":"2025-04-29T03:19:40.496680Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# DataLoader class\nclass DeepfakeDataLoader:\n    def __init__(self):\n        self.detector = MTCNN()\n        self.real_path = \"/kaggle/working/dataset/real\"\n        self.fake_path = \"/kaggle/working/dataset/fake\"\n\n    def _detect_face(self, frame):\n        \"\"\"Detect and crop face using MTCNN-tensorflow backend\"\"\"\n        results = self.detector.detect_faces(frame)\n        if not results:\n            return None\n        # change the logic here for multiple faces\n        best_face = max(results, key=lambda x: x['confidence'])\n        if best_face['confidence'] < MIN_FACE_CONFIDENCE:\n            return None\n        # change this to isotropic resizing\n        # bounding box\n        x, y, w, h = best_face['box']\n        x, y = max(0, x), max(0, y)\n\n        # crop and resize\n        face = frame[y:y+h, x:x+w]\n        face = cv2.resize(face, FACE_DETECTION_SIZE)\n        return face\n\n    def load_video_frames(self, video_path, max_frames=100):\n        \"\"\"Load and preprocess video frames\"\"\"\n        cap = cv2.VideoCapture(video_path)\n        frames = []\n        frame_count = 0\n\n        while cap.isOpened() and frame_count < max_frames:\n            ret, frame = cap.read()\n            if not ret:\n                break\n\n            if frame_count % FRAME_SAMPLE_RATE == 0:\n                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n                face = self._detect_face(frame)\n                if face is not None:\n                    frames.append(face)\n\n            frame_count += 1\n        cap.release()\n        return np.array(frames)\n\n    def create_video(self, df, loc_path):\n        w, h, _ = df[1][0].shape # first frame\n\n        codec = cv2.VideoWriter_fourcc(*'mp4v')\n        vid_writer = cv2.VideoWriter(os.path.join(loc_path, df[0]), codec, 30, (w, h))\n\n        # for frames in df[1]:\n        #     for _ in range(20):\n        #         vid_writer.write(frames)\n        for frames in df[1]:\n            frames = cv2.cvtColor(frames, cv2.COLOR_BGR2RGB)\n            vid_writer.write(frames)\n\n        vid_writer.release()\n\n    def load_dataset(self, df):\n        \"\"\"Load entire dataset\"\"\"\n        real_videos = []\n        fake_videos = []\n        real = df[df['label']==\"REAL\"].values.tolist()\n        fake = df[df['label']==\"FAKE\"].values.tolist()\n        false = \"off\"\n        try:\n            print(\"Loading real videos...\")\n            for video_file in tqdm(real):\n                frames = self.load_video_frames(video_file[0])\n                if len(frames) > 0:\n                    real_videos.append([video_file[0].split(\"/\")[-1], frames])\n            for video_file in real_videos:\n                self.create_video(video_file, self.real_path)\n                \n            # plot some samples\n            print(f\"Few samples of the Real class.\")\n            plt.subplot(1, 3, 1)\n            plt.imshow(real_videos[0][1][0])\n            plt.axis(false)\n            plt.subplot(1, 3, 2)\n            plt.imshow(real_videos[0][1][1])\n            plt.axis(false)\n            plt.subplot(1, 3, 3)\n            plt.imshow(real_videos[0][1][2])\n            plt.axis(false)\n            plt.show()\n        except Exception as e:\n            print(f\"Encountered an error : {e}\")\n            \n        try:\n            print(\"Loading fake vidoes...\")\n            for video_file in tqdm(fake):\n                frames = self.load_video_frames(video_file[0])\n                if len(frames) > 0:\n                    fake_videos.append([video_file[0].split(\"/\")[-1], frames])\n            for video_file in fake_videos:\n                self.create_video(video_file, self.fake_path)\n    \n            # plot some samples\n            print(f\"Few samples of the Fake class.\")\n            plt.subplot(1, 3, 1)\n            plt.imshow(fake_videos[0][1][0])\n            plt.axis(false)\n            plt.subplot(1, 3, 2)\n            plt.imshow(fake_videos[0][1][1])\n            plt.axis(false)\n            plt.subplot(1, 3, 3)\n            plt.imshow(fake_videos[0][1][2])\n            plt.axis(false)\n            plt.show()\n        except Exception as e:\n            print(f\"Encounterd an error : {e}\")\n        return real_videos, fake_videos","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:40.499621Z","iopub.execute_input":"2025-04-29T03:19:40.499995Z","iopub.status.idle":"2025-04-29T03:19:40.543555Z","shell.execute_reply.started":"2025-04-29T03:19:40.499938Z","shell.execute_reply":"2025-04-29T03:19:40.542510Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## <mark>Change it to complete dataset</mark> and move it to last ","metadata":{}},{"cell_type":"code","source":"dataloader = DeepfakeDataLoader()\n# make it run for all videos\nreal_videos, fake_videos = dataloader.load_dataset(df.iloc[0:4])","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:19:40.544917Z","iopub.execute_input":"2025-04-29T03:19:40.545171Z","iopub.status.idle":"2025-04-29T03:21:30.153330Z","shell.execute_reply.started":"2025-04-29T03:19:40.545118Z","shell.execute_reply":"2025-04-29T03:21:30.150624Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# convert sample videos into gif and present it for demo\ngif_dir = \"gifs\"\nreal_dir = \"real\"\nfake_dir = \"fake\"\n\ntry:\n    os.makedirs(os.path.join(gif_dir, real_dir))\n    os.makedirs(os.path.join(gif_dir, fake_dir))\nexcept FileExistsError:\n    print(f\"One or more directory with the given name already exists.\")\nexcept Exception as e:\n    print(f\"An error occured: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:30.155674Z","iopub.execute_input":"2025-04-29T03:21:30.156227Z","iopub.status.idle":"2025-04-29T03:21:30.166404Z","shell.execute_reply.started":"2025-04-29T03:21:30.156001Z","shell.execute_reply":"2025-04-29T03:21:30.165395Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import cv2\nimport imageio\nimport os\n\ndef video_to_gif(input_video_path, output_gif_path, fps=10, scale=1.0):\n    \"\"\"Convert videos to gifs\"\"\"\n    # Read the video using OpenCV\n    cap = cv2.VideoCapture(input_video_path)\n    if not cap.isOpened():\n        raise ValueError(\"Could not open the video file\")\n    \n    # Get video properties\n    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) * scale)\n    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) * scale)\n    frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))\n    \n    # Calculate skip frames to match target FPS\n    video_fps = cap.get(cv2.CAP_PROP_FPS)\n    skip_frames = max(1, int(video_fps / fps))\n    \n    print(f\"Converting video to GIF: {frame_count} frames, original FPS: {video_fps}, target FPS: {fps}\")\n    \n    frames = []\n    frame_idx = 0\n    \n    while True:\n        ret, frame = cap.read()\n        if not ret:\n            break\n            \n        # Skip frames to match target FPS\n        if frame_idx % skip_frames != 0:\n            frame_idx += 1\n            continue\n            \n        # Resize if needed\n        if scale != 1.0:\n            frame = cv2.resize(frame, (width, height), interpolation=cv2.INTER_AREA)\n            \n        # Convert BGR to RGB\n        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n        frames.append(frame)\n        frame_idx += 1\n    \n    cap.release()\n    \n    # Save as GIF using imageio\n    imageio.mimsave(output_gif_path, frames, fps=fps, loop=0)\n    print(f\"GIF saved to {output_gif_path}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:30.168071Z","iopub.execute_input":"2025-04-29T03:21:30.168415Z","iopub.status.idle":"2025-04-29T03:21:30.222398Z","shell.execute_reply.started":"2025-04-29T03:21:30.168359Z","shell.execute_reply":"2025-04-29T03:21:30.221746Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"video_to_gif(\"/kaggle/working/dataset/real/abarnvbtwb.mp4\", \"/kaggle/working/gifs/real/abarnvbtwb.gif\", fps=30)\nvideo_to_gif(\"/kaggle/working/dataset/fake/aagfhgtpmv.mp4\", \"/kaggle/working/gifs/fake/aagfhgtpmv.gif\", fps=30)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:30.223374Z","iopub.execute_input":"2025-04-29T03:21:30.223599Z","iopub.status.idle":"2025-04-29T03:21:33.584233Z","shell.execute_reply.started":"2025-04-29T03:21:30.223559Z","shell.execute_reply":"2025-04-29T03:21:33.583251Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"<h3>Sample Videos</h3>\n<div style=\"display:flex;\">\n    <div>\n    <h3>Real Video</h3> \n    <img src=\"./gifs/real/abarnvbtwb.gif\" />\n    </div>\n    <div>\n    <h3>Fake Video</h3> \n    <img src=\"./gifs/fake/aagfhgtpmv.gif\" />\n    </div>\n</div>","metadata":{}},{"cell_type":"markdown","source":"# Applying EVA","metadata":{}},{"cell_type":"code","source":"import cv2\nimport numpy as np\nfrom scipy import fftpack\n\nclass EulerianVideoAmplification:\n    def __init__(self):\n        self.levels = EVA_LEVELS\n        self.alpha = EVA_AMPLIFICATION_FACTOR\n        self.freq_min = EVA_FREQUENCY_MIN\n        self.freq_max = EVA_FREQUENCY_MAX\n\n    def _build_gaussian_pyramid(self, frame):\n        \"\"\"Building Gaussian Pyramid for spatial processing\"\"\"\n        pyramid = [frame]\n        for _ in range(self.levels -1 ):\n            frame = cv2.pyrDown(frame)\n            pyramid.append(frame)\n        return pyramid\n\n    def _reconstruct_frame(self, pyramid):\n        \"\"\"Reconstruct frame from Gaussian pyramid\"\"\"\n        current = pyramid[-1]\n        for i in range(len(pyramid)-2, -1, -1):\n            current = cv2.pyrUp(current, dstsize=(\n                pyramid[i].shape[1], pyramid[i].shape[0]\n            ))\n            current = cv2.add(current, pyramid[i])\n        return current\n\n    def _temporal_bandpass_filter(self, frames):\n        \"\"\"Apply temporal bandpass filter\"\"\"\n        fft = fftpack.fft(frames, axis=0)\n        frequencies = fftpack.fftfreq(len(frames))\n\n        # create bandpass mask\n        mask = (np.abs(frequencies) >= self.freq_min) & (np.abs(frequencies) <= self.freq_max)\n\n        # apply mask\n        fft[~mask] = 0\n        return fftpack.ifft(fft, axis=0).real\n\n    def amplify(self, video_frames):\n        \"\"\"Amplify subtle motion in video\"\"\"\n        if len(video_frames) < 2:\n            return video_frames\n\n        gray_frames = [cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY) for frame in video_frames]\n        gray_frames = np.array(gray_frames)\n\n        # apply temporal bandpass filter\n        filtered = self._temporal_bandpass_filter(gray_frames)\n\n        # process each frame\n        amplified_frames = []\n\n        for i, frame in enumerate(video_frames):\n            # build pyramid for current frame\n            pyramid = self._build_gaussian_pyramid(frame)\n\n            # amplify each level\n            for level in range(self.levels):\n                filtered_level = cv2.pyrDown(filtered[i]) if level > 0 else filtered[i]\n                for _ in range(level - 1):\n                    filtered_level = cv2.pyrDown(filtered_level)\n\n                # convert grayscale to color\n                if len(pyramid[level].shape) == 3:  # If color image\n                    filtered_level = cv2.cvtColor(filtered_level.astype(np.uint8), cv2.COLOR_GRAY2RGB)\n\n                pyramid[level] = pyramid[level] + self.alpha * filtered_level\n\n            amplified_frame = self._reconstruct_frame(pyramid)\n            amplified_frames.append(amplified_frame)\n\n        return np.array(amplified_frames)\n\n    def process_video(self, video_frames):\n        \"\"\"Process video with EVA and return amplified frames\"\"\"\n        amplified_frames = self.amplify(video_frames)\n        # additional processing - edge detection\n        processed_frames = []\n        for frame in amplified_frames:\n            # converting to LAB color space for edge enhancement\n            lab = cv2.cvtColor(frame.astype(np.uint8), cv2.COLOR_RGB2LAB)\n            l, a, b = cv2.split(lab)\n\n            clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))\n            l = clahe.apply(l)\n\n            enhanced_lab = cv2.merge((l, a, b))\n            enhanced_rgb = cv2.cvtColor(enhanced_lab, cv2.COLOR_LAB2RGB)\n\n            processed_frames.append(enhanced_rgb)\n            \n        return np.array(processed_frames)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:33.585612Z","iopub.execute_input":"2025-04-29T03:21:33.585913Z","iopub.status.idle":"2025-04-29T03:21:33.613802Z","shell.execute_reply.started":"2025-04-29T03:21:33.585857Z","shell.execute_reply":"2025-04-29T03:21:33.613248Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"dataset_dir = \"processed\"\nreal_dir = \"real\"\nfake_dir = \"fake\"\n\ntry:\n    os.makedirs(os.path.join(dataset_dir, real_dir))\n    os.makedirs(os.path.join(dataset_dir, fake_dir))\nexcept FileExistsError:\n    print(f\"One or more directory with the given name already exists.\")\nexcept Exception as e:\n    print(f\"An error occured: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:33.615376Z","iopub.execute_input":"2025-04-29T03:21:33.615692Z","iopub.status.idle":"2025-04-29T03:21:33.620918Z","shell.execute_reply.started":"2025-04-29T03:21:33.615636Z","shell.execute_reply":"2025-04-29T03:21:33.620133Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"dataset_dir = \"comparision\"\nreal_dir = \"real\"\nfake_dir = \"fake\"\n\ntry:\n    os.makedirs(os.path.join(dataset_dir, real_dir))\n    os.makedirs(os.path.join(dataset_dir, fake_dir))\nexcept FileExistsError:\n    print(f\"One or more directory with the given name already exists.\")\nexcept Exception as e:\n    print(f\"An error occured: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:33.622130Z","iopub.execute_input":"2025-04-29T03:21:33.622362Z","iopub.status.idle":"2025-04-29T03:21:33.632525Z","shell.execute_reply.started":"2025-04-29T03:21:33.622316Z","shell.execute_reply":"2025-04-29T03:21:33.631947Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## <mark>Running EVA on processed face video dataset</mark>","metadata":{}},{"cell_type":"code","source":"# create video of the processed files\ndef create_video(df, loc_path):\n    w, h, _ = df[1][0].shape # first frame\n\n    codec = cv2.VideoWriter_fourcc(*'mp4v')\n    vid_writer = cv2.VideoWriter(os.path.join(loc_path, df[0]), codec, 30, (w, h))\n\n    for frames in df[1]:\n        # frames = cv2.cvtColor(frames, cv2.COLOR_BGR2RGB)\n        vid_writer.write(frames)\n\n    vid_writer.release()\n\nreal_path = '/kaggle/working/processed/real'\nfake_path = '/kaggle/working/processed/fake'\n\neva = EulerianVideoAmplification()\n\ntry:\n    for video in real_videos:\n        processed_frames = eva.process_video(video[1])\n        create_video([video[0], processed_frames], real_path)\nexcept Exception as e:\n    print(f\"Encountered an error with real video: {e}\")\n\ntry:\n    for video in fake_videos:\n        processed_frames = eva.process_video(video[1])\n        create_video([video[0], processed_frames], fake_path)\nexcept Exception as e:\n    print(f\"Encountered an error with fake video: {e}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:33.633528Z","iopub.execute_input":"2025-04-29T03:21:33.633712Z","iopub.status.idle":"2025-04-29T03:21:35.581449Z","shell.execute_reply.started":"2025-04-29T03:21:33.633679Z","shell.execute_reply":"2025-04-29T03:21:35.580859Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"video_to_gif(\"/kaggle/working/processed/fake/aagfhgtpmv.mp4\", \"/kaggle/working/gifs/fake/aagfhgtpmv-processed.gif\", fps=30)\nvideo_to_gif(\"/kaggle/working/processed/real/abarnvbtwb.mp4\", \"/kaggle/working/gifs/real/abarnvbtwb-processed.gif\", fps=30)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:35.582498Z","iopub.execute_input":"2025-04-29T03:21:35.582693Z","iopub.status.idle":"2025-04-29T03:21:40.126009Z","shell.execute_reply.started":"2025-04-29T03:21:35.582659Z","shell.execute_reply":"2025-04-29T03:21:40.125163Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"<h3>Sample Videos</h3>\n<div style=\"display:flex;\">\n    <div>\n    <h3>Real Video</h3> \n    <img src=\"./gifs/real/abarnvbtwb-processed.gif\" />\n    </div>\n    <div>\n    <h3>Fake Video</h3> \n    <img src=\"./gifs/fake/aagfhgtpmv-processed.gif\" />\n    </div>\n</div>","metadata":{}},{"cell_type":"markdown","source":"# Feature Extractor","metadata":{}},{"cell_type":"code","source":"import tensorflow as tf\nimport numpy as np\nfrom tensorflow.keras.applications import ResNet50\nfrom tensorflow.keras.models import Model\nfrom tensorflow.keras.layers import Dense, GlobalAveragePooling2D, Input\n\nclass FeatureExtraction:\n    def __init__(self, model_type=ModelType.RESNET50):\n        self.model_type = model_type\n        self.model = self._build_feature_extractor()\n\n    def _build_feature_extractor(self):\n        \"\"\"Build feature extractor model\"\"\"\n        input_shape = FACE_DETECTION_SIZE + (3, )\n        inputs = Input(shape=input_shape)\n\n        if self.model_type == ModelType.RESNET50:\n            base_model = ResNet50(\n                weights='imagenet',\n                include_top=False,\n                input_shape=input_shape\n            )\n        elif self.model_type == ModelType.EFFICIENTNET:\n            base_model = EfficientNetB0(\n                weights='imagenet', \n                include_top=False,\n                input_shape=input_shape\n            )\n        else:\n            x = tf.keras.layers.Conv2D(32, (3, 3), activation='relu')(inputs)\n            x = tf.keras.layers.MaxPooling2D((2, 2))(x)\n            x = tf.keras.layers.Conv2D(64, (3, 3), activation='relu')(x)\n            x = tf.keras.layers.MaxPooling2D((2, 2))(x)\n            x = tf.keras.layers.Conv2D(128, (3, 3), activation='relu')(x)\n            x = tf.keras.layers.MaxPooling2D((2, 2))(x)\n            x = tf.keras.layers.Conv2D(256, (3, 3), activation='relu')(x)\n            x = tf.keras.layers.GlobalAveragePooling2D()(x)\n            return Model(inputs=inputs, outputs=x)\n\n        base_model.trainable = False\n\n        x = base_model(inputs)\n        x = GlobalAveragePooling2D()(x)\n        x = Dense(FEATURE_DIM, activation='relu')(x)\n\n        return Model(inputs=inputs, outputs=x)\n\n    def extract_features(self, frames):\n        \"\"\"Extract features from the video frames\"\"\"\n        if len(frames.shape) == 3:\n            frames = np.expand_dims(frames, axis=0)\n\n        if self.model_type == ModelType.RESNET50:\n            processed_frames = tf.keras.applications.resnet50.preprocess_input(frames)\n        elif self.model_type == ModelType.EFFICIENTNET:\n            processed_frames = tf.keras.applications.efficientnet.preprocess_input(frames)\n        else:\n            processed_frames = frames / 255.0\n\n        features = self.model.predict(processed_frames)\n        return features\n\n    def extract_video_features(self, video_frames):\n        \"\"\"Extract features of all frames in a video\"\"\"\n        frame_features = []\n        for frame in video_frames:\n            features = self.extract_features(frame)\n            frame_features.append(features)\n\n        video_features = np.mean(frame_features, axis=0)\n        return video_features","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:40.127293Z","iopub.execute_input":"2025-04-29T03:21:40.127537Z","iopub.status.idle":"2025-04-29T03:21:40.146071Z","shell.execute_reply.started":"2025-04-29T03:21:40.127488Z","shell.execute_reply":"2025-04-29T03:21:40.145399Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"fe = FeatureExtraction()\nfeatures = fe.extract_video_features(fake_videos[0][1])","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:40.147199Z","iopub.execute_input":"2025-04-29T03:21:40.147478Z","iopub.status.idle":"2025-04-29T03:21:50.102229Z","shell.execute_reply.started":"2025-04-29T03:21:40.147405Z","shell.execute_reply":"2025-04-29T03:21:50.101584Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"features.shape","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:50.103560Z","iopub.execute_input":"2025-04-29T03:21:50.103820Z","iopub.status.idle":"2025-04-29T03:21:50.108728Z","shell.execute_reply.started":"2025-04-29T03:21:50.103769Z","shell.execute_reply":"2025-04-29T03:21:50.108029Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Vanilla classification model","metadata":{}},{"cell_type":"code","source":"import tensorflow as tf\nimport numpy as np\nfrom tensorflow.keras.models import Sequential\nfrom tensorflow.keras.layers import Dense, Dropout, BatchNormalization\nfrom tensorflow.keras.regularizers import l2\n\nclass DeepfakeClassifier:\n    def __init__(self, input_dim=FEATURE_DIM):\n        self.model = self._build_classifier(input_dim)\n\n    def _build_classifier(self, input_dim):\n        \"\"\"Build binary classifier model\"\"\"\n        model = Sequential([\n            Dense(256, activation='relu', input_dim=input_dim, kernel_regularizer=l2(0.01)),\n            BatchNormalization(),\n            Dropout(0.5),\n            Dense(128, activation='relu', kernel_regularizer=l2(0.01)),\n            BatchNormalization(),\n            Dropout(0.5),\n            Dense(64, activation='relu', kernel_regularizer=l2(0.01)),\n            BatchNormalization(),\n            Dense(1, activation='sigmoid')\n        ])\n\n        model.compile(\n            optimizer=tf.keras.optimizers.Adam(LEARNING_RATE),\n            loss='binary_crossentropy',\n            metrics=['accuracy', tf.keras.metrics.AUC()]\n        )\n\n        return model\n\n    def train(self, X_train, y_train, X_val, y_val):\n        \"\"\"Train the classifier\"\"\"\n        callbacks = [\n            tf.keras.callbacks.EarlyStopping(patience=10,restore_best_weights=True),\n            tf.keras.callbacks.ModelCheckpoint(\n                f\"{MODEL_SAVE_PATH}/classifier.h5\",\n                save_best_only=True,\n                monitor='val_accuracy'\n            ),\n            tf.keras.callbacks.TensorBoard(log_dir='logs')\n        ]\n\n        history = self.model.fit(\n            X_train, y_train, \n            validation_data=(X_val, y_val),\n            epochs=EPOCHS,\n            batch_size=BATCH_SIZE,\n            callbacks=callbacks\n        )\n\n        return history\n\n    def evaluate(self, X_test, y_test):\n        \"\"\"Evaluate model performance\"\"\"\n        return self.model.evaluate(X_test, y_test)\n\n    def predict(self, X):\n        \"\"\"Make predictions\"\"\"\n        return self.model.predict(X)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:50.109835Z","iopub.execute_input":"2025-04-29T03:21:50.110101Z","iopub.status.idle":"2025-04-29T03:21:50.123550Z","shell.execute_reply.started":"2025-04-29T03:21:50.110060Z","shell.execute_reply":"2025-04-29T03:21:50.122679Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Visualizations","metadata":{}},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport numpy as np\nimport cv2\nimport tensorflow as tf\nfrom tensorflow.keras.models import Model\n\nclass DeepfakeVisualizer:\n    def __init__(self, feature_extractor, classifier):\n        self.feature_extractor = feature_extractor\n        self.classifier = classifier\n\n    def plot_training_history(self, history):\n        \"\"\"Plot training and validation metrics\"\"\"\n        plt.figure(figsize=(12, 4))\n\n        plt.subplot(1, 2, 1)\n        plt.plot(history.history['accuracy'], label=\"Train Accuracy\")\n        plt.plot(history.history['val_accuracy'], label=\"Val Accuracy\")\n        plt.title(\"Model Accuracy\")\n        plt.ylabel(\"Accuracy\")\n        plt.xlabel(\"Epoch\")\n        plt.legend()\n\n        plt.subplot(1, 2, 2)\n        plt.plot(history.history['loss'], label=\"Train Loss\")\n        plt.plot(history.history['val_loss'], label=\"Val Loss\")\n        plt.title(\"Model Loss\")\n        plt.ylabel(\"Loss\")\n        plt.xlabel(\"Epoch\")\n        plt.legend()\n\n        plt.tight_layout()\n        plt.savefig(f\"{VISUALIZATION_PATH}/training_history.png\")\n        plt.close()\n\n    # def generate_gradcam(self, frame, layer_name=None):\n    #     \"\"\"Generate Grad-CAM heatmap for a frame\"\"\"\n    #     if layer_name is None:\n    #         if self.feature_extractor.model_type == ModelType.RESNET50:\n    #             layer_name = \"resnet50\"\n    #         elif self.feature_extractor.model_type == ModelType.EFFICIENTNET:\n    #             layer_name = \"top_activation\"\n    #         else:\n    #             layer_name = \"conv2d_3\"\n\n    #     grad_model = Model(\n    #         inputs=[self.feature_extractor.model.inputs],\n    #         outputs=[self.feature_extractor.model.get_layer(layer_name).output, self.classifier.model.output]\n    #     )\n\n    #     with tf.GradientTape() as tape:\n    #         conv_outputs, predictions = grad_model(np.expand_dims(frame, axis=0))\n    #         loss = predictions[:, 0]\n\n    #     grads = tape.gradient(loss, conv_outputs)\n    #     pooled_grads = tf.reduce_mean(grads, axis=(0, 1, 2))\n    #     conv_outputs = conv_outputs[0]\n    #     heatmap = conv_outputs @ pooled_grads[..., tf.newaxis]\n    #     heatmap = tf.squeeze(heatmap)\n\n    #     heatmap = tf.maximum(heatmap, 0) / tf.math.reduce_max(heatmap)\n    #     heatmap = heatmap.numpy()\n\n    #     heatmap = cv2.resize(heatmap, FACE_DETECTION_SIZE)\n    #     heatmap = np.uint8(255 * heatmap)\n\n    #     heatmap = cv2.applyColorMap(heatmap, cv2.COLORMAP_JET)\n\n    #     superimposed_img = cv2.addWeighted(\n    #         cv2.cvtColor(frame, cv2.COLOR_RGB2BGR), 0.6, heatmap, 0.4, 0\n    #     )\n\n    #     return superimposed_img\n\n    def visualize_eva_effect(self, original_frames, processed_frames, n_frames=3):\n        \"\"\"Visualize the effect of eva preprocessing\"\"\"\n        plt.figure(figsize=(15, 5*n_frames))\n        \n        # Ensure inputs are numpy arrays\n        original_frames = np.array(original_frames)\n        processed_frames = np.array(processed_frames)\n        \n        # If single frame provided, convert to list of 1 frame\n        if original_frames.ndim == 3:\n            original_frames = [original_frames]\n            processed_frames = [processed_frames]\n        \n        for i in range(min(n_frames, len(original_frames))):  # Protect against short arrays\n            idx = i * len(original_frames) // n_frames\n            \n            # Original frame (convert to float [0,1] if needed)\n            plt.subplot(n_frames, 2, 2*i+1)\n            frame = original_frames[idx]\n            if frame.dtype == np.uint8:\n                frame = frame.astype(float)/255.0\n            plt.imshow(frame)\n            plt.title(f\"Original Frame {idx}\")\n            plt.axis(\"off\")\n            \n            # Processed frame (fixed subplot index)\n            plt.subplot(n_frames, 2, 2*i+2)  # Changed to 2*i+2\n            frame = processed_frames[idx]\n            if frame.dtype == np.uint8:\n                frame = frame.astype(float)/255.0\n            plt.imshow(frame)\n            plt.title(f\"Processed Frame {idx}\")\n            plt.axis(\"off\")\n    \n        plt.tight_layout()\n        plt.savefig(f\"{VISUALIZATION_PATH}/eva_effect.png\")\n        plt.close()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:50.125134Z","iopub.execute_input":"2025-04-29T03:21:50.125396Z","iopub.status.idle":"2025-04-29T03:21:50.143371Z","shell.execute_reply.started":"2025-04-29T03:21:50.125345Z","shell.execute_reply":"2025-04-29T03:21:50.142719Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Train the models","metadata":{}},{"cell_type":"code","source":"import os\nimport numpy as np\nfrom sklearn.model_selection import train_test_split\n\ndef create_directory(path):\n    if not os.path.exists(path):\n        os.makedirs(path)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:50.144406Z","iopub.execute_input":"2025-04-29T03:21:50.144702Z","iopub.status.idle":"2025-04-29T03:21:50.807213Z","shell.execute_reply.started":"2025-04-29T03:21:50.144659Z","shell.execute_reply":"2025-04-29T03:21:50.806663Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\ncreate_directory(MODEL_SAVE_PATH)\ncreate_directory(VISUALIZATION_PATH)\n\n# step 1: Load and process the complete dataset\nprint(\"Loading dataset\")\ndata_loader = DeepfakeDataLoader()\n# change it to full dataset\nreal_videos, fake_videos = data_loader.load_dataset(final_df)\n\nX = real_videos + fake_videos\ny = np.array([0]*len(real_videos) + [1]*len(fake_videos))\n\n# split dataset\nX_train, X_test, y_train, y_test = train_test_split(\n    X, y, test_size=1-TRAIN_RATIO, random_state=42, stratify=y\n)\nX_train, X_val, y_train, y_val = train_test_split(\n    X_train, y_train, test_size=0.2, random_state=42, stratify=y_train\n)\n\n# step 2: EVA\nprint(\"Processing videos with EVA...\")\neva_processor = EulerianVideoAmplification()\n\ndef process_videos(videos):\n    processed = []\n    for video in videos:\n        processed_frames = eva.process_video(video[1])\n        processed.append(processed_frames)\n    return processed\n    \nX_train_processed = process_videos(X_train)\nX_val_processed = process_videos(X_val)\nX_test_processed = process_videos(X_test)\n\n# step 3: Extract features\nprint(\"Extracting features...\")\nfeature_extractor = FeatureExtraction(model_type=ModelType.RESNET50)\n\ndef extract_features(videos):\n    features = []\n    for video in videos:\n        video_features = feature_extractor.extract_video_features(video)\n        features.append(video_features)\n    return np.vstack(features)\n\nX_train_features = extract_features(X_train_processed)\nX_val_features = extract_features(X_val_processed)\nX_test_features = extract_features(X_test_processed)\n\n# step 4: Train classifer\nprint(\"Training classifier...\")\nclassifier = DeepfakeClassifier()\nhistory = classifier.train(X_train_features, y_train, X_val_features, y_val)\n\n\n# step 5: Evaluate\nprint(\"Evaluating model...\")\ntest_loss, test_acc, test_auc = classifier.evaluate(X_test_features, y_test)\nprint(f\"Test Accuracy: {test_acc:.4f}, Test AUC: {test_auc:.4f}\")\n\n# step 6: Visualize results\nprint(\"Generating visualizations...\")\nvisualizer = DeepfakeVisualizer(feature_extractor, classifier)\n\n# plot the training history\nvisualizer.plot_training_history(history)\n# visualize eva effect\nsample_idx = np.random.randint(len(X_train))\nvisualizer.visualize_eva_effect(X_train[sample_idx][1], X_train_processed[sample_idx])\n\n# generate grad cam example\n# sample_frame = X_test_processed[0][0]\n# gradcam_img = visualizer.generate_gradcam(sample_frame)\n# cv2.imwrite(f\"{VISUALIZATION_PATH}/gradcam_example.png\", gradcam_img)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-29T03:21:50.808343Z","iopub.execute_input":"2025-04-29T03:21:50.808605Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## Zip all the outputs","metadata":{}},{"cell_type":"code","source":"classifier = DeepfakeClassifier()\nhistory = classifier.train(X_train_features, y_train, X_val_features, y_val)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}