Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
File size: 3,684 Bytes
c119738 4337a72 c119738 |
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 102 103 104 105 |
"""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 settings import DEPLOYMENT_PATH, SERVER_FILES, CLIENT_TYPES
from development.client_server_interface import MultiInputsFHEModelServer
# Load the server objects related to all currently available filters once and for all
FHE_SERVER = MultiInputsFHEModelServer(DEPLOYMENT_PATH)
def get_server_file_path(name, client_id, client_type):
"""Get the correct temporary file path for the server.
Args:
name (str): The desired file name (either 'evaluation_key' or 'encrypted_inputs').
client_id (int): The client ID to consider.
client_type (str): The type of user to consider (either 'user', 'bank' or 'third_party').
Returns:
pathlib.Path: The file path.
"""
return SERVER_FILES / f"{name}_{client_type}_{client_id}"
# Initialize an instance of FastAPI
app = FastAPI()
# Define the default route
@app.get("/")
def root():
return {"message": "Welcome to Credit Card Approval Prediction server!"}
@app.post("/send_input")
def send_input(
client_id: str = Form(),
client_type: str = Form(),
files: List[UploadFile] = File(),
):
"""Send the inputs to the server."""
# Retrieve the encrypted inputs and the evaluation key paths
encrypted_inputs_path = get_server_file_path("encrypted_inputs", client_id, client_type)
evaluation_key_path = get_server_file_path("evaluation_key", client_id, client_type)
# Write the files using the above paths
with encrypted_inputs_path.open("wb") as encrypted_inputs, evaluation_key_path.open(
"wb"
) as evaluation_key:
encrypted_inputs.write(files[0].file.read())
evaluation_key.write(files[1].file.read())
@app.post("/run_fhe")
def run_fhe(
client_id: str = Form(),
):
"""Execute the model on the encrypted inputs using FHE."""
# Retrieve the evaluation key
evaluation_key_path = get_server_file_path("evaluation_key", client_id, "user")
# Get the evaluation key
with evaluation_key_path.open("rb") as evaluation_key_file:
evaluation_key = evaluation_key_file.read()
# Get the encrypted inputs
encrypted_inputs = []
for client_type in CLIENT_TYPES:
encrypted_inputs_path = get_server_file_path("encrypted_inputs", client_id, client_type)
with encrypted_inputs_path.open("rb") as encrypted_inputs_file:
encrypted_input = encrypted_inputs_file.read()
encrypted_inputs.append(encrypted_input)
# Run the FHE execution
start = time.time()
encrypted_output = FHE_SERVER.run(*encrypted_inputs, serialized_evaluation_keys=evaluation_key)
fhe_execution_time = round(time.time() - start, 2)
# Retrieve the encrypted output path
encrypted_output_path = get_server_file_path("encrypted_output", client_id, client_type)
# Write the file using the above path
with encrypted_output_path.open("wb") as output_file:
output_file.write(encrypted_output)
return JSONResponse(content=fhe_execution_time)
@app.post("/get_output")
def get_output(
client_id: str = Form(),
client_type: str = Form(),
):
"""Retrieve the encrypted output."""
# Retrieve the encrypted output path
encrypted_output_path = get_server_file_path("encrypted_output", client_id, client_type)
# 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)
|