{"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":"nvidiaTeslaT4","dataSources":[{"sourceId":7795314,"sourceType":"datasetVersion","datasetId":4563655}],"dockerImageVersionId":29844,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport json\nimport cv2\nimport tensorflow as tf\nimport numpy as np\nimport pandas as pd\nfrom tensorflow.keras.utils import to_categorical\nfrom tensorflow.keras.preprocessing.image import img_to_array\nfrom tensorflow.keras import layers\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Define constants\nIMAGE_HEIGHT = 256\nIMAGE_WIDTH = 256\nCHANNELS = 3\nNUM_CLASSES = 2  # Real or Fake","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Function to read metadata file\ndef read_metadata(metadata_path):\n    with open(metadata_path, 'r') as f:\n        metadata = json.load(f)\n    return metadata","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# DataFrame of Meta-Data\nmeta_data = pd.DataFrame(pd.read_json(\"/kaggle/input/deepfakes-dataset/LAV-DF/metadata.min.json\"))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"meta_data.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"meta_data.info()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"meta_data.index = meta_data['file']","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"meta_data.columns","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# creating helper functions\n\n# generate data paths with labels\ndef define_paths(dir_path,extra='test/'):\n    filepaths = []\n    labels = []\n    vids = os.listdir(dir_path)\n    for vid in vids:\n        if extra+vid in meta_data['file'].values:\n            vid_path = os.path.join(dir_path,vid)\n            filepaths.append(vid_path)\n            labels.append(1 if meta_data.loc[extra+vid,'modify_video'] else 0)\n    return filepaths,labels\n\ndef define_df(files,labels):\n    Fseries = pd.Series(files,name='filepaths')\n    Lseries = pd.Series(labels,name='labels')\n    return pd.concat([Fseries,Lseries],axis=1)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"meta_data['file'].values","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_file_paths, train_file_labels = define_paths(\"/kaggle/input/deepfakes-dataset/LAV-DF/train\",\"train/\")\nval_file_paths, val_file_labels = define_paths(\"/kaggle/input/deepfakes-dataset/LAV-DF/dev\",\"dev/\")\ntest_file_paths, test_file_labels = define_paths(\"/kaggle/input/deepfakes-dataset/LAV-DF/test\",\"test/\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_df = define_df(train_file_paths,train_file_labels)\nval_df = define_df(val_file_paths,val_file_labels)\ntest_df = define_df(test_file_paths,test_file_labels)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"val_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import cv2\nimport numpy as np\nfrom tensorflow.keras.utils import to_categorical\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\n\ndef video_frame_generator(df, img_size=(256, 256), batch_size=32, color_mode='rgb', shuffle=True):\n    while True:\n        if shuffle:\n            df = df.sample(frac=1).reset_index(drop=True)\n\n        for start in range(0, len(df), batch_size):\n            end = min(start + batch_size, len(df))\n            batch_df = df[start:end]\n\n            X_batch = []\n            y_batch = []\n\n            for i, row in batch_df.iterrows():\n                video_path = row['filepaths']\n                label = row['labels']\n\n                frames = read_video_frames(video_path, img_size, color_mode)\n                X_batch.append(frames)\n                y_batch.append(label)\n\n            X_batch = np.array(X_batch)\n            y_batch = to_categorical(y_batch, num_classes=2)  # Assuming binary classification\n\n            yield X_batch, y_batch\n\ndef read_video_frames(video_path, img_size, color_mode):\n    cap = cv2.VideoCapture(video_path)\n    frames = []\n\n    while True:\n        ret, frame = cap.read()\n\n        if not ret:\n            break\n\n        frame = cv2.resize(frame, (img_size[1], img_size[0]))\n\n        if color_mode == 'rgb':\n            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)\n\n        frames.append(frame)\n\n    cap.release()\n    return np.array(frames)\n\n# Example usage:\ntrain_video_gen = video_frame_generator(train_df, img_size=(256, 256), batch_size=32, color_mode='rgb', shuffle=True)\nvalid_video_gen = video_frame_generator(val_df, img_size=(256, 256), batch_size=32, color_mode='rgb', shuffle=False)\ntest_video_gen = video_frame_generator(test_df, img_size=(256, 256), batch_size=32, color_mode='rgb', shuffle=False)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\ndef show_video_frames(video_gen, num_videos=5, frames_per_video=5):\n    plt.figure(figsize=(15, 10))\n    videos, labels = next(video_gen)\n\n    for i in range(num_videos):\n        plt.subplot(num_videos, 1, i + 1)\n        video_frames = videos[i]\n        label = labels[i]\n\n        for j in range(frames_per_video):\n            plt.subplot(num_videos, frames_per_video, i * frames_per_video + j + 1)\n            frame = video_frames[j] / 255.0\n            plt.imshow(frame)\n            plt.axis('off')\n\n        # Note: Assuming binary classification\n        class_name = 'Real' if label[0] == 1 else 'Fake'\n        plt.title(f'Class: {class_name}', color='blue', fontsize=10)\n\n    plt.show()\n\n# Example usage:\nshow_video_frames(train_video_gen, num_videos=5, frames_per_video=5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Hyperparameters\nepochs = 100\nbatch_size = 32\nlatent_dim = 100\nnoise_dim = 256\nnoise_shape = (batch_size, 1, 1, noise_dim)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import os\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom tensorflow.keras.models import Sequential, Model\nfrom tensorflow.keras.layers import Dense, LeakyReLU, BatchNormalization, Reshape, Flatten, Input, Conv2D, Dropout, Conv2DTranspose\nfrom tensorflow.keras.optimizers import Adam\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\n\n\n# Define constants\nimg_size = (256, 256, 3)  # Adjust the size to your video frame size\nlatent_dim = 100\n\n# Define the generator model\ndef build_generator(latent_dim):\n    model = Sequential()\n    model.add(Dense(128 * 64 * 64, input_dim=latent_dim))  # Adjust the input_dim based on the latent_dim\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(Reshape((64, 64, 128)))\n    model.add(BatchNormalization(momentum=0.8))\n    model.add(Conv2DTranspose(128, (4, 4), strides=(2, 2), padding='same'))\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(BatchNormalization(momentum=0.8))\n    model.add(Conv2DTranspose(128, (4, 4), strides=(2, 2), padding='same'))\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(BatchNormalization(momentum=0.8))\n    model.add(Conv2D(3, (7, 7), activation='tanh', padding='same'))\n    return model\n\n# Define the discriminator model\ndef build_discriminator(img_size):\n    model = Sequential()\n    model.add(Conv2D(64, (3,3), strides=(2,2), padding='same', input_shape=img_size))\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(Dropout(0.25))\n    model.add(Conv2D(128, (3,3), strides=(2,2), padding='same'))\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(Dropout(0.25))\n    model.add(Conv2D(256, (3,3), strides=(2,2), padding='same'))\n    model.add(LeakyReLU(alpha=0.2))\n    model.add(Dropout(0.25))\n    model.add(Flatten())\n    model.add(Dense(1, activation='sigmoid'))\n    return model\n\n# Build and compile the discriminator\ndiscriminator = build_discriminator(img_size)\ndiscriminator.compile(loss='binary_crossentropy', optimizer=Adam(0.001, 0.5), metrics=['accuracy','loss'])\n\n# Build the generator\ngenerator = build_generator(latent_dim)\n\n# Build the GAN model\ndiscriminator.trainable = False\ngan_input = Input(shape=(latent_dim,))\nfake_video = generator(gan_input)\ngan_output = discriminator(fake_video)\ngan = Model(gan_input, gan_output)\ngan.compile(loss='binary_crossentropy', optimizer=Adam(0.001, 0.5))\n\n# Training the GAN\ndef train_gan(generator, discriminator, gan, epochs=100, batch_size=32, save_interval=10):\n    for epoch in range(epochs):\n        for _ in range(2460):  # You need to define 'steps_per_epoch' based on your dataset size\n\n            # Train discriminator on real videos\n            real_videos, real_labels = next(train_video_gen)\n            fake_labels = np.zeros((batch_size, 1))\n\n            # Flatten the real_videos array\n            real_videos_flat = np.concatenate(real_videos, axis=0)\n\n            # Select a random batch of real labels\n            idx = np.random.randint(0, real_labels.shape[0], batch_size)\n            sampled_real_labels = real_labels[idx]\n\n            # Convert real_videos to TensorFlow tensor\n            real_videos_tensor = tf.convert_to_tensor(real_videos_flat, dtype=tf.float32)\n\n            d_loss_real = discriminator.train_on_batch(real_videos_tensor, sampled_real_labels)\n\n            # Train discriminator on fake videos\n            noise = np.random.normal(0, 1, (batch_size, latent_dim))\n            fake_videos = generator.predict(noise)\n            d_loss_fake = discriminator.train_on_batch(fake_videos, fake_labels)\n\n            # Calculate discriminator loss\n            d_loss = 0.5 * np.add(d_loss_real, d_loss_fake)\n\n            # Train generator\n            noise = np.random.normal(0, 1, (batch_size, latent_dim))\n            valid_labels = np.ones((batch_size, 1))\n            g_loss = gan.train_on_batch(noise, valid_labels)\n\n        # Print progress and save generated images at specified intervals\n        if epoch % save_interval == 0:\n            print(f\"{epoch} [D loss: {d_loss[0]} | D accuracy: {100 * d_loss[1]}] [G loss: {g_loss}]\")\n            save_generated_images(generator, epoch)\n\n\n\n# Function to save generated images\ndef save_generated_images(generator, epoch, examples=10, dim=(1, 10), figsize=(10, 1)):\n    noise = np.random.normal(0, 1, (examples, latent_dim))\n    generated_videos = generator.predict(noise)\n    generated_videos = 0.5 * generated_videos + 0.5\n\n    plt.figure(figsize=figsize)\n    for i in range(generated_videos.shape[0]):\n        plt.subplot(dim[0], dim[1], i + 1)\n        plt.imshow(generated_videos[i], interpolation='nearest', cmap='gray_r')\n        plt.axis('off')\n    plt.tight_layout()\n    plt.savefig(f\"gan_generated_image_epoch_{epoch}.png\")\n\n# Train the GAN\ntrain_gan(generator, discriminator, gan, epochs=100, batch_size=32, save_interval=10)\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Compile the GAN model\ngan_model.compile(loss_fn_gen, loss_fn_disc, optimizer_gen, optimizer_disc)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Lists to store training metrics\ngen_losses = []\ndisc_losses = []\n\n# Training loop\nfor epoch in range(epochs):\n    # Generate fake videos\n    noise_shape = (batch_size, 256, 256, 3, 3)\n    noise = np.random.normal(0, 1, noise_shape)\n\n    # Ensure the generator model is compiled before predicting\n    if not hasattr(generator, 'compiled') or not generator.compiled:\n        generator.compile(optimizer='Adam', loss='mean_squared_error')\n\n    # Check if the generator model is callable\n    if not callable(generator):\n        raise TypeError(\"Generator model is not callable. Check your model architecture.\")\n\n    fake_videos = generator.predict(noise)\n\n    # Train Discriminator\n    disc_loss = gan_model.train_step(fake_videos)\n\n    # Train Generator\n    with tf.GradientTape() as tape:\n        gen_loss = gan_model.generator_loss(fake_videos)\n\n    # Apply gradients\n    gen_grads = tape.gradient(gen_loss, generator.trainable_variables)\n    optimizer_gen.apply_gradients(zip(gen_grads, generator.trainable_variables))\n\n    # Store losses\n    gen_losses.append(gen_loss)\n    disc_losses.append(disc_loss)\n\n    # Print progress\n    if epoch % 100 == 0:\n        print(f\"Epoch {epoch}/{epochs} [D loss: {disc_loss.numpy()}] [G loss: {gen_loss.numpy()}]\")\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Plot training and validation losses\nplt.plot(gen_losses, label='Generator Loss')\nplt.plot(disc_losses, label='Discriminator Loss')\nplt.xlabel('Epoch')\nplt.ylabel('Loss')\nplt.legend()\nplt.show()\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}