ykcheungab
109 lines
4.2 KiB
Python
109 lines
4.2 KiB
Python
"""ORY-FINGER: UV-unrolled finger tactile pipeline.
|
|
|
|
Same shape as ``basic_pipeline.py`` - one object, one call per frame - with the finger-specific
|
|
capture path underneath:
|
|
|
|
FingerRuntime -> runtime.pipeline() -> FrameSnapshot per frame
|
|
|
|
``FingerRuntime`` owns the camera, the Blender export (mesh, intrinsics, world-to-camera), the UV
|
|
unroll map and the valid-region mask. ``runtime.pipeline()`` runs those stages concurrently and
|
|
returns one immutable snapshot per frame, so there is nothing to sequence and no tuple to unpack.
|
|
|
|
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
|
|
|
|
Keys: 'q' quit, 'r' reset the tracker.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
|
|
|
|
import cv2
|
|
|
|
from _display import draw_flow_arrows
|
|
from orisys.finger import FingerRuntime
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="UV-unrolled finger tactile pipeline")
|
|
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)")
|
|
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()
|
|
|
|
source = args.video if args.video else args.camera_id
|
|
|
|
# ------------------------------------------------------------------ create the object
|
|
try:
|
|
runtime = FingerRuntime.open(config=args.config, video=source, verbose=args.verbose)
|
|
except RuntimeError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
|
|
if args.verbose:
|
|
g = runtime.geometry
|
|
print(f"UV geometry: native={g.native_width}x{g.native_height} "
|
|
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.")
|
|
|
|
include = ("image", "flow") if args.show else ("image",)
|
|
try:
|
|
with runtime.pipeline(contact=True, include=include) as pipe:
|
|
while True:
|
|
# ------------------------------------------------------ read one frame
|
|
snap = pipe.wait()
|
|
if snap is None:
|
|
print("\nInput ended.")
|
|
break
|
|
|
|
# ------------------------------------------------------ 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
|
|
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",
|
|
end="\r",
|
|
)
|
|
|
|
if args.show and snap.image is not None and snap.flow is not None:
|
|
cv2.imshow("tactile (UV unrolled)", draw_flow_arrows(
|
|
snap.image, snap.flow, threshold=0.4, grid_spacing=10,
|
|
valid_mask=runtime.tactile.valid_mask))
|
|
|
|
key = cv2.waitKey(1) & 0xFF
|
|
if key == ord("q"):
|
|
print("\nExiting.")
|
|
break
|
|
if key == ord("r"):
|
|
pipe.reset()
|
|
print("\nTracker reset.")
|
|
except KeyboardInterrupt:
|
|
print("\nInterrupted.")
|
|
finally:
|
|
runtime.close()
|
|
if args.show:
|
|
cv2.destroyAllWindows()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |