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

clean up folder

Browse Source
This commit is contained in:
ykcheungab
2026-05-10 17:41:39 +08:00
parent b00b640261
commit d38ce82059
54 changed files with 556 additions and 2597 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/README.md
+50
View File
@@ -0,0 +1,50 @@
# Examples
This folder contains user-facing example scripts for the Orisys SDK.
## Files
- `basic_pipeline.py`: single-sensor basic processing and visualization flow
- `two_sensor_pipeline.py`: dual-sensor multiprocessing example
- `plot_data.py`: real-time plotting example with PyQtGraph
## Typical usage
Run examples from the repository root so relative paths such as `./config/ddjx01.json`
resolve correctly.
### Basic pipeline
```bash
python examples/basic_pipeline.py -v 0 -c ./config/ddjx01.json
```
### Two-sensor pipeline
```bash
python examples/two_sensor_pipeline.py
```
### Real-time plotting
```bash
python examples/plot_data.py -v 0 -c ./config/ddjx01.json
```
## Requirements
- Python environment with the SDK dependencies installed
- a supported camera or test video
- valid JSON config in `./config/`
Some examples also require:
- `pyqtgraph`
- `PyQt5`
- multiple cameras for multi-sensor demos
## Notes
- Most examples assume a tactile sensor camera is connected.
- Press `q` in the display window to quit where applicable.
- If your camera index is not `0`, change `-v` accordingly.
examples/basic_pipeline.py
+115
View File
@@ -0,0 +1,115 @@
"""
示例 1：单传感器基础流程
本示例演示 Orisys SDK 的标准处理流程：
1. 初始化传感器（摄像头或视频文件）
2. 获取图像并计算形变
3. 按需计算接触区域
4. 读取并可视化结果
SDK version: 0.3.1
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
import cv2
import orisys
import numpy as np
import argparse
def main():
parser = argparse.ArgumentParser()
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 为标定文件路径；每台设备建议使用独立的标定文件
# 输入既可以是摄像头编号，也可以是视频文件路径
try:
input = int(args.video)
is_camera = True
sensor = orisys.Sensor(input, config_name=args.config, verbose=args.verbose)
except:
is_camera = False
sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose)
print("\n按键说明：'q' 退出程序，'r' 重置追踪器。")
while True:
# 步骤 2：数据处理
# 步骤 2.1：获取并拼接图像（必需）
sensor.get_img()
# 步骤 2.2：计算形变场（必需）；返回值表示当前帧是否检测到接触/运动
is_contact_now = sensor.compute_deformation(check_motion=True, threshold=6)
# 步骤 2.3：仅在当前帧检测到接触/运动时计算接触区域
if is_contact_now:
sensor.compute_contact()
# 步骤 2.4：读取结果
fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info(
sensor.info.FPS,
sensor.info.VRAW,
sensor.info.VNORMAL,
sensor.info.IMG,
sensor.info.CONTOUR,
sensor.info.CENTROID,
sensor.info.DEPTH,
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY
)
# 步骤 3：可视化显示
# 步骤 3.1：显示形变矢量场
arrows = orisys.util.draw_arrows(
img,
flow,
threshold=2,
grid_spacing=20,
arrow_scale=1.0
)
cv2.imshow("形变矢量场", arrows)
# 步骤 3.2：显示接触区域和质心
image_with_foe = orisys.util.draw_contact(img, contour, centroid)
cv2.imshow("接触区域与质心", image_with_foe)
# 步骤 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)
# 应用伪彩色映射
cv2.imshow("深度图", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET))
# 步骤 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：键盘控制
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
print("\n正在退出示例程序...")
break
if key == ord("r"):
# 当光流追踪出现漂移时，可手动重置追踪器
sensor.reset()
print("追踪器已重置。")
# =========================================================================
# 4. 释放资源
# =========================================================================
sensor.disconnect() # 释放视频源及相关资源
if __name__ == '__main__':
main()
examples/plot_data.py
+221
View File
@@ -0,0 +1,221 @@
"""
示例 3：实时曲线与幅值图显示
本示例演示如何在读取传感器结果的同时：
1. 显示形变矢量场
2. 显示流场幅值热力图
3. 使用 PyQtGraph 实时绘制法向力与切向力曲线
4. 按键控制录像开始与结束
SDK version: 0.3.1
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
import cv2
import orisys
import numpy as np
import time
import argparse
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.setLabel('bottom', '时间', units='s')
win.nextRow()
p2 = win.addPlot(title="切向力 X")
p2.setLabel('bottom', '时间', units='s')
win.nextRow()
p3 = win.addPlot(title="切向力 Y")
p3.setLabel('bottom', '时间', units='s')
plot_af = p1.plot(pen='k')
plot_tfx = p2.plot(pen='k')
plot_tfy = p3.plot(pen='k')
return p1, p2, p3, plot_af, plot_tfx, plot_tfy
def update(plot_af, plot_tfx, plot_tfy, data, p1, p2, p3):
# global plot_af, plot_tfx, plot_tfy, data, p1, p2, p3
plot_af.setData(data[:,:,0])
p1.setYRange(min(data[:,1,0]),max(data[:,1,0]))
plot_tfx.setData(data[:,:,1])
p2.setYRange(min(data[:,1,1]),max(data[:,1,1]))
plot_tfy.setData(data[:,:,2])
p3.setYRange(min(data[:,1,2]),max(data[:,1,2]))
# 深度图
def draw_magnitude_map(vfield, threshold=3.0, colormap=cv2.COLORMAP_JET) -> np.ndarray:
"""
绘制流场幅值伪彩色图。
Args:
vfield: 需要可视化的二维矢量场
threshold: 最小显示阈值；低于阈值的区域显示为低值颜色
colormap: OpenCV 颜色映射类型（默认 COLORMAP_JET）
Returns:
np.ndarray: 400x400 的 BGR 伪彩图像
"""
flow_field = vfield
# 计算每个像素点的幅值
magnitude = np.sqrt(flow_field[:, :, 0]**2 + flow_field[:, :, 1]**2)
# 先降采样再插值回原尺寸，使显示更平滑
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)
# 应用阈值，将低于阈值的区域置零
magnitude_thresholded = magnitude_smoothed.copy()
magnitude_thresholded[magnitude_smoothed < threshold] = 0
# 归一化到 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)
# 应用伪彩色映射
magnitude_colored = cv2.applyColorMap(magnitude_norm, colormap)
# 缩放到固定输出尺寸
out_img = cv2.resize(magnitude_colored, (400, 400), interpolation=cv2.INTER_CUBIC)
return out_img
# 实时图表
npoints = 100 # 曲线缓冲区长度
pg.setConfigOption('background', 'w')
pg.setConfigOption('foreground', 'k')
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
for i in range(3):
data[:,0,i] = np.linspace(-npoints,0,npoints)
def main():
global start_time
parser = argparse.ArgumentParser()
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 为标定文件路径；每台设备建议使用独立的标定文件
# 输入既可以是摄像头编号，也可以是视频文件路径
try:
input = int(args.video)
is_camera = True
sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose)
except:
is_camera = False
sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose)
# 下面保留了历史示例写法，便于参考
# 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() # 获取并拼接复眼图像
if is_save and writer is not None and sensor.frame is not None:
frame_count += 1
save_img = sensor.frame
# 若为灰度图，则先转换为 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)
save_img = cv2.cvtColor(save_img, cv2.COLOR_GRAY2BGR)
writer.write(save_img)
sensor.compute_deformation(check_motion=True, threshold=0) # 计算形变与分解结果
# 读取信息：
# 向量场：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(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("形变矢量场", arrows)
# 显示流场幅值图
mag = draw_magnitude_map(flow)
cv2.imshow("流场幅值图", mag)
# 更新实时曲线
current_time = time.time() - start_time
for i in range(3):
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) # 更新曲线
key = cv2.waitKey(1)
if key & 0xFF == ord("q"):
print("\n正在退出示例程序...")
break
if key == ord("s"):
save_path = f"./data/pulse_test.mp4"
# 确保输出目录存在
os.makedirs(os.path.dirname(save_path), exist_ok=True)
# 从当前帧获取视频尺寸
save_img = sensor.frame
if save_img is not None:
# 如为灰度图，先转换为 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:
temp_img = temp_img.squeeze(2)
temp_img = cv2.cvtColor(temp_img, cv2.COLOR_GRAY2BGR)
frame_height, frame_width = temp_img.shape[:2]
writer = cv2.VideoWriter(save_path, cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width, frame_height))
if writer.isOpened():
is_save = True
print(f"开始录像：{save_path}")
else:
print(f"无法创建录像文件：{save_path}")
writer = None
else:
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"录像结束：{save_path}，总帧数：{frame_count}")
else:
print(f"录像结束，总帧数：{frame_count}")
frame_count = 0
sensor.disconnect() # 断开传感器连接并释放资源
if __name__ == '__main__':
main()
examples/two_sensor_pipeline.py
+281
View File
@@ -0,0 +1,281 @@
"""
示例 2：双传感器并行处理（多进程）
本示例演示如何使用 multiprocessing 同时运行两个传感器：
- 每个传感器运行在独立进程中
- 每个进程拥有独立的 CUDA 上下文
- 通过 Process 和 Queue 在主进程中汇总结果并显示
架构说明：
进程 1（传感器 0）：采集 → 光流 → HHD → 接触检测 → 队列
进程 2（传感器 1）：采集 → 光流 → HHD → 接触检测 → 队列
↓
主进程： 显示 ← ───────────────┘
适用场景：多指触觉传感器、多视角采集，以及需要 GPU 加速的多路处理。
注意事项：
- Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support()
- 每个传感器进程都会占用独立的 GPU 显存
SDK version: 0.3.1
"""
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
import argparse
def sensor_worker(sensor_id, vid_src, config_name, result_queue, stop_event, target_fps=None):
"""
传感器工作进程。
Args:
sensor_id: 传感器编号
vid_src: 视频源（文件路径或摄像头索引）
config_name: 配置名称
result_queue: 结果队列，用于向主进程发送数据
stop_event: 停止事件，用于通知工作进程退出
target_fps: 目标帧率；设置后将对处理循环限速，使多路传感器 FPS 更接近
"""
try:
# 在每个进程中创建独立的传感器实例
sensor = orisys.Sensor(
vid_src,
isstitch=True,
cuda=True,
verbose=False, # 减少子进程输出
config_name=config_name
)
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：获取图像
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 Exception:
# 如果队列已满，则丢弃当前帧，避免阻塞工作进程
pass
except Exception as e:
print(f"传感器 {sensor_id} 处理失败：{e}")
time.sleep(0.1)
# 步骤 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} 已停止。")
except Exception as e:
print(f"传感器 {sensor_id} 初始化失败：{e}")
result_queue.put({'sensor_id': sensor_id, 'error': str(e)})
def main():
# =========================================================================
# 1. 配置多路传感器
# =========================================================================
# 目标帧率：设为正数时，两路传感器会按该帧率限速；设为 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': args.video1,
'config_name': './config/ddjx01.json'
},
{
'sensor_id': 1,
'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. 启动工作进程
# =========================================================================
processes = []
for i, config in enumerate(sensor_configs):
p = Process(
target=sensor_worker,
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' 退出程序。")
# 保存每个传感器的最新一帧数据
latest_data = [None] * len(sensor_configs)
# 控制打印频率，减少 I/O 开销
print_counter = 0
print_interval = 30 # 每 30 帧打印一次
# 控制按键检查频率
key_check_counter = 0
key_check_interval = 5 # 每 5 帧检查一次按键
try:
# =====================================================================
# 4. 主循环：从队列获取结果并显示
# =====================================================================
while True:
# 从所有队列中快速获取最新数据，只保留最后一帧
for i, queue in enumerate(result_queues):
# 清空队列，仅保留最后一帧
data = None
# 读取所有可用数据，只保留最新结果
while True:
try:
data = queue.get_nowait()
except:
break
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'],
data['flow'],
threshold=5,
grid_spacing=10,
arrow_scale=1.0
)
cv2.imshow(f"传感器 {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))
# 降低打印频率，减少主循环中的 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_counter = 0
# 降低按键检查频率，减少额外开销
key_check_counter += 1
if key_check_counter >= key_check_interval:
key = cv2.waitKey(1) & 0xFF
if key == ord('q'):
print("\n正在退出示例程序...")
break
key_check_counter = 0
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()