File size: 3,250 Bytes
21c7197 7b32412 fa707a9 21c7197 fa707a9 21c7197 fa707a9 21c7197 a1cf584 21c7197 fa707a9 21c7197 3cf0931 fa707a9 a1cf584 3cf0931 fa707a9 0908a41 fa707a9 21c7197 fa707a9 3cf0931 fa707a9 21c7197 3cf0931 fa707a9 21c7197 fa707a9 21c7197 fa707a9 21c7197 fa707a9 21c7197 fa707a9 21c7197 3cf0931 21c7197 fa707a9 21c7197 fa707a9 21c7197 3cf0931 21c7197 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
"""Server that will listen for GET and POST requests from the client."""
import time
from typing import List
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.responses import JSONResponse, Response
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.
filter_name (str): The filter chosen by the user
Returns:
pathlib.Path: The file path.
"""
return SERVER_TMP_PATH / f"{name}_{user_id}"
# Initialize an instance of FastAPI
app = FastAPI()
# Define the default route
@app.get("/")
def root():
return {"message": "Welcome to Your Image FHE Filter Server!"}
@app.post("/send_input")
def send_input(
user_id: str = Form(),
files: List[UploadFile] = File(),
):
"""Send the inputs to the server."""
# Retrieve the encrypted input image and the evaluation key paths
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_embedding_path.open(
"wb"
) 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(),
):
"""Execute the filter on the encrypted input image using FHE."""
# Retrieve the encrypted input image and the evaluation key paths
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_image_path.open(
"rb"
) 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 filter
fhe_server = FHEModelServer("client_server")
# Run the FHE execution
start = time.time()
encrypted_output_image = fhe_server.run(encrypted_image, evaluation_key)
fhe_execution_time = round(time.time() - start, 2)
# 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_image)
return JSONResponse(content=fhe_execution_time)
@app.post("/get_output")
def get_output(
user_id: str = Form(),
):
"""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:
encrypted_output = encrypted_output_file.read()
return Response(encrypted_output)
|