Shape_Model_2D/pm2d/bench.py

"""Benchmark suite per LineShapeMatcher.

Usage:
    python -m pm2d.bench [--quick]

Misura tempi find() su 3 template-tipo × 3 scene-tipo × N config:
- Template: rettangolo 80×80, L-shape 120×120, cerchio 150×150
- Scene: pulita 800×600, cluttered 1080×1920, multi-pezzo 1080×1920
- Config: baseline, polarity, gpu, pyramid_propagate, greediness=0.7

Per ogni config stampa: ms/find, ms per fase (profile), n. match.
Output tabellare per detectare regressioni in CI.
"""

from __future__ import annotations

import argparse
import time

import cv2
import numpy as np

from pm2d.line_matcher import LineShapeMatcher, opencl_available


# ---------- Sintetizzatori template/scena ----------

def _tpl_rect() -> np.ndarray:
    t = np.zeros((80, 80, 3), np.uint8)
    cv2.rectangle(t, (15, 15), (65, 65), (255, 255, 255), 3)
    return t


def _tpl_lshape() -> np.ndarray:
    t = np.zeros((120, 120, 3), np.uint8)
    cv2.rectangle(t, (20, 20), (50, 100), (255, 255, 255), -1)
    cv2.rectangle(t, (20, 70), (100, 100), (255, 255, 255), -1)
    return t


def _tpl_circle() -> np.ndarray:
    t = np.zeros((150, 150, 3), np.uint8)
    cv2.circle(t, (75, 75), 60, (255, 255, 255), 4)
    return t


def _scene_clean(W: int, H: int, n_pieces: int = 1) -> np.ndarray:
    np.random.seed(0)
    s = np.zeros((H, W, 3), np.uint8)
    for _ in range(n_pieces):
        cx = np.random.randint(80, W - 80)
        cy = np.random.randint(80, H - 80)
        cv2.rectangle(s, (cx - 25, cy - 25), (cx + 25, cy + 25), (255, 255, 255), 3)
    return s


def _scene_cluttered(W: int, H: int) -> np.ndarray:
    np.random.seed(0)
    s = np.random.randint(50, 200, (H, W, 3), np.uint8)
    cv2.rectangle(s, (300, 200), (350, 250), (255, 255, 255), 3)
    cv2.rectangle(s, (1500, 800), (1550, 850), (255, 255, 255), 3)
    return s


# ---------- Single benchmark ----------

def _bench_config(template, scene, config_name: str,
                  init_kw: dict, find_kw: dict,
                  n_iter: int = 5) -> dict:
    m = LineShapeMatcher(**init_kw)
    t0 = time.perf_counter()
    n_var = m.train(template)
    t_train = time.perf_counter() - t0

    # Warmup (Numba JIT)
    m.find(scene, **find_kw)
    m.find(scene, **find_kw)

    # Run
    times_ms = []
    for _ in range(n_iter):
        t0 = time.perf_counter()
        matches = m.find(scene, **find_kw)
        times_ms.append((time.perf_counter() - t0) * 1000.0)

    # Profile (1 iter)
    m.find(scene, profile=True, **find_kw)
    prof = m.get_last_profile() or {}

    return {
        "config": config_name,
        "n_variants": n_var,
        "t_train_s": round(t_train, 3),
        "ms_avg": round(float(np.mean(times_ms)), 1),
        "ms_min": round(float(np.min(times_ms)), 1),
        "ms_max": round(float(np.max(times_ms)), 1),
        "n_matches": len(matches),
        "profile_ms": {k: round(v, 1) for k, v in prof.items()},
    }


# ---------- Suite ----------

CONFIGS = [
    ("baseline",
     {"angle_step_deg": 10, "pyramid_levels": 2},
     {"min_score": 0.4, "verify_threshold": 0.2}),
    ("polarity",
     {"angle_step_deg": 10, "pyramid_levels": 2, "use_polarity": True},
     {"min_score": 0.4, "verify_threshold": 0.2}),
    ("propagate",
     {"angle_step_deg": 10, "pyramid_levels": 3},
     {"min_score": 0.4, "verify_threshold": 0.2,
      "pyramid_propagate": True, "propagate_topk": 4}),
    ("greedy_07",
     {"angle_step_deg": 10, "pyramid_levels": 2},
     {"min_score": 0.4, "verify_threshold": 0.2, "greediness": 0.7}),
    ("stride2",
     {"angle_step_deg": 10, "pyramid_levels": 2},
     {"min_score": 0.4, "verify_threshold": 0.2, "coarse_stride": 2}),
]

if opencl_available():
    CONFIGS.append(
        ("gpu_umat",
         {"angle_step_deg": 10, "pyramid_levels": 2, "use_gpu": True},
         {"min_score": 0.4, "verify_threshold": 0.2})
    )


SCENARIOS = [
    ("rect_80 vs scene_800x600", _tpl_rect, lambda: _scene_clean(800, 600, 1)),
    ("lshape_120 vs scene_1080x1920_clutter",
     _tpl_lshape, lambda: _scene_cluttered(1920, 1080)),
    ("circle_150 vs scene_clean_3pieces",
     _tpl_circle, lambda: _scene_clean(1920, 1080, 3)),
]


def run(quick: bool = False) -> int:
    n_iter = 2 if quick else 5
    print(f"=== PM2D Benchmark Suite ({len(SCENARIOS)} scenarios x "
          f"{len(CONFIGS)} configs, n_iter={n_iter}) ===\n")
    rows = []
    for sc_name, tpl_fn, scn_fn in SCENARIOS:
        template = tpl_fn()
        scene = scn_fn()
        print(f"--- Scenario: {sc_name} (tpl={template.shape}, "
              f"scn={scene.shape}) ---")
        for cfg_name, init_kw, find_kw in CONFIGS:
            r = _bench_config(template, scene, cfg_name, init_kw, find_kw,
                              n_iter=n_iter)
            r["scenario"] = sc_name
            rows.append(r)
            prof_str = " ".join(
                f"{k}={v:.1f}" for k, v in r["profile_ms"].items()
            )
            print(f"  {cfg_name:14s}  {r['ms_avg']:6.1f}ms  "
                  f"(min {r['ms_min']:.1f} max {r['ms_max']:.1f})  "
                  f"vars={r['n_variants']:3d}  "
                  f"matches={r['n_matches']:2d}")
            if prof_str:
                print(f"     profile: {prof_str}")
        print()
    print("=== Done ===")
    return 0


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description="PM2D benchmark suite")
    p.add_argument("--quick", action="store_true",
                   help="2 iterazioni per config invece di 5 (smoke test)")
    args = p.parse_args(argv)
    return run(quick=args.quick)


if __name__ == "__main__":
    import sys
    sys.exit(main())