diff --git a/examples/finger_pipeline.py b/examples/finger_pipeline.py index 12c919a..1bee4d3 100644 --- a/examples/finger_pipeline.py +++ b/examples/finger_pipeline.py @@ -14,39 +14,118 @@ Run: python examples/finger_pipeline.py --camera_id 0 --config finger python examples/finger_pipeline.py --video path/to/clip.mp4 python examples/finger_pipeline.py --no-show # headless: print only + python examples/finger_pipeline.py --config path/to/sensor.json --video clip.mp4 --cal-path model.json --zero -Keys: 'q' quit, 'r' reset the tracker. +Geometry comes from the selected bundle's manifest.json assets_dir; its UV cache follows the +geometry directory. Schema 3 requires matching processing config/geometry and a session zero. +Use --zero only when the first 30 frames after warmup are known to be unloaded. Without a zero, +schema 3 reports zero_unavailable. Use --help-advanced for SN selection and diagnostic overrides. +Keys: 'q' quit, 'r' reset the tracker and clear the schema 3 session reference. """ import argparse +import math import os +import statistics import sys +from pathlib import Path os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") import cv2 - from _display import draw_flow_arrows + +from orisys.configs import resolve_sensor_bundle from orisys.finger import FingerRuntime +def configured_export_dir(config): + """Reuse the selected bundle's geometry, including an explicit sensor.json path.""" + path = Path(config) + bundle = resolve_sensor_bundle(str(path.resolve().parent) if path.is_file() else config) + if bundle is None: + return None + if path.is_file() and bundle.sensor_json != path.resolve(): + return None + if bundle.manifest_json is not None and bundle.assets_dir is None: + raise ValueError(f"No usable assets_dir in {bundle.manifest_json}; use --export-dir") + return str(bundle.assets_dir) if bundle.assets_dir is not None else None + + +def measure_unloaded_reference(runtime): + """Measure before pipeline threads start; --zero asserts these frames are unloaded.""" + tactile = runtime.tactile + model = tactile.raw_sensor.calibration + if getattr(model, "schema_version", None) != 3: + raise ValueError("--zero requires a loaded schema 3 calibration") + info = tactile.info + axes = [(axis, key) for axis, key in + (("fn", info.FNORMAL_RAW), ("fx", info.FSHEARX_RAW), ("fy", info.FSHEARY_RAW)) + if axis in model.channels] + samples = {axis: [] for axis, _ in axes} + print("Measuring session zero from 30 unloaded frames...") + for _ in range(30): + ok, _ = runtime.grab_and_process(color=False) + if not ok: + raise ValueError("Input ended before 30 unloaded frames; session zero was not set") + tactile.compute_deformation() + values = tactile.read_info(*(key for _, key in axes)) + if not all(math.isfinite(value) for value in values): + raise ValueError("Non-finite unloaded reading; session zero was not set") + for (axis, _), value in zip(axes, values): + samples[axis].append(value) + runtime.set_calibration_reference( + {axis: statistics.median(values) for axis, values in samples.items()}, + reference_id="startup-unloaded-reference") + + def main() -> int: - parser = argparse.ArgumentParser(description="UV-unrolled finger tactile pipeline") + advanced = "--help-advanced" in sys.argv + parser = argparse.ArgumentParser(description="UV-unrolled finger tactile pipeline", + epilog="Use --help-advanced for SN selection and diagnostic overrides.") + parser.add_argument("--help-advanced", action="help", help=argparse.SUPPRESS) + def advanced_help(message): + return message if advanced else argparse.SUPPRESS parser.add_argument("--camera_id", type=int, default=0, help="camera index (default 0)") parser.add_argument("-v", "--video", help="video file path; overrides --camera_id") parser.add_argument("-c", "--config", default="finger", help="product config name or path (default: finger)") + calibration = parser.add_mutually_exclusive_group() + calibration.add_argument("--cal-path", "--cal_path", help="explicit calibration JSON (schema 1/2/3)") + calibration.add_argument("--calibration-dir", "--calibration_dir", + help=advanced_help("select //.json")) + parser.add_argument("--device-id", "--device_id", help=advanced_help("actual device SN to validate")) + zero = parser.add_mutually_exclusive_group() + zero.add_argument("--zero", action="store_true", + help="measure session zero; first 30 frames after warmup MUST be unloaded") + zero.add_argument("--zero-fn-raw", type=float, + help=advanced_help("measured unloaded FN raw for this session (schema 3 only)")) + parser.add_argument("--export-dir", help=advanced_help("override bundle geometry export directory")) + parser.add_argument("--map-cache", help=advanced_help("override UV map base path; SDK appends scale suffix")) parser.add_argument("--no-show", dest="show", action="store_false", help="do not open the display window") parser.add_argument("-verbose", action="store_true", help="SDK verbose logs") args = parser.parse_args() + if args.device_id and not (args.cal_path or args.calibration_dir): + parser.error("--device-id requires --cal-path or --calibration-dir") + if args.calibration_dir and not args.device_id: + parser.error("--calibration-dir requires --device-id") + if args.zero_fn_raw is not None and not math.isfinite(args.zero_fn_raw): + parser.error("--zero-fn-raw must be finite") source = args.video if args.video else args.camera_id # ------------------------------------------------------------------ create the object + options = {key: getattr(args, key) for key in + ("cal_path", "calibration_dir", "device_id", "export_dir", "map_cache") + if getattr(args, key) is not None} try: - runtime = FingerRuntime.open(config=args.config, video=source, verbose=args.verbose) - except RuntimeError as exc: + if "export_dir" not in options: + export_dir = configured_export_dir(args.config) + if export_dir is not None: + options["export_dir"] = export_dir + runtime = FingerRuntime.open(config=args.config, video=source, verbose=args.verbose, **options) + except (RuntimeError, ValueError, OSError) as exc: print(str(exc), file=sys.stderr) return 1 @@ -56,14 +135,22 @@ def main() -> int: f"source={g.source_width}x{g.source_height} " f"output={g.out_width}x{g.out_height}") - # Fill the flow/HHD caches once so the first published frame is not a warm-up outlier. - runtime.warmup(color=False) - if args.show: - print("Keys: 'q' quit, 'r' reset tracker.") + print("Keys: 'q' quit, 'r' reset tracker and clear session zero.") include = ("image", "flow") if args.show else ("image",) try: + # Warmup may reset the session; apply a supplied reference afterwards and before threads start. + runtime.warmup(color=False) + if getattr(runtime.tactile.raw_sensor.calibration, "schema_version", None) == 3 and not args.device_id: + print("Calibration file selected; hardware SN has not been independently verified.") + if args.zero: + measure_unloaded_reference(runtime) + if args.zero_fn_raw is not None: + model = runtime.tactile.raw_sensor.calibration + if getattr(model, "schema_version", None) != 3: + raise ValueError("--zero-fn-raw requires a loaded schema 3 calibration") + runtime.set_calibration_reference({"fn": args.zero_fn_raw}, reference_id="cli-unloaded-reference") with runtime.pipeline(contact=True, include=include) as pipe: while True: # ------------------------------------------------------ read one frame @@ -75,12 +162,13 @@ def main() -> int: # ------------------------------------------------------ use the results coverage = runtime.tactile.mask_stats.get("coverage") coverage_text = f"{coverage:.2f}" if isinstance(coverage, float) else "n/a" - fn_unit, fx_unit, fy_unit = runtime.tactile.force_units + fn_unit, fx_unit, fy_unit = snap.force_units or runtime.tactile.force_units + status = ",".join(f"{axis}:{value}" for axis, value in (snap.calibration_status or {}).items()) print( f"FPS={snap.fps:5.1f} " f"normal={snap.normal_force:8.4f} {fn_unit} " f"shear=({snap.shear_x:7.4f} {fx_unit}, {snap.shear_y:7.4f} {fy_unit}) " - f"mask={coverage_text} age={snap.age_ms:4.1f} ms", + f"mask={coverage_text} age={snap.age_ms:4.1f} ms calibration={status or 'legacy/raw'}", end="\r", ) @@ -89,13 +177,16 @@ def main() -> int: snap.image, snap.flow, threshold=0.4, grid_spacing=10, valid_mask=runtime.tactile.valid_mask)) - key = cv2.waitKey(1) & 0xFF + key = cv2.waitKey(1) & 0xFF if args.show else -1 if key == ord("q"): print("\nExiting.") break if key == ord("r"): pipe.reset() - print("\nTracker reset.") + print("\nTracker reset. Schema 3 session zero cleared; a new unloaded reference is required.") + except (RuntimeError, ValueError, OSError) as exc: + print(str(exc), file=sys.stderr) + return 1 except KeyboardInterrupt: print("\nInterrupted.") finally: @@ -106,4 +197,4 @@ def main() -> int: if __name__ == "__main__": - sys.exit(main()) \ No newline at end of file + sys.exit(main()) diff --git a/tests/test_finger_pipeline_example.py b/tests/test_finger_pipeline_example.py new file mode 100644 index 0000000..2ed63e3 --- /dev/null +++ b/tests/test_finger_pipeline_example.py @@ -0,0 +1,149 @@ +"""Exercise the example's SDK boundary without opening a camera or display.""" + +import importlib.util +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +@pytest.fixture +def example(monkeypatch): + path = Path(__file__).resolve().parents[1] / "examples" / "finger_pipeline.py" + monkeypatch.syspath_prepend(str(path.parent)) + spec = importlib.util.spec_from_file_location("finger_pipeline_example", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.mark.parametrize("zero_mode", ["raw", "measure", "none"]) +def test_schema3_reference_is_applied_after_warmup_before_pipeline(example, monkeypatch, capsys, zero_mode): + events = [] + options = {} + snapshot = SimpleNamespace(fps=20., normal_force=1., shear_x=2., shear_y=3., age_ms=1., + force_units=("N", "raw", "raw"), calibration_status={"fn": "ok"}, + image=None, flow=None) + + class Pipe: + def __enter__(self): + events.append("pipeline") + self.frames = iter([snapshot, None]) + return self + + def __exit__(self, *args): + pass + + def wait(self): + return next(self.frames) + + runtime = SimpleNamespace( + tactile=SimpleNamespace(raw_sensor=SimpleNamespace(calibration=SimpleNamespace(schema_version=3)), + mask_stats={}, force_units=("wrong",) * 3), + warmup=lambda **kwargs: events.append("warmup"), + set_calibration_reference=lambda raw, **kwargs: events.append(("reference", raw, kwargs)), + pipeline=lambda **kwargs: Pipe(), close=lambda: events.append("close")) + + def open_runtime(**kwargs): + options.update(kwargs) + return runtime + + monkeypatch.setattr(example.FingerRuntime, "open", open_runtime) + monkeypatch.setattr(example, "measure_unloaded_reference", + lambda runtime: events.append("measure")) + monkeypatch.setattr(example.cv2, "waitKey", lambda *args: pytest.fail("headless must not call GUI")) + zero_args = {"raw": ["--zero-fn-raw", "1765.1"], "measure": ["--zero"], "none": []}[zero_mode] + monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--video", "test.mp4", "--no-show", + "--cal-path", "model.json", "--device-id", "dd02-gl", + "--export-dir", "assets", "--map-cache", "map", *zero_args]) + assert example.main() == 0 + assert options["cal_path"] == "model.json" and options["device_id"] == "dd02-gl" + assert options["export_dir"] == "assets" and options["map_cache"] == "map" + reference_events = {"raw": [("reference", {"fn": 1765.1}, {"reference_id": "cli-unloaded-reference"})], + "measure": ["measure"], "none": []}[zero_mode] + assert events == ["warmup", *reference_events, "pipeline", "close"] + output = capsys.readouterr().out + assert "calibration=fn:ok" in output and "wrong" not in output + + +def test_selected_calibration_error_is_reported_without_fallback(example, monkeypatch, capsys): + def fail(**kwargs): + raise ValueError("calibration identity mismatch") + + monkeypatch.setattr(example.FingerRuntime, "open", fail) + monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--no-show", "--cal_path", "bad.json"]) + assert example.main() == 1 + assert "calibration identity mismatch" in capsys.readouterr().err + + +@pytest.mark.parametrize("arguments", [["--zero-fn-raw", "nan"], ["--calibration-dir", "calibrations"], + ["--device-id", "SN1"], ["--cal-path", "a", "--calibration-dir", "b"]]) +def test_invalid_selection_fails_before_capture(example, monkeypatch, arguments): + monkeypatch.setattr(example.FingerRuntime, "open", lambda **kwargs: pytest.fail("must not open capture")) + monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", *arguments]) + with pytest.raises(SystemExit) as exc: + example.main() + assert exc.value.code == 2 + + +def test_bundle_geometry_follows_explicit_config(example, tmp_path, monkeypatch): + bundle = tmp_path / "custom" + bundle.mkdir() + (bundle / "sensor.json").write_text('{}') + (bundle / "manifest.json").write_text('{"assets_dir": "geometry"}') + (bundle / "geometry").mkdir() + monkeypatch.chdir(tmp_path) + assert Path(example.configured_export_dir("custom/sensor.json")) == bundle / "geometry" + + +def test_missing_declared_geometry_does_not_use_default_bundle(example, tmp_path): + (tmp_path / "sensor.json").write_text('{}') + (tmp_path / "manifest.json").write_text('{"assets_dir": "missing"}') + with pytest.raises(ValueError, match="assets_dir"): + example.configured_export_dir(str(tmp_path / "sensor.json")) + + +@pytest.mark.parametrize("advanced", [False, True]) +def test_help_separates_debug_overrides(example, monkeypatch, capsys, advanced): + monkeypatch.setattr(sys, "argv", ["finger_pipeline.py", "--help-advanced" if advanced else "--help"]) + with pytest.raises(SystemExit) as exc: + example.main() + assert exc.value.code == 0 + output = capsys.readouterr().out + assert "--cal-path" in output and "--zero " in output + assert ("--zero-fn-raw" in output) == advanced + assert ("--export-dir" in output) == advanced + assert ("--device-id" in output) == advanced + + +@pytest.mark.parametrize("failure", [None, "short", "nan", "legacy"]) +def test_unloaded_reference_is_measured_atomically(example, failure): + events = [] + frames = iter(range(30)) + count = 0 + + def grab(**kwargs): + nonlocal count + count += 1 + return not (failure == "short" and count == 30), None + + def read(*keys): + value = next(frames) + return (float("nan") if failure == "nan" and value == 29 else value,) + + runtime = SimpleNamespace( + tactile=SimpleNamespace( + raw_sensor=SimpleNamespace(calibration=SimpleNamespace( + schema_version=2 if failure == "legacy" else 3, channels={"fn": {}})), + info=SimpleNamespace(FNORMAL_RAW=0, FSHEARX_RAW=1, FSHEARY_RAW=2), + compute_deformation=lambda: None, read_info=read), + grab_and_process=grab, + set_calibration_reference=lambda raw, **kwargs: events.append((raw, kwargs))) + if failure: + with pytest.raises(ValueError): + example.measure_unloaded_reference(runtime) + assert events == [] + else: + example.measure_unloaded_reference(runtime) + assert events == [({"fn": 14.5}, {"reference_id": "startup-unloaded-reference"})]