signature-verification-with-triplet-loss.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "_cell_guid": "b1076dfc-b9ad-4769-8c92-a6c4dae69d19",
    "_kg_hide-output": true,
    "_uuid": "8f2839f25d086af736a60e9eeb907d3b93b6e0e5"
   },
   "outputs": [],
   "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"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Import Libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ[\"TF_CPP_MIN_LOG_LEVEL\"] = \"3\"\n",
    "\n",
    "import random\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import cv2\n",
    "import matplotlib.pyplot as plt\n",
    "from PIL import Image\n",
    "\n",
    "import tensorflow as tf\n",
    "from tensorflow.keras import layers, models, optimizers, losses, regularizers, metrics\n",
    "from tensorflow.keras.utils import to_categorical, plot_model\n",
    "from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping\n",
    "from tensorflow.keras.preprocessing.image import ImageDataGenerator\n",
    "from tensorflow.keras.applications.efficientnet import EfficientNetB7, preprocess_input\n",
    "\n",
    "from mlxtend.plotting import plot_confusion_matrix\n",
    "from sklearn.metrics import roc_curve, auc, classification_report, confusion_matrix\n",
    "from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score, roc_auc_score"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Load Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "root_dir = \"/kaggle/input/persons/CV_Signature_Dataset/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_csv = []\n",
    "for folder in os.listdir(root_dir):\n",
    "    for file in os.listdir(os.path.join(root_dir, folder, \"Train\")):\n",
    "        if file.endswith(\".csv\"):\n",
    "            train_csv.append(os.path.join(root_dir, folder, \"Train\", file))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_csv = []\n",
    "for folder in os.listdir(root_dir):\n",
    "    for file in os.listdir(os.path.join(root_dir, folder, \"Test\")):\n",
    "        if file.endswith(\".csv\"):\n",
    "            test_csv.append(os.path.join(root_dir, folder, \"Test\", file))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_triplets = []\n",
    "for csv in train_csv:\n",
    "    df = pd.read_csv(csv)\n",
    "    real_images = df[df[\"label\"] == \"real\"][\"image_name\"].tolist()\n",
    "    forged_images = df[df[\"label\"] == \"forged\"][\"image_name\"].tolist()\n",
    "\n",
    "    for anchor in real_images:\n",
    "        for image in real_images:\n",
    "            anchor_image = os.path.join(root_dir, image[:7], \"Train\", image)\n",
    "            positive_image = os.path.join(root_dir, image[:7], \"Train\", random.choice(real_images))\n",
    "            negative_image = os.path.join(root_dir, image[:7], \"Train\", random.choice(forged_images))\n",
    "            train_triplets.append((anchor_image, positive_image, negative_image))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_triplets = []\n",
    "for csv in test_csv:\n",
    "    df = pd.read_csv(csv)\n",
    "    real_images = df[df[\"label\"] == \"real\"][\"image_name\"].tolist()\n",
    "    forged_images = df[df[\"label\"] == \"forged\"][\"image_name\"].tolist()\n",
    "\n",
    "    for anchor in real_images:\n",
    "        for image in real_images:\n",
    "            anchor_image = os.path.join(root_dir, image[:7], \"Test\", image)\n",
    "            positive_image = os.path.join(root_dir, image[:7], \"Test\", random.choice(real_images))\n",
    "            negative_image = os.path.join(root_dir, image[:7], \"Test\", random.choice(forged_images))\n",
    "            test_triplets.append((anchor_image, positive_image, negative_image))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Train Size:\", len(train_triplets))\n",
    "print(\"Test Size:\", len(test_triplets))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Visualization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def plot_images(images):\n",
    "    plt.figure(figsize=(20, 6))\n",
    "\n",
    "    for i in range(8):\n",
    "        idx = np.random.randint(len(images))\n",
    "        plt.subplot(3, 8, i + 1)\n",
    "        img = Image.open(images[idx][0])\n",
    "        plt.imshow(img)\n",
    "        plt.axis(\"off\")\n",
    "        plt.title(\"Anchor\")\n",
    "\n",
    "        plt.subplot(3, 8, i + 9)\n",
    "        img = Image.open(images[idx][1])\n",
    "        plt.imshow(img)\n",
    "        plt.axis(\"off\")\n",
    "        plt.title(\"Positive\")\n",
    "\n",
    "        plt.subplot(3, 8, i + 17)\n",
    "        img = Image.open(images[idx][2])\n",
    "        plt.imshow(img)\n",
    "        plt.axis(\"off\")\n",
    "        plt.title(\"Negative\")\n",
    "\n",
    "    plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_images(train_triplets)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_image(index):\n",
    "    path = os.path.join(root_dir, index)\n",
    "    image = cv2.imread(path)\n",
    "    image = cv2.resize(image, (128, 128))\n",
    "    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n",
    "    return image"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_batch(triplet_list, batch_size):\n",
    "    batch_steps = len(triplet_list) // batch_size\n",
    "    \n",
    "    for i in range(batch_steps+1):\n",
    "        anchor   = []\n",
    "        positive = []\n",
    "        negative = []\n",
    "        \n",
    "        j = i * batch_size\n",
    "        while j < (i + 1) * batch_size and j < len(triplet_list):\n",
    "            a, p, n = triplet_list[j]\n",
    "            anchor.append(read_image(a))\n",
    "            positive.append(read_image(p))\n",
    "            negative.append(read_image(n))\n",
    "            j += 1\n",
    "            \n",
    "        anchor = np.array(anchor)\n",
    "        positive = np.array(positive)\n",
    "        negative = np.array(negative)\n",
    "        \n",
    "        anchor = preprocess_input(anchor)\n",
    "        positive = preprocess_input(positive)\n",
    "        negative = preprocess_input(negative)\n",
    "        \n",
    "        yield ([anchor, positive, negative])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pretrained_model = EfficientNetB7(weights=\"imagenet\",\n",
    "                            input_shape=(128, 128, 3),\n",
    "                            include_top=False,\n",
    "                            pooling=\"avg\")\n",
    "\n",
    "for i in range(len(pretrained_model.layers) - 25):\n",
    "    pretrained_model.layers[i].trainable = False\n",
    "\n",
    "encode_model = models.Sequential([\n",
    "    pretrained_model,\n",
    "    layers.Flatten(),\n",
    "    layers.Dense(512, activation='relu'),\n",
    "    layers.BatchNormalization(),\n",
    "    layers.Dropout(0.3),\n",
    "    layers.Dense(256, activation=\"relu\"),\n",
    "    layers.BatchNormalization(),\n",
    "    layers.Dropout(0.3),\n",
    "    layers.Dense(128, activation=\"relu\"),\n",
    "    layers.BatchNormalization(),\n",
    "    layers.Dropout(0.3),\n",
    "    layers.Dense(128)\n",
    "], name=\"Embedding\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "encode_model.summary()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class DistanceLayer(layers.Layer):\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "\n",
    "    def call(self, anchor, positive, negative):\n",
    "        ap_distance = tf.reduce_sum(tf.square(anchor - positive), -1)\n",
    "        an_distance = tf.reduce_sum(tf.square(anchor - negative), -1)\n",
    "        return ap_distance, an_distance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "anchor_input = layers.Input(name='anchor', shape=(128, 128, 3), dtype=tf.uint8)\n",
    "positive_input = layers.Input(name='positive', shape=(128, 128, 3), dtype=tf.uint8)\n",
    "negative_input = layers.Input(name='negative', shape=(128, 128, 3), dtype=tf.uint8)\n",
    "\n",
    "distances = DistanceLayer()(\n",
    "    encode_model(preprocess_input(anchor_input)),\n",
    "    encode_model(preprocess_input(positive_input)),\n",
    "    encode_model(preprocess_input(negative_input))\n",
    ")\n",
    "\n",
    "siamese_net = models.Model(\n",
    "    inputs=[anchor_input,\n",
    "            positive_input,\n",
    "            negative_input],\n",
    "    outputs=distances,\n",
    "    name = \"Siamese_Network\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plot_model(siamese_net, show_shapes=True, show_layer_names=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class SiameseModel(models.Model):\n",
    "    def __init__(self, siamese_network, margin=1.0):\n",
    "        super(SiameseModel, self).__init__()\n",
    "        self.siamese_network = siamese_network\n",
    "        self.margin = margin\n",
    "        self.loss_tracker = metrics.Mean(name=\"loss\")\n",
    "        self.accuracy_tracker = metrics.Mean(name=\"accuracy\")\n",
    "\n",
    "    def call(self, inputs):\n",
    "        return self.siamese_network(inputs)\n",
    "\n",
    "    def train_step(self, data):\n",
    "        with tf.GradientTape() as tape:\n",
    "            loss = self._compute_loss(data)\n",
    "            acc = self._compute_accuracy(data)\n",
    "            \n",
    "        gradients = tape.gradient(loss, self.siamese_network.trainable_weights)\n",
    "        self.optimizer.apply_gradients(zip(gradients, self.siamese_network.trainable_weights))\n",
    "        \n",
    "        self.loss_tracker.update_state(loss)\n",
    "        self.accuracy_tracker.update_state(acc)\n",
    "\n",
    "        return {\"loss\": self.loss_tracker.result(), \"accuracy\": self.accuracy_tracker.result()}\n",
    "\n",
    "    def test_step(self, data):\n",
    "        loss = self._compute_loss(data)\n",
    "        acc = self._compute_accuracy(data)\n",
    "\n",
    "        self.loss_tracker.update_state(loss)\n",
    "        self.accuracy_tracker.update_state(acc)\n",
    "\n",
    "        return {\"loss\": self.loss_tracker.result(), \"accuracy\": self.accuracy_tracker.result()}\n",
    "\n",
    "    def _compute_loss(self, data):\n",
    "        ap_distance, an_distance = self.siamese_network(data)\n",
    "        loss = tf.maximum(ap_distance - an_distance + self.margin, 0.0)\n",
    "        return loss\n",
    "\n",
    "    def _compute_accuracy(self, data):\n",
    "        ap_distance, an_distance = self.siamese_network(data)\n",
    "        return tf.reduce_mean(tf.cast(ap_distance < an_distance, tf.float32))\n",
    "\n",
    "    @property\n",
    "    def metrics(self):\n",
    "        return [self.loss_tracker, self.accuracy_tracker]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "siamese_model = SiameseModel(siamese_net)\n",
    "siamese_model.compile(optimizer=optimizers.Adam(learning_rate=0.001))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_on_triplets(batch_size):\n",
    "    pos_scores = [] \n",
    "    neg_scores = []\n",
    "\n",
    "    for data in get_batch(test_triplets, batch_size=batch_size):\n",
    "        prediction = siamese_model.predict(data, verbose=0)\n",
    "        pos_scores += list(prediction[0])\n",
    "        neg_scores += list(prediction[1])\n",
    "    \n",
    "    accuracy = np.sum(np.array(pos_scores) < np.array(neg_scores)) / len(pos_scores)\n",
    "    return accuracy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Train"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "epochs = 10\n",
    "batch_size = 128\n",
    "\n",
    "train_loss = []\n",
    "train_accuracy = []\n",
    "val_loss = []\n",
    "val_accuracy = []\n",
    "\n",
    "for epoch in range(epochs):\n",
    "    epoch_train_loss = []\n",
    "    epoch_train_accuracy = []\n",
    "    \n",
    "    for data in get_batch(train_triplets, batch_size=batch_size):\n",
    "        results = siamese_model.train_on_batch(data)\n",
    "        epoch_train_loss.append(results[0])\n",
    "        epoch_train_accuracy.append(results[1])\n",
    "    \n",
    "    avg_train_loss = sum(epoch_train_loss) / len(epoch_train_loss)\n",
    "    avg_train_accuracy = sum(epoch_train_accuracy) / len(epoch_train_accuracy)\n",
    "    train_loss.append(avg_train_loss)\n",
    "    train_accuracy.append(avg_train_accuracy)\n",
    "    \n",
    "    epoch_val_loss = []\n",
    "    epoch_val_accuracy = []\n",
    "    \n",
    "    for data in get_batch(test_triplets, batch_size=batch_size):\n",
    "        results = siamese_model.test_on_batch(data)\n",
    "        epoch_val_loss.append(results[0])\n",
    "        epoch_val_accuracy.append(results[1])\n",
    "    \n",
    "    avg_val_loss = sum(epoch_val_loss) / len(epoch_val_loss)\n",
    "    avg_val_accuracy = sum(epoch_val_accuracy) / len(epoch_val_accuracy)\n",
    "    val_loss.append(avg_val_loss)\n",
    "    val_accuracy.append(avg_val_accuracy)\n",
    "\n",
    "    print(f\"Epoch: {epoch} | Train Loss: {avg_train_loss:.5f} | Validation Loss: {avg_val_loss:.5f} | Train Accuracy: {avg_train_accuracy:.5f} | Test Accuracy: {avg_val_accuracy:.5f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.figure(figsize=(8, 6))\n",
    "plt.plot(train_loss)\n",
    "plt.plot(val_loss)\n",
    "plt.legend([\"Train Loss\", \"Validation Loss\"])\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.figure(figsize=(8, 6))\n",
    "plt.plot(train_accuracy)\n",
    "plt.plot(val_accuracy)\n",
    "plt.legend([\"Train Accuracy\", \"Validation Accuracy\"])\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pos_list = []\n",
    "neg_list = []\n",
    "\n",
    "for data in get_batch(test_triplets, batch_size=256):\n",
    "    anchor, positive, negative = data\n",
    "\n",
    "    anchor_pred = encode_model.predict(anchor, verbose=0)\n",
    "    pos_pred = encode_model.predict(positive, verbose=0)\n",
    "    neg_pred = encode_model.predict(negative, verbose=0)\n",
    "\n",
    "    pos_distance = np.sum(np.square(anchor_pred - pos_pred), axis=-1)\n",
    "    pos_prediction = np.where(pos_distance <= 1.3, 0, 1)\n",
    "    pos_list.append(pos_prediction)\n",
    "    \n",
    "    neg_distance = np.sum(np.square(anchor_pred - neg_pred), axis=-1)\n",
    "    neg_prediction = np.where(neg_distance <= 1.3, 0, 1)\n",
    "    neg_list.append(neg_prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_labels = np.array([0] * len(pos_list[0]) + [1] * len(neg_list[0]))\n",
    "pred_labels = np.append(pos_list, neg_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_precision_score = precision_score(test_labels, pred_labels, average='weighted')\n",
    "model_f1_score = f1_score(test_labels, pred_labels, average='weighted')\n",
    "model_recall_score = recall_score(test_labels, pred_labels, average='weighted')\n",
    "model_accuracy_score = accuracy_score(test_labels, pred_labels)\n",
    "\n",
    "print(f\"Precision Score = {model_precision_score * 100:.2f}%\")\n",
    "print(f\"F1 Score = {model_f1_score * 100:.2f}%\")\n",
    "print(f\"Recall Score = {model_recall_score * 100:.2f}%\")\n",
    "print(f\"Accuracy Score = {model_accuracy_score * 100:.2f}%\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cm = confusion_matrix(test_labels, pred_labels)\n",
    "fig, ax = plot_confusion_matrix(conf_mat=cm, show_absolute=True, show_normed=True, colorbar=True)\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}