Initialisation du repository de Beta

This commit is contained in:
Mathis 2026-02-06 22:23:20 +01:00
commit 14985f6dbb
9469 changed files with 1903273 additions and 0 deletions

View file

@ -0,0 +1,12 @@
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import os.path
import sys
sys.path.append(os.path.dirname(__file__))
transformers_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
if transformers_dir not in sys.path:
sys.path.append(transformers_dir)

View file

@ -0,0 +1,638 @@
# -------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
"""
Benchmark performance of SAM2 encoder with ORT or PyTorch. See benchmark_sam2.sh for usage.
"""
import argparse
import csv
import statistics
import time
from collections.abc import Mapping
from datetime import datetime
import torch
from image_decoder import SAM2ImageDecoder
from image_encoder import SAM2ImageEncoder
from sam2_utils import decoder_shape_dict, encoder_shape_dict, load_sam2_model
from onnxruntime import InferenceSession, SessionOptions, get_available_providers
from onnxruntime.transformers.io_binding_helper import CudaSession
class TestConfig:
def __init__(
self,
model_type: str,
onnx_path: str,
sam2_dir: str,
device: torch.device,
component: str = "image_encoder",
provider="CPUExecutionProvider",
torch_compile_mode="max-autotune",
batch_size: int = 1,
height: int = 1024,
width: int = 1024,
num_labels: int = 1,
num_points: int = 1,
num_masks: int = 1,
multi_mask_output: bool = False,
use_tf32: bool = True,
enable_cuda_graph: bool = False,
dtype=torch.float32,
prefer_nhwc: bool = False,
warm_up: int = 5,
enable_nvtx_profile: bool = False,
enable_ort_profile: bool = False,
enable_torch_profile: bool = False,
repeats: int = 1000,
verbose: bool = False,
):
assert model_type in ["sam2_hiera_tiny", "sam2_hiera_small", "sam2_hiera_large", "sam2_hiera_base_plus"]
assert height >= 160 and height <= 4096
assert width >= 160 and width <= 4096
self.model_type = model_type
self.onnx_path = onnx_path
self.sam2_dir = sam2_dir
self.component = component
self.provider = provider
self.torch_compile_mode = torch_compile_mode
self.batch_size = batch_size
self.height = height
self.width = width
self.num_labels = num_labels
self.num_points = num_points
self.num_masks = num_masks
self.multi_mask_output = multi_mask_output
self.device = device
self.use_tf32 = use_tf32
self.enable_cuda_graph = enable_cuda_graph
self.dtype = dtype
self.prefer_nhwc = prefer_nhwc
self.warm_up = warm_up
self.enable_nvtx_profile = enable_nvtx_profile
self.enable_ort_profile = enable_ort_profile
self.enable_torch_profile = enable_torch_profile
self.repeats = repeats
self.verbose = verbose
if self.component == "image_encoder":
assert self.height == 1024 and self.width == 1024, "Only image size 1024x1024 is allowed for image encoder."
def __repr__(self):
return f"{vars(self)}"
def shape_dict(self) -> Mapping[str, list[int]]:
if self.component == "image_encoder":
return encoder_shape_dict(self.batch_size, self.height, self.width)
else:
return decoder_shape_dict(self.height, self.width, self.num_labels, self.num_points, self.num_masks)
def random_inputs(self) -> Mapping[str, torch.Tensor]:
dtype = self.dtype
if self.component == "image_encoder":
return {"image": torch.randn(self.batch_size, 3, self.height, self.width, dtype=dtype, device=self.device)}
else:
return {
"image_features_0": torch.rand(1, 32, 256, 256, dtype=dtype, device=self.device),
"image_features_1": torch.rand(1, 64, 128, 128, dtype=dtype, device=self.device),
"image_embeddings": torch.rand(1, 256, 64, 64, dtype=dtype, device=self.device),
"point_coords": torch.randint(
0, 1024, (self.num_labels, self.num_points, 2), dtype=dtype, device=self.device
),
"point_labels": torch.randint(
0, 1, (self.num_labels, self.num_points), dtype=torch.int32, device=self.device
),
"input_masks": torch.zeros(self.num_labels, 1, 256, 256, dtype=dtype, device=self.device),
"has_input_masks": torch.ones(self.num_labels, dtype=dtype, device=self.device),
"original_image_size": torch.tensor([self.height, self.width], dtype=torch.int32, device=self.device),
}
def create_ort_session(config: TestConfig, session_options=None) -> InferenceSession:
if config.verbose:
print(f"create session for {vars(config)}")
if config.provider == "CUDAExecutionProvider":
device_id = torch.cuda.current_device() if isinstance(config.device, str) else config.device.index
provider_options = CudaSession.get_cuda_provider_options(device_id, config.enable_cuda_graph)
provider_options["use_tf32"] = int(config.use_tf32)
if config.prefer_nhwc:
provider_options["prefer_nhwc"] = 1
providers = [(config.provider, provider_options), "CPUExecutionProvider"]
else:
providers = ["CPUExecutionProvider"]
ort_session = InferenceSession(config.onnx_path, session_options, providers=providers)
return ort_session
def create_session(config: TestConfig, session_options=None) -> CudaSession:
ort_session = create_ort_session(config, session_options)
cuda_session = CudaSession(ort_session, config.device, config.enable_cuda_graph)
cuda_session.allocate_buffers(config.shape_dict())
return cuda_session
class OrtTestSession:
"""A wrapper of ORT session to test relevance and performance."""
def __init__(self, config: TestConfig, session_options=None):
self.ort_session = create_session(config, session_options)
self.feed_dict = config.random_inputs()
def infer(self):
return self.ort_session.infer(self.feed_dict)
def measure_latency(cuda_session: CudaSession, input_dict):
start = time.time()
_ = cuda_session.infer(input_dict)
end = time.time()
return end - start
def run_torch(config: TestConfig):
device_type = config.device.type
is_cuda = device_type == "cuda"
# Turn on TF32 for Ampere GPUs which could help when data type is float32.
if is_cuda and torch.cuda.get_device_properties(0).major >= 8 and config.use_tf32:
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.allow_tf32 = True
enabled_auto_cast = is_cuda and config.dtype != torch.float32
ort_inputs = config.random_inputs()
with torch.inference_mode(), torch.autocast(device_type=device_type, dtype=config.dtype, enabled=enabled_auto_cast):
sam2_model = load_sam2_model(config.sam2_dir, config.model_type, device=config.device)
if config.component == "image_encoder":
if is_cuda and config.torch_compile_mode != "none":
sam2_model.image_encoder.forward = torch.compile(
sam2_model.image_encoder.forward,
mode=config.torch_compile_mode, # "reduce-overhead" if you want to reduce latency of first run.
fullgraph=True,
dynamic=False,
)
image_shape = config.shape_dict()["image"]
img = torch.randn(image_shape).to(device=config.device, dtype=config.dtype)
sam2_encoder = SAM2ImageEncoder(sam2_model)
if is_cuda and config.torch_compile_mode != "none":
print(f"Running warm up. It will take a while since torch compile mode is {config.torch_compile_mode}.")
for _ in range(config.warm_up):
_image_features_0, _image_features_1, _image_embeddings = sam2_encoder(img)
if is_cuda and config.enable_nvtx_profile:
import nvtx # noqa: PLC0415
from cuda import cudart # noqa: PLC0415
cudart.cudaProfilerStart()
print("Start nvtx profiling on encoder ...")
with nvtx.annotate("one_run"):
sam2_encoder(img, enable_nvtx_profile=True)
cudart.cudaProfilerStop()
if is_cuda and config.enable_torch_profile:
with torch.profiler.profile(
activities=[torch.profiler.ProfilerActivity.CPU, torch.profiler.ProfilerActivity.CUDA],
record_shapes=True,
) as prof:
print("Start torch profiling on encoder ...")
with torch.profiler.record_function("encoder"):
sam2_encoder(img)
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
prof.export_chrome_trace("torch_image_encoder.json")
if config.repeats == 0:
return
print(f"Start {config.repeats} runs of performance tests...")
start = time.time()
for _ in range(config.repeats):
_image_features_0, _image_features_1, _image_embeddings = sam2_encoder(img)
if is_cuda:
torch.cuda.synchronize()
else:
torch_inputs = (
ort_inputs["image_features_0"],
ort_inputs["image_features_1"],
ort_inputs["image_embeddings"],
ort_inputs["point_coords"],
ort_inputs["point_labels"],
ort_inputs["input_masks"],
ort_inputs["has_input_masks"],
ort_inputs["original_image_size"],
)
sam2_decoder = SAM2ImageDecoder(
sam2_model,
multimask_output=config.multi_mask_output,
)
if is_cuda and config.torch_compile_mode != "none":
sam2_decoder.forward = torch.compile(
sam2_decoder.forward,
mode=config.torch_compile_mode,
fullgraph=True,
dynamic=False,
)
# warm up
for _ in range(config.warm_up):
_masks, _iou_predictions, _low_res_masks = sam2_decoder(*torch_inputs)
if is_cuda and config.enable_nvtx_profile:
import nvtx # noqa: PLC0415
from cuda import cudart # noqa: PLC0415
cudart.cudaProfilerStart()
print("Start nvtx profiling on decoder...")
with nvtx.annotate("one_run"):
sam2_decoder(*torch_inputs, enable_nvtx_profile=True)
cudart.cudaProfilerStop()
if is_cuda and config.enable_torch_profile:
with torch.profiler.profile(
activities=[torch.profiler.ProfilerActivity.CPU, torch.profiler.ProfilerActivity.CUDA],
record_shapes=True,
) as prof:
print("Start torch profiling on decoder ...")
with torch.profiler.record_function("decoder"):
sam2_decoder(*torch_inputs)
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
prof.export_chrome_trace("torch_image_decoder.json")
if config.repeats == 0:
return
print(f"Start {config.repeats} runs of performance tests...")
start = time.time()
for _ in range(config.repeats):
_masks, _iou_predictions, _low_res_masks = sam2_decoder(*torch_inputs)
if is_cuda:
torch.cuda.synchronize()
end = time.time()
return (end - start) / config.repeats
def run_test(
args: argparse.Namespace,
csv_writer: csv.DictWriter | None = None,
):
use_gpu: bool = args.use_gpu
enable_cuda_graph: bool = args.use_cuda_graph
repeats: int = args.repeats
if use_gpu:
device_id = torch.cuda.current_device()
device = torch.device("cuda", device_id)
provider = "CUDAExecutionProvider"
else:
device_id = 0
device = torch.device("cpu")
enable_cuda_graph = False
provider = "CPUExecutionProvider"
dtypes = {"fp32": torch.float32, "fp16": torch.float16, "bf16": torch.bfloat16}
config = TestConfig(
model_type=args.model_type,
onnx_path=args.onnx_path,
sam2_dir=args.sam2_dir,
component=args.component,
provider=provider,
batch_size=args.batch_size,
height=args.height,
width=args.width,
device=device,
use_tf32=True,
enable_cuda_graph=enable_cuda_graph,
dtype=dtypes[args.dtype],
prefer_nhwc=args.prefer_nhwc,
repeats=args.repeats,
warm_up=args.warm_up,
enable_nvtx_profile=args.enable_nvtx_profile,
enable_ort_profile=args.enable_ort_profile,
enable_torch_profile=args.enable_torch_profile,
torch_compile_mode=args.torch_compile_mode,
verbose=False,
)
if args.engine == "ort":
sess_options = SessionOptions()
sess_options.intra_op_num_threads = args.intra_op_num_threads
if config.enable_ort_profile:
sess_options.enable_profiling = True
sess_options.log_severity_level = 4
sess_options.log_verbosity_level = 0
session = create_session(config, sess_options)
input_dict = config.random_inputs()
# warm up session
try:
for _ in range(config.warm_up):
_ = measure_latency(session, input_dict)
except Exception as e:
print(f"Failed to run {config=}. Exception: {e}")
return
if config.enable_nvtx_profile:
import nvtx # noqa: PLC0415
from cuda import cudart # noqa: PLC0415
cudart.cudaProfilerStart()
with nvtx.annotate("one_run"):
_ = session.infer(input_dict)
cudart.cudaProfilerStop()
if config.enable_ort_profile:
session.ort_session.end_profiling()
if repeats == 0:
return
latency_list = []
for _ in range(repeats):
latency = measure_latency(session, input_dict)
latency_list.append(latency)
average_latency = statistics.mean(latency_list)
del session
else: # torch
with torch.no_grad():
try:
average_latency = run_torch(config)
except Exception as e:
print(f"Failed to run {config=}. Exception: {e}")
return
if repeats == 0:
return
engine = args.engine + ":" + ("cuda" if use_gpu else "cpu")
row = {
"model_type": args.model_type,
"component": args.component,
"dtype": args.dtype,
"use_gpu": use_gpu,
"enable_cuda_graph": enable_cuda_graph,
"prefer_nhwc": config.prefer_nhwc,
"use_tf32": config.use_tf32,
"batch_size": args.batch_size,
"height": args.height,
"width": args.width,
"multi_mask_output": args.multimask_output,
"num_labels": config.num_labels,
"num_points": config.num_points,
"num_masks": config.num_masks,
"intra_op_num_threads": args.intra_op_num_threads,
"warm_up": config.warm_up,
"repeats": repeats,
"enable_nvtx_profile": args.enable_nvtx_profile,
"torch_compile_mode": args.torch_compile_mode,
"engine": engine,
"average_latency": average_latency,
}
if csv_writer is not None:
csv_writer.writerow(row)
print(f"{vars(config)}")
print(f"{row}")
def run_perf_test(args):
features = "gpu" if args.use_gpu else "cpu"
csv_filename = "benchmark_sam_{}_{}_{}.csv".format(
features,
args.engine,
datetime.now().strftime("%Y%m%d-%H%M%S"),
)
with open(csv_filename, mode="a", newline="") as csv_file:
column_names = [
"model_type",
"component",
"dtype",
"use_gpu",
"enable_cuda_graph",
"prefer_nhwc",
"use_tf32",
"batch_size",
"height",
"width",
"multi_mask_output",
"num_labels",
"num_points",
"num_masks",
"intra_op_num_threads",
"warm_up",
"repeats",
"enable_nvtx_profile",
"torch_compile_mode",
"engine",
"average_latency",
]
csv_writer = csv.DictWriter(csv_file, fieldnames=column_names)
csv_writer.writeheader()
run_test(args, csv_writer)
def _parse_arguments():
parser = argparse.ArgumentParser(description="Benchmark SMA2 for ONNX Runtime and PyTorch.")
parser.add_argument(
"--component",
required=False,
choices=["image_encoder", "image_decoder"],
default="image_encoder",
help="component to benchmark. Choices are image_encoder and image_decoder.",
)
parser.add_argument(
"--dtype", required=False, choices=["fp32", "fp16", "bf16"], default="fp32", help="Data type for inference."
)
parser.add_argument(
"--use_gpu",
required=False,
action="store_true",
help="Use GPU for inference.",
)
parser.set_defaults(use_gpu=False)
parser.add_argument(
"--use_cuda_graph",
required=False,
action="store_true",
help="Use cuda graph in onnxruntime.",
)
parser.set_defaults(use_cuda_graph=False)
parser.add_argument(
"--intra_op_num_threads",
required=False,
type=int,
choices=[0, 1, 2, 4, 8, 16],
default=0,
help="intra_op_num_threads for onnxruntime. ",
)
parser.add_argument(
"--batch_size",
required=False,
type=int,
default=1,
help="batch size",
)
parser.add_argument(
"--height",
required=False,
type=int,
default=1024,
help="image height",
)
parser.add_argument(
"--width",
required=False,
type=int,
default=1024,
help="image width",
)
parser.add_argument(
"--repeats",
required=False,
type=int,
default=1000,
help="number of repeats for performance test. Default is 1000.",
)
parser.add_argument(
"--warm_up",
required=False,
type=int,
default=5,
help="number of runs for warm up. Default is 5.",
)
parser.add_argument(
"--engine",
required=False,
type=str,
default="ort",
choices=["ort", "torch"],
help="engine for inference",
)
parser.add_argument(
"--multimask_output",
required=False,
default=False,
action="store_true",
help="Export mask_decoder or image_decoder with multimask_output",
)
parser.add_argument(
"--prefer_nhwc",
required=False,
default=False,
action="store_true",
help="Use prefer_nhwc=1 provider option for CUDAExecutionProvider",
)
parser.add_argument(
"--enable_nvtx_profile",
required=False,
default=False,
action="store_true",
help="Enable nvtx profiling. It will add an extra run for profiling before performance test.",
)
parser.add_argument(
"--enable_ort_profile",
required=False,
default=False,
action="store_true",
help="Enable ORT profiling.",
)
parser.add_argument(
"--enable_torch_profile",
required=False,
default=False,
action="store_true",
help="Enable PyTorch profiling. It will add an extra run for profiling before performance test.",
)
parser.add_argument(
"--model_type",
required=False,
type=str,
default="sam2_hiera_large",
choices=["sam2_hiera_tiny", "sam2_hiera_small", "sam2_hiera_large", "sam2_hiera_base_plus"],
help="sam2 model name",
)
parser.add_argument(
"--sam2_dir",
required=False,
type=str,
default="./segment-anything-2",
help="The directory of segment-anything-2 git root directory",
)
parser.add_argument(
"--onnx_path",
required=False,
type=str,
default="./sam2_onnx_models/sam2_hiera_large_image_encoder.onnx",
help="path of onnx model",
)
parser.add_argument(
"--torch_compile_mode",
required=False,
type=str,
default=None,
choices=["reduce-overhead", "max-autotune", "max-autotune-no-cudagraphs", "none"],
help="torch compile mode. none will disable torch compile.",
)
args = parser.parse_args()
return args
if __name__ == "__main__":
args = _parse_arguments()
print(f"arguments:{args}")
if args.torch_compile_mode is None:
# image decoder will fail with compile modes other than "none".
args.torch_compile_mode = "max-autotune" if args.component == "image_encoder" else "none"
if args.use_gpu:
assert torch.cuda.is_available()
if args.engine == "ort":
assert "CUDAExecutionProvider" in get_available_providers()
args.enable_torch_profile = False
else:
# Only support cuda profiling for now.
assert not args.enable_nvtx_profile
assert not args.enable_torch_profile
if args.enable_nvtx_profile or args.enable_torch_profile:
run_test(args)
else:
run_perf_test(args)

