如何用OpenCV与multiprocessing在Python脚本间传递手势检测变量？

我需要实现两个Python脚本的并行运行与数据交互：

• 手势识别脚本：通过摄像头捕捉画面，当连续3次识别到同一手势时生成变量mvt_ok
• Psychopy实验脚本：接收mvt_ok信号后，切换展示新的实验刺激

流程要求：先启动摄像头，再启动Psychopy脚本展示初始刺激；当摄像头检测到目标手势后，立即将信号传递给Psychopy脚本完成刺激切换。

目前仅能通过如下简单函数传递mvt_ok，但不知道如何适配到Psychopy脚本中：

def f(child_conn,mvt_ok):
    print(mvt_ok)

手势识别核心代码：

if __name__ == '__main__':
    parent_conn,child_conn = Pipe()
    sentence = []

    while cap.isOpened(): 
        ret, frame = cap.read()
        image_np = np.array(frame)
        
        input_tensor = tf.convert_to_tensor(np.expand_dims(image_np, 0), dtype=tf.float32)
        detections = detect_fn(input_tensor)
        
        num_detections = int(detections.pop('num_detections'))
        detections = {key: value[0, :num_detections].numpy()
                      for key, value in detections.items()}
        detections['num_detections'] = num_detections
    
        # detection_classes should be ints.
        detections['detection_classes'] = detections['detection_classes'].astype(np.int64)
    
        label_id_offset = 1
        image_np_with_detections = image_np.copy()
    
        viz_utils.visualize_boxes_and_labels_on_image_array(
                    image_np_with_detections,
                    detections['detection_boxes'],
                    detections['detection_classes']+label_id_offset,
                    detections['detection_scores'],
                    category_index,
                    use_normalized_coordinates=True,
                    max_boxes_to_draw=5,
                    min_score_thresh=.8,
                    agnostic_mode=False)
        
        cv2.imshow('object detection',  cv2.resize(image_np_with_detections, (800, 600)))
        
        if np.max(detections['detection_scores'])>0.95:
            word = category_index[detections['detection_classes'][np.argmax(detections['detection_scores'])]+1]['name']
            sentence.append(word)
            
            if len(sentence)>=3:
                if sentence[-1]==sentence[-2] and sentence[-1]==sentence[-3]:
                    print('ok')
                    mvt_ok=1
                    p = Process(target=f, args=(child_conn,mvt_ok))
                    p.start()
                    p.join()
            
        
        if cv2.waitKey(10) & 0xFF == ord('q'):
            cap.release()
            cv2.destroyAllWindows()
            break

以下两种方法均可实现跨脚本的信号传递，适配你的需求：

方法一：多进程Pipe通信（父子进程管理） #

将Psychopy脚本作为子进程启动，通过multiprocessing.Pipe实现双向通信，手势识别脚本作为父进程发送信号。

1. 修改手势识别脚本 #

重构代码，直接启动Psychopy子进程并传递Pipe连接：

from multiprocessing import Process, Pipe
import cv2
import numpy as np
import tensorflow as tf
from object_detection.utils import viz_utils, label_map_util

# 加载你的模型和category_index（原有代码保留）
detect_fn = tf.saved_model.load('你的模型路径')
category_index = label_map_util.create_category_index_from_labelmap('标签路径', use_display_name=True)

def run_psychopy(parent_conn):
    # 导入并运行Psychopy实验脚本的核心逻辑
    from psychopy import visual, core
    win = visual.Window(size=(800,600))
    current_stim = visual.TextStim(win, text='初始刺激')
    current_stim.draw()
    win.flip()
    
    while True:
        # 监听Pipe信号
        if parent_conn.poll():
            signal = parent_conn.recv()
            if signal == 1:
                current_stim.text = '新刺激'
                current_stim.draw()
                win.flip()
                # 重置sentence列表，避免重复触发
                global sentence
                sentence = []
        # 处理Psychopy事件
        if win.checkEvents():
            break
    win.close()
    core.quit()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    # 启动Psychopy子进程
    psychopy_process = Process(target=run_psychopy, args=(child_conn,))
    psychopy_process.start()
    
    cap = cv2.VideoCapture(0)
    sentence = []

    while cap.isOpened(): 
        ret, frame = cap.read()
        image_np = np.array(frame)
        
        input_tensor = tf.convert_to_tensor(np.expand_dims(image_np, 0), dtype=tf.float32)
        detections = detect_fn(input_tensor)
        
        num_detections = int(detections.pop('num_detections'))
        detections = {key: value[0, :num_detections].numpy()
                      for key, value in detections.items()}
        detections['num_detections'] = num_detections
    
        detections['detection_classes'] = detections['detection_classes'].astype(np.int64)
    
        label_id_offset = 1
        image_np_with_detections = image_np.copy()
    
        viz_utils.visualize_boxes_and_labels_on_image_array(
                    image_np_with_detections,
                    detections['detection_boxes'],
                    detections['detection_classes']+label_id_offset,
                    detections['detection_scores'],
                    category_index,
                    use_normalized_coordinates=True,
                    max_boxes_to_draw=5,
                    min_score_thresh=.8,
                    agnostic_mode=False)
        
        cv2.imshow('object detection',  cv2.resize(image_np_with_detections, (800, 600)))
        
        if np.max(detections['detection_scores'])>0.95:
            word = category_index[detections['detection_classes'][np.argmax(detections['detection_scores'])]+1]['name']
            sentence.append(word)
            
            if len(sentence)>=3:
                if sentence[-1]==sentence[-2] and sentence[-1]==sentence[-3]:
                    print('ok')
                    # 发送信号给Psychopy进程
                    parent_conn.send(1)
                    # 重置sentence，防止重复触发
                    sentence = []
            
        
        if cv2.waitKey(10) & 0xFF == ord('q'):
            cap.release()
            cv2.destroyAllWindows()
            # 终止Psychopy进程
            psychopy_process.terminate()
            psychopy_process.join()
            break

