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

0.3.1 stable release

Browse Source
This commit is contained in:
ykcheungab
2026-04-17 09:58:30 +08:00
parent 7a49fd76e6
commit fdc4ffd25f
11 changed files with 208 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
example3_plot_data.py
+70 -55
View File
@@ -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() # 断开传感器连接并释放资源