Python多线程+OpenCV：无法从线程类获取numpy数组问题

问题 #

将目标检测类放在独立线程中运行后，主线程无法获取numpy数组形式的检测结果rectangles。线程内部能打印出预期的数组，但主线程调用detect.rectangles始终返回空数组，编写getRectangles方法也无效。相关代码如下：

主方法 #

detect = detector.Detection()

detect.start()

while(True): 
    # 获取游戏的最新截图
    screenshot = wincap.get_screenshot()

    detect.update(screenshot) 

    detection_image = Vision.draw_rectangles(screenshot, detect.rectangles)

说明：Vision.draw_rectangles是基于OpenCV的目标绘制方法，多线程改造前可正常工作，目前仍能输出正确结果，但主线程无法从线程中获取数据。

Detector类 #

class Detection(object):
    # 线程相关属性
    stopped = True
    lock = None
    rectangles = []
 
    # 属性
    screenshot = None
 
    def __init__(self):
        # 创建线程锁对象
        self.lock = Lock()
    
    def update(self, screenshot):
        self.lock.acquire()
        self.screenshot = screenshot
        self.lock.release()

    def start(self):
        self.stopped = False
        t = Thread(target=self.run)
        print("Started")
        t.start()
     
    def stop(self):
        print("Stopped")
        self.stopped = True
      
    def run(self):
        vision_mineral = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/Copper.JPG')
        # 初始化采矿图标检测器
        vision_mine_icon = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/mineIcon.JPG')
        # 初始化加载图标检测器
        vision_loading_icon = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/LoadingBar.JPG')
        
        while not self.stopped:
            if not self.screenshot is None:
                print("updating image")
                # 目标检测
                rectangles = vision_mineral.find(self.screenshot, 0.54)
                # 检测采矿图标
                mine_icon = vision_mine_icon.find(self.screenshot, 0.6)
                # 检测加载图标
                loading_icon = vision_loading_icon.find(self.screenshot, 0.6)

                rectangles = np.concatenate((rectangles, mine_icon, loading_icon))
                
                self.lock.acquire()
                self.rectangles = rectangles
                self.lock.release()

Vision类 #

class Vision:
    # 常量
    TRACKBAR_WINDOW = "Trackbars"

    # 属性
    img_tofind = None
    img_tofind_w = None
    img_tofind_h = None
    method = None

    def __init__(self, img_tofind_path, method=cv.TM_CCOEFF_NORMED):
        # 加载目标图像
        self.img_tofind = cv.imread(img_tofind_path)
        
        # 获取图像尺寸
        self.img_tofind_w = self.img_tofind.shape[1]
        self.img_tofind_h = self.img_tofind.shape[0]
        
        # 匹配方法可选6种：
        # TM_CCOEFF, TM_CCOEFF_NORMED, TM_CCORR, TM_CCORR_NORMED, TM_SQDIFF, TM_SQDIFF_NORMED
        method = cv.TM_CCOEFF_NORMED
        self.method = method
    
    def find(self, screen_img, threshold=0.5, max_results=10):
        # 运行OpenCV匹配算法
        result = cv.matchTemplate(screen_img, self.img_tofind, self.method)

        # 获取所有超过阈值的匹配位置
        locations = np.where(result >= threshold)
        locations = list(zip(*locations[::-1]))

        # 如果无匹配结果，返回特定形状的空数组，避免拼接报错
        if not locations:
            return np.array([], dtype=np.int32).reshape(0, 4)

        # 去除重叠矩形，使用groupRectangles()
        # 先创建[x, y, w, h]格式的矩形列表
        rectangles = []
        for loc in locations:
            rect = [int(loc[0]), int(loc[1]), self.img_tofind_w, self.img_tofind_h]
            # 每个矩形添加两次，保留非重叠的单个矩形
            rectangles.append(rect)
            rectangles.append(rect)
        # 应用矩形分组
        # groupThreshold通常设为1，设为0则不分组；设为2则需要至少3个重叠矩形才保留
        # eps是矩形合并的相对边长差异阈值，设为0.5
        rectangles, weights = cv.groupRectangles(rectangles, groupThreshold=1, eps=0.5)
        
        return rectangles

    # 将find()返回的[x, y, w, h]矩形列表转换为矩形中心的[x, y]点击点列表
    def get_click_points(self, rectangles):
        points = []

        # 遍历所有矩形
        for (x, y, w, h) in rectangles:
            # 计算中心位置
            center_x = x + int(w/2)
            center_y = y + int(h/2)
            # 保存点
            points.append((center_x, center_y))

        return points

    # 在画布图像上绘制所有[x, y, w, h]矩形，返回绘制后的图像
    def draw_rectangles(self, screen_img, rectangles):
        # 颜色为BGR格式
        line_color = (0, 255, 0)
        line_type = cv.LINE_4

        for (x, y, w, h) in rectangles:
            # 确定矩形的左上角和右下角
            top_left = (x, y)
            bottom_right = (x + w, y + h)
            # 绘制矩形
            cv.rectangle(screen_img, top_left, bottom_right, line_color, lineType=line_type)

        return screen_img

    # 在画布图像上绘制所有[x, y]点击点为十字标记，返回绘制后的图像
    def draw_crosshairs(self, screen_img, points):
        # 颜色为BGR格式
        marker_color = (255, 0, 255)
        marker_type = cv.MARKER_CROSS

        for (center_x, center_y) in points:
            # 绘制中心点
            cv.drawMarker(screen_img, (center_x, center_y), marker_color, marker_type)

        return screen_img

