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

update demo.py

Browse Source
This commit is contained in:
unknown
2026-01-02 14:27:31 +08:00
parent 7d2eed8261
commit 0d7ba7cd0b
5 changed files with 193 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/compound_eye.png
BIN
View File
Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 2.1 MiB

After

Width:  |  Height:  |  Size: 2.0 MiB

config/demo.json
+35
View File
@@ -0,0 +1,35 @@
{
"name": "demo",
"version": "1.0",
"camera": {
"resolution_raw": [1920, 1080],
"fps": 60,
"units": 15,
"units_used": 9,
"resolution_stitch": [400, 400]
},
"stitcher": {
"min_area_frac": 0.018,
"max_area_frac": 0.035,
"max_aspect_ratio": 100,
"index": [1, 2, 3, 6, 7, 8, 11, 12, 13],
"margin": 10,
"nominal_ov_pos": 100,
"n": [3, 3]
},
"tracker": {
"adaptive": true,
"lower_threshold": 12.0,
"upper_threshold": 18.0,
"loop_closure_interval": 1,
"dis_preset": "medium"
},
"contact": {
"min_contour_area": 3000,
"flow_threshold": 1,
"div_threshold": 0.02,
"expansion_threshold": 0.01,
"kernel_size": [5, 5],
"denoise_sigma": 15
}
}
config/exported_config.json
+1 -1
View File
@@ -1,5 +1,5 @@
{
"name": "prototype",
"name": "demo",
"version": "1.0",
"camera": {
"resolution_raw": [1920, 1080],
demo.py
+154 -145
View File
@@ -1,174 +1,183 @@
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
import cv2
import numpy as np
from orisys import Sensor
"""
示例 1: 基础用法 - 单传感器同步处理
def draw_arrows_fade(img, flow, depth, depth_scale=0.5, grid_spacing=20, arrow_scale=1.0, threshold=5.0):
本示例演示 orisys 的基本使用流程：
1. 初始化传感器（视频文件或摄像头）
2. 循环获取图像并计算形变和触碰点
3. 读取各类传感器数据
4. 可视化显示
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
import cv2
import orisys
import numpy as np
from scipy.interpolate import griddata
def draw_arrows(image, optical_flow,fps=True, threshold=5, grid_spacing=40, arrow_scale=1.0):
"""
Draw optical flow arrows on image with transparency based on magnitude.
Draw colored optical flow arrows on sensor image based on flow magnitude.
Args:
img: Input image (BGR or grayscale)
flow: Optical flow field with shape (flow_height, flow_width, 2)
threshold: Minimum flow magnitude to display arrow (default: 10.0)
grid_spacing: Spacing between arrows in pixels (default: 20)
arrow_scale: Scale factor for arrow length (default: 1.0)
threshold: Minimum flow magnitude to display arrow (default: 0.0)
arrow_scale: Scale factor for arrow length (default: 3.0)
Returns:
Image with arrows drawn, blended with original image
np.ndarray: 400x400 BGR image with colored arrows (JET colormap)
"""
depth_img = cv2.applyColorMap(((depth+1)*depth_scale* 255.0).astype('uint8'), cv2.COLORMAP_HOT)
# Make a copy of the image
img_copy = depth_img.copy()
# Create working copies to avoid modifying originals
# Convert grayscale to BGR if needed
if len(img_copy.shape) == 2 or img_copy.shape[2] == 1:
img_copy = cv2.cvtColor(img_copy, cv2.COLOR_GRAY2BGR)
# Get image and flow dimensions
img_height, img_width = img_copy.shape[:2]
flow_height, flow_width = flow.shape[:2]
# Convert grayscale to BGR to enable colored arrow drawing
if len(image.shape) == 2 or image.shape[2] == 1:
image = cv2.cvtColor(image, cv2.COLOR_GRAY2BGR)
# Calculate scaling factors between image and flow dimensions
scale_x = img_width / flow_width
scale_y = img_height / flow_height
img_height, img_width = image.shape[:2]
flow_height, flow_width = optical_flow.shape[:2]
scale_x, scale_y = img_width / flow_width, img_height / flow_height
# Create grid positions in flow coordinates
grid_y, grid_x = np.mgrid[grid_spacing//2:flow_height:grid_spacing,
grid_spacing//2:flow_width:grid_spacing]
# Create sampling grid for flow field (used for interpolation source points)
sample_step = 19 # Spacing between sample points
flow_start = np.stack(np.meshgrid(range(0, flow_width, sample_step), range(0, flow_height, sample_step)), 2)
flow_end = (optical_flow[flow_start[:, :, 1], flow_start[:, :, 0], :] * 2.0 + flow_start).astype(np.int32)
# Flatten grid positions
grid_y_flat = grid_y.flatten()
grid_x_flat = grid_x.flatten()
# Create regular grid for arrow placement in image coordinates
grid_x, grid_y = np.mgrid[grid_spacing//2:flow_height:grid_spacing, grid_spacing//2:flow_width:grid_spacing]
x_axis = (grid_y.reshape(-1) * scale_x).astype(np.int64) # Scale to image width
y_axis = (grid_x.reshape(-1) * scale_y).astype(np.int64) # Scale to image height
# Get flow vectors at grid positions
flow_vectors = flow[grid_y_flat, grid_x_flat, :] # Shape: (n_points, 2)
# Interpolate flow vectors from sample grid to regular arrow grid using cubic interpolation
grid_vector = griddata(
flow_start.reshape(-1, 2),
(flow_end - flow_start).reshape(-1, 2), # Flow from start to end (original to new position)
(grid_y, grid_x), method='cubic', fill_value=0
)
grid_vector[np.isnan(grid_vector)] = 0 # Replace NaN with zero for stability
grid_flat_scaled = grid_vector.reshape(-1, 2) * np.array([scale_x, scale_y])
# Calculate flow magnitudes
magnitudes = np.linalg.norm(flow_vectors, axis=1)
# Compute arrow endpoints and ensure they stay within image bounds
end_x = (grid_flat_scaled[:, 0] / arrow_scale + x_axis).astype(np.int64)
end_y = (grid_flat_scaled[:, 1] / arrow_scale + y_axis).astype(np.int64)
valid_mask = (y_axis >= 0) & (y_axis < img_height) & (x_axis >= 0) & (x_axis < img_width)
valid_mask &= (end_x >= 0) & (end_x < img_width) & (end_y >= 0) & (end_y < img_height)
# Filter arrows by threshold: only show arrows above threshold
valid_mask = magnitudes >= threshold
# Normalize magnitudes to 0-1 range for alpha blending
# Zero magnitude = transparent, higher magnitude = more opaque white
max_magnitude = magnitudes.max() if magnitudes.max() > 0 else 1.0
alpha_values = magnitudes / max_magnitude
# Scale grid positions to image coordinates
img_grid_x = (grid_x_flat * scale_x).astype(np.int32)
img_grid_y = (grid_y_flat * scale_y).astype(np.int32)
# Scale flow vectors to image coordinates
scaled_flow_x = flow_vectors[:, 0] * scale_x / arrow_scale
scaled_flow_y = flow_vectors[:, 1] * scale_y / arrow_scale
# Calculate arrow endpoints
end_x = (img_grid_x + scaled_flow_x).astype(np.int32)
end_y = (img_grid_y + scaled_flow_y).astype(np.int32)
# Create overlay image with white background for arrows
overlay = np.zeros_like(img_copy)
# Create weight map for blending based on arrow presence and magnitude
weight_map = np.zeros((img_height, img_width), dtype=np.float32)
# Draw arrows on overlay with white color and build weight map
for i in range(len(img_grid_x)):
# Skip if magnitude is below threshold
if not valid_mask[i]:
continue
# Check bounds
if (0 <= img_grid_x[i] < img_width and 0 <= img_grid_y[i] < img_height and
0 <= end_x[i] < img_width and 0 <= end_y[i] < img_height):
# Get alpha value (0 = transparent, 1 = opaque white)
alpha = alpha_values[i]
# Skip if magnitude is zero (transparent)
if alpha <= 0:
continue
# Calculate arrow thickness based on magnitude
thickness = max(1, int(1 + 2 * alpha))
# White color for arrows
color = (255, 255, 255)
# Draw arrow on overlay
cv2.arrowedLine(overlay,
(img_grid_x[i], img_grid_y[i]),
(end_x[i], end_y[i]),
color,
thickness=thickness,
tipLength=0.3)
# Create a line mask for this arrow in the weight map
line_mask = np.zeros((img_height, img_width), dtype=np.float32)
cv2.line(line_mask,
(img_grid_x[i], img_grid_y[i]),
(end_x[i], end_y[i]),
1.0,
thickness=thickness)
# Add arrow tip approximation
cv2.circle(line_mask, (end_x[i], end_y[i]), thickness, 1.0, -1)
# Weight by alpha (magnitude) - use maximum for overlapping arrows
weight_map = np.maximum(weight_map, line_mask * alpha)
# Expand weight map to 3 channels
weight_map_3ch = np.stack([weight_map] * 3, axis=2)
# Blend: result = original * (1 - weight) + overlay * weight
result = (img_copy.astype(np.float32) * (1 - weight_map_3ch) +
overlay.astype(np.float32) * weight_map_3ch)
return result.astype(np.uint8)
# Filter arrows by magnitude threshold and compute colors using JET colormap
magnitudes = np.linalg.norm(grid_flat_scaled, axis=1)
# Only show arrows where image is non-zero
# Check if image pixels at arrow positions are non-zero (sum across channels > 0)
# Ensure indices are within bounds before indexing
safe_y = np.clip(y_axis, 0, img_height - 1)
safe_x = np.clip(x_axis, 0, img_width - 1)
# Get pixel values at arrow positions and check if non-zero
pixel_values = image[safe_y, safe_x]
# When indexing 3D image with 1D arrays, result is 2D (N, channels)
# When indexing 2D image with 1D arrays, result is 1D (N,)
if pixel_values.ndim == 2: # BGR image (shape: N x 3)
pixel_sums = np.sum(pixel_values, axis=1)
else: # Grayscale image (shape: N,)
pixel_sums = pixel_values
valid_mask &= (pixel_sums > 0)
valid_indices = np.where(valid_mask)[0]
mag_range = magnitudes.max() - magnitudes.min()
# Normalize magnitudes to 0-255 range for colormap application, but set below-threshold values to 0 (blue)
norm_magnitudes = ((magnitudes - magnitudes.min()) / mag_range * 255.0).astype(np.uint8) if mag_range > 0 else np.zeros_like(magnitudes, dtype=np.uint8)
norm_magnitudes[magnitudes < threshold] = 0 # Keep lowest color (blue) for below-threshold magnitudes
colors_array = cv2.applyColorMap(norm_magnitudes.reshape(-1, 1), cv2.COLORMAP_BONE) # JET: blue (low) -> red (high)
# Draw arrows at each valid grid point with magnitude-based coloring
for i in valid_indices:
color = tuple(map(int, colors_array[i, 0]))
# Scale arrow length and thickness based on threshold
if magnitudes[i] < threshold:
# Below threshold: draw as small dot (zero-length arrow with larger tip)
cv2.circle(image, (x_axis[i], y_axis[i]), radius=1, color=color, thickness=-1)
else:
# Above threshold: draw full arrow with thickness proportional to magnitude
# Arrow rooted at original position (x_axis, y_axis), head points to new position (end_x, end_y)
thickness = max(1, int(2 * magnitudes[i] / magnitudes.max())) if magnitudes.max() > 0 else 2
cv2.arrowedLine(image, (x_axis[i], y_axis[i]), (end_x[i], end_y[i]), color, thickness=thickness, tipLength=0.3)
return image
def main():
global start_time
# 打开视频
# sensor1 = Sensor("./s1.mp4",isstitch=True,cuda = True,verbose = True)
# 打开摄像头
sensor1 = Sensor(0,isstitch=True,cuda = True,verbose = True)
# 步骤1：创建传感器对象 (必要步骤)，config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径）
# 方式1: 从视频文件读取
# sensor = orisys.Sensor("./s1.mp4", isstitch=True, cuda=True, verbose=True,config_name="prototype")
# 方式2: 从摄像头读取（取消注释使用）
sensor = orisys.Sensor(0, cuda=True, verbose=True, config_name="./config/demo.json")
sensor.export_config(path="./config/exported_config.json") # 导出配置文件供查看和修改
print("\n按 'q' 退出程序，按 'r' 重置光流追踪器")
while True:
sensor1.get_img() # 提取并拼接复眼图像
sensor1.compute_deformation() # 计算形变和分解
# 提取信息：
# 向量场: VRAW, VNORMAL, VSHEAR: 原始，法向，切向
# 帧率，时间戳: FPS, TIMESTAMP
# 变形量总和: FNORMAL , FSHEARX, FSHEARY : 法向，切向（X)，切向（Y)
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)
arrows = draw_arrows_fade(sensor1.img, flow, depth)
print(fps,np.min(depth),np.mean(depth))
# 步骤2：信息提取
# 步骤 2.1: 获取并拼接图像 (必要步骤)
sensor.get_img()
# Crop edges (15 pixels from each side) - only if image is large enough
h, w = arrows.shape[:2]
if h > 30 and w > 30:
cv2.imshow("Deformation", arrows[15:-15, 15:-15, :])
# 步骤 2.2: 计算形变场 (必要步骤)
sensor.compute_deformation()
# 步骤 2.3: 计算触碰点
sensor.compute_contact()
#步骤 2.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
)
#步骤 3: 可视化显示
# 步骤 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:
cv2.imshow("Deformation", arrows)
div_normalized = np.zeros_like(div_abs, dtype=np.uint8)
# 应用颜色映射（使用JET：蓝色=低散度，红色=高散度）
img_depth = cv2.applyColorMap(div_normalized, cv2.COLORMAP_HOT)
arrows = draw_arrows(
img_depth,
flow,
threshold=1,
grid_spacing=10,
arrow_scale=0.5
)
cv2.imshow("形变场 Deformation", arrows)
# Crop raw image edges if large enough
h_raw, w_raw = sensor1.img.shape[:2]
if h_raw > 30 and w_raw > 30:
cv2.imshow("raw", sensor1.img[15:-15, 15:-15])
else:
cv2.imshow("raw", sensor1.img)
key = cv2.waitKey(1)
if key & 0xFF == ord("q"):
print("\n 停止程序")
#步骤 3.4: 打印资料
print(f"fps: {fps}")
#步骤 3.5: 键盘控制
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
print("\n退出程序...")
break
if key == ord("r"):
# 重置光流追踪器（当追踪漂移时使用）
sensor.reset()
print("光流追踪器已重置")
sensor1.disconnect() # 断开传感器连接
# =========================================================================
# 4. 清理资源
# =========================================================================
sensor.disconnect() # 释放视频源和相关资源
if __name__ == '__main__':
main()
main()
src/orisys/sensor.py
+3 -2
View File
@@ -132,7 +132,7 @@ class Sensor:
# Initialize components
self.__stitcher = Stitcher(isstitch=isstitch, config_stitcher=self.__config_stitcher, verbose=self.verbose)
self.__tracker = Tracker(adaptive=True, cuda=cuda, verbose=self.verbose)
self.__tracker = Tracker(adaptive=True, cuda=cuda, verbose=self.verbose, dis_preset=self._config_dis_preset)
self.__contact3d = Contact3D(config_contact=self.__config_contact, verbose=self.verbose)
self.__hhd = None
@@ -342,7 +342,7 @@ class Sensor:
def reset(self):
"""Reset optical flow tracker."""
self.tracker.reset()
self.__tracker.reset()
def disconnect(self):
"""Release resources and stop async threads."""
@@ -403,6 +403,7 @@ class Sensor:
self._config_fps = cam["fps"]
self._config_units = cam["units"]
self._config_resolution_stitch = cam["resolution_stitch"]
self._config_dis_preset = self._config["tracker"]["dis_preset"]
self.__config_stitcher = self._config["stitcher"]
self.__config_contact = self._config["contact"]