team14: verio - working version 1

app.py

@@ -1,37 +1,49 @@
 """A local gradio app that detect matching images using FHE."""
 
-from PIL import Image
 import os
+from pathlib import Path
 import shutil
-import subprocess
 import time
-import gradio as gr
-import numpy
+from typing import Tuple
 import requests
+
+import numpy as np
+
+import subprocess
+import gradio as gr
 from itertools import chain
+import matplotlib.pyplot as plt
+import matplotlib.image as img
+import numpy as np
+from PIL import Image
+import torch
+import torchvision.transforms as transforms
+import torchvision.models as models
+import cv2
+from facenet_pytorch import InceptionResnetV1
+from concrete.ml.deployment import FHEModelClient, FHEModelServer
+from client_server_interface import FHEClient
 
 from common import (
-    AVAILABLE_MATCHERS,
     CLIENT_TMP_PATH,
-    ENCRYPTED_OUTPUT_NAME,
-    ENCRYPTED_QUERY_NAME,
-    ENCRYPTED_REFERENCE_NAME,
-    SERVER_TMP_PATH,
-    EXAMPLES,
-    MATCHERS_PATH,
-    INPUT_SHAPE,
+    ID_EXAMPLES,
+    SELFIE_EXAMPLES,
     KEYS_PATH,
+    MATCHERS_PATH,
     REPO_DIR,
+    SERVER_TMP_PATH,
     SERVER_URL,
 )
-from client_server_interface import FHEClient
+
+MODEL_PATH = "client_server"
+# CLIENT_TMP_PATH = "client_tmp"
 
 # Uncomment here to have both the server and client in the same terminal
 subprocess.Popen(["uvicorn", "server:app"], cwd=REPO_DIR)
 time.sleep(3)
 
 
-def decrypt_output_with_wrong_key(encrypted_image, matcher_name):
+def decrypt_output_with_wrong_key(encrypted_image):
     """Decrypt the encrypted output using a different private key."""
     # Retrieve the matcher's deployment path
     matcher_path = MATCHERS_PATH / f"{matcher_name}/deployment"
@@ -80,35 +92,31 @@ def shorten_bytes_object(bytes_object, limit=500):
     return bytes_object[shift : limit + shift].hex()
 
 
-def get_client(user_id, matcher_name):
+def get_client():
     """Get the client API.
 
     Args:
         user_id (int): The current user's ID.
-        matcher_name (str): The matcher chosen by the user
+        filter_name (str): The filter chosen by the user
 
     Returns:
         FHEClient: The client API.
     """
-    return FHEClient(
-        MATCHERS_PATH / f"{matcher_name}/deployment",
-        matcher_name,
-        key_dir=KEYS_PATH / f"{matcher_name}_{user_id}",
-    )
+    return FHEModelClient(MODEL_PATH)
 
 
-def get_client_file_path(name, user_id, matcher_name):
+def get_client_file_path(name, user_id):
     """Get the correct temporary file path for the client.
 
     Args:
         name (str): The desired file name.
         user_id (int): The current user's ID.
-        matcher_name (str): The matcher chosen by the user
+        filter_name (str): The filter chosen by the user
 
     Returns:
         pathlib.Path: The file path.
     """
-    return CLIENT_TMP_PATH / f"{name}_{matcher_name}_{user_id}"
+    return CLIENT_TMP_PATH / f"{name}_embedding_{user_id}"
 
 
 def clean_temporary_files(n_keys=20):
@@ -133,8 +141,8 @@ def clean_temporary_files(n_keys=20):
             shutil.rmtree(key_dir)
 
     # Get all the encrypted objects in the temporary folder
-    client_files = CLIENT_TMP_PATH.iterdir()
-    server_files = SERVER_TMP_PATH.iterdir()
+    client_files = Path(CLIENT_TMP_PATH).iterdir()
+    server_files = Path(SERVER_TMP_PATH).iterdir()
 
     # Delete all files related to the ids whose keys were deleted
     for file in chain(client_files, server_files):
@@ -157,10 +165,11 @@ def keygen(matcher_name):
     clean_temporary_files()
 
     # Create an ID for the current user
-    user_id = numpy.random.randint(0, 2**32)
+    user_id = np.random.randint(0, 2**32)
+    # user_id = 298147048
 
     # Retrieve the client API
-    client = get_client(user_id, matcher_name)
+    client = get_client()
 
     # Generate a private key
     client.generate_private_and_evaluation_keys(force=True)
@@ -172,7 +181,7 @@ def keygen(matcher_name):
 
     # Save evaluation_key as bytes in a file as it is too large to pass through regular Gradio
     # buttons (see https://github.com/gradio-app/gradio/issues/1877)
-    evaluation_key_path = get_client_file_path("evaluation_key", user_id, matcher_name)
+    evaluation_key_path = get_client_file_path("evaluation_key", user_id)
 
     with evaluation_key_path.open("wb") as evaluation_key_file:
         evaluation_key_file.write(evaluation_key)
@@ -180,18 +189,124 @@ def keygen(matcher_name):
     return (user_id, True)
 
 
