bluepulsar
243 lines
9.0 KiB
Python
243 lines
9.0 KiB
Python
"""
|
||
示例 4: 多传感器并行处理（多进程模式 - 不使用 MultiSensor）
|
||
|
||
本示例演示如何使用 multiprocessing 同时运行两个传感器：
|
||
- 每个传感器运行在独立的进程中
|
||
- 每个进程拥有独立的 CUDA 上下文
|
||
- 使用 multiprocessing.Process 和 Queue 进行进程间通信
|
||
|
||
架构说明：
|
||
进程 1 (传感器 0): 采集 → 光流 → HHD → 接触检测 → 队列
|
||
进程 2 (传感器 1): 采集 → 光流 → HHD → 接触检测 → 队列
|
||
↓
|
||
主进程: 显示 ← ───────────────┘
|
||
|
||
适用场景：多指触觉传感器、多视角采集、需要 GPU 加速的多路处理
|
||
|
||
注意事项：
|
||
- Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support()
|
||
- 每个传感器进程会占用独立的 GPU 显存
|
||
"""
|
||
import os
|
||
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
|
||
import cv2
|
||
import orisys
|
||
import numpy as np
|
||
import multiprocessing as mp
|
||
from multiprocessing import Queue, Process
|
||
import time
|
||
|
||
|
||
def sensor_worker(sensor_id, vid_src, config_name, cal_path, result_queue, stop_event):
|
||
"""
|
||
传感器工作进程函数
|
||
|
||
Args:
|
||
sensor_id: 传感器编号
|
||
vid_src: 视频源（文件路径或摄像头索引）
|
||
config_name: 配置名称
|
||
result_queue: 结果队列，用于向主进程发送数据
|
||
stop_event: 停止事件，用于通知工作进程退出
|
||
"""
|
||
try:
|
||
# 在每个进程中创建独立的传感器实例
|
||
sensor = orisys.Sensor(
|
||
vid_src,
|
||
isstitch=True,
|
||
cuda=True,
|
||
verbose=False, # 减少子进程输出
|
||
config_name=config_name,
|
||
cal_path=cal_path
|
||
)
|
||
|
||
print(f"传感器 {sensor_id} 已启动")
|
||
|
||
while not stop_event.is_set():
|
||
try:
|
||
# 步骤1: 获取图像
|
||
sensor.get_img()
|
||
|
||
# 步骤2: 计算形变场
|
||
sensor.compute_deformation()
|
||
|
||
# 步骤3: 计算触碰点
|
||
sensor.compute_contact()
|
||
|
||
# 步骤4: 读取信息
|
||
fps, flow, vnormal, img, contour, centroid, depth_map = sensor.read_info(
|
||
sensor.info.FPS,
|
||
sensor.info.VRAW,
|
||
sensor.info.VNORMAL,
|
||
sensor.info.IMG,
|
||
sensor.info.CONTOUR,
|
||
sensor.info.CENTROID,
|
||
sensor.info.DEPTH
|
||
)
|
||
|
||
# 步骤5: 准备数据字典（确保所有数据都是可序列化的）
|
||
data = {
|
||
'sensor_id': sensor_id,
|
||
'fps': fps,
|
||
'flow': flow.copy() if flow is not None else None,
|
||
'vnormal': vnormal.copy() if vnormal is not None else None,
|
||
'img': img.copy() if img is not None else None,
|
||
'contour': contour,
|
||
'centroid': centroid,
|
||
'depth_map': depth_map.copy() if depth_map is not None else None,
|
||
'timestamp': time.time()
|
||
}
|
||
|
||
# 步骤6: 发送数据到主进程（非阻塞）
|
||
try:
|
||
result_queue.put_nowait(data)
|
||
except:
|
||
# 如果队列已满，跳过这一帧
|
||
pass
|
||
|
||
# 短暂休眠，避免过度占用CPU
|
||
time.sleep(0.001)
|
||
|
||
except Exception as e:
|
||
print(f"传感器 {sensor_id} 处理错误: {e}")
|
||
time.sleep(0.1)
|
||
|
||
# 清理资源
|
||
sensor.disconnect()
|
||
print(f"传感器 {sensor_id} 已停止")
|
||
|
||
except Exception as e:
|
||
print(f"传感器 {sensor_id} 初始化失败: {e}")
|
||
result_queue.put({'sensor_id': sensor_id, 'error': str(e)})
|
||
|
||
|
||
def main():
|
||
# =========================================================================
|
||
# 1. 配置多个传感器
|
||
# =========================================================================
|
||
num_sensors = 2
|
||
sensor_configs = [
|
||
{
|
||
'sensor_id': 0,
|
||
'vid_src': './s1.mp4',
|
||
'config_name': 'prototype',
|
||
'cal_path': './config/cal.npy'
|
||
},
|
||
{
|
||
'sensor_id': 1,
|
||
'vid_src': './s1.mp4', # 可以改为不同的视频源或摄像头索引
|
||
'config_name': 'prototype',
|
||
'cal_path': './config/cal2.npy'
|
||
},
|
||
]
|
||
|
||
# =========================================================================
|
||
# 2. 创建进程间通信对象
|
||
# =========================================================================
|
||
result_queues = [Queue(maxsize=num_sensors) for _ in sensor_configs] # 每个传感器一个队列
|
||
stop_events = [mp.Event() for _ in sensor_configs] # 停止事件
|
||
|
||
# =========================================================================
|
||
# 3. 启动传感器工作进程
|
||
# =========================================================================
|
||
processes = []
|
||
for i, config in enumerate(sensor_configs):
|
||
p = Process(
|
||
target=sensor_worker,
|
||
args=(i, config['vid_src'], config['config_name'], config['cal_path'], result_queues[i], stop_events[i])
|
||
)
|
||
p.start()
|
||
processes.append(p)
|
||
|
||
print(f"\n已启动 {len(processes)} 个传感器进程")
|
||
print("按 'q' 退出程序")
|
||
|
||
# 用于存储每个传感器的最新数据
|
||
latest_data = [None] * len(sensor_configs)
|
||
|
||
try:
|
||
# =====================================================================
|
||
# 4. 主循环：从队列获取数据并显示
|
||
# =====================================================================
|
||
while True:
|
||
# 从所有队列中获取最新数据
|
||
for i, queue in enumerate(result_queues):
|
||
try:
|
||
# 非阻塞获取，获取队列中最新的数据
|
||
while True:
|
||
data = queue.get_nowait()
|
||
latest_data[i] = data
|
||
except:
|
||
# 队列为空，使用之前的数据
|
||
pass
|
||
|
||
# 显示所有传感器的数据
|
||
for data in latest_data:
|
||
if data is None or 'error' in data:
|
||
continue
|
||
|
||
sid = data['sensor_id']
|
||
|
||
# 显示形变场
|
||
if data['img'] is not None and data['flow'] is not None:
|
||
arrows = orisys.util.draw_arrows(
|
||
data['img'],
|
||
data['flow'],
|
||
threshold=5,
|
||
grid_spacing=10,
|
||
arrow_scale=1.0
|
||
)
|
||
cv2.imshow(f"形变场 Deformation {sid}", arrows)
|
||
|
||
# 显示触碰点和区域
|
||
if data['img'] is not None:
|
||
image_with_foe = orisys.util.draw_contact(
|
||
data['img'],
|
||
data['contour'],
|
||
data['centroid']
|
||
)
|
||
cv2.imshow(f"Contact Centroid {sid}", image_with_foe)
|
||
|
||
# 显示深度图
|
||
if data['depth_map'] is not None and data['depth_map'].size > 0:
|
||
div_abs = data['depth_map']
|
||
if div_abs.max() > div_abs.min():
|
||
div_normalized = ((div_abs - div_abs.min()) /
|
||
(div_abs.max() - div_abs.min()) * 255).astype(np.uint8)
|
||
else:
|
||
div_normalized = np.zeros_like(div_abs, dtype=np.uint8)
|
||
cv2.imshow(f"Depth Map {sid}",
|
||
cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET))
|
||
|
||
# 打印FPS
|
||
print(f"传感器 {sid}: FPS={data['fps']:5.1f}")
|
||
|
||
# 检查退出键
|
||
key = cv2.waitKey(1) & 0xFF
|
||
if key == ord('q'):
|
||
print("\n退出程序...")
|
||
break
|
||
|
||
finally:
|
||
# =====================================================================
|
||
# 5. 清理资源
|
||
# =====================================================================
|
||
# 通知所有工作进程停止
|
||
for event in stop_events:
|
||
event.set()
|
||
|
||
# 等待所有进程结束
|
||
for p in processes:
|
||
p.join(timeout=2)
|
||
if p.is_alive():
|
||
p.terminate()
|
||
p.join()
|
||
|
||
cv2.destroyAllWindows()
|
||
print("所有传感器进程已停止")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# Windows 下多进程必需的初始化
|
||
mp.freeze_support()
|
||
main()
|