""" HHD Downsample Factor Comparison Reads a video file and runs the full pipeline at multiple hhd_downsample factors. Computes per-frame HHD timing and vnormal quality metrics against the baseline (factor=1), then plots the results. Usage: python test_downsample.py -v jx01_hex1.5_2.mp4 -c ./config/prototype_of3_720p.json -f 1,2,3,4 -n 50 -w 30 """ import os import sys os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" _local_src = os.path.join(os.path.dirname(__file__), "src") if _local_src not in sys.path: sys.path.insert(0, _local_src) import cv2 import orisys import numpy as np import time import argparse import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt def run_sensor(vid_src, config, cal, downsample, n_warmup, n_measure): """Run sensor for n_warmup + n_measure frames. Capture timing and vnormal for every measurement frame.""" try: vid = int(vid_src) except ValueError: vid = vid_src sensor = orisys.Sensor( vid, config_name=config, cal_path=cal, isstitch=True, cuda=False, verbose=False, hhd_downsample=downsample, ) hhd_times = [] vnormals = [] frame_idx = 0 n_total = n_warmup + n_measure while frame_idx < n_total: try: sensor.get_img() except Exception: break sensor.compute_flow() t0 = time.perf_counter() sensor.compute_hhd() t_hhd = time.perf_counter() - t0 if frame_idx >= n_warmup: hhd_times.append(t_hhd) vn = sensor.vnormal.copy() if sensor.vnormal is not None else None vnormals.append(vn) frame_idx += 1 sensor.disconnect() return {"hhd_times": hhd_times, "vnormals": vnormals} def per_frame_rmse(a, b): if a is None or b is None: return float("nan") diff = a.astype(np.float64) - b.astype(np.float64) return float(np.sqrt(np.mean(diff ** 2))) def per_frame_max_abs_err(a, b): if a is None or b is None: return float("nan") return float(np.max(np.abs(a.astype(np.float64) - b.astype(np.float64)))) def per_frame_cosine_sim(a, b): if a is None or b is None: return float("nan") a64 = a.astype(np.float64) b64 = b.astype(np.float64) dot = np.sum(a64 * b64, axis=-1) mag_a = np.linalg.norm(a64, axis=-1) mag_b = np.linalg.norm(b64, axis=-1) denom = mag_a * mag_b valid = denom > 1e-12 if not np.any(valid): return float("nan") return float(np.mean(dot[valid] / denom[valid])) def main(): parser = argparse.ArgumentParser(description="Compare HHD downsample factors") parser.add_argument("--video", "-v", type=str, default="2", help="Camera ID or video file path") parser.add_argument("--config", "-c", type=str, default="./config/prototype_of3_720p.json") parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy") parser.add_argument("--factors", "-f", type=str, default="1,2,3", help="Comma-separated downsample factors") parser.add_argument("--warmup", "-w", type=int, default=5, help="Warmup frames (skipped)") parser.add_argument("--frames", "-n", type=int, default=30, help="Measurement frames") parser.add_argument("--output", "-o", type=str, default="downsample_comparison.png", help="Output plot filename") args = parser.parse_args() factors = [int(f) for f in args.factors.split(",")] print(f"Video: {args.video}") print(f"Config: {args.config}") print(f"Factors: {factors}") print(f"Warmup: {args.warmup} frames, Measure: {args.frames} frames") print() # --- collect data --- results = {} for factor in factors: print(f"--- Running factor {factor} ---") results[factor] = run_sensor( args.video, args.config, args.cal, factor, args.warmup, args.frames, ) n = len(results[factor]["hhd_times"]) if n > 0: arr = np.array(results[factor]["hhd_times"]) * 1000 print(f" {n} frames: hhd mean={arr.mean():.2f}ms, median={np.median(arr):.2f}ms, " f"min={arr.min():.2f}ms, max={arr.max():.2f}ms") else: print(" No frames captured") print() # --- timing summary table --- print("=" * 72) print(f"{'Factor':<10} {'Mean HHD (ms)':<16} {'Median (ms)':<14} {'Speedup':<10}") print("-" * 72) baseline_median = None for factor in factors: arr = np.array(results[factor]["hhd_times"]) * 1000 if len(arr) == 0: continue med = float(np.median(arr)) if baseline_median is None: baseline_median = med speedup = baseline_median / med if med > 0 else 0 print(f"{factor:<10} {arr.mean():<16.2f} {med:<14.2f} {speedup:<10.2f}x") print("=" * 72) print() # --- compute per-frame quality metrics vs baseline (factor=1) --- has_baseline = 1 in factors and len(factors) >= 2 metrics = {} # factor -> {"rmse": [...], "max_err": [...], "cosine": [...]} if has_baseline: base_vn = results[1]["vnormals"] n_base = len(base_vn) for factor in factors: if factor == 1: continue test_vn = results[factor]["vnormals"] n_common = min(n_base, len(test_vn)) rmse_arr, maxerr_arr, cos_arr = [], [], [] for i in range(n_common): rmse_arr.append(per_frame_rmse(base_vn[i], test_vn[i])) maxerr_arr.append(per_frame_max_abs_err(base_vn[i], test_vn[i])) cos_arr.append(per_frame_cosine_sim(base_vn[i], test_vn[i])) metrics[factor] = { "rmse": np.array(rmse_arr), "max_err": np.array(maxerr_arr), "cosine": np.array(cos_arr), } # print per-factor summary print(f"{'Factor':<10} {'Mean RMSE':<14} {'Mean MaxErr':<14} {'Mean CosSim':<14}") print("-" * 52) for factor in sorted(metrics): m = metrics[factor] print(f"{factor:<10} {np.nanmean(m['rmse']):<14.6f} {np.nanmean(m['max_err']):<14.4f} " f"{np.nanmean(m['cosine']):<14.6f}") print("=" * 52) print() # --- plot --- n_quality_factors = len(metrics) n_plots = 1 + (3 if has_baseline else 0) # timing + (rmse, max_err, cosine) fig, axes = plt.subplots(n_plots, 1, figsize=(12, 3.5 * n_plots), sharex=True) if n_plots == 1: axes = [axes] colors = plt.cm.tab10.colors frame_nums = None # 1) HHD timing per frame ax = axes[0] for i, factor in enumerate(factors): arr = np.array(results[factor]["hhd_times"]) * 1000 frames = np.arange(len(arr)) if frame_nums is None: frame_nums = frames ax.plot(frames, arr, label=f"ds={factor}", color=colors[i % len(colors)], linewidth=1) ax.set_ylabel("HHD time (ms)") ax.set_title("Per-frame HHD computation time") ax.legend(loc="upper right") ax.grid(True, alpha=0.3) if has_baseline: # 2) RMSE ax = axes[1] for i, factor in enumerate(sorted(metrics)): m = metrics[factor] frames = np.arange(len(m["rmse"])) ci = factors.index(factor) if factor in factors else i ax.plot(frames, m["rmse"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) ax.set_ylabel("RMSE (vnormal)") ax.set_title("Per-frame RMSE vs baseline (ds=1)") ax.legend(loc="upper right") ax.grid(True, alpha=0.3) # 3) Max absolute error ax = axes[2] for i, factor in enumerate(sorted(metrics)): m = metrics[factor] frames = np.arange(len(m["max_err"])) ci = factors.index(factor) if factor in factors else i ax.plot(frames, m["max_err"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) ax.set_ylabel("Max Abs Error (vnormal)") ax.set_title("Per-frame max absolute error vs baseline (ds=1)") ax.legend(loc="upper right") ax.grid(True, alpha=0.3) # 4) Cosine similarity ax = axes[3] for i, factor in enumerate(sorted(metrics)): m = metrics[factor] frames = np.arange(len(m["cosine"])) ci = factors.index(factor) if factor in factors else i ax.plot(frames, m["cosine"], label=f"ds={factor}", color=colors[ci % len(colors)], linewidth=1) ax.set_ylabel("Cosine Similarity") ax.set_title("Per-frame cosine similarity vs baseline (ds=1)") ax.legend(loc="upper right") ax.grid(True, alpha=0.3) ax.set_ylim(bottom=min(0.999, ax.get_ylim()[0])) axes[-1].set_xlabel("Frame") fig.tight_layout() fig.savefig(args.output, dpi=150) print(f"Plot saved to: {args.output}") if __name__ == "__main__": main()