Logic/FaceRecognition/Trainer.py

from fastapi import APIRouter,BackgroundTasks,HTTPException
from os.path import exists
from os import listdir
import subprocess
from passlib.context import CryptContext
import sqlite3
ModelTrainer=APIRouter(prefix="/Trainer")
TrainingProcess={}

def TrainModel(UserId):
    global TrainingProcess
    if TrainingProcess.get(UserId,False) and TrainingProcess[UserId].poll() is None:
        raise HTTPException(status_code=400,detail="Model is already training.")
    TrainingProcess=subprocess.Popen(["python","./FaceRecognition/ModelTrainer.py",f"{UserId}"])



@ModelTrainer.post("/TrainFaceModel")
async def SpeachToTextEndPoint(Tasks:BackgroundTasks,Email:str,Password:str):
    global TrainingProcess

    try:
        State=False
        pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")
        connect=sqlite3.connect("DataBase/DataBase.bd")
        cursor=connect.execute(f'''
                        SELECT UserId,Password FROM Users where Email='{Email}' 
                        ''')
        
        Data=cursor.fetchall()
        if len(Data) !=0 :
            if Data[0][0]==None:
                return {"Status":True,"Message":"Email or Password Is Incorrect"}
            HasedPassword=Data[0][1]
            UserId=Data[0][0]
            
            State=pwd_context.verify(Password,HasedPassword)
            
        if exists(f"./FaceRecognition/ExtactedFaces/{UserId}/Train"):
            for UserName in listdir(f"./FaceRecognition/ExtactedFaces/{UserId}/Train"):
                if len(listdir(f"./FaceRecognition/ExtactedFaces/{UserId}/Train/"+UserName))<2:
                    return  {"Status":False,"Message":f"{UserName} has only {len(listdir(f'./FaceRecognition/ExtactedFaces/{UserId}/Train/'+UserName))} image and it must be 2 or more"}
        if not State:
            return {"Status":False,"Message":"Email or Password is not correct"}
        if TrainingProcess.get(UserId,False) and TrainingProcess[UserId].poll() is None:
            raise HTTPException(status_code=400,detail="Model is already training.")
        Tasks.add_task(TrainModel ,args=[UserId])
        return{"message":"Training Started"}
    except Exception as e:
        return{"Stats":False,"message":f"{e}"}
    
@ModelTrainer.post("/TrainFaceModelStatus")
async def SpeachToTextEndPoint(Email:str,Password:str):
    global TrainingProcess
    try:
        State=False
        pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")
        connect=sqlite3.connect("DataBase/DataBase.bd")
        cursor=connect.execute(f'''
                        SELECT UserId,Password FROM Users where Email='{Email}' 
                        ''')
        if len(cursor.fetchall())==1:
            HasedPassword=cursor.fetchall()[0][1]
            UserId=cursor.fetchall()[0][0]
        State=pwd_context.verify(Password,HasedPassword)
        if not State:
            return {"Status":False,"Message":"Email or Password is not correct"}
        if TrainingProcess.get(UserId,False) and TrainingProcess[UserId].poll() is None:
            return{"message":"Model still Training "}
        else:
            return{"message":"Model Training ended"}
    except Exception as e:
        return{"Stats":True,"message":f"{e}"}