View file

@ -0,0 +1,270 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import argparse
import os
import pathlib
import sys
import torch
from image_decoder import export_decoder_onnx, test_decoder_onnx
from image_encoder import export_image_encoder_onnx, test_image_encoder_onnx
from mask_decoder import export_mask_decoder_onnx, test_mask_decoder_onnx
from prompt_encoder import export_prompt_encoder_onnx, test_prompt_encoder_onnx
from sam2_demo import run_demo, show_all_images
from sam2_utils import load_sam2_model, sam2_onnx_path, setup_logger
def parse_arguments():
parser = argparse.ArgumentParser(description="Export SAM2 models to ONNX")
parser.add_argument(
"--model_type",
required=False,
type=str,
choices=["sam2_hiera_tiny", "sam2_hiera_small", "sam2_hiera_large", "sam2_hiera_base_plus"],
default="sam2_hiera_large",
help="The model type to export",
)
parser.add_argument(
"--components",
required=False,
nargs="+",
choices=["image_encoder", "mask_decoder", "prompt_encoder", "image_decoder"],
default=["image_encoder", "image_decoder"],
help="Type of ONNX models to export. "
"Note that image_decoder is a combination of prompt_encoder and mask_decoder",
)
parser.add_argument(
"--output_dir",
type=str,
help="The output directory for the ONNX models",
default="sam2_onnx_models",
)
parser.add_argument(
"--dynamic_batch_axes",
required=False,
default=False,
action="store_true",
help="Export image_encoder with dynamic batch axes",
)
parser.add_argument(
"--multimask_output",
required=False,
default=False,
action="store_true",
help="Export mask_decoder or image_decoder with multimask_output",
)
parser.add_argument(
"--disable_dynamic_multimask_via_stability",
required=False,
action="store_true",
help="Disable mask_decoder dynamic_multimask_via_stability, and output first mask only."
"This option will be ignored when multimask_output is True",
)
parser.add_argument(
"--sam2_dir",
required=False,
type=str,
default="./segment-anything-2",
help="The directory of segment-anything-2 git repository",
)
parser.add_argument(
"--overwrite",
required=False,
default=False,
action="store_true",
help="Overwrite onnx model file if exists.",
)
parser.add_argument(
"--demo",
required=False,
default=False,
action="store_true",
help="Run demo with the exported ONNX models.",
)
parser.add_argument(
"--optimize",
required=False,
default=False,
action="store_true",
help="Optimize onnx models",
)
parser.add_argument(
"--dtype", required=False, choices=["fp32", "fp16"], default="fp32", help="Data type for inference."
)
parser.add_argument(
"--use_gpu",
required=False,
default=False,
action="store_true",
help="Optimize onnx models for GPU",
)
parser.add_argument(
"--dynamo",
required=False,
default=False,
action="store_true",
help="Use dynamo for exporting onnx model. Only image_encoder supports dynamo right now.",
)
parser.add_argument(
"--verbose",
required=False,
default=False,
action="store_true",
help="Print verbose information",
)
args = parser.parse_args()
return args
def optimize_sam2_model(onnx_model_path, optimized_model_path, float16: bool, use_gpu: bool):
print(f"Optimizing {onnx_model_path} to {optimized_model_path} with float16={float16} and use_gpu={use_gpu}...")
# Import from source directory.
transformers_dir = os.path.normpath(os.path.join(os.path.dirname(__file__), "..", ".."))
if transformers_dir not in sys.path:
sys.path.insert(0, transformers_dir)
from optimizer import optimize_model # noqa: PLC0415
optimized_model = optimize_model(onnx_model_path, model_type="sam2", opt_level=1, use_gpu=use_gpu)
if float16:
optimized_model.convert_float_to_float16(keep_io_types=False)
optimized_model.save_model_to_file(optimized_model_path)
def main():
args = parse_arguments()
sam2_model = load_sam2_model(args.sam2_dir, args.model_type, device="cpu")
pathlib.Path(args.output_dir).mkdir(parents=True, exist_ok=True)
for component in args.components:
onnx_model_path = sam2_onnx_path(args.output_dir, args.model_type, component, args.multimask_output)
if component == "image_encoder":
if args.overwrite or not os.path.exists(onnx_model_path):
export_image_encoder_onnx(
sam2_model, onnx_model_path, args.dynamic_batch_axes, args.verbose, args.dynamo
)
test_image_encoder_onnx(sam2_model, onnx_model_path, dynamic_batch_axes=args.dynamic_batch_axes)
elif component == "mask_decoder":
if args.overwrite or not os.path.exists(onnx_model_path):
export_mask_decoder_onnx(
sam2_model,
onnx_model_path,
args.multimask_output,
not args.disable_dynamic_multimask_via_stability,
args.verbose,
)
test_mask_decoder_onnx(
sam2_model,
onnx_model_path,
args.multimask_output,
not args.disable_dynamic_multimask_via_stability,
)
elif component == "prompt_encoder":
if args.overwrite or not os.path.exists(onnx_model_path):
export_prompt_encoder_onnx(sam2_model, onnx_model_path)
test_prompt_encoder_onnx(sam2_model, onnx_model_path)
else:
assert component == "image_decoder"
if args.overwrite or not os.path.exists(onnx_model_path):
export_decoder_onnx(sam2_model, onnx_model_path, args.multimask_output)
test_decoder_onnx(sam2_model, onnx_model_path, args.multimask_output)
suffix = ""
convert_to_fp16 = args.dtype == "fp16"
if args.optimize:
suffix = f"_{args.dtype}_" + ("gpu" if args.use_gpu else "cpu")
for component in args.components:
onnx_model_path = sam2_onnx_path(args.output_dir, args.model_type, component, args.multimask_output)
optimized_model_path = sam2_onnx_path(
args.output_dir, args.model_type, component, args.multimask_output, suffix
)
optimize_sam2_model(onnx_model_path, optimized_model_path, convert_to_fp16, args.use_gpu)
if args.demo:
# Export required ONNX models for demo if not already exported.
image_encoder_onnx_path = sam2_onnx_path(
args.output_dir, args.model_type, "image_encoder", args.multimask_output
)
if not os.path.exists(image_encoder_onnx_path):
export_image_encoder_onnx(sam2_model, image_encoder_onnx_path, args.dynamic_batch_axes, args.verbose)
image_decoder_onnx_path = sam2_onnx_path(args.output_dir, args.model_type, "image_decoder", False)
if not os.path.exists(image_decoder_onnx_path):
export_decoder_onnx(sam2_model, image_decoder_onnx_path, False)
image_decoder_multi_onnx_path = sam2_onnx_path(args.output_dir, args.model_type, "image_decoder", True)
if not os.path.exists(image_decoder_multi_onnx_path):
export_decoder_onnx(sam2_model, image_decoder_multi_onnx_path, True)
dtype = torch.float32 if args.dtype == "fp32" else torch.float16
if suffix:
optimized_image_encoder_onnx_path = image_encoder_onnx_path.replace(".onnx", f"{suffix}.onnx")
if not os.path.exists(optimized_image_encoder_onnx_path):
optimize_sam2_model(
image_encoder_onnx_path, optimized_image_encoder_onnx_path, convert_to_fp16, args.use_gpu
)
optimized_image_decoder_onnx_path = image_decoder_onnx_path.replace(".onnx", f"{suffix}.onnx")
if not os.path.exists(optimized_image_decoder_onnx_path):
optimize_sam2_model(
image_decoder_onnx_path, optimized_image_decoder_onnx_path, convert_to_fp16, args.use_gpu
)
optimized_image_decoder_multi_onnx_path = image_decoder_multi_onnx_path.replace(".onnx", f"{suffix}.onnx")
if not os.path.exists(optimized_image_decoder_multi_onnx_path):
optimize_sam2_model(
image_decoder_multi_onnx_path,
optimized_image_decoder_multi_onnx_path,
convert_to_fp16,
args.use_gpu,
)
# Use optimized models to run demo.
image_encoder_onnx_path = optimized_image_encoder_onnx_path
image_decoder_onnx_path = optimized_image_decoder_onnx_path
image_decoder_multi_onnx_path = optimized_image_decoder_multi_onnx_path
ort_image_files = run_demo(
args.sam2_dir,
args.model_type,
engine="ort",
dtype=dtype,
image_encoder_onnx_path=image_encoder_onnx_path,
image_decoder_onnx_path=image_decoder_onnx_path,
image_decoder_multi_onnx_path=image_decoder_multi_onnx_path,
use_gpu=args.use_gpu,
)
print("demo output files for ONNX Runtime:", ort_image_files)
# Get results from torch engine to compare.
torch_image_files = run_demo(args.sam2_dir, args.model_type, engine="torch", dtype=dtype, use_gpu=args.use_gpu)
print("demo output files for PyTorch:", torch_image_files)
show_all_images(ort_image_files, torch_image_files, suffix)
print(f"Combined demo output: sam2_demo{suffix}.png")
if __name__ == "__main__":
setup_logger(verbose=False)
with torch.no_grad():
main()

