From adaede435f4a47807230c6f6422b6014ded5c243 Mon Sep 17 00:00:00 2001 From: bluepulsar Date: Sat, 27 Dec 2025 16:30:36 +0800 Subject: [PATCH] update comment --- example1_basic.py | 142 +++++++++++++++++++++++++++++------- example4_multiple_sensor.py | 85 ++++++++++++++++++--- example5_async.py | 89 +++++++++++++++++++--- example6_multi_async.py | 100 +++++++++++++++++++------ example7_five_sensors.py | 99 ++++++++++++++++++++----- 5 files changed, 424 insertions(+), 91 deletions(-) diff --git a/example1_basic.py b/example1_basic.py index 895cc70..b5b04af 100644 --- a/example1_basic.py +++ b/example1_basic.py @@ -1,42 +1,130 @@ +""" +示例 1: 基础用法 - 单传感器同步处理 + +本示例演示 OrisysSensor 的基本使用流程： +1. 初始化传感器（视频文件或摄像头） +2. 循环获取图像并计算形变 +3. 读取各类传感器数据 +4. 可视化显示 + +适用场景：单传感器、简单应用、学习入门 +""" + import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" # Windows 下禁用硬件加速，避免兼容性问题 + import cv2 import orisys + def main(): + # ========================================================================= + # 1. 初始化传感器 + # ========================================================================= + # 参数说明： + # vid_src: 视频源，可以是文件路径（如 "./s1.mp4"）或摄像头索引（如 0） + # isstitch: 是否启用复眼拼接模式（多单元传感器需要开启） + # cuda: 是否使用 CUDA GPU 加速（需要 CuPy 和 CUDA 支持） + # verbose: 是否打印详细初始化日志 + + # 方式1: 从视频文件读取 + sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True) + + # 方式2: 从摄像头读取（取消注释使用） + # sensor = orisys.Sensor(0, isstitch=True, cuda=True, verbose=True) - # 打开视频 - sensor1 = orisys.Sensor("./s1.mp4",isstitch=True,cuda = True,verbose = True) - # 打开摄像头 - # sensor1 = Sensor(0,isstitch=True,cuda = True,verbose = True) - + print("\n按 'q' 退出程序，按 'r' 重置光流追踪器") + # ========================================================================= + # 2. 主循环：获取图像 → 计算形变 → 读取数据 → 显示 + # ========================================================================= while True: - sensor1.get_img() # 提取并拼接复眼图像 - sensor1.compute_deformation() # 计算形变和分解 - # 提取信息： - # 向量场: VRAW, VNORMAL, VSHEAR: 原始，法向，切向 - # 帧率，时间戳: FPS, TIMESTAMP - # 变形量总和: FNORMAL , FSHEARX, FSHEARY : 法向，切向（X)，切向（Y) - # 深度图: DEPTH - fps, fn, fx, fy, flow, depth = sensor1.read_info(sensor1.info.FPS, sensor1.info.FNORMAL, sensor1.info.FSHEARX, sensor1.info.FSHEARY, sensor1.info.VRAW, sensor1.info.DEPTH) - print(fps,fn,fx,fy) - # 显示向量场 - arrows = orisys.util.draw_arrows(sensor1.img, flow, threshold=5, grid_spacing=40, arrow_scale=1.0) - cv2.imshow("Deformation", arrows) + # 步骤 2.1: 获取并拼接图像 + # get_img() 完成以下操作： + # - 从视频源读取一帧 + # - 转换为灰度图 + # - 提取各个视觉单元并拼接 + # - 应用时域滤波降噪 + sensor.get_img() + + # 步骤 2.2: 计算形变场 + # compute_deformation() 完成以下操作： + # - 计算光流（追踪像素位移） + # - HHD 分解（分离法向和切向分量） + # - 计算力的大小 + # - 计算深度图 + sensor.compute_deformation() + + # 步骤 2.3: 读取传感器数据 + # read_info() 支持以下数据类型： + # 向量场数据： + # - VRAW: 原始光流场 (H, W, 2) 的 numpy 数组 + # - VNORMAL: 法向分量场 + # - VSHEAR: 切向分量场 + # 标量数据： + # - FNORMAL: 法向力总和（正压力） + # - FSHEARX: X方向切向力 + # - FSHEARY: Y方向切向力 + # 其他： + # - FPS: 当前帧率 + # - TIMESTAMP: 时间戳字符串 + # - DEPTH: 深度图（归一化 0-1） + fps, fn, fx, fy, flow, depth = sensor.read_info( + sensor.info.FPS, + sensor.info.FNORMAL, + sensor.info.FSHEARX, + sensor.info.FSHEARY, + sensor.info.VRAW, + sensor.info.DEPTH + ) + + # 打印力的数值 + print(f"FPS: {fps:5.1f}, 法向力: {fn:8.2f}, 切向力X: {fx:8.2f}, 切向力Y: {fy:8.2f}") + + # ===================================================================== + # 3. 可视化显示 + # ===================================================================== + # 绘制光流箭头图 + # 参数说明： + # img: 背景图像 + # flow: 光流场 + # threshold: 显示阈值（小于此值的箭头不显示） + # grid_spacing: 箭头网格间距（像素） + # arrow_scale: 箭头缩放比例 + arrows = orisys.util.draw_arrows( + sensor.img.copy(), # 使用 copy() 避免修改原图 + flow, + threshold=5, + grid_spacing=40, + arrow_scale=1.0 + ) + cv2.imshow("形变场 Deformation", arrows) - depth_img = cv2.applyColorMap((depth*1* 255.0).astype('uint8'), cv2.COLORMAP_HOT) - cv2.imshow('depth', depth_img) + # 绘制深度图（热力图） + depth_img = cv2.applyColorMap( + (depth * 255.0).astype('uint8'), + cv2.COLORMAP_HOT + ) + cv2.imshow("深度图 Depth", depth_img) - key = cv2.waitKey(1) - if key & 0xFF == ord("q"): - print("\n 停止程序") + # ===================================================================== + # 4. 键盘控制 + # ===================================================================== + key = cv2.waitKey(1) & 0xFF + if key == ord("q"): + print("\n退出程序...") break - if key & 0xFF == ord("r"): - sensor1.reset() - - sensor1.disconnect() # 断开传感器连接 + if key == ord("r"): + # 重置光流追踪器（当追踪漂移时使用） + sensor.reset() + print("光流追踪器已重置") + # ========================================================================= + # 5. 清理资源 + # ========================================================================= + sensor.disconnect() # 释放视频源和相关资源 + cv2.destroyAllWindows() + if __name__ == '__main__': main() diff --git a/example4_multiple_sensor.py b/example4_multiple_sensor.py index 246609c..b13991a 100644 --- a/example4_multiple_sensor.py +++ b/example4_multiple_sensor.py @@ -1,11 +1,26 @@ """ -Example 4: Multi-Sensor with Multiprocessing +示例 4: 多传感器并行处理（多进程模式） -Each sensor runs in its own process with independent CUDA context. +本示例演示如何同时运行多个传感器： +- 每个传感器运行在独立的进程中 +- 每个进程拥有独立的 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" +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" # Windows 兼容性设置 import cv2 import orisys @@ -13,33 +28,79 @@ import multiprocessing as mp def main(): + # ========================================================================= + # 1. 配置多个传感器 + # ========================================================================= + # 每个传感器的配置是一个字典，包含： + # vid_src: 视频源（文件路径或摄像头索引） + # isstitch: 是否启用拼接模式 + # cuda: 是否使用 GPU 加速 + # cal_path: 标定文件路径 configs = [ - {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, - {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + { + 'vid_src': './s1.mp4', + 'isstitch': True, + 'cuda': True, + 'cal_path': './config/cal.npy' + }, + { + 'vid_src': './s1_copy.mp4', + 'isstitch': True, + 'cuda': True, + 'cal_path': './config/cal.npy' + }, ] + # ========================================================================= + # 2. 使用上下文管理器启动多传感器 + # ========================================================================= + # MultiSensor 会自动： + # - 为每个配置创建独立的子进程 + # - 在子进程中初始化 Sensor 实例 + # - 持续采集和处理数据 + # - 通过队列将结果传回主进程 with orisys.MultiSensor(configs) as multi: - print("\nPress 'q' to quit") + 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"Sensor {sid}: FPS={data['fps']:5.1f}, Fn={data['fnormal']:8.2f}") + 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"Sensor {sid}", arrows) + arrows = orisys.util.draw_arrows( + data['img'].copy(), + data['flow'], + threshold=5 + ) + cv2.imshow(f"传感器 {sid} - 形变", arrows) + # 显示深度图 if data['depth'] is not None: - cv2.imshow(f"Sensor {sid} - Depth", - cv2.applyColorMap((data['depth'] * 255).astype('uint8'), cv2.COLORMAP_HOT)) + depth_img = cv2.applyColorMap( + (data['depth'] * 255).astype('uint8'), + cv2.COLORMAP_HOT + ) + cv2.imshow(f"传感器 {sid} - 深度", depth_img) + # 检查退出键 if cv2.waitKey(1) & 0xFF == ord('q'): break + # 上下文管理器退出时自动停止所有子进程 cv2.destroyAllWindows() if __name__ == '__main__': + # Windows 下多进程必需的初始化 mp.freeze_support() main() diff --git a/example5_async.py b/example5_async.py index 31b7fe3..e4a69ad 100644 --- a/example5_async.py +++ b/example5_async.py @@ -1,46 +1,111 @@ """ -Example 5: Async/Pipelined Processing +示例 5: 异步流水线处理（单传感器高吞吐模式） -Overlaps image capture with deformation computation for higher throughput. -Pipeline: [Capture Thread] -> [Queue] -> [Compute Thread] -> [Queue] -> [Main] +本示例演示传感器的异步三阶段流水线： +- 图像采集、光流计算、HHD分解并行执行 +- 通过队列在线程间传递数据 +- 实现更高的帧率和吞吐量 + +流水线架构： + [采集线程] → 队列 → [光流线程] → 队列 → [HHD线程] → 队列 → [主线程显示] + + 时间线示例（每格代表一帧的处理时间）： + 采集: |帧1|帧2|帧3|帧4|帧5|... + 光流: |帧1|帧2|帧3|帧4|... + HHD: |帧1|帧2|帧3|... + + 优势：三个阶段重叠执行，总吞吐量约为最慢阶段的速度 + +适用场景： +- 单传感器需要最大帧率 +- CPU/GPU 利用率需要最大化 +- 可以接受 2-3 帧的处理延迟 + +对比同步模式： +- 同步模式：采集→光流→HHD 串行执行，FPS 受限于总时间 +- 异步模式：三阶段并行，FPS 约为同步模式的 2-3 倍 """ import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" # Windows 兼容性设置 import cv2 import orisys def main(): - sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True) + # ========================================================================= + # 1. 初始化传感器 + # ========================================================================= + sensor = orisys.Sensor( + "./s1.mp4", + isstitch=True, + cuda=True, + verbose=True + ) + + # ========================================================================= + # 2. 启用异步流水线 + # ========================================================================= + # queue_size: 每个阶段间队列的大小 + # - 较小的值（1-2）：延迟低，但可能丢帧 + # - 较大的值（3-5）：吞吐稳定，但延迟增加 sensor.enable_async(queue_size=2) - print("\nPress 'q' to quit, 'r' to reset") + print("\n异步模式已启用") + print("按 'q' 退出，按 'r' 重置光流追踪器") + # ========================================================================= + # 3. 主循环：非阻塞获取数据 + # ========================================================================= try: while True: + # 非阻塞获取最新结果 + # timeout: 等待超时时间（秒） + # - 返回 None 表示暂无新数据 + # - 返回 dict 包含所有传感器数据 data = sensor.get_async(timeout=0.05) + if data is None: + # 暂无新数据，继续等待 continue - print(f"FPS: {data['fps']:5.1f}, Fn: {data['fnormal']:8.2f}, " - f"Fx: {data['fshearx']:8.2f}, Fy: {data['fsheary']:8.2f}") + # 打印力的数值 + print(f"FPS: {data['fps']:5.1f}, " + f"法向力: {data['fnormal']:8.2f}, " + f"切向力X: {data['fshearx']:8.2f}, " + f"切向力Y: {data['fsheary']: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("Deformation", arrows) + arrows = orisys.util.draw_arrows( + data['img'].copy(), + data['flow'], + threshold=5 + ) + cv2.imshow("异步模式 - 形变", arrows) + # 显示深度图 if data['depth'] is not None: - depth_img = cv2.applyColorMap((data['depth'] * 255).astype('uint8'), cv2.COLORMAP_HOT) - cv2.imshow("Depth", depth_img) + depth_img = cv2.applyColorMap( + (data['depth'] * 255).astype('uint8'), + cv2.COLORMAP_HOT + ) + cv2.imshow("异步模式 - 深度", depth_img) + # 键盘控制 key = cv2.waitKey(1) & 0xFF if key == ord('q'): break elif key == ord('r'): sensor.reset() + print("光流追踪器已重置") + finally: + # ===================================================================== + # 4. 清理资源 + # ===================================================================== + # disconnect() 会自动调用 disable_async() 停止所有线程 sensor.disconnect() cv2.destroyAllWindows() diff --git a/example6_multi_async.py b/example6_multi_async.py index 1b9bb9b..047f430 100644 --- a/example6_multi_async.py +++ b/example6_multi_async.py @@ -1,18 +1,31 @@ """ -Example 6: Multi-Sensor with Async Pipeline +示例 6: 多传感器 + 异步流水线（最高性能模式） -Combines multiprocessing (separate CUDA contexts) with async 3-stage pipeline -(Capture → Flow → HHD) for maximum throughput. +本示例结合了多进程和异步流水线的优势： +- 多进程：每个传感器运行在独立进程，拥有独立 CUDA 上下文 +- 异步流水线：每个进程内部使用三阶段流水线 -Architecture: - Process 1 (Sensor 0): [Capture] → [Flow] → [HHD] → Queue - Process 2 (Sensor 1): [Capture] → [Flow] → [HHD] → Queue - ↓ - Main Process: Display ← ─────────────────────┘ +这是性能最高的配置，适用于： +- 多个高分辨率传感器 +- 需要最大化总吞吐量 +- 有充足的 CPU 核心和 GPU 显存 + +架构说明： + 进程 1 (传感器 0): + [采集] → [光流] → [HHD] → 队列 ──┐ + 进程 2 (传感器 1): │ + [采集] → [光流] → [HHD] → 队列 ──┼──→ 主进程显示 + ... │ + 进程 N (传感器 N-1): │ + [采集] → [光流] → [HHD] → 队列 ──┘ + +资源占用估算（每个传感器）： +- CPU: 约 2-3 个核心（采集、光流、HHD 各一个线程） +- GPU 显存: 约 500MB-1GB（取决于分辨率和 CUDA 算法） """ import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" # Windows 兼容性设置 import cv2 import orisys @@ -20,24 +33,54 @@ import multiprocessing as mp def main(): + # ========================================================================= + # 1. 配置多个传感器 + # ========================================================================= configs = [ - {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, - {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, + { + 'vid_src': './s1.mp4', # 视频源 + 'isstitch': True, # 启用拼接 + 'cuda': True, # 启用 GPU 加速 + 'cal_path': './config/cal.npy' # 标定文件 + }, + { + 'vid_src': './s1_copy.mp4', + 'isstitch': True, + 'cuda': True, + 'cal_path': './config/cal.npy' + }, ] - # Method 1: Using async_context (recommended) + # ========================================================================= + # 2. 使用 async_context 启动（推荐方式） + # ========================================================================= + # async_context 是一个便捷的上下文管理器： + # - 自动启用每个传感器进程内的异步流水线 + # - 退出时自动清理所有资源 with orisys.MultiSensor.async_context(configs) as multi: - print("\nAsync multi-sensor started. Press 'q' to quit") + print("\n多传感器异步模式已启动，按 'q' 退出") + # ===================================================================== + # 3. 迭代获取所有传感器数据 + # ===================================================================== for all_data in multi: + # 过滤有效数据 for data in [d for d in all_data if d]: sid = data['sensor_id'] - print(f"Sensor {sid}: FPS={data['fps']:5.1f}, Fn={data['fnormal']:8.2f}") + # 打印状态 + 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"Sensor {sid}", arrows) + arrows = orisys.util.draw_arrows( + data['img'].copy(), + data['flow'], + threshold=5 + ) + cv2.imshow(f"传感器 {sid} - 形变", arrows) + # 检查退出键 if cv2.waitKey(1) & 0xFF == ord('q'): break @@ -45,27 +88,42 @@ def main(): def main_manual(): - """Alternative: Manual start with async flag""" + """ + 备选方式：手动控制启动和停止 + + 适用于需要更精细控制的场景，例如： + - 动态添加/移除传感器 + - 在运行时切换同步/异步模式 + - 自定义错误处理 + """ configs = [ {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, {'vid_src': './s1_copy.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, ] + # 创建 MultiSensor 实例 multi = orisys.MultiSensor(configs) - multi.start(use_async=True) # Enable async mode + + # 手动启动，use_async=True 启用异步模式 + multi.start(use_async=True) try: while True: - for data in [d for d in multi.get_all() if d]: - cv2.imshow(f"Sensor {data['sensor_id']}", data['img']) + # 获取所有传感器的最新数据 + all_data = multi.get_all() + + for data in [d for d in all_data if d]: + cv2.imshow(f"传感器 {data['sensor_id']}", data['img']) + if cv2.waitKey(1) & 0xFF == ord('q'): break finally: + # 手动停止所有进程 multi.stop() cv2.destroyAllWindows() if __name__ == '__main__': + # Windows 下多进程必需的初始化 mp.freeze_support() main() - diff --git a/example7_five_sensors.py b/example7_five_sensors.py index 029421a..9385879 100644 --- a/example7_five_sensors.py +++ b/example7_five_sensors.py @@ -1,12 +1,29 @@ """ -Example 7: Five Sensors with Async Pipeline +示例 7: 五传感器网格显示 -Demonstrates scaling to 5 sensors running in parallel with async mode. -Each sensor runs in its own process with 3-stage async pipeline. +本示例演示如何扩展到 5 个传感器： +- 使用异步多进程模式运行所有传感器 +- 将 5 个传感器画面拼接成 2x3 网格显示 +- 实时显示每个传感器的 FPS 和力数据 + +网格布局： + ┌─────────┬─────────┬─────────┐ + │ 传感器0 │ 传感器1 │ 传感器2 │ + ├─────────┼─────────┼─────────┤ + │ 传感器3 │ 传感器4 │ (空) │ + └─────────┴─────────┴─────────┘ + +硬件要求估算： +- CPU: 建议 8 核以上（每个传感器约 2-3 线程） +- GPU: 建议 8GB 显存以上（每个传感器约 500MB-1GB） +- 内存: 建议 16GB 以上 + +注意：本示例使用同一个视频文件模拟 5 个传感器， + 实际使用时请替换为不同的视频源或摄像头索引。 """ import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" # Windows 兼容性设置 import cv2 import orisys @@ -15,7 +32,13 @@ import numpy as np def main(): - # Configure 5 sensors (using same video for demo - replace with actual sources) + # ========================================================================= + # 1. 配置 5 个传感器 + # ========================================================================= + # 注意：这里使用同一个视频文件作为演示 + # 实际应用中应替换为不同的视频源： + # - 摄像头: 0, 1, 2, 3, 4 + # - 视频文件: './sensor0.mp4', './sensor1.mp4', ... configs = [ {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, @@ -24,43 +47,81 @@ def main(): {'vid_src': './s1.mp4', 'isstitch': True, 'cuda': True, 'cal_path': './config/cal.npy'}, ] + # ========================================================================= + # 2. 启动多传感器异步模式 + # ========================================================================= with orisys.MultiSensor.async_context(configs) as multi: - print(f"\n{len(configs)} sensors started. Press 'q' to quit") + print(f"\n{len(configs)} 个传感器已启动，按 'q' 退出") + # ===================================================================== + # 3. 主循环 + # ===================================================================== for all_data in multi: + # 过滤有效数据（有图像的传感器） valid_data = [d for d in all_data if d and d['img'] is not None] - # Print stats for all sensors + # ----------------------------------------------------------------- + # 3.1 打印所有传感器状态（单行显示） + # ----------------------------------------------------------------- for data in valid_data: - print(f"S{data['sensor_id']}: FPS={data['fps']:4.1f} Fn={data['fnormal']:7.1f}", end=" | ") + print(f"S{data['sensor_id']}: " + f"FPS={data['fps']:4.1f} " + f"Fn={data['fnormal']:7.1f}", end=" | ") if valid_data: - print() + print() # 换行 - # Create grid display (2 rows: 3 + 2) + # ----------------------------------------------------------------- + # 3.2 准备网格显示的图像 + # ----------------------------------------------------------------- imgs = [] for data in valid_data: + # 绘制光流箭头 if data['flow'] is not None: - img = orisys.util.draw_arrows(data['img'].copy(), data['flow'], threshold=5) + img = orisys.util.draw_arrows( + data['img'].copy(), + data['flow'], + threshold=5 + ) else: img = data['img'] - # Add sensor ID label - cv2.putText(img, f"S{data['sensor_id']}", (10, 30), - cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) + + # 在图像上标注传感器编号 + cv2.putText( + img, + f"S{data['sensor_id']}", + (10, 30), + cv2.FONT_HERSHEY_SIMPLEX, + 0.8, + (0, 255, 0), + 2 + ) imgs.append(img) + # ----------------------------------------------------------------- + # 3.3 创建 2x3 网格布局 + # ----------------------------------------------------------------- if len(imgs) >= 5: - # Create 2x3 grid (pad with black if needed) h, w = imgs[0].shape[:2] - black = np.zeros((h, w, 3), dtype=np.uint8) if len(imgs[0].shape) == 3 else np.zeros((h, w), dtype=np.uint8) + # 创建黑色占位图像（用于第6格） + if len(imgs[0].shape) == 3: + black = np.zeros((h, w, 3), dtype=np.uint8) + else: + black = np.zeros((h, w), dtype=np.uint8) + + # 拼接网格 + # 第一行: 传感器 0, 1, 2 row1 = np.hstack([imgs[0], imgs[1], imgs[2]]) + # 第二行: 传感器 3, 4, 空 row2 = np.hstack([imgs[3], imgs[4], black]) + # 合并两行 grid = np.vstack([row1, row2]) - # Resize for display + # 缩放到适合显示的尺寸 grid = cv2.resize(grid, (1200, 800)) - cv2.imshow("5 Sensors Grid", grid) + cv2.imshow("5 传感器网格视图", grid) + # 检查退出键 if cv2.waitKey(1) & 0xFF == ord('q'): break @@ -68,6 +129,6 @@ def main(): if __name__ == '__main__': + # Windows 下多进程必需的初始化 mp.freeze_support() main() -