From 97d7d55838142ba5f54bcfe2ed82975b01078b2e Mon Sep 17 00:00:00 2001 From: bluepulsar Date: Sat, 28 Mar 2026 18:20:20 +0800 Subject: [PATCH] update stitch algorithm and ver to 0.3.1 --- Readme.md | 2 +- build_nuitka.py | 4 +- doc/get_started.md | 10 +- setup.py | 2 +- src/orisys/Stitcher.py | 502 ++++++++++++++++++++++++++++++++++- src/orisys/__init__.py | 2 +- src/orisys/util.py | 3 +- test_stitch.py | 588 ++--------------------------------------- 8 files changed, 541 insertions(+), 572 deletions(-) diff --git a/Readme.md b/Readme.md index 07375c4..b731d13 100644 --- a/Readme.md +++ b/Readme.md @@ -25,7 +25,7 @@ pip install orisys-{version}-{python-version}-{python-version}-{os}_{machine}.wh ``` 如 ```bash -pip install orisys-0.3.0-cp310-cp310-win_amd64.whl +pip install orisys-0.3.1-cp310-cp310-win_amd64.whl ``` 3. **安装 CuPy（根据您的 CUDA 版本选择）：** diff --git a/build_nuitka.py b/build_nuitka.py index 323a6e6..51adb6d 100644 --- a/build_nuitka.py +++ b/build_nuitka.py @@ -21,11 +21,11 @@ def install_nuitka(): def get_version() -> str: init_file = Path("src/orisys/__init__.py") if not init_file.exists(): - return "0.3.0" + return "0.3.1" for line in init_file.read_text(encoding="utf-8").splitlines(): if line.startswith("__version__"): return line.split("=", 1)[1].strip().strip('"').strip("'") - return "0.3.0" + return "0.3.1" def find_compiled_extension(output_dir: Path) -> Path: diff --git a/doc/get_started.md b/doc/get_started.md index b580714..5ef11c0 100644 --- a/doc/get_started.md +++ b/doc/get_started.md @@ -1,5 +1,5 @@ 叠动科技软件使用手册 -版本： 0.3.0 +版本： 0.3.1 更新日期： 2026年3月6日 # 简介 @@ -57,9 +57,9 @@ orisys SDK 以 .whl 文件形式提供，您需要根据操作系统、Python ## 下载正确的 .whl 文件 从官方渠道获取 .whl 文件，文件名格式通常为： orisys-{版本}-{Python版本}-{Python版本}-{操作系统}_{硬件架构}.whl -例如：orisys-0.3.0-cp310-cp310-win_amd64.whl 表示： +例如：orisys-0.3.1-cp310-cp310-win_amd64.whl 表示： -版本：0.3.0 +版本：0.3.1 Python 版本：3.10（cp310） 操作系统：Windows（win） 硬件架构：64位（amd64） @@ -71,7 +71,7 @@ Python 版本：3.10（cp310） ``` bash # 请将文件名替换为您实际下载的文件名 -pip install orisys-0.3.0-cp310-cp310-win_amd64.whl +pip install orisys-0.3.1-cp310-cp310-win_amd64.whl ``` ## 安装 CuPy（可选） @@ -108,7 +108,7 @@ python import orisys print(orisys.__version__) ``` -如果输出版本号（例如 0.3.0），则说明安装成功。 +如果输出版本号（例如 0.3.1），则说明安装成功。 # 使用方法 ## 运行示例程序 diff --git a/setup.py b/setup.py index fa00e9e..6202818 100644 --- a/setup.py +++ b/setup.py @@ -9,7 +9,7 @@ if configs_dir.exists(): setup( name="orisys", - version="0.3.0", + version="0.3.1", description="Orisys - Optical Tactile Sensor SDK", author="Your Name", package_dir={"": "src"}, diff --git a/src/orisys/Stitcher.py b/src/orisys/Stitcher.py index 402e952..e506a1e 100644 --- a/src/orisys/Stitcher.py +++ b/src/orisys/Stitcher.py @@ -30,9 +30,29 @@ class Stitcher: self.row_tol = config_stitcher["row_tol"] self.box_size = config_stitcher["box_size"] self._blend_coef_cache = {} + self._feature_stitch_cache = { + "ref_idx": None, + "moving_indices": None, + "tile_shapes": None, + "transforms": None, + "canvas": {}, + } + self._feature_stitch_frame_idx = 0 + self._feature_stitch_print_every = 30 + self._feature_stitch_pad = 5 + self._feature_stitch_match_brightness = False if self.verbose: print(f"Stitcher initialized with config: {self.config_stitcher}") + + def configure_feature_stitch(self, *, match_brightness=None, pad=None, print_every=None): + """Update runtime options for feature-based stitching.""" + if match_brightness is not None: + self._feature_stitch_match_brightness = bool(match_brightness) + if pad is not None: + self._feature_stitch_pad = max(0, int(pad)) + if print_every is not None: + self._feature_stitch_print_every = max(1, int(print_every)) def _get_blend_coef(self, ov_pos, dim, ndim): key = (ov_pos, dim, ndim) @@ -48,6 +68,15 @@ class Stitcher: return blend_coef def stitch_rt_optimized(self, change_sv=False, s_mean=28., v_mean=180., blend=True): + feature_stitched = self.stitch_feature_optimized( + blend=blend, + match_brightness=self._feature_stitch_match_brightness, + pad=self._feature_stitch_pad, + print_every=self._feature_stitch_print_every, + ) + if feature_stitched is not None: + return feature_stitched + tile_list = self.tile_list stitch_calibration = self.stitch_calibration ncol = self.ncol @@ -101,6 +130,63 @@ class Stitcher: return img_stitched_height + def stitch_feature_optimized(self, blend=True, match_brightness=False, pad=5, print_every=30): + """ + Stitch the current tile grid by estimating tile-to-reference transforms once and + reusing them across frames. Falls back to overlap stitching when feature matching + cannot build a valid layout. + """ + plan = self._get_feature_stitch_plan() + if plan is None: + return None + + ref_idx, t_ref, moving_indices, moving_tiles = 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]) + cache = self._feature_stitch_cache + cache_changed = ( + cache["ref_idx"] != ref_idx + or cache["moving_indices"] != moving_indices + or cache["tile_shapes"] != tile_shapes + ) + + if cache_changed: + cache["ref_idx"] = ref_idx + cache["moving_indices"] = moving_indices + cache["tile_shapes"] = tile_shapes + cache["transforms"] = None + cache["canvas"] = {} + + H_use_list = cache["transforms"] + if H_use_list is None: + do_verbose_first = self.verbose and frame_idx % print_every == 0 + H_use_list = [] + for j, tile in enumerate(moving_tiles): + H_use_j, _ = self._estimate_bfmatcher_stitch_transform( + t_ref, + tile, + verbose=(do_verbose_first and j == 0), + ) + H_use_list.append(H_use_j) + cache["transforms"] = H_use_list + + if not any(H is not None for H in H_use_list): + return None + + return self._render_feature_tiles( + t_ref, + moving_tiles, + H_use_list, + pad=pad, + frame_idx=frame_idx, + print_every=print_every, + verbose=self.verbose, + match_brightness=match_brightness, + canvas_cache=cache["canvas"], + blend=blend, + ) + def __find_ov(self, img1, img2, ov_pos, dim=1): _dim = len(img1.shape) @@ -125,6 +211,418 @@ class Stitcher: 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: + 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: + 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 + + def _tile_to_bgr(self, img): + if img.ndim == 2: + return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR) + return img + + def _tile_to_gray(self, bgr): + if bgr.ndim == 2: + return bgr + return cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY) + + def _ensure_u8_gray(self, gray): + if gray.dtype == np.uint8: + return gray + return cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) + + def _preprocess_gray_for_orb(self, g0, g1): + g0 = cv2.GaussianBlur(g0, (5, 5), 0) + g1 = cv2.GaussianBlur(g1, (5, 5), 0) + g0 = cv2.equalizeHist(g0) + g1 = cv2.equalizeHist(g1) + return g0, g1 + + def _estimate_bfmatcher_stitch_transform( + self, + t0, + t1, + *, + ransac_reproj_thresh=5.0, + ratio=0.9, + orb_nfeatures=5000, + verbose=True, + ): + """ + Estimate a 3x3 transform mapping tile coordinates into the reference tile frame. + """ + t0_bgr = self._tile_to_bgr(t0) + t1_bgr = self._tile_to_bgr(t1) + g0 = self._ensure_u8_gray(self._tile_to_gray(t0_bgr)) + g1 = self._ensure_u8_gray(self._tile_to_gray(t1_bgr)) + + if verbose: + print( + f"BF stitcher debug: t0 dtype={t0.dtype}, t1 dtype={t1.dtype}, " + f"g0 dtype={g0.dtype}, g0 range=({np.min(g0)}, {np.max(g0)}), " + f"g1 dtype={g1.dtype}, g1 range=({np.min(g1)}, {np.max(g1)})" + ) + + g0, g1 = self._preprocess_gray_for_orb(g0, g1) + orb = cv2.ORB_create( + nfeatures=orb_nfeatures, + scaleFactor=1.2, + nlevels=8, + edgeThreshold=15, + fastThreshold=5, + ) + k0, d0 = orb.detectAndCompute(g0, None) + k1, d1 = orb.detectAndCompute(g1, None) + + info = { + "ok": False, + "method": None, + "inliers": 0, + "mean_err": None, + "good_matches": 0, + "k0": k0, + "k1": k1, + "t0_bgr": t0_bgr, + "t1_bgr": t1_bgr, + } + if d0 is None or d1 is None or len(k0) < 4 or len(k1) < 4: + if verbose: + print( + "OpenCV BFMatcher stitcher: not enough features.", + f"k0={0 if k0 is None else len(k0)}", + f"k1={0 if k1 is None else len(k1)}", + f"d0_is_none={d0 is None}", + f"d1_is_none={d1 is None}", + ) + return None, info + + matches = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=False).knnMatch(d0, d1, k=2) + good = [] + for m_n in matches: + if len(m_n) != 2: + continue + m, n = m_n + if m.distance < ratio * n.distance: + good.append(m) + + info["good_matches"] = len(good) + if verbose: + print(f"BFMatcher good matches: {len(good)}") + if len(good) < 4: + return None, info + + pts0 = np.float32([k0[m.queryIdx].pt for m in good]).reshape(-1, 1, 2) + pts1 = np.float32([k1[m.trainIdx].pt for m in good]).reshape(-1, 1, 2) + method = "affine_partial2d" + M, inlier_mask = cv2.estimateAffinePartial2D( + pts1, + pts0, + method=cv2.RANSAC, + ransacReprojThreshold=ransac_reproj_thresh, + ) + if M is not None: + H = np.vstack([M, [0, 0, 1]]).astype(np.float64) + else: + method = "homography" + H, inlier_mask = cv2.findHomography(pts1, pts0, cv2.RANSAC, ransac_reproj_thresh) + + info["method"] = method + info["inliers"] = int(inlier_mask.sum()) if inlier_mask is not None else 0 + if verbose: + print(f"Estimated {method} inliers: {info['inliers']}") + if H is None: + return None, info + + if abs(H[2, 2]) > 1e-12: + H = H / H[2, 2] + + mean_err = 1e9 + if inlier_mask is not None: + inlier_bool = inlier_mask.ravel().astype(bool) + if np.any(inlier_bool): + projected = cv2.perspectiveTransform(pts1[inlier_bool], H) + err = np.linalg.norm( + projected.reshape(-1, 2) - pts0[inlier_bool].reshape(-1, 2), + axis=1, + ) + mean_err = float(err.mean()) if err.size else 1e9 + info["mean_err"] = mean_err + info["ok"] = True + return H, info + + def _translation_from_H_to_ref(self, H): + if H is None: + return 0.0, 0.0 + H = np.asarray(H, dtype=np.float64) + if abs(H[2, 2]) > 1e-12: + H = H / H[2, 2] + pt = H @ np.array([[0.0], [0.0], [1.0]], dtype=np.float64) + return float(pt[0, 0]), float(pt[1, 0]) + + def _tile_weight_map(self, h, w, margin=16): + """ + Linear ramp weight that is 0 at tile edges and 1.0 at margin pixels inward. + Used for feathered blending at tile overlaps. + """ + wy = np.ones(h, dtype=np.float32) + wx = np.ones(w, dtype=np.float32) + m = min(margin, h // 2, w // 2) + if m > 0: + ramp = np.linspace(0.0, 1.0, m + 1, dtype=np.float32)[1:] + wy[:m] = ramp + wy[-m:] = ramp[::-1] + wx[:m] = ramp + wx[-m:] = ramp[::-1] + return wy[:, None] * wx[None, :] + + def _blend_tile_inplace(self, blend, wsum, cov, src, weight, y0, x0): + """Copy non-overlap tiles directly; blend only when this tile touches existing content.""" + h, w = src.shape[:2] + H, W = blend.shape[:2] + y0d, x0d = max(0, y0), max(0, x0) + y1d, x1d = min(H, y0 + h), min(W, x0 + w) + if y0d >= y1d or x0d >= x1d: + return + + sy0, sx0 = y0d - y0, x0d - x0 + sh, sw = y1d - y0d, x1d - x0d + w_patch = weight[sy0 : sy0 + sh, sx0 : sx0 + sw] + blend_patch = blend[y0d:y1d, x0d:x1d] + wsum_patch = wsum[y0d:y1d, x0d:x1d] + cov_patch = cov[y0d:y1d, x0d:x1d] + src_patch = src[sy0 : sy0 + sh, sx0 : sx0 + sw] + + if not np.any(cov_patch): + blend_patch[:] = src_patch + wsum_patch[:] = w_patch + else: + total = wsum_patch + w_patch + alpha = w_patch / np.maximum(total, 1e-6) + src_patch_f = src_patch.astype(np.float32, copy=False) + blend_patch += (src_patch_f - blend_patch) * alpha[:, :, None] + wsum_patch[:] = total + + cov_patch[:] = 1 + + def _mean_gray_bgr(self, bgr): + return float(cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY).mean()) + + def _match_bgr_brightness(self, tile_bgr, ref_mean_gray): + delta = int(round(ref_mean_gray - self._mean_gray_bgr(tile_bgr))) + if delta == 0: + return tile_bgr + return np.clip(tile_bgr.astype(np.int16) + delta, 0, 255).astype(np.uint8) + + def _largest_fully_covered_rect(self, cov): + if cov.size == 0: + return None + + mask = cov > 0 + if not np.any(mask): + return None + + h, w = mask.shape + heights = np.zeros(w, dtype=np.int32) + best_area = 0 + best_rect = None + for y in range(h): + heights = np.where(mask[y], heights + 1, 0) + stack = [] + for x in range(w + 1): + cur_h = heights[x] if x < w else 0 + start = x + while stack and stack[-1][1] > cur_h: + start0, h0 = stack.pop() + area = h0 * (x - start0) + if area > best_area: + best_area = area + best_rect = (y - h0 + 1, y + 1, start0, x) + start = start0 + if not stack or stack[-1][1] < cur_h: + stack.append((start, cur_h)) + return best_rect + + def _compute_final_crop_rect(self, cov): + valid = cov > 0 + if not np.any(valid): + return None + + ys, xs = np.where(valid) + by0, by1 = int(ys.min()), int(ys.max()) + 1 + bx0, bx1 = int(xs.min()), int(xs.max()) + 1 + rect2 = self._largest_fully_covered_rect(cov[by0:by1, bx0:bx1]) + if rect2 is None: + return by0, by1, bx0, bx1 + + y0, y1, x0, x1 = rect2 + return by0 + y0, by0 + y1, bx0 + x0, bx0 + x1 + + def _geom_cache_key(self, t_ref_bgr, tiles_bgr_c, pad): + ref_h, ref_w = t_ref_bgr.shape[:2] + shapes = tuple(tb.shape[:2] for tb in tiles_bgr_c) + return ref_h, ref_w, shapes, pad + + def _render_feature_tiles( + self, + t_ref_bgr, + tiles_bgr, + H_use_list, + *, + pad=5, + frame_idx=0, + print_every=30, + verbose=False, + match_brightness=False, + canvas_cache=None, + blend=True, + ): + """ + Composite multiple tiles into one canvas using translation only, where each tile + origin maps into the reference tile coordinates. + """ + if H_use_list is None: + return None + + if canvas_cache is None: + canvas_cache = {} + do_profile = verbose and frame_idx % print_every == 0 + t_profile0 = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + t_ref_bgr = self._tile_to_bgr(t_ref_bgr) + h0c, w0c = t_ref_bgr.shape[:2] + if match_brightness: + ref_mean_gray = self._mean_gray_bgr(t_ref_bgr) + tiles_bgr_c = [self._match_bgr_brightness(self._tile_to_bgr(tb), ref_mean_gray) for tb in tiles_bgr] + else: + tiles_bgr_c = [self._tile_to_bgr(tb) for tb in tiles_bgr] + t_after_brightness = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + geom_key = self._geom_cache_key(t_ref_bgr, tiles_bgr_c, pad) + geom_cache_hit = canvas_cache.get("geom_key") == geom_key + if not geom_cache_hit: + ref_corners = np.array([[0, 0], [w0c, 0], [w0c, h0c], [0, h0c]], dtype=np.float64) + pts = [ref_corners] + for tile_bgr, H_use in zip(tiles_bgr_c, H_use_list): + if H_use is None: + continue + hi, wi = tile_bgr.shape[:2] + tx, ty = self._translation_from_H_to_ref(H_use) + pts.append( + np.array( + [[tx, ty], [tx + wi, ty], [tx + wi, ty + hi], [tx, ty + hi]], + dtype=np.float64, + ) + ) + + all_xy = np.concatenate(pts, axis=0) + x_min_f = float(all_xy[:, 0].min()) - pad + y_min_f = float(all_xy[:, 1].min()) - pad + x_max_f = float(all_xy[:, 0].max()) + pad + y_max_f = float(all_xy[:, 1].max()) + pad + out_w = max(1, int(np.ceil(x_max_f - x_min_f))) + out_h = max(1, int(np.ceil(y_max_f - y_min_f))) + ref_y0 = int(np.floor(-y_min_f)) + ref_x0 = int(np.floor(-x_min_f)) + + pastes = [] + for H_use in H_use_list: + if H_use is None: + pastes.append(None) + continue + tx, ty = self._translation_from_H_to_ref(H_use) + pastes.append((int(np.floor(ty - y_min_f)), int(np.floor(tx - x_min_f)))) + + canvas_cache["geom_key"] = geom_key + canvas_cache["out_w"] = out_w + canvas_cache["out_h"] = out_h + canvas_cache["ref_y0"] = ref_y0 + canvas_cache["ref_x0"] = ref_x0 + canvas_cache["pastes"] = pastes + canvas_cache["x_min_f"] = x_min_f + canvas_cache["y_min_f"] = y_min_f + canvas_cache["crop_rect"] = None + canvas_cache["blend_buf"] = np.zeros((out_h, out_w, 3), dtype=np.float32) + canvas_cache["wsum_buf"] = np.zeros((out_h, out_w), dtype=np.float32) + canvas_cache["cov_buf"] = np.zeros((out_h, out_w), dtype=np.uint8) + canvas_cache["weight_maps"] = {} + else: + out_w = canvas_cache["out_w"] + out_h = canvas_cache["out_h"] + ref_y0 = canvas_cache["ref_y0"] + ref_x0 = canvas_cache["ref_x0"] + pastes = canvas_cache["pastes"] + t_after_geom = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + blend_buf = canvas_cache["blend_buf"] + wsum = canvas_cache["wsum_buf"] + cov = canvas_cache["cov_buf"] + weight_maps = canvas_cache["weight_maps"] + blend_buf.fill(0) + wsum.fill(0) + cov.fill(0) + t_after_reset = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + blend_margin = 16 if blend else 0 + blended_tiles = 0 + for tile_bgr, H_use, paste in zip(tiles_bgr_c, H_use_list, pastes): + if H_use is None or paste is None: + continue + y0, x0 = paste + th, tw = tile_bgr.shape[:2] + wkey = (th, tw, blend_margin) + if wkey not in weight_maps: + weight_maps[wkey] = self._tile_weight_map(th, tw, blend_margin) + self._blend_tile_inplace(blend_buf, wsum, cov, tile_bgr, weight_maps[wkey], y0, x0) + blended_tiles += 1 + + ref_wkey = (h0c, w0c, blend_margin) + if ref_wkey not in weight_maps: + weight_maps[ref_wkey] = self._tile_weight_map(h0c, w0c, blend_margin) + self._blend_tile_inplace(blend_buf, wsum, cov, t_ref_bgr, weight_maps[ref_wkey], ref_y0, ref_x0) + t_after_blend = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + crop_rect = canvas_cache.get("crop_rect") + if crop_rect is None: + crop_rect = self._compute_final_crop_rect(cov) + canvas_cache["crop_rect"] = crop_rect + if crop_rect is not None: + y0, y1, x0, x1 = crop_rect + stitched = np.clip(blend_buf[y0:y1, x0:x1], 0, 255).astype(np.uint8) + else: + stitched = np.clip(blend_buf, 0, 255).astype(np.uint8) + t_after_finalize = cv2.getTickCount() / cv2.getTickFrequency() if do_profile else 0.0 + + if do_profile: + xmf = canvas_cache.get("x_min_f", 0.0) + ymf = canvas_cache.get("y_min_f", 0.0) + print( + f"stitch_feature_optimized: canvas out_w={out_w}, out_h={out_h}, " + f"x_min={xmf:.2f}, y_min={ymf:.2f}" + ) + print( + "stitch_feature_optimized steps: " + f"brightness={t_after_brightness - t_profile0:.6f}s, " + f"geom={t_after_geom - t_after_brightness:.6f}s ({'hit' if geom_cache_hit else 'miss'}), " + f"reset={t_after_reset - t_after_geom:.6f}s, " + f"blend={t_after_blend - t_after_reset:.6f}s ({blended_tiles + 1} tiles incl ref), " + f"finalize={t_after_finalize - t_after_blend:.6f}s, " + f"total={t_after_finalize - t_profile0:.6f}s" + ) + + return stitched + def find_stitch_boxes(self,img_src,show = False): img_corr = self.__correct_edge_bloom(img_src) @@ -621,4 +1119,6 @@ class Stitcher: if area < min_area or area > max_area or (abs(ww - hh) > self.max_aspect_ratio): continue filtered.append((x,y,ww,hh)) - return filtered \ No newline at end of file + return filtered + + \ No newline at end of file diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index cb90086..f30f816 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -5,7 +5,7 @@ A Python library for processing and analyzing optical tactile sensor data, including image stitching, optical flow tracking, and deformation decomposition. """ -__version__ = "0.3.0" +__version__ = "0.3.1" __author__ = "Orisys" from .sensor import Sensor diff --git a/src/orisys/util.py b/src/orisys/util.py index 55a78a1..ee66820 100644 --- a/src/orisys/util.py +++ b/src/orisys/util.py @@ -44,7 +44,8 @@ def draw_arrows(frame, optical_flow, fps=True, threshold=5, grid_spacing=40, arr if mag[i] < threshold: cv2.circle(image, pt1, radius=1, color=below_threshold_color, thickness=-1) else: - thickness = max(1, int(2 * mag[i] / mag.max())) if mag.max() > 0 else 2 + # thickness = max(1, int(2 * mag[i] / mag.max())) if mag.max() > 0 else 2 + thickness = 1 cv2.arrowedLine(image, pt1, pt2, color, thickness=thickness, tipLength=0.2) return image diff --git a/test_stitch.py b/test_stitch.py index 0ec9138..35c4689 100644 --- a/test_stitch.py +++ b/test_stitch.py @@ -6,471 +6,44 @@ 2. 循环获取图像并计算形变和触碰点 3. 读取各类传感器数据 4. 可视化显示 - - """ import os -os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +import argparse + import cv2 import orisys -import numpy as np -import argparse -import time - -def _tile_to_bgr(img): - if img.ndim == 2: - return cv2.cvtColor(img, cv2.COLOR_GRAY2BGR) - return img - - -def _tile_to_gray_bgr(bgr): - if bgr.ndim == 2: - return bgr - return cv2.cvtColor(bgr, cv2.COLOR_BGR2GRAY) - - -def _ensure_u8_gray(gray): - if gray.dtype == np.uint8: - return gray - return cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX).astype(np.uint8) - - -def _preprocess_gray_for_orb(g0, g1): - g0 = cv2.GaussianBlur(g0, (5, 5), 0) - g1 = cv2.GaussianBlur(g1, (5, 5), 0) - g0 = cv2.equalizeHist(g0) - g1 = cv2.equalizeHist(g1) - return g0, g1 - - -def _translation_from_H_to_ref(H): - """ - Use only the translation part of H (tile -> reference): where tile origin (0,0) lands in ref coords. - Ignores rotation/scale/shear for fast compositing (slice paste). - """ - if H is None: - return 0.0, 0.0 - H = np.asarray(H, dtype=np.float64) - if abs(H[2, 2]) > 1e-12: - H = H / H[2, 2] - pt = H @ np.array([[0.0], [0.0], [1.0]], dtype=np.float64) - return float(pt[0, 0]), float(pt[1, 0]) - - -def _paste_bgr(dst, src, y0, x0): - """Paste src into dst at top-left (y0, x0), clipping to dst bounds.""" - h, w = src.shape[:2] - H, W = dst.shape[:2] - y0d, x0d = max(0, y0), max(0, x0) - y1d, x1d = min(H, y0 + h), min(W, x0 + w) - if y0d >= y1d or x0d >= x1d: - return - sy0 = y0d - y0 - sx0 = x0d - x0 - sh = y1d - y0d - sw = x1d - x0d - dst[y0d:y1d, x0d:x1d] = src[sy0 : sy0 + sh, sx0 : sx0 + sw] - - -def _accumulate_tile_bgr(acc, wsum, src, y0, x0): - """ - Add tile BGR into float accumulators for mean blending where tiles overlap. - acc: (H,W,3) float32, wsum: (H,W) float32 counts per pixel. - """ - h, w = src.shape[:2] - H, W = acc.shape[:2] - y0d, x0d = max(0, y0), max(0, x0) - y1d, x1d = min(H, y0 + h), min(W, x0 + w) - if y0d >= y1d or x0d >= x1d: - return - sy0 = y0d - y0 - sx0 = x0d - x0 - sh = y1d - y0d - sw = x1d - x0d - patch = src[sy0 : sy0 + sh, sx0 : sx0 + sw].astype(np.float32, copy=False) - acc[y0d:y1d, x0d:x1d] += patch - wsum[y0d:y1d, x0d:x1d] += 1.0 - - -def _finalize_mean_blend(acc, wsum): - """Average accumulated colors; pixels with no coverage stay 0.""" - out_h, out_w = wsum.shape - stitched = np.zeros((out_h, out_w, 3), dtype=np.uint8) - valid = wsum > 1e-6 - stitched[valid] = np.clip(acc[valid] / wsum[valid, None], 0.0, 255.0).astype(np.uint8) - return stitched - - -def _h_list_signature(H_use_list): - """Stable signature to detect when cached canvas geometry must be rebuilt.""" - parts = [] - for H in H_use_list: - if H is None: - parts.append("None") - else: - parts.append(np.asarray(H, dtype=np.float64).tobytes()) - return tuple(parts) - - -# Tuning (if stitch position is wrong or unstable): -# 1) ORB: nfeatures, fastThreshold, edgeThreshold in ORB_create. -# 2) Lowe ratio: lower => stricter matches; higher => more matches, more outliers. -# 3) ransac_reproj_thresh: affine/homography RANSAC inlier threshold (pixels). -# 4) Temporal: smooth_alpha, min_inliers, max_mean_reproj_err. -def estimate_bfmatcher_stitch_transform( - t0, - t1, - *, - H_prev=None, - smooth_alpha=0.2, - min_inliers=30, - max_mean_reproj_err=8.0, - ransac_reproj_thresh=5.0, - ratio=0.9, - orb_nfeatures=5000, - verbose=True, -): - """ - Estimate 3x3 transform H_use mapping t1 coordinates -> t0 coordinates (t0 is reference). - - Parameters - ---------- - t0, t1 : ndarray - Tile images (BGR or grayscale). - H_prev : ndarray or None - Previous frame's H_use for temporal smoothing; pass None on first frame. - - Returns - ------- - H_use : ndarray (3,3) or None - Normalized homography/affine-as-homography. None if not enough features or no valid model. - H_prev_out : ndarray or None - Value to pass as H_prev on the next call (same as H_use when successful, else unchanged H_prev). - info : dict - Keys may include: ok, method, inliers, mean_err, good_matches, k0, k1, t0_bgr, t1_bgr (for viz). - """ - t0_bgr = _tile_to_bgr(t0) - t1_bgr = _tile_to_bgr(t1) - g0 = _ensure_u8_gray(_tile_to_gray_bgr(t0_bgr)) - g1 = _ensure_u8_gray(_tile_to_gray_bgr(t1_bgr)) - - if verbose: - print( - f"BF stitcher debug: t0 dtype={t0.dtype}, t1 dtype={t1.dtype}, " - f"g0 dtype={g0.dtype}, g0 range=({np.min(g0)}, {np.max(g0)}), " - f"g1 dtype={g1.dtype}, g1 range=({np.min(g1)}, {np.max(g1)})" - ) - - g0, g1 = _preprocess_gray_for_orb(g0, g1) - - orb = cv2.ORB_create( - nfeatures=orb_nfeatures, scaleFactor=1.2, nlevels=8, edgeThreshold=15, fastThreshold=5 - ) - k0, d0 = orb.detectAndCompute(g0, None) - k1, d1 = orb.detectAndCompute(g1, None) - - info = { - "ok": False, - "method": None, - "inliers": 0, - "mean_err": None, - "good_matches": 0, - "k0": k0, - "k1": k1, - "t0_bgr": t0_bgr, - "t1_bgr": t1_bgr, - } - - if d0 is None or d1 is None or len(k0) < 4 or len(k1) < 4: - if verbose: - print( - "OpenCV BFMatcher stitcher: not enough features.", - f"k0={0 if k0 is None else len(k0)}", - f"k1={0 if k1 is None else len(k1)}", - f"d0_is_none={d0 is None}", - f"d1_is_none={d1 is None}", - ) - return None, H_prev, info - - bf = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=False) - matches = bf.knnMatch(d0, d1, k=2) - - good = [] - for m_n in matches: - if len(m_n) != 2: - continue - m, n = m_n - if m.distance < ratio * n.distance: - good.append(m) - info["good_matches"] = len(good) - info["good"] = good - if verbose: - print(f"BFMatcher good matches: {len(good)}") - - if len(good) < 4: - return None, H_prev, info - - pts0 = np.float32([k0[m.queryIdx].pt for m in good]).reshape(-1, 1, 2) - pts1 = np.float32([k1[m.trainIdx].pt for m in good]).reshape(-1, 1, 2) - - method = "affine_partial2d" - M, inliers_aff = cv2.estimateAffinePartial2D( - pts1, pts0, method=cv2.RANSAC, ransacReprojThreshold=ransac_reproj_thresh - ) - if M is not None: - H = np.vstack([M, [0, 0, 1]]).astype(np.float64) - inlier_mask = inliers_aff - else: - method = "homography" - H, inlier_mask = cv2.findHomography(pts1, pts0, cv2.RANSAC, ransac_reproj_thresh) - - inliers = int(inlier_mask.sum()) if inlier_mask is not None else 0 - info["method"] = method - info["inliers"] = inliers - if verbose: - print(f"Estimated {method} inliers: {inliers}") - - if H is None: - info["transform_none"] = True - return None, H_prev, info - - if abs(H[2, 2]) > 1e-12: - H = H / H[2, 2] - - mean_err = 1e9 - if inlier_mask is not None: - inlier_bool = inlier_mask.ravel().astype(bool) - if np.any(inlier_bool): - projected = cv2.perspectiveTransform(pts1[inlier_bool], H) - err = np.linalg.norm( - projected.reshape(-1, 2) - pts0[inlier_bool].reshape(-1, 2), axis=1 - ) - mean_err = float(err.mean()) if err.size else 1e9 - info["mean_err"] = mean_err - - H_use = H - if H_prev is not None: - if inliers < min_inliers or mean_err > max_mean_reproj_err: - H_use = H_prev - else: - H_use = smooth_alpha * H + (1.0 - smooth_alpha) * H_prev - if abs(H_use[2, 2]) > 1e-12: - H_use = H_use / H_use[2, 2] - - info["ok"] = True - return H_use, H_use.copy(), info - - -def render_stitched_preview( - t0_bgr, - t1_bgr, - H_use, - *, - pad=None, - frame_idx=0, - print_every=30, - verbose=True, -): - """ - Warp t1 with H_use (t1->t0), paste t0 on a fixed padded canvas. Returns stitched BGR image. - """ - if H_use is None: - return None - h0c, w0c = t0_bgr.shape[:2] - if pad is None: - pad = max(50, int(0.1 * w0c)) - out_w = w0c + 2 * pad - out_h = h0c + 2 * pad - T_pad = np.array([[1, 0, pad], [0, 1, pad], [0, 0, 1]], dtype=np.float64) - H_total = T_pad @ H_use - - if verbose and frame_idx % print_every == 0: - h1c, w1c = t1_bgr.shape[:2] - corners1 = np.float32([[0, 0], [w1c, 0], [w1c, h1c], [0, h1c]]).reshape(-1, 1, 2) - corners1_warped = cv2.perspectiveTransform(corners1, H_total).reshape(-1, 2) - corners0_out = np.float32( - [ - [pad, pad], - [pad + w0c, pad], - [pad + w0c, pad + h0c], - [pad, pad + h0c], - ] - ) - print("=== Stitch transform debug ===") - print("H_use (t1 -> t0):\n", H_use) - print("H_total (output canvas):\n", H_total) - print("t1 corners warped into output (x,y):", corners1_warped) - print("t0 corners in output (x,y):", corners0_out) - - stitched = cv2.warpPerspective(t1_bgr, H_total, (out_w, out_h), flags=cv2.INTER_LINEAR) - stitched[pad : pad + h0c, pad : pad + w0c] = t0_bgr - return stitched - - -def _geom_cache_key(t_ref_bgr, tiles_bgr_c, H_use_list, pad): - """Invalidate canvas layout when ref/tile sizes, transforms, or pad change.""" - ref_h, ref_w = t_ref_bgr.shape[:2] - shapes = tuple(tb.shape[:2] for tb in tiles_bgr_c) - return (ref_h, ref_w, shapes, pad, _h_list_signature(H_use_list)) - - -def render_multi_tiles_preview( - t_ref_bgr, - tiles_bgr, - H_use_list, - *, - pad=60, - frame_idx=0, - print_every=30, - verbose=False, - canvas_cache=None, -): - """ - Composite multiple tiles into one canvas using translation only (where each tile's - origin maps in ref coords). No per-tile warpPerspective. - - - t_ref_bgr: reference tile; output is in a shifted ref-aligned frame. - - each H_use maps tile_i -> reference; only the translation of origin (0,0) is used. - - tiles_bgr and H_use_list must have the same length. - - canvas_cache: optional dict, updated in place to reuse out_w/out_h and paste origins - when geometry is unchanged. - Overlapping regions use mean blending (equal weight per contributing tile). - """ - if H_use_list is None: - return None - - if canvas_cache is None: - canvas_cache = {} - - t_ref_bgr = _tile_to_bgr(t_ref_bgr) - h0c, w0c = t_ref_bgr.shape[:2] - tiles_bgr_c = [_tile_to_bgr(tb) for tb in tiles_bgr] - - geom_key = _geom_cache_key(t_ref_bgr, tiles_bgr_c, H_use_list, pad) - - if canvas_cache.get("geom_key") != geom_key: - ref_corners = np.array([[0, 0], [w0c, 0], [w0c, h0c], [0, h0c]], dtype=np.float64) - pts = [ref_corners] - for tile_bgr, H_use in zip(tiles_bgr_c, H_use_list): - if H_use is None: - continue - hi, wi = tile_bgr.shape[:2] - tx, ty = _translation_from_H_to_ref(H_use) - pts.append( - np.array( - [[tx, ty], [tx + wi, ty], [tx + wi, ty + hi], [tx, ty + hi]], - dtype=np.float64, - ) - ) - all_xy = np.concatenate(pts, axis=0) - x_min_f = float(all_xy[:, 0].min()) - pad - y_min_f = float(all_xy[:, 1].min()) - pad - x_max_f = float(all_xy[:, 0].max()) + pad - y_max_f = float(all_xy[:, 1].max()) + pad - - out_w = max(1, int(np.ceil(x_max_f - x_min_f))) - out_h = max(1, int(np.ceil(y_max_f - y_min_f))) - ref_y0 = int(np.floor(0.0 - y_min_f)) - ref_x0 = int(np.floor(0.0 - x_min_f)) - - pastes = [] - for tile_bgr, H_use in zip(tiles_bgr_c, H_use_list): - if H_use is None: - pastes.append(None) - continue - tx, ty = _translation_from_H_to_ref(H_use) - pastes.append((int(np.floor(ty - y_min_f)), int(np.floor(tx - x_min_f)))) - - canvas_cache["geom_key"] = geom_key - canvas_cache["out_w"] = out_w - canvas_cache["out_h"] = out_h - canvas_cache["ref_y0"] = ref_y0 - canvas_cache["ref_x0"] = ref_x0 - canvas_cache["pastes"] = pastes - canvas_cache["x_min_f"] = x_min_f - canvas_cache["y_min_f"] = y_min_f - else: - out_w = canvas_cache["out_w"] - out_h = canvas_cache["out_h"] - ref_y0 = canvas_cache["ref_y0"] - ref_x0 = canvas_cache["ref_x0"] - pastes = canvas_cache["pastes"] - - acc = np.zeros((out_h, out_w, 3), dtype=np.float32) - wsum = np.zeros((out_h, out_w), dtype=np.float32) - _accumulate_tile_bgr(acc, wsum, t_ref_bgr, ref_y0, ref_x0) - for tile_bgr, H_use, paste in zip(tiles_bgr_c, H_use_list, pastes): - if H_use is None or paste is None: - continue - y0, x0 = paste - _accumulate_tile_bgr(acc, wsum, tile_bgr, y0, x0) - stitched = _finalize_mean_blend(acc, wsum) - - # Crop away any uncovered regions (which correspond to black padding at borders). - # Use `wsum` for robustness even if a tile's actual content is black. - valid = wsum > 1e-6 - if np.any(valid): - ys, xs = np.where(valid) - y0c, y1c = int(ys.min()), int(ys.max()) + 1 - x0c, x1c = int(xs.min()), int(xs.max()) + 1 - stitched = stitched[y0c:y1c, x0c:x1c] - - stitched = cv2.GaussianBlur(stitched, (5, 5), 0) - - - if verbose and frame_idx % print_every == 0: - xmf = canvas_cache.get("x_min_f", 0.0) - ymf = canvas_cache.get("y_min_f", 0.0) - print( - f"render_multi_tiles_preview: canvas out_w={out_w}, out_h={out_h}, " - f"x_min={xmf:.2f}, y_min={ymf:.2f}" - ) - - return stitched +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" def main(): 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="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( + "--match-brightness", + dest="match_brightness", + action="store_true", + help="Enable per-tile brightness matching", + ) + parser.set_defaults(match_brightness=False) args = parser.parse_args() + try: + vid_src = int(args.video) + except ValueError: + vid_src = args.video + # 步骤1：创建传感器对象 (必要步骤)， # config_name为配置文件名（内置配置文件名）或路径（自定义配置文件路径） # cal_path为校准文件路径, 每一个设备需要创建/使用一个校准文件 - - # 方式1: 从视频文件读取 - try: - input = int(args.video) - is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - except: - is_camera = False - sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) + sensor = orisys.Sensor(vid_src, config_name=args.config, cal_path=args.cal, verbose=args.verbose) + sensor.stitcher.configure_feature_stitch(match_brightness=args.match_brightness) print("\n按 'q' 退出程序，按 'r' 重置传感器追踪器") - # For stabilizing frame-to-frame stitching (reduce oscillation). - # Stores the last good transform (mapping t1 -> t0) and exponentially smooths it. - smooth_alpha = 0.2 # 0..1; higher => less smoothing, more jitter - min_inliers = 30 # if inliers are too low, keep previous transform - max_mean_reproj_err = 8.0 # if reprojection error is too high, keep previous transform - frame_idx = 0 - print_every = 30 # reduce console spam; set to 1 for per-frame output - - # Stitch a 3x3 (9 tiles) block from the row-major `tile_list`. - # For your 3x5 layout (nrow=3, ncol=5), this stitches the "middle 3x3" - # using the center tile of that 3x3 as the reference frame. - H_prev_tiles = {} # key: tile_list index, value: H_prev - # Cache BFMatcher-estimated transforms from the first frame. - H_use_list_cached = None - moving_indices_cached = None - crop_pad = 5 # tighter crop in stitched canvas - canvas_cache = {} - while True: # 步骤2：信息提取 # 步骤 2.1: 获取并拼接图像 (必要步骤) @@ -478,118 +51,13 @@ def main(): if img is None: print("Failed to get image (end of stream or capture error). Exiting.") break - # boxes = sensor.stitcher.boxes - # draw boxes for checking - # for box in boxes: - # x0, y0, x1, y1 = box - # cv2.rectangle(sensor.frame, (x0, y0), (x1, y1), (0, 255, 0), 2) - # cv2.imshow("boxes", sensor.frame) - # cv2.waitKey(0) - - tiles = sensor.stitcher.tile_list - frame_idx += 1 - ncol = getattr(sensor.stitcher, "ncol", 0) or 0 - nrow = getattr(sensor.stitcher, "nrow", 0) or 0 + if img is not None: + cv2.imshow("OpenCV Stitch (feature-based)", img) - grid_r, grid_c = 3, 3 - needed = grid_r * grid_c - if len(tiles) < needed or ncol < grid_c or nrow < grid_r: - print(f"tile_list too small for 9-tile stitch: len={len(tiles)} ncol={ncol} nrow={nrow} needed={needed}") - cv2.waitKey(1) - continue - - # Center the 3x3 block in the full (nrow x ncol) layout. - # Use the center tile of that 3x3 as reference frame. - start_r = (nrow - grid_r) // 2 - start_c = (ncol - grid_c) // 2 - - idx_ref = (start_r + 1) * ncol + (start_c + 1) - if idx_ref < 0 or idx_ref >= len(tiles): - print("Computed reference tile index out of range:", idx_ref) - cv2.waitKey(1) - continue - - t_ref = tiles[idx_ref] - - # Stitcher: warp each moving tile into the reference tile coordinate system. - moving_tiles = [] - moving_indices = [] - for r in range(grid_r): - for c in range(grid_c): - idx = (start_r + r) * ncol + (start_c + c) - if idx == idx_ref: - continue - if idx < 0 or idx >= len(tiles): - continue - moving_tiles.append(tiles[idx]) - moving_indices.append(idx) - - # Only run BFMatcher transform estimation on the first frame. - if H_use_list_cached is None or moving_indices_cached != moving_indices: - H_use_list = [] - do_verbose_first = (frame_idx % print_every == 0) - for j, (tile, tile_idx) in enumerate(zip(moving_tiles, moving_indices)): - H_prev_j = H_prev_tiles.get(tile_idx) - H_use_j, H_prev_j_out, info_j = estimate_bfmatcher_stitch_transform( - t_ref, - tile, - H_prev=H_prev_j, - smooth_alpha=smooth_alpha, - min_inliers=min_inliers, - max_mean_reproj_err=max_mean_reproj_err, - verbose=(do_verbose_first and j == 0), - ) - H_prev_tiles[tile_idx] = H_prev_j_out - H_use_list.append(H_use_j) - - if ( - j == 0 - and do_verbose_first - and info_j.get("good") is not None - and len(info_j["good"]) >= 4 - ): - draw = cv2.drawMatches( - info_j["t0_bgr"], - info_j["k0"], - info_j["t1_bgr"], - info_j["k1"], - info_j["good"], - None, - matchColor=(0, 255, 0), - singlePointColor=(255, 0, 0), - flags=cv2.DrawMatchesFlags_NOT_DRAW_SINGLE_POINTS, - ) - cv2.imshow("BFMatcher Matches (t_ref vs t_moving0)", draw) - - H_use_list_cached = H_use_list - moving_indices_cached = list(moving_indices) - else: - H_use_list = H_use_list_cached - - start_time = time.perf_counter() - stitched = render_multi_tiles_preview( - t_ref, - moving_tiles, - H_use_list, - pad=crop_pad, - frame_idx=frame_idx, - print_every=print_every, - verbose=False, - canvas_cache=canvas_cache, - ) - end_time = time.perf_counter() - print(f"render_multi_tiles_preview: {end_time - start_time} seconds") - - if stitched is not None: - cv2.imshow("OpenCV Stitch (9 tiles, middle-ref stabilized)", stitched) - # cv2.imshow("tiles_1_2 (two rows, 2nd shifted)", canvas) - # cv2.waitKey(0) - # 步骤 2.2: 计算形变场 (必要步骤) - # is_touched = sensor.compute_deformation(check_motion=True, threshold=6) - # 步骤 2.3: 计算触碰点 - - #步骤 3.5: 键盘控制 + # 步骤 2.2: 计算形变场 (必要步骤) — sensor.compute_deformation(...) + # 步骤 2.3: 计算触碰点 + # 步骤 3: 键盘控制 key = cv2.waitKey(1) & 0xFF if key == ord("q"): print("\n退出程序...") @@ -598,12 +66,12 @@ def main(): # 重置光流追踪器（当追踪漂移时使用） sensor.reset() print("传感器已重置") - + # ========================================================================= # 4. 清理资源 # ========================================================================= sensor.disconnect() # 释放视频源和相关资源 -if __name__ == '__main__': +if __name__ == "__main__": main()