Python, FastAPI : 多线程/处理 - 使用 API 端点启动/停止进程

Question

我已经尝试解决这个问题已经有一段时间了，但无法弄清楚。将不胜感激一些帮助。所以我有一个 FastAPI 服务器，我在其中部署了一个嗜睡检测模型/脚本（dlib、opencv2、scipy）。现在我想要实现的是 - 通过 API 端点启动和停止 DDM。所以问题是 - uvicorn 服务器是单线程的，所以当我运行 DDM 时，它会在同一个线程中运行，当我尝试停止 DDM 时，它会停止整个服务器进程（这不是我想要的）。我曾尝试分叉该进程并在该进程上运行 DDM，但它会出错并崩溃。我认为使用多线程可能会有所帮助，我不确定。此外，如果它确实帮助我解决了我的问题，我不知道该如何解决。相关代码：

# 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")

可能的解决方法：

# 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"

我从 Reddit 获得了这个解决方案，但由于某种原因，它不起作用。任何帮助将不胜感激。

Answer 1

您也可以将非异步代码放入标准同步路由定义中。 （这实际上是 FastAPI 鼓励的方法）FastAPI 将在外部线程池中运行该代码并为您管理所有内容。从那里你可以简单地从你的while循环中检查任何东西（文件、redis、inMem dict、pub/sub）的状态，以停止睡意检测器。

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

Answer 2

虽然 FastAPI 文档中没有明确提到，BackgroundTasks.background_tasks 将在同一进程上创建一个新线程。

使用您发布的第一个代码 - 当您将 PID（进程 ID）存储到 detect_drowsiness() 函数中的文件中，然后杀死 stop_drowsiness_detection() 路由/函数上的进程时，您实际上是在杀死那个进程正在运行 FastAPI。

在 FastAPI 的后台任务部分，caveat，他们提到：

如果您需要执行繁重的后台计算并且您不一定需要它由同一进程运行（例如，您不需要共享内存、变量等），您可能会受益于使用其他像 Celery 这样更大的工具。

关于您发布的第二个代码，多处理的使用似乎朝着正确的方向发展。如果没有详细说明为什么特定的实现不起作用，很难进一步帮助您。