Add finetuning script and initial implementation

README.md

@@ -1 +1,43 @@
-# DeepTurne
+# DeepTune
+
+## Description
+
+DeepTune is an open source Python package for fine-tuning computer vision (CV) based deep models using Siamese architecture with a triplet loss function. The package supports various model backbones and provides tools for data preprocessing and evaluation metrics.
+
+## Installation
+
+To install the package, use the following command:
+
+```bash
+pip install DeepTune
+```
+
+## Usage
+
+### Fine-tuning Models with Siamese Architecture and Triplet Loss
+
+Here is an example of how to use the package for fine-tuning models with Siamese architecture and triplet loss:
+
+```python
+import DeepTune
+from DeepTune import triplet_loss, backbones, data_preprocessing, evaluation_metrics
+
+# Load and preprocess data
+data = data_preprocessing.load_data('path/to/dataset')
+triplets = data_preprocessing.create_triplets(data)
+
+# Initialize model backbone
+model = backbones.get_model('resnet')
+
+# Compile model with triplet loss
+model.compile(optimizer='adam', loss=triplet_loss.triplet_loss)
+
+# Train model
+model.fit(triplets, epochs=10, batch_size=32)
+
+# Evaluate model
+metrics = evaluation_metrics.evaluate_model(model, triplets)
+print(metrics)
+```
+
+For more detailed usage and examples, please refer to the documentation.

deeptune/__init__.py

@@ -0,0 +1 @@
+# Initialize the DeepTune package

deeptune/backbones.py

@@ -0,0 +1,45 @@
+import tensorflow as tf
+from tensorflow.keras.applications import ResNet50, VGG16, InceptionV3, MobileNet, EfficientNetB0
+from tensorflow.keras.models import Model
+from tensorflow.keras.layers import Input, GlobalAveragePooling2D, Dense, Lambda
+from tensorflow.keras.applications import InceptionResNetV2
+from tensorflow.keras import backend as K
+
+def get_model(backbone_name, input_shape=(224, 224, 3)):
+    """
+    Get the model backbone.
+
+    Args:
+        backbone_name: The name of the model backbone.
+        input_shape: The input shape of the model.
+
+    Returns:
+        The model backbone.
+    """
+    if backbone_name == 'resnet':
+        base_model = ResNet50(weights='imagenet', include_top=False, input_shape=input_shape)
+    elif backbone_name == 'vgg':
+        base_model = VGG16(weights='imagenet', include_top=False, input_shape=input_shape)
+    elif backbone_name == 'inception':
+        base_model = InceptionV3(weights='imagenet', include_top=False, input_shape=input_shape)
+    elif backbone_name == 'mobilenet':
+        base_model = MobileNet(weights='imagenet', include_top=False, input_shape=input_shape)
+    elif backbone_name == 'efficientnet':
+        base_model = EfficientNetB0(weights='imagenet', include_top=False, input_shape=input_shape)
+    elif backbone_name == 'facenet':
+        base_model = InceptionResNetV2(weights=None, include_top=False, input_shape=input_shape)
+        base_model.load_weights('path/to/facenet_weights.h5')
+    elif backbone_name == 'arcface':
+        base_model = InceptionResNetV2(weights=None, include_top=False, input_shape=input_shape)
+        base_model.load_weights('path/to/arcface_weights.h5')
+    else:
+        raise ValueError(f"Unsupported backbone: {backbone_name}")
+
+    model = tf.keras.Sequential([
+        base_model,
+        GlobalAveragePooling2D(),
+        Dense(128, activation='relu'),
+        Lambda(lambda x: tf.math.l2_normalize(x, axis=1))
+    ])
+
+    return model

deeptune/data_preprocessing.py

