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

update hex test video

Browse Source
This commit is contained in:
bluepulsar
2026-03-29 20:10:15 +08:00
parent 97d7d55838
commit 7609791aff
6 changed files with 271 additions and 1678 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DIS_scale_test.py
-106
View File
@@ -1,106 +0,0 @@
"""
示例 1: 基础用法 - 单传感器同步处理
本示例演示 orisys 的基本使用流程：
1. 初始化传感器（视频文件或摄像头）
2. 循环获取图像并计算形变和触碰点
3. 读取各类传感器数据
4. 可视化显示
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
import cv2
import orisys
import numpy as np
import time
def main():
# 步骤1：创建传感器对象 (必要步骤)，config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径）
# 方式1: 从视频文件读取
sensor1 = orisys.Sensor("s1.mp4", config_name="./config/prototype_of1.json", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True)
sensor2 = orisys.Sensor("s1.mp4", config_name="./config/prototype_of2.json", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True)
# 方式2: 从摄像头读取（取消注释使用）
# sensor = orisys.Sensor(0, config_name="./config/lens_test.json", cal_path= "./config/cal.npy", isstitch=True, cuda=True, verbose=True)
print("\n按 'q' 退出程序，按 'r' 重置光流追踪器")
while True:
# 步骤2：信息提取
# 步骤 2.1: 获取并拼接图像 (必要步骤)
sensor1.get_img()
sensor2.get_img()
# if sensor.detect_motion(threshold=2):
# print("motion detected")
# 步骤 2.2: 计算形变场 (必要步骤)
sensor1.compute_deformation()
sensor2.compute_deformation()
#步骤 2.4: 读取信息
fps, img, flow = sensor1.read_info(
sensor1.info.FPS,
sensor1.info.IMG,
sensor1.info.VRAW,
)
fps2, img2, flow2 = sensor2.read_info(
sensor2.info.FPS,
sensor2.info.IMG,
sensor2.info.VRAW,
)
# flow_mag = np.linalg.norm(flow, axis=2)
# flow_mag2 = np.linalg.norm(flow2, axis=2)
# flow_diff = np.linalg.norm(flow - flow2, axis=2)
# print(f"flow_mag: {flow_mag.mean()}, {flow_mag.min()}, flow_mag2: {flow_mag2.mean()}, {flow_mag2.min()}, flow_diff: {flow_diff.mean()}, {flow_diff.min()}")
# flow_diff_norm = cv2.applyColorMap((flow_diff*255).astype(np.uint8), cv2.COLORMAP_JET)
# cv2.imshow("flow_diff", flow_diff_norm)
#步骤 3: 可视化显示
# 步骤 3.1: 绘制形变场
arrows = orisys.util.draw_arrows(
img,
flow,
threshold=5,
grid_spacing=10,
arrow_scale=1.0
)
arrows2 = orisys.util.draw_arrows(
img2,
flow2,
threshold=5,
grid_spacing=10,
arrow_scale=1.0
)
cv2.imshow("形变场 Deformation", arrows)
cv2.imshow("形变场 Deformation2", arrows2)
#步骤 3.4: 打印资料
print(f"fps: {fps}, fps2: {fps2}")
#步骤 3.5: 键盘控制
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
print("\n退出程序...")
break
if key == ord("r"):
# 重置光流追踪器（当追踪漂移时使用）
sensor1.reset()
sensor2.reset()
print("传感器已重置")
# =========================================================================
# 4. 清理资源
# =========================================================================
sensor1.disconnect() # 释放视频源和相关资源
sensor2.disconnect() # 释放视频源和相关资源
if __name__ == '__main__':
main()
camera_gui.py
-1147
View File
File diff suppressed because it is too large Load Diff
demo.py
-183
View File
@@ -1,183 +0,0 @@
"""
示例 1: 基础用法 - 单传感器同步处理
本示例演示 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 colored optical flow arrows on sensor image based on flow magnitude.
Args:
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: 3.0)
Returns:
np.ndarray: 400x400 BGR image with colored arrows (JET colormap)
"""
# Create working copies to avoid modifying originals
# 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
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 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)
# 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
# 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])
# 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 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():
# 步骤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:
# 步骤2：信息提取
# 步骤 2.1: 获取并拼接图像 (必要步骤)
sensor.get_img()
# 步骤 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:
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)
#步骤 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("光流追踪器已重置")
# =========================================================================
# 4. 清理资源
# =========================================================================
sensor.disconnect() # 释放视频源和相关资源
if __name__ == '__main__':
main()
src/orisys/Stitcher.py
+235 -216
View File
@@ -29,6 +29,8 @@ class Stitcher:
self.bound=self.nominal_ov_pos - self.margin
self.row_tol = config_stitcher["row_tol"]
self.box_size = config_stitcher["box_size"]
# stich related (BF-matching)
self._blend_coef_cache = {}
self._feature_stitch_cache = {
"ref_idx": None,
@@ -77,58 +79,58 @@ class Stitcher:
if feature_stitched is not None:
return feature_stitched
tile_list = self.tile_list
stitch_calibration = self.stitch_calibration
ncol = self.ncol
nrow = self.nrow
# tile_list = self.tile_list
# stitch_calibration = self.stitch_calibration
# ncol = self.ncol
# nrow = self.nrow
img_stitched_height = None
# img_stitched_height = None
for i in range(nrow):
# Start with first tile of the row
img_stitched_width = tile_list[i * ncol]
# for i in range(nrow):
# # Start with first tile of the row
# img_stitched_width = tile_list[i * ncol]
# Stitch horizontally within the row
for j in range(1, ncol):
img_1 = img_stitched_width
img_2 = tile_list[i * ncol + j]
# # Stitch horizontally within the row
# for j in range(1, ncol):
# img_1 = img_stitched_width
# img_2 = tile_list[i * ncol + j]
ov_pos = int(stitch_calibration[i * ncol + j - 1][1])
ov_pos, ov1, ov2 = self.__find_ov(img_1, img_2, ov_pos, dim=1)
# ov_pos = int(stitch_calibration[i * ncol + j - 1][1])
# ov_pos, ov1, ov2 = self.__find_ov(img_1, img_2, ov_pos, dim=1)
img_stitched_width = self.__stitch_optimized(
img_1, img_2, ov_pos, ov1, ov2, dim=1, blend=blend
)
# img_stitched_width = self.__stitch_optimized(
# img_1, img_2, ov_pos, ov1, ov2, dim=1, blend=blend
# )
# Stitch rows vertically
if i == 0:
img_stitched_height = img_stitched_width
else:
width_h = img_stitched_height.shape[1]
width_w = img_stitched_width.shape[1]
wdiff = width_h - width_w
# # Stitch rows vertically
# if i == 0:
# img_stitched_height = img_stitched_width
# else:
# width_h = img_stitched_height.shape[1]
# width_w = img_stitched_width.shape[1]
# wdiff = width_h - width_w
if wdiff != 0:
if wdiff > 0:
padding = (
((0, 0), (0, wdiff), (0, 0))
if len(img_stitched_width.shape) == 3
else ((0, 0), (0, wdiff))
)
img_stitched_width = np.pad(img_stitched_width, padding, mode='edge')
else:
img_stitched_width = img_stitched_width[:, :width_h]
# if wdiff != 0:
# if wdiff > 0:
# padding = (
# ((0, 0), (0, wdiff), (0, 0))
# if len(img_stitched_width.shape) == 3
# else ((0, 0), (0, wdiff))
# )
# img_stitched_width = np.pad(img_stitched_width, padding, mode='edge')
# else:
# img_stitched_width = img_stitched_width[:, :width_h]
ov_pos_h = int(stitch_calibration[(i + 1) * nrow - 1][1])
ov_pos_h, ov_h1, ov_h2 = self.__find_ov(
img_stitched_height, img_stitched_width, ov_pos_h, dim=0
)
# ov_pos_h = int(stitch_calibration[(i + 1) * nrow - 1][1])
# ov_pos_h, ov_h1, ov_h2 = self.__find_ov(
# img_stitched_height, img_stitched_width, ov_pos_h, dim=0
# )
img_stitched_height = self.__stitch_optimized(
img_stitched_height, img_stitched_width, ov_pos_h, ov_h1, ov_h2, dim=0, blend=blend
)
# img_stitched_height = self.__stitch_optimized(
# img_stitched_height, img_stitched_width, ov_pos_h, ov_h1, ov_h2, dim=0, blend=blend
# )
return img_stitched_height
# return img_stitched_height
def stitch_feature_optimized(self, blend=True, match_brightness=False, pad=5, print_every=30):
"""
@@ -140,10 +142,10 @@ class Stitcher:
if plan is None:
return None
ref_idx, t_ref, moving_indices, moving_tiles = plan
ref_idx, t_ref, moving_indices, moving_tiles, selected_positions = plan
self._feature_stitch_frame_idx += 1
frame_idx = self._feature_stitch_frame_idx
tile_shapes = tuple(tile.shape[:2] for tile in self.tile_list[: self.nrow * self.ncol])
tile_shapes = tuple(self.tile_list[pos].shape[:2] for pos in selected_positions)
cache = self._feature_stitch_cache
cache_changed = (
cache["ref_idx"] != ref_idx
@@ -157,6 +159,11 @@ class Stitcher:
cache["tile_shapes"] = tile_shapes
cache["transforms"] = None
cache["canvas"] = {}
if self.verbose:
print(
"Feature stitch tile selection: "
f"ref_idx={ref_idx}, moving_indices={list(moving_indices)}"
)
H_use_list = cache["transforms"]
if H_use_list is None:
@@ -187,46 +194,58 @@ class Stitcher:
blend=blend,
)
def __find_ov(self, img1, img2, ov_pos, dim=1):
# def __find_ov(self, img1, img2, ov_pos, dim=1):
_dim = len(img1.shape)
# _dim = len(img1.shape)
if dim == 1:
if _dim == 3:
# RGB case
ov_1 = img1[:, -ov_pos:, :]
ov_2 = img2[:, :ov_pos, :]
else:
# gray scale
ov_1 = img1[:, -ov_pos:]
ov_2 = img2[:, :ov_pos]
elif dim == 0:
if _dim == 3:
ov_1 = img1[-ov_pos:, :, :]
ov_2 = img2[:ov_pos, :, :]
else:
# gray scale
ov_1 = img1[-ov_pos:, :]
ov_2 = img2[:ov_pos, :]
# if dim == 1:
# if _dim == 3:
# # RGB case
# ov_1 = img1[:, -ov_pos:, :]
# ov_2 = img2[:, :ov_pos, :]
# else:
# # gray scale
# ov_1 = img1[:, -ov_pos:]
# ov_2 = img2[:, :ov_pos]
# elif dim == 0:
# if _dim == 3:
# ov_1 = img1[-ov_pos:, :, :]
# ov_2 = img2[:ov_pos, :, :]
# else:
# # gray scale
# ov_1 = img1[-ov_pos:, :]
# ov_2 = img2[:ov_pos, :]
return ov_pos, ov_1, ov_2
# return ov_pos, ov_1, ov_2
def _get_feature_stitch_plan(self):
if not self.isstitch:
return None
total_tiles = self.nrow * self.ncol
if total_tiles <= 1 or len(self.tile_list) < total_tiles:
if isinstance(self.index, int):
configured_indices = (self.index,)
elif isinstance(self.index, np.ndarray):
configured_indices = tuple(int(i) for i in self.index.tolist())
else:
configured_indices = tuple(int(i) for i in self.index)
total_tiles = len(configured_indices)
expected_tiles = self.nrow * self.ncol
if total_tiles <= 1 or total_tiles != expected_tiles or len(self.tile_list) < total_tiles:
return None
# Use the center tile as the reference frame for a stable translation-only canvas.
ref_idx = (self.nrow // 2) * self.ncol + (self.ncol // 2)
if ref_idx >= total_tiles:
# Use the configured grid order from self.index and keep the same reference tile
# location as the original stitch setup.
ref_pos = (self.nrow // 2) * self.ncol + (self.ncol // 2)
if ref_pos >= total_tiles:
return None
moving_indices = tuple(i for i in range(total_tiles) if i != ref_idx)
moving_tiles = [self.tile_list[i] for i in moving_indices]
return ref_idx, self.tile_list[ref_idx], moving_indices, moving_tiles
selected_positions = tuple(range(total_tiles))
ref_idx = configured_indices[ref_pos]
moving_positions = tuple(pos for pos in selected_positions if pos != ref_pos)
moving_indices = tuple(configured_indices[pos] for pos in moving_positions)
moving_tiles = [self.tile_list[pos] for pos in moving_positions]
return ref_idx, self.tile_list[ref_pos], moving_indices, moving_tiles, selected_positions
def _tile_to_bgr(self, img):
if img.ndim == 2:
@@ -701,173 +720,173 @@ class Stitcher:
return tile_list
def stitch_cal(self):
# def stitch_cal(self):
# init
tile_list = self.tile_list
img_stitched_height = None
stitch_calibration = []
# Flip col and row for porting
nrow, ncol = self.ncol, self.nrow
if not self.isstitch:
return self.tile_list[0], None
# # init
# tile_list = self.tile_list
# img_stitched_height = None
# stitch_calibration = []
# # Flip col and row for porting
# nrow, ncol = self.ncol, self.nrow
# if not self.isstitch:
# return self.tile_list[0], None
for i in range(ncol):
img_stitched_width = None
for j in range(nrow):
if j == 0:
img_j0 = tile_list[i*ncol]
img_stitched_width = img_j0
else:
img_1 = tile_list[i*ncol + j-1]
img_2 = tile_list[i*ncol + j]
# for i in range(ncol):
# img_stitched_width = None
# for j in range(nrow):
# if j == 0:
# img_j0 = tile_list[i*ncol]
# img_stitched_width = img_j0
# else:
# img_1 = tile_list[i*ncol + j-1]
# img_2 = tile_list[i*ncol + j]
ov_pos, ov1, ov2, c_corr = self.__get_ov(img_1.astype(np.float32), img_2.astype(np.float32), dim=1)
# ov_pos, ov1, ov2, c_corr = self.__get_ov(img_1.astype(np.float32), img_2.astype(np.float32), dim=1)
stitch_calibration.append(["hor {}".format(j-1),ov_pos])
# stitch_calibration.append(["hor {}".format(j-1),ov_pos])
# stitch images with stacking
img_stitched_width = self.__stitch_optimized(img_stitched_width, img_2, ov_pos, ov1, ov2, blend=True, dim=1)
# # stitch images with stacking
# img_stitched_width = self.__stitch_optimized(img_stitched_width, img_2, ov_pos, ov1, ov2, blend=True, dim=1)
if i == 0:
img_stitched_height = img_stitched_width
stitch_calibration.append(["ver {}".format(i),0])
pass
else:
width = img_stitched_height.shape[1]
width_new = img_stitched_width.shape[1]
wdiff = width - width_new
if wdiff > 0:
img_stitched_width = np.hstack((img_stitched_width, img_stitched_width[:, -wdiff:]))
elif wdiff < 0:
img_stitched_width = img_stitched_width[:, :wdiff]
# if i == 0:
# img_stitched_height = img_stitched_width
# stitch_calibration.append(["ver {}".format(i),0])
# pass
# else:
# width = img_stitched_height.shape[1]
# width_new = img_stitched_width.shape[1]
# wdiff = width - width_new
# if wdiff > 0:
# img_stitched_width = np.hstack((img_stitched_width, img_stitched_width[:, -wdiff:]))
# elif wdiff < 0:
# img_stitched_width = img_stitched_width[:, :wdiff]
ov_pos_h, ov_h1, ov_h2, c_corr = self.__get_ov(img_stitched_height, img_stitched_width, dim=0)
# ov_pos_h, ov_h1, ov_h2, c_corr = self.__get_ov(img_stitched_height, img_stitched_width, dim=0)
stitch_calibration.append(["ver {}".format(i),ov_pos_h])
# stitch_calibration.append(["ver {}".format(i),ov_pos_h])
img_stitched_height = self.__stitch_optimized(img_stitched_height, img_stitched_width, ov_pos_h, ov_h1, ov_h2, blend=True, dim=0)
# img_stitched_height = self.__stitch_optimized(img_stitched_height, img_stitched_width, ov_pos_h, ov_h1, ov_h2, blend=True, dim=0)
self.stitch_calibration = stitch_calibration
# self.stitch_calibration = stitch_calibration
return img_stitched_height, stitch_calibration
# return img_stitched_height, stitch_calibration
def __stitch_optimized(self, img1, img2, ov_pos, ov1, ov2, blend=True, dim=1):
_dim = len(img1.shape)
# def __stitch_optimized(self, img1, img2, ov_pos, ov1, ov2, blend=True, dim=1):
# _dim = len(img1.shape)
# Clamp overlap for safety
if dim == 1:
ov_pos = int(max(0, min(ov_pos, img1.shape[1], img2.shape[1])))
else:
ov_pos = int(max(0, min(ov_pos, img1.shape[0], img2.shape[0])))
# # Clamp overlap for safety
# if dim == 1:
# ov_pos = int(max(0, min(ov_pos, img1.shape[1], img2.shape[1])))
# else:
# ov_pos = int(max(0, min(ov_pos, img1.shape[0], img2.shape[0])))
# Fast blend path with cached coefficient tensors
if not blend or ov_pos <= 1:
ov = ov1
else:
blend_coef = self._get_blend_coef(ov_pos, dim, _dim)
ov = (
ov1.astype(np.float32, copy=False) * blend_coef
+ ov2.astype(np.float32, copy=False) * (1.0 - blend_coef)
).astype(np.uint8)
# # Fast blend path with cached coefficient tensors
# if not blend or ov_pos <= 1:
# ov = ov1
# else:
# blend_coef = self._get_blend_coef(ov_pos, dim, _dim)
# ov = (
# ov1.astype(np.float32, copy=False) * blend_coef
# + ov2.astype(np.float32, copy=False) * (1.0 - blend_coef)
# ).astype(np.uint8)
# Build stitched output with one target allocation and slice writes
if dim == 1:
left_w = img1.shape[1] - ov_pos
right_w = img2.shape[1] - ov_pos
out_w = left_w + ov_pos + right_w
if _dim == 3:
stitched_img = np.empty((img1.shape[0], out_w, img1.shape[2]), dtype=img1.dtype)
if left_w > 0:
stitched_img[:, :left_w, :] = img1[:, :left_w, :]
if ov_pos > 0:
stitched_img[:, left_w:left_w + ov_pos, :] = ov
if right_w > 0:
stitched_img[:, left_w + ov_pos:, :] = img2[:, ov_pos:, :]
else:
stitched_img = np.empty((img1.shape[0], out_w), dtype=img1.dtype)
if left_w > 0:
stitched_img[:, :left_w] = img1[:, :left_w]
if ov_pos > 0:
stitched_img[:, left_w:left_w + ov_pos] = ov
if right_w > 0:
stitched_img[:, left_w + ov_pos:] = img2[:, ov_pos:]
else:
top_h = img1.shape[0] - ov_pos
bottom_h = img2.shape[0] - ov_pos
out_h = top_h + ov_pos + bottom_h
if _dim == 3:
stitched_img = np.empty((out_h, img1.shape[1], img1.shape[2]), dtype=img1.dtype)
if top_h > 0:
stitched_img[:top_h, :, :] = img1[:top_h, :, :]
if ov_pos > 0:
stitched_img[top_h:top_h + ov_pos, :, :] = ov
if bottom_h > 0:
stitched_img[top_h + ov_pos:, :, :] = img2[ov_pos:, :, :]
else:
stitched_img = np.empty((out_h, img1.shape[1]), dtype=img1.dtype)
if top_h > 0:
stitched_img[:top_h, :] = img1[:top_h, :]
if ov_pos > 0:
stitched_img[top_h:top_h + ov_pos, :] = ov
if bottom_h > 0:
stitched_img[top_h + ov_pos:, :] = img2[ov_pos:, :]
# # Build stitched output with one target allocation and slice writes
# if dim == 1:
# left_w = img1.shape[1] - ov_pos
# right_w = img2.shape[1] - ov_pos
# out_w = left_w + ov_pos + right_w
# if _dim == 3:
# stitched_img = np.empty((img1.shape[0], out_w, img1.shape[2]), dtype=img1.dtype)
# if left_w > 0:
# stitched_img[:, :left_w, :] = img1[:, :left_w, :]
# if ov_pos > 0:
# stitched_img[:, left_w:left_w + ov_pos, :] = ov
# if right_w > 0:
# stitched_img[:, left_w + ov_pos:, :] = img2[:, ov_pos:, :]
# else:
# stitched_img = np.empty((img1.shape[0], out_w), dtype=img1.dtype)
# if left_w > 0:
# stitched_img[:, :left_w] = img1[:, :left_w]
# if ov_pos > 0:
# stitched_img[:, left_w:left_w + ov_pos] = ov
# if right_w > 0:
# stitched_img[:, left_w + ov_pos:] = img2[:, ov_pos:]
# else:
# top_h = img1.shape[0] - ov_pos
# bottom_h = img2.shape[0] - ov_pos
# out_h = top_h + ov_pos + bottom_h
# if _dim == 3:
# stitched_img = np.empty((out_h, img1.shape[1], img1.shape[2]), dtype=img1.dtype)
# if top_h > 0:
# stitched_img[:top_h, :, :] = img1[:top_h, :, :]
# if ov_pos > 0:
# stitched_img[top_h:top_h + ov_pos, :, :] = ov
# if bottom_h > 0:
# stitched_img[top_h + ov_pos:, :, :] = img2[ov_pos:, :, :]
# else:
# stitched_img = np.empty((out_h, img1.shape[1]), dtype=img1.dtype)
# if top_h > 0:
# stitched_img[:top_h, :] = img1[:top_h, :]
# if ov_pos > 0:
# stitched_img[top_h:top_h + ov_pos, :] = ov
# if bottom_h > 0:
# stitched_img[top_h + ov_pos:, :] = img2[ov_pos:, :]
return stitched_img
# return stitched_img
def __get_ov(self, img1, img2, dim=1):
'''
default dimension is along axis 1 (width of the images)
'''
# pass parameters
nominal_ov = self.nominal_ov_pos
bound_ov = self.bound
# def __get_ov(self, img1, img2, dim=1):
# '''
# default dimension is along axis 1 (width of the images)
# '''
# # pass parameters
# nominal_ov = self.nominal_ov_pos
# bound_ov = self.bound
cross_corr = []
h1, w1 = img1.shape[:2]
h2, w2 = img2.shape[:2]
# clamp tested overlap to valid range (1..min dimension)
max_possible = w1 if dim == 1 else h1
max_possible = min(max_possible, w2 if dim == 1 else h2)
lo = max(1, nominal_ov - bound_ov)
hi = min(max_possible, nominal_ov + bound_ov)
ov_space = range(lo, hi + 1)
# cross_corr = []
# h1, w1 = img1.shape[:2]
# h2, w2 = img2.shape[:2]
# # clamp tested overlap to valid range (1..min dimension)
# max_possible = w1 if dim == 1 else h1
# max_possible = min(max_possible, w2 if dim == 1 else h2)
# lo = max(1, nominal_ov - bound_ov)
# hi = min(max_possible, nominal_ov + bound_ov)
# ov_space = range(lo, hi + 1)
for ov in ov_space:
if dim == 1:
ov_area1 = img1[:, max(0, w1 - ov):, :]
ov_area2 = img2[:, :min(ov, w2), :]
elif dim == 0:
ov_area1 = img1[max(0, h1 - ov):, :, :]
ov_area2 = img2[:min(ov, h2), :, :]
# for ov in ov_space:
# if dim == 1:
# ov_area1 = img1[:, max(0, w1 - ov):, :]
# ov_area2 = img2[:, :min(ov, w2), :]
# elif dim == 0:
# ov_area1 = img1[max(0, h1 - ov):, :, :]
# ov_area2 = img2[:min(ov, h2), :, :]
# require exact shape match (detect_nonblack_regions guarantees uniform tile size)
if ov_area1.size == 0 or ov_area2.size == 0 or ov_area1.shape != ov_area2.shape:
cross_corr.append(-1.0)
continue
# # require exact shape match (detect_nonblack_regions guarantees uniform tile size)
# if ov_area1.size == 0 or ov_area2.size == 0 or ov_area1.shape != ov_area2.shape:
# cross_corr.append(-1.0)
# continue
# multi-channel cross correlation (average over channels)
corr = 0.0
for i in range(min(3, ov_area1.shape[2])):
a = ov_area1[:, :, i].astype(np.float32)
b = ov_area2[:, :, i].astype(np.float32)
denom = np.sqrt(np.sum(a * a) * np.sum(b * b)) + 1e-12
corr += float(np.sum(a * b) / denom)
corr = corr / float(min(3, ov_area1.shape[2]))
cross_corr.append(corr)
# # multi-channel cross correlation (average over channels)
# corr = 0.0
# for i in range(min(3, ov_area1.shape[2])):
# a = ov_area1[:, :, i].astype(np.float32)
# b = ov_area2[:, :, i].astype(np.float32)
# denom = np.sqrt(np.sum(a * a) * np.sum(b * b)) + 1e-12
# corr += float(np.sum(a * b) / denom)
# corr = corr / float(min(3, ov_area1.shape[2]))
# cross_corr.append(corr)
# pick best overlap (ignore invalid entries)
best_idx = int(np.argmax(cross_corr))
ov_pos = list(ov_space)[best_idx]
# # pick best overlap (ignore invalid entries)
# best_idx = int(np.argmax(cross_corr))
# ov_pos = list(ov_space)[best_idx]
# extract the overlap regions (assume shapes match)
if dim == 1:
ov_1 = img1[:, max(0, w1 - ov_pos):, :].copy()
ov_2 = img2[:, :min(ov_pos, w2), :].copy()
else:
ov_1 = img1[max(0, h1 - ov_pos):, :, :].copy()
ov_2 = img2[:min(ov_pos, h2), :, :].copy()
return ov_pos, ov_1, ov_2, cross_corr
# # extract the overlap regions (assume shapes match)
# if dim == 1:
# ov_1 = img1[:, max(0, w1 - ov_pos):, :].copy()
# ov_2 = img2[:, :min(ov_pos, w2), :].copy()
# else:
# ov_1 = img1[max(0, h1 - ov_pos):, :, :].copy()
# ov_2 = img2[:min(ov_pos, h2), :, :].copy()
# return ov_pos, ov_1, ov_2, cross_corr
# Helper functions for stitching
src/orisys/sensor.py
+31 -25
View File
@@ -108,7 +108,7 @@ class Sensor:
self.compute_contact_valid = False
# Image Data Variables (public)
self.stitch_calibration = []
# self.stitch_calibration = []
self.frame = self.img = None
self.initial_img = None
shape = (self.__config_cam["resolution_stitch"][1], self.__config_cam["resolution_stitch"][0])
@@ -402,8 +402,10 @@ class Sensor:
def reset(self):
"""Reset optical flow tracker."""
self.__tracker.reset()
if self.img is not None:
self.initial_img = self.img.copy()
def disconnect(self):
"""Release resources and stop async threads."""
self._capture_running = False
@@ -438,31 +440,31 @@ class Sensor:
def stitch_cal(self):
"""
Load or create stitch calibration.
# def stitch_cal(self):
# """
# Load or create stitch calibration.
Returns:
np.ndarray: Stitched calibration image
"""
if self.cal_path is None:
self.cal_path = "./config/cal.npy"
# Returns:
# np.ndarray: Stitched calibration image
# """
# if self.cal_path is None:
# self.cal_path = "./config/cal.npy"
if os.path.exists(self.cal_path):
if self.verbose:
print("Loading calibration from", self.cal_path)
self.stitch_calibration = np.load(self.cal_path, allow_pickle=True)
self.stitcher.stitch_calibration = self.stitch_calibration
return self.stitcher.stitch_rt_optimized()
else:
if self.verbose:
print("Creating new calibration...")
img, self.stitch_calibration = self.stitcher.stitch_cal()
os.makedirs("./config", exist_ok=True)
np.save(self.cal_path, self.stitch_calibration)
if self.verbose:
print("Calibration saved to", self.cal_path)
return img
# if os.path.exists(self.cal_path):
# if self.verbose:
# print("Loading calibration from", self.cal_path)
# self.stitch_calibration = np.load(self.cal_path, allow_pickle=True)
# self.stitcher.stitch_calibration = self.stitch_calibration
# return self.stitcher.stitch_rt_optimized()
# else:
# if self.verbose:
# print("Creating new calibration...")
# img, self.stitch_calibration = self.stitcher.stitch_cal()
# os.makedirs("./config", exist_ok=True)
# np.save(self.cal_path, self.stitch_calibration)
# if self.verbose:
# print("Calibration saved to", self.cal_path)
# return img
# =========================================================================
# Private - Initialization Helpers
@@ -575,7 +577,11 @@ class Sensor:
else:
self._find_stitch_boxes(frame)
self._extract_tiles(frame)
img = self.stitch_cal()
# Feature-based stitching no longer needs eager overlap calibration here.
# Keep stitch_cal() as a fallback so older overlap-based stitching still works.
img = self.stitcher.stitch_rt_optimized()
# if img is None:
# img = self.stitch_cal()
img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
self.resolution_stitch = img.shape
self.initial_img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_CUBIC)
src/orisys/tracker.py
+4
View File
@@ -60,6 +60,10 @@ class Tracker:
# Aggressive optimization for CPU
self.__tracking_inst.setFinestScale(config_tracker["dis_mode"]) # Skip finest pyramid level (2x faster)
# self.__tracking_inst.setUseSpatialPropagation(False)
# self.__tracking_inst.setUseMeanNormalization(False)
# self.__tracking_inst.setPatchSize(4) # 8
# self.__tracking_inst.setPatchStride(2) # 3
# else:
# # CUDA flow method with stream support