orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add multisensor

Browse Source
This commit is contained in:
unknown
2025-12-30 17:35:06 +08:00
parent b4c3518ad5
commit daeb182497
5 changed files with 211 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/compound_eye.png
BIN
View File
Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 2.1 MiB

After

Width:  |  Height:  |  Size: 2.0 MiB

example3_two_sensor_pipeline.py
+118
View File
@@ -0,0 +1,118 @@
"""
示例 4: 多传感器并行处理（多进程模式）
本示例演示如何同时运行多个传感器：
- 每个传感器运行在独立的进程中
- 每个进程拥有独立的 CUDA 上下文
- 使用 MultiSensor 类简化多传感器管理
架构说明：
进程 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" # Windows 兼容性设置
import cv2
import orisys
import multiprocessing as mp
import numpy as np
def main():
# =========================================================================
# 1. 配置多个传感器
# =========================================================================
# 每个传感器的配置是一个字典，包含：
# vid_src: 视频源（文件路径或摄像头索引）
# isstitch: 是否启用拼接模式
# cuda: 是否使用 GPU 加速
# cal_path: 标定文件路径
configs = [
{
'vid_src': './s1.mp4',
'isstitch': True,
'cuda': True,
'cal_path': './config/cal.npy',
'config_name': "prototype"
},
{
'vid_src': './s1.mp4',
'isstitch': True,
'cuda': True,
'cal_path': './config/cal.npy',
'config_name': "prototype"
},
]
# =========================================================================
# 2. 使用上下文管理器启动多传感器
# =========================================================================
# MultiSensor 会自动：
# - 为每个配置创建独立的子进程
# - 在子进程中初始化 Sensor 实例
# - 持续采集和处理数据
# - 通过队列将结果传回主进程
# compute_contact=True 启用接触点检测（深度图、质心、轮廓）
with orisys.MultiSensor(configs, compute_contact=True) as multi:
print("\n多传感器已启动，按 'q' 退出")
# =====================================================================
# 3. 迭代获取所有传感器的数据
# =====================================================================
# multi 是可迭代的，每次迭代返回所有传感器的最新数据列表
# all_data = [data_sensor_0, data_sensor_1, ...]
for all_data in multi:
# 过滤掉 None（某些传感器可能暂时没有新数据）
for data in [d for d in all_data if d]:
sid = data['sensor_id'] # 传感器编号
# 打印传感器状态
print(f"传感器 {sid}: FPS={data['fps']:5.1f}, 法向力={data['fnormal']:8.2f}")
# 显示形变箭头图
if data['img'] is not None and data['flow'] is not None:
arrows = orisys.util.draw_arrows(
data['img'].copy(),
data['flow'],
threshold=5
)
cv2.imshow(f"Deformation {sid}", arrows)
if data['centroid'] is not None:
# print(f"触碰点: {data['centroid']}")
image_with_foe = orisys.util.draw_contact(data['img'].copy(), 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:
# 归一化散度值到0-255范围（使用绝对值，然后归一化）
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))
# 检查退出键
if cv2.waitKey(1) & 0xFF == ord('q'):
break
# 上下文管理器退出时自动停止所有子进程
cv2.destroyAllWindows()
if __name__ == '__main__':
# Windows 下多进程必需的初始化
mp.freeze_support()
main()
example6_multi_async.py
+11 -12
View File
@@ -1,11 +1,11 @@
"""
示例 6: 多传感器 + 异步流水线（最高性能模式）
示例 6: 多传感器并行处理
本示例结合了多进程和异步流水线的优势：
本示例演示多进程并行处理多个传感器：
- 多进程：每个传感器运行在独立进程，拥有独立 CUDA 上下文
- 异步流水线：每个进程内部使用三阶段流水线
- 每个进程独立处理：采集 → 光流 → HHD
这是性能最高的配置，适用于：
适用于：
- 多个高分辨率传感器
- 需要最大化总吞吐量
- 有充足的 CPU 核心和 GPU 显存
@@ -20,7 +20,7 @@
[采集] → [光流] → [HHD] → 队列 ──┘
资源占用估算（每个传感器）：
- CPU: 约 2-3 个核心（采集、光流、HHD 各一个线程）
- CPU: 约 1-2 个核心
- GPU 显存: 约 500MB-1GB（取决于分辨率和 CUDA 算法）
"""
@@ -52,12 +52,11 @@ def main():
]
# =========================================================================
# 2. 使用 async_context 启动（推荐方式）
# 2. 使用上下文管理器启动（推荐方式）
# =========================================================================
# async_context 是一个便捷的上下文管理器：
# - 自动启用每个传感器进程内的异步流水线
# - 退出时自动清理所有资源
with orisys.MultiSensor.async_context(configs) as multi:
# 上下文管理器会自动启动所有传感器进程
# 退出时自动清理所有资源
with orisys.MultiSensor(configs) as multi:
print("\n多传感器异步模式已启动，按 'q' 退出")
# =====================================================================
@@ -104,8 +103,8 @@ def main_manual():
# 创建 MultiSensor 实例
multi = orisys.MultiSensor(configs)
# 手动启动，use_async=True 启用异步模式
multi.start(use_async=True)
# 手动启动
multi.start()
try:
while True:
example7_five_sensors.py
+3 -3
View File
@@ -2,7 +2,7 @@
示例 7: 五传感器网格显示
本示例演示如何扩展到 5 个传感器：
- 使用异步多进程模式运行所有传感器
- 使用多进程模式并行运行所有传感器
- 将 5 个传感器画面拼接成 2x3 网格显示
- 实时显示每个传感器的 FPS 和力数据
@@ -48,9 +48,9 @@ def main():
]
# =========================================================================
# 2. 启动多传感器异步模式
# 2. 启动多传感器并行模式
# =========================================================================
with orisys.MultiSensor.async_context(configs) as multi:
with orisys.MultiSensor(configs) as multi:
print(f"\n{len(configs)} 个传感器已启动，按 'q' 退出")
# =====================================================================
src/orisys/MultiSensor.py
+79 -81
View File
@@ -9,19 +9,16 @@ independent CUDA contexts.
import os
def _sensor_worker(sensor_id, config, data_queue, stop_event, use_async=False):
def _sensor_worker(sensor_id, config, data_queue, stop_event, compute_contact=False):
"""
Worker function that runs in a separate PROCESS for each sensor.
Each process has its own Python interpreter and CUDA context.
Args:
use_async: If True, use 3-stage async pipeline within the process
"""
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
from .sensor import Sensor
print(f"[Sensor {sensor_id}] Process started (PID: {os.getpid()}), async={use_async}")
print(f"[Sensor {sensor_id}] Process started (PID: {os.getpid()})")
sensor = Sensor(
vid_src=config.get('vid_src'),
@@ -32,10 +29,6 @@ def _sensor_worker(sensor_id, config, data_queue, stop_event, use_async=False):
cal_path=config.get('cal_path', None)
)
# Enable async pipeline if requested
if use_async:
sensor.enable_async(queue_size=2)
frame_count = 0
def queue_put(data):
@@ -48,43 +41,76 @@ def _sensor_worker(sensor_id, config, data_queue, stop_event, use_async=False):
while not stop_event.is_set():
try:
if use_async:
# Use async get
result = sensor.get_async(timeout=0.05)
if result is None:
continue
data = {
'sensor_id': sensor_id,
'img': result['img'].copy() if result['img'] is not None else None,
'flow': result['flow'].copy() if result['flow'] is not None else None,
'normal': result['normal'].copy() if result['normal'] is not None else None,
'shear': result['shear'].copy() if result['shear'] is not None else None,
'depth': result['depth'].copy() if result['depth'] is not None else None,
'fps': result['fps'],
'fnormal': result['fnormal'],
'fshearx': result['fshearx'],
'fsheary': result['fsheary'],
'ftor': result['ftor'],
'frame_count': frame_count,
}
# Sync mode
sensor.get_img()
sensor.compute_deformation()
if compute_contact:
sensor.compute_contact()
# Use read_info to get all sensor data
img, flow, normal, shear, fps, fnormal, fshearx, fsheary = sensor.read_info(
sensor.info.IMG,
sensor.info.VRAW,
sensor.info.VNORMAL,
sensor.info.VSHEAR,
sensor.info.FPS,
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY
)
if compute_contact:
depth, centroid, contour = sensor.read_info(
sensor.info.DEPTH,
sensor.info.CENTROID,
sensor.info.CONTOUR
)
else:
# Sync mode
sensor.get_img()
sensor.compute_deformation()
data = {
'sensor_id': sensor_id,
'img': sensor.img.copy() if sensor.img is not None else None,
'flow': sensor.flow.copy() if sensor.flow is not None else None,
'normal': sensor.vnormal.copy() if sensor.vnormal is not None else None,
'shear': sensor.shear.copy() if sensor.shear is not None else None,
'depth': sensor.depth.copy() if sensor.depth is not None else None,
'fps': sensor.fps,
'fnormal': sensor.fnormal,
'fshearx': sensor.fshearx,
'fsheary': sensor.fsheary,
'ftor': sensor.ftor,
'frame_count': frame_count,
}
depth = None
centroid = None
contour = None
# Build data dictionary
data = {
'sensor_id': sensor_id,
'img': img if img is not None and img.size > 0 else None,
'flow': flow if flow is not None and flow.size > 0 else None,
'normal': normal if normal is not None and normal.size > 0 else None,
'shear': shear if shear is not None and shear.size > 0 else None,
'fps': fps,
'fnormal': fnormal,
'fshearx': fshearx,
'fsheary': fsheary,
'frame_count': frame_count,
}
# Handle contact-related fields
if compute_contact:
# depth and depth_map
data['depth'] = depth if depth is not None and depth.size > 0 else None
data['depth_map'] = data['depth'] # Alias for compatibility
# centroid - can be None or numpy array
if centroid is not None:
# Check if it's a valid array with elements
if hasattr(centroid, 'size') and centroid.size > 0:
data['centroid'] = centroid
data['centroid_found'] = True
else:
data['centroid'] = None
data['centroid_found'] = False
else:
data['centroid'] = None
data['centroid_found'] = False
# contour
data['contour'] = contour if contour is not None and contour.size > 0 else None
else:
data['depth'] = None
data['depth_map'] = None
data['centroid'] = None
data['centroid_found'] = False
data['contour'] = None
queue_put(data)
frame_count += 1
@@ -130,7 +156,7 @@ class MultiSensor:
multi.stop()
"""
def __init__(self, sensor_configs):
def __init__(self, sensor_configs, compute_contact=False):
"""
Initialize MultiSensor manager.
@@ -155,7 +181,7 @@ class MultiSensor:
self.sensor_configs = sensor_configs
self.num_sensors = len(sensor_configs)
self._mp = mp
self._use_async = False
self.compute_contact = compute_contact
self.processes = []
self.data_queues = []
@@ -167,33 +193,24 @@ class MultiSensor:
self.data_queues.append(mp.Queue(maxsize=2))
self.stop_events.append(mp.Event())
def start(self, use_async=False):
"""
Start all sensor worker processes.
Args:
use_async: If True, each sensor process uses 3-stage async pipeline
for maximum throughput (Capture → Flow → HHD in parallel)
"""
def start(self):
"""Start all sensor worker processes."""
if self._started:
print("MultiSensor already started")
return
self._use_async = use_async
for i, config in enumerate(self.sensor_configs):
process = self._mp.Process(
target=_sensor_worker,
args=(i, config, self.data_queues[i], self.stop_events[i], use_async),
args=(i, config, self.data_queues[i], self.stop_events[i], self.compute_contact),
daemon=True
)
process.start()
self.processes.append(process)
print(f"Started sensor {i} (PID: {process.pid})")
mode = "async (3-stage pipeline)" if use_async else "sync"
self._started = True
print(f"All {len(self.processes)} sensors started in {mode} mode")
print(f"All {len(self.processes)} sensors started")
def get(self, sensor_idx, timeout=0.01):
"""
@@ -283,8 +300,8 @@ class MultiSensor:
print("All sensors stopped.")
def __enter__(self):
"""Context manager entry - starts sensors (sync mode by default)."""
self.start(use_async=False)
"""Context manager entry - starts sensors."""
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
@@ -296,22 +313,3 @@ class MultiSensor:
"""Cleanup on deletion."""
self.stop()
@classmethod
def async_context(cls, sensor_configs):
"""
Create a MultiSensor that uses async mode when used as context manager.
Example:
with MultiSensor.async_context(configs) as multi:
for all_data in multi:
...
"""
return _AsyncMultiSensorContext(sensor_configs)
class _AsyncMultiSensorContext(MultiSensor):
"""Helper class for async context manager."""
def __enter__(self):
self.start(use_async=True)
return self