@@ -0,0 +1,101 @@
+import tensorflow as tf
+from tensorflow.keras.preprocessing.image import ImageDataGenerator
+import numpy as np
+import os
+from sklearn.model_selection import train_test_split
+
+def load_data(data_dir, img_size=(224, 224)):
+    """
+    Load and preprocess data from the given directory.
+
+    Args:
+        data_dir: The directory containing the dataset.
+        img_size: The target size of the images.
+
+    Returns:
+        A tuple of (images, labels).
+    """
+    images = []
+    labels = []
+    class_names = os.listdir(data_dir)
+    for label, class_name in enumerate(class_names):
+        class_dir = os.path.join(data_dir, class_name)
+        for img_name in os.listdir(class_dir):
+            img_path = os.path.join(class_dir, img_name)
+            img = tf.keras.preprocessing.image.load_img(img_path, target_size=img_size)
+            img = tf.keras.preprocessing.image.img_to_array(img)
+            img = img / 255.0  # Normalize the image
+            images.append(img)
+            labels.append(label)
+    return np.array(images), np.array(labels)
+
+def create_triplets(images, labels):
+    """
+    Create triplets of images (anchor, positive, negative) for training.
+
+    Args:
+        images: The array of images.
+        labels: The array of labels.
+
+    Returns:
+        A tuple of (anchor_images, positive_images, negative_images).
+    """
+    anchor_images = []
+    positive_images = []
+    negative_images = []
+    num_classes = len(np.unique(labels))
+    for i in range(len(images)):
+        anchor = images[i]
+        anchor_label = labels[i]
+        positive_indices = np.where(labels == anchor_label)[0]
+        negative_indices = np.where(labels != anchor_label)[0]
+        positive = images[np.random.choice(positive_indices)]
+        negative = images[np.random.choice(negative_indices)]
+        anchor_images.append(anchor)
+        positive_images.append(positive)
+        negative_images.append(negative)
+    return np.array(anchor_images), np.array(positive_images), np.array(negative_images)
+
+def preprocess_data(data_dir, img_size=(224, 224), test_size=0.2):
+    """
+    Preprocess data for training and testing.
+
+    Args:
+        data_dir: The directory containing the dataset.
+        img_size: The target size of the images.
+        test_size: The proportion of the dataset to include in the test split.
+
+    Returns:
+        A tuple of (train_triplets, test_triplets).
+    """
+    images, labels = load_data(data_dir, img_size)
+    train_images, test_images, train_labels, test_labels = train_test_split(images, labels, test_size=test_size, stratify=labels)
+    train_triplets = create_triplets(train_images, train_labels)
+    test_triplets = create_triplets(test_images, test_labels)
+    return train_triplets, test_triplets
+
+def augment_data(images, rotation_range=20, width_shift_range=0.2, height_shift_range=0.2, horizontal_flip=True):
+    """
+    Augment the images to increase the diversity of the training data.
+
+    Args:
+        images: The array of images to augment.
+        rotation_range: The range of rotation for augmentation.
+        width_shift_range: The range of width shift for augmentation.
+        height_shift_range: The range of height shift for augmentation.
+        horizontal_flip: Whether to randomly flip images horizontally.
+
+    Returns:
+        The augmented images.
+    """
+    datagen = ImageDataGenerator(rotation_range=rotation_range,
+                                 width_shift_range=width_shift_range,
+                                 height_shift_range=height_shift_range,
+                                 horizontal_flip=horizontal_flip)
+    augmented_images = []
+    for img in images:
+        img = np.expand_dims(img, axis=0)
+        aug_iter = datagen.flow(img)
+        aug_img = next(aug_iter)[0]
+        augmented_images.append(aug_img)
+    return np.array(augmented_images)

deeptune/datasets.py

@@ -0,0 +1,52 @@
+import tensorflow_datasets as tfds
+
+def load_dataset(dataset_name, split='train'):
+    """
+    Load a dataset using TensorFlow Datasets (TFDS).
+
+    Args:
+        dataset_name: The name of the dataset to load.
+        split: The split of the dataset to load (e.g., 'train', 'test').
+
+    Returns:
+        A tf.data.Dataset object.
+    """
+    dataset, info = tfds.load(dataset_name, split=split, with_info=True)
+    return dataset, info
+
+def preprocess_dataset(dataset, img_size=(224, 224)):
+    """
+    Preprocess the dataset by resizing and normalizing the images.
+
+    Args:
+        dataset: The tf.data.Dataset object to preprocess.
+        img_size: The target size of the images.
+
+    Returns:
+        A preprocessed tf.data.Dataset object.
+    """
+    def _preprocess(image, label):
+        image = tf.image.resize(image, img_size)
+        image = tf.cast(image, tf.float32) / 255.0
+        return image, label
+
+    dataset = dataset.map(_preprocess, num_parallel_calls=tf.data.experimental.AUTOTUNE)
+    dataset = dataset.cache()
+    dataset = dataset.shuffle(buffer_size=1000)
+    dataset = dataset.batch(32)
+    dataset = dataset.prefetch(buffer_size=tf.data.experimental.AUTOTUNE)
+    return dataset
+
+def get_lfw_dataset(split='train'):
+    """
+    Get the LFW dataset.
+
+    Args:
+        split: The split of the dataset to load (e.g., 'train', 'test').
+
+    Returns:
+        A preprocessed tf.data.Dataset object.
+    """
+    dataset, info = load_dataset('lfw', split=split)
+    dataset = preprocess_dataset(dataset)
+    return dataset

