diff --git a/Readme.md b/Readme.md index b731d13..665bc80 100644 --- a/Readme.md +++ b/Readme.md @@ -73,7 +73,7 @@ conda install conda-forge::pyvkfft * **vid_src** (`int | str`, 必填): 传感器 ID、相机编号或视频路径。默认为 `0`。 * **config_name** (`str`，可选): 传感器配置`json`文件路径，用于加载对应的参数配置，默认为 `./config/ddjx01`。 -* **cal_path** (`str`，可选): 复眼拼接校准文件路径，默认为 `./config/cal.npy`。有多个传感器同时运行时建议传感器类创建时建议填写，避免校准文件覆写。在首次运行时程序会在找不到文件时自动创建。 +* * **cal_path** (`str`，可选，已弃用): 该参数已弃用，当前版本中不再使用，传入后也会被忽略。为兼容旧代码暂时保留。: 复眼拼接校准文件路径，默认为 `./config/cal.npy`。有多个传感器同时运行时建议传感器类创建时建议填写，避免校准文件覆写。在首次运行时程序会在找不到文件时自动创建。 * **backend** (`str`，可选): 光流场解耦计算后端，默认为 `auto`，可选`cupy`,`opencl`,`cpu`。 * **verbose** (`bool`，可选): 是否输出初始化及运行过程中的详细日志信息，默认为 `False`。 @@ -99,12 +99,12 @@ sensor = Sensor("/data/data.mp4") 采集并预处理一帧图像并存到`Sensor` 类别中。 + ### 返回 -* `np.ndarray`：处理后的BGR格式图像(`uint8`，尺寸为json文件中`resolution_stitch`对应分辨率)。 -* `None`：当输入源读取失败（例如视频结束、摄像头断开）时返回。 +* `bool`：是否成功读取图像。 ### 状态更新 -* 成功时更新`sensor`类里面的图像。 +* 成功时更新`sensor`类里面的`sensor.img`。处理后的BGR格式图像(`uint8`，尺寸为json文件中`resolution_stitch`对应分辨率)。 ### 示例 ```python @@ -123,6 +123,8 @@ sensor.get_img() ### 输入参数 * **check_motion**(`bool`，可选): 为 `True` 时先根据光流判断是否有运动，有运动才做 HHD 分解；为 `False` 时每帧都做 HHD。默认 `True`。 * **threshold**(`float`，可选): **check_motion** 功能的阈值，默认为`6`。 +* **tail_frames** (`int`，可选): 光流变化低于阈值后继续解算多少帧，确保传感器回复，，默认为 `10`. + ### 返回 * `bool`：是否接触到物体，如果**check_motion**为`False`时返回`None`。 diff --git a/build_nuitka.py b/build_nuitka.py index 51adb6d..e58c054 100644 --- a/build_nuitka.py +++ b/build_nuitka.py @@ -39,6 +39,16 @@ def find_compiled_extension(output_dir: Path) -> Path: return sorted(candidates, key=lambda p: len(p.name))[0] +def move_wheel_to_sdk(wheel_path: Path, version: str) -> Path: + sdk_version_dir = Path("SDK") / version + sdk_version_dir.mkdir(parents=True, exist_ok=True) + dest_wheel = sdk_version_dir / wheel_path.name + if dest_wheel.exists(): + dest_wheel.unlink() + shutil.move(str(wheel_path), str(dest_wheel)) + return dest_wheel + + def build(): install_nuitka() @@ -143,9 +153,10 @@ setup( print("\n=== Building Wheel ===") subprocess.check_call([sys.executable, "setup.py", "bdist_wheel"], cwd=dist_dir) - wheels = list((dist_dir / "dist").glob("*.whl")) + wheels = sorted((dist_dir / "dist").glob("*.whl")) if wheels: - print(f"\n[SUCCESS] Wheel created: {wheels[0]}") + final_wheel = move_wheel_to_sdk(wheels[0], version) + print(f"\n[SUCCESS] Wheel created and moved to: {final_wheel}") else: print("\n[ERROR] Wheel not found.") diff --git a/config/ddjx01.json b/config/ddjx01.json index 988ab1e..21b5e96 100644 --- a/config/ddjx01.json +++ b/config/ddjx01.json @@ -11,7 +11,7 @@ }, "stitcher": { "min_area_frac": 0.01, - "max_area_frac": 0.04, + "max_area_frac": 0.05, "max_aspect_ratio": 50, "index": [1,2,3,6,7,8,11,12,13], "margin": 10, diff --git a/doc/get_started.docx b/doc/get_started.docx index 5426117..8fad3c3 100644 Binary files a/doc/get_started.docx and b/doc/get_started.docx differ diff --git a/doc/get_started.md b/doc/get_started.md index 5ef11c0..053ce13 100644 --- a/doc/get_started.md +++ b/doc/get_started.md @@ -1,6 +1,6 @@ 叠动科技软件使用手册 版本： 0.3.1 -更新日期： 2026年3月6日 +更新日期： 2026年4月16日 # 简介 本文件介绍如何上手和运行叠动视触觉传感器配套的开发套件。本套件是配合叠动视触觉传感器进行图像处理，计算形变、力以及接触信息。它可以从摄像头或视频文件中读取图像，处理后打印和展示处理后信息。 diff --git a/example1_basic_pipeline.py b/example1_basic_pipeline.py index d8763c6..8faaa95 100644 --- a/example1_basic_pipeline.py +++ b/example1_basic_pipeline.py @@ -1,13 +1,13 @@ """ -示例 1: 基础用法 - 单传感器同步处理 - -本示例演示 orisys 的基本使用流程： -1. 初始化传感器（视频文件或摄像头） -2. 循环获取图像并计算形变和触碰点 -3. 读取各类传感器数据 -4. 可视化显示 +示例 1：单传感器基础流程 +本示例演示 Orisys SDK 的标准处理流程： +1. 初始化传感器（摄像头或视频文件） +2. 获取图像并计算形变 +3. 按需计算接触区域 +4. 读取并可视化结果 +SDK version: 0.3.1 """ import os os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" @@ -18,39 +18,38 @@ import argparse def main(): parser = argparse.ArgumentParser() - parser.add_argument("--video","-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose","-verbose", type=bool, default=True, help="Print detailed information") + parser.add_argument("--video","-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径") + parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="配置文件名称或路径") + parser.add_argument("--verbose","-verbose", type=bool, default=True, help="是否输出详细日志") args = parser.parse_args() - # 步骤1：创建传感器对象 (必要步骤)， - # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） - # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 + # 步骤 1：创建传感器对象（必需） + # config_name 可以是内置配置名称，也可以是自定义配置文件路径 + # cal_path 为标定文件路径；每台设备建议使用独立的标定文件 - # 方式1: 从视频文件读取 + # 输入既可以是摄像头编号，也可以是视频文件路径 try: input = int(args.video) is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose) + sensor = orisys.Sensor(input, config_name=args.config, verbose=args.verbose) except: is_camera = False - sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) + sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose) - print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") + print("\n按键说明：'q' 退出程序，'r' 重置追踪器。") while True: - # 步骤2：信息提取 - # 步骤 2.1: 获取并拼接图像 (必要步骤) + # 步骤 2：数据处理 + # 步骤 2.1：获取并拼接图像（必需） sensor.get_img() - # 步骤 2.2: 计算形变场 (必要步骤) + # 步骤 2.2：计算形变场（必需） is_touched = sensor.compute_deformation(check_motion=True, threshold=6) - # 步骤 2.3: 计算触碰点 + # 步骤 2.3：检测到接触后再计算接触区域 if is_touched: sensor.compute_contact() - #步骤 2.4: 读取信息 + # 步骤 2.4：读取结果 fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info( sensor.info.FPS, sensor.info.VRAW, @@ -64,8 +63,8 @@ def main(): sensor.info.FSHEARY ) - #步骤 3: 可视化显示 - # 步骤 3.1: 绘制形变场 + # 步骤 3：可视化显示 + # 步骤 3.1：显示形变矢量场 arrows = orisys.util.draw_arrows( img, flow, @@ -73,40 +72,43 @@ def main(): grid_spacing=20, arrow_scale=1.0 ) - cv2.imshow("Deformation", arrows) + cv2.imshow("形变矢量场", arrows) - # 步骤 3.2: 在图像上触碰点和区域 + # 步骤 3.2：显示接触区域和质心 image_with_foe = orisys.util.draw_contact(img, contour, centroid) - cv2.imshow("Contact Centroid", image_with_foe) + cv2.imshow("接触区域与质心", image_with_foe) - # 步骤 3.3: 绘制深度图 - # 归一化散度值到0-255范围（使用绝对值，然后归一化） + # 步骤 3.3：显示深度图 + # 将深度值归一化到 0-255 范围 div_abs = 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) - # 应用颜色映射（使用JET：蓝色=低散度，红色=高散度） - cv2.imshow("Depth Map", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) + # 应用伪彩色映射 + cv2.imshow("深度图", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) - #步骤 3.4: 打印资料 - print(f"fps: {fps}, fnormal: {fnormal}, fshearx: {fshearx}, fsheary: {fsheary}, depth_shape: {depth_map.shape}, sensor.resolution_raw: {sensor.img_size_raw}") + # 步骤 3.4：打印关键结果 + print( + f"FPS={fps:.2f}, 法向力={fnormal:.4f}, 切向力X={fshearx:.4f}, " + f"切向力Y={fsheary:.4f}, 深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}" + ) - #步骤 3.5: 键盘控制 + # 步骤 3.5：键盘控制 key = cv2.waitKey(1) & 0xFF if key == ord("q"): - print("\n退出程序...") + print("\n正在退出示例程序...") break if key == ord("r"): - # 重置光流追踪器（当追踪漂移时使用） + # 当光流追踪出现漂移时，可手动重置追踪器 sensor.reset() - print("传感器已重置") + print("追踪器已重置。") # ========================================================================= - # 4. 清理资源 + # 4. 释放资源 # ========================================================================= - sensor.disconnect() # 释放视频源和相关资源 + sensor.disconnect() # 释放视频源及相关资源 if __name__ == '__main__': diff --git a/example2_two_sensor_pipeline.py b/example2_two_sensor_pipeline.py index 60e5fd4..0bf7c5a 100644 --- a/example2_two_sensor_pipeline.py +++ b/example2_two_sensor_pipeline.py @@ -1,22 +1,24 @@ """ -示例 4: 多传感器并行处理（多进程模式 - 不使用 MultiSensor） +示例 2：双传感器并行处理（多进程） 本示例演示如何使用 multiprocessing 同时运行两个传感器： -- 每个传感器运行在独立的进程中 +- 每个传感器运行在独立进程中 - 每个进程拥有独立的 CUDA 上下文 -- 使用 multiprocessing.Process 和 Queue 进行进程间通信 +- 通过 Process 和 Queue 在主进程中汇总结果并显示 架构说明： - 进程 1 (传感器 0): 采集 → 光流 → HHD → 接触检测 → 队列 - 进程 2 (传感器 1): 采集 → 光流 → HHD → 接触检测 → 队列 + 进程 1（传感器 0）：采集 → 光流 → HHD → 接触检测 → 队列 + 进程 2（传感器 1）：采集 → 光流 → HHD → 接触检测 → 队列 ↓ - 主进程: 显示 ← ───────────────┘ + 主进程： 显示 ← ───────────────┘ -适用场景：多指触觉传感器、多视角采集、需要 GPU 加速的多路处理 +适用场景：多指触觉传感器、多视角采集，以及需要 GPU 加速的多路处理。 注意事项： - Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support() -- 每个传感器进程会占用独立的 GPU 显存 +- 每个传感器进程都会占用独立的 GPU 显存 + +SDK version: 0.3.1 """ import os os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" @@ -26,20 +28,20 @@ import numpy as np import multiprocessing as mp from multiprocessing import Queue, Process import time +import argparse -def sensor_worker(sensor_id, vid_src, config_name, cal_path, result_queue, stop_event, target_fps=None): +def sensor_worker(sensor_id, vid_src, config_name, result_queue, stop_event, target_fps=None): """ - 传感器工作进程函数 + 传感器工作进程。 Args: sensor_id: 传感器编号 vid_src: 视频源（文件路径或摄像头索引） config_name: 配置名称 - cal_path: 标定路径 result_queue: 结果队列，用于向主进程发送数据 stop_event: 停止事件，用于通知工作进程退出 - target_fps: 目标帧率；若设置则用定时器限速，使多传感器 FPS 接近一致 + target_fps: 目标帧率；设置后将对处理循环限速，使多路传感器 FPS 更接近 """ try: # 在每个进程中创建独立的传感器实例 @@ -48,27 +50,26 @@ def sensor_worker(sensor_id, vid_src, config_name, cal_path, result_queue, stop_ isstitch=True, cuda=True, verbose=False, # 减少子进程输出 - config_name=config_name, - cal_path=cal_path + config_name=config_name ) - print(f"传感器 {sensor_id} 已启动" + (f" (target_fps={target_fps})" if target_fps else "")) + print(f"传感器 {sensor_id} 已启动" + (f"（目标帧率：{target_fps} FPS）" if target_fps else "")) frame_interval = 1.0 / target_fps if target_fps and target_fps > 0 else 0.0 while not stop_event.is_set(): t_start = time.perf_counter() try: - # 步骤1: 获取图像 + # 步骤 1：获取图像 sensor.get_img() - # 步骤2: 计算形变场 + # 步骤 2：计算形变场 sensor.compute_deformation() - # 步骤3: 计算触碰点 + # 步骤 3：计算接触区域（如需要可取消注释） # sensor.compute_contact() - # 步骤4: 读取信息 + # 步骤 4：读取结果 fps, flow, vnormal, img, contour, centroid, depth_map = sensor.read_info( sensor.info.FPS, sensor.info.VRAW, @@ -79,7 +80,7 @@ def sensor_worker(sensor_id, vid_src, config_name, cal_path, result_queue, stop_ sensor.info.DEPTH ) - # 步骤5: 准备数据字典（确保所有数据都是可序列化的） + # 步骤 5：整理结果字典（确保数据可序列化） data = { 'sensor_id': sensor_id, 'fps': fps, @@ -92,100 +93,104 @@ def sensor_worker(sensor_id, vid_src, config_name, cal_path, result_queue, stop_ 'timestamp': time.time() } - # 步骤6: 发送数据到主进程（非阻塞） + # 步骤 6：将结果发送到主进程（非阻塞） try: result_queue.put_nowait(data) except Exception: - # 如果队列已满，跳过这一帧 + # 如果队列已满，则丢弃当前帧，避免阻塞工作进程 pass except Exception as e: - print(f"传感器 {sensor_id} 处理错误: {e}") + print(f"传感器 {sensor_id} 处理失败：{e}") time.sleep(0.1) - # 步骤7: 按 target_fps 限速，使多传感器 FPS 接近 + # 步骤 7：按 target_fps 限速，使多路传感器 FPS 更接近 if frame_interval > 0: elapsed = time.perf_counter() - t_start sleep_time = frame_interval - elapsed if sleep_time > 0.001: # 至少 1ms 才 sleep，避免无意义调度 time.sleep(sleep_time) - # 清理资源 + # 释放资源 sensor.disconnect() - print(f"传感器 {sensor_id} 已停止") + print(f"传感器 {sensor_id} 已停止。") except Exception as e: - print(f"传感器 {sensor_id} 初始化失败: {e}") + print(f"传感器 {sensor_id} 初始化失败：{e}") result_queue.put({'sensor_id': sensor_id, 'error': str(e)}) def main(): # ========================================================================= - # 1. 配置多个传感器 + # 1. 配置多路传感器 # ========================================================================= - # 目标帧率：设为正数时，两路传感器会按此帧率限速，使 FPS 接近一致；设为 None 则不限速 + # 目标帧率：设为正数时，两路传感器会按该帧率限速；设为 None 则不限速 + + parser = argparse.ArgumentParser() + parser.add_argument("--video1","-v1", type=str, default="0", help="视频文件路径") + parser.add_argument("--video2","-v2", type=str, default="0", help="视频文件路径") + args = parser.parse_args() + target_fps = 30 num_sensors = 2 sensor_configs = [ { 'sensor_id': 0, - 'vid_src': 2, - 'config_name': './config/prototype_of3_720p.json', - 'cal_path': './config/ddjx01.npy' + 'vid_src': args.video1, + 'config_name': './config/ddjx01.json' }, { 'sensor_id': 1, - 'vid_src': 4, # 可以改为不同的视频源或摄像头索引 - 'config_name': './config/prototype_of3_720p.json', - 'cal_path': './config/ddjx01.npy' + 'vid_src': args.video2, # 可替换为其他视频源或摄像头索引 + 'config_name': './config/ddjx01.json' }, ] # ========================================================================= # 2. 创建进程间通信对象 # ========================================================================= - # 增大队列大小以缓冲更多帧，避免因主进程处理延迟导致的丢帧 + # 适当增大队列容量，以缓冲更多帧并减轻主进程偶发延迟带来的丢帧 result_queues = [Queue(maxsize=30) for _ in sensor_configs] stop_events = [mp.Event() for _ in sensor_configs] # ========================================================================= - # 3. 启动传感器工作进程 + # 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'], + args=(i, config['vid_src'], config['config_name'], result_queues[i], stop_events[i], target_fps) ) p.start() processes.append(p) - print(f"\n已启动 {len(processes)} 个传感器进程") - print("按 'q' 退出程序") + print(f"\n已启动 {len(processes)} 个传感器进程。") + print("按键说明：'q' 退出程序。") - # 用于存储每个传感器的最新数据 + # 保存每个传感器的最新一帧数据 latest_data = [None] * len(sensor_configs) - # 用于控制打印频率，减少I/O开销 + # 控制打印频率，减少 I/O 开销 print_counter = 0 - print_interval = 30 # 每30帧打印一次，进一步减少I/O + print_interval = 30 # 每 30 帧打印一次 - # 用于控制按键检查频率 + # 控制按键检查频率 key_check_counter = 0 - key_check_interval = 5 # 每5帧检查一次按键 + key_check_interval = 5 # 每 5 帧检查一次按键 try: # ===================================================================== - # 4. 主循环：从队列获取数据并显示 + # 4. 主循环：从队列获取结果并显示 # ===================================================================== while True: - # 从所有队列中快速获取最新数据（高效清空队列，只保留最新帧） + # 从所有队列中快速获取最新数据，只保留最后一帧 for i, queue in enumerate(result_queues): - # 快速清空队列，只保留最后一帧 + # 清空队列，仅保留最后一帧 data = None - # 一次性获取所有可用数据，只保留最后一个 + # 读取所有可用数据，只保留最新结果 while True: try: data = queue.get_nowait() @@ -195,14 +200,14 @@ def main(): if data is not None: latest_data[i] = data - # 显示所有传感器的数据 + # 显示所有传感器的最新结果 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'], @@ -211,7 +216,7 @@ def main(): grid_spacing=10, arrow_scale=1.0 ) - cv2.imshow(f"Deformation {sid}", arrows) + cv2.imshow(f"传感器 {sid} - 形变矢量场", arrows) # # 显示触碰点和区域 # if data['img'] is not None: @@ -233,33 +238,33 @@ def main(): # cv2.imshow(f"Depth Map {sid}", # cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) - # 降低打印频率以减少I/O开销（在主循环级别控制） + # 降低打印频率，减少主循环中的 I/O 开销 print_counter += 1 if print_counter >= print_interval: for data in latest_data: if data is not None and 'error' not in data: sid = data['sensor_id'] - print(f"传感器 {sid}: FPS={data['fps']:5.1f}") + print(f"传感器 {sid}：FPS={data['fps']:5.1f}") print_counter = 0 - # 降低按键检查频率以减少开销 + # 降低按键检查频率，减少额外开销 key_check_counter += 1 if key_check_counter >= key_check_interval: key = cv2.waitKey(1) & 0xFF if key == ord('q'): - print("\n退出程序...") + print("\n正在退出示例程序...") break key_check_counter = 0 finally: # ===================================================================== - # 5. 清理资源 + # 5. 释放资源 # ===================================================================== # 通知所有工作进程停止 for event in stop_events: event.set() - # 等待所有进程结束 + # 等待所有工作进程结束 for p in processes: p.join(timeout=2) if p.is_alive(): @@ -267,10 +272,10 @@ def main(): p.join() cv2.destroyAllWindows() - print("所有传感器进程已停止") + print("所有传感器进程已停止。") if __name__ == '__main__': - # Windows 下多进程必需的初始化 + # Windows 下运行多进程示例所需的初始化 mp.freeze_support() main() diff --git a/example3_plot_data.py b/example3_plot_data.py index 1463545..ca28bb5 100644 --- a/example3_plot_data.py +++ b/example3_plot_data.py @@ -1,3 +1,15 @@ +""" +示例 3：实时曲线与幅值图显示 + +本示例演示如何在读取传感器结果的同时： +1. 显示形变矢量场 +2. 显示流场幅值热力图 +3. 使用 PyQtGraph 实时绘制法向力与切向力曲线 +4. 按键控制录像开始与结束 + +SDK version: 0.3.1 +""" + import os os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" import cv2 @@ -5,10 +17,14 @@ import orisys import numpy as np import time import argparse -import pyqtgraph as pg +try: + import pyqtgraph as pg +except: + print("PyQtGraph is not installed. Please install it using 'pip install pyqt5 pyqtgraph'.") + exit() def create_plot(): - p1 = win.addPlot(title="法相力") + p1 = win.addPlot(title="法向力") p1.setLabel('bottom', '时间', units='s') win.nextRow() p2 = win.addPlot(title="切向力 X") @@ -33,53 +49,53 @@ def update(plot_af, plot_tfx, plot_tfy, data, p1, p2, p3): # 深度图 def draw_magnitude_map(vfield, threshold=3.0, colormap=cv2.COLORMAP_JET) -> np.ndarray: """ - Draw color map of pointwise magnitude of velocity field. + 绘制流场幅值伪彩色图。 Args: - vfield: Velocity field to visualize (2D vector field) - threshold: Minimum magnitude to display (below threshold shows as zero/blue) - colormap: OpenCV colormap to use (default: COLORMAP_JET) + vfield: 需要可视化的二维矢量场 + threshold: 最小显示阈值；低于阈值的区域显示为低值颜色 + colormap: OpenCV 颜色映射类型（默认 COLORMAP_JET） Returns: - np.ndarray: 400x400 BGR image with colored magnitude map + np.ndarray: 400x400 的 BGR 伪彩图像 """ flow_field = vfield - # Compute magnitude at each point + # 计算每个像素点的幅值 magnitude = np.sqrt(flow_field[:, :, 0]**2 + flow_field[:, :, 1]**2) - # Smooth by downsampling and interpolating back + # 先降采样再插值回原尺寸，使显示更平滑 downsample_factor = 10 h, w = magnitude.shape magnitude_downsampled = magnitude[::downsample_factor, ::downsample_factor] magnitude_smoothed = cv2.resize(magnitude_downsampled, (w, h), interpolation=cv2.INTER_CUBIC) - # Apply threshold - set below-threshold values to zero + # 应用阈值，将低于阈值的区域置零 magnitude_thresholded = magnitude_smoothed.copy() magnitude_thresholded[magnitude_smoothed < threshold] = 0 - # Normalize to 0-255 range + # 归一化到 0-255 mag_min, mag_max = magnitude_thresholded.min(), magnitude_thresholded.max() if mag_max > mag_min: magnitude_norm = ((magnitude_thresholded - mag_min) / (mag_max - mag_min) * 255).astype(np.uint8) else: magnitude_norm = np.zeros_like(magnitude_thresholded, dtype=np.uint8) - # Apply colormap + # 应用伪彩色映射 magnitude_colored = cv2.applyColorMap(magnitude_norm, colormap) - # Resize to output size + # 缩放到固定输出尺寸 out_img = cv2.resize(magnitude_colored, (400, 400), interpolation=cv2.INTER_CUBIC) return out_img # 实时图表 -npoints = 100 # 图表数据长度 +npoints = 100 # 曲线缓冲区长度 pg.setConfigOption('background', 'w') pg.setConfigOption('foreground', 'k') -app = pg.mkQApp("Plotting Example") -win = pg.GraphicsLayoutWidget(show=True, title="Basic plotting examples") -win.setWindowTitle('pyqtgraph example: Plotting') +app = pg.mkQApp("Orisys Plot Example") +win = pg.GraphicsLayoutWidget(show=True, title="Orisys 实时力曲线") +win.setWindowTitle('Orisys 实时力曲线') p1, p2, p3, plot_af, plot_tfx, plot_tfy = create_plot() data = np.zeros([npoints,2,3]) start_time = None # Will be set when sensor starts @@ -91,17 +107,17 @@ for i in range(3): def main(): global start_time parser = argparse.ArgumentParser() - parser.add_argument("--video","-v", type=str, default="0", help="Camera ID / Video file path") - parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="Configuration file path") - parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="Calibration file path") - parser.add_argument("--verbose","-verbose", type=bool, default=True, help="Print detailed information") + parser.add_argument("--video","-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径") + parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="配置文件名称或路径") + parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="标定文件路径") + parser.add_argument("--verbose","-verbose", type=bool, default=True, help="是否输出详细日志") args = parser.parse_args() - # 步骤1：创建传感器对象 (必要步骤)， - # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） - # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 + # 步骤 1：创建传感器对象（必需） + # config_name 可以是内置配置名称，也可以是自定义配置文件路径 + # cal_path 为标定文件路径；每台设备建议使用独立的标定文件 - # 方式1: 从视频文件读取 + # 输入既可以是摄像头编号，也可以是视频文件路径 try: input = int(args.video) is_camera = True @@ -109,23 +125,23 @@ def main(): except: is_camera = False sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - # 打开视频 - # sensor1 = Sensor("./s1.mp4",isstitch=True,cuda = True,verbose = True) - # 方式1: 从视频文件读取 + # 下面保留了历史示例写法，便于参考 # sensor1 = orisys.Sensor("./s1.mp4", config_name="./config/ddjx01.json", cal_path= "./config/ddjx01.npy", cuda=False, verbose=True) start_time = time.time() is_save = False frame_count = 0 writer = None + save_path = None + print("\n按键说明：'q' 退出程序，'s' 开始录像，'e' 结束录像。") while True: - sensor.get_img() # 提取并拼接复眼图像 + sensor.get_img() # 获取并拼接复眼图像 if is_save and writer is not None and sensor.frame is not None: frame_count += 1 save_img = sensor.frame - # Convert grayscale (1 channel) to BGR (3 channels) if needed + # 若为灰度图，则先转换为 BGR 再写入视频 if len(save_img.shape) == 2 or (len(save_img.shape) == 3 and save_img.shape[2] == 1): if len(save_img.shape) == 3: save_img = save_img.squeeze(2) @@ -133,44 +149,43 @@ def main(): writer.write(save_img) - sensor.compute_deformation(check_motion=False) # 计算形变和分解 - # 提取信息： - # 向量场: VRAW, VNORMAL, VSHEAR: 原始，法向，切向 - # 帧率，时间戳: FPS, TIMESTAMP - # 变形量总和: FNORMAL , FSHEARX, FSHEARY : 法向，切向（X)，切向（Y) + sensor.compute_deformation(check_motion=False) # 计算形变与分解结果 + # 读取信息： + # 向量场：VRAW、VNORMAL、VSHEAR -> 原始 / 法向 / 切向 + # 标量：FNORMAL、FSHEARX、FSHEARY -> 法向力 / X 向切向力 / Y 向切向力 fps, fn, fx, fy, flow, vnormal, vshear = sensor.read_info(sensor.info.FPS, sensor.info.FNORMAL, sensor.info.FSHEARX, sensor.info.FSHEARY, sensor.info.VRAW, sensor.info.VNORMAL,sensor.info.VSHEAR) - print(fps,fn,fx,fy) - # 显示向量场 + print(f"FPS={fps:.2f}, 法向力={fn:.4f}, 切向力X={fx:.4f}, 切向力Y={fy:.4f}") + # 显示形变矢量场 arrows = orisys.util.draw_arrows(sensor.img, flow, threshold=5, grid_spacing=10, arrow_scale=1.0) - cv2.imshow("Deformation", arrows) - # 显示深度图 + cv2.imshow("形变矢量场", arrows) + # 显示流场幅值图 mag = draw_magnitude_map(flow) - cv2.imshow("Magnitude", mag) + cv2.imshow("流场幅值图", mag) - # 显示图表 - 更新时间轴 + # 更新实时曲线 current_time = time.time() - start_time for i in range(3): - data[:-1, 0, i] = data[1:, 0, i] # shift time data left - data[:-1, 1, i] = data[1:, 1, i] # shift force data left - data[-1, 0, :] = current_time # set current time for all plots + data[:-1, 0, i] = data[1:, 0, i] # 时间轴左移 + data[:-1, 1, i] = data[1:, 1, i] # 力数据左移 + data[-1, 0, :] = current_time # 写入当前时间 data[-1, 1, 0] = fn data[-1, 1, 1] = fx data[-1, 1, 2] = fy - update(plot_af, plot_tfx, plot_tfy, data, p1, p2, p3) # Update plot + update(plot_af, plot_tfx, plot_tfy, data, p1, p2, p3) # 更新曲线 key = cv2.waitKey(1) if key & 0xFF == ord("q"): - print("\n 停止程序") + print("\n正在退出示例程序...") break if key == ord("s"): save_path = f"./data/pulse_test.mp4" - # Ensure directory exists + # 确保输出目录存在 os.makedirs(os.path.dirname(save_path), exist_ok=True) - # Get frame dimensions from the actual frame + # 从当前帧获取视频尺寸 save_img = sensor.frame if save_img is not None: - # Convert grayscale (1 channel) to BGR (3 channels) if needed for dimension check + # 如为灰度图，先转换为 BGR 以获取视频尺寸 temp_img = save_img.copy() if len(temp_img.shape) == 2 or (len(temp_img.shape) == 3 and temp_img.shape[2] == 1): if len(temp_img.shape) == 3: @@ -181,24 +196,24 @@ def main(): writer = cv2.VideoWriter(save_path, cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width, frame_height)) if writer.isOpened(): is_save = True - print(f"Start writing to {save_path}") + print(f"开始录像：{save_path}") else: - print(f"Failed to open video writer for {save_path}") + print(f"无法创建录像文件：{save_path}") writer = None else: - print("No frame available to determine video dimensions") + print("当前没有可用帧，无法确定录像尺寸。") if key == ord("e"): is_save = False if writer is not None: writer.release() writer = None if save_path is not None: - print(f"Stop writing to {save_path}, total frames: {frame_count}") + print(f"录像结束：{save_path}，总帧数：{frame_count}") else: - print(f"Stop writing, total frames: {frame_count}") + print(f"录像结束，总帧数：{frame_count}") frame_count = 0 - sensor.disconnect() # 断开传感器连接 + sensor.disconnect() # 断开传感器连接并释放资源 diff --git a/src/orisys/configs/ddjx01.json b/src/orisys/configs/ddjx01.json index dc1e174..ac8eba6 100644 --- a/src/orisys/configs/ddjx01.json +++ b/src/orisys/configs/ddjx01.json @@ -1,6 +1,6 @@ { "name": "prototype", - "version": "1.0", + "version": "0.3.1", "camera": { "resolution_raw": [1920, 1080], "fps": 30, @@ -11,14 +11,14 @@ }, "stitcher": { "min_area_frac": 0.01, - "max_area_frac": 0.04, - "max_aspect_ratio": 100, + "max_area_frac": 0.05, + "max_aspect_ratio": 50, "index": [1,2,3,6,7,8,11,12,13], "margin": 10, "nominal_ov_pos": 100, "n": [3,3], "row_tol": 0.6, - "box_size": [220, 220] + "box_size": [250, 250] }, "tracker": { "adaptive": true, @@ -28,15 +28,16 @@ "algorithm": "dis", "dis_preset": "medium", "dis_mode": 3, - "algorithm_list": ["raft", "dis"] + "algorithm_list": ["raft", "dis"], + "hhd_downsample": 3 }, "contact": { "min_contour_area": 3000, - "flow_threshold":1, - "div_threshold":0.02, + "flow_threshold":5, + "div_threshold":0.05, "expansion_threshold": 0.01, "kernel_size": [5,5], - "denoise_sigma": 15 + "denoise_sigma": 12 } } diff --git a/src/orisys/configs/prototype.json b/src/orisys/configs/prototype.json deleted file mode 100644 index b4aecb7..0000000 --- a/src/orisys/configs/prototype.json +++ /dev/null @@ -1,39 +0,0 @@ -{ - "name": "prototype", - "version": "1.0", - "camera": { - "resolution_raw": [1920, 1080], - "fps": 30, - "units": 15, - "units_used": 9, - "resolution_stitch": [400, 400] - }, - "stitcher": { - "min_area_frac": 0.01, - "max_area_frac": 0.04, - "max_aspect_ratio": 100, - "index": [1,2,3,6,7,8,11,12,13], - "margin": 10, - "nominal_ov_pos": 100, - "n": [3,3], - "row_tol": 0.6 - }, - "tracker": { - "adaptive": true, - "lower_threshold": 12.0, - "upper_threshold": 18.0, - "loop_closure_interval": 1, - "dis_preset": "medium", - "finest_scale": 3 - }, - "contact": { - "min_contour_area": 3000, - "flow_threshold":1, - "div_threshold":0.02, - "expansion_threshold": 0.01, - "kernel_size": [5,5], - "denoise_sigma": 15 - } -} - - diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 04fc6d6..577d223 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -282,7 +282,7 @@ class Sensor: self.compute_hhd() return True return False - + else: return None # return True