Python多线程图像处理：智能门锁单摄像头多特征检测实现问题

核心思路 #

不要为每个检测任务单独打开摄像头，而是只启动一个摄像头读取帧，通过线程安全的队列共享帧数据，让手部检测、面部特征检测（微笑、眉毛）等线程从队列中获取同一帧并行处理，最后统一展示或输出检测结果。

具体实现步骤 #

1. 代码类重构 #

将各个模块封装为独立类，实现职责单一：

• Camera类：负责摄像头初始化、帧读取，将帧写入共享队列。
• HandDetector类：封装手部关键点检测逻辑，从队列取帧处理。
• FaceFeatureDetector类：封装面部关键点、微笑、眉毛状态检测逻辑，从队列取帧处理。
• DoorLockController类：负责整合各检测结果，判断是否符合解锁条件。

2. 线程安全的帧共享 #

使用queue.Queue实现帧的线程安全传递：

• 摄像头线程持续读取帧，将帧对象放入队列（注意传递帧的副本，避免多线程数据竞争）。
• 各检测线程从队列中获取帧，执行各自的检测逻辑。
• 队列设置最大长度，避免内存溢出，当队列满时摄像头线程暂停写入。

3. 多线程协同逻辑 #

• 主线程初始化队列、摄像头类、各检测类。
• 启动摄像头线程和所有检测线程。
• 监听退出信号（比如按下q键），终止所有线程并释放摄像头资源。

关键代码示例 #

Camera类实现

import cv2
import queue
import threading

class Camera:
    def __init__(self, queue, cam_id=0):
        self.queue = queue
        self.cam_id = cam_id
        self.cap = None
        self.running = False

    def start(self):
        self.cap = cv2.VideoCapture(self.cam_id)
        if not self.cap.isOpened():
            raise Exception("无法打开摄像头")
        self.running = True
        threading.Thread(target=self._read_frames, daemon=True).start()

    def _read_frames(self):
        while self.running:
            ret, frame = self.cap.read()
            if not ret:
                break
            # 队列满时阻塞，避免内存占用过高
            if not self.queue.full():
                self.queue.put(frame.copy())
        self.cap.release()

    def stop(self):
        self.running = False

手部检测线程类

class HandDetector:
    def __init__(self, queue, result_callback):
        self.queue = queue
        self.result_callback = result_callback
        self.running = False
        # 导入手部检测模块逻辑
        from HandTrackingModule import HandDetector as HTMDetector
        self.detector = HTMDetector()

    def start(self):
        self.running = True
        threading.Thread(target=self._detect, daemon=True).start()

    def _detect(self):
        while self.running:
            try:
                frame = self.queue.get(timeout=1)
                # 执行手部检测逻辑
                hands = self.detector.findHands(frame, draw=False)
                if hands:
                    finger_count = self.detector.fingersUp(hands[0])
                    self.result_callback("hand", finger_count)
                self.queue.task_done()
            except queue.Empty:
                continue

    def stop(self):
        self.running = False

面部特征检测线程类

class FaceFeatureDetector:
    def __init__(self, queue, result_callback):
        self.queue = queue
        self.result_callback = result_callback
        self.running = False
        # 导入面部关键点检测模块逻辑
        from FaceLandMarks import FaceLandmarkDetector
        self.detector = FaceLandmarkDetector()

    def start(self):
        self.running = True
        threading.Thread(target=self._detect, daemon=True).start()

    def _detect(self):
        while self.running:
            try:
                frame = self.queue.get(timeout=1)
                # 执行面部关键点检测
                landmarks = self.detector.get_landmarks(frame)
                if landmarks:
                    is_smiling = self.detector.detect_smile(landmarks)
                    eyebrow_state = self.detector.detect_eyebrow(landmarks)
                    self.result_callback("face", {"smile": is_smiling, "eyebrow": eyebrow_state})
                self.queue.task_done()
            except queue.Empty:
                continue

    def stop(self):
        self.running = False

主线程控制逻辑

def main():
    # 初始化队列，设置最大长度避免内存溢出
    frame_queue = queue.Queue(maxsize=5)
    results = {}

    def result_callback(feature_type, data):
        nonlocal results
        results[feature_type] = data
        # 这里可添加解锁判断逻辑，比如当检测结果符合预设密码时触发解锁
        if "hand" in results and "face" in results:
            print(f"当前检测结果：手指{results['hand']}，微笑{results['face']['smile']}，眉毛{results['face']['eyebrow']}")

    # 初始化并启动各个模块
    camera = Camera(frame_queue)
    hand_detector = HandDetector(frame_queue, result_callback)
    face_detector = FaceFeatureDetector(frame_queue, result_callback)

    camera.start()
    hand_detector.start()
    face_detector.start()

    # 等待退出信号
    while True:
        key = cv2.waitKey(1) & 0xFF
        if key == ord('q'):
            break

    # 停止所有线程
    camera.stop()
    hand_detector.stop()
    face_detector.stop()
    cv2.destroyAllWindows()

if __name__ == "__main__":
    main()

注意事项 #

• 避免在多个线程中同时操作同一帧的像素数据，必须传递帧的副本（如frame.copy()），防止数据竞争导致的异常。
• 检测线程的timeout参数要合理设置，避免线程无限阻塞。
• 摄像头资源必须在主线程中统一释放，避免内存泄漏。

内容的提问来源于stack exchange，提问作者Can Secgin