-def encrypt(
-    user_id, input_image, matcher_name, encrypted_image_name: str = "encrypted_image"
-):
-    """Encrypt the given image for a specific user and matcher.
+def detect_and_crop_face(
+    image: str,
+    min_aspect_ratio: float = 0.5,
+    max_aspect_ratio: float = 1.5,
+    min_face_size: float = 0.01,
+    max_face_size: float = 0.6,
+) -> Tuple[np.ndarray, Tuple[int, int, int, int], np.ndarray]:
+    # Read the image
+    # image = cv2.imread(image_path)
+    image_path = "test"
+    if image is None:
+        print(f"Failed to load image: {image_path}")
+        return None
+
+    # Print the image depth to debug
+    print(f"Image Depth: {image.dtype}, Shape: {image.shape}")
+
+    # Check if the image is of type CV_64F (float64) and convert to uint8
+    if image.dtype == np.float64:
+        print(f"Converting image from float64 to uint8 for {image_path}")
+        image = cv2.convertScaleAbs(image)  # Scale and convert to 8-bit
+
+    elif image.dtype != np.uint8:
+        print(f"Converting image from {image.dtype} to uint8 for {image_path}")
+        image = cv2.convertScaleAbs(image)  # Convert to 8-bit unsigned
+
+    # Convert to grayscale
+    try:
+        gray_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+    except cv2.error as e:
+        print(f"Error converting image to grayscale: {e} for {image_path}")
+        return None
+
+    # Load the face classifier
+    face_classifier = cv2.CascadeClassifier(
+        cv2.data.haarcascades + "haarcascade_frontalface_default.xml"
+    )
+
+    # Detect faces
+    faces = face_classifier.detectMultiScale(
+        gray_image,
+        scaleFactor=1.1,
+        minNeighbors=5,
+        minSize=(int(image.shape[1] * 0.1), int(image.shape[0] * 0.1)),
+    )
+
+    valid_faces = []
+    for x, y, w, h in faces:
+        aspect_ratio = w / h
+        face_area = w * h
+        image_area = image.shape[0] * image.shape[1]
+        face_size_ratio = face_area / image_area
+
+        if (
+            min_aspect_ratio <= aspect_ratio <= max_aspect_ratio
+            and min_face_size <= face_size_ratio <= max_face_size
+        ):
+            valid_faces.append((x, y, w, h))
+
+    if not valid_faces:
+        print(f"No suitable faces detected in {image_path}")
+        return None
+
+    # Sort faces by area (descending) and select the largest
+    valid_faces.sort(key=lambda f: f[2] * f[3], reverse=True)
+    (x, y, w, h) = valid_faces[0]
+
+    # Crop the face
+    try:
+        face_crop = image[
+            int(y - h * 0.1) : int(y + h * 1.1), int(x - w * 0.1) : int(x + w * 1.1)
+        ]
+        if face_crop.size == 0:
+            print(f"Failed to crop face for {image_path}: resulting crop is empty")
+            return None
+    except Exception as e:
+        print(f"Error cropping face from {image_path}: {e}")
+        return None
+
+    # Convert to RGB for display
+    try:
+        face_crop_rgb = cv2.cvtColor(face_crop, cv2.COLOR_BGR2RGB)
+    except cv2.error as e:
+        print(f"Error converting cropped face to RGB: {e} for {image_path}")
+        return None
+
+    return face_crop_rgb, (x, y, w, h), image
+
+
+def preprocess_image(input_image):
+    # TODO change for facenet
+    model = InceptionResnetV1(pretrained="vggface2").eval()
+    input_image = np.array(input_image)
+    image_crop = detect_and_crop_face(image=input_image)
+    preprocess = transforms.Compose(
+        [
+            transforms.Resize((160, 160)),  # Resize to 160x160 as required by the model
+            transforms.ToTensor(),  # Convert to tensor
+            transforms.Normalize(
+                [0.5, 0.5, 0.5], [0.5, 0.5, 0.5]
+            ),  # Normalize to [-1, 1]
+        ]
+    )
+    if image_crop[0] is not None:
+        img_tensor = preprocess(Image.fromarray(image_crop[0]))
+        img_tensor = img_tensor.unsqueeze(0)
+        with torch.no_grad():
+            embedding = model(img_tensor)
+        return embedding.numpy().flatten()
+
+
+def encrypt(user_id, selfie_image, id_image):
+    """Encrypt the given image for a specific user and filter.
 
     Args:
         user_id (int): The current user's ID.
-        input_image (numpy.ndarray): The image to encrypt.
-        matcher_name (str): The current matcher to consider.
-        encrypted_image_name (str): how to name the encrypted image
-            to distinguish between the query and the reference images.
-            Defaults to "encrypted_image"
+        selfie_image (np.ndarray): The image to encrypt.
+        id_image (np.ndarray): The image to encrypt.
 
     Returns:
         (input_image, encrypted_image_short) (Tuple[bytes]): The encrypted image and one of its
