Python, FastAPI : Multi-Threading/Processing - Start/Stop Process with API Endpoints

Question

I have been trying to solve this one for quite a while now and cannot figure it out. Would appreciate some help with it. So I have a FastAPI server in which I have deployed a Drowsiness Detection Model/Script (dlib, opencv2, scipy). Now what I am trying to achieve is - Start and stop the DDM via API Endpoints. So the problem is - the uvicorn server is single-threaded, so when I run the DDM it will run in the same thread and when I try to stop the DDM it stops the entire server process (which is not something I want). I have tried forking the process and running the DDM on that process but it gives an error and crashes. I think using multithreading might help, I am not sure. Also if it does help me solve my issue I don't know how exactly to approach it. Relevant Code :

# Drowsiness Detection Script
def eye_aspect_ratio(eye):
    A = distance.euclidean(eye[1], eye[5])
    B = distance.euclidean(eye[2], eye[4])
    C = distance.euclidean(eye[0], eye[3])
    ear = (A + B) / (2.0 * C)
    return ear
 
 
def detect_drowsiness(monitor: bool):
    pid_file = open("intelligence/drowsiness_detection/dataset/pid.txt", "w")
    pid_str = str(os.getpid())
    pid_file.write(pid_str)
    pid_file.close()
 
    thresh = 0.25
    frame_check = 18
    detect = dlib.get_frontal_face_detector()
    # Dat file is the crux of the code
    predict = dlib.shape_predictor(
        "intelligence/drowsiness_detection/dataset/shape_predictor_68_face_landmarks.dat")
 
    (lStart, lEnd) = face_utils.FACIAL_LANDMARKS_68_IDXS["left_eye"]
    (rStart, rEnd) = face_utils.FACIAL_LANDMARKS_68_IDXS["right_eye"]
    cap = cv2.VideoCapture(0)
    flag = 0
    while monitor:
        ret, frame = cap.read()
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        subjects = detect(gray, 0)
        for subject in subjects:
            shape = predict(gray, subject)
            shape = face_utils.shape_to_np(
                shape)  # converting to NumPy Array
            leftEye = shape[lStart:lEnd]
            rightEye = shape[rStart:rEnd]
            leftEAR = eye_aspect_ratio(leftEye)
            rightEAR = eye_aspect_ratio(rightEye)
            ear = (leftEAR + rightEAR) / 2.0
            if ear < thresh:
                flag += 1
                print("Detecting,{}".format(flag))
                if flag >= frame_check:
                    print("ALERT - Drowsy")
 
            else:
                flag = 0
    cap.release()
 
 
 
 
# Drowsiness detection for a user
@ router.get("/face/drowsy/start", response_description="Drowsiness monitoring for the user")
async def start_drowsiness_detection(background_tasks: BackgroundTasks):
    background_tasks.add_task(detect_drowsiness, True)
    return("Drowsiness monitoring ON")
 
 
@ router.get("/face/drowsy/stop", response_description="Drowsiness monitoring for the user")
async def stop_drowsiness_detection():
    pid_file_path = f"intelligence/drowsiness_detection/dataset/pid.txt"
    pid_file = open(pid_file_path, "r")
    if not os.path.exists(pid_file_path):
        return("Please start monitoring first")
    pid_str = pid_file.read()
    remove_file(pid_file_path)
    os.kill(int(pid_str), signal.SIGKILL)
 
    return("Drowsiness monitoring OFF")

Possible workaround :

# Drowsiness Detection Script
def eye_aspect_ratio(eye):
    A = distance.euclidean(eye[1], eye[5])
    B = distance.euclidean(eye[2], eye[4])
    C = distance.euclidean(eye[0], eye[3])
    ear = (A + B) / (2.0 * C)
    return ear


class DrowsinessDetector(Process):

    running = Event()

    def stop_monitoring(self):
        if self.running.is_set():
            self.running.clear()

    def start_monitoring(self):
        if self.running.is_set():
            return

        self.running.set()
        self.detect_drowsiness()

    def detect_drowsiness(self):
        thresh = 0.25
        frame_check = 18
        detect = dlib.get_frontal_face_detector()
        # Dat file is the crux of the code
        predict = dlib.shape_predictor("./shape_predictor_68_face_landmarks.dat")

        (lStart, lEnd) = face_utils.FACIAL_LANDMARKS_68_IDXS["left_eye"]
        (rStart, rEnd) = face_utils.FACIAL_LANDMARKS_68_IDXS["right_eye"]
        cap = cv2.VideoCapture(0)
        flag = 0
        while self.running.is_set():
            ret, frame = cap.read()
            gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
            subjects = detect(gray, 0)
            for subject in subjects:
                shape = predict(gray, subject)
                shape = face_utils.shape_to_np(shape)  # converting to NumPy Array
                leftEye = shape[lStart:lEnd]
                rightEye = shape[rStart:rEnd]
                leftEAR = eye_aspect_ratio(leftEye)
                rightEAR = eye_aspect_ratio(rightEye)
                ear = (leftEAR + rightEAR) / 2.0
                if ear < thresh:
                    flag += 1
                    print("Detecting - {}".format(flag))
                    if flag >= frame_check:
                        print("ALERT - Drowsy")

                else:
                    flag = 0
        cap.release()


# Drowsiness detection for a user
drowsy = DrowsinessDetector()

@router.get("/face/drowsy/start", response_description="Drowsiness monitoring for the user")
async def start_drowsiness_detection(background_tasks: BackgroundTasks):
    background_tasks.add_task(drowsy.start_monitoring())
    return "Drowsiness monitoring ON"


@router.get("/face/drowsy/stop", response_description="Drowsiness monitoring for the user")
async def stop_drowsiness_detection(background_tasks: BackgroundTasks):
    background_tasks.add_task(drowsy.stop_monitoring())
    return "Drowsiness monitoring OFF"

I got this solution from Reddit but for some reason, it doesn't work. Any help will be much appreciated.

Answer 1

You can also just put your non async code into a standard sync route def. (This is actually the encouraged approach from FastAPI) FastAPI will then run that code in an external threadpool and manage it all for you. From there you could simply check the status of literally anything(files, redis, inMem dict, pub/sub) from within your while loop to stop the drowsiness detector.

https://fastapi.tiangolo.com/async/#path-operation-functions

Answer 2

While not explicitly mentioned in the FastAPI documentation, BackgroundTasks.background_tasks will create a new thread on the same process.

Using the first code you posted - when you store the PID (process ID) into a file in the detect_drowsiness() function, and then kill the process on stop_drowsiness_detection() route/function, you are effectively killing the very process that is running FastAPI.

In the background tasks section of FastAPI, in caveat, they mention:

If you need to perform heavy background computation and you don't necessarily need it to be run by the same process (for example, you don't need to share memory, variables, etc), you might benefit from using other bigger tools like Celery.

Relating to the second code you posted, the use of multiprocessing there seems to be towards the right direction. Without more details on why that specific implementation is not working, it's hard to help you further.

Answer 3

Had a similar issue (involving image processing through cv2 + fastapi) , solved this by opening a new process for the task without the use of BackgroundTasks. Here's a snippet:

import concurrent.futures

@app.post("/do_image_thing")
async def do_image_thing(params: Params):
    detect_bool = params.detect_bool # params is just an input of type pydantic BaseModel, irrelevant for this problem

    with concurrent.futures.ProcessPoolExecutor() as executor:
        executor.submit(detect_drowsiness, detect_bool)
    return {'stuff': 'stuff'}