{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":63103,"databundleVersionId":6877691,"sourceType":"competition"}],"dockerImageVersionId":30646,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"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\n# import numpy as np # linear algebra\n# import 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\n# import os\n# for 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","execution":{"iopub.status.busy":"2024-02-04T03:39:50.911185Z","iopub.execute_input":"2024-02-04T03:39:50.911534Z","iopub.status.idle":"2024-02-04T03:39:50.917327Z","shell.execute_reply.started":"2024-02-04T03:39:50.911501Z","shell.execute_reply":"2024-02-04T03:39:50.916408Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Prebuild face detection model ","metadata":{}},{"cell_type":"code","source":"pip install facenet-pytorch","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:39:50.919330Z","iopub.execute_input":"2024-02-04T03:39:50.919630Z","iopub.status.idle":"2024-02-04T03:40:05.128688Z","shell.execute_reply.started":"2024-02-04T03:39:50.919607Z","shell.execute_reply":"2024-02-04T03:40:05.127614Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# General imports ","metadata":{}},{"cell_type":"code","source":"import torch \n\nuse_cuda  = torch.cuda.is_available()\ndevice = torch.device(\"cuda\" if use_cuda else \"cpu\")\ntorch.manual_seed(42)\nprint(device)","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:40:05.130113Z","iopub.execute_input":"2024-02-04T03:40:05.130411Z","iopub.status.idle":"2024-02-04T03:40:09.393516Z","shell.execute_reply.started":"2024-02-04T03:40:05.130382Z","shell.execute_reply":"2024-02-04T03:40:09.392603Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Transforms Dataset ","metadata":{}},{"cell_type":"code","source":"import os\nfrom PIL import Image\nimport torch \nfrom torch.utils.data import Dataset   # to create a dataset for reterving the data from the diectories\nfrom torchvision import transforms\nfrom facenet_pytorch import MTCNN\n\nroot_directory = r'/kaggle/input/face-classification-deep-learning-cs-405/dataset/train'\n\n# Define the data transform \ntransform  = transforms.Compose([\n    transforms.Resize((160 , 160)),\n    transforms.ToTensor(),\n])\n\nclass CustomDataset(Dataset):\n    def __init__(self , root_directory):\n        self.root_directory = root_directory\n        self.identity_directories = os.listdir(root_directory)\n        self.transform = transform\n        self.mtcnn = MTCNN()\n        self.identity_mapping  = {identity:label for label,identity in enumerate(self.identity_directories)}\n        self.data = []\n        for identity_label , identity in enumerate(self.identity_directories):\n            images  = os.listdir(os.path.join(root_directory , identity))\n            for image in images:\n                image_path = os.path.join(root_directory , identity , image)\n                self.data.append((image_path , identity_label))       \n                \n    def __len__(self):\n        return len(self.data)\n    \n    def __getitem__(self,index):\n        image_path, identity_label = self.data[index]\n        img = Image.open(image_path)\n\n        # Use MTCNN for face detection and alignment\n        boxes, probs = self.mtcnn.detect(img)\n\n        # If faces are detected, crop and align them\n        if boxes is not None:\n            face = img.crop((boxes[0][0], boxes[0][1], boxes[0][2], boxes[0][3]))  # Cropping the face\n\n            # Apply transform to the cropped face\n            face_transformed = self.transform(face)\n            return face_transformed, torch.tensor(identity_label)\n\n        else:\n            return None\n        \n                \ndataset = CustomDataset(root_directory)","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:40:09.395110Z","iopub.execute_input":"2024-02-04T03:40:09.395757Z","iopub.status.idle":"2024-02-04T03:40:52.040752Z","shell.execute_reply.started":"2024-02-04T03:40:09.395719Z","shell.execute_reply":"2024-02-04T03:40:52.039940Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from torch.utils.data import DataLoader\n\n#Spliting the data into traning ans validation and test sets\ntrain_size = int(0.7 * len(dataset))\nvalidation_size = int(0.15 * len(dataset))\ntest_size = len(dataset)  - train_size - validation_size \n\n#spliting the data\ntrain_data , validation_data , test_data = torch.utils.data.random_split(dataset , [train_size , validation_size , test_size])\n\n#define the batch size for the dataloaders\nbatch_size = 128\n\n#creating the data loader\ntrain_loader = DataLoader(train_data, batch_size=batch_size, shuffle=True, collate_fn=lambda x: [sample for sample in x if sample is not None], num_workers=2)\nvalidation_loader = DataLoader(validation_data, batch_size=batch_size, shuffle=False, collate_fn=lambda x: [sample for sample in x if sample is not None])\ntest_loader = DataLoader(test_data, batch_size=batch_size, shuffle=False, collate_fn=lambda x: [sample for sample in x if sample is not None])\n","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:40:52.043580Z","iopub.execute_input":"2024-02-04T03:40:52.044076Z","iopub.status.idle":"2024-02-04T03:40:52.071802Z","shell.execute_reply.started":"2024-02-04T03:40:52.044042Z","shell.execute_reply":"2024-02-04T03:40:52.071098Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import matplotlib.pyplot as plt\nimport numpy as np\n\ndef display_images(loader):\n    # Get a batch of data\n    data_iter = iter(loader)\n    batch = next(data_iter)\n\n    inputs, labels = zip(*batch)\n\n    # Convert the batch to NumPy arrays\n    inputs = np.array([item.numpy() for item in inputs])\n\n    # Define a function to display images and labels\n    def show_images(images, labels):\n        num_images = len(images)\n        num_cols = 5  # Number of columns for image display\n        num_rows = (num_images + num_cols - 1) // num_cols\n\n        plt.figure(figsize=(15, 3 * num_rows))\n        for i in range(num_images):\n            plt.subplot(num_rows, num_cols, i + 1)\n            plt.imshow(np.transpose(images[i], (1, 2, 0)))\n            if labels is not None:\n                plt.title(f\"Label: {labels[i]}\")  # Show label as title\n            plt.axis(\"off\")\n\n    show_images(inputs, labels)\n    plt.show()\n\ndisplay_images(train_loader)","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:40:52.072764Z","iopub.execute_input":"2024-02-04T03:40:52.073034Z","iopub.status.idle":"2024-02-04T03:41:22.277146Z","shell.execute_reply.started":"2024-02-04T03:40:52.073011Z","shell.execute_reply":"2024-02-04T03:41:22.275959Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from facenet_pytorch import InceptionResnetV1\n\n# Set num_classes to 7000\nnum_classes = 7000\n\n# Initialize the InceptionResnetV1 model\nmodel = InceptionResnetV1(pretrained='vggface2', device=device, classify=True, num_classes=num_classes)\n\nlayers_to_unfreeze = 15  # Number of layers to keep unfrozen\nfor param in model.parameters():\n    param.requires_grad = False\n\n# Unfreeze the last layers\nfor param in list(model.parameters())[-layers_to_unfreeze:]:\n    param.requires_grad = True\n    ","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:41:22.279005Z","iopub.execute_input":"2024-02-04T03:41:22.279391Z","iopub.status.idle":"2024-02-04T03:41:23.887171Z","shell.execute_reply.started":"2024-02-04T03:41:22.279357Z","shell.execute_reply":"2024-02-04T03:41:23.886370Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import torch.nn as nn\nfrom torch import optim\n\n#using cross-entropy for loss for classification\ncriterion  = nn.CrossEntropyLoss()\n\n\n#using the Adam optimizer\noptimizer = optim.Adam(model.parameters() , lr = 0.001)\n\ncheckpoint_path = '/kaggle/working/checkpoint.pth'\nbest_val_checkpoint_path = '/kaggle/working/best_val_checkpoint.pth'\n\ntorch.save(model.state_dict(), checkpoint_path)\ntorch.save(model.state_dict(), best_val_checkpoint_path)","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:41:23.888284Z","iopub.execute_input":"2024-02-04T03:41:23.888619Z","iopub.status.idle":"2024-02-04T03:41:24.300299Z","shell.execute_reply.started":"2024-02-04T03:41:23.888592Z","shell.execute_reply":"2024-02-04T03:41:24.299516Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Model traning script","metadata":{}},{"cell_type":"code","source":"from tqdm import tqdm\n\nnum_epochs = 2\n\ndef train_model(model):\n    model.to(device)\n    model.train()\n\n    train_losses = []\n    train_accuracies = []\n    val_losses = []\n    val_accuracies = []\n\n    best_val_accuracy = 0.0\n\n    for epoch in range(num_epochs):\n        running_loss = 0.0\n        correct_predictions = 0\n        total_samples = 0\n\n        # Wrap the training loader with tqdm for a progress bar\n        for batch in tqdm(train_loader):\n            if batch is not None:\n                inputs, labels = zip(*batch)\n                inputs = torch.stack(inputs).to(device)\n                labels = torch.stack(labels).to(device)\n\n                optimizer.zero_grad()\n\n                outputs = model(inputs)\n                loss = criterion(outputs, labels)\n                \n                loss.backward()\n                optimizer.step()\n\n                running_loss += loss.item()\n\n                _, predicted = torch.max(outputs, 1)\n                total_samples += labels.size(0)\n                correct_predictions += (predicted == labels).sum().item()\n\n        epoch_loss = running_loss / len(train_loader)\n        epoch_accuracy = correct_predictions / total_samples\n\n        train_losses.append(epoch_loss)\n        train_accuracies.append(epoch_accuracy)\n\n        print(f'Epoch {epoch + 1}/{num_epochs}, Training Loss: {epoch_loss:.4f}, Training Accuracy: {epoch_accuracy * 100:.2f}%')\n\n        # Save a checkpoint every epoch\n        torch.save({\n            'epoch': epoch,\n            'model_state_dict': model.state_dict(),\n            'optimizer_state_dict': optimizer.state_dict(),\n            'loss': epoch_loss,\n            'accuracy': epoch_accuracy\n        }, checkpoint_path)\n\n        # Check validation accuracy and save the best model\n        val_loss, val_accuracy = evaluate_model(model, validation_loader)\n\n        print(f'Epoch {epoch + 1}/{num_epochs}, Validation Loss: {val_loss:.4f}, Validation Accuracy: {val_accuracy * 100:.2f}%')\n\n\n        val_losses.append(val_loss)\n        val_accuracies.append(val_accuracy)\n\n        if val_accuracy > best_val_accuracy:\n            best_val_accuracy = val_accuracy\n            torch.save({\n                'epoch': epoch,\n                'model_state_dict': model.state_dict(),\n                'optimizer_state_dict': optimizer.state_dict(),\n                'loss': epoch_loss,\n                'accuracy': epoch_accuracy\n            }, best_val_checkpoint_path)\n\n    print('Training complete.')\n\n    # Plotting training loss and accuracy\n    plt.figure(figsize=(12, 4))\n    plt.subplot(1, 2, 1)\n    plt.plot(train_losses, label='Training Loss')\n    plt.title('Training Loss')\n    plt.xlabel('Epoch')\n    plt.ylabel('Loss')\n    plt.legend()\n\n    plt.subplot(1, 2, 2)\n    plt.plot(train_accuracies, label='Training Accuracy')\n    plt.title('Training Accuracy')\n    plt.xlabel('Epoch')\n    plt.ylabel('Accuracy')\n    plt.legend()\n\n    plt.show()\n\n    # Plotting validation loss and accuracy\n    plt.figure(figsize=(12, 4))\n    plt.subplot(1, 2, 1)\n    plt.plot(val_losses, label='Validation Loss')\n    plt.title('Validation Loss')\n    plt.xlabel('Epoch')\n    plt.ylabel('Loss')\n    plt.legend()\n\n    plt.subplot(1, 2, 2)\n    plt.plot(val_accuracies, label='Validation Accuracy')\n    plt.title('Validation Accuracy')\n    plt.xlabel('Epoch')\n    plt.ylabel('Accuracy')\n    plt.legend()\n\n    plt.show()\n\ndef evaluate_model(model, dataloader):\n    model.eval()\n    correct_predictions = 0\n    total_samples = 0\n\n    valLoss = 0\n\n    with torch.no_grad():\n        for batch in dataloader:\n            inputs, labels = zip(*batch)\n            inputs = torch.stack(inputs).to(device)\n            labels = torch.stack(labels).to(device)\n\n            outputs = model(inputs)\n            loss = criterion(outputs, labels)\n\n            valLoss += loss.item()\n\n            _, predicted = torch.max(outputs, 1)\n\n            total_samples += labels.size(0)\n            correct_predictions += (predicted == labels).sum().item()\n\n    valLoss = valLoss / len(validation_loader)\n    accuracy = correct_predictions / total_samples\n    return valLoss, accuracy\n\n# Usage example\ntrain_model(model)","metadata":{"execution":{"iopub.status.busy":"2024-02-04T03:41:24.301649Z","iopub.execute_input":"2024-02-04T03:41:24.301912Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def validate_model(model):\n    # Load model from the checkpoint\n    checkpoint = torch.load(best_val_checkpoint_path)\n    model.load_state_dict(checkpoint['model_state_dict'])\n    model.to(device)\n\n    model.eval()\n    running_loss = 0.0\n    correct_predictions = 0\n    total_samples = 0\n\n    with torch.no_grad():\n        for batch in tqdm(validation_loader):\n            if batch is not None:\n                val_inputs, val_labels = zip(*batch)\n                val_inputs = torch.stack(val_inputs).to(device)\n                val_labels = torch.stack(val_labels).to(device)\n\n                val_outputs = model(val_inputs)\n                val_loss = criterion(val_outputs, val_labels)\n\n                running_loss += val_loss.item()\n\n                _, val_predicted = torch.max(val_outputs, 1)\n                total_samples += val_labels.size(0)\n                correct_predictions += (val_predicted == val_labels).sum().item()\n\n    val_loss = running_loss / len(validation_loader)\n    val_accuracy = correct_predictions / total_samples\n\n    print(f'Validation Loss: {val_loss:.4f}, Validation Accuracy: {val_accuracy * 100:.2f}%')\n\n# Usage example\nvalidate_model(model)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def test_model(model):\n    # Load model from the best validation checkpoint\n    checkpoint = torch.load(best_val_checkpoint_path)\n    model.load_state_dict(checkpoint['model_state_dict'])\n    model.to(device)\n\n    model.eval()\n    running_loss = 0.0\n    correct_predictions = 0\n    total_samples = 0\n\n    with torch.no_grad():\n        for batch in tqdm(test_loader):\n            if batch is not None:\n                test_inputs, test_labels = zip(*batch)\n                test_inputs = torch.stack(test_inputs).to(device)\n                test_labels = torch.stack(test_labels).to(device)\n\n                test_outputs = model(test_inputs)\n                test_loss = criterion(test_outputs, test_labels)\n\n                running_loss += test_loss.item()\n\n                _, test_predicted = torch.max(test_outputs, 1)\n                total_samples += test_labels.size(0)\n                correct_predictions += (test_predicted == test_labels).sum().item()\n\n    # Calculate the mean loss and accuracy over all batches\n    test_loss = running_loss / len(test_loader)\n    test_accuracy = correct_predictions / total_samples\n\n    print(f'Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy * 100:.2f}%')\n\n# Usage example\ntest_model(model)","metadata":{"trusted":true},"execution_count":null,"outputs":[]}]}