解决方案 #

核心问题是主线程读取rectangles时未加锁，导致线程安全问题：线程更新rectangles时持有锁，但主线程直接读取可能在更新中途获取到不完整数据，或者因Python内存可见性问题无法感知到更新后的值。

修复步骤 #

1. 给rectangles的读取添加锁保护，通过封装方法确保读写操作都在锁的控制下；
2. 优化线程同步逻辑，避免线程空转浪费CPU资源。

修改后的Detector类代码 #

from threading import Lock, Thread, Event
import numpy as np

class Detection(object):
    stopped = True
    lock = None
    rectangles = []
    screenshot = None
    # 新增事件用于同步截图更新
    screenshot_updated = None
 
    def __init__(self):
        self.lock = Lock()
        self.screenshot_updated = Event()
    
    def update(self, screenshot):
        self.lock.acquire()
        self.screenshot = screenshot
        self.lock.release()
        # 通知线程有新截图可用
        self.screenshot_updated.set()

    def start(self):
        self.stopped = False
        t = Thread(target=self.run)
        print("Started")
        t.start()
     
    def stop(self):
        print("Stopped")
        self.stopped = True
        # 唤醒线程，避免阻塞在wait()
        self.screenshot_updated.set()
      
    # 新增带锁的获取方法
    def get_rectangles(self):
        self.lock.acquire()
        # 返回数组副本，避免外部操作影响内部数据
        res = self.rectangles.copy() if isinstance(self.rectangles, np.ndarray) else np.array([], dtype=np.int32).reshape(0,4)
        self.lock.release()
        return res
      
    def run(self):
        vision_mineral = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/Copper.JPG')
        vision_mine_icon = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/mineIcon.JPG')
        vision_loading_icon = Vision('C:/Users/Jordan/Desktop/Wakfu_Bot/Mining_bot/Resources/LoadingBar.JPG')
        
        while not self.stopped:
            # 等待新截图，避免空转
            self.screenshot_updated.wait()
            self.screenshot_updated.clear()
            
            self.lock.acquire()
            # 复制截图，避免主线程更新时冲突
            current_screenshot = self.screenshot.copy()
            self.lock.release()
            
            if current_screenshot is not None:
                print("updating image")
                rectangles = vision_mineral.find(current_screenshot, 0.54)
                mine_icon = vision_mine_icon.find(current_screenshot, 0.6)
                loading_icon = vision_loading_icon.find(current_screenshot, 0.6)

                rectangles = np.concatenate((rectangles, mine_icon, loading_icon))
                
                self.lock.acquire()
                self.rectangles = rectangles
                self.lock.release()

修改后的主方法代码 #

detect = detector.Detection()

detect.start()

while(True): 
    screenshot = wincap.get_screenshot()
    detect.update(screenshot) 
    # 使用带锁的方法获取结果
    detection_rects = detect.get_rectangles()
    detection_image = Vision.draw_rectangles(screenshot, detection_rects)

关键说明 #

• 锁的双向保护：读写rectangles和screenshot都必须加锁，确保同一时间只有一个线程操作这些共享变量；
• 数组副本返回：获取rectangles时返回副本，避免外部代码修改内部数据，同时减少锁持有时间；
• 事件同步：用Event替代空循环等待，降低CPU占用，线程只有在有新截图时才会执行检测逻辑。

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