树莓派4B+Python+OpenCV视频帧处理异常及效率优化求助

• 预期流程：打开罗技USB摄像头，捕获一帧画面后处理识别目标，向Arduino发送指令控制电机翻转目标，等待0.5秒后重复流程直至满足条件。
• 实际问题：初始无目标时识别正常，但放入目标后有5-8帧识别延迟，移除目标后仍持续识别5-8帧，导致平台重复转动5-8次。
• 尝试方案：每次循环开关摄像头解决了延迟，但出现启动超时；添加sleep(1.2)后超时解决，但整体耗时增至3-5秒，无法满足整体流程压缩至2秒内的需求。

原始代码 #

def predict(self):
        print('start predicting : ' + str(self.started))
        cap = cv2.VideoCapture(0)
        cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
        cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 640)
        cap.set(cv2.CAP_PROP_FPS, 30)
        while self.started is True and self.token is not None:
            print("predicting")
            ret, frame = cap.read()
            if not ret:
                return
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            cv2_im = frame.copy()
            cv2_im = cv2.resize(cv2_im, (self.height, self.width))
            img_array = np.asarray(cv2_im * 1. / 255, dtype='float32')
            predictions = self.lite_model(img_array[None, ...])[0] # Prediction process, it takes 1.3-1.7 sec
            y_lite = np.argmax(predictions)
            self.save_operation(y_lite) # this is a thread
            class_name = self.getClass(y_lite)
            print("class_name = "+class_name)
            if class_name == "object1":
                self.serial.write(b'r')
                print("turn right")
                sleep(.8)
            elif class_name == "object2":
                self.serial.write(b'l')
                print("turn left")
                sleep(.8)
            self.save_img(frame, class_name) # this is a thread

        cap.release()

修改后的尝试代码 #

def predict(self):
        print('start predicting : ' + str(self.started))
        
        while self.started is True and self.token is not None:
            print("predicting")
            cap = cv2.VideoCapture(0) # Turn on the camera
            cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
            cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 640)
            cap.set(cv2.CAP_PROP_FPS, 30)
            ret, frame = cap.read() # read a frame
            if not ret:
                return
            cap.release() # Release Camera resources
            frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            cv2_im = frame.copy()
            cv2_im = cv2.resize(cv2_im, (self.height, self.width))
            img_array = np.asarray(cv2_im * 1. / 255, dtype='float32')
            predictions = self.lite_model(img_array[None, ...])[0] # Prediction process, it takes 1.3-1.7 sec
            y_lite = np.argmax(predictions)
            self.save_operation(y_lite) # this is a thread
            class_name = self.getClass(y_lite)
            print("class_name = "+class_name)
            if class_name == "object1":
                self.serial.write(b'r')
                print("turn right")
            elif class_name == "object2":
                self.serial.write(b'l')
                print("turn left")
            self.save_img(frame, class_name) # this is a thread
            sleep(1.2) # For the camera to shutdown properly

1. 核心解决缓存延迟问题 #

USB摄像头默认会缓存多帧，导致你读取的是旧帧而非实时画面。保持摄像头持续打开，通过清空缓存+限制缓存大小解决延迟：

def predict(self):
    print('start predicting : ' + str(self.started))
    cap = cv2.VideoCapture(0)
    cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 640)
    cap.set(cv2.CAP_PROP_FPS, 30)
    # 强制摄像头仅缓存1帧，减少旧帧残留
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
    
    while self.started is True and self.token is not None:
        print("predicting")
        # 清空缓存：连续读取2帧，确保最后一帧是最新的
        for _ in range(2):
            ret, _ = cap.read()
            if not ret:
                cap.release()
                return
        # 读取实时帧
        ret, frame = cap.read()
        if not ret:
            cap.release()
            return
        
        # 后续处理流程不变
        frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        cv2_im = frame.copy()
        cv2_im = cv2.resize(cv2_im, (self.height, self.width))
        img_array = np.asarray(cv2_im * 1. / 255, dtype='float32')
        predictions = self.lite_model(img_array[None, ...])[0] # 耗时1.3-1.7秒
        y_lite = np.argmax(predictions)
        self.save_operation(y_lite) # 线程操作不阻塞主流程
        class_name = self.getClass(y_lite)
        print("class_name = "+class_name)
        
        # 调整电机等待时间为需求的0.5秒
        if class_name == "object1":
            self.serial.write(b'r')
            print("turn right")
            sleep(0.5)
        elif class_name == "object2":
            self.serial.write(b'l')
            print("turn left")
            sleep(0.5)
        
        self.save_img(frame, class_name) # 线程操作不阻塞主流程

    cap.release()

2. 优化流程耗时 #

• 保留摄像头持续打开：避免每次开关的启动/关闭耗时（这是修改后方案超时的核心原因）。
• 确认异步线程：确保save_operation和save_img确实是异步执行，不会阻塞主流程的帧捕获与识别。
• 可选降分辨率：如果摄像头支持，将分辨率降到480x480，进一步压缩图像处理和推理耗时。

3. 推理优化建议 #

如果整体流程仍接近2秒阈值，可对lite_model做优化：

• 使用模型量化（int8量化）减少推理耗时。
• 启用硬件加速（如OpenCV的CUDA后端、TensorRT），将推理耗时压缩到1.2秒以内，确保整体流程控制在2秒内。

内容的提问来源于stack exchange，提问作者Younes ezouagh