|
import numpy as np |
|
import time |
|
import os, sys |
|
|
|
from pathlib import Path |
|
|
|
from sklearn.datasets import make_classification |
|
from sklearn.model_selection import train_test_split |
|
|
|
from concrete.ml.deployment import FHEModelClient |
|
|
|
import requests |
|
|
|
|
|
def to_json(python_object): |
|
if isinstance(python_object, bytes): |
|
return {"__class__": "bytes", "__value__": list(python_object)} |
|
raise TypeError(repr(python_object) + " is not JSON serializable") |
|
|
|
|
|
def from_json(python_object): |
|
if "__class__" in python_object: |
|
return bytes(python_object["__value__"]) |
|
|
|
|
|
API_URL = "https://puqif7goarh132kl.us-east-1.aws.endpoints.huggingface.cloud" |
|
headers = { |
|
"Authorization": "Bearer " + os.environ.get("HF_TOKEN"), |
|
"Content-Type": "application/json", |
|
} |
|
|
|
|
|
def query(payload): |
|
response = requests.post(API_URL, headers=headers, json=payload) |
|
print(f"{response=}") |
|
print(f"{response.json()=}") |
|
return response.json() |
|
|
|
|
|
path_to_model = Path("compiled_model") |
|
x, y = make_classification(n_samples=1000, class_sep=2, n_features=30, random_state=42) |
|
_, X_test, _, y_test = train_test_split(x, y, test_size=0.2, random_state=42) |
|
|
|
|
|
fhemodel_client = FHEModelClient(path_to_model) |
|
|
|
|
|
fhemodel_client.generate_private_and_evaluation_keys() |
|
evaluation_keys = fhemodel_client.get_serialized_evaluation_keys() |
|
|
|
|
|
payload = { |
|
"inputs": "fake", |
|
"evaluation_keys": to_json(evaluation_keys), |
|
"method": "save_key", |
|
} |
|
|
|
uid = query(payload) |
|
print(f"Storing the key in the database under {uid=}") |
|
|
|
|
|
nb_good = 0 |
|
nb_samples = len(X_test) |
|
verbose = False |
|
time_start = time.time() |
|
duration = 0 |
|
|
|
for i in range(nb_samples): |
|
|
|
|
|
encrypted_inputs = fhemodel_client.quantize_encrypt_serialize([X_test[i]]) |
|
|
|
|
|
payload = { |
|
"inputs": "fake", |
|
"encrypted_inputs": to_json(encrypted_inputs), |
|
"method": "inference", |
|
"uid": uid, |
|
} |
|
|
|
|
|
duration -= time.time() |
|
duration_inference = -time.time() |
|
encrypted_prediction = query(payload) |
|
duration += time.time() |
|
duration_inference += time.time() |
|
|
|
encrypted_prediction = from_json(encrypted_prediction) |
|
|
|
|
|
prediction_proba = fhemodel_client.deserialize_decrypt_dequantize(encrypted_prediction)[0] |
|
prediction = np.argmax(prediction_proba) |
|
|
|
if verbose or True: |
|
print( |
|
f"for {i}-th input, {prediction=} with expected {y_test[i]} in {duration_inference:.3f} seconds" |
|
) |
|
|
|
|
|
nb_good += y_test[i] == prediction |
|
|
|
print(f"Accuracy on {nb_samples} samples is {nb_good * 1. / nb_samples}") |
|
print(f"Total time: {time.time() - time_start} seconds") |
|
print(f"Duration in inferences: {duration} seconds") |
|
print(f"Duration per inference: {duration / nb_samples} seconds") |
|
|