View file

@ -0,0 +1,272 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import warnings
import torch
import torch.nn.functional as F
from image_encoder import SAM2ImageEncoder, random_sam2_input_image
from mask_decoder import SAM2MaskDecoder
from prompt_encoder import SAM2PromptEncoder
from sam2.modeling.sam2_base import SAM2Base
from sam2_utils import compare_tensors_with_tolerance
from torch import nn
logger = logging.getLogger(__name__)
class SAM2ImageDecoder(nn.Module):
def __init__(
self,
sam_model: SAM2Base,
multimask_output: bool,
dynamic_multimask_via_stability: bool = True,
return_logits: bool = False,
mask_threshold: float = 0.0,
) -> None:
super().__init__()
self.prompt_encoder = SAM2PromptEncoder(sam_model)
self.mask_decoder = SAM2MaskDecoder(sam_model, multimask_output, dynamic_multimask_via_stability)
self.return_logits = return_logits
self.mask_threshold = mask_threshold
@torch.no_grad()
def forward(
self,
image_features_0: torch.Tensor,
image_features_1: torch.Tensor,
image_embeddings: torch.Tensor,
point_coords: torch.Tensor,
point_labels: torch.Tensor,
input_masks: torch.Tensor,
has_input_masks: torch.Tensor,
original_image_size: torch.Tensor,
enable_nvtx_profile: bool = False,
):
"""
Decode masks from image features and prompts. Batched images are not supported. H=W=1024.
Args:
image_features_0 (torch.Tensor): [1, 32, H/4, W/4]. high resolution features of level 0 from image encoder.
image_features_1 (torch.Tensor): [1, 64, H/8, W/8]. high resolution features of level 1 from image encoder.
image_embeddings (torch.Tensor): [1, 256, H/16, W/16]. image embedding from image encoder.
point_coords (torch.Tensor): [L, P, 2] shape and float32 dtype and contains the absolute pixel
coordinate in (x, y) format of the P input points in image of size 1024x1024.
point_labels (torch.Tensor): shape [L, P] and int32 dtype, where 1 means
positive (foreground), 0 means negative (background), -1 means padding,
2 (box left upper corner), 3 (box right bottom corner).
input_masks (torch.Tensor): [L, 1, H/4, W/4]. Low resolution mask input to the model.
Typically coming from a previous iteration.
has_input_masks (torch.Tensor): [L]. 1.0 if input_masks is used, 0.0 otherwise.
original_image_size(torch.Tensor): [2]. original image size H_o, W_o.
enable_nvtx_profile (bool): enable NVTX profiling.
Returns:
masks (torch.Tensor): [1, M, H_o, W_o] where M=3 or 1. Masks of original image size.
iou_predictions (torch.Tensor): [1, M]. scores for M masks.
low_res_masks (torch.Tensor, optional): [1, M, H/4, W/4]. low resolution masks.
"""
nvtx_helper = None
if enable_nvtx_profile:
from nvtx_helper import NvtxHelper # noqa: PLC0415
nvtx_helper = NvtxHelper(["prompt_encoder", "mask_decoder", "post_process"])
if nvtx_helper is not None:
nvtx_helper.start_profile("prompt_encoder", color="blue")
sparse_embeddings, dense_embeddings, image_pe = self.prompt_encoder(
point_coords, point_labels, input_masks, has_input_masks
)
if nvtx_helper is not None:
nvtx_helper.stop_profile("prompt_encoder")
nvtx_helper.start_profile("mask_decoder", color="red")
low_res_masks, iou_predictions = self.mask_decoder(
image_features_0, image_features_1, image_embeddings, image_pe, sparse_embeddings, dense_embeddings
)
if nvtx_helper is not None:
nvtx_helper.stop_profile("mask_decoder")
nvtx_helper.start_profile("post_process", color="green")
# Interpolate the low resolution masks back to the original image size.
masks = F.interpolate(
low_res_masks,
(original_image_size[0], original_image_size[1]),
mode="bilinear",
align_corners=False, # Note that align_corners=True has less mismatches during comparing ORT and PyTorch.
)
low_res_masks = torch.clamp(low_res_masks, -32.0, 32.0)
if not self.return_logits:
masks = masks > self.mask_threshold
if nvtx_helper is not None:
nvtx_helper.stop_profile("post_process")
nvtx_helper.print_latency()
return masks, iou_predictions, low_res_masks
def export_decoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
multimask_output: bool = False,
verbose: bool = False,
):
batch_size = 1
image = random_sam2_input_image(batch_size)
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image)
logger.info("image_features_0.shape: %s", image_features_0.shape)
logger.info("image_features_1.shape: %s", image_features_1.shape)
logger.info("image_embeddings.shape: %s", image_embeddings.shape)
sam2_decoder = SAM2ImageDecoder(
sam2_model,
multimask_output=multimask_output,
dynamic_multimask_via_stability=True,
).cpu()
num_labels = 2
num_points = 3
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.int32)
input_masks = torch.zeros(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.ones(1, dtype=torch.float)
original_image_size = torch.tensor([1200, 1800], dtype=torch.int32)
example_inputs = (
image_features_0,
image_features_1,
image_embeddings,
point_coords,
point_labels,
input_masks,
has_input_masks,
original_image_size,
)
logger.info("point_coords.shape: %s", point_coords.shape)
logger.info("point_labels.shape: %s", point_labels.shape)
logger.info("input_masks.shape: %s", input_masks.shape)
logger.info("has_input_masks.shape: %s", has_input_masks.shape)
logger.info("original_image_size.shape: %s", original_image_size.shape)
if verbose:
masks, iou_predictions, low_res_masks = sam2_decoder(*example_inputs)
logger.info("masks.shape: %s", masks.shape)
logger.info("iou_predictions.shape: %s", iou_predictions.shape)
logger.info("low_res_masks.shape: %s", low_res_masks.shape)
input_names = [
"image_features_0",
"image_features_1",
"image_embeddings",
"point_coords",
"point_labels",
"input_masks",
"has_input_masks",
"original_image_size",
]
output_names = ["masks", "iou_predictions", "low_res_masks"]
dynamic_axes = {
"point_coords": {0: "num_labels", 1: "num_points"},
"point_labels": {0: "num_labels", 1: "num_points"},
"input_masks": {0: "num_labels"},
"has_input_masks": {0: "num_labels"},
"masks": {0: "num_labels", 2: "original_image_height", 3: "original_image_width"},
"low_res_masks": {0: "num_labels"},
"iou_predictions": {0: "num_labels"},
}
with warnings.catch_warnings():
if not verbose:
warnings.filterwarnings("ignore", category=torch.jit.TracerWarning)
warnings.filterwarnings("ignore", category=UserWarning)
torch.onnx.export(
sam2_decoder,
example_inputs,
onnx_model_path,
export_params=True,
opset_version=16,
do_constant_folding=True,
input_names=input_names,
output_names=output_names,
dynamic_axes=dynamic_axes,
)
logger.info("decoder onnx model saved to %s", onnx_model_path)
def test_decoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
multimask_output=False,
):
batch_size = 1
image = random_sam2_input_image(batch_size)
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image)
sam2_image_decoder = SAM2ImageDecoder(
sam2_model,
multimask_output=multimask_output,
dynamic_multimask_via_stability=True,
).cpu()
num_labels = 1
num_points = 5
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.int32)
input_masks = torch.zeros(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.zeros(1, dtype=torch.float)
original_image_size = torch.tensor([1500, 1500], dtype=torch.int32)
example_inputs = (
image_features_0,
image_features_1,
image_embeddings,
point_coords,
point_labels,
input_masks,
has_input_masks,
original_image_size,
)
masks, iou_predictions, low_res_masks = sam2_image_decoder(*example_inputs)
import onnxruntime # noqa: PLC0415
ort_session = onnxruntime.InferenceSession(onnx_model_path, providers=["CPUExecutionProvider"])
model_inputs = ort_session.get_inputs()
input_names = [model_inputs[i].name for i in range(len(model_inputs))]
logger.info("input_names: %s", input_names)
model_outputs = ort_session.get_outputs()
output_names = [model_outputs[i].name for i in range(len(model_outputs))]
logger.info("output_names: %s", output_names)
inputs = {model_inputs[i].name: example_inputs[i].numpy() for i in range(len(model_inputs))}
outputs = ort_session.run(output_names, inputs)
for i, output_name in enumerate(output_names):
logger.info(f"{output_name}.shape: %s", outputs[i].shape)
ort_masks, ort_iou_predictions, ort_low_res_masks = outputs
if (
compare_tensors_with_tolerance("masks", masks.float(), torch.tensor(ort_masks).float())
and compare_tensors_with_tolerance("iou_predictions", iou_predictions, torch.tensor(ort_iou_predictions))
and compare_tensors_with_tolerance("low_res_masks", low_res_masks, torch.tensor(ort_low_res_masks))
):
print("onnx model has been verified:", onnx_model_path)
else:
print("onnx model verification failed:", onnx_model_path)

View file

@ -0,0 +1,236 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import warnings
import torch
from sam2.modeling.sam2_base import SAM2Base
from sam2_utils import compare_tensors_with_tolerance, random_sam2_input_image
from torch import nn
import onnxruntime
logger = logging.getLogger(__name__)
class SAM2ImageEncoder(nn.Module):
def __init__(self, sam_model: SAM2Base) -> None:
super().__init__()
self.model = sam_model
self.image_encoder = sam_model.image_encoder
self.no_mem_embed = sam_model.no_mem_embed
def forward(
self,
image: torch.Tensor,
enable_nvtx_profile: bool = False,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Encodes images into features.
Only supports H=W=1024. If you want to use different image sizes like 512x512,
see https://github.com/facebookresearch/segment-anything-2/issues/138.
Args:
image (torch.Tensor): images of shape [B, 3, H, W], B is batch size, H and W are height and width.
enable_nvtx_profile (bool): enable NVTX profiling.
Returns:
image_features_0: image features of shape [B, 32, H/4, W/4] - high resolution features of level 0
image_features_1: image features of shape [B, 64, H/8, W/8] - high resolution features of level 1
image_embeddings: image features of shape [B, 256, H/16, W/16] - 16 is the backbone_stride
"""
nvtx_helper = None
if enable_nvtx_profile:
from nvtx_helper import NvtxHelper # noqa: PLC0415
nvtx_helper = NvtxHelper(["image_encoder", "post_process"])
if nvtx_helper is not None:
nvtx_helper.start_profile("image_encoder")
backbone_out = self.image_encoder(image)
if nvtx_helper is not None:
nvtx_helper.stop_profile("image_encoder")
nvtx_helper.start_profile("post_process")
# precompute projected level 0 and level 1 features in SAM decoder
# to avoid running it again on every SAM click
backbone_out["backbone_fpn"][0] = self.model.sam_mask_decoder.conv_s0(backbone_out["backbone_fpn"][0])
backbone_out["backbone_fpn"][1] = self.model.sam_mask_decoder.conv_s1(backbone_out["backbone_fpn"][1])
# Prepare and flatten visual features.
feature_maps = backbone_out["backbone_fpn"][-self.model.num_feature_levels :]
vision_pos_embeds = backbone_out["vision_pos_enc"][-self.model.num_feature_levels :]
feat_sizes = [(x.shape[-2], x.shape[-1]) for x in vision_pos_embeds]
# flatten NxCxHxW to HWxNxC
# TODO: we should avoid this transpose since it will be transposed back to NCHW later.
vision_feats = [x.flatten(2).permute(2, 0, 1) for x in feature_maps]
vision_feats[-1] = vision_feats[-1] + self.no_mem_embed
feats = [
feat.permute(1, 2, 0).reshape(1, -1, *feat_size)
for feat, feat_size in zip(vision_feats[::-1], feat_sizes[::-1], strict=False)
][::-1]
if nvtx_helper is not None:
nvtx_helper.stop_profile("post_process")
nvtx_helper.print_latency()
return feats[0], feats[1], feats[2]
def export_image_encoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
dynamic_batch_axes: bool = False,
verbose: bool = False,
dynamo: bool = False,
clear_dynamo_metadata: bool = False,
):
image = random_sam2_input_image()
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image)
logger.info("image.shape: %s", image.shape)
logger.info("image_features_0.shape: %s", image_features_0.shape)
logger.info("image_features_1.shape: %s", image_features_1.shape)
logger.info("image_embeddings.shape: %s", image_embeddings.shape)
dynamic_axes = None
if dynamic_batch_axes:
dynamic_axes = {
"image": {0: "batch_size"},
"image_features_0": {0: "batch_size"},
"image_features_1": {0: "batch_size"},
"image_embeddings": {0: "batch_size"},
}
with warnings.catch_warnings():
if not verbose:
warnings.filterwarnings("ignore", category=torch.jit.TracerWarning)
warnings.filterwarnings("ignore", category=UserWarning)
if not dynamo:
torch.onnx.export(
sam2_encoder,
image,
onnx_model_path,
export_params=True,
opset_version=17,
do_constant_folding=True,
input_names=["image"],
output_names=["image_features_0", "image_features_1", "image_embeddings"],
dynamic_axes=dynamic_axes,
)
else:
torch._dynamo.config.capture_scalar_outputs = True
ep = torch.export.export(
sam2_encoder,
args=(image,),
strict=False,
dynamic_shapes=[
{0: torch.export.Dim.AUTO},
],
)
onnx_program = torch.onnx.export(
ep,
(),
opset_version=17,
input_names=["image"],
output_names=["image_features_0", "image_features_1", "image_embeddings"],
dynamo=True,
)
onnx_program.optimize()
onnx_program.save(onnx_model_path + ".dynamo.onnx", external_data=False)
import onnx # noqa: PLC0415
from onnxruntime.transformers.dynamo_onnx_helper import DynamoOnnxHelper # noqa: PLC0415
onnx_model = onnx.load_model(onnx_model_path + ".dynamo.onnx", load_external_data=True)
if dynamic_batch_axes:
# Fix labels of dynamic axes since they can't be specified during Dynamo export currently
onnx_model.graph.input[0].type.tensor_type.shape.dim[0].dim_param = "batch_size"
for i in range(3):
onnx_model.graph.output[i].type.tensor_type.shape.dim[0].dim_param = "batch_size"
onnx_model_helper = DynamoOnnxHelper(onnx_model)
onnx_model_helper.convert_constants_to_initializers()
if clear_dynamo_metadata:
onnx_model_helper.clear_metadata()
import os # noqa: PLC0415
if os.path.exists(onnx_model_path):
os.remove(onnx_model_path)
if os.path.exists(onnx_model_path + ".data"):
os.remove(onnx_model_path + ".data")
onnx_model_helper.model.save_model_to_file(
onnx_model_path, use_external_data_format=True, all_tensors_to_one_file=True, convert_attribute=True
)
print("encoder onnx model saved to", onnx_model_path)
def test_image_encoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
dynamic_batch_axes=False,
):
ort_session = onnxruntime.InferenceSession(onnx_model_path, providers=["CPUExecutionProvider"])
model_inputs = ort_session.get_inputs()
input_names = [model_inputs[i].name for i in range(len(model_inputs))]
logger.info("input_names: %s", input_names)
model_outputs = ort_session.get_outputs()
output_names = [model_outputs[i].name for i in range(len(model_outputs))]
logger.info("output_names: %s", output_names)
batch_sizes = [1, 2] if dynamic_batch_axes else [1]
for batch_size in batch_sizes:
image = random_sam2_input_image(batch_size)
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image.clone())
logger.info("image.shape: %s", image.shape)
logger.info("image_features_0.shape: %s", image_features_0.shape)
logger.info("image_features_1.shape: %s", image_features_1.shape)
logger.info("image_embeddings.shape: %s", image_embeddings.shape)
outputs = ort_session.run(output_names, {"image": image.numpy()})
for i, output_name in enumerate(output_names):
logger.info("output %s shape %s", output_name, outputs[i].shape)
ort_image_features_0, ort_image_features_1, ort_image_embeddings = outputs
# ONNXRuntime and PyTorch has about 0.75% mismatched elements, but seems not impacting segmentation results.
if (
compare_tensors_with_tolerance(
"image_features_0",
image_features_0,
torch.tensor(ort_image_features_0),
mismatch_percentage_tolerance=1,
)
and compare_tensors_with_tolerance(
"image_features_1",
image_features_1,
torch.tensor(ort_image_features_1),
mismatch_percentage_tolerance=1,
)
and compare_tensors_with_tolerance(
"image_embeddings",
image_embeddings,
torch.tensor(ort_image_embeddings),
mismatch_percentage_tolerance=1,
)
):
print(f"onnx model has been verified for batch_size={batch_size}: {onnx_model_path}")
else:
print(f"onnx model verification failed for batch_size={batch_size}: {onnx_model_path}")

View file

@ -0,0 +1,208 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import warnings
import torch
from image_encoder import SAM2ImageEncoder, random_sam2_input_image
from prompt_encoder import SAM2PromptEncoder
from sam2.modeling.sam2_base import SAM2Base
from torch import nn
logger = logging.getLogger(__name__)
class SAM2MaskDecoder(nn.Module):
def __init__(
self,
sam_model: SAM2Base,
multimask_output: bool,
dynamic_multimask_via_stability: bool = True,
) -> None:
super().__init__()
self.mask_decoder = sam_model.sam_mask_decoder
self.prompt_encoder = sam_model.sam_prompt_encoder
self.model = sam_model
self.multimask_output = multimask_output
self.dynamic_multimask_via_stability = dynamic_multimask_via_stability
@torch.no_grad()
def forward(
self,
image_features_0: torch.Tensor,
image_features_1: torch.Tensor,
image_embeddings: torch.Tensor,
image_pe: torch.Tensor,
sparse_embeddings: torch.Tensor,
dense_embeddings: torch.Tensor,
):
"""
Decode masks from image and prompt embeddings. Only support H=W=1024.
Args:
image_features_0 (torch.Tensor): [1, 32, H/4, W/4]. high resolution features of level 0 from image encoder.
image_features_1 (torch.Tensor): [1, 64, H/8, W/8]. high resolution features of level 1 from image encoder.
image_embeddings (torch.Tensor): [1, 256, H/16, W/16]. image embedding from image encoder.
image_pe (torch.Tensor): [1, 256, H/16, W/16]. image positional encoding.
sparse_embeddings (torch.Tensor): [L, P+1, 256], embedding for points and boxes.
dense_embeddings (torch.Tensor): [L, 256, H/16, W/16]. embedding for input masks.
Returns:
low_res_masks (torch.Tensor, optional): [1, M, H/4, W/4]. low resolution masks.
iou_predictions (torch.Tensor): [1, M]. scores for M masks.
"""
low_res_masks, iou_predictions, _, _ = self.mask_decoder.predict_masks(
image_embeddings=image_embeddings,
image_pe=image_pe,
sparse_prompt_embeddings=sparse_embeddings,
dense_prompt_embeddings=dense_embeddings,
repeat_image=sparse_embeddings.shape[0] > 1, # batch mode
high_res_features=[image_features_0, image_features_1],
)
if self.multimask_output:
low_res_masks = low_res_masks[:, 1:, :, :]
iou_predictions = iou_predictions[:, 1:]
elif self.dynamic_multimask_via_stability:
# When outputting a single mask, if the stability score from the current single-mask
# output (based on output token 0) falls below a threshold, we instead select from
# multi-mask outputs (based on output token 1~3) the mask with the highest predicted IoU score.
low_res_masks, iou_predictions = self.mask_decoder._dynamic_multimask_via_stability(
low_res_masks, iou_predictions
)
else:
low_res_masks = low_res_masks[:, 0:1, :, :]
iou_predictions = iou_predictions[:, 0:1]
return low_res_masks, iou_predictions
def export_mask_decoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
multimask_output: bool,
dynamic_multimask_via_stability: bool = True,
verbose=False,
):
sam2_prompt_encoder = SAM2PromptEncoder(sam2_model).cpu()
image = random_sam2_input_image()
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image)
logger.info("image_features_0.shape: %s", image_features_0.shape)
logger.info("image_features_1.shape: %s", image_features_1.shape)
logger.info("image_embeddings.shape: %s", image_embeddings.shape)
# encode an random prompt
num_labels = 2
num_points = 3
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.float)
input_masks = torch.zeros(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.ones(1, dtype=torch.float)
sparse_embeddings, dense_embeddings, image_pe = sam2_prompt_encoder(
point_coords, point_labels, input_masks, has_input_masks
)
logger.info("sparse_embeddings.shape: %s", sparse_embeddings.shape)
logger.info("dense_embeddings.shape: %s", dense_embeddings.shape)
logger.info("image_pe.shape: %s", image_pe.shape)
sam2_mask_decoder = SAM2MaskDecoder(sam2_model, multimask_output, dynamic_multimask_via_stability)
inputs = (image_features_0, image_features_1, image_embeddings, image_pe, sparse_embeddings, dense_embeddings)
low_res_masks, iou_predictions = sam2_mask_decoder(*inputs)
logger.info("low_res_masks.shape: %s", low_res_masks.shape)
logger.info("iou_predictions.shape: %s", iou_predictions.shape)
with warnings.catch_warnings():
if not verbose:
warnings.filterwarnings("ignore", category=torch.jit.TracerWarning)
warnings.filterwarnings("ignore", category=UserWarning)
torch.onnx.export(
sam2_mask_decoder,
inputs,
onnx_model_path,
export_params=True,
opset_version=18,
do_constant_folding=True,
input_names=[
"image_features_0",
"image_features_1",
"image_embeddings",
"image_pe",
"sparse_embeddings",
"dense_embeddings",
],
output_names=["low_res_masks", "iou_predictions"],
dynamic_axes={
"sparse_embeddings": {0: "num_labels", 1: "num_points+1"},
"dense_embeddings": {0: "num_labels"},
"low_res_masks": {0: "num_labels"},
"iou_predictions": {0: "num_labels"},
},
)
print("mask decoder onnx model saved to", onnx_model_path)
def test_mask_decoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
multimask_output: bool,
dynamic_multimask_via_stability: bool,
):
sam2_prompt_encoder = SAM2PromptEncoder(sam2_model).cpu()
image = random_sam2_input_image()
sam2_encoder = SAM2ImageEncoder(sam2_model).cpu()
image_features_0, image_features_1, image_embeddings = sam2_encoder(image)
num_labels = 1
num_points = 5
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.float)
input_masks = torch.rand(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.ones(1, dtype=torch.float)
sparse_embeddings, dense_embeddings, image_pe = sam2_prompt_encoder(
point_coords, point_labels, input_masks, has_input_masks
)
sam2_mask_decoder = SAM2MaskDecoder(sam2_model, multimask_output, dynamic_multimask_via_stability)
inputs = (image_features_0, image_features_1, image_embeddings, image_pe, sparse_embeddings, dense_embeddings)
low_res_masks, iou_predictions = sam2_mask_decoder(*inputs)
import onnxruntime # noqa: PLC0415
ort_session = onnxruntime.InferenceSession(onnx_model_path, providers=["CPUExecutionProvider"])
model_inputs = ort_session.get_inputs()
input_names = [model_inputs[i].name for i in range(len(model_inputs))]
logger.info("input_names: %s", input_names)
model_outputs = ort_session.get_outputs()
output_names = [model_outputs[i].name for i in range(len(model_outputs))]
logger.info("output_names: %s", output_names)
outputs = ort_session.run(
output_names,
{
"image_features_0": image_features_0.numpy(),
"image_features_1": image_features_1.numpy(),
"image_embeddings": image_embeddings.numpy(),
"image_pe": image_pe.numpy(),
"sparse_embeddings": sparse_embeddings.numpy(),
"dense_embeddings": dense_embeddings.numpy(),
},
)
for i, output_name in enumerate(output_names):
logger.info("output %s shape: %s", output_name, outputs[i].shape)
ort_low_res_masks, ort_iou_predictions = outputs
torch.testing.assert_close(low_res_masks, torch.tensor(ort_low_res_masks), atol=5e-3, rtol=1e-4)
torch.testing.assert_close(iou_predictions, torch.tensor(ort_iou_predictions), atol=5e-3, rtol=1e-4)
print(f"onnx model has been verified: {onnx_model_path}")

View file

@ -0,0 +1,33 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import nvtx
from cuda import cudart
class NvtxHelper:
def __init__(self, stages):
self.stages = stages
self.events = {}
for stage in stages:
for marker in ["start", "stop"]:
self.events[stage + "-" + marker] = cudart.cudaEventCreate()[1]
self.markers = {}
def start_profile(self, stage, color="blue"):
self.markers[stage] = nvtx.start_range(message=stage, color=color)
event_name = stage + "-start"
if event_name in self.events:
cudart.cudaEventRecord(self.events[event_name], 0)
def stop_profile(self, stage):
event_name = stage + "-stop"
if event_name in self.events:
cudart.cudaEventRecord(self.events[event_name], 0)
nvtx.end_range(self.markers[stage])
def print_latency(self):
for stage in self.stages:
latency = cudart.cudaEventElapsedTime(self.events[f"{stage}-start"], self.events[f"{stage}-stop"])[1]
print(f"{stage}: {latency:.2f} ms")

View file

@ -0,0 +1,189 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import torch
from sam2.modeling.sam2_base import SAM2Base
from sam2_utils import compare_tensors_with_tolerance
from torch import nn
logger = logging.getLogger(__name__)
class SAM2PromptEncoder(nn.Module):
def __init__(self, sam_model: SAM2Base):
super().__init__()
self.prompt_encoder = sam_model.sam_prompt_encoder
self.model = sam_model
@torch.no_grad()
def forward(
self,
point_coords: torch.Tensor,
point_labels: torch.Tensor,
input_masks: torch.Tensor,
has_input_masks: torch.Tensor,
):
"""Encode prompts.
Args:
point_coords (torch.Tensor): [L, P, 2] shape and float32 dtype and contains the absolute pixel
coordinate in (x, y) format of the P input points in image of size 1024x1024.
point_labels (torch.Tensor): shape [L, P] and int32 dtype, where 1 means
positive (foreground), 0 means negative (background), -1 means padding,
2 (box left upper corner), 3 (box right bottom corner).
input_masks (torch.Tensor): [L, 1, H/4, W/4]. Low resolution mask input to the model.
Typically coming from a previous iteration.
has_input_masks (torch.Tensor): [L]. 1.0 if input_masks is used, 0.0 otherwise.
Returns:
sparse_embeddings (torch.Tensor): [L, P+1, 256], embedding for points and boxes.
dense_embeddings (torch.Tensor): [L, 256, 64, 64]. embedding for input masks.
image_pe (torch.Tensor, optional): [1, 256, 64, 64]. image positional encoding.
"""
sparse_embeddings = self._embed_points(point_coords, point_labels)
dense_embeddings = self._embed_masks(input_masks, has_input_masks)
image_pe = self.prompt_encoder.get_dense_pe()
return sparse_embeddings, dense_embeddings, image_pe
def _embed_points(self, point_coords: torch.Tensor, point_labels: torch.Tensor) -> torch.Tensor:
point_coords = point_coords + 0.5
padding_point = torch.zeros((point_coords.shape[0], 1, 2), device=point_coords.device)
padding_label = -torch.ones((point_labels.shape[0], 1), device=point_labels.device)
point_coords = torch.cat([point_coords, padding_point], dim=1)
point_labels = torch.cat([point_labels, padding_label], dim=1)
# Note that the input coordinates are based on image size 1024x1024. Here we normalize it to [0.0, 1.0).
point_coords[:, :, 0] = point_coords[:, :, 0] / self.model.image_size
point_coords[:, :, 1] = point_coords[:, :, 1] / self.model.image_size
point_embedding = self.prompt_encoder.pe_layer._pe_encoding(point_coords)
point_labels = point_labels.unsqueeze(-1).expand_as(point_embedding)
point_embedding = point_embedding * (point_labels != -1)
point_embedding = point_embedding + self.prompt_encoder.not_a_point_embed.weight * (point_labels == -1)
for i in range(self.prompt_encoder.num_point_embeddings):
point_embedding = point_embedding + self.prompt_encoder.point_embeddings[i].weight * (point_labels == i)
return point_embedding
def _embed_masks(self, input_masks: torch.Tensor, has_input_masks: torch.Tensor) -> torch.Tensor:
mask_embedding = self.prompt_encoder.mask_downscaling(input_masks)
no_mask_embedding = self.prompt_encoder.no_mask_embed.weight.reshape(1, -1, 1, 1)
logger.info("no_mask_embedding.shape: %s", no_mask_embedding.shape)
mask_embedding = has_input_masks * mask_embedding + (1.0 - has_input_masks) * no_mask_embedding
logger.info("mask_embedding.shape: %s", mask_embedding.shape)
return mask_embedding
def export_prompt_encoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
):
sam2_prompt_encoder = SAM2PromptEncoder(sam2_model).cpu()
num_labels = 2
num_points = 3
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.int32)
input_masks = torch.zeros(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.ones(1, dtype=torch.float)
sparse_embeddings, dense_embeddings, image_pe = sam2_prompt_encoder(
point_coords, point_labels, input_masks, has_input_masks
)
logger.info("point_coords.shape: %s", point_coords.shape)
logger.info("point_labels.shape: %s", point_labels.shape)
logger.info("input_masks.shape: %s", input_masks.shape)
logger.info("has_input_masks.shape: %s", has_input_masks.shape)
logger.info("sparse_embeddings.shape: %s", sparse_embeddings.shape)
logger.info("dense_embeddings.shape: %s", dense_embeddings.shape)
logger.info("image_pe.shape: %s", image_pe.shape)
torch.onnx.export(
sam2_prompt_encoder,
(point_coords, point_labels, input_masks, has_input_masks),
onnx_model_path,
export_params=True,
opset_version=18,
do_constant_folding=True,
input_names=["point_coords", "point_labels", "input_masks", "has_input_masks"],
output_names=["sparse_embeddings", "dense_embeddings", "image_pe"],
dynamic_axes={
"point_coords": {0: "num_labels", 1: "num_points"},
"point_labels": {0: "num_labels", 1: "num_points"},
"input_masks": {0: "num_labels"},
"sparse_embeddings": {0: "num_labels", 1: "num_points+1"},
"dense_embeddings": {0: "num_labels"},
},
)
print("prompt encoder onnx model saved to ", onnx_model_path)
def test_prompt_encoder_onnx(
sam2_model: SAM2Base,
onnx_model_path: str,
):
sam2_prompt_encoder = SAM2PromptEncoder(sam2_model).cpu()
num_labels = 1
num_points = 5
point_coords = torch.randint(low=0, high=1024, size=(num_labels, num_points, 2), dtype=torch.float)
point_labels = torch.randint(low=0, high=1, size=(num_labels, num_points), dtype=torch.int32)
input_masks = torch.rand(num_labels, 1, 256, 256, dtype=torch.float)
has_input_masks = torch.ones(1, dtype=torch.float)
sparse_embeddings, dense_embeddings, image_pe = sam2_prompt_encoder(
point_coords, point_labels, input_masks, has_input_masks
)
import onnxruntime # noqa: PLC0415
ort_session = onnxruntime.InferenceSession(onnx_model_path, providers=["CPUExecutionProvider"])
model_inputs = ort_session.get_inputs()
input_names = [model_inputs[i].name for i in range(len(model_inputs))]
logger.info("input_names: %s", input_names)
model_outputs = ort_session.get_outputs()
output_names = [model_outputs[i].name for i in range(len(model_outputs))]
logger.info("output_names: %s", output_names)
outputs = ort_session.run(
output_names,
{
"point_coords": point_coords.numpy(),
"point_labels": point_labels.numpy(),
"input_masks": input_masks.numpy(),
"has_input_masks": has_input_masks.numpy(),
},
)
for i, output_name in enumerate(output_names):
logger.info("output %s shape: %s", output_name, outputs[i].shape)
ort_sparse_embeddings, ort_dense_embeddings, ort_image_pe = outputs
if (
compare_tensors_with_tolerance(
"sparse_embeddings",
sparse_embeddings,
torch.tensor(ort_sparse_embeddings),
mismatch_percentage_tolerance=0.2,
)
and compare_tensors_with_tolerance(
"dense_embeddings", dense_embeddings, torch.tensor(ort_dense_embeddings), mismatch_percentage_tolerance=0.2
)
and compare_tensors_with_tolerance(
"image_pe", image_pe, torch.tensor(ort_image_pe), mismatch_percentage_tolerance=0.2
)
):
print(f"onnx model has been verified: {onnx_model_path}")
else:
print(f"onnx model verification failed: {onnx_model_path}")

View file

@ -0,0 +1,321 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import os
import matplotlib.image as mpimg
import matplotlib.pyplot as plt
import numpy as np
import torch
from matplotlib.patches import Rectangle
from PIL import Image
from sam2.sam2_image_predictor import SAM2ImagePredictor
from sam2_image_onnx_predictor import SAM2ImageOnnxPredictor
from sam2_utils import load_sam2_model
import onnxruntime
def show_mask(mask, ax, random_color=False, borders=True):
if random_color:
color = np.concatenate([np.random.random(3), np.array([0.6])], axis=0)
else:
color = np.array([30 / 255, 144 / 255, 255 / 255, 0.6])
h, w = mask.shape[-2:]
mask = mask.astype(np.uint8)
mask_image = mask.reshape(h, w, 1) * color.reshape(1, 1, -1)
if borders:
import cv2 # noqa: PLC0415
contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_NONE)
# Try to smooth contours
contours = [cv2.approxPolyDP(contour, epsilon=0.01, closed=True) for contour in contours]
mask_image = cv2.drawContours(mask_image, contours, -1, (1, 1, 1, 0.5), thickness=2)
ax.imshow(mask_image)
def show_points(coords, labels, ax, marker_size=375):
pos_points = coords[labels == 1]
neg_points = coords[labels == 0]
ax.scatter(
pos_points[:, 0], pos_points[:, 1], color="green", marker="*", s=marker_size, edgecolor="white", linewidth=1.25
)
ax.scatter(
neg_points[:, 0], neg_points[:, 1], color="red", marker="*", s=marker_size, edgecolor="white", linewidth=1.25
)
def show_box(box, ax):
x0, y0 = box[0], box[1]
w, h = box[2] - box[0], box[3] - box[1]
ax.add_patch(Rectangle((x0, y0), w, h, edgecolor="green", facecolor=(0, 0, 0, 0), lw=2))
def show_masks(
image,
masks,
scores,
point_coords=None,
box_coords=None,
input_labels=None,
borders=True,
output_image_file_prefix=None,
image_files=None,
):
for i, (mask, score) in enumerate(zip(masks, scores, strict=False)):
plt.figure(figsize=(10, 10))
plt.imshow(image)
show_mask(mask, plt.gca(), borders=borders)
if point_coords is not None:
assert input_labels is not None
show_points(point_coords, input_labels, plt.gca())
if box_coords is not None:
show_box(box_coords, plt.gca())
if len(scores) > 1:
plt.title(f"Mask {i + 1}, Score: {score:.3f}", fontsize=18)
plt.axis("off")
if output_image_file_prefix:
filename = f"{output_image_file_prefix}_{i}.png"
if os.path.exists(filename):
os.remove(filename)
plt.savefig(filename, format="png", bbox_inches="tight", pad_inches=0)
if isinstance(image_files, list):
image_files.append(filename)
plt.show(block=False)
plt.close()
def get_predictor(
sam2_dir: str,
device: str | torch.device,
dtype: torch.dtype,
model_type="sam2_hiera_large",
engine="torch",
image_encoder_onnx_path: str = "",
image_decoder_onnx_path: str = "",
image_decoder_multi_onnx_path: str = "",
provider: str = "CUDAExecutionProvider",
):
sam2_model = load_sam2_model(sam2_dir, model_type, device=device)
if engine == "torch":
predictor = SAM2ImagePredictor(sam2_model)
else:
predictor = SAM2ImageOnnxPredictor(
sam2_model,
image_encoder_onnx_path=image_encoder_onnx_path,
image_decoder_onnx_path=image_decoder_onnx_path,
image_decoder_multi_onnx_path=image_decoder_multi_onnx_path,
provider=provider,
device=device,
onnx_dtype=dtype,
)
return predictor
def run_demo(
sam2_dir: str,
model_type: str = "sam2_hiera_large",
engine: str = "torch",
dtype: torch.dtype = torch.float32,
image_encoder_onnx_path: str = "",
image_decoder_onnx_path: str = "",
image_decoder_multi_onnx_path: str = "",
use_gpu: bool = True,
enable_batch: bool = False,
):
if use_gpu:
assert torch.cuda.is_available()
assert "CUDAExecutionProvider" in onnxruntime.get_available_providers()
provider = "CUDAExecutionProvider"
else:
provider = "CPUExecutionProvider"
device = torch.device("cuda" if use_gpu else "cpu")
if use_gpu and engine == "torch" and torch.cuda.get_device_properties(0).major >= 8:
# Turn on tfloat32 for Ampere GPUs.
torch.backends.cuda.matmul.allow_tf32 = True
torch.backends.cudnn.allow_tf32 = True
np.random.seed(3)
image = Image.open("truck.jpg")
image = np.array(image.convert("RGB"))
predictor = get_predictor(
sam2_dir,
device,
dtype,
model_type,
engine,
image_encoder_onnx_path,
image_decoder_onnx_path,
image_decoder_multi_onnx_path,
provider=provider,
)
predictor.set_image(image)
prefix = f"sam2_demo_{engine}_"
# The model returns masks, quality predictions for those masks,
# and low resolution mask logits that can be passed to the next iteration of prediction.
# With multimask_output=True (the default setting), SAM 2 outputs 3 masks, where
# scores gives the model's own estimation of the quality of these masks.
# For ambiguous prompts such as a single point, it is recommended to use multimask_output=True
# even if only a single mask is desired;
input_point = np.array([[500, 375]])
input_label = np.array([1])
masks, scores, logits = predictor.predict(
point_coords=input_point,
point_labels=input_label,
multimask_output=True,
)
sorted_ind = np.argsort(scores)[::-1]
masks = masks[sorted_ind]
scores = scores[sorted_ind]
logits = logits[sorted_ind]
image_files = []
show_masks(
image,
masks,
scores,
point_coords=input_point,
input_labels=input_label,
borders=True,
output_image_file_prefix=prefix + "multimask",
image_files=image_files,
)
# Multiple points.
input_point = np.array([[500, 375], [1125, 625]])
input_label = np.array([1, 1])
mask_input = logits[np.argmax(scores), :, :] # Choose the model's best mask
masks, scores, _ = predictor.predict(
point_coords=input_point,
point_labels=input_label,
mask_input=mask_input[None, :, :],
multimask_output=False,
)
show_masks(
image,
masks,
scores,
point_coords=input_point,
input_labels=input_label,
output_image_file_prefix=prefix + "multi_points",
image_files=image_files,
)
# Specify a window and a background point.
input_point = np.array([[500, 375], [1125, 625]])
input_label = np.array([1, 0])
mask_input = logits[np.argmax(scores), :, :] # Choose the model's best mask
masks, scores, _ = predictor.predict(
point_coords=input_point,
point_labels=input_label,
mask_input=mask_input[None, :, :],
multimask_output=False,
)
show_masks(
image,
masks,
scores,
point_coords=input_point,
input_labels=input_label,
output_image_file_prefix=prefix + "background_point",
image_files=image_files,
)
# Take a box as input
input_box = np.array([425, 600, 700, 875])
masks, scores, _ = predictor.predict(
point_coords=None,
point_labels=None,
box=input_box[None, :],
multimask_output=False,
)
show_masks(
image,
masks,
scores,
box_coords=input_box,
output_image_file_prefix=prefix + "box",
image_files=image_files,
)
# Combining points and boxes
input_box = np.array([425, 600, 700, 875])
input_point = np.array([[575, 750]])
input_label = np.array([0])
masks, scores, logits = predictor.predict(
point_coords=input_point,
point_labels=input_label,
box=input_box,
multimask_output=False,
)
show_masks(
image,
masks,
scores,
box_coords=input_box,
point_coords=input_point,
input_labels=input_label,
output_image_file_prefix=prefix + "box_and_point",
image_files=image_files,
)
# TODO: support batched prompt inputs
if enable_batch:
input_boxes = np.array(
[
[75, 275, 1725, 850],
[425, 600, 700, 875],
[1375, 550, 1650, 800],
[1240, 675, 1400, 750],
]
)
masks, scores, _ = predictor.predict(
point_coords=None,
point_labels=None,
box=input_boxes,
multimask_output=False,
)
plt.figure(figsize=(10, 10))
plt.imshow(image)
for mask in masks:
show_mask(mask.squeeze(0), plt.gca(), random_color=True)
for box in input_boxes:
show_box(box, plt.gca())
plt.axis("off")
plt.show()
plt.savefig(prefix + "batch_prompt.png")
image_files.append(prefix + "batch_prompt.png")
return image_files
def show_all_images(left_images, right_images, suffix=""):
# Show images in two rows since display screen is horizontal in most cases.
fig, axes = plt.subplots(nrows=2, ncols=len(left_images), figsize=(19.20, 10.80))
for i, (left_img_path, right_img_path) in enumerate(zip(left_images, right_images, strict=False)):
left_img = mpimg.imread(left_img_path)
right_img = mpimg.imread(right_img_path)
axes[0, i].imshow(left_img)
axes[0, i].set_title(left_img_path.replace("sam2_demo_", "").replace(".png", ""), fontsize=10)
axes[0, i].axis("off")
axes[0, i].set_aspect(left_img.shape[1] / left_img.shape[0])
axes[1, i].imshow(right_img)
axes[1, i].set_title(right_img_path.replace("sam2_demo_", "").replace(".png", ""), fontsize=10)
axes[1, i].axis("off")
axes[1, i].set_aspect(right_img.shape[1] / right_img.shape[0])
plt.tight_layout()
plt.savefig(f"sam2_demo{suffix}.png", format="png", bbox_inches="tight", dpi=1000)
plt.show()

View file

@ -0,0 +1,279 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import numpy as np
import torch
from PIL.Image import Image
from sam2.modeling.sam2_base import SAM2Base
from sam2.sam2_image_predictor import SAM2ImagePredictor
from sam2_utils import decoder_shape_dict, encoder_shape_dict
from onnxruntime import InferenceSession
from onnxruntime.transformers.io_binding_helper import CudaSession
logger = logging.getLogger(__name__)
def create_ort_session(
onnx_path: str,
session_options=None,
provider="CUDAExecutionProvider",
enable_cuda_graph=False,
use_tf32=True,
) -> InferenceSession:
if provider == "CUDAExecutionProvider":
device_id = torch.cuda.current_device()
provider_options = CudaSession.get_cuda_provider_options(device_id, enable_cuda_graph)
provider_options["use_tf32"] = int(use_tf32)
providers = [(provider, provider_options), "CPUExecutionProvider"]
else:
providers = ["CPUExecutionProvider"]
logger.info("Using providers: %s", providers)
return InferenceSession(onnx_path, session_options, providers=providers)
def create_session(
onnx_path: str,
session_options=None,
provider="CUDAExecutionProvider",
device: str | torch.device = "cuda",
enable_cuda_graph=False,
) -> CudaSession:
ort_session = create_ort_session(
onnx_path, session_options, provider, enable_cuda_graph=enable_cuda_graph, use_tf32=True
)
cuda_session = CudaSession(ort_session, device=torch.device(device), enable_cuda_graph=enable_cuda_graph)
return cuda_session
class SAM2ImageOnnxPredictor(SAM2ImagePredictor):
def __init__(
self,
sam_model: SAM2Base,
image_encoder_onnx_path: str = "",
image_decoder_onnx_path: str = "",
image_decoder_multi_onnx_path: str = "",
provider: str = "CUDAExecutionProvider",
device: str | torch.device = "cuda",
onnx_dtype: torch.dtype = torch.float32,
mask_threshold=0.0,
max_hole_area=0.0,
max_sprinkle_area=0.0,
**kwargs,
) -> None:
"""
Uses SAM-2 to compute the image embedding for an image, and then allow mask prediction given prompts.
Arguments:
sam_model (SAM2Base): The model to use for mask prediction.
onnx_directory (str): The path of the directory that contains encoder and decoder onnx models.
onnx_dtype (torch.dtype): The data type to use for ONNX inputs.
mask_threshold (float): The threshold to convert mask logits to binary masks. Default is 0.0.
max_hole_area (float): If max_hole_area > 0, we fill small holes in up to
the maximum area of max_hole_area in low_res_masks.
max_sprinkle_area (float): If max_sprinkle_area > 0, we remove small sprinkles up to
the maximum area of max_sprinkle_area in low_res_masks.
"""
super().__init__(
sam_model, mask_threshold=mask_threshold, max_hole_area=max_hole_area, max_sprinkle_area=max_sprinkle_area
)
logger.debug("self.device=%s, device=%s", self.device, device)
# This model is exported by image_encoder.py.
self.encoder_session = create_session(
image_encoder_onnx_path,
session_options=None,
provider=provider,
device=device,
enable_cuda_graph=False,
)
self.onnx_dtype = onnx_dtype
# This model is exported by image_decoder.py. It outputs only one mask.
self.decoder_session = create_session(
image_decoder_onnx_path,
session_options=None,
provider=provider,
device=device,
enable_cuda_graph=False,
)
# This model is exported by image_decoder.py. It outputs multiple (3) masks.
self.decoder_session_multi_out = create_session(
image_decoder_multi_onnx_path,
session_options=None,
provider=provider,
device=device,
enable_cuda_graph=False,
)
@torch.no_grad()
def set_image(self, image: np.ndarray | Image):
"""
Calculates the image embeddings for the provided image.
Arguments:
image (np.ndarray or PIL Image): The input image to embed in RGB format.
The image should be in HWC format if np.ndarray, or WHC format if PIL Image with pixel values in [0, 255].
"""
self.reset_predictor()
# Transform the image to the form expected by the model
if isinstance(image, np.ndarray):
# For numpy array image, we assume (HxWxC) format.
self._orig_hw = [image.shape[:2]]
elif isinstance(image, Image):
w, h = image.size
self._orig_hw = [(h, w)]
else:
raise NotImplementedError("Image format not supported")
input_image = self._transforms(image)
input_image = input_image[None, ...].to(self.device)
assert len(input_image.shape) == 4 and input_image.shape[1] == 3, (
f"input_image must be of size 1x3xHxW, got {input_image.shape}"
)
# Computing image embeddings for the provided image
io_shapes = encoder_shape_dict(batch_size=1, height=input_image.shape[2], width=input_image.shape[3])
self.encoder_session.allocate_buffers(io_shapes)
feed_dict = {"image": input_image.to(self.onnx_dtype).to(self.device)}
for key, value in feed_dict.items():
logger.debug(f"{key}: {value.shape}, {value.dtype}")
logger.debug(f"encoder onnx: {self.encoder_session.ort_session._model_path}")
ort_outputs = self.encoder_session.infer(feed_dict)
self._features = {
"image_embed": ort_outputs["image_embeddings"],
"high_res_feats": [ort_outputs[f"image_features_{i}"] for i in range(2)],
}
self._is_image_set = True
logging.info("Image embeddings computed.")
@torch.no_grad()
def _predict(
self,
point_coords: torch.Tensor | None,
point_labels: torch.Tensor | None,
boxes: torch.Tensor | None = None,
mask_input: torch.Tensor | None = None,
multimask_output: bool = True,
return_logits: bool = False,
img_idx: int = -1,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
"""
Predict masks for the given input prompts, using the currently set image.
Input prompts are batched torch tensors and are expected to already be
transformed to the input frame using SAM2Transforms.
Arguments:
point_coords (torch.Tensor or None): A BxNx2 array of point prompts to the
model. Each point is in (X,Y) in pixels.
point_labels (torch.Tensor or None): A BxN array of labels for the
point prompts. 1 indicates a foreground point and 0 indicates a
background point.
boxes (np.ndarray or None): A Bx4 array given a box prompt to the
model, in XYXY format.
mask_input (np.ndarray): A low resolution mask input to the model, typically
coming from a previous prediction iteration. Has form Bx1xHxW, where
for SAM, H=W=256. Masks returned by a previous iteration of the
predict method do not need further transformation.
multimask_output (bool): If true, the model will return three masks.
For ambiguous input prompts (such as a single click), this will often
produce better masks than a single prediction. If only a single
mask is needed, the model's predicted quality score can be used
to select the best mask. For non-ambiguous prompts, such as multiple
input prompts, multimask_output=False can give better results.
return_logits (bool): If true, returns un-thresholded masks logits
instead of a binary mask.
Returns:
(torch.Tensor): The output masks in BxCxHxW format, where C is the
number of masks, and (H, W) is the original image size.
(torch.Tensor): An array of shape BxC containing the model's
predictions for the quality of each mask.
(torch.Tensor): An array of shape BxCxHxW, where C is the number
of masks and H=W=256. These low res logits can be passed to
a subsequent iteration as mask input.
"""
assert not return_logits # onnx model is exported for returning bool masks.
if not self._is_image_set:
raise RuntimeError("An image must be set with .set_image(...) before mask prediction.")
if point_coords is not None:
concat_points = (point_coords, point_labels)
else:
concat_points = None
# Embed prompts
if boxes is not None:
box_coords = boxes.reshape(-1, 2, 2)
box_labels = torch.tensor([[2, 3]], dtype=torch.int, device=boxes.device)
box_labels = box_labels.repeat(boxes.size(0), 1)
# we merge "boxes" and "points" into a single "concat_points" input (where
# boxes are added at the beginning) to sam_prompt_encoder
if concat_points is not None:
concat_coords = torch.cat([box_coords, concat_points[0]], dim=1)
concat_labels = torch.cat([box_labels, concat_points[1]], dim=1)
concat_points = (concat_coords, concat_labels)
else:
concat_points = (box_coords, box_labels)
assert concat_points is not None
num_labels = concat_points[0].shape[0]
shape_dict = decoder_shape_dict(
original_image_height=self._orig_hw[img_idx][0],
original_image_width=self._orig_hw[img_idx][1],
num_labels=num_labels,
max_points=concat_points[0].shape[1],
num_masks=3 if multimask_output else 1,
)
if multimask_output:
decoder_session = self.decoder_session_multi_out
else:
decoder_session = self.decoder_session
decoder_session.allocate_buffers(shape_dict)
image_features_0 = self._features["high_res_feats"][0][img_idx].unsqueeze(0)
image_features_1 = self._features["high_res_feats"][1][img_idx].unsqueeze(0)
image_embeddings = self._features["image_embed"][img_idx].unsqueeze(0)
if mask_input is None:
input_masks = torch.zeros(num_labels, 1, 256, 256, dtype=self.onnx_dtype, device=self.device)
has_input_masks = torch.zeros(num_labels, dtype=self.onnx_dtype, device=self.device)
else:
input_masks = mask_input[img_idx].unsqueeze(0).repeat(num_labels, 1, 1, 1)
has_input_masks = torch.ones(num_labels, dtype=self.onnx_dtype, device=self.device)
feed_dict = {
"image_embeddings": image_embeddings.contiguous().to(dtype=self.onnx_dtype).to(self.device),
"image_features_0": image_features_0.contiguous().to(dtype=self.onnx_dtype).to(self.device),
"image_features_1": image_features_1.contiguous().to(dtype=self.onnx_dtype).to(self.device),
"point_coords": concat_points[0].to(dtype=self.onnx_dtype).to(self.device),
"point_labels": concat_points[1].to(dtype=torch.int32).to(self.device),
"input_masks": input_masks.to(dtype=self.onnx_dtype).to(self.device),
"has_input_masks": has_input_masks.to(dtype=self.onnx_dtype).to(self.device),
"original_image_size": torch.tensor(self._orig_hw[img_idx], dtype=torch.int32, device=self.device),
}
for key, value in feed_dict.items():
logger.debug(f"{key}: {value.shape}, {value.dtype}")
logger.debug(f"decoder onnx: {self.decoder_session.ort_session._model_path}")
ort_outputs = decoder_session.infer(feed_dict)
masks = ort_outputs["masks"]
iou_predictions = ort_outputs["iou_predictions"]
low_res_masks = ort_outputs["low_res_masks"]
return torch.Tensor(masks), torch.Tensor(iou_predictions), torch.Tensor(low_res_masks)

View file

@ -0,0 +1,147 @@
# -------------------------------------------------------------------------
# Copyright (R) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
# --------------------------------------------------------------------------
import logging
import os
import sys
from collections.abc import Mapping
import torch
from sam2.build_sam import build_sam2
from sam2.modeling.sam2_base import SAM2Base
logger = logging.getLogger(__name__)
def _get_model_cfg(model_type) -> str:
assert model_type in ["sam2_hiera_tiny", "sam2_hiera_small", "sam2_hiera_large", "sam2_hiera_base_plus"]
if model_type == "sam2_hiera_tiny":
model_cfg = "sam2_hiera_t.yaml"
elif model_type == "sam2_hiera_small":
model_cfg = "sam2_hiera_s.yaml"
elif model_type == "sam2_hiera_base_plus":
model_cfg = "sam2_hiera_b+.yaml"
else:
model_cfg = "sam2_hiera_l.yaml"
return model_cfg
def load_sam2_model(sam2_dir, model_type, device: str | torch.device = "cpu") -> SAM2Base:
checkpoints_dir = os.path.join(sam2_dir, "checkpoints")
sam2_config_dir = os.path.join(sam2_dir, "sam2_configs")
if not os.path.exists(sam2_dir):
raise FileNotFoundError(f"{sam2_dir} does not exist. Please specify --sam2_dir correctly.")
if not os.path.exists(checkpoints_dir):
raise FileNotFoundError(f"{checkpoints_dir} does not exist. Please specify --sam2_dir correctly.")
if not os.path.exists(sam2_config_dir):
raise FileNotFoundError(f"{sam2_config_dir} does not exist. Please specify --sam2_dir correctly.")
checkpoint_path = os.path.join(checkpoints_dir, f"{model_type}.pt")
if not os.path.exists(checkpoint_path):
raise FileNotFoundError(f"{checkpoint_path} does not exist. Please download checkpoints under the directory.")
if sam2_dir not in sys.path:
sys.path.append(sam2_dir)
model_cfg = _get_model_cfg(model_type)
sam2_model = build_sam2(model_cfg, checkpoint_path, device=device)
return sam2_model
def sam2_onnx_path(output_dir, model_type, component, multimask_output=False, suffix=""):
if component == "image_encoder":
return os.path.join(output_dir, f"{model_type}_image_encoder{suffix}.onnx")
elif component == "mask_decoder":
return os.path.join(output_dir, f"{model_type}_mask_decoder{suffix}.onnx")
elif component == "prompt_encoder":
return os.path.join(output_dir, f"{model_type}_prompt_encoder{suffix}.onnx")
else:
assert component == "image_decoder"
return os.path.join(
output_dir, f"{model_type}_image_decoder" + ("_multi" if multimask_output else "") + f"{suffix}.onnx"
)
def encoder_shape_dict(batch_size: int, height: int, width: int) -> Mapping[str, list[int]]:
assert height == 1024 and width == 1024, "Only 1024x1024 images are supported."
return {
"image": [batch_size, 3, height, width],
"image_features_0": [batch_size, 32, height // 4, width // 4],
"image_features_1": [batch_size, 64, height // 8, width // 8],
"image_embeddings": [batch_size, 256, height // 16, width // 16],
}
def decoder_shape_dict(
original_image_height: int,
original_image_width: int,
num_labels: int = 1,
max_points: int = 16,
num_masks: int = 1,
) -> dict:
height: int = 1024
width: int = 1024
return {
"image_features_0": [1, 32, height // 4, width // 4],
"image_features_1": [1, 64, height // 8, width // 8],
"image_embeddings": [1, 256, height // 16, width // 16],
"point_coords": [num_labels, max_points, 2],
"point_labels": [num_labels, max_points],
"input_masks": [num_labels, 1, height // 4, width // 4],
"has_input_masks": [num_labels],
"original_image_size": [2],
"masks": [num_labels, num_masks, original_image_height, original_image_width],
"iou_predictions": [num_labels, num_masks],
"low_res_masks": [num_labels, num_masks, height // 4, width // 4],
}
def compare_tensors_with_tolerance(
name: str,
tensor1: torch.Tensor,
tensor2: torch.Tensor,
atol=5e-3,
rtol=1e-4,
mismatch_percentage_tolerance=0.1,
) -> bool:
assert tensor1.shape == tensor2.shape
a = tensor1.clone().float()
b = tensor2.clone().float()
differences = torch.abs(a - b)
mismatch_count = (differences > (rtol * torch.max(torch.abs(a), torch.abs(b)) + atol)).sum().item()
total_elements = a.numel()
mismatch_percentage = (mismatch_count / total_elements) * 100
passed = mismatch_percentage < mismatch_percentage_tolerance
log_func = logger.error if not passed else logger.info
log_func(
"%s: mismatched elements percentage %.2f (%d/%d). Verification %s (threshold=%.2f).",
name,
mismatch_percentage,
mismatch_count,
total_elements,
"passed" if passed else "failed",
mismatch_percentage_tolerance,
)
return passed
def random_sam2_input_image(batch_size=1, image_height=1024, image_width=1024) -> torch.Tensor:
image = torch.randn(batch_size, 3, image_height, image_width, dtype=torch.float32).cpu()
return image
def setup_logger(verbose=True):
if verbose:
logging.basicConfig(format="[%(filename)s:%(lineno)s - %(funcName)20s()] %(message)s")
logging.getLogger().setLevel(logging.INFO)
else:
logging.basicConfig(format="[%(message)s")
logging.getLogger().setLevel(logging.WARNING)