TensorRT checklist OpenVINO: The Performance Battle performance for Engineers

import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np
import os
import glob
from PIL import Image
import argparse

# Define TensorRT logger
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

def load_calibration_images(image_dir, num_images=1000, input_shape=(3, 224, 224)):
    """Load and preprocess calibration images for INT8 quantization."""
    image_paths = glob.glob(os.path.join(image_dir, "*.jpg"))[:num_images]
    if not image_paths:
        raise FileNotFoundError(f"No JPEG images found in {image_dir}")

    images = []
    for path in image_paths:
        try:
            img = Image.open(path).resize((input_shape[2], input_shape[1]))
            img = np.array(img).transpose(2, 0, 1).astype(np.float32) / 255.0
            img = (img - np.array([0.485, 0.456, 0.406])[:, None, None]) / np.array([0.229, 0.224, 0.225])[:, None, None]
            images.append(img)
        except Exception as e:
            print(f"Skipping {path}: {e}")
            continue
    if not images:
        raise ValueError("No valid calibration images loaded")
    return np.stack(images)

class ResNet50Calibrator(trt.IInt8Calibrator):
    """INT8 calibrator for ResNet-50 using ImageNet calibration set."""
    def __init__(self, calibration_images, batch_size=32):
        trt.IInt8Calibrator.__init__(self)
        self.calibration_images = calibration_images
        self.batch_size = batch_size
        self.current_index = 0
        self.device_input = cuda.mem_alloc(self.calibration_images[0].nbytes * batch_size)

    def get_batch_size(self):
        return self.batch_size

    def get_batch(self, names):
        if self.current_index + self.batch_size > len(self.calibration_images):
            return None
        batch = self.calibration_images[self.current_index:self.current_index + self.batch_size]
        self.current_index += self.batch_size
        cuda.memcpy_htod(self.device_input, batch)
        return [int(self.device_input)]

    def read_calibration_cache(self):
        return None

    def write_calibration_cache(self, cache):
        pass

def build_tensorrt_engine(onnx_path, engine_path, precision="fp16", calibration_images=None, batch_size=32):
    """Build TensorRT engine from ONNX model with specified precision."""
    with trt.Builder(TRT_LOGGER) as builder, builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)) as network, trt.OnnxParser(network, TRT_LOGGER) as parser:
        # Parse ONNX model
        with open(onnx_path, "rb") as f:
            if not parser.parse(f.read()):
                for error in range(parser.num_errors):
                    print(f"ONNX parse error: {parser.get_error(error)}")
                raise RuntimeError("Failed to parse ONNX model")

        # Configure builder flags
        config = builder.create_builder_config()
        config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)  # 1GB workspace

        if precision == "fp16":
            if not builder.platform_has_fast_fp16():
                print("Warning: Platform does not support fast FP16, falling back to FP32")
            else:
                config.set_flag(trt.BuilderFlag.FP16)
        elif precision == "int8":
            if not builder.platform_has_fast_int8():
                raise RuntimeError("Platform does not support INT8")
            config.set_flag(trt.BuilderFlag.INT8)
            if calibration_images is None:
                raise ValueError("Calibration images required for INT8 precision")
            calibrator = ResNet50Calibrator(calibration_images, batch_size)
            config.int8_calibrator = calibrator

        # Set optimization profile for dynamic batch size
        profile = builder.create_optimization_profile()
        input_name = network.get_input(0).name
        profile.set_shape(input_name, (1, 3, 224, 224), (batch_size, 3, 224, 224), (batch_size, 3, 224, 224))
        config.add_optimization_profile(profile)

        # Build engine
        print(f"Building TensorRT engine with {precision} precision...")
        engine = builder.build_engine(network, config)
        if engine is None:
            raise RuntimeError("Failed to build TensorRT engine")

        # Serialize and save engine
        with open(engine_path, "wb") as f:
            f.write(engine.serialize())
        print(f"Engine saved to {engine_path}")
        return engine

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Build TensorRT engine from ONNX model")
    parser.add_argument("--onnx", required=True, help="Path to ONNX model")
    parser.add_argument("--engine", required=True, help="Path to save TensorRT engine")
    parser.add_argument("--precision", choices=["fp32", "fp16", "int8"], default="fp16")
    parser.add_argument("--calibration_dir", help="Directory of calibration images for INT8")
    parser.add_argument("--batch_size", type=int, default=32)
    args = parser.parse_args()

    try:
        calibration_images = None
        if args.precision == "int8":
            if not args.calibration_dir:
                raise ValueError("--calibration_dir required for INT8")
            calibration_images = load_calibration_images(args.calibration_dir, num_images=1000)
        build_tensorrt_engine(args.onnx, args.engine, args.precision, calibration_images, args.batch_size)
    except Exception as e:
        print(f"Failed to build engine: {e}")
        exit(1)

