quantification-MobileNetV5-V4 / scripts /benchmark_scale_machine_1.py

Initial release: technical documentation, quantization scripts, and NPU benchmarks (part 2)

2248c41 verified 3 days ago

7.61 kB

import os
import sys
import time
import argparse
from pathlib import Path
import numpy as np
import pandas as pd
from PIL import Image

# 1. Résolution dynamique et relative des répertoires (BYOM / Portabilité absolue)
BASE_DIR = Path(__file__).resolve().parent.parent

def resolve_models_dir(custom_path: str = None) -> Path:
    if custom_path and Path(custom_path).exists():
        return Path(custom_path)
    env_root = os.environ.get("MODELS_ROOT")
    if env_root and Path(env_root).exists():
        return Path(env_root)
    # Candidats relatifs prioritaires
    candidates = [
        BASE_DIR / "models",
        BASE_DIR.parent / "fine tuning rock model",
        BASE_DIR.parent / "models",
        BASE_DIR / "ScaleDetection_MobileNetV4" / "onnx",
        BASE_DIR.parent / "fine tuning rock model" / "ScaleDetection_MobileNetV4" / "onnx"
    ]
    for c in candidates:
        if c.exists():
            return c
    return BASE_DIR / "models"

# 2. Support dynamique de cuDNN si installé dans l'environnement virtuel actif
if "VIRTUAL_ENV" in os.environ:
    venv_cudnn = Path(os.environ["VIRTUAL_ENV"]) / "Lib" / "site-packages" / "nvidia" / "cudnn" / "bin"
    if venv_cudnn.exists():
        os.environ["PATH"] = str(venv_cudnn) + os.pathsep + os.environ.get("PATH", "")

try:
    import onnxruntime as ort
except ImportError:
    print("onnxruntime not installed. Please install onnxruntime-gpu or onnxruntime.")
    sys.exit(0)

OUTPUT_DIR = BASE_DIR / "results" / "ScaleDetection" / "machine_1"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
CSV_OUT = OUTPUT_DIR / "benchmark_machine_1_scale.csv"

def find_scale_model(fmt: str, models_root: Path) -> Path:
    candidates = [
        models_root / f"v4_small_scale_{fmt.lower()}.onnx",
        models_root / "ScaleDetection_MobileNetV4" / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx",
        models_root / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx",
        BASE_DIR / "models" / f"v4_small_scale_{fmt.lower()}.onnx",
        BASE_DIR.parent / "fine tuning rock model" / "ScaleDetection_MobileNetV4" / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx"
    ]
    for c in candidates:
        if c.exists():
            return c.resolve()
    return candidates[0]

def load_real_images(test_dir: Path, target_shape=(1, 1024, 1024, 3)):
    """
    Charge et prétraite de véritables images réelles (INT8 [-128, 127]) depuis un répertoire de test.
    """
    if not test_dir or not test_dir.exists():
        return []
    
    extensions = ("*.jpg", "*.jpeg", "*.png", "*.bmp", "*.webp")
    image_files = []
    for ext in extensions:
        image_files.extend(list(test_dir.glob(ext)))
    
    if not image_files:
        return []
    
    preprocessed_tensors = []
    _, h, w, _ = target_shape
    for img_path in sorted(image_files)[:50]:
        try:
            with Image.open(img_path) as pil_img:
                img_rgb = pil_img.convert("RGB").resize((w, h), Image.Resampling.BILINEAR)
                arr = np.array(img_rgb, dtype=np.float32)
                # Formule bit-accurate : conversion [0, 255] float vers [-128, 127] signed int8
                arr_int8 = np.clip(np.round(arr - 128.0), -128, 127).astype(np.int8)
                tensor = np.expand_dims(arr_int8, axis=0)
                preprocessed_tensors.append(tensor)
        except Exception as e:
            print(f"[!] Warning: Impossible de charger {img_path.name}: {e}")
            
    return preprocessed_tensors

