diff --git a/config/compound_eye.png b/config/compound_eye.png index 0b03dc8..ef3b17b 100644 Binary files a/config/compound_eye.png and b/config/compound_eye.png differ diff --git a/example3_two_sensor_pipeline.py b/example3_two_sensor_pipeline.py new file mode 100644 index 0000000..c25a0bb --- /dev/null +++ b/example3_two_sensor_pipeline.py @@ -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() diff --git a/example6_multi_async.py b/example6_multi_async.py index 3fc369e..d3e6168 100644 --- a/example6_multi_async.py +++ b/example6_multi_async.py @@ -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: diff --git a/example7_five_sensors.py b/example7_five_sensors.py index 9385879..b907edf 100644 --- a/example7_five_sensors.py +++ b/example7_five_sensors.py @@ -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' 退出") # ===================================================================== diff --git a/src/orisys/MultiSensor.py b/src/orisys/MultiSensor.py index 374adc0..8fb8c77 100644 --- a/src/orisys/MultiSensor.py +++ b/src/orisys/MultiSensor.py @@ -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): @@ -295,23 +312,4 @@ class MultiSensor: def __del__(self): """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