import openvino as ov
import openvino.properties as props
from openvino.runtime import Core
import numpy as np
import argparse
import time
from PIL import Image
import os

def preprocess_image(image_path, input_shape=(3, 224, 224)):
    """Preprocess image for OpenVINO inference."""
    try:
        img = Image.open(image_path).resize((input_shape[2], input_shape[1]))
        img = np.array(img).transpose(2, 0, 1).astype(np.float32) / 255.0
        img = (img - np.array([0.485, 0.456, 0.406])[:, None, None]) / np.array([0.229, 0.224, 0.225])[:, None, None]
        return np.expand_dims(img, axis=0)  # Add batch dimension
    except Exception as e:
        raise RuntimeError(f"Failed to preprocess {image_path}: {e}")

def optimize_model_with_openvino(onnx_path, optimized_model_path, precision="fp16", device="CPU"):
    """Optimize ONNX model using OpenVINO 2024.3 quantization and compilation."""
    core = Core()

    # Read ONNX model
    print(f"Reading ONNX model from {onnx_path}...")
    model = core.read_model(onnx_path)

    # Apply quantization if INT8 is requested
    if precision == "int8":
        try:
            from openvino.tools import nncf
            print("Applying INT8 quantization with NNCF...")
            # Load calibration dataset (using same ImageNet subset as TensorRT example)
            calibration_images = []
            calib_dir = "imagenet_calib"
            if not os.path.exists(calib_dir):
                raise FileNotFoundError(f"Calibration directory {calib_dir} not found")
            for img_path in os.listdir(calib_dir)[:1000]:
                if img_path.endswith(".jpg"):
                    img = preprocess_image(os.path.join(calib_dir, img_path))
                    calibration_images.append(img)
            calibration_dataset = np.concatenate(calibration_images, axis=0)

            # Quantize model
            quantized_model = nncf.quantize(
                model,
                ov.nncf.Dataset(calibration_dataset),
                preset=ov.nncf.QuantizationPreset.PERFORMANCE,
                target_device=ov.nncf.TargetDevice(device)
            )
            model = quantized_model
            print("INT8 quantization complete")
        except ImportError:
            raise RuntimeError("NNCF not installed. Install with: pip install nncf")
        except Exception as e:
            raise RuntimeError(f"Quantization failed: {e}")

    # Compile model for target device
    print(f"Compiling model for {device} with {precision} precision...")
    compile_config = {props.compile_cache_enable: True}
    if precision == "fp16":
        compile_config[props.hint_performance_mode] = ov.hint.PerformanceMode.LATENCY
        compile_config[props.hint_num_requests] = 4
    compiled_model = core.compile_model(model=model, device_name=device, config=compile_config)

    # Save optimized model
    ov.save_model(model, optimized_model_path)
    print(f"Optimized model saved to {optimized_model_path}")
    return compiled_model

