{"metadata":{"kernelspec":{"name":"python3","display_name":"Python 3","language":"python"},"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":"none","dataSources":[{"sourceId":16880,"databundleVersionId":858837,"sourceType":"competition"},{"sourceId":9288677,"sourceType":"datasetVersion","datasetId":5623031},{"sourceId":107974,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":90434,"modelId":114652}],"dockerImageVersionId":30762,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false},"colab":{"provenance":[],"gpuType":"T4"},"accelerator":"GPU"},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n for filename in filenames:\n print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\"\n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","id":"IXQIXQGTtT6p","outputId":"d63ff830-75e5-4526-cc43-605855e39ebe","execution":{"iopub.status.busy":"2024-09-06T04:52:36.277635Z","iopub.execute_input":"2024-09-06T04:52:36.278673Z","iopub.status.idle":"2024-09-06T04:52:40.513347Z","shell.execute_reply.started":"2024-09-06T04:52:36.278605Z","shell.execute_reply":"2024-09-06T04:52:40.512122Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\nimport numpy as np\nimport cv2\nimport matplotlib.pyplot as plt\n\ndef create_high_pass_filter(shape):\n \"\"\"\n Creates a high-pass filter that zeros out low-frequency components.\n\n Parameters:\n shape (tuple): The shape of the filter (H, W)\n\n Returns:\n high_pass_filter (Tensor): The high-pass filter of shape (H, W)\n \"\"\"\n #H, W = shape\n\n # Create a grid of indices\n #i = np.arange(H) - H // 2 # Shift indices so that 0 is in the center\n #j = np.arange(W) - W // 2\n #I, J = np.meshgrid(i, j, indexing='ij')\n\n # Initialize the filter with ones\n #high_pass_filter = np.ones((H, W), dtype=np.float32)\n\n # Apply the filter condition\n #high_pass_filter[(np.abs(I) < H // 4) & (np.abs(J) < W // 4)] = 0\n\n #return tf.convert_to_tensor(high_pass_filter)\n scale = 4\n H, W = shape\n # Create a grid of indices\n i = tf.range(H, dtype=tf.float32)\n j = tf.range(W, dtype=tf.float32)\n I, J = tf.meshgrid(i, j, indexing='ij')\n # Calculate the center of the image\n h_center, w_center = H // 2, W // 2\n h_start, h_end = h_center - H // scale, h_center + H // scale\n w_start, w_end = w_center - W // scale, w_center + W // scale\n high_pass_filter = tf.ones((H, W), dtype=tf.float32)\n low_freq_mask = tf.logical_and(\n tf.logical_and(I >= h_start, I < h_end),\n tf.logical_and(J >= w_start, J < w_end)\n )\n\n high_pass_filter = tf.where(low_freq_mask, 0.0, high_pass_filter)\n return high_pass_filter\n\n\ndef high_frequency_representation_image(image):\n \"\"\"\n Applies the high-pass filter to an image in the frequency domain.\n\n Parameters:\n image (Tensor): The input image of shape (B, H, W, C)\n\n Returns:\n high_freq_image (Tensor): The high-frequency component of the image\n \"\"\"\n # Convert the image to frequency domain using FFT\n freq_rep = tf.signal.fft2d(tf.cast(image, tf.complex64))\n\n # Shift zero-frequency component to the center\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-3, -2))\n\n # Get the shape of the image and create the high-pass filter\n B, H, W, C = image.shape\n high_pass_filter = create_high_pass_filter((H, W))\n\n # Apply the high-pass filter to each channel\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=0) # for batch\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=-1) # for channels\n\n # Multiply the frequency representation by the high-pass filter\n high_freq_rep = freq_rep * tf.cast(high_pass_filter, tf.complex64)\n\n # Convert back to image domain using iFFT\n high_freq_rep = tf.signal.ifftshift(high_freq_rep, axes=(-3, -2))\n high_freq_image = tf.signal.ifft2d(high_freq_rep)\n print(tf.math.real(high_freq_image).shape)\n\n # Return the real part of the image\n return tf.math.real(high_freq_image)\n\n\n\n","metadata":{"id":"ZVFybCo8tT6t","execution":{"iopub.status.busy":"2024-09-06T04:52:48.964218Z","iopub.execute_input":"2024-09-06T04:52:48.964637Z","iopub.status.idle":"2024-09-06T04:53:05.062Z","shell.execute_reply.started":"2024-09-06T04:52:48.964597Z","shell.execute_reply":"2024-09-06T04:53:05.060739Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import cv2\nimport matplotlib.pyplot as plt\nimport numpy as np\n\ndef split_and_transform(image,img_path=False):\n # Load the image\n if img_path==True:\n image = cv2.imread(image)\n\n # Split the image into R, G, B channels\n # Get the shape of the image and create the high-pass filter\n # Convert the image to frequency domain using FFT\n freq_rep = tf.signal.fft2d(tf.cast(image, tf.complex64))\n\n # Shift zero-frequency component to the center\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-3, -2))\n B, H, W, C = tf.expand_dims(image,axis=0).shape\n high_pass_filter = create_high_pass_filter((H, W))\n\n # Apply the high-pass filter to each channel\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=0) # for batch\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=-1) # for channels\n\n # Multiply the frequency representation by the high-pass filter\n high_freq_rep = freq_rep * tf.cast(high_pass_filter, tf.complex64)\n high_freq_rep = tf.signal.ifftshift(high_freq_rep, axes=(-3, -2))\n high_freq_image = tf.signal.ifft2d(high_freq_rep)\n image_hp = tf.math.real(high_freq_image)\n print(image_hp.shape)\n B,G,R = cv2.split(np.array(tf.squeeze(image_hp)))\n\n b, g, r = cv2.split(image)\n\n # Apply Fourier Transform to each channel\n f_transform_r = np.fft.fft2(r)\n f_transform_g = np.fft.fft2(g)\n f_transform_b = np.fft.fft2(b)\n f_transform_R = np.fft.fft2(R)\n f_transform_G = np.fft.fft2(G)\n f_transform_B = np.fft.fft2(B)\n\n # Shift the zero frequency component to the center\n f_shift_r = np.fft.fftshift(f_transform_r)\n f_shift_g = np.fft.fftshift(f_transform_g)\n f_shift_b = np.fft.fftshift(f_transform_b)\n f_shift_R = np.fft.fftshift(f_transform_R)\n f_shift_G = np.fft.fftshift(f_transform_G)\n f_shift_B = np.fft.fftshift(f_transform_B)\n\n # Calculate the magnitude spectrum\n magnitude_spectrum_r = 20 * np.log(np.abs(f_shift_r))\n magnitude_spectrum_g = 20 * np.log(np.abs(f_shift_g))\n magnitude_spectrum_b = 20 * np.log(np.abs(f_shift_b))\n\n magnitude_spectrum_R = 20 * np.log(np.abs(f_shift_R))\n magnitude_spectrum_G = 20 * np.log(np.abs(f_shift_G))\n magnitude_spectrum_B = 20 * np.log(np.abs(f_shift_B))\n\n return r, g, b,R,G,B, magnitude_spectrum_r, magnitude_spectrum_g, magnitude_spectrum_b,magnitude_spectrum_R , magnitude_spectrum_G, magnitude_spectrum_B\n\ndef plot_results(r, g, b,R,G,B, magnitude_spectrum_r, magnitude_spectrum_g, magnitude_spectrum_b,magnitude_spectrum_R , magnitude_spectrum_G, magnitude_spectrum_B):\n plt.figure(figsize=(12, 6))\n\n plt.subplot(231), plt.imshow(r)\n plt.title('Original Red Channel'), plt.axis('off')\n\n plt.subplot(232), plt.imshow(g)\n plt.title('Original Green Channel'), plt.axis('off')\n\n plt.subplot(233), plt.imshow(b)\n plt.title('Original Blue Channel'), plt.axis('off')\n\n plt.subplot(234), plt.imshow(R)\n plt.title(' Red Channel High Pass'), plt.axis('off')\n\n plt.subplot(235), plt.imshow(G)\n plt.title(' Green Channel Highpass'), plt.axis('off')\n\n plt.subplot(236), plt.imshow(G)\n plt.title('Blue Channel Highpass'), plt.axis('off')\n\n plt.subplot(237), plt.imshow(magnitude_spectrum_r)\n plt.title('Magnitude Spectrum (Red)'), plt.axis('off')\n\n plt.subplot(238), plt.imshow(magnitude_spectrum_g)\n plt.title('Magnitude Spectrum (Green)'), plt.axis('off')\n\n plt.subplot(239), plt.imshow(magnitude_spectrum_b)\n plt.title('Magnitude Spectrum (Blue)'), plt.axis('off')\n\n plt.subplot(240), plt.imshow(magnitude_spectrum_R)\n plt.title('Magnitude Spectrum (Red Highpass)'), plt.axis('off')\n\n plt.subplot(241), plt.imshow(magnitude_spectrum_G)\n plt.title('Magnitude Spectrum (Green Highpass)'), plt.axis('off')\n\n plt.subplot(242), plt.imshow(magnitude_spectrum_B)\n plt.title('Magnitude Spectrum (Blue Highpass)'), plt.axis('off')\n\n plt.show()","metadata":{"id":"DOJWKpl09jOO","execution":{"iopub.status.busy":"2024-09-05T17:27:29.386202Z","iopub.execute_input":"2024-09-05T17:27:29.386609Z","iopub.status.idle":"2024-09-05T17:27:29.403591Z","shell.execute_reply.started":"2024-09-05T17:27:29.386575Z","shell.execute_reply":"2024-09-05T17:27:29.40253Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"image = cv2.imread('/kaggle/input/deepfakedet-p2/1000_videos/validation/fake/063_041_16.png')\nr,g,b,R,G,B,magnitude_spectrum_r,magnitude_spectrum_g,magnitude_spectrum_b,magnitude_spectrum_R , magnitude_spectrum_G, magnitude_spectrum_B = split_and_transform(image)\nplot_results(r,g,b,R,G,B,magnitude_spectrum_r,magnitude_spectrum_g,magnitude_spectrum_b,magnitude_spectrum_R , magnitude_spectrum_G, magnitude_spectrum_B)","metadata":{"id":"wNhagZmA-33U","outputId":"30a3b40d-91db-4218-99bc-22fd514bfa02","execution":{"iopub.status.busy":"2024-09-05T17:27:31.732527Z","iopub.execute_input":"2024-09-05T17:27:31.73294Z","iopub.status.idle":"2024-09-05T17:27:32.299049Z","shell.execute_reply.started":"2024-09-05T17:27:31.732906Z","shell.execute_reply":"2024-09-05T17:27:32.297664Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import cv2\nimport matplotlib.pyplot as plt\nx = high_frequency_representation_image(tf.expand_dims(image,axis=0))\nplt.imshow(tf.squeeze(x))","metadata":{"id":"vieR1m55t7Q4","outputId":"3b1a27d2-53f7-4e4f-9965-c7ca5610c8d7","execution":{"iopub.status.busy":"2024-09-05T15:02:49.096118Z","iopub.execute_input":"2024-09-05T15:02:49.097031Z","iopub.status.idle":"2024-09-05T15:02:49.64172Z","shell.execute_reply.started":"2024-09-05T15:02:49.096987Z","shell.execute_reply":"2024-09-05T15:02:49.640637Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Detect hardware, return appropriate distribution strategy\n\ntry:\n # TPU detection. No parameters necessary if TPU_NAME environment variable is set. On Kaggle this is always the case.\n tpu = tf.distribute.cluster_resolver.TPUClusterResolver() \n print('Running on TPU ', tpu.master())\nexcept ValueError:\n tpu = None\n\nif tpu:\n tf.config.experimental_connect_to_cluster(tpu)\n tf.tpu.experimental.initialize_tpu_system(tpu)\n strategy = tf.distribute.experimental.TPUStrategy(tpu)\nelse:\n strategy = tf.distribute.get_strategy() # default distribution strategy in Tensorflow. Works on CPU and single GPU.\n\nprint(\"REPLICAS: \", strategy.num_replicas_in_sync)","metadata":{"id":"casuJqiN9_ae","outputId":"70a68e50-044c-47aa-e2cd-c97894b049bc","execution":{"iopub.status.busy":"2024-09-05T14:56:47.354954Z","iopub.execute_input":"2024-09-05T14:56:47.356084Z","iopub.status.idle":"2024-09-05T14:56:52.544929Z","shell.execute_reply.started":"2024-09-05T14:56:47.356036Z","shell.execute_reply":"2024-09-05T14:56:52.544134Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def high_frequency_representation_feature(Mk, dim):\n if dim == 'spatial':\n freq_rep = tf.signal.fft2d(tf.cast(Mk, tf.complex64))\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-2, -1))\n H, W = Mk.shape[1], Mk.shape[2]\n high_pass_filter = np.ones((H, W))\n high_pass_filter[H//4:H*3//4, W//4:W*3//4] = 0\n high_pass_filter = tf.convert_to_tensor(high_pass_filter, dtype=tf.complex64)\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=-1)\n high_pass_rep = freq_rep * high_pass_filter\n high_freq_feature = tf.signal.ifftshift(high_pass_rep, axes=(-2, -1))\n high_freq_feature = tf.signal.ifft2d(high_freq_feature)\n elif dim == 'channel':\n freq_rep = tf.signal.fft(tf.cast(Mk, tf.complex64))\n freq_rep = tf.signal.fftshift(freq_rep, axes=(1,))\n C = Mk.shape[-1]\n high_pass_filter = np.ones((C,))\n high_pass_filter[C//4:C*3//4] = 0\n high_pass_filter = tf.convert_to_tensor(high_pass_filter, dtype=tf.complex64)\n high_pass_filter = tf.reshape(high_pass_filter, [1, 1, 1, C])\n high_pass_rep = freq_rep * high_pass_filter\n high_freq_feature = tf.signal.ifftshift(high_pass_rep, axes=(1,))\n high_freq_feature = tf.signal.ifft2d(high_freq_feature)\n else:\n raise ValueError(\"dim must be either 'spatial' or 'channel'\")\n\n return tf.math.real(high_freq_feature)\n","metadata":{"id":"WgtVAUEttT6u","execution":{"iopub.status.busy":"2024-09-06T04:53:09.92531Z","iopub.execute_input":"2024-09-06T04:53:09.926213Z","iopub.status.idle":"2024-09-06T04:53:09.940784Z","shell.execute_reply.started":"2024-09-06T04:53:09.926154Z","shell.execute_reply":"2024-09-06T04:53:09.939434Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from tensorflow.keras.layers import Conv2D\n\n#class FrequencyConvLayer(tf.keras.layers.Layer):\n # def __init__(self, filters, kernel_size=3, stride=1, padding='same'):\n # super(FrequencyConvLayer, self).__init__()\n # self.conv_amp = Conv2D(filters, kernel_size, stride, padding)\n # self.conv_phase = Conv2D(filters, kernel_size, stride, padding)\n\n # def call(self, x):\n # # Convert to frequency domain\n # freq_rep = tf.signal.fft2d(tf.cast(x, tf.complex64))\n # freq_rep = tf.signal.fftshift(freq_rep, axes=(-2, -1))\n\n # Split into amplitude and phase\n # amp = tf.abs(freq_rep)\n #phase = tf.math.angle(freq_rep)\n\n # Apply convolutions on amplitude and phase\n #amp_conv = self.conv_amp(amp)\n #phase_conv = self.conv_phase(phase)\n\n # Combine and convert back to spatial domain\n #combined_freq = tf.complex(amp_conv, phase_conv)\n #combined_freq = tf.signal.ifftshift(combined_freq, axes=(-2, -1))\n #out = tf.signal.ifft2d(combined_freq)\n\n #return tf.math.real(out)\n","metadata":{"id":"9gQkcRiGtT6w","execution":{"iopub.status.busy":"2024-09-05T15:03:00.558813Z","iopub.execute_input":"2024-09-05T15:03:00.559209Z","iopub.status.idle":"2024-09-05T15:03:00.592219Z","shell.execute_reply.started":"2024-09-05T15:03:00.55918Z","shell.execute_reply":"2024-09-05T15:03:00.591509Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from tensorflow.keras.layers import Dense, GlobalAveragePooling2D, Flatten\nfrom tensorflow.keras.models import Model\n\n#class FreqNet(Model):\n # def __init__(self, num_classes=2):\n # super(FreqNet, self).__init__()\n # self.hfri = high_frequency_representation_image\n # self.hfrf_blocks = [high_frequency_representation_feature for _ in range(3)]\n # self.fcl_blocks = [FrequencyConvLayer(64) for _ in range(3)]\n # self.global_pool = GlobalAveragePooling2D()\n # self.fc = Dense(num_classes, activation='softmax')\n\n #def call(self, x):\n # x = self.hfri(x)\n # for hfrf, fcl in zip(self.hfrf_blocks, self.fcl_blocks):\n # x = hfrf(x, dim='spatial') # Process spatial dimension; can also use 'channel'\n # x = fcl(x)\n # x = self.global_pool(x)\n # x = Flatten()(x)\n # out = self.fc(x)\n # return out\n","metadata":{"id":"Qy2YO2yatT6z","execution":{"iopub.status.busy":"2024-09-05T11:18:28.526999Z","iopub.execute_input":"2024-09-05T11:18:28.527645Z","iopub.status.idle":"2024-09-05T11:18:28.534213Z","shell.execute_reply.started":"2024-09-05T11:18:28.527608Z","shell.execute_reply":"2024-09-05T11:18:28.533297Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from tensorflow.keras.preprocessing.image import ImageDataGenerator\nbatch_size = 128\n\n# Path to the dataset\ndataset_path = '/kaggle/input/deepfakedet-p2/1000_videos/train'\nval_dataset_path = '/kaggle/input/deepfakedet-p2/1000_videos/validation'\n\n# Create an ImageDataGenerator instance\ndatagen = ImageDataGenerator(\n rescale=1./255, # Normalize pixel values to [0, 1]\n # Additional augmentation can be added here\n)\n\n# Create a data generator\ntrain_generator = datagen.flow_from_directory(\n dataset_path, # Path to the dataset\n target_size=(128, 128), # Resize images\n batch_size=batch_size, # Number of images per batch\n class_mode='binary' # Use 'categorical' for multi-class classification\n)\nval_generator = datagen.flow_from_directory(\n val_dataset_path, # Path to the dataset\n target_size=(128, 128), # Resize images\n batch_size=batch_size, # Number of images per batch\n class_mode='binary'\n)\n\n\nnan_image = []\nnan_label = []\n\n\n","metadata":{"id":"nut5SMl26b5V","outputId":"d7d8cc1e-2c45-48d8-edc7-c49141d68089","execution":{"iopub.status.busy":"2024-09-06T04:53:29.034281Z","iopub.execute_input":"2024-09-06T04:53:29.034766Z","iopub.status.idle":"2024-09-06T04:53:32.908342Z","shell.execute_reply.started":"2024-09-06T04:53:29.034719Z","shell.execute_reply":"2024-09-06T04:53:32.907033Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\n\nclass FCLBlock(tf.keras.layers.Layer):\n def __init__(self, filters, kernel_size=3):\n super(FCLBlock, self).__init__()\n\n # Convolutions for the Phase and Amplitude spectra\n self.phase_conv = tf.keras.layers.Conv2D(filters, kernel_size, padding='same')\n self.amplitude_conv = tf.keras.layers.Conv2D(filters, kernel_size, padding='same')\n\n # Frequency Conv Layer\n self.freq_conv = tf.keras.layers.Conv2D(filters, kernel_size, padding='same')\n\n def call(self, inputs):\n \n fft_output = tf.signal.fft2d(tf.cast(inputs, tf.complex64)) # Convert to complex64 for FFT\n\n # Step 1: Extract amplitude and phase from the FFT output\n amplitude = tf.abs(fft_output)\n phase = tf.math.angle(fft_output)\n\n # Step 2: Apply convolutions on phase and amplitude\n amplitude = self.amplitude_conv(amplitude) # No need for expand_dims\n phase = self.phase_conv(phase) # No need for expand_dims\n\n # Step 3: Recombine the amplitude and phase into a complex tensor\n processed_fft = tf.complex(amplitude, phase)\n\n # Step 4: Apply iFFT to bring data back to the spatial domain\n ifft_output = tf.signal.ifft2d(processed_fft)\n ifft_output = tf.math.real(ifft_output) # Take the real part only\n\n # Step 5: Apply the Frequency Conv Layer\n freq_output = self.freq_conv(ifft_output)\n\n # Step 7: Add residual connection\n output =freq_output\n\n return output\n\n# Test the FCL Block\ninput_tensor = tf.random.normal([1, 128, 128, 64]) # Example input [batch_size, height, width, channels]\nfcl_block = FCLBlock(filters=64) # Define the FCL Block\noutput_tensor = fcl_block(input_tensor) # Get the output\nfcl_block_1 = FCLBlock(filters=64)\nop = fcl_block_1(output_tensor)\n\nprint(\"Output shape:\", op.shape)\n","metadata":{"id":"IN4JI6pytYrH","outputId":"feed3ce5-3448-4aad-870c-137600384871","execution":{"iopub.status.busy":"2024-09-06T04:53:56.899874Z","iopub.execute_input":"2024-09-06T04:53:56.900849Z","iopub.status.idle":"2024-09-06T04:53:57.421378Z","shell.execute_reply.started":"2024-09-06T04:53:56.900791Z","shell.execute_reply":"2024-09-06T04:53:57.419434Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\n\nclass ResidualConvBlock(tf.keras.layers.Layer):\n def __init__(self, filters, kernel_size=3, downsample=False):\n super(ResidualConvBlock, self).__init__()\n self.filters = filters\n self.kernel_size = kernel_size\n self.downsample = downsample\n\n # Convolutional layers, batch normalization, and activations\n self.conv1 = tf.keras.layers.Conv2D(filters, kernel_size, padding='same')\n self.bn1 = tf.keras.layers.BatchNormalization()\n self.relu1 = tf.keras.layers.ReLU()\n\n self.conv2 = tf.keras.layers.Conv2D(filters, kernel_size, padding='same')\n self.bn2 = tf.keras.layers.BatchNormalization()\n\n # Skip connection layers (for downsampling or channel mismatch)\n self.conv_skip = None\n self.bn_skip = None\n\n def build(self, input_shape):\n # Check if input and output shapes need adjustment for residual connection\n if self.downsample or input_shape[-1] != self.filters:\n # Apply 1x1 convolution to adjust dimensions\n self.conv_skip = tf.keras.layers.Conv2D(self.filters, kernel_size=1, padding='same')\n self.bn_skip = tf.keras.layers.BatchNormalization()\n\n def call(self, inputs, training=False):\n # Save input for residual connection\n x = inputs\n\n # First convolution block\n out = self.conv1(inputs)\n out = self.bn1(out, training=training)\n out = self.relu1(out)\n\n # Second convolution block\n out = self.conv2(out)\n out = self.bn2(out, training=training)\n out = self.relu1(out)\n\n # Adjust input (skip connection) if downsampling or channel mismatch occurs\n if self.conv_skip is not None:\n x = self.conv_skip(x)\n x = self.bn_skip(x, training=training)\n\n # Add the skip connection (input) to the output\n out = out + x\n \n\n # Final activation\n out = tf.nn.relu(out)\n\n return out\n\n# Test the Residual Conv Block\ninput_tensor = tf.random.normal([1, 64, 64, 128]) # Example input [batch_size, height, width, channels]\nres_block = ResidualConvBlock(filters=64, downsample=False) # Define a residual block\noutput_tensor = res_block(input_tensor) # Pass input tensor through the residual block\n\nprint(\"Output shape:\", output_tensor.shape)\n","metadata":{"id":"OwJj1_E1wJvZ","outputId":"f0809757-2168-4aef-cf17-fd1162d496f3","execution":{"iopub.status.busy":"2024-09-06T04:53:58.211964Z","iopub.execute_input":"2024-09-06T04:53:58.21245Z","iopub.status.idle":"2024-09-06T04:53:58.339056Z","shell.execute_reply.started":"2024-09-06T04:53:58.212404Z","shell.execute_reply":"2024-09-06T04:53:58.337657Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\n\nclass HighFrequencyImageLayer(tf.keras.layers.Layer):\n def __init__(self):\n super(HighFrequencyImageLayer, self).__init__()\n\n def create_high_pass_filter(self, shape):\n \"\"\"\n Creates a high-pass filter that zeros out a specific region in the frequency domain.\n \n This filter zeros out the low-frequency components based on the scale factor.\n \n Parameters:\n shape (tuple): The shape of the filter (H, W)\n \n Returns:\n high_pass_filter (Tensor): The high-pass filter of shape (H, W)\n \"\"\"\n scale = 4\n H, W = shape\n # Create a grid of indices\n i = tf.range(H, dtype=tf.float32)\n j = tf.range(W, dtype=tf.float32)\n I, J = tf.meshgrid(i, j, indexing='ij')\n # Calculate the center of the image\n h_center, w_center = H // 2, W // 2\n h_start, h_end = h_center - H // scale, h_center + H // scale\n w_start, w_end = w_center - W // scale, w_center + W // scale\n high_pass_filter = tf.ones((H, W), dtype=tf.float32)\n # Cast bounds to float32 to match the types of I and J\n h_start = tf.cast(h_start, tf.float32)\n h_end = tf.cast(h_end, tf.float32)\n w_start = tf.cast(w_start, tf.float32)\n w_end = tf.cast(w_end, tf.float32)\n low_freq_mask = tf.logical_and(\n tf.logical_and(I >= h_start, I < h_end),\n tf.logical_and(J >= w_start, J < w_end)\n )\n\n high_pass_filter = tf.where(low_freq_mask, 0.0, high_pass_filter)\n return high_pass_filter\n\n def call(self, inputs):\n \"\"\"\n Applies the high-pass filter to an image in the frequency domain.\n\n Parameters:\n inputs (Tensor): The input image of shape (B, H, W, C)\n\n Returns:\n high_freq_image (Tensor): The high-frequency component of the image\n \"\"\"\n # Convert the image to frequency domain using FFT\n freq_rep = tf.signal.fft2d(tf.cast(inputs, tf.complex64))\n\n # Shift zero-frequency component to the center\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-2, -1))\n\n # Get the shape of the image and create the high-pass filter\n B, H, W, C = tf.shape(inputs)[0], tf.shape(inputs)[1], tf.shape(inputs)[2], tf.shape(inputs)[3]\n high_pass_filter = self.create_high_pass_filter((H, W))\n\n # Apply the high-pass filter to each channel\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=0) # for batch\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=-1) # for channels\n high_pass_filter = tf.cast(high_pass_filter, tf.complex64)\n\n # Multiply the frequency representation by the high-pass filter\n high_freq_rep = freq_rep * high_pass_filter\n\n # Convert back to image domain using iFFT\n high_freq_rep = tf.signal.ifftshift(high_freq_rep, axes=(-2, -1))\n high_freq_image = tf.signal.ifft2d(high_freq_rep)\n \n high_freq_image = tf.math.real(high_freq_image)\n \n high_freq_image = tf.keras.layers.ReLU()(high_freq_image)\n \n \n\n # Return the real part of the image\n return high_freq_image\n\nclass HighFrequencyFeatureLayer(tf.keras.layers.Layer):\n def __init__(self, dim='spatial'):\n super(HighFrequencyFeatureLayer, self).__init__()\n self.dim = dim\n\n def create_high_pass_filter(self, shape):\n \"\"\"\n Creates a high-pass filter that zeros out low-frequency components.\n\n Parameters:\n shape (tuple): The shape of the filter (H, W) or (C,)\n\n Returns:\n high_pass_filter (Tensor): The high-pass filter of shape (H, W) or (C,)\n \"\"\"\n if len(shape) == 2:\n scale = 4\n H, W = shape\n # Create a grid of indices\n i = tf.range(H, dtype=tf.float32)\n j = tf.range(W, dtype=tf.float32)\n I, J = tf.meshgrid(i, j, indexing='ij')\n # Calculate the center of the image\n h_center, w_center = H // 2, W // 2\n h_start, h_end = h_center - H // scale, h_center + H // scale\n w_start, w_end = w_center - W // scale, w_center + W // scale\n # Cast bounds to float32 to match the types of I and J\n h_start = tf.cast(h_start, tf.float32)\n h_end = tf.cast(h_end, tf.float32)\n w_start = tf.cast(w_start, tf.float32)\n w_end = tf.cast(w_end, tf.float32)\n high_pass_filter = tf.ones((H, W), dtype=tf.float32)\n low_freq_mask = tf.logical_and(\n tf.logical_and(I >= h_start, I < h_end),\n tf.logical_and(J >= w_start, J < w_end)\n )\n\n high_pass_filter = tf.where(low_freq_mask, 0.0, high_pass_filter)\n return high_pass_filter\n elif len(shape) == 1:\n C = shape[0]\n scale =4\n # Create a ones tensor for the entire channel axis\n high_pass_filter = tf.ones((C,), dtype=tf.float32)\n\n # Determine the center and the range of channels to zero out\n c_center = C // 2\n c_start, c_end = c_center - C // scale, c_center + C // scale\n\n # Create a mask to zero out the middle channels\n low_freq_mask = tf.logical_and(tf.range(C) >= c_start, tf.range(C) < c_end)\n\n # Apply the mask to set channels to 0 where the condition is true\n high_pass_filter = tf.where(low_freq_mask, 0.0, high_pass_filter)\n\n\n return high_pass_filter\n else:\n raise ValueError(\"Invalid shape for high-pass filter\")\n\n def call(self, inputs):\n \"\"\"\n Applies the high-pass filter to a feature map in the frequency domain.\n\n Parameters:\n inputs (Tensor): The input feature map of shape (B, H, W, C)\n dim (str): The dimension to apply the FFT ('spatial' or 'channel')\n\n Returns:\n high_freq_feature (Tensor): The high-frequency component of the feature map\n \"\"\"\n if self.dim == 'spatial':\n freq_rep = tf.signal.fft2d(tf.cast(inputs, tf.complex64))\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-2, -1))\n H, W = tf.shape(inputs)[1], tf.shape(inputs)[2]\n high_pass_filter = self.create_high_pass_filter((H, W))\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=-1) # for channels\n high_pass_filter = tf.expand_dims(high_pass_filter, axis=0) # for batch\n high_pass_filter = tf.cast(high_pass_filter, tf.complex64)\n high_pass_rep = freq_rep * high_pass_filter\n high_freq_feature = tf.signal.ifftshift(high_pass_rep, axes=(-2, -1))\n high_freq_feature = tf.signal.ifft2d(high_freq_feature)\n \n high_freq_image = tf.math.real(high_freq_feature)\n \n high_freq_image = tf.keras.layers.ReLU()(high_freq_image)\n \n\n elif self.dim == 'channel':\n freq_rep = tf.signal.fft2d(tf.cast(inputs, tf.complex64))\n freq_rep = tf.signal.fftshift(freq_rep, axes=(-2, -1))\n C = tf.shape(inputs)[-1]\n high_pass_filter = self.create_high_pass_filter((C,))\n high_pass_filter = tf.reshape(high_pass_filter, [1, 1, 1, C])\n high_pass_filter = tf.cast(high_pass_filter, tf.complex64)\n high_pass_rep = freq_rep * high_pass_filter\n high_freq_feature = tf.signal.ifftshift(high_pass_rep, axes=(-2, -1))\n high_freq_feature = tf.signal.ifft2d(high_freq_feature)\n \n high_freq_image = tf.math.real(high_freq_feature)\n \n high_freq_image = tf.keras.layers.ReLU()(high_freq_image)\n\n else:\n raise ValueError(\"dim must be either 'spatial' or 'channel'\")\n\n return high_freq_image\n# Test the layers inside a simple model\n\n\n","metadata":{"id":"faQUk7QD0mGv","outputId":"55bea51a-d1e6-4aaf-8b3e-4617a6eb9f4d","execution":{"iopub.status.busy":"2024-09-06T04:54:00.243752Z","iopub.execute_input":"2024-09-06T04:54:00.244307Z","iopub.status.idle":"2024-09-06T04:54:00.287267Z","shell.execute_reply.started":"2024-09-06T04:54:00.244259Z","shell.execute_reply":"2024-09-06T04:54:00.285654Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# IMPLEMENT FREPNET\ndef create_freqnet():\n input_layer = tf.keras.layers.Input(shape=(128, 128, 3))\n ### HFRI\n x = HighFrequencyImageLayer()(input_layer)\n x = tf.keras.layers.Conv2D(32,(3,3),activation=\"relu\")(x)\n \n ### HFRFC\n x = HighFrequencyFeatureLayer('channel')(x)\n \n ### FCL\n x = FCLBlock(filters=64)(x)\n x = tf.keras.layers.ReLU()(x)\n \n ### HFRFS\n x = HighFrequencyFeatureLayer('spatial')(x)\n x = tf.keras.layers.Conv2D(64,(3,3),strides=2,activation=\"relu\")(x)\n \n ### HFRFC\n x = HighFrequencyFeatureLayer('channel')(x)\n \n ###FCL\n x = FCLBlock(filters=64)(x)\n x = tf.keras.layers.ReLU()(x)\n \n ### RESBLOCK\n x = ResidualConvBlock(filters=64)(x)\n \n ### HFRFS\n x = HighFrequencyFeatureLayer('spatial')(x)\n x = tf.keras.layers.Conv2D(64,(3,3),strides=2,activation=\"relu\")(x)\n \n ### FCL\n x = FCLBlock(filters=256)(x)\n x = tf.keras.layers.ReLU()(x)\n \n ### HFRFS\n x = HighFrequencyFeatureLayer('spatial')(x)\n x = tf.keras.layers.Conv2D(64,(3,3),strides=2,activation=\"relu\")(x)\n \n ### FCL\n x = FCLBlock(filters=256)(x)\n x = tf.keras.layers.ReLU()(x)\n\n \n ### RESBLOCK\n x = ResidualConvBlock(filters=64)(x)\n \n ### GLOBAL AVG POOLING\n x = tf.keras.layers.GlobalAveragePooling2D()(x)\n \n x = tf.keras.layers.Dense(512, activation='relu')(x)\n output_layer = tf.keras.layers.Dense(1, activation='sigmoid')(x)\n\n model = tf.keras.Model(inputs=input_layer, outputs=output_layer)\n\n return model","metadata":{"id":"n1blITW1GoVC","execution":{"iopub.status.busy":"2024-09-06T04:54:01.687365Z","iopub.execute_input":"2024-09-06T04:54:01.687801Z","iopub.status.idle":"2024-09-06T04:54:01.703452Z","shell.execute_reply.started":"2024-09-06T04:54:01.687761Z","shell.execute_reply":"2024-09-06T04:54:01.701888Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\n\n# detect and init the TPU\ntpu = tf.distribute.cluster_resolver.TPUClusterResolver()\nprint('Running on TPU ', tpu.master())\n\n# instantiate a distribution strategy\ntf.tpu.experimental.initialize_tpu_system(tpu)\ntpu_strategy = tf.distribute.TPUStrategy(tpu)\n\n# instantiating the model in the strategy scope creates the model on the TPU\nwith tpu_strategy.scope():\n model = create_freqnet()\n model.compile(optimizer=tf.keras.optimizers.Adam(),\n loss='binary_crossentropy',\n metrics=['accuracy'])\n\n","metadata":{"id":"B2GwzCmbtT60","execution":{"iopub.status.busy":"2024-09-05T17:27:56.966288Z","iopub.execute_input":"2024-09-05T17:27:56.966694Z","iopub.status.idle":"2024-09-05T17:28:05.975447Z","shell.execute_reply.started":"2024-09-05T17:27:56.96666Z","shell.execute_reply":"2024-09-05T17:28:05.974365Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"REPLICAS: \", tpu_strategy.num_replicas_in_sync)","metadata":{"execution":{"iopub.status.busy":"2024-09-05T17:28:14.128076Z","iopub.execute_input":"2024-09-05T17:28:14.128483Z","iopub.status.idle":"2024-09-05T17:28:14.133384Z","shell.execute_reply.started":"2024-09-05T17:28:14.128442Z","shell.execute_reply":"2024-09-05T17:28:14.132549Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#model.compile(optimizer=tf.keras.optimizers.Adam(),\n # loss='binary_crossentropy',\n # metrics=['accuracy'])\n","metadata":{"execution":{"iopub.status.busy":"2024-09-05T15:27:52.388371Z","iopub.execute_input":"2024-09-05T15:27:52.388825Z","iopub.status.idle":"2024-09-05T15:27:52.408166Z","shell.execute_reply.started":"2024-09-05T15:27:52.388784Z","shell.execute_reply":"2024-09-05T15:27:52.407177Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.summary()","metadata":{"id":"UVemg0if7hQ9","outputId":"20364c5b-ef16-4a3e-cfc2-a4ded43b254b","execution":{"iopub.status.busy":"2024-09-05T16:03:56.985971Z","iopub.execute_input":"2024-09-05T16:03:56.986851Z","iopub.status.idle":"2024-09-05T16:03:57.019225Z","shell.execute_reply.started":"2024-09-05T16:03:56.986816Z","shell.execute_reply":"2024-09-05T16:03:57.018412Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.predict(tf.expand_dims(image,axis=0))","metadata":{"id":"PbfRODXeRFpe","outputId":"fb57b6ab-c783-4a3a-b649-0c62dcd88aa4","execution":{"iopub.status.busy":"2024-09-05T15:05:55.209233Z","iopub.execute_input":"2024-09-05T15:05:55.209598Z","iopub.status.idle":"2024-09-05T15:05:55.241562Z","shell.execute_reply.started":"2024-09-05T15:05:55.209565Z","shell.execute_reply":"2024-09-05T15:05:55.240343Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class CustomModelCheckpoint(tf.keras.callbacks.Callback):\n def __init__(self, filepath, save_freq=5):\n super(CustomModelCheckpoint, self).__init__()\n self.filepath = filepath\n self.save_freq = save_freq\n\n def on_epoch_end(self, epoch, logs=None):\n if (epoch + 1) % self.save_freq == 0:\n self.model.save(self.filepath.format(epoch + 1))\n print(f\"Model saved at epoch {epoch + 1} to {self.filepath.format(epoch + 1)}\")\n\n#def lr_scheduler(epoch, lr):\n # if epoch > 0 and epoch % 10 == 0:\n # return lr * 0.8\n # return lr","metadata":{"execution":{"iopub.status.busy":"2024-09-05T17:28:18.470164Z","iopub.execute_input":"2024-09-05T17:28:18.470572Z","iopub.status.idle":"2024-09-05T17:28:18.476296Z","shell.execute_reply.started":"2024-09-05T17:28:18.470535Z","shell.execute_reply":"2024-09-05T17:28:18.475376Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Compile model\n#model.compile(optimizer=tf.keras.optimizers.Adam(),\n # loss='binary_crossentropy',\n # metrics=['accuracy'])\n\n# Define learning rate scheduler\nearly_stopping_callback = tf.keras.callbacks.EarlyStopping(\n monitor='val_loss', # Metric to monitor\n min_delta=0.001, # Minimum change to qualify as an improvement\n patience=10, # Number of epochs to wait for improvement\n verbose=1, # Verbosity mode\n restore_best_weights=True # Restore model weights from the epoch with the best value of the monitored metric\n)\n# Create the ReduceLROnPlateau callback\nreduce_lr_callback = tf.keras.callbacks.ReduceLROnPlateau(\n monitor='val_loss',\n factor=0.5,\n patience=10,\n min_lr=1e-6\n)\ncheckpoint_callback = CustomModelCheckpoint(filepath='/kaggle/working/model_epoch_{:02d}.h5', save_freq=5)\n\n\n# Train model\nhistory=model.fit(train_generator, epochs=100, callbacks=[reduce_lr_callback,early_stopping_callback,checkpoint_callback], validation_data=val_generator)","metadata":{"id":"9iGnLDnuBdoL","outputId":"dba38a1a-57a1-4814-c415-8d5bceb5dd4a","execution":{"iopub.status.busy":"2024-09-05T17:28:30.225438Z","iopub.execute_input":"2024-09-05T17:28:30.226018Z","iopub.status.idle":"2024-09-05T17:31:38.759843Z","shell.execute_reply.started":"2024-09-05T17:28:30.225951Z","shell.execute_reply":"2024-09-05T17:31:38.75855Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Add data augmentation to prevent overfitting","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:56.393301Z","iopub.execute_input":"2024-09-04T05:24:56.39361Z","iopub.status.idle":"2024-09-04T05:24:56.423705Z","shell.execute_reply.started":"2024-09-04T05:24:56.393578Z","shell.execute_reply":"2024-09-04T05:24:56.42282Z"},"id":"8djtzv0AtT64","outputId":"b829160d-a39d-4185-c3c0-b24621a955e4"}},{"cell_type":"code","source":"from tensorflow.keras.preprocessing.image import ImageDataGenerator\n\n# Path to the dataset\ndataset_path = '/kaggle/input/deepfakedet-p2/1000_videos/train'\nval_dataset_path = '/kaggle/input/deepfakedet-p2/1000_videos/validation'\n\n# Create an ImageDataGenerator instance with augmentation for training\ntrain_datagen = ImageDataGenerator(\n rescale=1./255, # Normalize pixel values to [0, 1]\n rotation_range=20, # Random rotation between 0-20 degrees # Random vertical shift\n shear_range=0.2, # Shear transformations\n zoom_range=0.2, # Random zoom\n horizontal_flip=True # Random horizontal flip # Fill missing pixels after transformations\n)\n\n# Create a separate ImageDataGenerator for validation (no augmentation)\nval_datagen = ImageDataGenerator(rescale=1./255) # Only normalization\n\n# Create a data generator for training with augmentation\ntrain_generator = train_datagen.flow_from_directory(\n dataset_path, # Path to the dataset\n target_size=(128, 128), # Resize images\n batch_size=16, # Number of images per batch\n class_mode='binary', # Use 'categorical' for multi-class classification\n shuffle=True # Shuffle the dataset\n)\n\n# Create a data generator for validation without augmentation\nval_generator = val_datagen.flow_from_directory(\n val_dataset_path, # Path to the dataset\n target_size=(128, 128), # Resize images\n batch_size=16, # Number of images per batch\n class_mode='binary', # Use 'categorical' for multi-class classification\n shuffle=False # No need to shuffle validation data\n)\n","metadata":{"execution":{"iopub.status.busy":"2024-09-05T12:23:24.751494Z","iopub.execute_input":"2024-09-05T12:23:24.751998Z","iopub.status.idle":"2024-09-05T12:23:27.623363Z","shell.execute_reply.started":"2024-09-05T12:23:24.75196Z","shell.execute_reply":"2024-09-05T12:23:27.622445Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Create model\nmodel_1 = create_freqnet()\nmodel_1.load_weights('/kaggle/input/frepnet/other/var1/1/model_epoch_20.h5')","metadata":{"execution":{"iopub.status.busy":"2024-09-06T05:52:37.808558Z","iopub.execute_input":"2024-09-06T05:52:37.808993Z","iopub.status.idle":"2024-09-06T05:52:38.995541Z","shell.execute_reply.started":"2024-09-06T05:52:37.808949Z","shell.execute_reply":"2024-09-06T05:52:38.994314Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"img = tf.expand_dims(cv2.imread('/kaggle/input/deepfakedet-p2/1000_videos/test/fake/068_028_9.png'),axis=0)","metadata":{"execution":{"iopub.status.busy":"2024-09-06T05:52:40.018667Z","iopub.execute_input":"2024-09-06T05:52:40.019162Z","iopub.status.idle":"2024-09-06T05:52:40.029588Z","shell.execute_reply.started":"2024-09-06T05:52:40.019115Z","shell.execute_reply":"2024-09-06T05:52:40.028234Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom tensorflow.keras.preprocessing import image\n\ndef load_and_preprocess_image(img_path, target_size=(128, 128)):\n # Load the image from the file path\n img = image.load_img(img_path, target_size=target_size)\n \n # Convert the image to a numpy array\n img_array = image.img_to_array(img)\n \n # Expand dimensions to match model input (batch_size, height, width, channels)\n img_array = np.expand_dims(img_array, axis=0)\n \n # Normalize the image array (assuming the model expects values between 0 and 1)\n img_array /= 255.0\n \n return img_array\n\ndef predict_image(model, img_array):\n # Predict the class (assuming binary classification: 0 = real, 1 = fake)\n predictions = model.predict(img_array)\n \n # Convert predictions to scalar if it's still an array\n confidence = float(predictions[0]) # Convert to a scalar value\n \n # Assuming model outputs a single prediction (e.g., sigmoid activation for binary classification)\n predicted_class = 1 if confidence > 0.5 else 0\n \n return predicted_class, confidence\n\ndef plot_images_with_predictions(model, img_paths):\n # Create a figure for displaying images\n plt.figure(figsize=(10, 10))\n \n # Loop through the image paths\n for i, img_path in enumerate(img_paths):\n # Load and preprocess the image\n img_array = load_and_preprocess_image(img_-path)\n \n # Predict the class and confidence\n predicted_class, confidence = predict_image(model, img_array)\n confidence = 1-confidence\n \n # Load the original image for display\n img = image.load_img(img_path)\n \n # Create a subplot for each image\n plt.subplot(2, 2, i + 1)\n plt.imshow(img)\n plt.axis('off')\n \n # Display prediction result on the image\n label = \"Real\" if predicted_class == 1 else \"Fake\"\n plt.title(f\"Prediction: {label}\\nConfidence: {confidence*100:.2f}%\")\n \n # Show the plot with 4 images\n plt.tight_layout()\n plt.show()\n\n# Example usage (assuming you have a trained model and 4 image paths):\n# model = your_trained_model\n# img_paths = [\"image1.jpg\", \"image2.jpg\", \"image3.jpg\", \"image4.jpg\"]\n# plot_images_with_predictions(model, img_paths)\n\n\n\nimg_paths = [\"/kaggle/input/deepfakedet-p2/1000_videos/test/fake/080_061_17.png\", \"/kaggle/input/deepfakedet-p2/1000_videos/train/fake/128_896_24.png\", \"/kaggle/input/deepfakedet-p2/1000_videos/train/fake/130_139_17.png\",'/kaggle/input/deepfakedet-p2/1000_videos/train/fake/132_007_9.png']\nplot_images_with_predictions(model_1, img_paths)\n\n\n\n","metadata":{"execution":{"iopub.status.busy":"2024-09-06T07:34:44.793809Z","iopub.execute_input":"2024-09-06T07:34:44.794812Z","iopub.status.idle":"2024-09-06T07:34:46.445719Z","shell.execute_reply.started":"2024-09-06T07:34:44.794758Z","shell.execute_reply":"2024-09-06T07:34:46.444265Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"!pip install dlib","metadata":{"execution":{"iopub.status.busy":"2024-09-06T06:52:32.61545Z","iopub.execute_input":"2024-09-06T06:52:32.615994Z","iopub.status.idle":"2024-09-06T07:02:19.674009Z","shell.execute_reply.started":"2024-09-06T06:52:32.615945Z","shell.execute_reply":"2024-09-06T07:02:19.672342Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import cv2\nimport numpy as np\nimport dlib\nfrom keras.preprocessing.image import img_to_array\nfrom keras.models import load_model\n\ndef prediction_on_video(model, video_path, output_video_path):\n # Set up video capture and output writer\n cap = cv2.VideoCapture(video_path)\n frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))\n frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))\n fps = cap.get(cv2.CAP_PROP_FPS)\n \n # Video writer to save the output video\n out = cv2.VideoWriter(output_video_path, cv2.VideoWriter_fourcc(*'XVID'), fps, (frame_width, frame_height))\n \n # Initialize face detector\n detector = dlib.get_frontal_face_detector()\n \n while cap.isOpened():\n ret, frame = cap.read()\n if not ret:\n break\n \n # Detect faces in the frame\n face_rects = detector(frame, 1)\n \n for rect in face_rects:\n x1 = rect.left()\n y1 = rect.top()\n x2 = rect.right()\n y2 = rect.bottom()\n \n # Crop and resize the face region\n crop_img = frame[y1:y2, x1:x2]\n if crop_img.size == 0:\n continue\n \n data = img_to_array(cv2.resize(crop_img, (128, 128))).flatten() / 255.0\n data = data.reshape(-1, 128, 128, 3)\n \n # Predict if the face is REAL or FAKE\n # Convert predictions to scalar if it's still an array\n prediction = model.predict(data)\n confidence = float(prediction[0]) # Convert to a scalar value\n \n # Assuming model outputs a single prediction (e.g., sigmoid activation for binary classification)\n predicted = 1 if confidence > 0.5 else 0\n label = \"REAL\" if predicted == 1 else \"FAKE\"\n \n # Draw the bounding box and label on the frame\n color = (0, 0, 255) if label == \"FAKE\" else (0, 255, 0)\n cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)\n cv2.putText(frame, label, (x1, y1 - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.9, color, 2)\n \n # Write the modified frame to the output video\n out.write(frame)\n \n # Release video capture and writer\n cap.release()\n out.release()\n print(f\"Video saved to {output_video_path}\")\n\n# Path to the input and output video files\nvideo_path = '/kaggle/input/deepfake-detection-challenge/train_sample_videos/aettqgevhz.mp4'\noutput_video_path = '/kaggle/working/processed_video_xyz_1.avi' # Specify full path including filename\n\n\n# Process the video and save the result\nprediction_on_video(model_1, video_path, output_video_path)","metadata":{"execution":{"iopub.status.busy":"2024-09-06T07:22:08.5239Z","iopub.execute_input":"2024-09-06T07:22:08.527357Z","iopub.status.idle":"2024-09-06T07:31:20.024526Z","shell.execute_reply.started":"2024-09-06T07:22:08.527296Z","shell.execute_reply":"2024-09-06T07:31:20.023162Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from IPython.display import HTML\nfrom base64 import b64encode\n\ndef play(filename):\n html = ''\n video = open(filename, 'rb').read()\n src = 'data:video/avi;base64,' + b64encode(video).decode()\n html += '' % src \n return HTML(html)\n\nplay('/kaggle/working/processed_video_xyz_1.avi')\n","metadata":{"execution":{"iopub.status.busy":"2024-09-06T07:33:36.820898Z","iopub.execute_input":"2024-09-06T07:33:36.821855Z","iopub.status.idle":"2024-09-06T07:33:37.20133Z","shell.execute_reply.started":"2024-09-06T07:33:36.821752Z","shell.execute_reply":"2024-09-06T07:33:37.19921Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import json\n\ndef read_metadata(file_path):\n # Open and read the JSON file\n with open(file_path, 'r') as f:\n metadata = json.load(f) # Load the content as a dictionary\n \n # Return the parsed JSON data\n return metadata\n\n# Example usage\nfile_path = '/kaggle/input/deepfake-detection-challenge/train_sample_videos/metadata.json' # Adjust the file path if needed\nmetadata = read_metadata(file_path)\n\n# Printing metadata to check the structure\nprint(json.dumps(metadata, indent=4)) # Pretty print JSON with indentation\n","metadata":{"execution":{"iopub.status.busy":"2024-09-06T07:21:22.152278Z","iopub.execute_input":"2024-09-06T07:21:22.152892Z","iopub.status.idle":"2024-09-06T07:21:22.178789Z","shell.execute_reply.started":"2024-09-06T07:21:22.152839Z","shell.execute_reply":"2024-09-06T07:21:22.177508Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom tensorflow.keras.preprocessing import image\nfrom tensorflow.keras.models import Model\n\ndef load_and_preprocess_image(img_path, target_size=(128, 128)):\n # Load the image from the file path\n img = image.load_img(img_path, target_size=target_size)\n \n # Convert the image to a numpy array\n img_array = image.img_to_array(img)\n \n # Expand dimensions to match model input (batch_size, height, width, channels)\n img_array = np.expand_dims(img_array, axis=0)\n \n # Normalize the image array (assuming the model expects values between 0 and 1)\n img_array /= 255.0\n \n return img_array\n\ndef create_feature_extractor(model):\n # Extract the output of the last layer before the final prediction\n feature_extractor = Model(inputs=model.input, outputs=model.layers[-4].output)\n return feature_extractor\n\ndef get_feature_map(feature_extractor, img_array):\n # Pass the image through the feature extractor model to get features\n feature_map = feature_extractor.predict(img_array)\n # Print shape for debugging\n print(\"Feature map shape:\", feature_map.shape)\n return feature_map\n\ndef plot_feature_map(feature_map):\n # Assuming the feature map is 2D or 3D (height, width, channels)\n num_features = feature_map.shape[-1] # Number of feature maps/channels\n size = int(np.sqrt(num_features)) # Create a grid based on number of features\n \n # Create a grid plot of feature maps\n plt.figure(figsize=(12, 12))\n for i in range(num_features):\n plt.subplot(size, size, i + 1)\n feature = feature_map[0, :, :, i]\n plt.imshow(feature, cmap='viridis')\n plt.axis('off')\n \n plt.tight_layout()\n plt.show()\n\n# Example of combining the functions\ndef visualize_features_from_last_layer(model, img_path):\n # Load and preprocess the image\n img_array = load_and_preprocess_image(img_path)\n \n # Create a feature extractor from the model\n feature_extractor = create_feature_extractor(model)\n \n # Get the feature map from the last layer\n feature_map = get_feature_map(feature_extractor, img_array)\n \n # Plot the feature map\n plot_feature_map(feature_map)\n\n# Example usage (assuming you have a trained model and an image path):\n# model = your_trained_model\nimg_path = \"/kaggle/input/deepfakedet-p2/1000_videos/test/fake/067_025_6.png\"\nvisualize_features_from_last_layer(model_1, img_path)\n","metadata":{"execution":{"iopub.status.busy":"2024-09-06T05:59:09.569144Z","iopub.execute_input":"2024-09-06T05:59:09.569614Z","iopub.status.idle":"2024-09-06T05:59:13.507743Z","shell.execute_reply.started":"2024-09-06T05:59:09.569568Z","shell.execute_reply":"2024-09-06T05:59:13.50624Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import numpy as np\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom sklearn.metrics import confusion_matrix\nfrom tensorflow.keras.models import load_model\n\n# detect and init the TPU\ntpu = tf.distribute.cluster_resolver.TPUClusterResolver()\nprint('Running on TPU ', tpu.master())\n\n# instantiate a distribution strategy\ntf.tpu.experimental.initialize_tpu_system(tpu)\ntpu_strategy = tf.distribute.TPUStrategy(tpu)\n\n# instantiating the model in the strategy scope creates the model on the TPU\nwith tpu_strategy.scope():\n\n # Create model\n model_1 = create_freqnet()\n model_1.load_weights('/kaggle/input/frepnet/other/var1/1/model_epoch_20.h5')\n # 3. Make predictions\n y_pred = model_1.predict(train_generator, steps=train_generator.samples // train_generator.batch_size + 1)\n y_pred = (y_pred > 0.5).astype(int) # Convert probabilities to binary labels\n\n # 4. Get true labels\n y_true = train_generator.classes\n\n # 5. Compute confusion matrix\n cm = confusion_matrix(y_true, y_pred)\n\n # 6. Plot confusion matrix\n plt.figure(figsize=(10, 7))\n sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', \n xticklabels=train_generator.class_indices.keys(), \n yticklabels=train_generator.class_indices.keys())\n plt.xlabel('Predicted labels')\n plt.ylabel('True labels')\n plt.title('Confusion Matrix')\n plt.show()","metadata":{"execution":{"iopub.status.busy":"2024-09-05T15:45:46.449274Z","iopub.execute_input":"2024-09-05T15:45:46.44965Z","iopub.status.idle":"2024-09-05T15:47:38.179233Z","shell.execute_reply.started":"2024-09-05T15:45:46.44962Z","shell.execute_reply":"2024-09-05T15:47:38.177995Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":" # 5. Compute confusion matrix\ncm = confusion_matrix(y_true, y_pred)\n\n# 6. Plot confusion matrix\nplt.figure(figsize=(10, 7))\nsns.heatmap(cm, annot=True, fmt='d', cmap='Blues', \n xticklabels=train_generator.class_indices.keys(), \n yticklabels=train_generator.class_indices.keys())\nplt.xlabel('Predicted labels')\nplt.ylabel('True labels')\nplt.title('Confusion Matrix')\nplt.show()","metadata":{"execution":{"iopub.status.busy":"2024-09-05T15:48:25.496629Z","iopub.execute_input":"2024-09-05T15:48:25.497111Z","iopub.status.idle":"2024-09-05T15:48:25.651897Z","shell.execute_reply.started":"2024-09-05T15:48:25.497049Z","shell.execute_reply":"2024-09-05T15:48:25.650955Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"checkpoint_callback = CustomModelCheckpoint(filepath='/kaggle/working/model_epoch_{:02d}.h5', save_freq=5)\n\nlr_callback = tf.keras.callbacks.LearningRateScheduler(lr_scheduler)","metadata":{"execution":{"iopub.status.busy":"2024-09-05T12:29:11.000212Z","iopub.execute_input":"2024-09-05T12:29:11.000864Z","iopub.status.idle":"2024-09-05T12:29:11.005515Z","shell.execute_reply.started":"2024-09-05T12:29:11.000827Z","shell.execute_reply":"2024-09-05T12:29:11.004554Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model_1.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=2e-2),\n loss='binary_crossentropy',\n metrics=['accuracy'])","metadata":{"execution":{"iopub.status.busy":"2024-09-05T12:30:02.143922Z","iopub.execute_input":"2024-09-05T12:30:02.144715Z","iopub.status.idle":"2024-09-05T12:30:02.159981Z","shell.execute_reply.started":"2024-09-05T12:30:02.144674Z","shell.execute_reply":"2024-09-05T12:30:02.159212Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"history_1=model_1.fit(train_generator, epochs=100,initial_epoch=20, callbacks=[lr_callback,checkpoint_callback], validation_data=val_generator)","metadata":{"execution":{"iopub.status.busy":"2024-09-05T12:30:07.771264Z","iopub.execute_input":"2024-09-05T12:30:07.771653Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def test_differentiability():\n # Create a dummy image tensor\n img = tf.random.uniform((1, 128, 128, 3), dtype=tf.float32)\n\n with tf.GradientTape() as tape:\n tape.watch(img)\n img_freq = fft2d_rgb(img)\n img_reconstructed = ifft2d_rgb(img_freq)\n\n # Compute gradients\n grads = tape.gradient(img_reconstructed, img)\n print(\"Gradients w.r.t input image:\", grads)\n\ntest_differentiability()\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:56.424836Z","iopub.execute_input":"2024-09-04T05:24:56.425156Z","iopub.status.idle":"2024-09-04T05:24:56.795867Z","shell.execute_reply.started":"2024-09-04T05:24:56.425125Z","shell.execute_reply":"2024-09-04T05:24:56.794889Z"},"id":"kNE3WvfxtT67","outputId":"ec795643-df28-4caa-e850-b64feb9c62fb","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"image = cv2.imread('/kaggle/input/deepfakedet-p2/1000_videos/train/fake/119_123_8.png')","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:56.797121Z","iopub.execute_input":"2024-09-04T05:24:56.797505Z","iopub.status.idle":"2024-09-04T05:24:56.804142Z","shell.execute_reply.started":"2024-09-04T05:24:56.797464Z","shell.execute_reply":"2024-09-04T05:24:56.803165Z"},"id":"UrNtN20UtT68","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"r, g, b = tf.split(image, num_or_size_splits=3, axis=-1)","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:56.807325Z","iopub.execute_input":"2024-09-04T05:24:56.807642Z","iopub.status.idle":"2024-09-04T05:24:56.814954Z","shell.execute_reply.started":"2024-09-04T05:24:56.807609Z","shell.execute_reply":"2024-09-04T05:24:56.814066Z"},"id":"MpcUi68CtT69","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x = m(tf.expand_dims(fft2d_rgb(image),axis=0))","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:56.816044Z","iopub.execute_input":"2024-09-04T05:24:56.816375Z","iopub.status.idle":"2024-09-04T05:24:58.519507Z","shell.execute_reply.started":"2024-09-04T05:24:56.81634Z","shell.execute_reply":"2024-09-04T05:24:58.518401Z"},"id":"m_sckbzFtT6-","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"r.shape","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.520875Z","iopub.execute_input":"2024-09-04T05:24:58.521281Z","iopub.status.idle":"2024-09-04T05:24:58.527781Z","shell.execute_reply.started":"2024-09-04T05:24:58.521237Z","shell.execute_reply":"2024-09-04T05:24:58.526648Z"},"id":"IC5lcwDLtT6_","outputId":"ebddaf75-5d4a-44f5-a7a9-9413d0dccafe","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x.shape","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.52905Z","iopub.execute_input":"2024-09-04T05:24:58.529405Z","iopub.status.idle":"2024-09-04T05:24:58.536556Z","shell.execute_reply.started":"2024-09-04T05:24:58.529367Z","shell.execute_reply":"2024-09-04T05:24:58.535737Z"},"id":"bqRY1cnGtT6_","outputId":"d1417bda-1dba-4fd2-8a69-9fae93c60c96","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x = ifft2d_rgb(x)","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.537666Z","iopub.execute_input":"2024-09-04T05:24:58.538014Z","iopub.status.idle":"2024-09-04T05:24:58.548594Z","shell.execute_reply.started":"2024-09-04T05:24:58.537974Z","shell.execute_reply":"2024-09-04T05:24:58.547754Z"},"id":"SkCN__sGtT7A","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x.shape","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.549623Z","iopub.execute_input":"2024-09-04T05:24:58.549979Z","iopub.status.idle":"2024-09-04T05:24:58.556144Z","shell.execute_reply.started":"2024-09-04T05:24:58.549945Z","shell.execute_reply":"2024-09-04T05:24:58.555251Z"},"id":"0LY_-183tT7B","outputId":"0c90ee6d-2930-452f-fdbc-f3149093c530","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"type(x)","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.557271Z","iopub.execute_input":"2024-09-04T05:24:58.557747Z","iopub.status.idle":"2024-09-04T05:24:58.564816Z","shell.execute_reply.started":"2024-09-04T05:24:58.557715Z","shell.execute_reply":"2024-09-04T05:24:58.564027Z"},"id":"eHQp9ZQCtT7C","outputId":"cab15cd0-70ea-4bde-af2c-b2fa424c2ce0","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"r, g, b, magnitude_spectrum_r, magnitude_spectrum_g, magnitude_spectrum_b = split_and_transform(np.array(tf.squeeze(x)))\nplot_results(r, g, b, magnitude_spectrum_r, magnitude_spectrum_g, magnitude_spectrum_b)","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:58.566047Z","iopub.execute_input":"2024-09-04T05:24:58.566425Z","iopub.status.idle":"2024-09-04T05:24:59.45211Z","shell.execute_reply.started":"2024-09-04T05:24:58.566384Z","shell.execute_reply":"2024-09-04T05:24:59.451239Z"},"id":"rdYcACXhtT7D","outputId":"2755ab78-a903-47af-be0d-41d136c7e6be","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"y = build_dcgan_discriminator((128,128,3))","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:59.453321Z","iopub.execute_input":"2024-09-04T05:24:59.453639Z","iopub.status.idle":"2024-09-04T05:24:59.540659Z","shell.execute_reply.started":"2024-09-04T05:24:59.453607Z","shell.execute_reply":"2024-09-04T05:24:59.539641Z"},"id":"D-NWcurTtT7D","outputId":"06531bc0-62a8-4ca0-949a-f0449403ebb7","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(y.predict(x))","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:24:59.54194Z","iopub.execute_input":"2024-09-04T05:24:59.542381Z","iopub.status.idle":"2024-09-04T05:25:00.335121Z","shell.execute_reply.started":"2024-09-04T05:24:59.542334Z","shell.execute_reply":"2024-09-04T05:25:00.334089Z"},"id":"mWB5VZLytT7E","outputId":"d3d421f0-5eb1-4f6a-e4fc-55f0da026dc7","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\n\ndef load_and_preprocess_image(image_path, img_size=(128, 128)):\n \"\"\"\n Load an image from a file, resize it to a fixed size, and normalize pixel values.\n\n Args:\n - image_path (str): The file path of the image to be loaded.\n - img_size (tuple): The desired image size (height, width).\n\n Returns:\n - image (tf.Tensor): The preprocessed image tensor.\n \"\"\"\n # Load the image file\n image = tf.io.read_file(image_path)\n\n # Decode the image as JPEG, channels last (RGB format)\n image = tf.image.decode_jpeg(image, channels=3)\n\n # Resize the image to the target size\n image = tf.image.resize(image, img_size)\n\n # Normalize the pixel values to [0, 1] range\n image = tf.cast(image, tf.float32) / 255.0\n\n return image\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:25:00.33827Z","iopub.execute_input":"2024-09-04T05:25:00.338585Z","iopub.status.idle":"2024-09-04T05:25:00.344847Z","shell.execute_reply.started":"2024-09-04T05:25:00.338552Z","shell.execute_reply":"2024-09-04T05:25:00.34388Z"},"id":"FdWCaLmHtT7F","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"For the training for freqgan ,the dataset will contain both the real and fake images..\nas written in the paper 'https://arxiv.org/pdf/2202.03347'\n'both real and fake images are added with the generated perturbation maps of FrePGAN'","metadata":{"id":"aDRw3stutT7G"}},{"cell_type":"code","source":"\ndef data_generator(directory, batch_size=16, img_size=(128, 128)):\n real_images = []\n fake_images = []\n labels = []\n\n # Load real and fake images\n for img_path in os.listdir(os.path.join(directory, \"real\")):\n img = load_and_preprocess_image(os.path.join(directory, \"real\", img_path), img_size)\n real_images.append(img)\n labels.append(0) # Label 0 for real\n\n for img_path in os.listdir(os.path.join(directory, \"fake\")):\n img = load_and_preprocess_image(os.path.join(directory, \"fake\", img_path), img_size)\n fake_images.append(img)\n labels.append(1) # Label 1 for fake\n\n all_images = np.concatenate([real_images, fake_images], axis=0)\n all_labels = np.array(labels)\n\n # Shuffle the dataset\n dataset = tf.data.Dataset.from_tensor_slices((all_images, all_labels))\n dataset = dataset.shuffle(buffer_size=1024).batch(batch_size)\n\n for images, labels in dataset:\n yield images, labels\ntrain_dataset = tf.data.Dataset.from_generator(\n lambda: data_generator('/kaggle/input/deepfakedet-p2/1000_videos/train', batch_size=16),\n output_signature=(\n tf.TensorSpec(shape=(None, 128, 128, 3), dtype=tf.float32),\n tf.TensorSpec(shape=(None,), dtype=tf.int32)\n )\n)\n\n\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:25:00.346085Z","iopub.execute_input":"2024-09-04T05:25:00.346431Z","iopub.status.idle":"2024-09-04T05:25:00.388241Z","shell.execute_reply.started":"2024-09-04T05:25:00.3464Z","shell.execute_reply":"2024-09-04T05:25:00.387503Z"},"id":"UDOjSQXutT7G","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Training OF FrepGAN","metadata":{"id":"p6xNdcKZtT7H"}},{"cell_type":"code","source":"def discriminator_loss(real_preds, fake_preds):\n \"\"\"\n Compute the adversarial loss for the discriminator.\n\n Parameters:\n real_preds (tf.Tensor): Predictions of the discriminator for real images.\n fake_preds (tf.Tensor): Predictions of the discriminator for fake images.\n\n Returns:\n tf.Tensor: The computed discriminator loss.\n \"\"\"\n #real_loss = tf.keras.losses.binary_crossentropy(tf.ones_like(real_preds), real_preds, from_logits=True)\n #fake_loss = tf.keras.losses.binary_crossentropy(tf.zeros_like(fake_preds), fake_preds, from_logits=True)\n #return tf.reduce_mean(real_loss + fake_loss)\n # Use tf.clip_by_value to prevent log(0) issues\n log_real_preds = tf.math.log(tf.clip_by_value(real_preds, 1e-10, 1.0))\n log_fake_preds = tf.math.log(tf.clip_by_value(1 - fake_preds, 1e-10, 1.0))\n\n # Step 3: Compute the mean loss\n d_loss_real = -tf.reduce_mean(log_real_preds)\n d_loss_fake = -tf.reduce_mean(log_fake_preds)\n d_loss = d_loss_real + d_loss_fake\n\n return d_loss\n\ndef generator_loss(fake_preds):\n \"\"\"\n Compute the adversarial loss for the generator.\n\n Parameters:\n fake_preds (tf.Tensor): Predictions of the discriminator for fake images.\n\n Returns:\n tf.Tensor: The computed generator loss.\n \"\"\"\n # Use tf.clip_by_value to prevent log(0) issues\n log_fake_preds = tf.math.log(tf.clip_by_value(1 - fake_preds, 1e-10, 1.0))\n\n # Step 4: Compute the mean loss\n g_loss_adv = -tf.reduce_mean(log_fake_preds)\n\n return g_loss_adv\n #return tf.reduce_mean(tf.keras.losses.binary_crossentropy(tf.ones_like(fake_preds), fake_preds, from_logits=True))\n\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:25:13.344589Z","iopub.execute_input":"2024-09-04T05:25:13.345411Z","iopub.status.idle":"2024-09-04T05:25:13.354634Z","shell.execute_reply.started":"2024-09-04T05:25:13.345371Z","shell.execute_reply":"2024-09-04T05:25:13.352944Z"},"id":"leWDdJAmtT7H","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"\n################### DUMMY IMPLEMENTATION #######################################\n\nimport tensorflow as tf\n\n# Assume models G, D, C are defined and compiled using TensorFlow/Keras\n# generator (G), discriminator (D), and classifier (C)\n# Assume train_dataset is a tf.data.Dataset that yields (imgs, labels)\n# Initialize epochs and batch size\nauc_metric = tf.keras.metrics.AUC()\nprecision_metric = tf.keras.metrics.Precision()\nf1_metric = tf.keras.metrics.F1Score()\nepochs = 20\nbatch_size = 16\n\n# Instantiate models\ngenerator = build_vgg_like_generator(input_shape=(128, 128, 6))\ndiscriminator = build_dcgan_discriminator((128, 128, 3))\nclassifier = build_deepfake_classifier()\n\n# Optimizers\noptimizer_D = tf.keras.optimizers.Adam(learning_rate=0.1)\noptimizer_G = tf.keras.optimizers.Adam(learning_rate=1e-4)\noptimizer_C = tf.keras.optimizers.Adam(learning_rate=0.01)\n\n# Loss functions\n\nadversarial_loss = tf.keras.losses.BinaryCrossentropy(from_logits=False)\ncompression_loss = lambda x: tf.reduce_mean(tf.square(tf.norm(x, ord='euclidean', axis=1))) # Example compression loss\nclassification_loss = tf.keras.losses.BinaryCrossentropy()\n\n# Custom training loop\nfor epoch in range(epochs):\n print(f\"Epoch {epoch + 1}/{epochs}\")\n auc_metric.reset_state()\n precision_metric.reset_state()\n f1_metric.reset_state()\n for step, (imgs, labels) in enumerate(train_dataset):\n\n # Ensure images are float32 for TensorFlow operations\n imgs = tf.cast(imgs, tf.float32)\n\n # ============================\n # FFT Transformation\n # ============================\n imgs_freq = fft2d_rgb(imgs) # Transform to frequency domain\n\n with tf.GradientTape() as tape_G, tf.GradientTape() as tape_D, tf.GradientTape() as tape_C:\n # ============================\n # Generator Forward Pass\n # ============================\n perturbations_freq = generator(imgs_freq, training=True)\n perturbations = ifft2d_rgb(perturbations_freq) # Back to spatial domain\n\n # ============================\n # Discriminator Forward Pass\n # ============================\n real_preds = discriminator(imgs, training=True)\n fake_imgs = imgs + perturbations\n fake_preds = discriminator(fake_imgs, training=True)\n\n # ============================\n # Discriminator Loss\n # ============================\n #real_loss = adversarial_loss(tf.ones_like(real_preds), real_preds)\n #fake_loss = adversarial_loss(tf.zeros_like(fake_preds), fake_preds)\n #d_loss = (real_loss + fake_loss) / 2\n d_loss = discriminator_loss(real_preds, fake_preds)\n\n # ============================\n # Generator Loss (Dummy Example)\n # ============================\n #dummy_loss = tf.reduce_mean(perturbations_freq) + 0.1 * tf.reduce_sum(perturbations_freq)\n #print(\"g loss\", dummy_loss)\n #g_loss_adv = adversarial_loss(tf.ones_like(fake_preds), fake_preds)\n #g_loss_com = compression_loss(perturbations_freq)\n #g_loss = 0.5 * g_loss_adv + 0.5 * g_loss_com\n #print(\"g loss\", g_loss)\n\n #### adv loss = E[1-LOG(D(G(x)))] , com loss E[|G(X)|2]\n d = discriminator(perturbations,training=False)\n g_loss = 0.5*generator_loss(d) + 0.5*compression_loss(perturbations)\n #print(\"g loss\", g_loss)\n\n # ============================\n # Classifier Forward Pass\n # ============================\n c_preds = classifier(fake_imgs, training=True)\n c_loss = classification_loss(labels, c_preds)\n\n # Calculate gradients\n gradients_D = tape_D.gradient(d_loss, discriminator.trainable_variables)\n gradients_G = tape_G.gradient(g_loss, generator.trainable_variables) # Using dummy_loss for G\n gradients_C = tape_C.gradient(c_loss, classifier.trainable_variables)\n\n # Debug: Print gradients to check if they are None\n #print(f\"Gradients D: {[grad is None for grad in gradients_D]}\")\n #print(f\"Gradients G: {[grad is None for grad in gradients_G]}\")\n #print(f\"Gradients C: {[grad is None for grad in gradients_C]}\")\n\n # Apply gradients\n optimizer_D.apply_gradients(zip(gradients_D, discriminator.trainable_variables))\n optimizer_G.apply_gradients(zip(gradients_G, generator.trainable_variables))\n optimizer_C.apply_gradients(zip(gradients_C, classifier.trainable_variables))\n\n # Update metrics\n auc_metric.update_state(labels, c_preds)\n precision_metric.update_state(labels, c_preds)\n f1_metric.update_state(tf.expand_dims(labels,axis=1), c_preds)\n\n # Print losses\n print(f\"Epoch [{epoch + 1}/{epochs}], Step [{step + 1}], \"\n f\"D Loss: {d_loss.numpy()}, G Loss: {g_loss.numpy()}, C Loss: {c_loss.numpy()}\")\n # Print metrics at the end of the epoch\n print(f\"Epoch [{epoch + 1}/{epochs}] Metrics - \"\n f\"classifier_AUC: {auc_metric.result().numpy():.4f}, \"\n f\"classifier_Precision: {precision_metric.result().numpy():.4f}, \"\n f\"classifier_Accuracy: {accuracy_metric.result().numpy():.4f}\")\n\nprint(\"Training completed.\")\n\n############################# ENDS #####################################33\n","metadata":{"execution":{"iopub.status.busy":"2024-09-03T13:41:08.03728Z","iopub.execute_input":"2024-09-03T13:41:08.037773Z","iopub.status.idle":"2024-09-03T13:44:05.795465Z","shell.execute_reply.started":"2024-09-03T13:41:08.037727Z","shell.execute_reply":"2024-09-03T13:44:05.793426Z"},"id":"oHJdJ-kBtT7I","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import tensorflow as tf\n\ndef train_gan_with_metrics(epochs, batch_size, train_dataset, generator, discriminator, classifier,\n optimizer_D, optimizer_G, optimizer_C, fft2d_rgb, ifft2d_rgb,\n discriminator_loss, generator_loss, compression_loss, classification_loss,save_model_interval=10, save_path='./models'):\n # Metrics\n auc_metric = tf.keras.metrics.AUC()\n precision_metric = tf.keras.metrics.Precision()\n accuracy_metric = tf.keras.metrics.BinaryAccuracy()\n f1_metric = tf.keras.metrics.F1Score()\n\n # Lists to store losses\n d_losses = []\n g_losses = []\n c_losses = []\n\n for epoch in range(epochs):\n print(f\"Epoch {epoch + 1}/{epochs}\")\n\n # Reset metrics at the start of each epoch\n auc_metric.reset_state()\n precision_metric.reset_state()\n accuracy_metric.reset_state()\n f1_metric.reset_state()\n\n epoch_d_loss = 0.0\n epoch_g_loss = 0.0\n epoch_c_loss = 0.0\n steps = 0\n try:\n for step, (imgs, labels) in enumerate(train_dataset):\n steps += 1\n\n # Ensure images are float32 for TensorFlow operations\n imgs = tf.cast(imgs, tf.float32)\n\n # ============================\n # FFT Transformation\n # ============================\n imgs_freq = fft2d_rgb(imgs) # Transform to frequency domain\n\n with tf.GradientTape() as tape_G, tf.GradientTape() as tape_D, tf.GradientTape() as tape_C:\n # ============================\n # Generator Forward Pass\n # ============================\n perturbations_freq = generator(imgs_freq, training=True)\n perturbations = ifft2d_rgb(perturbations_freq) # Back to spatial domain\n\n # ============================\n # Discriminator Forward Pass\n # ============================\n real_preds = discriminator(imgs, training=True)\n fake_imgs = imgs + perturbations\n fake_preds = discriminator(fake_imgs, training=True)\n\n # ============================\n # Discriminator Loss\n # ============================\n d_loss = discriminator_loss(real_preds, fake_preds)\n\n # ============================\n # Generator Loss\n # ============================\n d = discriminator(perturbations, training=False)\n g_loss = 0.5 * generator_loss(d) + 0.5 * compression_loss(perturbations)\n\n # ============================\n # Classifier Forward Pass\n # ============================\n c_preds = classifier(fake_imgs, training=True)\n c_loss = classification_loss(labels, c_preds)\n\n # Calculate gradients\n gradients_D = tape_D.gradient(d_loss, discriminator.trainable_variables)\n gradients_G = tape_G.gradient(g_loss, generator.trainable_variables)\n gradients_C = tape_C.gradient(c_loss, classifier.trainable_variables)\n\n # Apply gradients\n optimizer_D.apply_gradients(zip(gradients_D, discriminator.trainable_variables))\n optimizer_G.apply_gradients(zip(gradients_G, generator.trainable_variables))\n optimizer_C.apply_gradients(zip(gradients_C, classifier.trainable_variables))\n\n # Update metrics\n auc_metric.update_state(labels, c_preds)\n precision_metric.update_state(labels, c_preds)\n accuracy_metric.update_state(labels, c_preds)\n f1_metric.update_state(tf.expand_dims(labels, axis=1), c_preds)\n\n # Accumulate losses for the epoch\n epoch_d_loss += d_loss.numpy()\n epoch_g_loss += g_loss.numpy()\n epoch_c_loss += c_loss.numpy()\n\n # Print losses for each step\n if(step%100==0):\n print(f\"Epoch [{epoch + 1}/{epochs}], Step [{step + 1}], \",f\"D Loss: {d_loss.numpy()}, G Loss: {g_loss.numpy()}, C Loss: {c_loss.numpy()}\")\n except tf.errors.OutOfRangeError:\n # If dataset is exhausted, handle it\n print(f\"Dataset exhausted on epoch {epoch + 1}. Moving to next epoch.\")\n\n # Compute average losses for the epoch\n avg_d_loss = epoch_d_loss / steps\n avg_g_loss = epoch_g_loss / steps\n avg_c_loss = epoch_c_loss / steps\n\n d_losses.append(avg_d_loss)\n g_losses.append(avg_g_loss)\n c_losses.append(avg_c_loss)\n\n # Print metrics at the end of the epoch\n f1_value = f1_metric.result().numpy()\n if isinstance(f1_value, np.ndarray):\n f1_value = f1_value.mean() # Take the mean if it's an array\n\n print(f\"Epoch [{epoch + 1}/{epochs}] Metrics - \"\n f\"classifier_AUC: {auc_metric.result().numpy():.4f}, \"\n f\"classifier_Precision: {precision_metric.result().numpy():.4f}, \"\n f\"classifier_Accuracy: {accuracy_metric.result().numpy():.4f}, \"\n f\"classifier_F1: {f1_value:.4f}\")\n if (epoch + 1) % save_model_interval == 0:\n # Save generator model architecture and weights\n generator_json = generator.to_json()\n with open(f\"{save_path}/generator_epoch_{epoch + 1}.json\", \"w\") as json_file:\n json_file.write(generator_json)\n generator.save_weights(f\"{save_path}/generator_epoch_{epoch + 1}.weights.h5\")\n\n # Save discriminator model architecture and weights\n discriminator_json = discriminator.to_json()\n with open(f\"{save_path}/discriminator_epoch_{epoch + 1}.json\", \"w\") as json_file:\n json_file.write(discriminator_json)\n discriminator.save_weights(f\"{save_path}/discriminator_epoch_{epoch + 1}.weights.h5\")\n\n # Save classifier model architecture and weights\n classifier_json = classifier.to_json()\n with open(f\"{save_path}/classifier_epoch_{epoch + 1}.json\", \"w\") as json_file:\n json_file.write(classifier_json)\n classifier.save_weights(f\"{save_path}/classifier_epoch_{epoch + 1}.weights.h5\")\n\n print(f\"Models saved for epoch {epoch + 1}\")\n\n\n print(\"Training completed.\")\n\n return d_losses, g_losses, c_losses\ncompression_loss = lambda x: tf.reduce_mean(tf.square(tf.norm(x, ord='euclidean', axis=1))) # Example compression loss\n\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:25:23.227097Z","iopub.execute_input":"2024-09-04T05:25:23.227658Z","iopub.status.idle":"2024-09-04T05:25:23.262186Z","shell.execute_reply.started":"2024-09-04T05:25:23.227607Z","shell.execute_reply":"2024-09-04T05:25:23.261015Z"},"id":"NdlEGNactT7K","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"initially:\nlr of d , g:1e-1.1e-4\nlr ''': 1e-3,1e-4\nlr ''': 1e-3,1e-5\n\noptimize classifier learning rate","metadata":{"id":"51-XtichtT7L"}},{"cell_type":"code","source":"# Example call to the function\nd_losses, g_losses, c_losses = train_gan_with_metrics(\n epochs=20,\n batch_size=16,\n train_dataset=train_dataset,\n generator=build_vgg_like_generator(input_shape=(128, 128, 6)),\n discriminator=build_dcgan_discriminator((128, 128, 3)),\n classifier=build_deepfake_classifier(),\n optimizer_D=tf.keras.optimizers.Adam(learning_rate=1e-4),\n optimizer_G=tf.keras.optimizers.Adam(learning_rate=1e-5),\n optimizer_C=tf.keras.optimizers.Adam(learning_rate=1e-4),\n fft2d_rgb=fft2d_rgb,\n ifft2d_rgb=ifft2d_rgb,\n discriminator_loss=discriminator_loss,\n generator_loss=generator_loss,\n compression_loss=compression_loss,\n classification_loss=tf.keras.losses.BinaryCrossentropy(),\n save_path='/kaggle/working/'\n)\n","metadata":{"execution":{"iopub.status.busy":"2024-09-04T05:25:32.332251Z","iopub.execute_input":"2024-09-04T05:25:32.33307Z"},"id":"hN_kGCFWtT7M","outputId":"0e2cc4d7-60c0-40ed-a0f9-7f8cd274df60","trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{"id":"3g4XRpKLtT7N"},"execution_count":null,"outputs":[]}]}