@@ -201,85 +316,75 @@ def encrypt(
     if user_id == "":
         raise gr.Error("Please generate the private key first.")
 
-    if input_image is None:
-        raise gr.Error("Please choose an image first.")
+    # for input_image in [selfie_image, id_image]:
+    #     if input_image is None:
+    #         raise gr.Error("Please choose an image first.")
 
-    if input_image.shape[-1] not in {3, 4}:
-        raise ValueError(
-            f"Input image must have 3 channels (RGB) or 4 channels. Current shape: {input_image.shape}"
-        )
+    #     if input_image.shape[-1] != 3:
+    #         raise ValueError(
+    #     f"Input image must have 3 channels (RGB). Current shape: {input_image.shape}"
+    # )
 
-    if input_image.shape[-1] == 4:
-        # Discarding alpha channel from images stored as Numpy arrays
-        # (reference https://stackoverflow.com/questions/35902302/discarding-alpha-channel-from-images-stored-as-numpy-arrays)
-        input_image = input_image[:, :, :3]
+    # Resize the image if it hasn't the shape (100, 100, 3)
 
-    # Resize the image if it hasn't the shape (INPUT_SHAPE[0], INPUT_SHAPE[1], 3)
-    if input_image.shape != (INPUT_SHAPE[0], INPUT_SHAPE[1], 3):
-        input_image_pil = Image.fromarray(input_image)
-        input_image_pil = input_image_pil.resize((INPUT_SHAPE[0], INPUT_SHAPE[1]))
-        input_image = numpy.array(input_image_pil)
+    selfie_image_orig = selfie_image.copy()
+    id_image_orig = id_image.copy()
 
+    selfie_image = Image.fromarray(selfie_image).convert("RGB")
+    id_image = Image.fromarray(id_image).convert("RGB")
+    embeddings_selfie = preprocess_image(selfie_image)
+    embeddings_id = preprocess_image(id_image)
+    X = np.concatenate((embeddings_selfie, embeddings_id))[np.newaxis, ...]
     # Retrieve the client API
-    client = get_client(user_id, matcher_name)
+    client: FHEModelClient = get_client()
 
     # Pre-process, encrypt and serialize the image
-    encrypted_image = client.encrypt_serialize(input_image)
+    encrypted_image = client.quantize_encrypt_serialize(X)
 
     # Save encrypted_image to bytes in a file, since too large to pass through regular Gradio
     # buttons, https://github.com/gradio-app/gradio/issues/1877
-    encrypted_image_path = get_client_file_path(
-        encrypted_image_name, user_id, matcher_name
-    )
+    encrypted_embedding = get_client_file_path("encrypted_embedding", user_id)
 
-    with encrypted_image_path.open("wb") as encrypted_image_file:
+    with encrypted_embedding.open("wb") as encrypted_image_file:
         encrypted_image_file.write(encrypted_image)
 
     # Create a truncated version of the encrypted image for display
     encrypted_image_short = shorten_bytes_object(encrypted_image)
 
-    return (resize_img(input_image), encrypted_image_short)
+    return (
+        encrypted_image_short,
+        resize_img(selfie_image_orig),
+        resize_img(id_image_orig),
+    )
 
 
-def send_input(user_id, matcher_name):
-    """Send the encrypted input images as well as the evaluation key to the server.
+def send_input(user_id):
+    """Send the encrypted input image as well as the evaluation key to the server.
 
     Args:
         user_id (int): The current user's ID.
-        matcher_name (str): The current matcher to consider.
+        filter_name (str): The current filter to consider.
     """
     # Get the evaluation key path
-    evaluation_key_path = get_client_file_path("evaluation_key", user_id, matcher_name)
+    evaluation_key_path = get_client_file_path("evaluation_key", user_id)
 
     if user_id == "" or not evaluation_key_path.is_file():
         raise gr.Error("Please generate the private key first.")
 
-    encrypted_query_image_path = get_client_file_path(
-        ENCRYPTED_QUERY_NAME, user_id, matcher_name
-    )
-
-    encrypted_reference_image_path = get_client_file_path(
-        ENCRYPTED_REFERENCE_NAME, user_id, matcher_name
-    )
+    encrypted_input_path = get_client_file_path("encrypted_embedding", user_id)
 
-    for encrypted_input_path in {
-        encrypted_query_image_path,
-        encrypted_reference_image_path,
-    }:
-        if not encrypted_input_path.is_file():
-            raise gr.Error(
-                f"Please generate the private key and then encrypt an image first: {encrypted_input_path}"
-            )
+    if not encrypted_input_path.is_file():
+        raise gr.Error(
+            "Please generate the private key and then encrypt an image first."
+        )
 
     # Define the data and files to post
     data = {
         "user_id": user_id,
-        "matcher": matcher_name,
     }
 
     files = [
-        ("files", open(encrypted_query_image_path, "rb")),
-        ("files", open(encrypted_reference_image_path, "rb")),
+        ("files", open(encrypted_input_path, "rb")),
         ("files", open(evaluation_key_path, "rb")),
     ]
 
@@ -293,16 +398,15 @@ def send_input(user_id, matcher_name):
         return response.ok
 
 
-def run_fhe(user_id, matcher_name):
-    """Apply the matcher on the encrypted image previously sent using FHE.
+def run_fhe(user_id):
+    """Apply the filter on the encrypted image previously sent using FHE.
 
     Args:
         user_id (int): The current user's ID.
-        matcher_name (str): The current matcher to consider.
+        filter_name (str): The current filter to consider.
     """
     data = {
         "user_id": user_id,
-        "matcher": matcher_name,
     }
 
     # Trigger the FHE execution on the encrypted image previously sent
@@ -314,25 +418,22 @@ def run_fhe(user_id, matcher_name):
         if response.ok:
             return response.json()
         else:
-            print(f"ERROR run_fhe: {response}")
-            
-            raise gr.Error("Please wait for the input images to be sent to the server.")
+            raise gr.Error("Please wait for the input image to be sent to the server.")
 
 
-def get_output(user_id, matcher_name):
-    """Retrieve the encrypted output.
+def get_output(user_id):
+    """Retrieve the encrypted output image.
 
     Args:
         user_id (int): The current user's ID.
-        matcher_name (str): The current filter to consider.
+        filter_name (str): The current filter to consider.
 
     Returns:
-        encrypted_output_short (bytes): A representation of the encrypted result.
+        encrypted_output_image_short (bytes): A representation of the encrypted result.
 
     """
     data = {
         "user_id": user_id,
-        "matcher": matcher_name,
     }
 
     # Retrieve the encrypted output image
@@ -346,38 +447,40 @@ def get_output(user_id, matcher_name):
 
             # Save the encrypted output to bytes in a file as it is too large to pass through regular
             # Gradio buttons (see https://github.com/gradio-app/gradio/issues/1877)
-            encrypted_output_path = get_client_file_path(
-                ENCRYPTED_OUTPUT_NAME, user_id, matcher_name
-            )
+            encrypted_output_path = get_client_file_path("encrypted_output", user_id)
 
             with encrypted_output_path.open("wb") as encrypted_output_file:
                 encrypted_output_file.write(encrypted_output)
 
-            # Decrypt the output using a different (wrong) key for display
-            output_representation = decrypt_output_with_wrong_key(
-                encrypted_output, matcher_name
-            )
+            # # Decrypt the image using a different (wrong) key for display
+            # output_image_representation = decrypt_output_with_wrong_key(
+            #     encrypted_output
+            # )
 
-            return {
-                encrypted_output_representation: gr.update(
-                    value=output_representation
-                    # value=resize_img(output_image_representation)
-                )
-            }
+            # return {
+            #     encrypted_output_representation: gr.update(
+            #         value=resize_img(output_image_representation)
+            #     )
+            # }
+
+            # Create a truncated version of the encrypted image for display
+            encrypted_output_short = shorten_bytes_object(encrypted_output)
+
+            return encrypted_output_short
 
         else:
             raise gr.Error("Please wait for the FHE execution to be completed.")
 
 
-def decrypt_output(user_id, matcher_name):
+def decrypt_output(user_id):
     """Decrypt the result.
 
     Args:
         user_id (int): The current user's ID.
-        matcher_name (str): The current matcher to consider.
+        filter_name (str): The current filter to consider.
 
     Returns:
-        (output_image, False, False) ((Tuple[numpy.ndarray, bool, bool]): The decrypted output, as
+        (output_image, False, False) ((Tuple[np.ndarray, bool, bool]): The decrypted output, as
             well as two booleans used for resetting Gradio checkboxes
 
     """
@@ -385,9 +488,7 @@ def decrypt_output(user_id, matcher_name):
         raise gr.Error("Please generate the private key first.")
 
     # Get the encrypted output path
-    encrypted_output_path = get_client_file_path(
-        ENCRYPTED_OUTPUT_NAME, user_id, matcher_name
-    )
+    encrypted_output_path = get_client_file_path("encrypted_output", user_id)
 
     if not encrypted_output_path.is_file():
         raise gr.Error("Please run the FHE execution first.")
@@ -397,25 +498,28 @@ def decrypt_output(user_id, matcher_name):
         encrypted_output = encrypted_output_file.read()
 
     # Retrieve the client API
-    client = get_client(user_id, matcher_name)
+    client = get_client()
 
     # Deserialize, decrypt and post-process the encrypted output
-    decrypted_ouput = client.deserialize_decrypt_post_process(encrypted_output)
+    decrypted_ouput = client.deserialize_decrypt_dequantize(encrypted_output)
 
     print(f"Decrypted output: {decrypted_ouput.shape=}")
+    print(f"Decrypted output: {decrypted_ouput=}")
 
-    return {output_result: gr.update(value=decrypted_ouput)}
+    predicted_class_id = np.argmax(decrypted_ouput)
+    print(f"{predicted_class_id=}")
+    return "PASS" if predicted_class_id == 1 else "FAIL"
 
 
 def resize_img(img, width=256, height=256):
     """Resize the image."""
-    if img.dtype != numpy.uint8:
-        img = img.astype(numpy.uint8)
+    if img.dtype != np.uint8:
+        img = img.astype(np.uint8)
     img_pil = Image.fromarray(img)
     # Resize the image
     resized_img_pil = img_pil.resize((width, height))
-    # Convert back to a NumPy array
-    return numpy.array(resized_img_pil)
+    # Convert back to a np array
+    return np.array(resized_img_pil)
 
 
 demo = gr.Blocks()
@@ -428,9 +532,9 @@ with demo:
         <!--p align="center">
             <img width=200 src="https://user-images.githubusercontent.com/5758427/197816413-d9cddad3-ba38-4793-847d-120975e1da11.png">
         </p-->
-        <h1 align="center"> #ppaihackteam14 </h1>
-        <h1 align="center">Biometric image matching Using Fully Homomorphic Encryption</h1>
+        <h1 align="center">Verio “Privacy-Preserving Biometric Verification for Authentication”</h1>
         <p align="center">
+            #ppaihackteam14  
             <a href="https://github.com/zama-ai/concrete-ml"> <img style="vertical-align: middle; display:inline-block; margin-right: 3px;" width=15 src="https://user-images.githubusercontent.com/5758427/197972109-faaaff3e-10e2-4ab6-80f5-7531f7cfb08f.png">Concrete-ML</a>
             —
             <a href="https://docs.zama.ai/concrete-ml"> <img style="vertical-align: middle; display:inline-block; margin-right: 3px;" width=15 src="https://user-images.githubusercontent.com/5758427/197976802-fddd34c5-f59a-48d0-9bff-7ad1b00cb1fb.png">Documentation</a>
@@ -447,11 +551,11 @@ with demo:
 
     gr.Markdown("## Client side")
     gr.Markdown("### Step 1: Upload input images. ")
-    gr.Markdown(
-        f"The image will automatically be resized to shape ({INPUT_SHAPE[0]}x{INPUT_SHAPE[1]}). "
-        "The image here, however, is displayed in its original resolution. The true image used "
-        "in this demo can be seen in Step 8."
-    )
+    # gr.Markdown(
+    #     f"The image will automatically be resized to shape ({INPUT_SHAPE[0]}x{INPUT_SHAPE[1]}). "
+    #     "The image here, however, is displayed in its original resolution. The true image used "
+    #     "in this demo can be seen in Step 8."
+    # )
     gr.Markdown("The query image to certify.")
     with gr.Row():
         input_query_img = gr.Image(
@@ -463,8 +567,8 @@ with demo:
             interactive=True,
         )
 
-        examples = gr.Examples(
-            examples=EXAMPLES,
+        selfie_examples = gr.Examples(
+            examples=SELFIE_EXAMPLES,
             inputs=[input_query_img],
             examples_per_page=5,
             label="Examples to use.",
@@ -480,28 +584,28 @@ with demo:
             interactive=True,
         )
 
-        examples = gr.Examples(
-            examples=EXAMPLES,
+        id_examples = gr.Examples(
+            examples=ID_EXAMPLES,
             inputs=[input_reference_img],
             examples_per_page=5,
             label="Examples to use.",
         )
 
-    gr.Markdown("### Step 2: Choose your matcher.")
-    matcher_name = gr.Dropdown(
-        choices=AVAILABLE_MATCHERS,
-        value="random guessing",
-        label="Choose your matcher",
-        interactive=True,
-    )
-
-    gr.Markdown("#### Notes")
-    gr.Markdown(
-        """
-        - The private key is used to encrypt and decrypt the data and will never be shared.
-        - No public key is required for these matcher operators.
-        """
-    )
+    # gr.Markdown("### Step 2: Choose your matcher.")
+    # matcher_name = gr.Dropdown(
+    #     choices=AVAILABLE_MATCHERS,
+    #     value="random guessing",
+    #     label="Choose your matcher",
+    #     interactive=True,
+    # )
+
+    # gr.Markdown("#### Notes")
+    # gr.Markdown(
+    #     """
+    #     - The private key is used to encrypt and decrypt the data and will never be shared.
+    #     - No public key is required for these matcher operators.
+    #     """
+    # )
 
     gr.Markdown("### Step 3: Generate the private key.")
     keygen_button = gr.Button("Generate the private key.")
@@ -510,35 +614,27 @@ with demo:
         keygen_checkbox = gr.Checkbox(label="Private key generated:", interactive=False)
 
     user_id = gr.Textbox(label="", max_lines=2, interactive=False, visible=False)
-    encrypted_query_image = gr.Textbox(
-        value=ENCRYPTED_QUERY_NAME,
-        label="",
-        max_lines=2,
-        interactive=False,
-        visible=False,
-    )
-    encrypted_reference_image = gr.Textbox(
-        value=ENCRYPTED_REFERENCE_NAME,
-        label="",
-        max_lines=2,
-        interactive=False,
-        visible=False,
-    )
-
-    gr.Markdown("### Step 4: Encrypt the images using FHE.")
-    encrypt_query_button = gr.Button("Encrypt the query image using FHE.")
-
-    with gr.Row():
-        encrypted_input_query = gr.Textbox(
-            label="Encrypted input query representation:",
-            max_lines=2,
-            interactive=False,
-        )
+    # encrypted_query_image = gr.Textbox(
+    #     value=ENCRYPTED_QUERY_NAME,
+    #     label="",
+    #     max_lines=2,
+    #     interactive=False,
+    #     visible=False,
+    # )
+    # encrypted_reference_image = gr.Textbox(
+    #     value=ENCRYPTED_REFERENCE_NAME,
+    #     label="",
+    #     max_lines=2,
+    #     interactive=False,
+    #     visible=False,
+    # )
+
+    gr.Markdown("### Step 4: Encrypt the input images using FHE.")
+    encrypt_button = gr.Button("Encrypt the images using FHE.")
 
-    encrypt_reference_button = gr.Button("Encrypt the reference image using FHE.")
     with gr.Row():
-        encrypted_input_reference = gr.Textbox(
-            label="Encrypted input reference representation:",
+        encrypted_input = gr.Textbox(
+            label="Encrypted input images representation:",
             max_lines=2,
             interactive=False,
         )
@@ -596,14 +692,14 @@ with demo:
     with gr.Row():
         original_query_image = gr.Image(
             input_query_img.value,
-            label=f"Input query image ({INPUT_SHAPE[0]}x{INPUT_SHAPE[1]}):",
+            label=f"Input query image:",
             interactive=False,
             height=256,
             width=256,
         )
         original_reference_image = gr.Image(
             input_reference_img.value,
-            label=f"Input reference image ({INPUT_SHAPE[0]}x{INPUT_SHAPE[1]}):",
+            label=f"Input reference image:",
             interactive=False,
             height=256,
             width=256,
@@ -619,46 +715,36 @@ with demo:
     # Button to generate the private key
     keygen_button.click(
         keygen,
-        inputs=[matcher_name],
+        inputs=[],
         outputs=[user_id, keygen_checkbox],
     )
 
     # Button to encrypt input query on the client side
-    encrypt_query_button.click(
+    encrypt_button.click(
         encrypt,
-        inputs=[user_id, input_query_img, matcher_name, encrypted_query_image],
-        outputs=[original_query_image, encrypted_input_query],
-    )
-
-    # Button to encrypt input reference on the client side
-    encrypt_reference_button.click(
-        encrypt,
-        inputs=[user_id, input_reference_img, matcher_name, encrypted_reference_image],
-        outputs=[original_reference_image, encrypted_input_reference],
+        inputs=[user_id, input_query_img, input_reference_img],
+        outputs=[encrypted_input, original_query_image, original_reference_image],
     )
 
     # Button to send the encodings to the server using post method
-    send_input_button.click(
-        send_input, inputs=[user_id, matcher_name], outputs=[send_input_checkbox]
-    )
+    send_input_button.click(send_input, inputs=[user_id], outputs=[send_input_checkbox])
 
     # Button to send the encodings to the server using post method
-    execute_fhe_button.click(
-        run_fhe, inputs=[user_id, matcher_name], outputs=[fhe_execution_time]
-    )
+    execute_fhe_button.click(run_fhe, inputs=[user_id], outputs=[fhe_execution_time])
 
     # Button to send the encodings to the server using post method
     get_output_button.click(
         get_output,
-        inputs=[user_id, matcher_name],
+        inputs=[user_id],
         outputs=[encrypted_output_representation],
     )
 
     # Button to decrypt the output on the client side
     decrypt_button.click(
         decrypt_output,
-        inputs=[user_id, matcher_name],
-        outputs=[output_result, keygen_checkbox, send_input_checkbox],
+        inputs=[user_id],
+        # outputs=[output_result, original_query_image, original_reference_image],
+        outputs=[output_result],
     )
 
     gr.Markdown(

client_server/client.zip

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:71104d20cae502527c0e5f9f25813c5d339c2629aa640fd12d53ffcb8c78c4d5
+size 1115569

matchers/filter blur/deployment/client.zip → client_server/server.zip

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c6b45511096f6da9aa73601b1372ffc42ec9272b40348d92f90427fbddff0571
-size 360
+oid sha256:9351a05f346e608a2263f0c6e77895ba9c0659151ae761251e21e39d49bb6853
+size 17205

common.py

@@ -33,7 +33,10 @@ INPUT_SHAPE = (100, 100)
 INPUT_EXAMPLES_DIR = REPO_DIR / "input_examples"
 
 # List of all image examples suggested in the demo
-EXAMPLES = [str(image) for image in INPUT_EXAMPLES_DIR.glob("**/*")]
+ID_EXAMPLES = [str(image) for image in (INPUT_EXAMPLES_DIR / "ids").glob("**/*")]
+SELFIE_EXAMPLES = [
+    str(image) for image in (INPUT_EXAMPLES_DIR / "selfies").glob("**/*")
+]
 
 # Encrypted image and output names
 ENCRYPTED_QUERY_NAME = "encrypted_query_image"

generate_deployment_files.py

@@ -8,7 +8,7 @@ print("Generating deployment files for all available filters")
 # This repository's directory
 REPO_DIR = Path(__file__).parent
 # This repository's main necessary folders
-MATCHERS_PATH = REPO_DIR / "data/compiled_models"
+MATCHERS_PATH = REPO_DIR / "matchers"
 for matcher_name in AVAILABLE_MATCHERS:
     print("Matcher:", matcher_name, "\n")
 

matchers.py

@@ -17,19 +17,17 @@ class TorchRandomGuessing(nn.Module):
         super().__init__()
         self.classes_ = classes_
 
-    def forward(self, q, r):
+    def forward(self, x):
         """Random guessing forward pass.
 
         Args:
-            q (torch.Tensor): The input query.
-            r (torch.Tensor): The input reference.
+            x (torch.Tensor): concat of query and reference.
 
         Returns:
              (torch.Tensor): .
         """
-        q = q.sum()
-        r = r.sum()
-        return torch.tensor([random.choice([0, 1])]) + q - q + r - r
+        x = x.sum()
+        return torch.tensor([random.choice([0, 1])]) + x - x
 
 
 class Matcher:
@@ -46,17 +44,14 @@ class Matcher:
 
     def compile(self):
 
-        inputset = [(np.array([10]), np.array([5]))]
+        inputset = (np.array([10]), np.array([5]))
 
         print("torch module > numpy module ...")
         numpy_module = NumpyModule(
             # torch_model, dummy_input=torch.from_numpy(np.array([10], dtype=np.int64))
             self.torch_model,
             # dummy_input=(torch.tensor([10]), torch.tensor([5])),
-            dummy_input=(
-                torch.from_numpy(inputset[0][1]),
-                torch.from_numpy(inputset[0][1]),
-            ),
+            dummy_input=torch.from_numpy(inputset[0]),
         )
 
         print("get proxy function ...")
@@ -64,15 +59,14 @@ class Matcher:
         # This is done in order to be able to provide any modules with arbitrary numbers of
         # encrypted arguments to Concrete Numpy's compiler
         numpy_filter_proxy, parameters_mapping = generate_proxy_function(
-            numpy_module.numpy_forward, ["query", "reference"]
+            numpy_module.numpy_forward, ["inputs"]
         )
 
         print("Compile the filter and retrieve its FHE circuit ...")
         compiler = Compiler(
             numpy_filter_proxy,
             {
-                parameters_mapping["query"]: "encrypted",
-                parameters_mapping["reference"]: "encrypted",
+                parameters_mapping["inputs"]: "encrypted",
             },
         )
         self.fhe_circuit = compiler.compile(inputset)

matchers/filter blur/deployment/circuit.mlir

@@ -1,11 +0,0 @@
-module {
-  func.func @main(%arg0: tensor<100x100x3x!FHE.eint<12>>) -> tensor<98x98x3x!FHE.eint<12>> {
-    %0 = "FHELinalg.transpose"(%arg0) {axes = [2, 1, 0]} : (tensor<100x100x3x!FHE.eint<12>>) -> tensor<3x100x100x!FHE.eint<12>>
-    %expanded = tensor.expand_shape %0 [[0, 1], [2], [3]] : tensor<3x100x100x!FHE.eint<12>> into tensor<1x3x100x100x!FHE.eint<12>>
-    %cst = arith.constant dense<1> : tensor<3x1x3x3xi13>
-    %1 = "FHELinalg.conv2d"(%expanded, %cst) {dilations = dense<1> : tensor<2xi64>, group = 3 : i64, padding = dense<0> : tensor<4xi64>, strides = dense<1> : tensor<2xi64>} : (tensor<1x3x100x100x!FHE.eint<12>>, tensor<3x1x3x3xi13>) -> tensor<1x3x98x98x!FHE.eint<12>>
-    %2 = "FHELinalg.transpose"(%1) {axes = [0, 3, 2, 1]} : (tensor<1x3x98x98x!FHE.eint<12>>) -> tensor<1x98x98x3x!FHE.eint<12>>
-    %collapsed = tensor.collapse_shape %2 [[0, 1], [2], [3]] : tensor<1x98x98x3x!FHE.eint<12>> into tensor<98x98x3x!FHE.eint<12>>
-    return %collapsed : tensor<98x98x3x!FHE.eint<12>>
-  }
-}

matchers/filter blur/deployment/configuration.json

@@ -1,22 +0,0 @@
-{
-    "verbose": false,
-    "show_graph": null,
-    "show_mlir": null,
-    "show_optimizer": null,
-    "dump_artifacts_on_unexpected_failures": true,
-    "enable_unsafe_features": false,
-    "use_insecure_key_cache": false,
-    "insecure_key_cache_location": null,
-    "loop_parallelize": true,
-    "dataflow_parallelize": false,
-    "auto_parallelize": false,
-    "jit": false,
-    "p_error": null,
-    "global_p_error": null,
-    "auto_adjust_rounders": false,
-    "single_precision": true,
-    "parameter_selection_strategy": "mono",
-    "show_progress": false,
-    "progress_title": "",
-    "progress_tag": false
-}

requirements.txt

@@ -1,3 +1,6 @@
-concrete-ml==1.6.1
+concrete-ml==1.6.0
 gradio
-more_itertools
+more_itertools
+torchvision==0.14.1
+facenet-pytorch==2.5.3
+opencv-python==4.10.0.84

server.py

@@ -5,35 +5,22 @@ from typing import List
 from fastapi import FastAPI, File, Form, UploadFile
 from fastapi.responses import JSONResponse, Response
 
-from common import (
-    ENCRYPTED_OUTPUT_NAME,
-    ENCRYPTED_QUERY_NAME,
-    ENCRYPTED_REFERENCE_NAME,
-    MATCHERS_PATH,
-    SERVER_TMP_PATH,
-    AVAILABLE_MATCHERS,
-)
-from client_server_interface import FHEServer
-
-# Load the server objects related to all currently available matchers once and for all
-FHE_SERVERS = {
-    matcher: FHEServer(MATCHERS_PATH / f"{matcher}/deployment")
-    for matcher in AVAILABLE_MATCHERS
-}
-
-
-def get_server_file_path(name, user_id, matcher_name):
+from common import SERVER_TMP_PATH
+from concrete.ml.deployment import FHEModelClient, FHEModelServer
+
+
+def get_server_file_path(name, user_id):
     """Get the correct temporary file path for the server.
 
     Args:
         name (str): The desired file name.
         user_id (int): The current user's ID.
-        matcher_name (str): The matcher chosen by the user
+        filter_name (str): The filter chosen by the user
 
     Returns:
         pathlib.Path: The file path.
     """
-    return SERVER_TMP_PATH / f"{name}_{matcher_name}_{user_id}"
+    return SERVER_TMP_PATH / f"{name}_{user_id}"
 
 
 # Initialize an instance of FastAPI
@@ -43,83 +30,57 @@ app = FastAPI()
 # Define the default route
 @app.get("/")
 def root():
-    return {"message": "Welcome to Your biometric image matching!"}
+    return {"message": "Welcome to Your Image FHE Filter Server!"}
 
 
 @app.post("/send_input")
 def send_input(
     user_id: str = Form(),
-    matcher: str = Form(),
     files: List[UploadFile] = File(),
 ):
     """Send the inputs to the server."""
     # Retrieve the encrypted input image and the evaluation key paths
-    encrypted_query_image_path = get_server_file_path(
-        ENCRYPTED_QUERY_NAME, user_id, matcher
-    )
-    encrypted_reference_image_path = get_server_file_path(
-        ENCRYPTED_REFERENCE_NAME, user_id, matcher
-    )
-    evaluation_key_path = get_server_file_path("evaluation_key", user_id, matcher)
+    encrypted_embedding_path = get_server_file_path("encrypted_embedding", user_id)
+    evaluation_key_path = get_server_file_path("evaluation_key", user_id)
 
     # Write the files using the above paths
-    with encrypted_query_image_path.open(
-        "wb"
-    ) as encrypted_query_image_file, encrypted_reference_image_path.open(
+    with encrypted_embedding_path.open(
         "wb"
-    ) as encrypted_reference_image_file, evaluation_key_path.open(
-        "wb"
-    ) as evaluation_key:
-        encrypted_query_image_file.write(files[0].file.read())
-        encrypted_reference_image_file.write(files[1].file.read())
-        evaluation_key.write(files[2].file.read())
+    ) as encrypted_embedding, evaluation_key_path.open("wb") as evaluation_key:
+        encrypted_embedding.write(files[0].file.read())
+        evaluation_key.write(files[1].file.read())
 
 
 @app.post("/run_fhe")
 def run_fhe(
     user_id: str = Form(),
-    matcher: str = Form(),
 ):
-    """Execute the matcher on the encrypted input images using FHE."""
+    """Execute the filter on the encrypted input image using FHE."""
     # Retrieve the encrypted input image and the evaluation key paths
-    encrypted_query_image_path = get_server_file_path(
-        "encrypted_query_image", user_id, matcher
-    )
-    encrypted_reference_image_path = get_server_file_path(
-        "encrypted_reference_image", user_id, matcher
-    )
-    evaluation_key_path = get_server_file_path("evaluation_key", user_id, matcher)
+    encrypted_image_path = get_server_file_path("encrypted_embedding", user_id)
+    evaluation_key_path = get_server_file_path("evaluation_key", user_id)
 
     # Read the files using the above paths
-    with encrypted_query_image_path.open(
-        "rb"
-    ) as encrypted_query_image_file, encrypted_reference_image_path.open(
-        "rb"
-    ) as encrypted_reference_image_file, evaluation_key_path.open(
+    with encrypted_image_path.open(
         "rb"
-    ) as evaluation_key_file:
-        encrypted_query_image = encrypted_query_image_file.read()
-        encrypted_reference_image = encrypted_reference_image_file.read()
+    ) as encrypted_image_file, evaluation_key_path.open("rb") as evaluation_key_file:
+        encrypted_image = encrypted_image_file.read()
         evaluation_key = evaluation_key_file.read()
 
-    # Load the FHE server related to the chosen matcher
-    fhe_server = FHE_SERVERS[matcher]
+    # Load the FHE server related to the chosen filter
+    fhe_server = FHEModelServer("client_server")
 
     # Run the FHE execution
     start = time.time()
-    encrypted_output = fhe_server.run(
-        encrypted_query_image, encrypted_reference_image, evaluation_key
-    )
+    encrypted_output_image = fhe_server.run(encrypted_image, evaluation_key)
     fhe_execution_time = round(time.time() - start, 2)
 
-    # Retrieve the encrypted output path
-    encrypted_output_path = get_server_file_path(
-        ENCRYPTED_OUTPUT_NAME, user_id, matcher
-    )
+    # Retrieve the encrypted output image path
+    encrypted_output_path = get_server_file_path("encrypted_output", user_id)
 
     # Write the file using the above path
     with encrypted_output_path.open("wb") as encrypted_output:
-        encrypted_output.write(encrypted_output)
+        encrypted_output.write(encrypted_output_image)
 
     return JSONResponse(content=fhe_execution_time)
 
@@ -127,13 +88,10 @@ def run_fhe(
 @app.post("/get_output")
 def get_output(
     user_id: str = Form(),
-    matcher: str = Form(),
 ):
-    """Retrieve the encrypted output."""
-    # Retrieve the encrypted output path
-    encrypted_output_path = get_server_file_path(
-        ENCRYPTED_OUTPUT_NAME, user_id, matcher
-    )
+    """Retrieve the encrypted output image."""
+    # Retrieve the encrypted output image path
+    encrypted_output_path = get_server_file_path("encrypted_output", user_id)
 
     # Read the file using the above path
     with encrypted_output_path.open("rb") as encrypted_output_file: