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

add blend control on stitch

Browse Source
This commit is contained in:
ykcheungab
2026-03-21 19:27:07 +08:00
parent 88c8b8fa01
commit 966fc74313
9 changed files with 189 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/cal.npy
BIN
View File
Binary file not shown.
config/dd01a_0.npy
BIN
View File
Binary file not shown.
config/dd01a_1.npy
BIN
View File
Binary file not shown.
config/dd01a_2.npy
BIN
View File
Binary file not shown.
config/ddjx01_480p.json
+43
View File
@@ -0,0 +1,43 @@
{
"name": "prototype",
"version": "1.0",
"camera": {
"resolution_raw": [854, 480],
"fps": 30,
"units": 15,
"units_used": 9,
"resolution_stitch": [123, 123],
"stitch_res_list": [365,263,123,63]
},
"stitcher": {
"min_area_frac": 0.01,
"max_area_frac": 0.06,
"max_aspect_ratio": 50,
"index": [1,2,3,6,7,8,11,12,13],
"margin": 4,
"nominal_ov_pos": 44,
"n": [3,3],
"row_tol": 0.6,
"box_size": [98, 98]
},
"tracker": {
"adaptive": true,
"lower_threshold": 12.0,
"upper_threshold": 18.0,
"loop_closure_interval": 1,
"algorithm": "dis",
"dis_preset": "medium",
"dis_mode": 2,
"algorithm_list": ["raft", "dis"]
},
"contact": {
"min_contour_area": 3000,
"flow_threshold":1,
"div_threshold":0.02,
"expansion_threshold": 0.01,
"kernel_size": [5,5],
"denoise_sigma": 6
}
}
config/prototype_of3_720p.json
+2 -2
View File
@@ -12,9 +12,9 @@
"stitcher": {
"min_area_frac": 0.01,
"max_area_frac": 0.04,
"max_aspect_ratio": 100,
"max_aspect_ratio": 25,
"index": [1,2,3,6,7,8,11,12,13],
"margin": 6,
"margin": 10,
"nominal_ov_pos": 67,
"n": [3,3],
"row_tol": 0.6,
example2_two_sensor_pipeline.py
+6 -6
View File
@@ -130,15 +130,15 @@ def main():
sensor_configs = [
{
'sensor_id': 0,
'vid_src': 4,
'config_name': './config/prototype_of3_480p.json',
'cal_path': './config/dd01a_0_480p.npy'
'vid_src': 0,
'config_name': './config/ddjx01.json',
'cal_path': './config/ddjx01.npy'
},
{
'sensor_id': 1,
'vid_src': 6, # 可以改为不同的视频源或摄像头索引
'config_name': './config/prototype_of3_480p.json',
'cal_path': './config/dd01a_0_480p.npy'
'vid_src': 1, # 可以改为不同的视频源或摄像头索引
'config_name': './config/ddjx01.json',
'cal_path': './config/ddjx01.npy'
},
]
src/orisys/Stitcher.py
+119 -93
View File
@@ -29,17 +29,30 @@ class Stitcher:
self.bound=self.nominal_ov_pos - self.margin
self.row_tol = config_stitcher["row_tol"]
self.box_size = config_stitcher["box_size"]
self._blend_coef_cache = {}
if self.verbose:
print(f"Stitcher initialized with config: {self.config_stitcher}")
def stitch_rt_optimized(self, change_sv=False, s_mean=28., v_mean=180.):
def _get_blend_coef(self, ov_pos, dim, ndim):
key = (ov_pos, dim, ndim)
coef = self._blend_coef_cache.get(key)
if coef is not None:
return coef
blend_coef = np.linspace(1.0, 0.0, ov_pos, dtype=np.float32)
if dim == 1:
blend_coef = blend_coef.reshape(1, ov_pos, 1) if ndim == 3 else blend_coef.reshape(1, ov_pos)
else:
blend_coef = blend_coef.reshape(ov_pos, 1, 1) if ndim == 3 else blend_coef.reshape(ov_pos, 1)
self._blend_coef_cache[key] = blend_coef
return blend_coef
def stitch_rt_optimized(self, change_sv=False, s_mean=28., v_mean=180., blend=True):
tile_list = self.tile_list
stitch_calibration = self.stitch_calibration
ncol = self.ncol
nrow = self.nrow
processed_tiles = tile_list
img_stitched_height = None
for i in range(nrow):
@@ -48,39 +61,42 @@ class Stitcher:
# Stitch horizontally within the row
for j in range(1, ncol):
img_1 = img_stitched_width # Already stitched part
img_2 = processed_tiles[i * ncol + j] # if change_sv else tile_list[i * ncol + j]
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)
# Use optimized stitch function
img_stitched_width = self.__stitch_optimized(img_1, img_2, ov_pos, ov1, ov2, dim=1, blend=True)
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:
# Handle width mismatch more efficiently
width_h = img_stitched_height.shape[1]
width_w = img_stitched_width.shape[1]
wdiff = width_h - width_w
if wdiff != 0:
# Use numpy pad instead of hstack for better performance
if wdiff > 0:
# Pad right side of the narrower image
padding = ((0, 0), (0, wdiff), (0, 0)) if len(img_stitched_width.shape) == 3 else ((0, 0), (0, wdiff))
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:
# Crop the wider image
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, 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=True
img_stitched_height, img_stitched_width, ov_pos_h, ov_h1, ov_h2, dim=0, blend=blend
)
return img_stitched_height
@@ -118,6 +134,16 @@ class Stitcher:
# return self.boxes
def _copy_into_tile_buffer(self, src_view, dst, flip_x=True, flip_y=True):
if flip_x and flip_y:
cv2.flip(src_view, -1, dst)
elif flip_x:
cv2.flip(src_view, 1, dst)
elif flip_y:
cv2.flip(src_view, 0, dst)
else:
np.copyto(dst, src_view)
def get_stitch_unit(self, img, flip_x=True, flip_y=True,index = np.arange(15)):
"""
Extract each tile given by boxes (x0,y0,x1,y1), default flip both X and Y due to inverting lens.
@@ -129,43 +155,35 @@ class Stitcher:
Returns: canvas image and list of flipped tiles as (x0,y0,flipped_tile)
"""
h, w = img.shape[:2]
if self.units is None:
flipped_units = []
for i, (x0, y0, x1, y1) in enumerate(self.boxes):
# clamp coordinates
x0 = max(0, int(x0)); y0 = max(0, int(y0))
x1 = min(w, int(x1)); y1 = min(h, int(y1))
if x1 <= x0 or y1 <= y0:
continue
tile = img[y0:y1, x0:x1].copy()
# flip X then Y in default
if flip_x:
tile = cv2.flip(tile, 1) # horizontal
if flip_y:
tile = cv2.flip(tile, 0) # vertical
flipped_units.append((x0, y0, tile))
self.units = flipped_units
active_index = index if self.units is None else self.index
if isinstance(active_index, int):
selected_index = {active_index}
else:
flipped_units = self.units
index = self.index
for i, (x0, y0, x1, y1) in enumerate(self.boxes):
if i not in index:
continue
# clamp coordinates
x0 = max(0, int(x0)); y0 = max(0, int(y0))
x1 = min(w, int(x1)); y1 = min(h, int(y1))
if x1 <= x0 or y1 <= y0:
continue
tile = img[y0:y1, x0:x1].copy()
# flip X then Y in default
if flip_x:
tile = cv2.flip(tile, 1) # horizontal
if flip_y:
tile = cv2.flip(tile, 0) # vertical
# Assign instead of append
selected_index = set(active_index)
flipped_units = [] if self.units is None else self.units
for i, (x0, y0, x1, y1) in enumerate(self.boxes):
if i not in selected_index:
continue
x0 = max(0, int(x0)); y0 = max(0, int(y0))
x1 = min(w, int(x1)); y1 = min(h, int(y1))
if x1 <= x0 or y1 <= y0:
continue
src_view = img[y0:y1, x0:x1]
if self.units is None:
tile = np.empty_like(src_view)
self._copy_into_tile_buffer(src_view, tile, flip_x=flip_x, flip_y=flip_y)
flipped_units.append((x0, y0, tile))
else:
_, _, tile = flipped_units[i]
if tile.shape != src_view.shape or tile.dtype != src_view.dtype:
tile = np.empty_like(src_view)
self._copy_into_tile_buffer(src_view, tile, flip_x=flip_x, flip_y=flip_y)
flipped_units[i] = (x0, y0, tile)
self.units = flipped_units
self.units = flipped_units
return True
@@ -239,55 +257,63 @@ class Stitcher:
def __stitch_optimized(self, img1, img2, ov_pos, ov1, ov2, blend=True, dim=1):
_dim = len(img1.shape)
# Early return if no blending needed
if not blend:
ov = ov1
elif ov_pos <= 1:
ov = ov1
else:
# Vectorized blending
blend_coef = np.linspace(1.0, 0.0, ov_pos, dtype=np.float32)
# Prepare blend_coef for broadcasting
if dim == 1: # Horizontal blending
if _dim == 3:
# Shape: (1, ov_pos, 1)
blend_coef = blend_coef.reshape(1, ov_pos, 1)
else:
# Shape: (1, ov_pos)
blend_coef = blend_coef.reshape(1, ov_pos)
else: # Vertical blending
if _dim == 3:
# Shape: (ov_pos, 1, 1)
blend_coef = blend_coef.reshape(ov_pos, 1, 1)
else:
# Shape: (ov_pos, 1)
blend_coef = blend_coef.reshape(ov_pos, 1)
# Compute blended overlap in one operation
ov = (ov1.astype(np.float32) * blend_coef +
ov2.astype(np.float32) * (1 - blend_coef)).astype(np.uint8)
# Avoid unnecessary .copy() calls when possible
# Clamp overlap for safety
if dim == 1:
if _dim == 3:
# Use np.take or slicing without copy when safe
left = img1[:, :img1.shape[1] - ov_pos, :] if img1.shape[1] > ov_pos else img1[:, :0, :]
right = img2[:, ov_pos:, :]
stitched_img = np.concatenate((left, ov, right), axis=1)
else:
left = img1[:, :img1.shape[1] - ov_pos] if img1.shape[1] > ov_pos else img1[:, :0]
right = img2[:, ov_pos:]
stitched_img = np.concatenate((left, ov, right), axis=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)
# 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:
top = img1[:img1.shape[0] - ov_pos, :, :] if img1.shape[0] > ov_pos else img1[:0, :, :]
bottom = img2[ov_pos:, :, :]
stitched_img = np.concatenate((top, ov, bottom), axis=0)
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:
top = img1[:img1.shape[0] - ov_pos, :] if img1.shape[0] > ov_pos else img1[:0, :]
bottom = img2[ov_pos:, :]
stitched_img = np.concatenate((top, ov, bottom), axis=0)
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
src/orisys/sensor.py
+18 -1
View File
@@ -134,6 +134,13 @@ class Sensor:
self._fps = 0
self._last_capture_time = 0.0
self._last_processing_time = 0.0
self._last_stage_times = {
"extract_tiles": 0.0,
"stitch": 0.0,
"resize": 0.0,
"mirror": 0.0,
"astype": 0.0,
}
# Initialize components
self.stitcher = Stitcher(isstitch=isstitch, config_stitcher=self.__config_stitcher, verbose=self.verbose)
@@ -202,20 +209,30 @@ class Sensor:
# Extract and stitch tiles
# self.stitcher.get_stitch_unit(frame)
self._extract_tiles(frame)
t_after_extract = time.perf_counter()
img = np.ascontiguousarray(self.stitcher.stitch_rt_optimized())
img = np.ascontiguousarray(self.stitcher.stitch_rt_optimized(blend=True))
t_after_stitch = time.perf_counter()
self.img_size_raw = img.shape
# Resize
img = cv2.resize(img, tuple(self._config_resolution_stitch), interpolation=cv2.INTER_LINEAR)
t_after_resize = time.perf_counter()
if self.mirror:
img = cv2.flip(img, 1)
t_after_mirror = time.perf_counter()
self.img = img.astype(np.uint8)
t_after_astype = time.perf_counter()
t_processing_end = time.perf_counter()
self._last_processing_time = t_processing_end - t_processing_start
self._last_stage_times["extract_tiles"] = t_after_extract - t_processing_start
self._last_stage_times["stitch"] = t_after_stitch - t_after_extract
self._last_stage_times["resize"] = t_after_resize - t_after_stitch
self._last_stage_times["mirror"] = t_after_mirror - t_after_resize
self._last_stage_times["astype"] = t_after_astype - t_after_mirror
# FPS calculation
self._fps = 1.0 / (time.perf_counter() - self.last_log_time)