def run_openvino_inference(compiled_model, image_path, num_iterations=100):
    """Run inference with OpenVINO compiled model and return latency stats."""
    input_tensor = preprocess_image(image_path)
    infer_request = compiled_model.create_infer_request()

    # Warmup
    for _ in range(10):
        infer_request.infer({0: input_tensor})

    # Benchmark
    latencies = []
    for _ in range(num_iterations):
        start = time.perf_counter()
        infer_request.infer({0: input_tensor})
        end = time.perf_counter()
        latencies.append((end - start) * 1000)  # ms

    p50 = np.percentile(latencies, 50)
    p99 = np.percentile(latencies, 99)
    throughput = num_iterations / (sum(latencies) / 1000)
    print(f"Inference Results:")
    print(f"  p50 Latency: {p50:.2f} ms")
    print(f"  p99 Latency: {p99:.2f} ms")
    print(f"  Throughput: {throughput:.2f} inf/sec")
    return {"p50": p50, "p99": p99, "throughput": throughput}

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Optimize and run inference with OpenVINO 2024.3")
    parser.add_argument("--onnx", required=True, help="Path to ONNX model")
    parser.add_argument("--optimized_model", default="resnet50_ov.xml", help="Path to save optimized OpenVINO model")
    parser.add_argument("--precision", choices=["fp32", "fp16", "int8"], default="fp16")
    parser.add_argument("--device", default="CPU", choices=["CPU", "GPU", "NPU"])
    parser.add_argument("--image", required=True, help="Path to input image for inference")
    args = parser.parse_args()

    try:
        compiled_model = optimize_model_with_openvino(
            args.onnx, args.optimized_model, args.precision, args.device
        )
        run_openvino_inference(compiled_model, args.image)
    except Exception as e:
        print(f"OpenVINO pipeline failed: {e}")
        exit(1)

import tensorrt as trt
import openvino as ov
from openvino.runtime import Core
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np
import argparse
import time
import json
from typing import Dict, List

# Initialize loggers
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
OV_CORE = ov.Core()

class TensorRTInfer:
    """Wrapper for TensorRT inference."""
    def __init__(self, engine_path: str):
        self.engine = self._load_engine(engine_path)
        self.context = self.engine.create_execution_context()
        self.input_name = self.engine.get_tensor_name(0)
        self.output_name = self.engine.get_tensor_name(1)

        # Allocate device memory
        self.input_shape = self.engine.get_tensor_shape(self.input_name)
        self.output_shape = self.engine.get_tensor_shape(self.output_name)
        self.input_size = np.prod(self.input_shape) * np.float32().nbytes
        self.output_size = np.prod(self.output_shape) * np.float32().nbytes
        self.d_input = cuda.mem_alloc(self.input_size)
        self.d_output = cuda.mem_alloc(self.output_size)
        self.stream = cuda.Stream()

    def _load_engine(self, engine_path: str):
        with open(engine_path, "rb") as f:
            engine_data = f.read()
        runtime = trt.Runtime(TRT_LOGGER)
        engine = runtime.deserialize_cuda_engine(engine_data)
        if engine is None:
            raise RuntimeError(f"Failed to load TensorRT engine from {engine_path}")
        return engine

    def infer(self, input_tensor: np.ndarray) -> np.ndarray:
        # Copy input to device
        cuda.memcpy_htod_async(self.d_input, input_tensor, self.stream)
        # Set tensor addresses
        self.context.set_tensor_address(self.input_name, int(self.d_input))
        self.context.set_tensor_address(self.output_name, int(self.d_output))
        # Run inference
        self.context.execute_async_v3(self.stream.handle)
        # Copy output to host
        output = np.empty(self.output_shape, dtype=np.float32)
        cuda.memcpy_dtoh_async(output, self.d_output, self.stream)
        self.stream.synchronize()
        return output

class OpenVINOInfer:
    """Wrapper for OpenVINO inference."""
    def __init__(self, model_path: str, device: str = "CPU"):
        self.compiled_model = OV_CORE.compile_model(model=model_path, device_name=device)
        self.infer_request = self.compiled_model.create_infer_request()

    def infer(self, input_tensor: np.ndarray) -> np.ndarray:
        return self.infer_request.infer({0: input_tensor})[0]