def main():
    parser = argparse.ArgumentParser(description="Lithotheque Scale Detection ONNX Benchmark (Machine 1)")
    parser.add_argument("--models_dir", type=str, default=None, help="Chemin relatif ou absolu vers les modèles")
    parser.add_argument("--test_dir", type=str, default=None, help="Chemin vers le répertoire d'images réelles de test")
    parser.add_argument("--runs", type=int, default=100, help="Nombre d'itérations d'inférence par backend")
    args = parser.parse_args()

    models_root = resolve_models_dir(args.models_dir)
    print(f"[*] Répertoire modèles résolu : {models_root}")

    # Résolution relative du répertoire de test
    test_dir_path = None
    if args.test_dir:
        test_dir_path = Path(args.test_dir)
    else:
        test_candidates = [
            BASE_DIR / "dataset" / "echelle" / "test",
            BASE_DIR.parent / "fine tuning rock model" / "dataset" / "echelle" / "test",
            BASE_DIR / "test_images"
        ]
        for tc in test_candidates:
            if tc.exists():
                test_dir_path = tc
                break

    real_samples = load_real_images(test_dir_path) if test_dir_path else []
    if real_samples:
        print(f"[*] Jeu de données de test réel chargé : {len(real_samples)} images réelles depuis {test_dir_path}")
    else:
        print("[!] Aucun jeu de test réel trouvé, utilisation de tenseurs de calibration normalisés.")

    input_shape = (1, 1024, 1024, 3)
    configs = [
        ("INT8", "GPU_RTX5060", ["CUDAExecutionProvider", "CPUExecutionProvider"]),
        ("W4A8", "GPU_RTX5060", ["CUDAExecutionProvider", "CPUExecutionProvider"]),
        ("INT8", "CPU_AVX2", ["CPUExecutionProvider"]),
        ("W4A8", "CPU_AVX2", ["CPUExecutionProvider"])
    ]

    print("\n=== Démarrage du Banc d'Essai ScaleDetection (Machine 1) ===")
    records = []

    for fmt, backend_label, providers in configs:
        model_path = find_scale_model(fmt, models_root)
        if not model_path.exists():
            print(f"[!] Modèle introuvable, passage : {model_path}")
            continue

        print(f"\n--> Évaluation de {fmt} sur {backend_label} (Providers: {providers})")
        sess = ort.InferenceSession(str(model_path), providers=providers)
        active = sess.get_providers()[0]
        print(f"Active Provider: {active}")
        input_name = sess.get_inputs()[0].name

        # Préchauffage (Warmup)
        warmup_input = real_samples[0] if real_samples else np.zeros(input_shape, dtype=np.int8)
        for _ in range(5):
            _ = sess.run(None, {input_name: warmup_input})

        latencies = []
        for i in range(1, args.runs + 1):
            if real_samples:
                inp = real_samples[(i - 1) % len(real_samples)]
            else:
                inp = np.zeros(input_shape, dtype=np.int8)

            t0 = time.perf_counter()
            _ = sess.run(None, {input_name: inp})
            lat = (time.perf_counter() - t0) * 1000.0
            latencies.append(lat)

            records.append({
                "Model": "ScaleDetection_MobileNetV4",
                "Task": "Scale_Metrology",
                "Format": fmt,
                "Backend": backend_label,
                "Inference_ID": i,
                "Latency_ms": round(lat, 3),
                "Precision_Type": f"{fmt}_Tensor_Cores" if "GPU" in backend_label else f"{fmt}_AVX2",
                "Hardware_Environment": "Machine_1_Workstation"
            })

        print(f"Résultats pour {fmt} sur {backend_label}:")
        print(f"  Moyenne: {np.mean(latencies):.2f} ms")
        print(f"  Min:     {np.min(latencies):.2f} ms")
        print(f"  Max:     {np.max(latencies):.2f} ms")
        print(f"  p95:     {np.percentile(latencies, 95):.2f} ms")
        print(f"  FPS:     {1000.0 / np.mean(latencies):.1f} FPS")

    if records:
        df_results = pd.DataFrame(records)
        df_results.to_csv(CSV_OUT, index=False)
        print(f"\n[SUCCÈS] {len(records)} enregistrements de latence empirique sauvegardés dans : {CSV_OUT}")

