|
import numpy as np |
|
from deepface import DeepFace |
|
from PIL import Image |
|
from io import BytesIO |
|
from typing import Union, List |
|
import os |
|
from fastapi import HTTPException |
|
import uuid |
|
import cv2 |
|
from .preprocess import load_dataset, RegNet |
|
from sklearn.linear_model import LogisticRegression |
|
from concrete.ml.torch.compile import compile_torch_model |
|
import logging |
|
|
|
|
|
class InferenceService: |
|
def __init__(self): |
|
self.model = None |
|
|
|
def save_video(self, video_content: bytes, user_id: str) -> None: |
|
""" |
|
Extract frames from the video content asynchronously using FFmpeg. |
|
""" |
|
try: |
|
video_id = str(uuid.uuid4()) |
|
os.makedirs(f"temp/{user_id}", exist_ok=True) |
|
with open(file=f"temp/{user_id}/video.mp4", mode="wb") as f: |
|
f.write(video_content) |
|
except Exception as e: |
|
raise HTTPException(status_code=500, detail=f"Error saving video: {e}") |
|
|
|
def save_frames(self,user_id: str) -> str: |
|
""" |
|
Save 10 random frames from the video content using cv2 and return the path to the frames. |
|
""" |
|
video_path = f"temp/{user_id}/video.mp4" |
|
if not os.path.exists(video_path): |
|
raise HTTPException(status_code=404, detail="Video not found") |
|
|
|
vidcap = cv2.VideoCapture(video_path) |
|
success, image = vidcap.read() |
|
count = 0 |
|
while success and count < 10: |
|
cv2.imwrite( |
|
f"temp/{user_id}/frame%d.jpg" % count, image |
|
) |
|
success, image = vidcap.read() |
|
count += 1 |
|
return f"temp/{user_id}/" |
|
|
|
def train_model(self, frames_path: str, user_id: str) -> str: |
|
""" |
|
Train a model on the frames and return the path to the model. |
|
""" |
|
""" |
|
embeddings, labels = load_dataset(frames_path, cache=True) |
|
embeddings = (embeddings - np.mean(embeddings, axis=0)) / np.std( |
|
embeddings, axis=0 |
|
) |
|
model = LogisticRegression(C=1 / 5) |
|
model.fit(embeddings, y=labels) |
|
nb_sample = 100 |
|
W = model.coef_ |
|
b = model.intercept_.reshape(-1, 1) |
|
X_train_rand = np.random.normal(0, 1, [nb_sample, embeddings.shape[1]]) |
|
W_rand = np.random.normal(0, 1, [nb_sample, W.shape[1]]) |
|
X_rand_stack = np.hstack([X_train_rand, W_rand]) |
|
reg_net = RegNet(b) |
|
quantized_module = compile_torch_model( |
|
reg_net, # our model |
|
X_rand_stack, # a representative input-set to be used for both quantization and compilation |
|
n_bits=6, |
|
rounding_threshold_bits={"n_bits": 6, "method": "approximate"}, |
|
) |
|
save_model_path: str = f"temp/model/{user_id}.zip" |
|
quantized_module.fhe_circuit.server.save(save_model_path) |
|
return save_model_path |
|
""" |
|
return "temp/model/fake_id.zip" |
|
|
|
def push_model(self, model_path: str, user_id: str) -> None: |
|
""" |
|
Push the model to the servrer |
|
""" |
|
|
|
|
