Facerecogniton.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db806a69",
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import numpy as np\n",
    "import torch \n",
    "import cv2\n",
    "import os\n",
    "import uuid \n",
    "import path\n",
    "import pydot\n",
    "from typing import List, Tuple\n",
    "from matplotlib.pyplot import imshow\n",
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt\n",
    "import PIL.Image\n",
    "import pathlib\n",
    "import shutil\n",
    "from mtcnn import MTCNN\n",
    "from tensorflow.keras.layers import Layer\n",
    "import tqdm\n",
    "import tensorflow as tf\n",
    "from tensorflow import keras\n",
    "from tensorflow.keras.metrics import Precision, Recall\n",
    "from tensorflow.keras.initializers import glorot_uniform\n",
    "from tensorflow.keras.models import Model\n",
    "from tensorflow.keras.losses import Loss\n",
    "from tensorflow.keras.preprocessing import image\n",
    "from tensorflow.keras.layers import Input, Add, Dense, Activation, ZeroPadding2D, BatchNormalization, Flatten, Conv2D, AveragePooling2D, MaxPooling2D, GlobalMaxPooling2D\n",
    "from tensorflow.keras.initializers import glorot_uniform\n",
    "from tensorflow.keras.models import Model, load_model\n",
    "from tensorflow.keras.applications.imagenet_utils import preprocess_input\n",
    "from IPython.display import SVG\n",
    "import scipy.misc\n",
    "import tensorflow.keras.backend as K\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79f8d262",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Avoid OOM errors by setting GPU Memory Consumption Growth\n",
    "gpus = tf.config.experimental.list_physical_devices('GPU')\n",
    "for gpu in gpus: \n",
    "    tf.config.experimental.set_memory_growth(gpu, True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c9485db0",
   "metadata": {},
   "outputs": [],
   "source": [
    "#setting up the Path \n",
    "POS_PATH='/Users/siddharth/Desktop/Siamesedata/positive'\n",
    "NEG_PATH='/Users/siddharth/Desktop/Siamesedata/negative'\n",
    "ANC_PATH='/Users/siddharth/Desktop/Siamesedata/anchor'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "67290e31",
   "metadata": {},
   "outputs": [],
   "source": [
    "#seting up directories \n",
    "os.makedirs(POS_PATH,exist_ok= True)\n",
    "os.makedirs(NEG_PATH,exist_ok= True)\n",
    "os.makedirs(ANC_PATH,exist_ok= True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6eed3f3e",
   "metadata": {},
   "source": [
    "# Input Additional Pictures from your webcam in the data by pressing key \"A\" to add in anchor images or Press \"P\" to add in Postive images "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a164c75",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import required libraries\n",
    "import os\n",
    "import cv2\n",
    "import uuid\n",
    "from mtcnn import MTCNN\n",
    "\n",
    "# Function to crop face from the frame\n",
    "def crop_face(frame, face_box, target_size=(100, 100)):\n",
    "    x1, y1, width, height = face_box\n",
    "    x2, y2 = x1 + width, y1 + height\n",
    "    face = frame[y1:y2, x1:x2]\n",
    "    return cv2.resize(face, target_size)\n",
    "\n",
    "# Establish a connection to the webcam\n",
    "cap = cv2.VideoCapture(1)\n",
    "\n",
    "# Initialize MTCNN for face detection\n",
    "detector = MTCNN()\n",
    "\n",
    "while cap.isOpened(): \n",
    "    ret, frame = cap.read()\n",
    "   \n",
    "    # Detect faces in the frame\n",
    "    faces = detector.detect_faces(frame)\n",
    "\n",
    "    # Draw rectangles around the detected faces\n",
    "    for face in faces:\n",
    "        x, y, width, height = face['box']\n",
    "        cv2.rectangle(frame, (x, y), (x+width, y+height), (0, 255, 0), 2)\n",
    "    \n",
    "    # Show the frame with rectangles\n",
    "    cv2.imshow('Image Collection', frame)\n",
    "    \n",
    "    # Capture images\n",
    "    if cv2.waitKey(1) & 0xFF == ord('a'):  # Anchor\n",
    "        if len(faces) > 0:\n",
    "            # Get the first detected face and crop it to 100x100 pixels\n",
    "            face = crop_face(frame, faces[0]['box'], target_size=(100, 100))\n",
    "            # Create the unique file path \n",
    "            imgname = os.path.join(ANC_PATH, '{}.png'.format(uuid.uuid1()))\n",
    "            # Write out anchor image\n",
    "            cv2.imwrite(imgname, face)\n",
    "    \n",
    "    if cv2.waitKey(1) & 0xFF == ord('p'):  # Positive\n",
    "        if len(faces) > 0:\n",
    "            # Get the first detected face and crop it to 100x100 pixels\n",
    "            face = crop_face(frame, faces[0]['box'], target_size=(100, 100))\n",
    "            # Create the unique file path \n",
    "            imgname = os.path.join(POS_PATH, '{}.png'.format(uuid.uuid1()))\n",
    "            # Write out positive image\n",
    "            cv2.imwrite(imgname, face)\n",
    "    \n",
    "    # Breaking gracefully\n",
    "    if cv2.waitKey(1) & 0xFF == ord('q'):\n",
    "        break\n",
    "        \n",
    "# Release the webcam\n",
    "cap.release()\n",
    "# Close the image show frame\n",
    "cv2.destroyAllWindows()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a05b264",
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess(file_path):\n",
    "    \n",
    "    # Read in image from file path\n",
    "    byte_img = tf.io.read_file(file_path)\n",
    "    # Load in the image \n",
    "    img = tf.io.decode_jpeg(byte_img)\n",
    "    \n",
    "    # Preprocessing steps - resizing the image to be 100x100x3\n",
    "    img = tf.image.resize(img, (100,100))\n",
    "    # Scale image to be between 0 and 1 \n",
    "    img = img / 255.0\n",
    "\n",
    "    # Return image\n",
    "    return img"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f4bc9d6",
   "metadata": {},
   "source": [
    "# Joining the (Anchor,Positive) and (Anchor,Negative) pair in a tensor to make a Dataset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a9dbcfe2",
   "metadata": {},
   "outputs": [],
   "source": [
    "anchor = tf.data.Dataset.list_files(ANC_PATH + '/*.png')\n",
    "positive = tf.data.Dataset.list_files(POS_PATH + '/*.png')\n",
    "negative = tf.data.Dataset.list_files(NEG_PATH + '/*.png')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "55ee68e4",
   "metadata": {},
   "outputs": [],
   "source": [
    "positives = tf.data.Dataset.zip((anchor, positive, tf.data.Dataset.from_tensor_slices(tf.ones(len(anchor)))))\n",
    "negatives = tf.data.Dataset.zip((anchor, negative, tf.data.Dataset.from_tensor_slices(tf.zeros(len(anchor)))))\n",
    "data = positives.concatenate(negatives)\n",
    "samples = data.as_numpy_iterator()\n",
    "example = samples.next()\n",
    "example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "47a781b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess_twin(input_img, validation_img, label):\n",
    "    return(preprocess(input_img), preprocess(validation_img), label)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1cbf0a4a",
   "metadata": {},
   "source": [
    "# Data Pipeline "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c92ecd4a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Build dataloader pipeline\n",
    "data = data.map(preprocess_twin)\n",
    "data = data.shuffle(buffer_size=10000)\n",
    "\n",
    "# Take a subset of the dataset and cache it\n",
    "train_data = data.take(round(len(data) * 0.7)).cache()\n",
    "train_data = train_data.batch(16)\n",
    "train_data = train_data.prefetch(8)\n",
    "\n",
    "# Take another subset for testing and cache it\n",
    "test_data = data.skip(round(len(data) * 0.7)).take(round(len(data) * 0.3)).cache()\n",
    "test_data = test_data.batch(16)\n",
    "test_data = test_data.prefetch(8)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "81884e6f",
   "metadata": {},
   "source": [
    "# ResNet50 (generates 512 embedding of images) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8515b322",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from typing import Tuple\n",
    "from tensorflow.keras.layers import (\n",
    "    Input,\n",
    "    Conv2D,\n",
    "    BatchNormalization,\n",
    "    Activation,\n",
    "    MaxPooling2D,\n",
    "    ZeroPadding2D,\n",
    "    Add,\n",
    "    AveragePooling2D,\n",
    "    Flatten,\n",
    "    Dense,\n",
    ")\n",
    "from tensorflow.keras.initializers import glorot_uniform\n",
    "from tensorflow.keras.models import Model\n",
    "\n",
    "\n",
    "def identity_block(X, level, block, filters):\n",
    "    \"\"\"\n",
    "    Implementation of the identity block as defined in the ResNet model.\n",
    "\n",
    "    Arguments:\n",
    "    X -- input tensor of shape (m, n_H_prev, n_W_prev, n_C_prev)\n",
    "    level -- level/layer of the block (used for naming)\n",
    "    block -- block identifier (used for naming)\n",
    "    filters -- python list of integers, defining the number of filters in the CONV layers\n",
    "\n",
    "    Returns:\n",
    "    X -- output of the identity block, tensor of shape (m, n_H, n_W, n_C)\n",
    "    \"\"\"\n",
    "\n",
    "    # Retrieve filters\n",
    "    f1, f2, f3 = filters\n",
    "\n",
    "    # Define names for the layers using the provided level and block numbers\n",
    "    conv_name = 'res' + str(level) + '_' + str(block) + '_{}_{}'\n",
    "\n",
    "    # Save the input value to be used in the skip connection\n",
    "    X_shortcut = X\n",
    "\n",
    "    # First component of main path\n",
    "    X = Conv2D(filters=f1, kernel_size=(1, 1), strides=(1, 1), padding='valid',\n",
    "               name=conv_name.format(1, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(1, 'bn'))(X)\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    # Second component of main path\n",
    "    X = Conv2D(filters=f2, kernel_size=(3, 3), strides=(1, 1), padding='same',\n",
    "               name=conv_name.format(2, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(2, 'bn'))(X)\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    # Third component of main path\n",
    "    X = Conv2D(filters=f3, kernel_size=(1, 1), strides=(1, 1), padding='valid',\n",
    "               name=conv_name.format(3, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(3, 'bn'))(X)\n",
    "\n",
    "    # Add shortcut value to main path and pass it through a ReLU activation\n",
    "    X = Add()([X, X_shortcut])\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    return X\n",
    "\n",
    "\n",
    "def convolutional_block(X, level, block, filters, s=2):\n",
    "    \"\"\"\n",
    "    Implementation of the convolutional block as defined in the ResNet model.\n",
    "\n",
    "    Arguments:\n",
    "    X -- input tensor of shape (m, n_H_prev, n_W_prev, n_C_prev)\n",
    "    level -- level/layer of the block (used for naming)\n",
    "    block -- block identifier (used for naming)\n",
    "    filters -- python list of integers, defining the number of filters in the CONV layers\n",
    "    s -- stride for the first convolutional layer in the block\n",
    "\n",
    "    Returns:\n",
    "    X -- output of the convolutional block, tensor of shape (m, n_H, n_W, n_C)\n",
    "    \"\"\"\n",
    "\n",
    "    # Retrieve filters\n",
    "    f1, f2, f3 = filters\n",
    "\n",
    "    # Define names for the layers using the provided level and block numbers\n",
    "    conv_name = 'res' + str(level) + '_' + str(block) + '_{}_{}'\n",
    "\n",
    "    # Save the input value to be used in the skip connection\n",
    "    X_shortcut = X\n",
    "\n",
    "    # First component of main path\n",
    "    X = Conv2D(filters=f1, kernel_size=(1, 1), strides=(s, s), padding='valid',\n",
    "               name=conv_name.format(1, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(1, 'bn'))(X)\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    # Second component of main path\n",
    "    X = Conv2D(filters=f2, kernel_size=(3, 3), strides=(1, 1), padding='same',\n",
    "               name=conv_name.format(2, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(2, 'bn'))(X)\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    # Third component of main path\n",
    "    X = Conv2D(filters=f3, kernel_size=(1, 1), strides=(1, 1), padding='valid',\n",
    "               name=conv_name.format(3, 'conv'), kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "    X = BatchNormalization(axis=3, name=conv_name.format(3, 'bn'))(X)\n",
    "\n",
    "    # Shortcut path\n",
    "    X_shortcut = Conv2D(filters=f3, kernel_size=(1, 1), strides=(s, s), padding='valid',\n",
    "                        name=conv_name.format('shortcut', 'conv'), kernel_initializer=glorot_uniform(seed=0))(X_shortcut)\n",
    "    X_shortcut = BatchNormalization(axis=3, name=conv_name.format('shortcut', 'bn'))(X_shortcut)\n",
    "\n",
    "    # Add shortcut value to main path and pass it through a ReLU activation\n",
    "    X = Add()([X, X_shortcut])\n",
    "    X = Activation('relu')(X)\n",
    "\n",
    "    return X\n",
    "\n",
    "\n",
    "def ResNet50(input_size: Tuple[int, int, int], embedding_size: int) -> Model:\n",
    "    \"\"\"\n",
    "    Implementation of the ResNet50 architecture.\n",
    "\n",
    "    Arguments:\n",
    "    input_size -- shape of the input images (height, width, channels)\n",
    "    embedding_size -- size of the embedding vector\n",
    "\n",
    "    Returns:\n",
    "    model -- a Model() instance in Keras\n",
    "    \"\"\"\n",
    "\n",
    "    # Define the input placeholder as a tensor with shape input_size\n",
    "    X_input = Input(input_size)\n",
    "\n",
    "    # Zero-padding\n",
    "    X = ZeroPadding2D((3, 3))(X_input)\n",
    "\n",
    "    # Stage 1\n",
    "    X = Conv2D(64, (7, 7), strides=(2, 2), name='conv1')(X)\n",
    "    X = BatchNormalization(axis=3, name='bn_conv1')(X)\n",
    "    X = Activation('relu')(X)\n",
    "    X = MaxPooling2D((3, 3), strides=(2, 2))(X)\n",
    "\n",
    "    # Stage 2\n",
    "    X = convolutional_block(X, level=2, block=1, filters=[64, 64, 256], s=1)\n",
    "    X = identity_block(X, level=2, block=2, filters=[64, 64, 256])\n",
    "    X = identity_block(X, level=2, block=3, filters=[64, 64, 256])\n",
    "\n",
    "    # Stage 3\n",
    "    X = convolutional_block(X, level=3, block=1, filters=[128, 128, 512], s=2)\n",
    "    X = identity_block(X, level=3, block=2, filters=[128, 128, 512])\n",
    "    X = identity_block(X, level=3, block=3, filters=[128, 128, 512])\n",
    "    X = identity_block(X, level=3, block=4, filters=[128, 128, 512])\n",
    "\n",
    "    # Stage 4\n",
    "    X = convolutional_block(X, level=4, block=1, filters=[256, 256, 1024], s=2)\n",
    "    X = identity_block(X, level=4, block=2, filters=[256, 256, 1024])\n",
    "    X = identity_block(X, level=4, block=3, filters=[256, 256, 1024])\n",
    "    X = identity_block(X, level=4, block=4, filters=[256, 256, 1024])\n",
    "    X = identity_block(X, level=4, block=5, filters=[256, 256, 1024])\n",
    "    X = identity_block(X, level=4, block=6, filters=[256, 256, 1024])\n",
    "\n",
    "    # Stage 5\n",
    "    X = convolutional_block(X, level=5, block=1, filters=[512, 512, 2048], s=2)\n",
    "    X = identity_block(X, level=5, block=2, filters=[512, 512, 2048])\n",
    "    X = identity_block(X, level=5, block=3, filters=[512, 512, 2048])\n",
    "\n",
    "    # Average pooling\n",
    "    X = AveragePooling2D((2, 2), name='avg_pool')(X)\n",
    "\n",
    "    # Flatten the output\n",
    "    X = Flatten()(X)\n",
    "\n",
    "    # Fully connected layer\n",
    "    X = Dense(embedding_size, activation='linear', name='fc' + str(embedding_size),\n",
    "              kernel_initializer=glorot_uniform(seed=0))(X)\n",
    "\n",
    "    # Create model instance\n",
    "    model = Model(inputs=X_input, outputs=X, name='ResNet50')\n",
    "\n",
    "    return model\n",
    "\n",
    "\n",
    "# Create ResNet50 model for input images of size 100x100 and embedding size of 128\n",
    "input_size = (100, 100, 3)\n",
    "embedding_size = 512\n",
    "model = ResNet50(input_size, embedding_size)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4706847a",
   "metadata": {},
   "source": [
    "# Siamese Model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb342a2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Siamese L1 Distance class\n",
    "class L1Dist(Layer):\n",
    "    \n",
    "    # Init method - inheritance\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__()\n",
    "       \n",
    "    # Magic happens here - similarity calculation\n",
    "    def call(self, input_embedding, validation_embedding):\n",
    "        return tf.math.abs(input_embedding - validation_embedding)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "212d5270",
   "metadata": {},
   "outputs": [],
   "source": [
    "l1 = L1Dist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "edaea696",
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_siamese_model():\n",
    "    # Anchor image input in the network\n",
    "    input_image = Input(name='input_img', shape=(100, 100, 3))\n",
    "\n",
    "    # Validation image in the network\n",
    "    validation_image = Input(name='validation_img', shape=(100, 100, 3))\n",
    "\n",
    "    # Get embeddings using your ResNet model\n",
    "    inp_embedding = model(input_image)\n",
    "    val_embedding = model(validation_image)\n",
    "\n",
    "    # Siamese L1 Distance class\n",
    "    class L1Dist(Layer):\n",
    "        def __init__(self, **kwargs):\n",
    "            super().__init__()\n",
    "\n",
    "        def call(self, input_embedding, validation_embedding):\n",
    "            return tf.math.abs(input_embedding - validation_embedding)\n",
    "\n",
    "    l1 = L1Dist()\n",
    "    distances = l1(inp_embedding, val_embedding)\n",
    "\n",
    "    # Classification layer\n",
    "    classifier = Dense(1, activation='sigmoid')(distances)\n",
    "\n",
    "    return Model(inputs=[input_image, validation_image], outputs=classifier, name='SiameseNetwork')\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a89fde68",
   "metadata": {},
   "outputs": [],
   "source": [
    "siamese_model = make_siamese_model()\n",
    "siamese_model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3ef7ddb8",
   "metadata": {},
   "outputs": [],
   "source": [
    "binary_cross_loss = tf.losses.BinaryCrossentropy()\n",
    "opt = tf.keras.optimizers.legacy.Adam(1e-4)  # Adjusted learning rate: 0.00005"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6d24432a",
   "metadata": {},
   "outputs": [],
   "source": [
    "checkpoint_dir = '/Users/siddharth/Desktop/Siamesedata/training_checkpoints'\n",
    "checkpoint_prefix = os.path.join(checkpoint_dir, 'ckpt')\n",
    "checkpoint = tf.train.Checkpoint(opt=opt, siamese_model=siamese_model)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "336bdc16",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_batch = train_data.as_numpy_iterator()\n",
    "batch_1 = test_batch.next()\n",
    "X = batch_1[:2]\n",
    "y = batch_1[2]\n",
    "y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4bb2e780",
   "metadata": {},
   "outputs": [],
   "source": [
    "tf.losses.BinaryCrossentropy??"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9f556177",
   "metadata": {},
   "source": [
    "# Compiling the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "12859de0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the Siamese model\n",
    "siamese_model = make_siamese_model()\n",
    "\n",
    "# Compile the model for training\n",
    "siamese_model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b45b7400",
   "metadata": {},
   "source": [
    "# Custom call back when loss reduces to a certain point "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fce6df58",
   "metadata": {},
   "outputs": [],
   "source": [
    "class CustomStopCallback(tf.keras.callbacks.Callback):\n",
    "    def on_epoch_end(self, epoch, logs=None):\n",
    "        if logs is None:\n",
    "            logs = {}\n",
    "        \n",
    "        # Stop training if the loss is below or equal to 0.002, FAR is less than 0.1, and FRR is less than 2.5\n",
    "        if logs.get('loss') <= 0.002 and logs.get('FAR') < 0.1 and logs.get('FRR') < 2.5:\n",
    "            print(\"\\nReached the desired loss, FAR, and FRR. Stopping training.\")\n",
    "            self.model.stop_training = True\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63ca2327",
   "metadata": {},
   "source": [
    "# FAR and FRR calculation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "58c8c5b9",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_far_frr(y_true, y_pred):\n",
    "    # Convert the predictions to binary (0 or 1) based on a threshold of 0.5\n",
    "    y_pred_binary = tf.where(y_pred >= 0.5, 1.0, 0.0)\n",
    "\n",
    "    # Calculate the confusion matrix\n",
    "    tp = tf.reduce_sum(tf.multiply(y_true, y_pred_binary))\n",
    "    tn = tf.reduce_sum(tf.multiply(1 - y_true, 1 - y_pred_binary))\n",
    "    fp = tf.reduce_sum(tf.multiply(1 - y_true, y_pred_binary))\n",
    "    fn = tf.reduce_sum(tf.multiply(y_true, 1 - y_pred_binary))\n",
    "\n",
    "    # Calculate FAR and FRR\n",
    "    far = fp / (fp + tn)\n",
    "    frr = fn / (fn + tp)\n",
    "\n",
    "    return far, frr"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "55ac599e",
   "metadata": {},
   "source": [
    "# Training the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c9faaa0",
   "metadata": {},
   "outputs": [],
   "source": [
    "@tf.function\n",
    "def train_step(batch):\n",
    "    \n",
    "    # Record all of our operations \n",
    "    with tf.GradientTape() as tape:     \n",
    "        # Get anchor and positive/negative image\n",
    "        X = batch[:2]\n",
    "        # Get label\n",
    "        y = batch[2]\n",
    "        \n",
    "        # Forward pass\n",
    "        yhat = siamese_model(X, training=True)\n",
    "        # Calculate loss\n",
    "        loss = binary_cross_loss(y, yhat)\n",
    "    print(loss)\n",
    "        \n",
    "    # Calculate gradients\n",
    "    grad = tape.gradient(loss, siamese_model.trainable_variables)\n",
    "    \n",
    "    # Calculate updated weights and apply to siamese model\n",
    "    opt.apply_gradients(zip(grad, siamese_model.trainable_variables))\n",
    "        \n",
    "    # Return loss\n",
    "    return loss"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c86b36a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import metric calculations\n",
    "from tensorflow.keras.metrics import Precision, Recall"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfb23860",
   "metadata": {},
   "outputs": [],
   "source": [
    "import tensorflow as tf\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import csv\n",
    "\n",
    "# Define the training function\n",
    "def train(data, EPOCHS):\n",
    "    # Check if there is a previous checkpoint to resume training\n",
    "    latest_checkpoint = tf.train.latest_checkpoint(checkpoint_dir)\n",
    "    if latest_checkpoint:\n",
    "        checkpoint.restore(latest_checkpoint)\n",
    "        print(f\"Resuming training from checkpoint: {latest_checkpoint}\")\n",
    "\n",
    "    # Create a checkpoint manager\n",
    "    checkpoint_manager = tf.train.CheckpointManager(checkpoint, checkpoint_dir, max_to_keep=3)\n",
    "\n",
    "    # Initialize lists to store loss and metrics values after each epoch\n",
    "    loss_values = []\n",
    "    recall_values = []\n",
    "    precision_values = []\n",
    "    far_values = []\n",
    "    frr_values = []\n",
    "\n",
    "    # Loop through epochs\n",
    "    for epoch in range(1, EPOCHS + 1):\n",
    "        print('\\n Epoch {}/{}'.format(epoch, EPOCHS))\n",
    "        progbar = tf.keras.utils.Progbar(len(data))\n",
    "        epoch_loss = 0.0\n",
    "\n",
    "        # Creating a metric object\n",
    "        r = tf.keras.metrics.Recall()\n",
    "        p = tf.keras.metrics.Precision()\n",
    "\n",
    "        # Loop through each batch\n",
    "        for idx, batch in enumerate(data):\n",
    "            # Run train step here and accumulate the loss\n",
    "            loss = train_step(batch)\n",
    "            epoch_loss += loss\n",
    "\n",
    "            yhat = siamese_model.predict(batch[:2])\n",
    "            r.update_state(batch[2], yhat)\n",
    "            p.update_state(batch[2], yhat)\n",
    "            progbar.update(idx + 1)\n",
    "\n",
    "        # Calculate the average loss for the epoch\n",
    "        avg_loss = epoch_loss / len(data)\n",
    "\n",
    "        # Calculate FAR and FRR for the current epoch\n",
    "        y_true = np.concatenate([batch[2] for batch in data], axis=0)\n",
    "        y_pred = np.concatenate([siamese_model.predict(batch[:2]) for batch in data], axis=0)\n",
    "        far, frr = calculate_far_frr(y_true, y_pred)\n",
    "\n",
    "        # Append metrics to lists\n",
    "        loss_values.append(avg_loss.numpy())\n",
    "        recall_values.append(r.result().numpy())\n",
    "        precision_values.append(p.result().numpy())\n",
    "        far_values.append(far)\n",
    "        frr_values.append(frr)\n",
    "\n",
    "        # Print metrics for the epoch\n",
    "        print(f\"Avg. Loss: {avg_loss.numpy()}, Recall: {r.result().numpy()}, Precision: {p.result().numpy()}, FAR: {far}, FRR: {frr}\")\n",
    "\n",
    "        # Save the checkpoint\n",
    "        checkpoint_manager.save()\n",
    "\n",
    "        # Save loss values to a CSV file after each epoch\n",
    "        with open('loss_values.csv', 'w', newline='') as csvfile:\n",
    "            writer = csv.writer(csvfile)\n",
    "            writer.writerow(loss_values)\n",
    "\n",
    "    # Plot the loss graph after all epochs\n",
    "    plt.figure()\n",
    "    plt.plot(loss_values)\n",
    "    plt.xlabel('Epoch')\n",
    "    plt.ylabel('Loss')\n",
    "    plt.title('Loss Graph')\n",
    "    plt.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "811ce1dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "EPOCHS = 50\n",
    "train(train_data, EPOCHS)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2aee9b15",
   "metadata": {},
   "source": [
    "# Saving the model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3f4fbaba",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Save weights\n",
    "siamese_model.save('siamesemodelv2.h5')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "984570e5",
   "metadata": {},
   "source": [
    "# Reloading the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02cbcce1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Reload model \n",
    "siamese_model = tf.keras.models.load_model('siamesemodelv2.h5', \n",
    "                                   custom_objects={'L1Dist':L1Dist, 'BinaryCrossentropy':tf.losses.BinaryCrossentropy})\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f290d66e",
   "metadata": {},
   "source": [
    "# Testing the model on the test data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "99504bf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import metric calculations\n",
    "from tensorflow.keras.metrics import Precision, Recall\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f01392f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get a batch of test data\n",
    "test_input, test_val, y_true = test_data.as_numpy_iterator().next()\n",
    "y_hat = siamese_model.predict([test_input, test_val])\n",
    "# Post processing the results \n",
    "[1 if prediction > 0.5 else 0 for prediction in y_hat ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "850191f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert y_hat probabilities to binary predictions using a threshold of 0.9\n",
    "y_pred_binary = [1 if prediction > 0.9 else 0 for prediction in y_hat]\n",
    "\n",
    "# Calculate the number of genuine samples (true label = 1) and impostor samples (true label = 0)\n",
    "num_genuine = sum(y_true)\n",
    "num_impostor = len(y_true) - num_genuine\n",
    "\n",
    "# Initialize variables to count false acceptance and false rejection\n",
    "false_acceptance = 0\n",
    "false_rejection = 0\n",
    "\n",
    "# Loop through the predictions and compare with true labels to calculate FAR and FRR\n",
    "for y_true_val, y_pred_val in zip(y_true, y_pred_binary):\n",
    "    if y_true_val == 1 and y_pred_val == 0:\n",
    "        false_rejection += 1\n",
    "    elif y_true_val == 0 and y_pred_val == 1:\n",
    "        false_acceptance += 1\n",
    "\n",
    "# Calculate FAR and FRR as ratios\n",
    "far = false_acceptance / num_genuine\n",
    "frr = false_rejection / num_impostor\n",
    "\n",
    "print(\"False Acceptance Rate (FAR): {:.2%}\".format(far))\n",
    "print(\"False Rejection Rate (FRR): {:.2%}\".format(frr))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a3dad98",
   "metadata": {},
   "outputs": [],
   "source": [
    "y_true"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b167de3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating a metric object \n",
    "m = Recall()\n",
    "\n",
    "# Calculating the recall value \n",
    "m.update_state(y_true, y_hat)\n",
    "\n",
    "# Return Recall Result\n",
    "m.result().numpy()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f8949528",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creating a metric object \n",
    "m = Precision()\n",
    "\n",
    "# Calculating the recall value \n",
    "m.update_state(y_true, y_hat)\n",
    "\n",
    "# Return Recall Result\n",
    "m.result().numpy()"
   ]
  },
  {
   "cell_type": "raw",
   "id": "2a117dbe",
   "metadata": {},
   "source": [
    "r = Recall()\n",
    "p = Precision()\n",
    "\n",
    "for test_input, test_val, y_true in test_data.as_numpy_iterator():\n",
    "    yhat = siamese_model.predict([test_input, test_val])\n",
    "    r.update_state(y_true, yhat)\n",
    "    p.update_state(y_true,yhat) \n",
    "\n",
    "print(r.result().numpy(), p.result().numpy())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "02abd269",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set plot size \n",
    "plt.figure(figsize=(10,8))\n",
    "\n",
    "# Set first subplot\n",
    "plt.subplot(1,2,1)\n",
    "plt.imshow(test_input[1])\n",
    "\n",
    "# Set second subplot\n",
    "plt.subplot(1,2,2)\n",
    "plt.imshow(test_val[0])\n",
    "\n",
    "# Renders cleanly\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "54b6038f",
   "metadata": {},
   "source": [
    "# Take Image of person You need to verify "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a94f745d",
   "metadata": {},
   "outputs": [],
   "source": [
    "import cv2\n",
    "from mtcnn import MTCNN\n",
    "from PIL import Image\n",
    "\n",
    "def capture_and_process_face(save_folder):\n",
    "    # Initialize the camera capture\n",
    "    cap = cv2.VideoCapture(0)\n",
    "\n",
    "    # Initialize the MTCNN detector\n",
    "    detector = MTCNN()\n",
    "\n",
    "    while True:\n",
    "        # Capture frame-by-frame from the camera\n",
    "        ret, frame = cap.read()\n",
    "\n",
    "        # Detect faces in the frame using MTCNN\n",
    "        faces = detector.detect_faces(frame)\n",
    "\n",
    "        for idx, face_info in enumerate(faces):\n",
    "            # Extract the face bounding box coordinates\n",
    "            x, y, width, height = face_info['box']\n",
    "\n",
    "            # Draw a rectangle around the detected face on the frame\n",
    "            cv2.rectangle(frame, (x, y), (x + width, y + height), (0, 255, 0), 2)\n",
    "\n",
    "        # Display the original frame with the detected faces\n",
    "        cv2.imshow('Camera', frame)\n",
    "\n",
    "        # Check for user input to capture the image (press 'c')\n",
    "        key = cv2.waitKey(1) & 0xFF\n",
    "        if key == ord('c'):\n",
    "            # Save the first detected face (assuming there's only one face in the frame)\n",
    "            if len(faces) > 0:\n",
    "                x, y, width, height = faces[0]['box']\n",
    "                face = frame[y:y+height, x:x+width]\n",
    "                face_resized = cv2.resize(face, (100, 100))\n",
    "                save_path = f\"{save_folder}/face_0.png\"\n",
    "                cv2.imwrite(save_path, face_resized)\n",
    "                print(\"Face captured and saved!\")\n",
    "            else:\n",
    "                print(\"No face detected!\")\n",
    "\n",
    "        # Press 'q' to exit the loop\n",
    "        elif key == ord('q'):\n",
    "            break\n",
    "\n",
    "    # Release the camera and close the window\n",
    "    cap.release()\n",
    "    cv2.destroyAllWindows()\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    # Specify the folder to save the cropped and resized face images\n",
    "    save_faces_folder = \"/Users/siddharth/Siamesedata/application_data/verification_images\"\n",
    "\n",
    "    # Call the function to capture from the camera and process faces\n",
    "    capture_and_process_face(save_faces_folder)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "545dee5b",
   "metadata": {},
   "source": [
    "# Prediction "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c39feca2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make predictions with reloaded model\n",
    "siamese_model.predict([test_input, test_val])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7dad2d2a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import cv2\n",
    "import mtcnn\n",
    "import os\n",
    "import numpy as np\n",
    "import tensorflow as tf\n",
    "\n",
    "# Load the Siamese model\n",
    "siamese_model = tf.keras.models.load_model('siamesemodelv2.h5', custom_objects={'L1Dist': L1Dist, 'BinaryCrossentropy': tf.losses.BinaryCrossentropy})\n",
    "\n",
    "# Load the pre-trained MTCNN face detection model\n",
    "face_detector = mtcnn.MTCNN()\n",
    "\n",
    "def preprocess_image(image):\n",
    "    # Resize the image to the required input size of the Siamese model\n",
    "    image = cv2.resize(image, (100, 100))\n",
    "    # Convert image to float and normalize to [0, 1]\n",
    "    image = image.astype(np.float32) / 255.0\n",
    "    # Expand dimensions to make it a batch of 1\n",
    "    image = np.expand_dims(image, axis=0)\n",
    "    return image\n",
    "\n",
    "def verify_face(image1, image2):\n",
    "    # Preprocess both images\n",
    "    image1 = preprocess_image(image1)\n",
    "    image2 = preprocess_image(image2)\n",
    "\n",
    "    # Use the Siamese model to get the similarity score\n",
    "    similarity_score = siamese_model.predict([image1, image2])[0][0]\n",
    "\n",
    "    return similarity_score\n",
    "\n",
    "def real_time_facial_verification():\n",
    "    # Path to the folder with correct images\n",
    "    verification_images_folder = '/Users/siddharth/Siamesedata/application_data/verification_images'\n",
    "    correct_images = []\n",
    "\n",
    "    # Load the correct images from the folder\n",
    "    for file in os.listdir(verification_images_folder):\n",
    "        image_path = os.path.join(verification_images_folder, file)\n",
    "        img = cv2.imread(image_path)\n",
    "        if img is not None:\n",
    "            correct_images.append(img)\n",
    "\n",
    "    # Open the webcam\n",
    "    cap = cv2.VideoCapture(0)\n",
    "\n",
    "    while True:\n",
    "        # Read a frame from the webcam\n",
    "        ret, frame = cap.read()\n",
    "\n",
    "        # Detect faces in the frame using MTCNN\n",
    "        faces = face_detector.detect_faces(frame)\n",
    "\n",
    "        for face in faces:\n",
    "            # Extract the bounding box coordinates for the face\n",
    "            x, y, width, height = face['box']\n",
    "            x2, y2 = x + width, y + height\n",
    "\n",
    "            # Crop the face region from the frame\n",
    "            face_region = frame[y:y2, x:x2]\n",
    "\n",
    "            # Perform verification on each correct image\n",
    "            verified = False\n",
    "            for reference_face in correct_images:\n",
    "                similarity_score = verify_face(face_region, reference_face)\n",
    "                if similarity_score >= 0.5:  # You can adjust the threshold as needed\n",
    "                    verified = True\n",
    "                    break\n",
    "\n",
    "            # Draw the bounding box around the face\n",
    "            if verified:\n",
    "                cv2.putText(frame, 'Verified', (x, y - 10), cv2.FONT_HERSHEY_, 0.7, (0, 255, 0), 2)\n",
    "            else:\n",
    "                cv2.putText(frame, 'Unknown', (x, y - 10), cv2.FONT_HERSHEY_COMPLEX, 0.7, (0, 0, 255), 2)\n",
    "\n",
    "            cv2.rectangle(frame, (x, y), (x2, y2), (0, 255, 0), 2)\n",
    "\n",
    "        # Display the frame\n",
    "        cv2.imshow('Facial Verification', frame)\n",
    "\n",
    "        # Exit the loop if 'q' is pressed\n",
    "        if cv2.waitKey(1) & 0xFF == ord('q'):\n",
    "            break\n",
    "\n",
    "    # Release the video capture and close the OpenCV windows\n",
    "    cap.release()\n",
    "    cv2.destroyAllWindows()\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    real_time_facial_verification()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}