Download infer.py from gatilin/sam2_segmenter: direct link, hf CLI and curl.
- Browser
- Download file 15.4 kB
-
https://huggingface.co/gatilin/sam2_segmenter/resolve/main/infer.py
- Command line
-
hf download hf://gatilin/sam2_segmenter/infer.py
-
curl -L -o infer.py https://huggingface.co/gatilin/sam2_segmenter/resolve/main/infer.py
15.4 kB
| #!/usr/bin/env python3 | |
| """ | |
| SAM2 ONNX 独立推理脚本 (sam2-hiera-base-plus) | |
| 依赖: numpy, opencv-python, onnxruntime | |
| 无需 torch / CUDA | |
| 用法: | |
| # 单点分割 | |
| python infer.py --image photo.jpg --point 500 400 | |
| # 多点分割 | |
| python infer.py --image photo.jpg --point 500 400 --point 600 300 | |
| # bbox 中心点分割 | |
| python infer.py --image photo.jpg --bbox 100 200 500 600 | |
| # 保存结果 | |
| python infer.py --image photo.jpg --point 500 400 --save mask.png | |
| """ | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import cv2 | |
| import numpy as np | |
| SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| VE_ONNX = os.path.join(SCRIPT_DIR, "vision_encoder.onnx") | |
| MD_ONNX = os.path.join(SCRIPT_DIR, "mask_decoder.onnx") | |
| CONV_S0_NPZ = os.path.join(SCRIPT_DIR, "conv_s0.npz") | |
| CONV_S1_NPZ = os.path.join(SCRIPT_DIR, "conv_s1.npz") | |
| PE_NPZ = os.path.join(SCRIPT_DIR, "prompt_encoder.npz") | |
| PROC_CONFIG = os.path.join(SCRIPT_DIR, "preprocessor_config.json") | |
| INPUT_SIZE = 1024 | |
| PE_INPUT_SIZE = 1024 | |
| # ============================================================================ | |
| # 图像预处理 | |
| # ============================================================================ | |
| def preprocess_image(image_rgb: np.ndarray): | |
| """ | |
| SAM2 图像预处理: resize longest edge to 1024, pad to 1024×1024, normalize | |
| 返回: pixel_values (1, 3, 1024, 1024), (orig_h, orig_w), scale | |
| """ | |
| h, w = image_rgb.shape[:2] | |
| scale = INPUT_SIZE / max(h, w) | |
| new_h, new_w = int(h * scale), int(w * scale) | |
| resized = cv2.resize(image_rgb, (new_w, new_h), interpolation=cv2.INTER_AREA) | |
| # Normalize (ImageNet-like, from SAM2 processor) | |
| img = resized.astype(np.float32) / 255.0 | |
| mean = np.array([0.485, 0.456, 0.406], dtype=np.float32) | |
| std = np.array([0.229, 0.224, 0.225], dtype=np.float32) | |
| img = (img - mean) / std | |
| # Pad to INPUT_SIZE × INPUT_SIZE | |
| padded = np.zeros((INPUT_SIZE, INPUT_SIZE, 3), dtype=np.float32) | |
| padded[:new_h, :new_w] = img | |
| # HWC → CHW → NCHW | |
| pixel_values = padded.transpose(2, 0, 1)[np.newaxis] | |
| return pixel_values, (h, w), scale | |
| # ============================================================================ | |
| # Conv2d 推理 (conv_s0, conv_s1) | |
| # ============================================================================ | |
| def conv2d_forward(x, weight, bias): | |
| """纯 numpy Conv2d(1×1) 实现""" | |
| # weight: (out_ch, in_ch, 1, 1), x: (1, in_ch, H, W) | |
| B, C, H, W = x.shape | |
| out_ch = weight.shape[0] | |
| # 1×1 conv = matrix multiply on channel dimension | |
| x_flat = x.reshape(B, C, -1) # (B, C, H*W) | |
| w_flat = weight.reshape(out_ch, C) # (out_ch, C) | |
| out = np.einsum('ij,bjk->bik', w_flat, x_flat) + bias.reshape(1, -1, 1) | |
| return out.reshape(B, out_ch, H, W) | |
| # ============================================================================ | |
| # Prompt Encoder (纯 numpy 实现) | |
| # ============================================================================ | |
| class PromptEncoder: | |
| """SAM2 Prompt Encoder — 纯 numpy 实现""" | |
| def __init__(self, pe_weights_path): | |
| state = np.load(pe_weights_path) | |
| # Sinusoidal positional embedding matrix: (2, 128) | |
| self.pos_embed = state["shared_embedding.positional_embedding"] | |
| # Point type embeddings | |
| self.point_embed_weight = state["point_embed.weight"] # (4, 256) | |
| self.not_a_point_weight = state["not_a_point_embed.weight"] # (1, 256) | |
| self.no_mask_weight = state["no_mask_embed.weight"] # (1, 256) | |
| def _embed_points(self, coords): | |
| """ | |
| Sinusoidal positional encoding for 2D coordinates. | |
| coords: (N, n_points, 2) — in PE coordinate space (0 to PE_INPUT_SIZE) | |
| """ | |
| # Shift to center of pixel | |
| coords = coords + 0.5 | |
| # Normalize to [0, 1] | |
| coords[..., 0] = coords[..., 0] / PE_INPUT_SIZE | |
| coords[..., 1] = coords[..., 1] / PE_INPUT_SIZE | |
| # Scale to [-1, 1] | |
| coords = 2 * coords - 1 | |
| # Apply positional embedding matrix | |
| # coords: (N, P, 2) @ pos_embed: (2, D) → (N, P, D) | |
| embedded = coords @ self.pos_embed | |
| # Scale by 2π | |
| embedded = 2 * np.pi * embedded | |
| # Concat sin and cos | |
| sin_emb = np.sin(embedded) # (N, P, D) | |
| cos_emb = np.cos(embedded) # (N, P, D) | |
| return np.concatenate([sin_emb, cos_emb], axis=-1) # (N, P, 2D) | |
| def encode_points(self, points, labels): | |
| """ | |
| Encode point prompts. | |
| points: (B, n_prompts, n_points, 2) — raw pixel coordinates | |
| labels: (B, n_prompts, n_points) — 1=foreground, 0=background, -1=not_a_point | |
| Returns: sparse_emb (B, n_prompts, n_points, embed_dim), dense_emb (B, embed_dim, 64, 64) | |
| """ | |
| B, n_prompts, n_points, _ = points.shape | |
| embed_dim = self.pos_embed.shape[1] * 2 # 2D = sin + cos | |
| # Flatten for encoding | |
| coords_flat = points.reshape(B, -1, 2) # (B, n_prompts*n_points, 2) | |
| labels_flat = labels.reshape(B, -1) # (B, n_prompts*n_points) | |
| # Sinusoidal position encoding | |
| pos_enc = self._embed_points(coords_flat) # (B, n_total, embed_dim) | |
| # Add type embeddings based on labels | |
| # label=1 → point_embed (foreground) | |
| # label=0 → not_a_point (should not appear normally) | |
| # label=-1 → not_a_point_embed | |
| fg_mask = (labels_flat == 1).astype(np.float32)[..., np.newaxis] | |
| bg_mask = (labels_flat == 0).astype(np.float32)[..., np.newaxis] | |
| na_mask = (labels_flat == -1).astype(np.float32)[..., np.newaxis] | |
| # point_embed.weight: (4, 256) — nn.Embedding(4, 256) | |
| # label=0 → row 0 (background), label=1 → row 1 (foreground) | |
| fg_emb = self.point_embed_weight[1] # (256,) — label=1 foreground | |
| bg_emb = self.point_embed_weight[0] # (256,) — label=0 background | |
| na_emb = self.not_a_point_weight[0] # (256,) — label=-1 not_a_point | |
| type_emb = (fg_mask * fg_emb + | |
| bg_mask * bg_emb + | |
| na_mask * na_emb) | |
| sparse_emb = pos_enc + type_emb # (B, n_total, embed_dim) | |
| # Reshape: (B, n_prompts, n_points, embed_dim) → (B, n_prompts, n_points*embed_dim) | |
| sparse_emb = sparse_emb.reshape(B, n_prompts, n_points, embed_dim) | |
| # For the ONNX mask decoder, sparse_prompt_emb shape is (B, n_prompts, n_points, embed_dim) | |
| # and dense_prompt_emb shape is (B, embed_dim, 64, 64) | |
| # Dense embedding: start with no_mask_embed, broadcast to spatial size | |
| dense_emb = np.broadcast_to( | |
| self.no_mask_weight.reshape(1, embed_dim, 1, 1), | |
| (B, embed_dim, 64, 64) | |
| ).copy() | |
| return sparse_emb.astype(np.float32), dense_emb.astype(np.float32) | |
| # ============================================================================ | |
| # 主推理流程 | |
| # ============================================================================ | |
| class SAM2Segmenter: | |
| """SAM2 独立分割器 — 使用 ONNX VE + MD""" | |
| def __init__(self, providers=None): | |
| import onnxruntime as ort | |
| if providers is None: | |
| providers = ["CPUExecutionProvider"] | |
| self.ve_sess = ort.InferenceSession(VE_ONNX, providers=providers) | |
| self.md_sess = ort.InferenceSession(MD_ONNX, providers=providers) | |
| # Load conv weights from .npz (no torch dependency) | |
| s0 = np.load(CONV_S0_NPZ) | |
| s1 = np.load(CONV_S1_NPZ) | |
| self.conv_s0_w = s0["weight"] | |
| self.conv_s0_b = s0["bias"] | |
| self.conv_s1_w = s1["weight"] | |
| self.conv_s1_b = s1["bias"] | |
| # Load prompt encoder | |
| self.pe = PromptEncoder(PE_NPZ) | |
| def encode_image(self, image_rgb): | |
| """运行 Vision Encoder, 返回图像特征缓存""" | |
| pixel_values, (orig_h, orig_w), scale = preprocess_image(image_rgb) | |
| ve_out = self.ve_sess.run(None, {"pixel_values": pixel_values}) | |
| # Outputs: last_hidden_state, fpn0, fpn1, fpn2, pos0, pos1, pos2 | |
| _, fpn0, fpn1, fpn2, _, _, pos2 = ve_out | |
| # Apply conv_s0 / conv_s1 | |
| hres0 = conv2d_forward(fpn0, self.conv_s0_w, self.conv_s0_b) | |
| hres1 = conv2d_forward(fpn1, self.conv_s1_w, self.conv_s1_b) | |
| return { | |
| "image_embeddings": fpn2, | |
| "image_pos_enc": pos2, | |
| "high_res_feat_0": hres0, | |
| "high_res_feat_1": hres1, | |
| "orig_h": orig_h, | |
| "orig_w": orig_w, | |
| "scale": scale, | |
| } | |
| def _scale_points(self, ve_cache, points_xy): | |
| """将原始图像像素坐标缩放到 1024×1024 预处理空间""" | |
| scale = ve_cache["scale"] | |
| return [(float(x) * scale, float(y) * scale) for x, y in points_xy] | |
| def segment_point(self, ve_cache, point_xy, label=1): | |
| """ | |
| 使用点 prompt 分割. | |
| point_xy: (x, y) 原始图像像素坐标 | |
| label: 1=前景, 0=背景 | |
| 注意: ONNX mask_decoder 固定 n_points=2, 单点自动填充 not-a-point | |
| """ | |
| scaled = self._scale_points(ve_cache, [point_xy]) | |
| x, y = scaled[0] | |
| # 填充第二个点为 not-a-point (label=-1) | |
| points = np.array([[[[x, y], [0.0, 0.0]]]], dtype=np.float32) # (1, 1, 2, 2) | |
| labels = np.array([[[label, -1]]], dtype=np.int64) # (1, 1, 2) | |
| return self._run_decoder(ve_cache, points, labels) | |
| def segment_points(self, ve_cache, points_xy, labels_list): | |
| """ | |
| 使用多个点 prompt 分割. 每次取 2 个点运行, 合并 mask. | |
| points_xy: list of (x, y) — 原始图像像素坐标 | |
| labels_list: list of int (1=fg, 0=bg) | |
| """ | |
| scaled = self._scale_points(ve_cache, points_xy) | |
| if len(scaled) == 0: | |
| raise ValueError("No points provided") | |
| # 分组: 每 2 个点一组, 不足 2 个则填充 not-a-point | |
| combined_mask = None | |
| best_iou = 0.0 | |
| total_pixels = 0 | |
| for i in range(0, len(scaled), 2): | |
| chunk_pts = scaled[i:i+2] | |
| chunk_lbl = labels_list[i:i+2] | |
| if len(chunk_pts) == 1: | |
| chunk_pts.append((0.0, 0.0)) | |
| chunk_lbl.append(-1) | |
| points = np.array([[list(p) for p in chunk_pts]], dtype=np.float32) # (1, 1, 2, 2) | |
| labels = np.array([[chunk_lbl]], dtype=np.int64) # (1, 1, 2) | |
| result = self._run_decoder(ve_cache, points, labels) | |
| if combined_mask is None: | |
| combined_mask = result["mask"] | |
| else: | |
| combined_mask = np.maximum(combined_mask, result["mask"]) | |
| best_iou = max(best_iou, result["iou"]) | |
| total_pixels += result["pixels"] | |
| return { | |
| "mask": combined_mask, | |
| "iou": best_iou, | |
| "pixels": total_pixels, | |
| "raw_mask": None, | |
| } | |
| def segment_bbox_center(self, ve_cache, bbox_xyxy): | |
| """使用 bbox 中心点作为 prompt 分割 (bbox 为原始图像坐标)""" | |
| x1, y1, x2, y2 = bbox_xyxy | |
| cx = (x1 + x2) / 2.0 | |
| cy = (y1 + y2) / 2.0 | |
| return self.segment_point(ve_cache, (cx, cy), label=1) | |
| def _run_decoder(self, ve_cache, points, labels): | |
| """运行 Prompt Encoder + Mask Decoder""" | |
| sparse_emb, dense_emb = self.pe.encode_points(points, labels) | |
| md_out = self.md_sess.run(None, { | |
| "image_embeddings": ve_cache["image_embeddings"], | |
| "image_pos_enc": ve_cache["image_pos_enc"], | |
| "sparse_prompt_emb": sparse_emb, | |
| "dense_prompt_emb": dense_emb, | |
| "high_res_feat_0": ve_cache["high_res_feat_0"], | |
| "high_res_feat_1": ve_cache["high_res_feat_1"], | |
| }) | |
| iou_scores, pred_masks, obj_score = md_out | |
| # Pick best mask from 3 candidates (multimask_output=True) | |
| best_idx = np.argmax(iou_scores[0, 0]) | |
| best_iou = float(iou_scores[0, 0, best_idx]) | |
| mask = pred_masks[0, 0, best_idx] # (256, 256) | |
| # Upsample to original image size | |
| orig_h, orig_w = ve_cache["orig_h"], ve_cache["orig_w"] | |
| mask_resized = cv2.resize(mask, (orig_w, orig_h), interpolation=cv2.INTER_LINEAR) | |
| binary_mask = (mask_resized > 0.0).astype(np.uint8) | |
| return { | |
| "mask": binary_mask, | |
| "iou": best_iou, | |
| "pixels": int(binary_mask.sum()), | |
| "raw_mask": mask, | |
| } | |
| # ============================================================================ | |
| # 可视化 | |
| # ============================================================================ | |
| def overlay_mask(image_rgb, mask, color=(0, 255, 0), alpha=0.5): | |
| """将 mask 叠加到图像上""" | |
| overlay = image_rgb.copy().astype(np.float32) | |
| color_arr = np.array(color, dtype=np.float32) | |
| overlay[mask > 0] = overlay[mask > 0] * (1 - alpha) + color_arr * alpha | |
| return np.clip(overlay, 0, 255).astype(np.uint8) | |
| # ============================================================================ | |
| # Main | |
| # ============================================================================ | |
| def main(): | |
| parser = argparse.ArgumentParser(description="SAM2 ONNX 独立分割推理") | |
| parser.add_argument("--image", required=True, help="输入图像路径") | |
| parser.add_argument("--point", nargs=2, type=float, action="append", | |
| metavar=("X", "Y"), help="点 prompt (可多次指定)") | |
| parser.add_argument("--bbox", nargs=4, type=float, | |
| metavar=("X1", "Y1", "X2", "Y2"), help="bbox (使用中心点)") | |
| parser.add_argument("--save", default=None, help="保存结果图像路径") | |
| parser.add_argument("--save-mask", default=None, help="保存纯 mask 图像路径") | |
| parser.add_argument("--providers", nargs="+", default=["CPUExecutionProvider"]) | |
| args = parser.parse_args() | |
| # 加载图像 | |
| image_bgr = cv2.imread(args.image) | |
| if image_bgr is None: | |
| raise FileNotFoundError(f"Image not found: {args.image}") | |
| image_rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB) | |
| h, w = image_rgb.shape[:2] | |
| # 初始化分割器 | |
| seg = SAM2Segmenter(args.providers) | |
| # 编码图像 (VE — 只需一次) | |
| print(f"Encoding image ({w}x{h})...") | |
| ve_cache = seg.encode_image(image_rgb) | |
| # 分割 | |
| if args.point: | |
| for i, (x, y) in enumerate(args.point): | |
| result = seg.segment_point(ve_cache, (x, y), label=1) | |
| print(f" Point ({x}, {y}): IoU={result['iou']:.4f}, pixels={result['pixels']}") | |
| if i == 0: | |
| combined_mask = result["mask"] | |
| else: | |
| combined_mask = np.maximum(combined_mask, result["mask"]) | |
| elif args.bbox: | |
| result = seg.segment_bbox_center(ve_cache, args.bbox) | |
| print(f" BBox center: IoU={result['iou']:.4f}, pixels={result['pixels']}") | |
| combined_mask = result["mask"] | |
| else: | |
| print("Error: specify --point or --bbox") | |
| sys.exit(1) | |
| # 保存结果 | |
| if args.save: | |
| result_img = overlay_mask(image_rgb, combined_mask) | |
| cv2.imwrite(args.save, cv2.cvtColor(result_img, cv2.COLOR_RGB2BGR)) | |
| print(f"Saved result to: {args.save}") | |
| if args.save_mask: | |
| cv2.imwrite(args.save_mask, combined_mask * 255) | |
| print(f"Saved mask to: {args.save_mask}") | |
| if __name__ == "__main__": | |
| main() | |