def run_benchmark(infer_obj, num_iterations: int = 1000, batch_size: int = 32) -> Dict[str, float]:
    """Run benchmark and return latency/throughput stats."""
    # Generate dummy input
    dummy_input = np.random.randn(batch_size, 3, 224, 224).astype(np.float32)

    # Warmup
    for _ in range(10):
        infer_obj.infer(dummy_input)

    # Benchmark
    latencies = []
    start_total = time.perf_counter()
    for _ in range(num_iterations):
        iter_start = time.perf_counter()
        infer_obj.infer(dummy_input)
        iter_end = time.perf_counter()
        latencies.append((iter_end - iter_start) * 1000)  # ms
    total_time = time.perf_counter() - start_total

    return {
        "p50_latency_ms": np.percentile(latencies, 50),
        "p99_latency_ms": np.percentile(latencies, 99),
        "throughput_inf_per_sec": (num_iterations * batch_size) / total_time,
        "avg_latency_ms": np.mean(latencies)
    }

def main():
    parser = argparse.ArgumentParser(description="Cross-framework benchmark for TensorRT and OpenVINO")
    parser.add_argument("--trt_engine", help="Path to TensorRT engine")
    parser.add_argument("--ov_model", help="Path to OpenVINO model (XML)")
    parser.add_argument("--device", default="CUDA", choices=["CUDA", "CPU"])
    parser.add_argument("--iterations", type=int, default=1000)
    parser.add_argument("--batch_size", type=int, default=32)
    parser.add_argument("--output", default="benchmark_results.json")
    args = parser.parse_args()

    results = []

    if args.trt_engine and args.device == "CUDA":
        print("Running TensorRT benchmark...")
        try:
            trt_infer = TensorRTInfer(args.trt_engine)
            trt_stats = run_benchmark(trt_infer, args.iterations, args.batch_size)
            trt_stats["framework"] = "TensorRT"
            trt_stats["device"] = args.device
            results.append(trt_stats)
            print(f"TensorRT Results: {trt_stats}")
        except Exception as e:
            print(f"TensorRT benchmark failed: {e}")

    if args.ov_model:
        print(f"Running OpenVINO benchmark on {args.device}...")
        try:
            ov_infer = OpenVINOInfer(args.ov_model, args.device)
            ov_stats = run_benchmark(ov_infer, args.iterations, args.batch_size)
            ov_stats["framework"] = "OpenVINO"
            ov_stats["device"] = args.device
            results.append(ov_stats)
            print(f"OpenVINO Results: {ov_stats}")
        except Exception as e:
            print(f"OpenVINO benchmark failed: {e}")

    # Save results
    with open(args.output, "w") as f:
        json.dump(results, f, indent=2)
    print(f"Results saved to {args.output}")

if __name__ == "__main__":
    main()

import subprocess
import os

def detect_inference_runtime():
    # Check for NVIDIA GPU
    try:
        nvidia_smi = subprocess.check_output(["nvidia-smi", "-L"]).decode()
        if "NVIDIA" in nvidia_smi:
            return "tensorrt"
    except:
        pass
    # Check for Intel CPU
    if "INTEL" in os.popen("cat /proc/cpuinfo | grep vendor_id").read().upper():
        return "openvino"
    return "onnxruntime"

from openvino.tools import nncf
import openvino as ov

def quantize_with_nncf(onnx_path, calibration_data):
    core = ov.Core()
    model = core.read_model(onnx_path)
    # Create NNCF dataset from production samples
    nncf_dataset = nncf.Dataset(calibration_data)
    # Quantize for performance (vs accuracy preset)
    quantized_model = nncf.quantize(
        model,
        nncf_dataset,
        preset=nncf.QuantizationPreset.PERFORMANCE,
        target_device=nncf.TargetDevice.CPU
    )
    ov.save_model(quantized_model, "resnet50_int8.xml")
    return quantized_model

from openvino.runtime import Core
import openvino.tools.benchmark as benchmark

def run_ov_benchmark(model_path, device="CPU", num_iterations=1000):
    core = Core()
    compiled_model = core.compile_model(model_path, device)
    # Configure benchmark parameters
    config = benchmark.BenchmarkConfig(
        num_iterations=num_iterations,
        batch_size=32,
        api_type=benchmark.ApiType.SYNC
    )
    results = benchmark.run(compiled_model, config)
    print(f"p99 Latency: {results.p99_latency_ms:.2f} ms")
    print(f"Throughput: {results.throughput_inf_per_sec:.2f} inf/sec")
    return results