ykcheungab
143 lines
6.6 KiB
Python
143 lines
6.6 KiB
Python
"""ORY-BOX / ORY-MINI: read a sensor and get tactile information, in one loop.
|
|
|
|
The whole recommended flow:
|
|
|
|
sensor -> StreamPipeline -> FrameSnapshot per frame
|
|
|
|
One object reads the camera, pre-processes (stitch, grey, resize), computes the deformation field
|
|
and the forces, and hands back one immutable snapshot per frame. There is no stage sequence to
|
|
remember and no tuple to unpack by position.
|
|
|
|
What you do NOT need to touch: the compute backend, the capture queue, the stitch mode, the
|
|
low-pass filter, or the order in which the stages run. Those are the SDK's concern and
|
|
``StreamPipeline`` sequences them on three concurrent threads. If you later need that control, the
|
|
manual form is preserved in ``examples/archive/basic_pipeline_manual.py``.
|
|
|
|
Run:
|
|
|
|
python examples/basic_pipeline.py # first camera, box config
|
|
python examples/basic_pipeline.py -v 0 -c mini
|
|
python examples/basic_pipeline.py -v recording.mp4 -c box
|
|
python examples/basic_pipeline.py --no-show # headless: print only
|
|
python examples/basic_pipeline.py -v 0 -c box --calibration-dir data/calibrations --device-id BOX-001
|
|
|
|
Keys: 'q' quit, 'r' reset the tracker.
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
# Set before OpenCV is imported: stops the capture backend applying its own transform on top of
|
|
# the SDK's preprocessing. This is a capture detail, not a tuning knob you need to own - it is
|
|
# here so the file can be copied as a whole.
|
|
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
|
|
|
|
import cv2
|
|
|
|
import orisys
|
|
from _display import depth_to_color, draw_contact, draw_flow_arrows
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Read a sensor and print its tactile information.")
|
|
parser.add_argument("-v", "--video", default="0",
|
|
help="camera index or video file path (default: 0)")
|
|
parser.add_argument("-c", "--config", default="box",
|
|
help="product config: box, mini, finger, or a config path")
|
|
calibration = parser.add_mutually_exclusive_group()
|
|
calibration.add_argument("--cal", default=None,
|
|
help="force calibration file; omit to use the sensor bundle's own")
|
|
calibration.add_argument("--calibration-dir", default=None,
|
|
help="root containing <product>/<device-id>.json")
|
|
parser.add_argument("--device-id", default=None,
|
|
help="physical sensor ID; required with --calibration-dir")
|
|
parser.add_argument("--no-show", dest="show", action="store_false",
|
|
help="do not open the display windows")
|
|
parser.add_argument("-verbose", action="store_true", help="SDK verbose logs")
|
|
args = parser.parse_args()
|
|
if args.calibration_dir is not None and args.device_id is None:
|
|
parser.error("--calibration-dir requires --device-id")
|
|
if args.device_id is not None and args.calibration_dir is None and args.cal is None:
|
|
parser.error("--device-id requires --calibration-dir or --cal")
|
|
|
|
# ------------------------------------------------------------------ create the object
|
|
# That is the whole setup. The source may be a camera index or a video path.
|
|
source = int(args.video) if args.video.isdigit() else args.video
|
|
sensor = orisys.Sensor(
|
|
source, config_name=args.config, cal_path=args.cal,
|
|
calibration_dir=args.calibration_dir, device_id=args.device_id,
|
|
verbose=args.verbose,
|
|
)
|
|
fn_unit, fx_unit, fy_unit = sensor.force_units
|
|
selected = sensor.calibration.source_path if sensor.calibration else None
|
|
print(f"Force calibration: {selected or 'none'}")
|
|
print(f"Force units: FN={fn_unit}, FX={fx_unit}, FY={fy_unit}")
|
|
if sensor.device_calibration_missing:
|
|
print(
|
|
f"Warning: no calibration for {args.device_id!r} under "
|
|
f"{args.calibration_dir}; force is raw."
|
|
)
|
|
|
|
# contact=True runs contact analysis. include=... selects which arrays each snapshot carries;
|
|
# arrays that are not included are None, so you pay only for what you use.
|
|
include = ("image", "flow", "depth", "centroid", "contour") if args.show else ("image",)
|
|
|
|
if args.show:
|
|
print("Keys: 'q' quit, 'r' reset tracker.")
|
|
|
|
try:
|
|
with orisys.StreamPipeline(sensor, sensor.get_img, contact=True,
|
|
include=include) as pipe:
|
|
while True:
|
|
# ------------------------------------------------------ read one frame
|
|
# Blocks until a frame newer than the last one is ready. Everything in the
|
|
# snapshot belongs to the same frame.
|
|
snap = pipe.wait()
|
|
if snap is None:
|
|
print("\nInput ended.")
|
|
break
|
|
|
|
# ------------------------------------------------------ use the results
|
|
# The *_raw pair is always the unfiltered integrator output.
|
|
touching = "yes" if snap.centroid is not None else "no"
|
|
fn_unit, fx_unit, fy_unit = sensor.force_units
|
|
print(
|
|
f"FPS={snap.fps:5.1f} "
|
|
f"normal={snap.normal_force:8.4f} {fn_unit} (raw {snap.normal_force_raw:10.2f}) "
|
|
f"shear=({snap.shear_x:7.4f} {fx_unit}, {snap.shear_y:7.4f} {fy_unit}) "
|
|
f"contact={touching} age={snap.age_ms:4.1f} ms",
|
|
end="\r",
|
|
)
|
|
|
|
if args.show and snap.image is not None:
|
|
if snap.flow is not None:
|
|
cv2.imshow("deformation", draw_flow_arrows(
|
|
snap.image, snap.flow, threshold=2, grid_spacing=20))
|
|
if snap.contour is not None and snap.centroid is not None:
|
|
cv2.imshow("contact", draw_contact(
|
|
snap.image, snap.contour, snap.centroid))
|
|
colour = depth_to_color(snap.depth)
|
|
if colour is not None:
|
|
cv2.imshow("depth", colour)
|
|
|
|
key = cv2.waitKey(1) & 0xFF if args.show else -1
|
|
if key == ord("q"):
|
|
print("\nExiting.")
|
|
break
|
|
if key == ord("r"):
|
|
# Queued onto the flow stage, so this is safe from the consumer loop.
|
|
pipe.reset()
|
|
print("\nTracker reset.")
|
|
except KeyboardInterrupt:
|
|
print("\nInterrupted.")
|
|
finally:
|
|
sensor.disconnect()
|
|
if args.show:
|
|
cv2.destroyAllWindows()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|