2. 核心改动说明 #

• 新增run_psychopy函数，包含Psychopy实验的核心逻辑，监听Pipe接收信号
• 手势识别脚本启动时直接创建Psychopy子进程，无需单独启动
• 检测到目标手势时，通过parent_conn.send(1)传递信号，同时重置sentence避免重复触发

方法二：Socket网络通信（独立脚本启动） #

如果需要两个脚本完全独立启动（先开手势识别，再开Psychopy），可以用本地Socket实现通信。

1. Psychopy脚本（服务端） #

作为Socket服务端，监听本地端口等待信号：

from psychopy import visual, core
import socket

# 启动Socket服务端
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('localhost', 12345))
server_socket.listen(1)
conn, addr = server_socket.accept()

# 初始化Psychopy窗口和刺激
win = visual.Window(size=(800,600))
current_stim = visual.TextStim(win, text='初始刺激')
current_stim.draw()
win.flip()

while True:
    # 接收手势识别脚本的信号
    data = conn.recv(1024).decode()
    if data == '1':
        current_stim.text = '新刺激'
        current_stim.draw()
        win.flip()
    # 处理退出事件
    if win.checkEvents():
        break

# 清理资源
conn.close()
server_socket.close()
win.close()
core.quit()

2. 修改手势识别脚本（客户端） #

检测到目标手势时，向Socket服务端发送信号：

import cv2
import numpy as np
import tensorflow as tf
from object_detection.utils import viz_utils, label_map_util
import socket

# 加载模型和标签（原有代码保留）
detect_fn = tf.saved_model.load('你的模型路径')
category_index = label_map_util.create_category_index_from_labelmap('标签路径', use_display_name=True)

# 连接到Socket服务端
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(('localhost', 12345))

if __name__ == '__main__':
    cap = cv2.VideoCapture(0)
    sentence = []

    while cap.isOpened(): 
        ret, frame = cap.read()
        image_np = np.array(frame)
        
        input_tensor = tf.convert_to_tensor(np.expand_dims(image_np, 0), dtype=tf.float32)
        detections = detect_fn(input_tensor)
        
        num_detections = int(detections.pop('num_detections'))
        detections = {key: value[0, :num_detections].numpy()
                      for key, value in detections.items()}
        detections['num_detections'] = num_detections
    
        detections['detection_classes'] = detections['detection_classes'].astype(np.int64)
    
        label_id_offset = 1
        image_np_with_detections = image_np.copy()
    
        viz_utils.visualize_boxes_and_labels_on_image_array(
                    image_np_with_detections,
                    detections['detection_boxes'],
                    detections['detection_classes']+label_id_offset,
                    detections['detection_scores'],
                    category_index,
                    use_normalized_coordinates=True,
                    max_boxes_to_draw=5,
                    min_score_thresh=.8,
                    agnostic_mode=False)
        
        cv2.imshow('object detection',  cv2.resize(image_np_with_detections, (800, 600)))
        
        if np.max(detections['detection_scores'])>0.95:
            word = category_index[detections['detection_classes'][np.argmax(detections['detection_scores'])]+1]['name']
            sentence.append(word)
            
            if len(sentence)>=3:
                if sentence[-1]==sentence[-2] and sentence[-1]==sentence[-3]:
                    print('ok')
                    # 发送信号到Socket服务端
                    client_socket.send('1'.encode())
                    # 重置sentence
                    sentence = []
            
        
        if cv2.waitKey(10) & 0xFF == ord('q'):
            cap.release()
            cv2.destroyAllWindows()
            client_socket.close()
            break

3. 使用步骤 #

1. 先运行Psychopy脚本（启动服务端）
2. 再运行手势识别脚本（启动客户端）
3. 检测到目标手势后，Psychopy自动切换刺激

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