Siamese model code

.gitignore

@@ -5,3 +5,5 @@ src/model/train/**
 src/model/test/**
 src/model/david/**
 venv/**
+src/model/keras/**
+src/model/*.h5

requirements.txt

@@ -1,4 +1,42 @@
-
+absl-py==2.0.0
+astunparse==1.6.3
+cachetools==5.3.2
+certifi==2023.7.22
+charset-normalizer==3.3.2
+flatbuffers==23.5.26
+gast==0.4.0
+google-auth==2.23.4
+google-auth-oauthlib==1.0.0
+google-pasta==0.2.0
+grpcio==1.59.2
+h5py==3.10.0
+idna==3.4
+jax==0.4.20
 keras==2.12.0
-Pillow==10.0.1
-tensorflow-macos==2.12.0
+libclang==16.0.6
+Markdown==3.5.1
+MarkupSafe==2.1.3
+ml-dtypes==0.3.1
+numpy==1.23.5
+oauthlib==3.2.2
+opencv-python==4.8.1.78
+opt-einsum==3.3.0
+packaging==23.2
+protobuf==4.25.0
+pyasn1==0.5.0
+pyasn1-modules==0.3.0
+requests==2.31.0
+requests-oauthlib==1.3.1
+rsa==4.9
+scipy==1.11.3
+six==1.16.0
+tensorboard==2.12.3
+tensorboard-data-server==0.7.2
+tensorflow-estimator==2.12.0
+tensorflow-io-gcs-filesystem==0.34.0
+tensorflow-macos==2.12.0
+termcolor==2.3.0
+typing_extensions==4.8.0
+urllib3==2.0.7
+Werkzeug==3.0.1
+wrapt==1.14.1

src/model/Siamese_Network.py

@@ -0,0 +1,232 @@
+# https://keras.io/examples/vision/siamese_network/
+
+import matplotlib.pyplot as plt
+import numpy as np
+import os
+import random
+import tensorflow as tf
+from pathlib import Path
+from tensorflow.keras import applications
+from tensorflow.keras import layers
+from tensorflow.keras import losses
+from tensorflow.keras import optimizers
+from tensorflow.keras import metrics
+from tensorflow.keras import Model
+from tensorflow.keras.applications import resnet
+
+target_shape = (200, 200)
+
+cache_dir = Path().resolve() / "keras"
+anchor_images_path = cache_dir / "left"
+positive_images_path = cache_dir / "right"
+
+
+def preprocess_image(filename):
+    """
+    Load the specified file as a JPEG image, preprocess it and
+    resize it to the target shape.
+    """
+
+    image_string = tf.io.read_file(filename)
+    image = tf.image.decode_jpeg(image_string, channels=3)
+    image = tf.image.convert_image_dtype(image, tf.float32)
+    image = tf.image.resize(image, target_shape)
+    return image
+
+
+def preprocess_triplets(anchor, positive, negative):
+    """
+    Given the filenames corresponding to the three images, load and
+    preprocess them.
+    """
+
+    return (
+        preprocess_image(anchor),
+        preprocess_image(positive),
+        preprocess_image(negative),
+    )
+
+
+# We need to make sure both the anchor and positive images are loaded in
+# sorted order so we can match them together.
+anchor_images = sorted(
+    [str(anchor_images_path / f) for f in os.listdir(anchor_images_path)]
+)
+
+positive_images = sorted(
+    [str(positive_images_path / f) for f in os.listdir(positive_images_path)]
+)
+
+all_images = anchor_images + positive_images
+random.shuffle(all_images)
+print(all_images)
+
+image_count = len(anchor_images)
+print(image_count)
+
+anchor_dataset = tf.data.Dataset.from_tensor_slices(anchor_images)
+positive_dataset = tf.data.Dataset.from_tensor_slices(positive_images)
+negative_dataset = tf.data.Dataset.from_tensor_slices(all_images[:100])
+
+print(len(negative_dataset))
+print(len(anchor_dataset))
+
+dataset = tf.data.Dataset.zip((anchor_dataset, positive_dataset, negative_dataset))
+dataset = dataset.shuffle(buffer_size=1024)
+dataset = dataset.map(preprocess_triplets)
+
+# Let's now split our dataset in train and validation.
+train_dataset = dataset.take(round(image_count * 0.8))
+val_dataset = dataset.skip(round(image_count * 0.8))
+
+batch_size = 16
+
+train_dataset = train_dataset.batch(batch_size, drop_remainder=False)
+train_dataset = train_dataset.prefetch(8)
+
+val_dataset = val_dataset.batch(batch_size, drop_remainder=False)
+val_dataset = val_dataset.prefetch(8)
+
+
+def visualize(anchor, positive, negative):
+    """Visualize a few triplets from the supplied batches."""
+
+    def show(ax, image):
+        ax.imshow(image)
+        ax.get_xaxis().set_visible(False)
+        ax.get_yaxis().set_visible(False)
+
+    fig = plt.figure(figsize=(9, 9))
+
+    axs = fig.subplots(3, 3)
+    for i in range(3):
+        show(axs[i, 0], anchor[i])
+        show(axs[i, 1], positive[i])
+        show(axs[i, 2], negative[i])
+
+
+visualize(*list(train_dataset.take(1).as_numpy_iterator())[0])
+
+base_cnn = resnet.ResNet50(
+    weights="imagenet", input_shape=target_shape + (3,), include_top=False
+)
+
+flatten = layers.Flatten()(base_cnn.output)
+dense1 = layers.Dense(512, activation="relu")(flatten)
+dense1 = layers.BatchNormalization()(dense1)
+dense2 = layers.Dense(256, activation="relu")(dense1)
+dense2 = layers.BatchNormalization()(dense2)
+output = layers.Dense(256)(dense2)
+
+embedding = Model(base_cnn.input, output, name="Embedding")
+
+trainable = False
+for layer in base_cnn.layers:
+    if layer.name == "conv5_block1_out":
+        trainable = True
+    layer.trainable = trainable
+
+
+class DistanceLayer(layers.Layer):
+    """
+    This layer is responsible for computing the distance between the anchor
+    embedding and the positive embedding, and the anchor embedding and the
+    negative embedding.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def call(self, anchor, positive, negative):
+        ap_distance = tf.reduce_sum(tf.square(anchor - positive), -1)
+        an_distance = tf.reduce_sum(tf.square(anchor - negative), -1)
+        return (ap_distance, an_distance)
+
+
+anchor_input = layers.Input(name="anchor", shape=target_shape + (3,))
+positive_input = layers.Input(name="positive", shape=target_shape + (3,))
+negative_input = layers.Input(name="negative", shape=target_shape + (3,))
+
+distances = DistanceLayer()(
+    embedding(resnet.preprocess_input(anchor_input)),
+    embedding(resnet.preprocess_input(positive_input)),
+    embedding(resnet.preprocess_input(negative_input)),
+)
+
+siamese_network = Model(
+    inputs=[anchor_input, positive_input, negative_input], outputs=distances
+)
+
+
+class SiameseModel(Model):
+    """The Siamese Network model with a custom training and testing loops.
+
+    Computes the triplet loss using the three embeddings produced by the
+    Siamese Network.
+
+    The triplet loss is defined as:
+       L(A, P, N) = max(‖f(A) - f(P)‖² - ‖f(A) - f(N)‖² + margin, 0)
+    """
+
+    def __init__(self, siamese_network, margin=0.5):
+        super().__init__()
+        self.siamese_network = siamese_network
+        self.margin = margin
+        self.loss_tracker = metrics.Mean(name="loss")
+
+    def call(self, inputs):
+        return self.siamese_network(inputs)
+
+    def train_step(self, data):
+        # GradientTape is a context manager that records every operation that
+        # you do inside. We are using it here to compute the loss so we can get
+        # the gradients and apply them using the optimizer specified in
+        # `compile()`.
+        with tf.GradientTape() as tape:
+            loss = self._compute_loss(data)
+
+        # Storing the gradients of the loss function with respect to the
+        # weights/parameters.
+        gradients = tape.gradient(loss, self.siamese_network.trainable_weights)
+
+        # Applying the gradients on the model using the specified optimizer
+        self.optimizer.apply_gradients(
+            zip(gradients, self.siamese_network.trainable_weights)
+        )
+
+        # Let's update and return the training loss metric.
+        self.loss_tracker.update_state(loss)
+        return {"loss": self.loss_tracker.result()}
+
+    def test_step(self, data):
+        loss = self._compute_loss(data)
+
+        # Let's update and return the loss metric.
+        self.loss_tracker.update_state(loss)
+        return {"loss": self.loss_tracker.result()}
+
+    def _compute_loss(self, data):
+        # The output of the network is a tuple containing the distances
+        # between the anchor and the positive example, and the anchor and
+        # the negative example.
+        ap_distance, an_distance = self.siamese_network(data)
+
+        # Computing the Triplet Loss by subtracting both distances and
+        # making sure we don't get a negative value.
+        loss = ap_distance - an_distance
+        loss = tf.maximum(loss + self.margin, 0.0)
+        return loss
+
+    @property
+    def metrics(self):
+        # We need to list our metrics here so the `reset_states()` can be
+        # called automatically.
+        return [self.loss_tracker]
+
+
+# Training
+
+siamese_model = SiameseModel(siamese_network)
+siamese_model.compile(optimizer=optimizers.Adam(0.0001))  # 0..0001
+siamese_model.fit(train_dataset, epochs=7, validation_data=val_dataset)
+embedding.save("siamese_feature.h5")

src/model/Siamese_Predictor.py

@@ -0,0 +1,48 @@
+# from Siamese_Network import preprocess_image
+import tensorflow as tf
+from tensorflow.keras.applications import resnet
+import argparse
+from tensorflow.keras import metrics
+
+
+def preprocess_image(filename):
+    """
+    Load the specified file as a JPEG image, preprocess it and
+    resize it to the target shape.
+    """
+    target_shape = (200, 200)
+
+    image_string = tf.io.read_file(filename)
+    image = tf.image.decode_jpeg(image_string, channels=3)
+    image = tf.image.convert_image_dtype(image, tf.float32)
+    image = tf.image.resize(image, target_shape)
+    return image
+
+
+# Initialize parser
+parser = argparse.ArgumentParser()
+parser.add_argument(
+    "filename", type=str, help="Path to the input image file (e.g., img.jpg)"
+)
+
+args = parser.parse_args()
+
+# load the model
+embedding = tf.keras.models.load_model("siamese_feature.h5")
+
+
+david_1 = preprocess_image(args.filename)
+david_2 = preprocess_image("david_base.jpg")
+# add a dimension to the tensor
+david_1 = tf.expand_dims(david_1, axis=0)
+david_2 = tf.expand_dims(david_2, axis=0)
+
+anchor_embedding, positive_embedding = (
+    embedding(resnet.preprocess_input(david_1)),
+    embedding(resnet.preprocess_input(david_2)),
+)
+
+cosine_similarity = metrics.CosineSimilarity()
+
+positive_similarity = cosine_similarity(anchor_embedding, positive_embedding)
+print("Similarity score: ", positive_similarity.numpy())