树莓派Python多线程CV项目段错误问题求助

核心问题定位 #

你的代码出现段错误（Segmentation Error），结合树莓派环境和代码细节，主要原因包括：

• 语法缩进错误：capture_frames函数中while循环缩进错误，直接导致代码运行时崩溃
• pyttsx3线程不安全：语音引擎pyttsx3并非线程安全组件，在子线程中频繁调用runAndWait()和stop()会引发资源冲突
• 锁范围不合理：将整个帧处理逻辑包裹在锁内，导致采集线程长时间无法获取锁，同时增加了线程阻塞风险
• OpenCV GUI操作跨线程：子线程中调用cv2.waitKey()会引发GUI线程冲突，导致崩溃
• 共享数据未正确隔离：帧处理过程中，共享的data.image可能被采集线程覆盖，导致处理异常

修复步骤 #

1. 修正代码缩进错误
2. 将语音输出逻辑移至主线程，通过队列传递语音文本
3. 缩小锁的作用范围，仅保护共享数据的读写操作
4. 移除子线程中的cv2.waitKey()，统一在主线程处理退出信号
5. 处理帧时先复制共享图像数据，避免被采集线程覆盖
6. 若使用多进程，需在子进程内重新加载模型（PyTorch模型无法跨进程共享）

修改后的完整代码 #

from ultralytics import YOLO
import cv2
import torch
import numpy as np
import pyttsx3
from collections import Counter
from threading import Thread, Event, Lock
from picamera2 import Picamera2
from queue import Queue

labels={0: 'person', 1: 'bicycle', 2: 'car', 3: 'motorcycle', 4: 'airplane', 5: 'bus', 6: 'train', 7: 'truck', 8: 'boat', 9: 'traffic light', 10: 'fire hydrant', 11: 'stop sign', 12: 'parking meter', 13: 'bench', 14: 'bird', 15: 'cat', 16: 'dog', 17: 'horse', 18: 'sheep', 19: 'cow', 20: 'elephant', 21: 'bear', 22: 'zebra', 23: 'giraffe', 24: 'backpack', 25: 'umbrella', 26: 'handbag', 27: 'tie', 28: 'suitcase', 29: 'frisbee', 30: 'skis', 31: 'snowboard', 32: 'sports ball', 33: 'kite', 34: 'baseball bat', 35: 'baseball glove', 36: 'skateboard', 37: 'surfboard', 38: 'tennis racket', 39: 'bottle', 40: 'wine glass', 41: 'cup', 42: 'fork', 43: 'knife', 44: 'spoon', 45: 'bowl', 46: 'banana', 47: 'apple', 48: 'sandwich', 49: 'orange', 50: 'broccoli', 51: 'carrot', 52: 'hot dog', 53: 'pizza', 54: 'donut', 55: 'cake', 56: 'chair', 57: 'couch', 58: 'potted plant', 59: 'bed', 60: 'dining table', 61: 'toilet', 62: 'tv', 63: 'laptop', 64: 'mouse', 65: 'remote', 66: 'keyboard', 67: 'cell phone', 68: 'microwave', 69: 'oven', 70: 'toaster', 71: 'sink', 72: 'refrigerator', 73: 'book', 74: 'clock', 75: 'vase', 76: 'scissors', 77: 'teddy bear', 78: 'hair drier', 79: 'toothbrush'}

# 初始化模型
model = YOLO("yolov8n.pt")
midas = torch.hub.load("intel-isl/MiDaS", "MiDaS_small")
device = torch.device("cpu")
midas.to(device)
midas.eval()
midas_transforms = torch.hub.load("intel-isl/MiDaS", "transforms")
transform = midas_transforms.small_transform

# 语音队列（线程安全）
voice_queue = Queue()

class SharedData:
    def __init__(self):
        self.image = None
        self.box = None
        self.label = None
        self.prob = None
        self.stop_event = Event()
  
lock = Lock()
        
def capture_frames(data):
    cam = Picamera2()
    cam.start()
    while not data.stop_event.is_set():
        frame = cam.capture_array("main")[:, :, :3]
        if frame is not None:
            with lock:
                # 复制帧数据，避免处理时被覆盖
                data.image = frame.copy()
    cam.stop()
    cv2.destroyAllWindows()
    

def process_frames(data):
    while not data.stop_event.is_set():
        current_frame = None
        # 仅在读取共享图像时加锁
        with lock:
            if data.image is not None:
                current_frame = data.image.copy()
        
        if current_frame is not None:
            img = cv2.cvtColor(current_frame, cv2.COLOR_BGR2RGB)
            input_batch = transform(img).to(device)

            with torch.no_grad():
                prediction = midas(input_batch)
                prediction = torch.nn.functional.interpolate(
                    prediction.unsqueeze(1),
                    size=img.shape[:2],
                    mode="bicubic",
                    align_corners=False,
                ).squeeze()

            output = prediction.cpu().numpy()
            o = ((output - np.min(output))/(np.max(output)-np.min(output)))
            overlay = cv2.merge([o,o,o])
            feed = (current_frame * overlay).astype(np.uint8)

            results = model(feed)
            for result in results:
                with lock:
                    data.box = result.boxes.xyxy
                    data.prob = result.probs
                    data.label = result.boxes.cls.numpy()
            
            # 生成语音文本，放入队列
            if results:
                str_text = ""
                for k,v in Counter(results[0].boxes.cls.numpy()).items():
                    str_text += f"{v} {labels[k]} "
                voice_queue.put(str_text)
                
    cv2.destroyAllWindows()

def voice_speaker():
    # 主线程处理语音输出，避免线程安全问题
    engine = pyttsx3.init()
    voices = engine.getProperty('voices') 
    rate = engine.getProperty('rate')
    engine.setProperty('voice', voices[1].id)
    engine.setProperty('rate', 160)
    
    while True:
        text = voice_queue.get()
        if text == "STOP":
            break
        engine.say(text)
        engine.runAndWait()
    engine.stop()
    
if __name__ == "__main__":
    data = SharedData()

    # 启动语音线程（主线程关联）
    voice_thread = Thread(target=voice_speaker, daemon=True)
    voice_thread.start()
    
    capture_thread = Thread(target=capture_frames, args=(data,))
    process_thread = Thread(target=process_frames, args=(data,))

    capture_thread.start()
    process_thread.start()

    try:
        # 主线程处理退出信号
        while True:
            if cv2.waitKey(1) & 0xFF == ord("q") or cv2.waitKey(1) == 27:
                break
    finally:
        data.stop_event.set()
        # 发送停止信号给语音线程
        voice_queue.put("STOP")
        capture_thread.join()
        process_thread.join()
        voice_thread.join()
        cv2.destroyAllWindows()

额外说明 #

• 若使用多进程替代多线程，需在每个子进程内重新加载YOLO和MiDaS模型，因为PyTorch模型无法跨进程共享
• 单帧预测耗时1.7秒，可考虑优化：使用YOLOv8n的量化版本、降低输入分辨率、启用树莓派的硬件加速（如OpenCV的GPU支持）
• 发送目标框和视频流到笔记本，可使用socket或zeromq实现，建议在处理完帧后，将目标框数据和帧（可选）打包发送

内容的提问来源于stack exchange，提问作者Ayushman Singh