examples: make StreamPipeline the recommended entry, archive the manual loop
The two canonical examples taught the advanced path: they sequenced get_img() -> compute_deformation() -> compute_contact() -> read_info() by hand and unpacked a 13-element tuple positionally, which is what led developers to reach for the compute backend, the capture queue and the stitch mode. The managed alternative already existed and was documented nowhere. Library (add-only, nothing removed): - FrameSnapshot.image_size_raw carries the capture resolution, so a caller can compare it with the processed frame instead of reaching into Sensor. - StreamPipeline.reset() resets the tracker and force filters *safely*. It is queued and applied by the flow stage before its next frame, never inline: Sensor.reset rebases initial_img on the current Sensor.img, which the flow stage is reading concurrently, so a caller-thread reset would capture a half-written reference frame. Idempotent. - TactileProcessor.pipeline(produce) and FingerRuntime.pipeline() build a pipeline without the caller naming a backend concept; the latter is `with runtime.pipeline(contact=True) as pipe`. Examples: - examples/basic_pipeline.py and examples/finger_pipeline.py rewritten onto StreamPipeline + FrameSnapshot: one object, one call per frame, no tuple, and no compute backend, queue or stitch setting in sight. - examples/archive/basic_pipeline_manual.py and _finger_pipeline_manual.py keep the previous versions byte-for-byte apart from a supersede header, so code copied from them keeps working. The finger header also records the archived --show defect (store_true with default=True, so it could never be disabled). - examples/_display.py holds the shared display helpers, so an example no longer hand-rolls depth normalisation and colormap application. - examples/README.md leads with the recommended pair and documents the manual form as still-supported, and states that examples/gui/ is moving to orisys-studio. Tests: three added to tests/test_pipeline.py, covering the reset queue (it must not run on the caller's thread and must land before the first compute_flow), image_size_raw, and the None case when a producer reports no size. Verified the reset test fails when reset() is made inline, so it discriminates. Verified end-to-end against real media: basic_pipeline on data/sample/s1_480p.mp4 (box) reaches ~113 FPS and exits 0; finger_pipeline on data/dd02-0703/1.mp4 (finger) reaches ~140 FPS with mask coverage 0.41 and exits 0. Both archived manual versions still run. tests/test_pipeline.py: 20 passed. Full suite: 342 passed, 3 failed - all three failures reproduce on the pristine tree (import precedence, longrun timing, and a stale ruff version literal in docs/code-review.md) and are unrelated to this change.
This commit is contained in:
ykcheungab
+101
-118
@@ -1,137 +1,120 @@
|
||||
"""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
|
||||
|
||||
Keys: 'q' quit, 'r' reset the tracker.
|
||||
"""
|
||||
ORY-BOX：单传感器基础流程
|
||||
|
||||
本示例演示 Orisys SDK 的标准处理流程：
|
||||
1. 初始化传感器（摄像头或视频文件）
|
||||
2. 获取图像并计算形变
|
||||
3. 按需计算接触区域
|
||||
4. 读取并可视化结果
|
||||
|
||||
"""
|
||||
import os
|
||||
|
||||
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
|
||||
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 numpy as np
|
||||
|
||||
import orisys
|
||||
from _display import depth_to_color, draw_contact, draw_flow_arrows
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--video","-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径")
|
||||
parser.add_argument("--config","-c", type=str, default="box", help="配置文件名称或路径")
|
||||
parser.add_argument("--verbose","-verbose", type=bool, default=True, help="是否输出详细日志")
|
||||
parser.add_argument("--cal_path","-cal", type=str, default=None, help="标定文件路径")
|
||||
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")
|
||||
parser.add_argument("--cal", default=None,
|
||||
help="force calibration file; omit to use the sensor bundle's own")
|
||||
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()
|
||||
|
||||
# 步骤 1：创建传感器对象（必需）
|
||||
# config_name 可以是内置配置名称，也可以是自定义配置文件路径
|
||||
# cal_path 为标定文件路径；每台设备建议使用独立的标定文件
|
||||
# ------------------------------------------------------------------ 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,
|
||||
verbose=args.verbose)
|
||||
|
||||
# 输入既可以是摄像头编号，也可以是视频文件路径
|
||||
if args.video.isdigit():
|
||||
sensor = orisys.Sensor(int(args.video), config_name=args.config, verbose=args.verbose, cal_path=args.cal_path)
|
||||
else:
|
||||
sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose, cal_path=args.cal_path)
|
||||
# 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",)
|
||||
|
||||
print("\n按键说明：'q' 退出程序，'r' 重置追踪器。")
|
||||
if args.show:
|
||||
print("Keys: 'q' quit, 'r' reset tracker.")
|
||||
|
||||
while True:
|
||||
# 步骤 2：数据处理
|
||||
# 步骤 2.1：获取并拼接图像（必需）
|
||||
sensor.get_img()
|
||||
# 步骤 2.2：计算形变场（必需）；返回值表示当前帧是否检测到接触/运动
|
||||
is_contact_now = sensor.compute_deformation()
|
||||
# 步骤 2.3：仅在当前帧检测到接触/运动时计算接触区域
|
||||
# if is_contact_now:
|
||||
sensor.compute_contact()
|
||||
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
|
||||
|
||||
|
||||
# 步骤 2.4：读取结果（FNORMAL/FSHEAR* 在有 sensors/box/force_calibration.json 时为标定后牛顿力；
|
||||
# *_RAW 始终为原始积分量）
|
||||
(
|
||||
fps,
|
||||
flow,
|
||||
vnormal,
|
||||
img,
|
||||
contour,
|
||||
centroid,
|
||||
depth_map,
|
||||
fnormal,
|
||||
fshearx,
|
||||
fsheary,
|
||||
fnormal_raw,
|
||||
fshearx_raw,
|
||||
fsheary_raw,
|
||||
) = sensor.read_info(
|
||||
sensor.info.FPS,
|
||||
sensor.info.VRAW,
|
||||
sensor.info.VNORMAL,
|
||||
sensor.info.IMG,
|
||||
sensor.info.CONTOUR,
|
||||
sensor.info.CENTROID,
|
||||
sensor.info.DEPTH,
|
||||
sensor.info.FNORMAL,
|
||||
sensor.info.FSHEARX,
|
||||
sensor.info.FSHEARY,
|
||||
sensor.info.FNORMAL_RAW,
|
||||
sensor.info.FSHEARX_RAW,
|
||||
sensor.info.FSHEARY_RAW,
|
||||
)
|
||||
|
||||
# 步骤 3：可视化显示
|
||||
# 步骤 3.1：显示形变矢量场
|
||||
arrows = orisys.util.draw_arrows(
|
||||
img,
|
||||
flow,
|
||||
threshold=2,
|
||||
grid_spacing=20,
|
||||
arrow_scale=1.0
|
||||
)
|
||||
cv2.imshow("arrow", arrows)
|
||||
|
||||
# 步骤 3.2：显示接触区域和质心
|
||||
image_with_foe = orisys.util.draw_contact(img, contour, centroid)
|
||||
cv2.imshow("contact", image_with_foe)
|
||||
# ------------------------------------------------------ use the results
|
||||
# Scalars are calibrated Newtons when the bundle ships a calibration; the *_raw
|
||||
# pair is always the unfiltered integrator output.
|
||||
touching = "yes" if snap.centroid is not None else "no"
|
||||
print(
|
||||
f"FPS={snap.fps:5.1f} "
|
||||
f"normal={snap.normal_force:8.4f} N (raw {snap.normal_force_raw:10.2f}) "
|
||||
f"shear=({snap.shear_x:7.4f}, {snap.shear_y:7.4f}) "
|
||||
f"contact={touching} age={snap.age_ms:4.1f} ms",
|
||||
end="\r",
|
||||
)
|
||||
|
||||
# 步骤 3.3：显示深度图
|
||||
# 将深度值归一化到 0-255 范围
|
||||
div_abs = depth_map
|
||||
if div_abs.max() > div_abs.min():
|
||||
div_normalized = ((div_abs - div_abs.min()) / (div_abs.max() - div_abs.min()) * 255).astype(np.uint8)
|
||||
else:
|
||||
div_normalized = np.zeros_like(div_abs, dtype=np.uint8)
|
||||
|
||||
# 应用伪彩色映射
|
||||
cv2.imshow("depth", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET))
|
||||
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)
|
||||
|
||||
# 步骤 3.4：打印关键结果（标定力 + 原始力）
|
||||
print(
|
||||
f"FPS={fps:.2f}, "
|
||||
f"法向力={fnormal:.4f} N (raw={fnormal_raw:.4f}), "
|
||||
f"切向力X={fshearx:.4f} N (raw={fshearx_raw:.4f}), "
|
||||
f"切向力Y={fsheary:.4f} N (raw={fsheary_raw:.4f}), "
|
||||
f"深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}"
|
||||
)
|
||||
|
||||
# 步骤 3.5：键盘控制
|
||||
key = cv2.waitKey(1) & 0xFF
|
||||
if key == ord("q"):
|
||||
print("\n正在退出示例程序...")
|
||||
break
|
||||
if key == ord("r"):
|
||||
# 当光流追踪出现漂移时，可手动重置追踪器
|
||||
sensor.reset()
|
||||
print("追踪器已重置。")
|
||||
|
||||
# =========================================================================
|
||||
# 4. 释放资源
|
||||
# =========================================================================
|
||||
sensor.disconnect() # 释放视频源及相关资源
|
||||
key = cv2.waitKey(1) & 0xFF
|
||||
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__':
|
||||
main()
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user