deeptune/evaluation_metrics.py

@@ -0,0 +1,42 @@
+import tensorflow as tf
+from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, average_precision_score
+
+def evaluate_model(model, triplets, labels):
+    """
+    Evaluate the model using various metrics.
+
+    Args:
+        model: The trained model.
+        triplets: The triplets of images (anchor, positive, negative).
+        labels: The true labels of the triplets.
+
+    Returns:
+        A dictionary containing the evaluation metrics.
+    """
+    anchor, positive, negative = triplets
+    anchor_embeddings = model.predict(anchor)
+    positive_embeddings = model.predict(positive)
+    negative_embeddings = model.predict(negative)
+
+    # Calculate distances
+    pos_distances = tf.reduce_sum(tf.square(anchor_embeddings - positive_embeddings), axis=-1)
+    neg_distances = tf.reduce_sum(tf.square(anchor_embeddings - negative_embeddings), axis=-1)
+
+    # Calculate metrics
+    accuracy = accuracy_score(labels, pos_distances < neg_distances)
+    precision = precision_score(labels, pos_distances < neg_distances)
+    recall = recall_score(labels, pos_distances < neg_distances)
+    f1 = f1_score(labels, pos_distances < neg_distances)
+    roc_auc = roc_auc_score(labels, pos_distances < neg_distances)
+    mean_avg_precision = average_precision_score(labels, pos_distances < neg_distances)
+    triplet_loss_value = tf.reduce_mean(tf.maximum(pos_distances - neg_distances + 1.0, 0.0))
+
+    return {
+        'accuracy': accuracy,
+        'precision': precision,
+        'recall': recall,
+        'f1_score': f1,
+        'roc_auc': roc_auc,
+        'mean_avg_precision': mean_avg_precision,
+        'triplet_loss': triplet_loss_value
+    }

deeptune/test_finetuning.py

@@ -0,0 +1,22 @@
+import tensorflow as tf
+from DeepTune import triplet_loss, backbones, data_preprocessing, evaluation_metrics, datasets
+
+def test_finetuning():
+    # Load and preprocess data
+    train_triplets, test_triplets = data_preprocessing.preprocess_data('path/to/dataset')
+
+    # Initialize model backbone
+    model = backbones.get_model('resnet')
+
+    # Compile model with triplet loss
+    model.compile(optimizer='adam', loss=triplet_loss.triplet_loss)
+
+    # Train model
+    model.fit(train_triplets, epochs=5, batch_size=32)
+
+    # Evaluate model
+    metrics = evaluation_metrics.evaluate_model(model, test_triplets)
+    print(metrics)
+
+if __name__ == "__main__":
+    test_finetuning()

deeptune/triplet_loss.py

@@ -0,0 +1,25 @@
+import tensorflow as tf
+
+def triplet_loss(anchor, positive, negative, margin=1.0):
+    """
+    Custom triplet loss function.
+
+    Args:
+        anchor: The anchor input tensor.
+        positive: The positive input tensor.
+        negative: The negative input tensor.
+        margin: The margin by which the positive example should be closer to the anchor than the negative example.
+
+    Returns:
+        The triplet loss value.
+    """
+    # Calculate the distance between the anchor and the positive example
+    pos_dist = tf.reduce_sum(tf.square(anchor - positive), axis=-1)
+
+    # Calculate the distance between the anchor and the negative example
+    neg_dist = tf.reduce_sum(tf.square(anchor - negative), axis=-1)
+
+    # Compute the triplet loss
+    loss = tf.maximum(pos_dist - neg_dist + margin, 0.0)
+
+    return tf.reduce_mean(loss)