if __name__ == "__main__":
    main()

	import os
	import sys
	import time
	import argparse
	from pathlib import Path
	import numpy as np
	import pandas as pd
	from PIL import Image

	# 1. Résolution dynamique et relative des répertoires (BYOM / Portabilité absolue)
	BASE_DIR = Path(__file__).resolve().parent.parent

	def resolve_models_dir(custom_path: str = None) -> Path:
	if custom_path and Path(custom_path).exists():
	return Path(custom_path)
	env_root = os.environ.get("MODELS_ROOT")
	if env_root and Path(env_root).exists():
	return Path(env_root)
	# Candidats relatifs prioritaires
	candidates = [
	BASE_DIR / "models",
	BASE_DIR.parent / "fine tuning rock model",
	BASE_DIR.parent / "models",
	BASE_DIR / "ScaleDetection_MobileNetV4" / "onnx",
	BASE_DIR.parent / "fine tuning rock model" / "ScaleDetection_MobileNetV4" / "onnx"
	]
	for c in candidates:
	if c.exists():
	return c
	return BASE_DIR / "models"

	# 2. Support dynamique de cuDNN si installé dans l'environnement virtuel actif
	if "VIRTUAL_ENV" in os.environ:
	venv_cudnn = Path(os.environ["VIRTUAL_ENV"]) / "Lib" / "site-packages" / "nvidia" / "cudnn" / "bin"
	if venv_cudnn.exists():
	os.environ["PATH"] = str(venv_cudnn) + os.pathsep + os.environ.get("PATH", "")

	try:
	import onnxruntime as ort
	except ImportError:
	print("onnxruntime not installed. Please install onnxruntime-gpu or onnxruntime.")
	sys.exit(0)

	OUTPUT_DIR = BASE_DIR / "results" / "ScaleDetection" / "machine_1"
	OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
	CSV_OUT = OUTPUT_DIR / "benchmark_machine_1_scale.csv"

	def find_scale_model(fmt: str, models_root: Path) -> Path:
	candidates = [
	models_root / f"v4_small_scale_{fmt.lower()}.onnx",
	models_root / "ScaleDetection_MobileNetV4" / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx",
	models_root / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx",
	BASE_DIR / "models" / f"v4_small_scale_{fmt.lower()}.onnx",
	BASE_DIR.parent / "fine tuning rock model" / "ScaleDetection_MobileNetV4" / "onnx" / f"v4_small_scale_{fmt.lower()}.onnx"
	]
	for c in candidates:
	if c.exists():
	return c.resolve()
	return candidates[0]

	def load_real_images(test_dir: Path, target_shape=(1, 1024, 1024, 3)):
	"""
	Charge et prétraite de véritables images réelles (INT8 [-128, 127]) depuis un répertoire de test.
	"""
	if not test_dir or not test_dir.exists():
	return []

	extensions = (".jpg", ".jpeg", ".png", ".bmp", "*.webp")
	image_files = []
	for ext in extensions:
	image_files.extend(list(test_dir.glob(ext)))

	if not image_files:
	return []

	preprocessed_tensors = []
	_, h, w, _ = target_shape
	for img_path in sorted(image_files)[:50]:
	try:
	with Image.open(img_path) as pil_img:
	img_rgb = pil_img.convert("RGB").resize((w, h), Image.Resampling.BILINEAR)
	arr = np.array(img_rgb, dtype=np.float32)
	# Formule bit-accurate : conversion [0, 255] float vers [-128, 127] signed int8
	arr_int8 = np.clip(np.round(arr - 128.0), -128, 127).astype(np.int8)
	tensor = np.expand_dims(arr_int8, axis=0)
	preprocessed_tensors.append(tensor)
	except Exception as e:
	print(f"[!] Warning: Impossible de charger {img_path.name}: {e}")

	return preprocessed_tensors

	def main():
	parser = argparse.ArgumentParser(description="Lithotheque Scale Detection ONNX Benchmark (Machine 1)")
	parser.add_argument("--models_dir", type=str, default=None, help="Chemin relatif ou absolu vers les modèles")
	parser.add_argument("--test_dir", type=str, default=None, help="Chemin vers le répertoire d'images réelles de test")
	parser.add_argument("--runs", type=int, default=100, help="Nombre d'itérations d'inférence par backend")
	args = parser.parse_args()

	models_root = resolve_models_dir(args.models_dir)
	print(f"[*] Répertoire modèles résolu : {models_root}")

	# Résolution relative du répertoire de test
	test_dir_path = None
	if args.test_dir:
	test_dir_path = Path(args.test_dir)
	else:
	test_candidates = [
	BASE_DIR / "dataset" / "echelle" / "test",
	BASE_DIR.parent / "fine tuning rock model" / "dataset" / "echelle" / "test",
	BASE_DIR / "test_images"
	]
	for tc in test_candidates:
	if tc.exists():
	test_dir_path = tc
	break

	real_samples = load_real_images(test_dir_path) if test_dir_path else []
	if real_samples:
	print(f"[*] Jeu de données de test réel chargé : {len(real_samples)} images réelles depuis {test_dir_path}")
	else:
	print("[!] Aucun jeu de test réel trouvé, utilisation de tenseurs de calibration normalisés.")

	input_shape = (1, 1024, 1024, 3)
	configs = [
	("INT8", "GPU_RTX5060", ["CUDAExecutionProvider", "CPUExecutionProvider"]),
	("W4A8", "GPU_RTX5060", ["CUDAExecutionProvider", "CPUExecutionProvider"]),
	("INT8", "CPU_AVX2", ["CPUExecutionProvider"]),
	("W4A8", "CPU_AVX2", ["CPUExecutionProvider"])
	]

	print("\n=== Démarrage du Banc d'Essai ScaleDetection (Machine 1) ===")
	records = []

	for fmt, backend_label, providers in configs:
	model_path = find_scale_model(fmt, models_root)
	if not model_path.exists():
	print(f"[!] Modèle introuvable, passage : {model_path}")
	continue

	print(f"\n--> Évaluation de {fmt} sur {backend_label} (Providers: {providers})")
	sess = ort.InferenceSession(str(model_path), providers=providers)
	active = sess.get_providers()[0]
	print(f"Active Provider: {active}")
	input_name = sess.get_inputs()[0].name

	# Préchauffage (Warmup)
	warmup_input = real_samples[0] if real_samples else np.zeros(input_shape, dtype=np.int8)
	for _ in range(5):
	_ = sess.run(None, {input_name: warmup_input})

	latencies = []
	for i in range(1, args.runs + 1):
	if real_samples:
	inp = real_samples[(i - 1) % len(real_samples)]
	else:
	inp = np.zeros(input_shape, dtype=np.int8)

	t0 = time.perf_counter()
	_ = sess.run(None, {input_name: inp})
	lat = (time.perf_counter() - t0) * 1000.0
	latencies.append(lat)

	records.append({
	"Model": "ScaleDetection_MobileNetV4",
	"Task": "Scale_Metrology",
	"Format": fmt,
	"Backend": backend_label,
	"Inference_ID": i,
	"Latency_ms": round(lat, 3),
	"Precision_Type": f"{fmt}_Tensor_Cores" if "GPU" in backend_label else f"{fmt}_AVX2",
	"Hardware_Environment": "Machine_1_Workstation"
	})

	print(f"Résultats pour {fmt} sur {backend_label}:")
	print(f" Moyenne: {np.mean(latencies):.2f} ms")
	print(f" Min: {np.min(latencies):.2f} ms")
	print(f" Max: {np.max(latencies):.2f} ms")
	print(f" p95: {np.percentile(latencies, 95):.2f} ms")
	print(f" FPS: {1000.0 / np.mean(latencies):.1f} FPS")

	if records:
	df_results = pd.DataFrame(records)
	df_results.to_csv(CSV_OUT, index=False)
	print(f"\n[SUCCÈS] {len(records)} enregistrements de latence empirique sauvegardés dans : {CSV_OUT}")

	if __name__ == "__main__":
	main()