import os os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" import cv2 import orisys import numpy as np import time import pyqtgraph as pg 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: """ 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) Returns: np.ndarray: 400x400 BGR image with colored magnitude map """ 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 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 # 图表数据长度 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') 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 # 打开视频 # sensor1 = Sensor("./s1.mp4",isstitch=True,cuda = True,verbose = True) # 方式1: 从视频文件读取 sensor1 = orisys.Sensor(0, config_name="./config/prototype_of3.json", cal_path= "./config/dd01a_98.npy", isstitch=True, cuda=False, verbose=True) start_time = time.time() is_save = False frame_count = 0 writer = None while True: sensor1.get_img() # 提取并拼接复眼图像 if is_save and writer is not None and sensor1.frame is not None: frame_count += 1 save_img = sensor1.frame # Convert grayscale (1 channel) to BGR (3 channels) if needed 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) sensor1.compute_deformation() # 计算形变和分解 # 提取信息： # 向量场: VRAW, VNORMAL, VSHEAR: 原始，法向，切向 # 帧率，时间戳: FPS, TIMESTAMP # 变形量总和: FNORMAL , FSHEARX, FSHEARY : 法向，切向（X)，切向（Y) fps, fn, fx, fy, flow = sensor1.read_info(sensor1.info.FPS, sensor1.info.FNORMAL, sensor1.info.FSHEARX, sensor1.info.FSHEARY, sensor1.info.VRAW) print(fps,fn,fx,fy) # 显示向量场 arrows = orisys.util.draw_arrows(sensor1.img, flow, threshold=5, grid_spacing=10, arrow_scale=1.0) cv2.imshow("Deformation", arrows) # 显示深度图 mag = draw_magnitude_map(flow) cv2.imshow("Magnitude", 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, 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 key = cv2.waitKey(1) if key & 0xFF == ord("q"): 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 = sensor1.frame if save_img is not None: # Convert grayscale (1 channel) to BGR (3 channels) if needed for dimension check 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"Start writing to {save_path}") else: print(f"Failed to open video writer for {save_path}") writer = None else: print("No frame available to determine video dimensions") 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}") else: print(f"Stop writing, total frames: {frame_count}") frame_count = 0 sensor1.disconnect() # 断开传感器连接 if __name__ == '__main__': main()