diff --git a/.gitignore b/.gitignore index b8ce174..035bb1a 100644 --- a/.gitignore +++ b/.gitignore @@ -9,7 +9,7 @@ __pycache__/ /dist_obfuscated/ /dist_pyd/ /dist_single_pyd/ -/dist_nuitka/ +/nuitka_wheel/ /build_nuitka/ /release/ /SDK/ diff --git a/README.md b/README.md index a9ac9ce..5a76526 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,7 @@ cfg = orisys.export_config("ddjx01", "./config/exported_config.json") - Example entry points: `examples/README.md` - User documentation: `docs/get_started.md` -- Packaging helper: `scripts/build_nuitka.py` +- Packaging helper: `build/build_nuitka.py` - Specialized utilities: `tools/` ## Notes for Collaborators diff --git a/assets/finger/30fps/camera_intrinsics.npz b/assets/finger/30fps/camera_intrinsics.npz new file mode 100644 index 0000000..59ac687 Binary files /dev/null and b/assets/finger/30fps/camera_intrinsics.npz differ diff --git a/assets/finger/30fps/flatten_remap.npz b/assets/finger/30fps/flatten_remap.npz new file mode 100644 index 0000000..02ca287 Binary files /dev/null and b/assets/finger/30fps/flatten_remap.npz differ diff --git a/assets/finger/30fps/mesh.npz b/assets/finger/30fps/mesh.npz new file mode 100644 index 0000000..ad81542 Binary files /dev/null and b/assets/finger/30fps/mesh.npz differ diff --git a/assets/finger/30fps/unroll_map_uv.npz b/assets/finger/30fps/unroll_map_uv.npz new file mode 100644 index 0000000..8415977 Binary files /dev/null and b/assets/finger/30fps/unroll_map_uv.npz differ diff --git a/assets/finger/30fps/unroll_map_uv_s0.4.npz b/assets/finger/30fps/unroll_map_uv_s0.4.npz new file mode 100644 index 0000000..365fcb5 Binary files /dev/null and b/assets/finger/30fps/unroll_map_uv_s0.4.npz differ diff --git a/assets/finger/30fps/world_to_cam.npy b/assets/finger/30fps/world_to_cam.npy new file mode 100644 index 0000000..2d5e411 Binary files /dev/null and b/assets/finger/30fps/world_to_cam.npy differ diff --git a/config/dd02-ov.json b/config/dd02-ov.json index 35e39c4..0fb0a18 100644 --- a/config/dd02-ov.json +++ b/config/dd02-ov.json @@ -5,7 +5,7 @@ "camera": { "resolution_raw": [1280, 720], "fps": 30, - "resolution_stitch": [280, 300], + "resolution_stitch": [672, 728], "source_scale": 0.4, "old":[672, 728], "work": [504, 546,1], @@ -27,7 +27,7 @@ "contact": { "min_contour_area": 3000, "flow_threshold": 1, - "div_threshold": 0.05, + "div_threshold": 0.03, "expansion_threshold": 0.01, "kernel_size": [5, 5], "denoise_sigma": 12 diff --git a/examples/README.md b/examples/README.md index b05b54e..0db07e1 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,50 +1,41 @@ -# Examples - -This folder contains user-facing example scripts for the Orisys SDK. - -## Files - -- `basic_pipeline.py`: single-sensor basic processing and visualization flow -- `two_sensor_pipeline.py`: dual-sensor multiprocessing example -- `plot_data.py`: real-time plotting example with PyQtGraph - -## Typical usage - -Run examples from the repository root so relative paths such as `./config/ddjx01.json` -resolve correctly. - -### Basic pipeline - -```bash -python examples/basic_pipeline.py -v 0 -c ./config/ddjx01.json -``` - -### Two-sensor pipeline - -```bash -python examples/two_sensor_pipeline.py -``` - -### Real-time plotting - -```bash -python examples/plot_data.py -v 0 -c ./config/ddjx01.json -``` - -## Requirements - -- Python environment with the SDK dependencies installed -- a supported camera or test video -- valid JSON config in `./config/` - -Some examples also require: - -- `pyqtgraph` -- `PyQt5` -- multiple cameras for multi-sensor demos - -## Notes - -- Most examples assume a tactile sensor camera is connected. -- Press `q` in the display window to quit where applicable. -- If your camera index is not `0`, change `-v` accordingly. +# Examples + +User-facing demo scripts for the Orisys SDK. Run from the repository root. + +## Layout + +- `basic_pipeline.py` — compound-eye sensor, OpenCV visualization +- `plot_data.py` — compound-eye sensor, force charts (PyQtGraph) +- `two_sensor_pipeline.py` — dual-sensor multiprocessing demo +- `finger_pipeline.py` — finger UV minimal pipeline +- `finger_pipeline_3dview.py` — finger UV + Open3D mesh viewer +- `finger_plot_data.py` — finger UV + force charts +- `viz/` — shared visualization helpers used by plot and finger examples + - `mesh_viewer.py` — re-exports `UVMeshViewer` / `build_uv_surface_mesh` from `orisys.finger` + - `finger_panels.py` — 2D OpenCV panels for finger demos + +## Typical usage + +```bash +python examples/basic_pipeline.py -v 0 -c ./config/ddjx01.json +python examples/plot_data.py -v 0 -c ./config/ddjx01.json +python examples/finger_plot_data.py -v 0 -c ./config/dd02-ov.json +``` + +## Requirements + +- SDK installed (`pip install -e .`) +- camera or test video +- JSON config under `./config/` + +Plot examples additionally require: + +```bash +pip install pyqt5 pyqtgraph +``` + +## Notes + +- Press `q` to quit OpenCV windows where applicable. +- Plot examples: `s` starts recording, `e` stops recording. + \ No newline at end of file diff --git a/examples/finger_pipeline.py b/examples/finger_pipeline.py index c7ec68f..8f19656 100644 --- a/examples/finger_pipeline.py +++ b/examples/finger_pipeline.py @@ -3,7 +3,7 @@ Minimum UV-unroll finger tactile pipeline. Offline: load Blender export -> build_or_load_domain -> unroll_map_uv.npz -Runtime: camera frame -> UVUnrollFrameProcessor -> Sensor -> forces/contact +Runtime: camera frame -> UVUnrollFrameProcessor -> TactileProcessor -> forces/contact For the full demo (UV debug panels, dashboard, benchmarks) see: tools/3d_patch_generation_blender/unroll_realtime.py @@ -17,190 +17,67 @@ os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") import cv2 import orisys -from orisys.configs import load_config -from orisys.frame_processors import UVUnrollFrameProcessor -from orisys.unroll import ( - build_or_load_domain, - compute_uv_distortion_metrics, - load_calib_json, - load_exported_data, - recommend_edge_expand, - resolve_uv_runtime_sizes, - scale_camera_geometry, - scaled_map_cache_path, -) - - -def open_source(video_path, camera_id): - return cv2.VideoCapture(video_path) if video_path else cv2.VideoCapture(camera_id) - - -def resolve_export_paths(export_dir): - """Return Blender export paths expected inside one folder.""" - export_dir = os.path.abspath(export_dir) - paths = { - "mesh": os.path.join(export_dir, "mesh.npz"), - "intrinsics": os.path.join(export_dir, "camera_intrinsics.npz"), - "pose": os.path.join(export_dir, "world_to_cam.npy"), - } - missing = [name for name, path in paths.items() if not os.path.isfile(path)] - if missing: - raise FileNotFoundError( - f"Missing export file(s) in {export_dir}: {', '.join(missing)}" - ) - return paths["mesh"], paths["intrinsics"], paths["pose"] +from orisys.finger import FingerRuntime def main(): parser = argparse.ArgumentParser(description="Minimum UV-unroll finger tactile pipeline") - parser.add_argument( - "--export-dir", - required=True, - help="Blender export folder containing mesh.npz, camera_intrinsics.npz, world_to_cam.npy", - ) - parser.add_argument("--calib", default=None, help="OpenCV calibration JSON for the live camera") - parser.add_argument("--map_cache", default="unroll_map_uv.npz", help="Cached UV domain .npz") parser.add_argument("--camera_id", type=int, default=0) parser.add_argument("--video", help="Video file path (overrides camera)") parser.add_argument("--config", "-c", default="./config/dd02-ov.json", help="Orisys SDK config") - parser.add_argument( - "--source_scale", - type=float, - default=None, - help="Resize live source before UV remap (default: camera.source_scale in config)", - ) - parser.add_argument("--no_flip_source_vertical", action="store_true", help="Skip vertical flip before UV remap") - parser.add_argument("--show", action="store_true", help="Show unrolled image window") + parser.add_argument("--show", action="store_true", default=True, help="Show unrolled image window") parser.add_argument("--verbose", action="store_true", help="SDK verbose logs") args = parser.parse_args() - sdk_config = load_config(args.config) - camera_config = sdk_config.get("camera", {}) - out_width, out_height, source_scale = resolve_uv_runtime_sizes( - camera_config, - source_scale=args.source_scale, - ) - map_cache = scaled_map_cache_path(args.map_cache, source_scale) + try: + video = args.camera_id + if args.video: + video = args.video + runtime = FingerRuntime.open( + config=args.config, + video=video, + verbose=args.verbose, + ) + except RuntimeError as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) - mesh_path, intrinsics_path, pose_path = resolve_export_paths(args.export_dir) - - data = load_exported_data(mesh_path, intrinsics_path, pose_path) - distortion = compute_uv_distortion_metrics(data) - edge_expand = recommend_edge_expand(distortion) - - if args.calib: - source_K, dist_coeffs, img_size = load_calib_json(args.calib) - native_width, native_height = img_size - else: - source_K = data["K"] - dist_coeffs = None - native_width = data["width"] - native_height = data["height"] - - source_K, source_width, source_height = scale_camera_geometry( - source_K, - native_width, - native_height, - source_scale, - ) if args.verbose: + geometry = runtime.geometry print( - f"UV geometry: native={native_width}x{native_height}, " - f"source={source_width}x{source_height} (scale={source_scale:g}), " - f"output={out_width}x{out_height}, map_cache={map_cache}" + f"UV geometry: native={geometry.native_width}x{geometry.native_height}, " + f"source={geometry.source_width}x{geometry.source_height} " + f"output={geometry.out_width}x{geometry.out_height} " ) - domain = build_or_load_domain( - mesh_path, - intrinsics_path, - pose_path, - map_cache, - source_K, - source_width, - source_height, - out_width, - out_height, - edge_expand=edge_expand, - include_surface=True, - verbose=args.verbose, - source_scale=source_scale, - ) - domain.remap.ensure_fast_maps() - - cap = open_source(args.video, args.camera_id) - if not cap.isOpened(): - print("Cannot open camera/video source", file=sys.stderr) - sys.exit(1) - cap.set(cv2.CAP_PROP_FRAME_WIDTH, native_width) - cap.set(cv2.CAP_PROP_FRAME_HEIGHT, native_height) - - sensor = orisys.Sensor("external", config_name=args.config, verbose=args.verbose, mirror=False) - processor = UVUnrollFrameProcessor( - domain.remap, - camera_matrix=source_K if dist_coeffs is not None else None, - dist_coeffs=dist_coeffs, - source_width=source_width, - source_height=source_height, - flip_source_vertical=not args.no_flip_source_vertical, - ) - - ok, frame = cap.read() - if ok: - if args.show: - prepared = processor.prepare_source(frame) - sensor.get_img( - frame=processor.flatten_prepared(prepared), - already_processed=True, - valid_mask=processor.valid_mask, - ) - else: - prepared_gray = processor.prepare_source_gray(frame) - sensor.get_img( - frame=processor.flatten_prepared_gray(prepared_gray), - already_processed=True, - already_gray=True, - valid_mask=processor.valid_mask, - ) - sensor.warmup_deformation() + runtime.warmup(color=False) + tactile = runtime.tactile print("Press 'q' to quit, 'r' to reset tracker.") try: while True: - ok, frame = cap.read() + ok, unrolled = runtime.grab_and_process(color=False) if not ok: break + tactile.compute_deformation(check_motion=True, threshold=0) + tactile.compute_contact() - if args.show: - prepared = processor.prepare_source(frame) - unrolled = processor.flatten_prepared(prepared) - sensor.get_img( - frame=unrolled, - already_processed=True, - valid_mask=processor.valid_mask, - ) - else: - prepared_gray = processor.prepare_source_gray(frame) - unrolled = processor.flatten_prepared_gray(prepared_gray) - sensor.get_img( - frame=unrolled, - already_processed=True, - already_gray=True, - valid_mask=processor.valid_mask, - ) - sensor.compute_deformation(check_motion=True, threshold=0) - sensor.compute_contact() - - fps, flow,fnormal, fshearx, fsheary = sensor.read_info( - sensor.info.FPS, - sensor.info.VRAW, - sensor.info.FNORMAL, - sensor.info.FSHEARX, - sensor.info.FSHEARY, + fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = tactile.read_info( + tactile.info.FPS, + tactile.info.VRAW, + tactile.info.VNORMAL, + tactile.info.IMG, + tactile.info.CONTOUR, + tactile.info.CENTROID, + tactile.info.DEPTH, + tactile.info.FNORMAL, + tactile.info.FSHEARX, + tactile.info.FSHEARY ) print( f"FPS={fps:.2f} normal={fnormal:.2f} shear=({fshearx:.2f}, {fsheary:.2f}) " - f"mask={sensor.mask_stats['coverage']:.2f}", + f"mask={tactile.mask_stats['coverage']:.2f}", end="\r", ) @@ -214,6 +91,9 @@ def main(): grid_spacing=10, arrow_scale=1.0, ) + # 步骤 3.2：显示接触区域和质心 + image_with_foe = orisys.util.draw_contact(img, contour, centroid) + cv2.imshow("接触区域与质心", image_with_foe) cv2.imshow("UV Unrolled", arrows) key = cv2.waitKey(1) & 0xFF @@ -221,11 +101,10 @@ def main(): print("\nExiting.") break if key == ord("r"): - sensor.reset() + tactile.reset() print("\nTracker reset.") finally: - cap.release() - sensor.disconnect() + runtime.close() cv2.destroyAllWindows() diff --git a/examples/finger_pipeline_3dview.py b/examples/finger_pipeline_3dview.py new file mode 100644 index 0000000..bdf1183 --- /dev/null +++ b/examples/finger_pipeline_3dview.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +""" +UV-unroll finger tactile pipeline with 3D mesh visualization. + +Offline: load Blender export -> build_or_load_domain -> unroll_map_uv.npz +Runtime: camera frame -> UVUnrollFrameProcessor -> TactileProcessor -> forces/contact + UV-space flow/HHD -> grey-white 3D finger mesh (Open3D) + +For the full 2D dashboard demo see: + tools/3d_patch_generation_blender/unroll_realtime.py +""" + +import argparse +import os +import sys +from pathlib import Path + +os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0") + +import cv2 +from orisys.finger import DEFAULT_FINGER_EXPORT_DIR, DEFAULT_FINGER_MAP_CACHE, FingerRuntime + +_EXAMPLES_DIR = Path(__file__).resolve().parent +if str(_EXAMPLES_DIR) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_DIR)) + +from viz import UVMeshViewer, build_uv_surface_mesh, show_finger_uv_panels + + +def open_mesh_viewer(runtime, args): + domain = runtime.domain + if domain.surface is None: + raise RuntimeError( + "UV domain has no world_xyz surface map. Rebuild cache with include_surface=True." + ) + + uv_mesh = build_uv_surface_mesh(domain.surface.world_xyz, domain.valid_mask) + if args.no_3d: + return None, uv_mesh + + viewer = UVMeshViewer( + uv_mesh, + max_magnitude=args.magnitude_max, + threshold=args.depth_threshold if args.mesh_map == "depth" else args.flow_threshold, + color_mode=args.mesh_map, + ) + print( + f"3D mesh viewer: {len(uv_mesh.rest_vertices)} vertices, " + f"{len(uv_mesh.triangles)} triangles, {args.mesh_map} surface mode" + ) + return viewer, uv_mesh + + +def main(): + parser = argparse.ArgumentParser( + description="UV-unroll finger tactile pipeline with 3D mesh viewer" + ) + parser.add_argument( + "--export-dir", + default=DEFAULT_FINGER_EXPORT_DIR, + help="Blender export folder containing mesh.npz, camera_intrinsics.npz, world_to_cam.npy", + ) + parser.add_argument("--map_cache", default=DEFAULT_FINGER_MAP_CACHE, help="Cached UV domain .npz") + parser.add_argument("--camera_id", type=int, default=0) + parser.add_argument("--video", help="Video file path (overrides camera)") + parser.add_argument("--config", "-c", default="./config/dd02-ov.json", help="Orisys SDK config") + parser.add_argument( + "--source_scale", + type=float, + default=None, + help="Resize live source before UV remap (default: camera.source_scale in config)", + ) + parser.add_argument("--no_flip_source_vertical", action="store_true", help="Skip vertical flip before UV remap") + parser.add_argument("--show", action="store_true", help="Show 2D unrolled image window with flow arrows") + parser.add_argument("--verbose", action="store_true", help="SDK verbose logs") + parser.add_argument( + "--magnitude-max", + type=float, + default=None, + help="Upper optical-flow magnitude for 3D surface heat-map scaling; default uses frame percentile.", + ) + parser.add_argument( + "--mesh-map", + choices=["flow", "depth"], + default="depth", + help="Color the 3D mesh from optical-flow magnitude or the contact depth/divergence map.", + ) + parser.add_argument( + "--flow-threshold", + type=float, + default=0.4, + help="Minimum optical-flow magnitude to color 3D surface / draw 2D arrows (matches draw_arrows).", + ) + parser.add_argument( + "--depth-threshold", + type=float, + default=0.0, + help="Minimum depth/divergence value to color on the 3D mesh.", + ) + parser.add_argument("--no-3d", action="store_true", help="Disable Open3D mesh viewer") + args = parser.parse_args() + + video = args.camera_id + if args.video: + video = args.video + + try: + runtime = FingerRuntime.open( + export_dir=args.export_dir, + config=args.config, + video=video, + verbose=args.verbose, + ) + except RuntimeError as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) + + if args.verbose: + geometry = runtime.geometry + print( + f"UV geometry: native={geometry.native_width}x{geometry.native_height}, " + f"source={geometry.source_width}x{geometry.source_height} " + f"(scale={geometry.source_scale:g}), output={geometry.out_width}x{geometry.out_height}, " + f"map_cache={geometry.map_cache}" + ) + + mesh_viewer, _uv_mesh = open_mesh_viewer(runtime, args) + runtime.warmup(color=False) + tactile = runtime.tactile + + print("Press 'q' to quit, 'r' to reset tracker. Close the 3D window to stop.") + try: + while True: + ok, unrolled = runtime.grab_and_process(color=False) + if not ok: + break + + tactile.compute_deformation(decompose=True, check_motion=True, threshold=0) + tactile.compute_contact() + + fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = tactile.read_info( + tactile.info.FPS, + tactile.info.VRAW, + tactile.info.VNORMAL, + tactile.info.IMG, + tactile.info.CONTOUR, + tactile.info.CENTROID, + tactile.info.DEPTH, + tactile.info.FNORMAL, + tactile.info.FSHEARX, + tactile.info.FSHEARY, + ) + flow_mag = float((flow[:, :, 0] ** 2 + flow[:, :, 1] ** 2).mean() ** 0.5) if flow.size else 0.0 + + print( + f"FPS={fps:.2f} flow_rms={flow_mag:.2f} " + f"mask={tactile.mask_stats['coverage']:.2f} normal={fnormal:.2f} shear=({fshearx:.2f}, {fsheary:.2f})", + end="\r", + ) + + if mesh_viewer is not None: + if not mesh_viewer.update(flow=flow, scalar_map=depth_map): + print("\n3D viewer closed.") + break + + if args.show: + show_finger_uv_panels( + unrolled, + vnormal, + depth_map, + flow_threshold=args.flow_threshold, + ) + + key = cv2.waitKey(1) & 0xFF + if key == ord("q"): + print("\nExiting.") + break + if key == ord("r"): + tactile.reset() + print("\nTracker reset.") + finally: + runtime.close() + if mesh_viewer is not None: + mesh_viewer.close() + cv2.destroyAllWindows() + + +if __name__ == "__main__": + main() diff --git a/examples/finger_plot_data.py b/examples/finger_plot_data.py new file mode 100644 index 0000000..c966066 --- /dev/null +++ b/examples/finger_plot_data.py @@ -0,0 +1,101 @@ +""" +Finger UV pipeline with real-time force charts and OpenCV panels. + +Demonstrates: +1. UV-unrolled deformation arrows +2. Contact region overlay +3. PyQtGraph normal/shear force plots +4. Optional raw-frame recording (s/e keys) +""" + +import argparse +import os +import sys +import time +from pathlib import Path + +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" + +import cv2 +import orisys +from orisys.finger import FingerRuntime + +_EXAMPLES_DIR = Path(__file__).resolve().parent +if str(_EXAMPLES_DIR) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_DIR)) + +from viz import ForceChartWindow, FrameRecorder, lowpass_filter + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--video", "-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径") + parser.add_argument("--config", "-c", type=str, default="./config/dd02-ov.json", help="配置文件名称或路径") + parser.add_argument("--verbose", "-verbose", type=bool, default=True, help="是否输出详细日志") + args = parser.parse_args() + + charts = ForceChartWindow() + recorder = FrameRecorder() + + runtime = FingerRuntime.open( + config=args.config, + video=args.video, + verbose=args.verbose, + ) + runtime.warmup(color=True) + tactile = runtime.tactile + start_time = time.time() + + print("\n按键说明：'q' 退出程序，'s' 开始录像，'e' 结束录像。") + try: + while True: + ok, unrolled = runtime.grab_and_process(color=True) + if not ok: + break + + recorder.write_if_recording(tactile.frame) + + tactile.compute_deformation(check_motion=True, threshold=0) + tactile.compute_contact() + fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = tactile.read_info( + tactile.info.FPS, + tactile.info.VRAW, + tactile.info.VNORMAL, + tactile.info.IMG, + tactile.info.CONTOUR, + tactile.info.CENTROID, + tactile.info.DEPTH, + tactile.info.FNORMAL, + tactile.info.FSHEARX, + tactile.info.FSHEARY, + ) + print(f"FPS={fps:.2f}, 法向力={fnormal:.4f}, 切向力X={fshearx:.4f}, 切向力Y={fsheary:.4f}") + + arrows = orisys.util.draw_arrows(unrolled, flow, threshold=3, grid_spacing=10, arrow_scale=1.0) + cv2.imshow("arrows", arrows) + cv2.imshow("contact", orisys.util.draw_contact(img, contour, centroid)) + + charts.push(time.time() - start_time, fnormal, fshearx, fsheary) + display = charts.data.copy() + for channel in range(3): + display[:, 1, channel] = lowpass_filter( + charts.data[:, 1, channel], + cutoff_hz=5, + sample_rate_hz=30, + ) + charts.refresh(display) + + key = cv2.waitKey(1) + if key & 0xFF == ord("q"): + print("\n正在退出示例程序...") + break + if key == ord("s"): + recorder.start(tactile.frame) + if key == ord("e"): + recorder.stop() + finally: + runtime.close() + + +if __name__ == "__main__": + main() diff --git a/examples/plot_data.py b/examples/plot_data.py index 65aa22c..9e3cb7f 100644 --- a/examples/plot_data.py +++ b/examples/plot_data.py @@ -6,252 +6,90 @@ 2. 显示流场幅值热力图 3. 使用 PyQtGraph 实时绘制法向力与切向力曲线 4. 按键控制录像开始与结束 - -SDK version: 0.3.1 """ +import argparse import os +import sys +import time +from pathlib import Path + os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" + import cv2 import orisys -import numpy as np -import time -import argparse -try: - import pyqtgraph as pg -except: - print("PyQtGraph is not installed. Please install it using 'pip install pyqt5 pyqtgraph'.") - exit() -def lp_filter(series, alpha=0.5): - series = np.asarray(series) - if series.ndim == 0: - return float(series) - if series.shape[0] < 2: - return float(series[-1]) - return float(series[-1] * alpha + series[-2] * (1 - alpha)) +_EXAMPLES_DIR = Path(__file__).resolve().parent +if str(_EXAMPLES_DIR) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_DIR)) -def median_filter(series, window_size=5): - series = np.asarray(series) - if series.ndim == 0: - return float(series) - if series.shape[0] == 0: - return 0.0 +from viz import ForceChartWindow, FrameRecorder, draw_magnitude_map, lowpass_filter + + +def open_sensor(video: str, config: str, cal: str, verbose: bool): try: - window_n = max(1, int(window_size)) - except (TypeError, ValueError): - window_n = 1 - return float(np.median(series[-window_n:])) -def moving_average(series, window_size=5): - series = np.asarray(series) - if series.ndim == 0: - return float(series) - if series.shape[0] == 0: - return 0.0 - try: - window_n = max(1, int(window_size)) - except (TypeError, ValueError): - window_n = 1 - return float(np.mean(series[-window_n:])) - -def create_plot(): - p1 = win.addPlot(title="法向力") - p1.setLabel('bottom', '时间', units='s') - win.nextRow() - p2 = win.addPlot(title="切向力 X") - p2.setLabel('bottom', '时间', units='s') - win.nextRow() - p3 = win.addPlot(title="切向力 Y") - p3.setLabel('bottom', '时间', units='s') - plot_af = p1.plot(pen='k') - plot_tfx = p2.plot(pen='k') - plot_tfy = p3.plot(pen='k') - return p1, p2, p3, plot_af, plot_tfx, plot_tfy - -def update(plot_af, plot_tfx, plot_tfy, data, p1, p2, p3): - # global plot_af, plot_tfx, plot_tfy, data, p1, p2, p3 - plot_af.setData(data[:,:,0]) - p1.setYRange(min(data[:,1,0]),max(data[:,1,0])) - plot_tfx.setData(data[:,:,1]) - p2.setYRange(min(data[:,1,1]),max(data[:,1,1])) - plot_tfy.setData(data[:,:,2]) - p3.setYRange(min(data[:,1,2]),max(data[:,1,2])) - -# 深度图 -def draw_magnitude_map(vfield, threshold=3.0, colormap=cv2.COLORMAP_JET) -> np.ndarray: - """ - 绘制流场幅值伪彩色图。 - - Args: - vfield: 需要可视化的二维矢量场 - threshold: 最小显示阈值；低于阈值的区域显示为低值颜色 - colormap: OpenCV 颜色映射类型（默认 COLORMAP_JET） - - Returns: - np.ndarray: 400x400 的 BGR 伪彩图像 - """ - flow_field = vfield - - # 计算每个像素点的幅值 - magnitude = np.sqrt(flow_field[:, :, 0]**2 + flow_field[:, :, 1]**2) - - # 先降采样再插值回原尺寸，使显示更平滑 - downsample_factor = 10 - h, w = magnitude.shape - magnitude_downsampled = magnitude[::downsample_factor, ::downsample_factor] - magnitude_smoothed = cv2.resize(magnitude_downsampled, (w, h), interpolation=cv2.INTER_CUBIC) - - # 应用阈值，将低于阈值的区域置零 - magnitude_thresholded = magnitude_smoothed.copy() - magnitude_thresholded[magnitude_smoothed < threshold] = 0 - - # 归一化到 0-255 - mag_min, mag_max = magnitude_thresholded.min(), magnitude_thresholded.max() - if mag_max > mag_min: - magnitude_norm = ((magnitude_thresholded - mag_min) / (mag_max - mag_min) * 255).astype(np.uint8) - else: - magnitude_norm = np.zeros_like(magnitude_thresholded, dtype=np.uint8) - - # 应用伪彩色映射 - magnitude_colored = cv2.applyColorMap(magnitude_norm, colormap) - - # 缩放到固定输出尺寸 - out_img = cv2.resize(magnitude_colored, (400, 400), interpolation=cv2.INTER_CUBIC) - - return out_img - -# 实时图表 -npoints = 100 # 曲线缓冲区长度 -pg.setConfigOption('background', 'w') -pg.setConfigOption('foreground', 'k') -app = pg.mkQApp("Orisys Plot Example") -win = pg.GraphicsLayoutWidget(show=True, title="Orisys 实时力曲线") -win.setWindowTitle('Orisys 实时力曲线') -p1, p2, p3, plot_af, plot_tfx, plot_tfy = create_plot() -data = np.zeros([npoints,2,3]) -start_time = None # Will be set when sensor starts -for i in range(3): - data[:,0,i] = np.linspace(-npoints,0,npoints)/30 - + source = int(video) + except ValueError: + source = video + return orisys.Sensor(source, config_name=config, cal_path=cal, verbose=verbose) def main(): - global start_time parser = argparse.ArgumentParser() - parser.add_argument("--video","-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径") - parser.add_argument("--config","-c", type=str, default="./config/ddjx01.json", help="配置文件名称或路径") - parser.add_argument("--cal","-cal", type=str, default="./config/ddjx01.npy", help="标定文件路径") - parser.add_argument("--verbose","-verbose", type=bool, default=True, help="是否输出详细日志") + parser.add_argument("--video", "-v", type=str, default="0", help="视频源：摄像头编号或视频文件路径") + parser.add_argument("--config", "-c", type=str, default="./config/ddjx01.json", help="配置文件名称或路径") + parser.add_argument("--cal", "-cal", type=str, default="./config/ddjx01.npy", help="标定文件路径") + parser.add_argument("--verbose", "-verbose", type=bool, default=True, help="是否输出详细日志") args = parser.parse_args() - # 步骤 1：创建传感器对象（必需） - # config_name 可以是内置配置名称，也可以是自定义配置文件路径 - # cal_path 为标定文件路径；每台设备建议使用独立的标定文件 - - # 输入既可以是摄像头编号，也可以是视频文件路径 - try: - input = int(args.video) - is_camera = True - sensor = orisys.Sensor(input, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - except: - is_camera = False - sensor = orisys.Sensor(args.video, config_name=args.config, cal_path= args.cal, verbose=args.verbose) - # 下面保留了历史示例写法，便于参考 - # sensor1 = orisys.Sensor("./s1.mp4", config_name="./config/ddjx01.json", cal_path= "./config/ddjx01.npy", cuda=False, verbose=True) - + charts = ForceChartWindow() + recorder = FrameRecorder() + sensor = open_sensor(args.video, args.config, args.cal, args.verbose) start_time = time.time() - is_save = False - frame_count = 0 - writer = None - save_path = None + print("\n按键说明：'q' 退出程序，'s' 开始录像，'e' 结束录像。") - while True: - sensor.get_img() # 获取并拼接复眼图像 - - if is_save and writer is not None and sensor.frame is not None: - frame_count += 1 - save_img = sensor.frame - - # 若为灰度图，则先转换为 BGR 再写入视频 - if len(save_img.shape) == 2 or (len(save_img.shape) == 3 and save_img.shape[2] == 1): - if len(save_img.shape) == 3: - save_img = save_img.squeeze(2) - save_img = cv2.cvtColor(save_img, cv2.COLOR_GRAY2BGR) - - writer.write(save_img) + try: + while True: + sensor.get_img() + recorder.write_if_recording(sensor.frame) - sensor.compute_deformation(check_motion=True, threshold=0) # 计算形变与分解结果 - # 读取信息： - # 向量场：VRAW、VNORMAL、VSHEAR -> 原始 / 法向 / 切向 - # 标量：FNORMAL、FSHEARX、FSHEARY -> 法向力 / X 向切向力 / Y 向切向力 - fps, fn, fx, fy, flow, vnormal, vshear = sensor.read_info(sensor.info.FPS, sensor.info.FNORMAL, sensor.info.FSHEARX, sensor.info.FSHEARY, sensor.info.VRAW, sensor.info.VNORMAL,sensor.info.VSHEAR) - print(f"FPS={fps:.2f}, 法向力={fn:.4f}, 切向力X={fx:.4f}, 切向力Y={fy:.4f}") - # 显示形变矢量场 - arrows = orisys.util.draw_arrows(sensor.img, flow, threshold=5, grid_spacing=10, arrow_scale=1.0) - cv2.imshow("形变矢量场", arrows) - # 显示流场幅值图 - mag = draw_magnitude_map(flow) - cv2.imshow("流场幅值图", mag) + sensor.compute_deformation(check_motion=True, threshold=0) + fps, fn, fx, fy, flow, vnormal, vshear = sensor.read_info( + sensor.info.FPS, + sensor.info.FNORMAL, + sensor.info.FSHEARX, + sensor.info.FSHEARY, + sensor.info.VRAW, + sensor.info.VNORMAL, + sensor.info.VSHEAR, + ) + print(f"FPS={fps:.2f}, 法向力={fn:.4f}, 切向力X={fx:.4f}, 切向力Y={fy:.4f}") - # 更新实时曲线 - current_time = time.time() - start_time - data_filtered = data.copy() - for i in range(3): - data[:-1, 0, i] = data[1:, 0, i] # 时间轴左移 - data[:-1, 1, i] = data[1:, 1, i] # 力数据左移 - data[-1, 0, :] = current_time # 写入当前时间 - data[-1, 1, 0] = fn - data[-1, 1, 1] = fx - data[-1, 1, 2] = fy - # 对最新采样点做一阶低通，滤波对象是历史曲线而不是单个标量。 - data_filtered[-1, 1, 0] = moving_average(data[:, 1, 0]) - data_filtered[-1, 1, 1] = moving_average(data[:, 1, 1]) - data_filtered[-1, 1, 2] = moving_average(data[:, 1, 2]) - update(plot_af, plot_tfx, plot_tfy, data_filtered, p1, p2, p3) # 更新曲线 + arrows = orisys.util.draw_arrows(sensor.img, flow, threshold=5, grid_spacing=10, arrow_scale=1.0) + cv2.imshow("形变矢量场", arrows) + cv2.imshow("流场幅值图", draw_magnitude_map(flow)) - key = cv2.waitKey(1) - if key & 0xFF == ord("q"): - print("\n正在退出示例程序...") - break - if key == ord("s"): - save_path = f"./data/pulse_test.mp4" - # 确保输出目录存在 - os.makedirs(os.path.dirname(save_path), exist_ok=True) - # 从当前帧获取视频尺寸 - save_img = sensor.frame + charts.push(time.time() - start_time, fn, fx, fy) + display = charts.data.copy() + for channel in range(3): + display[:, 1, channel] = lowpass_filter( + charts.data[:, 1, channel], + cutoff_hz=5, + sample_rate_hz=30, + ) + charts.refresh(display) - if save_img is not None: - # 如为灰度图，先转换为 BGR 以获取视频尺寸 - temp_img = save_img.copy() - if len(temp_img.shape) == 2 or (len(temp_img.shape) == 3 and temp_img.shape[2] == 1): - if len(temp_img.shape) == 3: - temp_img = temp_img.squeeze(2) - temp_img = cv2.cvtColor(temp_img, cv2.COLOR_GRAY2BGR) - - frame_height, frame_width = temp_img.shape[:2] - writer = cv2.VideoWriter(save_path, cv2.VideoWriter_fourcc(*'mp4v'), 30, (frame_width, frame_height)) - if writer.isOpened(): - is_save = True - print(f"开始录像：{save_path}") - else: - print(f"无法创建录像文件：{save_path}") - writer = None - else: - print("当前没有可用帧，无法确定录像尺寸。") - if key == ord("e"): - is_save = False - if writer is not None: - writer.release() - writer = None - if save_path is not None: - print(f"录像结束：{save_path}，总帧数：{frame_count}") - else: - print(f"录像结束，总帧数：{frame_count}") - frame_count = 0 - - sensor.disconnect() # 断开传感器连接并释放资源 - + key = cv2.waitKey(1) + if key & 0xFF == ord("q"): + print("\n正在退出示例程序...") + break + if key == ord("s"): + recorder.start(sensor.frame) + if key == ord("e"): + recorder.stop() + finally: + sensor.disconnect() -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/examples/viz/__init__.py b/examples/viz/__init__.py new file mode 100644 index 0000000..e8b31d7 --- /dev/null +++ b/examples/viz/__init__.py @@ -0,0 +1,20 @@ +"""Shared visualization helpers for Orisys example scripts.""" + +from .finger_panels import show_finger_uv_panels +from .force_charts import ForceChartWindow +from .mesh_viewer import UVMeshViewer, build_uv_surface_mesh +from .opencv_maps import depth_colormap, draw_magnitude_map, frame_to_bgr +from .recording import FrameRecorder +from .signal_filters import lowpass_filter + +__all__ = [ + "ForceChartWindow", + "FrameRecorder", + "UVMeshViewer", + "build_uv_surface_mesh", + "depth_colormap", + "draw_magnitude_map", + "frame_to_bgr", + "lowpass_filter", + "show_finger_uv_panels", +] diff --git a/examples/viz/finger_panels.py b/examples/viz/finger_panels.py new file mode 100644 index 0000000..18ad80b --- /dev/null +++ b/examples/viz/finger_panels.py @@ -0,0 +1,29 @@ +"""Finger-specific OpenCV panels for example scripts.""" + +from __future__ import annotations + +import cv2 +import orisys + +from .opencv_maps import depth_colormap + + +def show_finger_uv_panels( + unrolled, + vnormal, + depth_map, + *, + flow_threshold: float = 0.4, + grid_spacing: int = 20, + arrow_scale: float = 1.0, +) -> None: + """Show depth colormap and UV unrolled flow arrows.""" + cv2.imshow("Depth", depth_colormap(depth_map)) + arrows = orisys.util.draw_arrows( + unrolled, + vnormal, + threshold=flow_threshold, + grid_spacing=grid_spacing, + arrow_scale=arrow_scale, + ) + cv2.imshow("UV Unrolled", arrows) diff --git a/examples/viz/force_charts.py b/examples/viz/force_charts.py new file mode 100644 index 0000000..dc6acfc --- /dev/null +++ b/examples/viz/force_charts.py @@ -0,0 +1,69 @@ +"""PyQtGraph real-time force chart window for examples.""" + +from __future__ import annotations + +import numpy as np + + +def _require_pyqtgraph(): + try: + import pyqtgraph as pg + except ImportError as exc: + raise ImportError( + "PyQtGraph is not installed. Install with: pip install pyqt5 pyqtgraph" + ) from exc + return pg + + +class ForceChartWindow: + """Rolling normal/shear force charts backed by PyQtGraph.""" + + def __init__( + self, + *, + npoints: int = 100, + sample_rate_hz: float = 30.0, + title: str = "Orisys 实时力曲线", + app_name: str = "Orisys Plot Example", + ) -> None: + pg = _require_pyqtgraph() + pg.setConfigOption("background", "w") + pg.setConfigOption("foreground", "k") + + self._pg = pg + self.npoints = npoints + self.sample_rate_hz = sample_rate_hz + self.app = pg.mkQApp(app_name) + self.win = pg.GraphicsLayoutWidget(show=True, title=title) + self.win.setWindowTitle(title) + + self.data = np.zeros((npoints, 2, 3), dtype=np.float64) + for channel in range(3): + self.data[:, 0, channel] = np.linspace(-npoints, 0, npoints) / sample_rate_hz + + p1 = self.win.addPlot(title="法向力") + p1.setLabel("bottom", "时间", units="s") + self.win.nextRow() + p2 = self.win.addPlot(title="切向力 X") + p2.setLabel("bottom", "时间", units="s") + self.win.nextRow() + p3 = self.win.addPlot(title="切向力 Y") + p3.setLabel("bottom", "时间", units="s") + + self._plots = (p1, p2, p3) + self._curves = (p1.plot(pen="k"), p2.plot(pen="k"), p3.plot(pen="k")) + + def push(self, timestamp: float, fn: float, fx: float, fy: float) -> None: + for channel in range(3): + self.data[:-1, 0, channel] = self.data[1:, 0, channel] + self.data[:-1, 1, channel] = self.data[1:, 1, channel] + self.data[-1, 0, :] = timestamp + self.data[-1, 1, 0] = fn + self.data[-1, 1, 1] = fx + self.data[-1, 1, 2] = fy + + def refresh(self, display_data: np.ndarray | None = None) -> None: + plot_data = self.data if display_data is None else display_data + for channel, (plot, curve) in enumerate(zip(self._plots, self._curves)): + curve.setData(plot_data[:, :, channel]) + plot.setYRange(float(plot_data[:, 1, channel].min()), float(plot_data[:, 1, channel].max())) diff --git a/examples/viz/mesh_viewer.py b/examples/viz/mesh_viewer.py new file mode 100644 index 0000000..cb25b11 --- /dev/null +++ b/examples/viz/mesh_viewer.py @@ -0,0 +1,4 @@ +"""Re-exports for finger 3D mesh examples; implementation lives in orisys.finger.""" + +from orisys.finger.mesh import * # noqa: F403 +from orisys.finger.viewer import * # noqa: F403 diff --git a/examples/viz/opencv_maps.py b/examples/viz/opencv_maps.py new file mode 100644 index 0000000..5a74c6f --- /dev/null +++ b/examples/viz/opencv_maps.py @@ -0,0 +1,57 @@ +"""OpenCV visualization helpers for example scripts.""" + +from __future__ import annotations + +import cv2 +import numpy as np + + +def depth_colormap(depth_map, colormap=cv2.COLORMAP_JET) -> np.ndarray: + """Normalize a scalar map to uint8 and apply an OpenCV colormap.""" + values = np.asarray(depth_map) + if values.size == 0: + return np.zeros((1, 1, 3), dtype=np.uint8) + if values.max() > values.min(): + normalized = ((values - values.min()) / (values.max() - values.min()) * 255).astype(np.uint8) + else: + normalized = np.zeros_like(values, dtype=np.uint8) + return cv2.applyColorMap(normalized, colormap) + + +def draw_magnitude_map( + vfield, + threshold=3.0, + colormap=cv2.COLORMAP_JET, + output_size=(400, 400), +) -> np.ndarray: + """Render a smoothed optical-flow magnitude map as a BGR pseudo-color image.""" + magnitude = np.sqrt(vfield[:, :, 0] ** 2 + vfield[:, :, 1] ** 2) + + downsample_factor = 10 + h, w = magnitude.shape + magnitude_downsampled = magnitude[::downsample_factor, ::downsample_factor] + magnitude_smoothed = cv2.resize(magnitude_downsampled, (w, h), interpolation=cv2.INTER_CUBIC) + + magnitude_thresholded = magnitude_smoothed.copy() + magnitude_thresholded[magnitude_smoothed < threshold] = 0 + + mag_min, mag_max = magnitude_thresholded.min(), magnitude_thresholded.max() + if mag_max > mag_min: + magnitude_norm = ((magnitude_thresholded - mag_min) / (mag_max - mag_min) * 255).astype(np.uint8) + else: + magnitude_norm = np.zeros_like(magnitude_thresholded, dtype=np.uint8) + + magnitude_colored = cv2.applyColorMap(magnitude_norm, colormap) + out_w, out_h = output_size + return cv2.resize(magnitude_colored, (out_w, out_h), interpolation=cv2.INTER_CUBIC) + + +def frame_to_bgr(frame: np.ndarray) -> np.ndarray: + """Convert a grayscale or single-channel frame to BGR for display/recording.""" + if frame is None: + raise ValueError("frame is None") + if len(frame.shape) == 2: + return cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR) + if len(frame.shape) == 3 and frame.shape[2] == 1: + return cv2.cvtColor(frame.squeeze(2), cv2.COLOR_GRAY2BGR) + return frame diff --git a/examples/viz/recording.py b/examples/viz/recording.py new file mode 100644 index 0000000..5376746 --- /dev/null +++ b/examples/viz/recording.py @@ -0,0 +1,71 @@ +"""Simple frame recording helper for example scripts.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import cv2 + +from .opencv_maps import frame_to_bgr + + +class FrameRecorder: + """Write camera frames to disk on demand (start/stop via keyboard hooks).""" + + def __init__(self, save_path: str | Path = "./data/pulse_test.mp4", fps: int = 30) -> None: + self.save_path = Path(save_path) + self.fps = fps + self._writer: cv2.VideoWriter | None = None + self._is_recording = False + self._frame_count = 0 + + @property + def is_recording(self) -> bool: + return self._is_recording + + def write_if_recording(self, frame) -> None: + if not self._is_recording or self._writer is None or frame is None: + return + self._writer.write(frame_to_bgr(frame)) + self._frame_count += 1 + + def start(self, frame) -> bool: + if frame is None: + print("当前没有可用帧，无法确定录像尺寸。") + return False + + self.save_path.parent.mkdir(parents=True, exist_ok=True) + bgr = frame_to_bgr(frame) + height, width = bgr.shape[:2] + writer = cv2.VideoWriter( + str(self.save_path), + cv2.VideoWriter_fourcc(*"mp4v"), + self.fps, + (width, height), + ) + if not writer.isOpened(): + print(f"无法创建录像文件：{self.save_path}") + return False + + self._writer = writer + self._is_recording = True + self._frame_count = 0 + print(f"开始录像：{self.save_path}") + return True + + def stop(self) -> tuple[Path | None, int]: + self._is_recording = False + if self._writer is not None: + self._writer.release() + self._writer = None + + path = self.save_path if self._frame_count else None + if path is not None: + print(f"录像结束：{path}，总帧数：{self._frame_count}") + else: + print(f"录像结束，总帧数：{self._frame_count}") + + count = self._frame_count + self._frame_count = 0 + return path, count diff --git a/examples/viz/signal_filters.py b/examples/viz/signal_filters.py new file mode 100644 index 0000000..beec846 --- /dev/null +++ b/examples/viz/signal_filters.py @@ -0,0 +1,27 @@ +"""Signal filtering helpers for example force charts.""" + +from __future__ import annotations + +import numpy as np +from scipy.signal import butter, filtfilt + +def lowpass_filter(data, cutoff_hz, sample_rate_hz=30, order=2) -> np.ndarray: + """Zero-phase Butterworth low-pass filter for uniformly sampled 1-D data.""" + y = np.asarray(data, dtype=np.float64) + if cutoff_hz == 0: + return y.copy() + if y.size == 0: + return y + + nyquist = 0.5 * sample_rate_hz + if cutoff_hz < 0 or cutoff_hz >= nyquist: + raise ValueError( + f"cutoff_hz must be in (0, {nyquist}) for sample_rate_hz={sample_rate_hz}" + ) + + b, a = butter(order, cutoff_hz / nyquist, btype="low") + padlen = 3 * (max(len(b), len(a)) - 1) + if y.size <= padlen: + return y.copy() + + return filtfilt(b, a, y) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..fa04ff4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,69 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "orisys" +description = "Orisys - Vision Based Tactile Sensor SDK" +readme = "README.md" +requires-python = ">=3.10" +authors = [{ name = "Orisys" }] +dynamic = ["version"] +dependencies = [ + "numpy>=1.24.4", + "opencv-python>=4.10.0.84", + "scipy>=1.10.1", + "open3d>=0.17.0", +] + +[project.optional-dependencies] +cuda12 = ["cupy-cuda12x>=12.3.0"] +cuda13 = ["cupy-cuda13x>=13.0.0"] +# Backward compatibility: 'cuda' defaults to CUDA 12.x +cuda = ["cupy-cuda12x>=12.3.0"] + +[tool.setuptools] +package-dir = { "" = "src" } +include-package-data = true + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.dynamic] +version = { attr = "orisys.__version__" } + +# Single source of truth for bundled non-code files. +# - dev_install = true -> pip install / editable install (via setup.py) +# - all items -> Nuitka wheel (build/wheel_layout.py) +[tool.orisys.bundle] + +# Use patterns (globs) and/or files (explicit names under source/). +[[tool.orisys.bundle.items]] +name = "configs" +source = "config" +target = "configs" +patterns = ["*.json"] +# files = ["dd02-ov.json", "ddjx01.json"] +dev_install = true + +[[tool.orisys.bundle.items]] +name = "examples" +source = "examples" +target = "examples" +patterns = ["*.py", "README.md", "viz/*.py"] +preserve_tree = true + +[[tool.orisys.bundle.items]] +name = "docs" +source = "docs" +target = "docs" +patterns = ["*.md", "*.pdf"] + +[[tool.orisys.bundle.items]] +name = "assets" +source = "assets" +target = "assets" +patterns = ["finger/30fps/*"] +preserve_tree = true +dev_install = true + diff --git a/scripts/build_nuitka.py b/scripts/build_nuitka.py deleted file mode 100644 index 99b82b2..0000000 --- a/scripts/build_nuitka.py +++ /dev/null @@ -1,228 +0,0 @@ -import os -import sys -import shutil -import subprocess -from pathlib import Path -from importlib.machinery import EXTENSION_SUFFIXES - -INCLUDE_LAYOUT = { - "configs": { - "source_dir": Path("config"), - "target_subdir": Path("configs"), - "patterns": ["*.json"], - }, - "examples": { - "source_dir": Path("examples"), - "target_subdir": Path("examples"), - "files": [ - Path("examples/basic_pipeline.py"), - Path("examples/two_sensor_pipeline.py"), - Path("examples/plot_data.py"), - ], - }, - "docs": { - "source_dir": Path("docs"), - "target_subdir": Path("docs"), - "patterns": ["*.md", "*.pdf"], - }, -} - -def install_nuitka(): - print("Checking Nuitka installation...") - try: - import nuitka # noqa: F401 - print("Nuitka is installed.") - except ImportError: - print("Installing Nuitka...") - subprocess.check_call( - [sys.executable, "-m", "pip", "install", "nuitka", "ordered-set", "zstandard"] - ) - - -def get_version() -> str: - init_file = Path("src/orisys/__init__.py") - if not init_file.exists(): - return "0.3.1" - for line in init_file.read_text(encoding="utf-8").splitlines(): - if line.startswith("__version__"): - return line.split("=", 1)[1].strip().strip('"').strip("'") - return "0.3.1" - - -def find_compiled_extension(output_dir: Path, module_name: str) -> Path: - candidates = [] - for suffix in EXTENSION_SUFFIXES: - candidates.extend(output_dir.glob(f"{module_name}*{suffix}")) - if not candidates: - all_files = ", ".join(p.name for p in output_dir.iterdir()) if output_dir.exists() else "(none)" - raise RuntimeError( - f"Nuitka did not produce a compiled extension for {module_name}. Found: {all_files}" - ) - return sorted(candidates, key=lambda p: len(p.name))[0] - - -def move_wheel_to_sdk(wheel_path: Path, version: str) -> Path: - sdk_version_dir = Path("SDK") / version - sdk_version_dir.mkdir(parents=True, exist_ok=True) - dest_wheel = sdk_version_dir / wheel_path.name - if dest_wheel.exists(): - dest_wheel.unlink() - shutil.move(str(wheel_path), str(dest_wheel)) - return dest_wheel - - -def compile_orisys(build_root: Path, env: dict[str, str]) -> Path: - module_output_dir = build_root / "orisys" - if module_output_dir.exists(): - shutil.rmtree(module_output_dir) - module_output_dir.mkdir(parents=True) - - cmd = [ - sys.executable, - "-m", - "nuitka", - "--module", - "--include-package=orisys", - "--output-dir=" + str(module_output_dir), - "--remove-output", - "--no-pyi-file", - "src/orisys", - ] - - print(f"Running: {' '.join(cmd)}") - subprocess.check_call(cmd, env=env) - - ext_file = find_compiled_extension(module_output_dir, "orisys") - print(f"Generated compiled extension for orisys: {ext_file}") - return ext_file - - -def copy_compiled_orisys(pkg_dir: Path, ext_file: Path) -> str: - ext_suffix = "".join(ext_file.suffixes) - dest_ext = pkg_dir / f"__init__{ext_suffix}" - shutil.copy(ext_file, dest_ext) - print(f"Copied compiled module orisys to {dest_ext}") - return ext_suffix - - -def copy_included_files(pkg_dir: Path) -> dict[str, int]: - copied_counts = {} - for group_name, group_cfg in INCLUDE_LAYOUT.items(): - src_dir = group_cfg["source_dir"] - dst_dir = pkg_dir / group_cfg["target_subdir"] - dst_dir.mkdir(exist_ok=True) - - copied = 0 - if "files" in group_cfg: - for file_path in group_cfg["files"]: - if file_path.exists(): - shutil.copy(file_path, dst_dir) - copied += 1 - print(f"Copied {group_name}: {file_path.name}") - else: - for pattern in group_cfg.get("patterns", []): - for src_file in src_dir.glob(pattern): - shutil.copy(src_file, dst_dir) - copied += 1 - print(f"Copied {group_name}: {src_file.name}") - - if group_name == "examples": - (dst_dir / "__init__.py").write_text("# Examples for orisys package\n", encoding="utf-8") - - copied_counts[group_name] = copied - return copied_counts - - -def get_package_data_entries(ext_suffix: str) -> dict[str, list[str]]: - package_data = { - "orisys": [f"*{ext_suffix}"], - "orisys.examples": ["*.py"], - } - for group_name, group_cfg in INCLUDE_LAYOUT.items(): - if group_name == "examples": - continue - target_subdir = str(group_cfg["target_subdir"]).replace("\\", "/") - for pattern in group_cfg.get("patterns", []): - package_data["orisys"].append(f"{target_subdir}/{pattern}") - return package_data - - -def build(): - install_nuitka() - - version = get_version() - output_dir = Path("build_nuitka") - dist_dir = Path("dist_nuitka") - - if output_dir.exists(): - shutil.rmtree(output_dir) - if dist_dir.exists(): - shutil.rmtree(dist_dir) - output_dir.mkdir() - dist_dir.mkdir() - - print("\n=== Compiling with Nuitka ===") - env = os.environ.copy() - env["PYTHONPATH"] = "src" + os.pathsep + env.get("PYTHONPATH", "") - ext_file = compile_orisys(output_dir, env) - - print("\n=== Preparing Wheel Structure ===") - wheel_root = dist_dir / "wheel_root" - pkg_dir = wheel_root / "orisys" - pkg_dir.mkdir(parents=True) - - ext_suffix = copy_compiled_orisys(pkg_dir, ext_file) - - copied_counts = copy_included_files(pkg_dir) - print(f"Copied {copied_counts['configs']} JSON config files") - print(f"Copied {copied_counts['examples']} example files to orisys/examples/") - print(f"Copied {copied_counts['docs']} doc files to orisys/docs/") - package_data = get_package_data_entries(ext_suffix) - - setup_py_path = dist_dir / "setup.py" - setup_content = f""" -from setuptools import setup, Distribution - -class BinaryDistribution(Distribution): - def has_ext_modules(self): - return True - -setup( - name="orisys", - version="{version}", - description="Orisys - Optical Tactile Sensor SDK (Nuitka Compiled)", - author="Orisys", - packages=['orisys', 'orisys.examples'], - package_dir={{'': 'wheel_root'}}, - package_data={package_data}, - include_package_data=True, - install_requires=[ - "numpy>=1.24.4", - "opencv-python>=4.10.0.84", - "scipy>=1.10.1", - ], - extras_require={{ - "cuda12": ["cupy-cuda12x>=12.3.0"], - "cuda13": ["cupy-cuda13x>=13.0.0"], - "cuda": ["cupy-cuda12x>=12.3.0"], - }}, - distclass=BinaryDistribution, - zip_safe=False, -) -""" - setup_py_path.write_text(setup_content, encoding="utf-8") - - print("\n=== Building Wheel ===") - subprocess.check_call([sys.executable, "setup.py", "bdist_wheel"], cwd=dist_dir) - - wheels = sorted((dist_dir / "dist").glob("*.whl")) - if wheels: - final_wheel = move_wheel_to_sdk(wheels[0], version) - print(f"\n[SUCCESS] Wheel created and moved to: {final_wheel}") - else: - print("\n[ERROR] Wheel not found.") - - -if __name__ == "__main__": - build() - diff --git a/scripts/qt_viewer_finger.py b/scripts/qt_viewer_finger.py new file mode 100644 index 0000000..b7889dc --- /dev/null +++ b/scripts/qt_viewer_finger.py @@ -0,0 +1,858 @@ +""" +Orisys SDK 实时监控 — Qt 统一界面 + +将所有可视化和曲线整合到一个窗口中，适合销售演示使用。 +SDK version: 0.3.1 +""" + +import os +import sys +os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" +import cv2 +def _fix_qt_plugin_path(): + """OpenCV redirects Qt to cv2/qt/plugins; use PyQt5's plugins instead.""" + try: + from PyQt5.QtCore import QLibraryInfo + os.environ["QT_QPA_PLATFORM_PLUGIN_PATH"] = QLibraryInfo.location( + QLibraryInfo.PluginsPath + ) + except Exception: + path = os.environ.get("QT_QPA_PLATFORM_PLUGIN_PATH", "") + if "cv2" in path.replace("\\", "/"): + os.environ.pop("QT_QPA_PLATFORM_PLUGIN_PATH", None) +_fix_qt_plugin_path() +import orisys +from orisys.finger import FingerRuntime, UVMeshViewer, build_uv_surface_mesh +import numpy as np +import time +import argparse +from datetime import datetime +from scipy.signal import butter, filtfilt + +# Force curve sampling rate (Hz), matches typical sensor loop ~30 FPS. +FORCE_SAMPLE_RATE_HZ = 30.0 +PLOT_BUFFER_POINTS = 120 +PLOT_WINDOW_SEC = PLOT_BUFFER_POINTS / FORCE_SAMPLE_RATE_HZ # 4 s window +# Minimum y-axis span so autoscale does not over-zoom on small fluctuations. +Y_AXIS_MIN_SPAN = 20000 +CURVE_THEME_COLOR = '#00E5FF' +NORMAL_FORCE_CAL_SLOPE = 3.92685554668795e-05 +NORMAL_FORCE_CAL_OFFSET = 0.0217596967278535 + +from PyQt5.QtWidgets import ( + QMainWindow, QWidget, QHBoxLayout, QVBoxLayout, + QLabel, QApplication, QSplitter, QPlainTextEdit, + QPushButton, QFileDialog, QLineEdit, +) +from PyQt5.QtCore import Qt, QTimer +from PyQt5 import QtGui +from PyQt5.QtGui import QImage, QPixmap, QFont, QIcon + +try: + import pyqtgraph as pg +except Exception: + print("PyQtGraph is not installed. Please install it using 'pip install pyqt5 pyqtgraph'.") + sys.exit(1) + +# --------------------------------------------------------------------------- +# 工具 +# --------------------------------------------------------------------------- + +def cv2_to_pixmap(img_bgr: np.ndarray) -> QPixmap: + rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB) + h, w, ch = rgb.shape + qimg = QImage(rgb.data, w, h, ch * w, QImage.Format_RGB888) + return QPixmap.fromImage(qimg) + +def draw_magnitude_map(vfield, threshold=3, colormap=cv2.COLORMAP_JET) -> np.ndarray: + flow_field = vfield + magnitude = np.sqrt(flow_field[:, :, 0] ** 2 + flow_field[:, :, 1] ** 2) + + downsample_factor = 10 + h, w = magnitude.shape + magnitude_downsampled = magnitude[::downsample_factor, ::downsample_factor] + magnitude_smoothed = cv2.resize(magnitude_downsampled, (w, h), interpolation=cv2.INTER_CUBIC) + + magnitude_thresholded = magnitude_smoothed.copy() + magnitude_thresholded[magnitude_smoothed < threshold] = 0 + + mag_min, mag_max = magnitude_thresholded.min(), magnitude_thresholded.max() + if mag_max > mag_min: + magnitude_norm = ((magnitude_thresholded - mag_min) / (mag_max - mag_min) * 255).astype(np.uint8) + else: + magnitude_norm = np.zeros_like(magnitude_thresholded, dtype=np.uint8) + + magnitude_colored = cv2.applyColorMap(magnitude_norm, colormap) + out_img = cv2.resize(magnitude_colored, (400, 400), interpolation=cv2.INTER_CUBIC) + return out_img + + +def lowpass_filter(data, cutoff_hz, sample_rate_hz=FORCE_SAMPLE_RATE_HZ, order=2): + """ + Zero-phase Butterworth low-pass filter for uniformly sampled 1-D data. + + Parameters + ---------- + data : array-like + Input samples acquired at `sample_rate_hz` (default 30 Hz). + cutoff_hz : float + -3 dB cutoff frequency in Hz; use 0 to skip filtering (return raw data). + Otherwise must satisfy 0 < cutoff_hz < sample_rate_hz / 2. + sample_rate_hz : float + Sampling rate of `data` in Hz. + order : int + Butterworth filter order. + + Returns + ------- + np.ndarray + Filtered 1-D array with the same length as `data`. + """ + y = np.asarray(data, dtype=np.float64) + if cutoff_hz == 0: + return y.copy() + if y.size == 0: + return y + + nyquist = 0.5 * sample_rate_hz + if cutoff_hz < 0 or cutoff_hz >= nyquist: + raise ValueError( + f"cutoff_hz must be in (0, {nyquist}) for sample_rate_hz={sample_rate_hz}" + ) + + b, a = butter(order, cutoff_hz / nyquist, btype="low") + padlen = 3 * (max(len(b), len(a)) - 1) + if y.size <= padlen: + return y.copy() + + return filtfilt(b, a, y) + + +def calibrate_normal_force(raw_value): + """Convert raw normal-force reading to Newtons using y = mx + c.""" + return NORMAL_FORCE_CAL_SLOPE * raw_value + NORMAL_FORCE_CAL_OFFSET + + +class TeeStream: + """同时写入多个流""" + def __init__(self, *streams): + self.streams = streams + + def write(self, text): + for s in self.streams: + s.write(text) + + def flush(self): + for s in self.streams: + s.flush() + + +class WidgetLogHandler: + """将 print 输出实时显示到 QPlainTextEdit""" + def __init__(self, widget: QPlainTextEdit): + self.widget = widget + self.buf = "" + + def write(self, text: str): + self.buf += text + while "\n" in self.buf: + line, self.buf = self.buf.split("\n", 1) + if line: + self.widget.appendPlainText(line) + + def flush(self): + if self.buf: + self.widget.appendPlainText(self.buf) + self.buf = "" + +# --------------------------------------------------------------------------- +# 样式表 +# --------------------------------------------------------------------------- + +STYLE = """ +QMainWindow { + background-color: #000000; +} + +/* ---- 图像面板 ---- */ +QLabel#arrowsLabel { + border: 1px solid #262626; + border-radius: 4px; + background-color: #0b0b0b; +} + +/* ---- 顶栏按钮 & 输入 - — */ +QPushButton { + background-color: #1a1a1a; + color: #999999; + border: 1px solid #333333; + border-radius: 4px; + padding: 6px 16px; + font-family: "SimHei"; + font-size: 13px; +} +QPushButton:hover { + background-color: #262626; + color: #cccccc; + border: 1px solid #555555; +} +QPushButton#btnStart { + background-color: #0d2818; + color: #4caf84; + border: 1px solid #1a4d30; +} +QPushButton#btnStart:hover { + background-color: #143d24; + color: #6fd4a0; +} +QPushButton#btnStart:disabled { + background-color: #0f0f0f; + color: #333333; + border: 1px solid #222222; +} +QPushButton#btnStop { + background-color: #2b1010; + color: #e05555; + border: 1px solid #4d1a1a; +} +QPushButton#btnStop:hover { + background-color: #3d1818; + color: #f07070; +} +QPushButton#btnStop:disabled { + background-color: #0f0f0f; + color: #333333; + border: 1px solid #222222; +} +QLineEdit { + background-color: #0d0d0d; + color: #999999; + border: 1px solid #333333; + border-radius: 4px; + padding: 5px 10px; + font-size: 13px; +} + +/* ---- 力曲线容器 ---- */ +GraphicsLayoutWidget { + border: 1px solid #262626; + border-radius: 4px; +} + +/* ---- 日志面板 ---- */ +QPlainTextEdit { + background-color: #0b0b0b; + color: #777777; + border: 1px solid #262626; + border-radius: 4px; + font-family: "Cascadia Code", "Consolas", "Courier New", monospace; + font-size: 13px; + padding: 8px; + selection-background-color: #1a3a4a; + selection-color: #cccccc; +} + +QPlainTextEdit QScrollBar:vertical { + background-color: #0b0b0b; + width: 6px; + margin: 1px; + border-radius: 3px; +} +QPlainTextEdit QScrollBar::handle:vertical { + background-color: #262626; + border-radius: 3px; + min-height: 30px; +} +QPlainTextEdit QScrollBar::handle:vertical:hover { + background-color: #3a3a3a; +} +QPlainTextEdit QScrollBar::add-line:vertical, +QPlainTextEdit QScrollBar::sub-line:vertical { + height: 0px; +} +QPlainTextEdit QScrollBar::add-page:vertical, +QPlainTextEdit QScrollBar::sub-page:vertical { + background: none; +} + +QSplitter::handle { + background-color: #1a1a1a; +} +QSplitter::handle:vertical { + height: 1px; +} +QSplitter::handle:horizontal { + width: 1px; +} +""" + +# --------------------------------------------------------------------------- +# 主窗口 +# --------------------------------------------------------------------------- + +class MainWindow(QMainWindow): + def __init__(self, args, log_file): + super().__init__() + self.setWindowTitle("Orisys SDK") + self.setStyleSheet(STYLE) + + logo_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "logo.png") + if os.path.exists(logo_path): + self.setWindowIcon(QIcon(logo_path)) + + self.sensor = None + self.mesh_viewer = None + self.start_time = None + self.args = args + + # 录像状态 + self.is_save = False + self.frame_count = 0 + self.writer = None + self.save_path = None + + # 曲线数据缓冲（先填满 npoints 再滚动；120 点 @ 30 Hz = 4 s） + self.npoints = PLOT_BUFFER_POINTS + self.lowpass_cutoff_hz = getattr(args, "cutoff", 5.0) + self._plot_sample_count = 0 + self._init_plot_buffer() + + self._setup_ui() + self._create_plots() + + # 同步参数到 UI + self.camera_edit.setText(args.video) + self.config_edit.setText(args.config) + self.cal_edit.setText(getattr(args, "cal", "")) + + # stdout → 日志文件 + 界面日志面板 + self.log_handler = WidgetLogHandler(self.log_widget) + sys.stdout = TeeStream(log_file, self.log_handler) + sys.stderr = sys.stdout + + self._y_range_alpha = 0.5 + + self.timer = QTimer() + self.timer.timeout.connect(self._on_timer) + self._running = False + + def _init_plot_buffer(self): + """Empty buffer; x-axis grows until npoints real samples, then scrolls.""" + self.data = np.full((self.npoints, 2, 3), np.nan, dtype=np.float64) + self._plot_sample_count = 0 + self._y_range_smooth = [None, None, None] + + # ---- UI 构建 ---- + + def _setup_ui(self): + central = QWidget() + self.setCentralWidget(central) + root_layout = QVBoxLayout(central) + root_layout.setContentsMargins(6, 6, 6, 6) + root_layout.setSpacing(6) + + # ---- 顶栏 ---- + bar = QHBoxLayout() + bar.setSpacing(8) + + self.btn_camera = QPushButton("摄像机") + self.btn_camera.setObjectName("btnCamera") + self.btn_camera.clicked.connect(self._select_camera) + bar.addWidget(self.btn_camera) + self.camera_edit = QLineEdit("1") + self.camera_edit.setPlaceholderText("摄像头编号 或 视频路径") + bar.addWidget(self.camera_edit, 1) + + self.btn_config = QPushButton("配置") + self.btn_config.setObjectName("btnConfig") + self.btn_config.clicked.connect(self._select_config) + bar.addWidget(self.btn_config) + self.config_edit = QLineEdit("./config/ddjx01.json") + bar.addWidget(self.config_edit, 1) + + self.btn_reset = QPushButton("重置") + self.btn_reset.setObjectName("btnReset") + self.btn_reset.setEnabled(False) + self.btn_reset.clicked.connect(self._on_reset) + bar.addWidget(self.btn_reset) + self.cal_edit = QLineEdit("./config/ddjx01.npy") + bar.addWidget(self.cal_edit, 1) + + self.btn_start = QPushButton("启动") + self.btn_start.setObjectName("btnStart") + self.btn_start.clicked.connect(self._on_start) + bar.addWidget(self.btn_start) + + self.btn_stop = QPushButton("暂停") + self.btn_stop.setObjectName("btnStop") + self.btn_stop.setEnabled(False) + self.btn_stop.clicked.connect(self._on_stop) + bar.addWidget(self.btn_stop) + + root_layout.addLayout(bar) + + # ---- 外层纵向 splitter：主区域 / 日志 ---- + v_splitter = QSplitter(Qt.Vertical) + + # ---- 主区域：图像 (左) | 曲线 (右) ---- + h_splitter = QSplitter(Qt.Horizontal) + + # 左侧图像面板 + left = QWidget() + left_layout = QVBoxLayout(left) + left_layout.setContentsMargins(0, 0, 0, 0) + left_layout.setSpacing(6) + + self.arrows_label = QLabel() + self.arrows_label.setObjectName("arrowsLabel") + self.arrows_label.setMinimumSize(160, 160) + self.arrows_label.setAlignment(Qt.AlignCenter) + self.arrows_label.setWordWrap(True) + self.arrows_label.setText(self._mesh_viewer_status_text()) + left_layout.addWidget(self.arrows_label) + + h_splitter.addWidget(left) + + # 右侧力曲线 + pg.setConfigOption('background', '#000000') + pg.setConfigOption('foreground', '#777777') + pg.setConfigOptions(antialias=True) + self.graph_widget = pg.GraphicsLayoutWidget() + self.graph_widget.setMinimumWidth(300) + h_splitter.addWidget(self.graph_widget) + + self.h_splitter = h_splitter + + v_splitter.addWidget(h_splitter) + + # ---- 底部日志面板 ---- + self.log_widget = QPlainTextEdit() + self.log_widget.setReadOnly(True) + self.log_widget.setMaximumBlockCount(2000) + font = QFont("Consolas", 12) + font.setStyleHint(QFont.Monospace) + self.log_widget.setFont(font) + v_splitter.addWidget(self.log_widget) + + v_splitter.setStretchFactor(0, 4) + v_splitter.setStretchFactor(1, 1) + + root_layout.addWidget(v_splitter) + + def _create_plots(self): + titles = ["法向力 (N)", "切向力 X", "切向力 Y"] + self.plots = [] + self.curves = [] + self.dot_glows = [] + self.dot_cores = [] + + axis_pen = pg.mkPen(color=(38, 38, 38), width=1) # #262626 + label_font = QFont("SimHei", 9) + + for i, title in enumerate(titles): + p = self.graph_widget.addPlot(title=None) + + # 标题：黑体 + p.setTitle(title, color='#999999', size='14pt') + p.titleLabel.setFont(QFont("SimHei", 14)) + + # X 轴标签 + p.setLabel('bottom', '时间', units='s') + p.getAxis('bottom').label.setFont(label_font) + + # 仅保留左 + 下轴线，隐藏右 + 上边框 + p.showAxis('top', False) + p.showAxis('right', False) + p.getAxis('left').setPen(axis_pen) + p.getAxis('bottom').setPen(axis_pen) + + # 仅横向网格，极淡 + p.showGrid(x=False, y=True, alpha=0.12) + + # 等宽刻度字体 + tick_font = QFont("Consolas", 11) + tick_font.setStyleHint(QFont.Monospace) + p.getAxis('left').setTickFont(tick_font) + p.getAxis('bottom').setTickFont(tick_font) + + # 曲线: 青色描边 + 向下渐变填充（与 demo_wuxi 一致） + line_pen = pg.mkPen(color=CURVE_THEME_COLOR, width=2) + # grad_brush = make_gradient_brush(CURVE_THEME_COLOR, max_alpha=120) + curve = p.plot(pen=line_pen, fillLevel=0) + + # 终点光晕: #00e5ff 30% 透明，半径 8 + dot_glow = p.plot([], [], pen=None, symbol='o', symbolSize=8, + symbolPen=None, symbolBrush=(0, 229, 255, 76)) + # 终点核心: 纯白，半径 4 + dot_core = p.plot([], [], pen=None, symbol='o', symbolSize=4, + symbolPen=None, symbolBrush=(255, 255, 255, 255)) + + p.setXRange(0, PLOT_WINDOW_SEC, padding=0) + p.enableAutoRange(x=False, y=False) + + self.plots.append(p) + self.curves.append(curve) + self.dot_glows.append(dot_glow) + self.dot_cores.append(dot_core) + + if i < 2: + self.graph_widget.nextRow() + + def adjust_square_images(self): + """根据窗口高度调整左侧面板宽度，使左侧图像保持正方形""" + top_bar_h = 48 + log_h = self.log_widget.height() + spacing = 20 + available = self.height() - top_bar_h - log_h - spacing + image_w = max(available, 160) + self.h_splitter.setSizes([image_w, self.width() - image_w]) + + def _mesh_viewer_status_text(self): + if self.mesh_viewer is None: + return "3D mesh viewer will open in a separate window\nwhen acquisition starts." + return "3D mesh viewer is running in a separate window." + + def _update_mesh_status_label(self): + self.arrows_label.setText(self._mesh_viewer_status_text()) + + def _init_mesh_viewer(self): + if self.sensor is None: + self._update_mesh_status_label() + return + domain = getattr(self.sensor, "domain", None) + surface = getattr(domain, "surface", None) + if surface is None or getattr(surface, "world_xyz", None) is None: + self.arrows_label.setText("3D mesh viewer unavailable.\nUV domain has no surface map.") + return + try: + uv_mesh = build_uv_surface_mesh(domain.surface.world_xyz, domain.valid_mask) + self.mesh_viewer = UVMeshViewer( + uv_mesh, + color_mode="flow", + threshold=2.0, + ) + self._update_mesh_status_label() + except Exception as exc: + self.mesh_viewer = None + self.arrows_label.setText(f"3D mesh viewer init failed.\n{exc}") + + def _close_mesh_viewer(self): + if self.mesh_viewer is not None: + self.mesh_viewer.close() + self.mesh_viewer = None + self._update_mesh_status_label() + + # ---- 帧处理 ---- + + def _on_timer(self): + if self.sensor is None: + return + runtime = self.sensor + tactile = runtime.tactile + + # isimg = sensor.get_img() + ok, unrolled = runtime.grab_and_process(color=True) + if not ok: + print("Failed to get image") + return + + # 旋转原始图像，使力方向与实际物理方向对齐 + # sensor.img = cv2.rotate(sensor.img, cv2.ROTATE_90_COUNTERCLOCKWISE) + # if sensor.frame is not None: + # sensor.frame = cv2.rotate(sensor.frame, cv2.ROTATE_90_COUNTERCLOCKWISE) + + # 录像写入 + if self.is_save and self.writer is not None and tactile.frame is not None: + self.frame_count += 1 + save_img = tactile.frame + if len(save_img.shape) == 2 or (len(save_img.shape) == 3 and save_img.shape[2] == 1): + if len(save_img.shape) == 3: + save_img = save_img.squeeze(2) + save_img = cv2.cvtColor(save_img, cv2.COLOR_GRAY2BGR) + self.writer.write(save_img) + + tactile.compute_deformation(check_motion=True, threshold=1) + tactile.compute_contact() + + fps, fn_raw, fx, fy, flow, vnormal, vshear, _ = tactile.read_info( + tactile.info.FPS, tactile.info.FNORMAL, + tactile.info.FSHEARX, tactile.info.FSHEARY, + tactile.info.VRAW, tactile.info.VNORMAL, + tactile.info.VSHEAR, + tactile.info.DEPTH + ) + fn = calibrate_normal_force(fn_raw) + + + print(f"FPS={fps:.2f}, 法向力={fn:.4f}, 切向力X={fx:.4f}, 切向力Y={fy:.4f}") + # print(f"CUPY Available: {sensor.CUPY_AVAILABLE}") + + if self.mesh_viewer is not None: + if not self.mesh_viewer.update(flow=flow): + self.mesh_viewer.close() + self.mesh_viewer = None + self.arrows_label.setText("3D mesh viewer closed.\nRestart acquisition to reopen it.") + + # 曲线：先顺序写入 npoints 个真实样本，满后再左移滚动 + current_time = time.time() - self.start_time + if self._plot_sample_count < self.npoints: + idx = self._plot_sample_count + self.data[idx, 0, :] = current_time + self.data[idx, 1, 0] = fn + self.data[idx, 1, 1] = fx + self.data[idx, 1, 2] = fy + self._plot_sample_count += 1 + else: + for i in range(3): + self.data[:-1, 0, i] = self.data[1:, 0, i] + self.data[:-1, 1, i] = self.data[1:, 1, i] + self.data[-1, 0, :] = current_time + self.data[-1, 1, 0] = fn + self.data[-1, 1, 1] = fx + self.data[-1, 1, 2] = fy + + n = self._plot_sample_count + if n == 0: + return + + data_plot = self.data[:n].copy() + for i in range(3): + data_plot[:, 1, i] = lowpass_filter( + data_plot[:, 1, i], + self.lowpass_cutoff_hz, + sample_rate_hz=FORCE_SAMPLE_RATE_HZ, + ) + + for i, curve in enumerate(self.curves): + x = data_plot[:, 0, i] + y = data_plot[:, 1, i] + curve.setData(x, y) + ymin, ymax = y.min(), y.max() + if self._y_range_smooth[i] is None: + self._y_range_smooth[i] = (ymin, ymax) + else: + prev_min, prev_max = self._y_range_smooth[i] + alpha = self._y_range_alpha + ymin = alpha * ymin + (1 - alpha) * prev_min + ymax = alpha * ymax + (1 - alpha) * prev_max + span = ymax - ymin + if span < Y_AXIS_MIN_SPAN: + mid = (ymin + ymax) / 2 + half = Y_AXIS_MIN_SPAN / 2 + ymin = mid - half + ymax = mid + half + self._y_range_smooth[i] = (ymin, ymax) + pad = max((ymax - ymin) * 0.12, 0.01) + self.plots[i].setYRange(ymin - pad, ymax + pad) + last_x = x[-1:] + last_y = y[-1:] + self.dot_glows[i].setData(last_x, last_y) + self.dot_cores[i].setData(last_x, last_y) + + if n < self.npoints: + for p in self.plots: + p.setXRange(0, PLOT_WINDOW_SEC, padding=0) + else: + x0 = float(data_plot[0, 0, 0]) + x1 = float(data_plot[-1, 0, 0]) + if x1 <= x0: + x1 = x0 + PLOT_WINDOW_SEC + for p in self.plots: + p.setXRange(x0, x1, padding=0) + + # ---- 键盘 & 录像 ---- + + def _on_start(self): + if self._running: + return + self.btn_start.setText("启动中...") + self.btn_start.setEnabled(False) + QApplication.processEvents() + + try: + src = self.camera_edit.text().strip() + cfg = self.config_edit.text() + cal = self.cal_edit.text().strip() + verbose = getattr(self.args, 'verbose', True) + try: + video = int(src) + except ValueError: + video = src + # self.sensor = orisys.Sensor(src, config_name=cfg, cal_path=cal, verbose=verbose) + self.sensor = FingerRuntime.open( + config=cfg, + calib=cal or None, + source_scale=None, + video=video, + verbose=verbose, + ) + self.sensor.warmup(color=True) + self._init_mesh_viewer() + except Exception as e: + self._close_mesh_viewer() + self.btn_start.setText("启动") + self.btn_start.setEnabled(True) + print(f"启动失败: {e}") + return + + self.start_time = time.time() + self._init_plot_buffer() + self._running = True + self.timer.start(10) + self.btn_start.setText("启动") + self.btn_stop.setEnabled(True) + self.btn_reset.setEnabled(True) + print(f"▶ 启动 — 摄像机={src}, 配置={cfg}, 标定={cal}") + + def _on_stop(self): + if not self._running: + return + self._running = False + self.timer.stop() + self.btn_stop.setText("停止中...") + self.btn_stop.setEnabled(False) + QApplication.processEvents() + + if self.sensor is not None: + self.sensor.close() + self.sensor = None + self._close_mesh_viewer() + + self.btn_start.setEnabled(True) + self.btn_stop.setText("暂停") + self.btn_reset.setEnabled(False) + print("⏸ 已暂停") + + def _select_camera(self): + text = self.camera_edit.text() + path, _ = QFileDialog.getOpenFileName( + self, "选择视频文件", "data", "视频 (*.mp4 *.avi *.mov);;所有文件 (*)", + ) + if path: + self.camera_edit.setText(path) + + def _select_config(self): + path, _ = QFileDialog.getOpenFileName( + self, "选择配置文件", "config", "JSON (*.json);;所有文件 (*)", + ) + if path: + self.config_edit.setText(path) + + def _on_reset(self): + """Reset optical flow tracker when drift occurs (same as basic_pipeline 'r' key).""" + if self.sensor is None or not self._running: + print("请先启动传感器后再重置追踪器。") + return + self.sensor.tactile.reset() + print("追踪器已重置。") + + def keyPressEvent(self, event): + key = event.key() + if key == Qt.Key_Escape: + if self.isFullScreen(): + self.showNormal() + else: + self.showFullScreen() + elif key == Qt.Key_Q: + print("\n正在退出示例程序...") + self.close() + elif key == Qt.Key_S: + self._start_recording() + elif key == Qt.Key_E: + self._stop_recording() + elif key == Qt.Key_R: + self._on_reset() + else: + super().keyPressEvent(event) + + def _start_recording(self): + if self.sensor is None: + print("请先启动采集") + return + self.save_path = "./data/pulse_test.mp4" + os.makedirs(os.path.dirname(self.save_path), exist_ok=True) + + save_img = self.sensor.tactile.frame + if save_img is None: + print("当前没有可用帧，无法确定录像尺寸。") + return + + temp_img = save_img.copy() + if len(temp_img.shape) == 2 or (len(temp_img.shape) == 3 and temp_img.shape[2] == 1): + if len(temp_img.shape) == 3: + temp_img = temp_img.squeeze(2) + temp_img = cv2.cvtColor(temp_img, cv2.COLOR_GRAY2BGR) + + fh, fw = temp_img.shape[:2] + self.writer = cv2.VideoWriter( + self.save_path, cv2.VideoWriter_fourcc(*'mp4v'), 30, (fw, fh), + ) + if self.writer.isOpened(): + self.is_save = True + print(f"开始录像：{self.save_path}") + else: + print(f"无法创建录像文件：{self.save_path}") + self.writer = None + + def _stop_recording(self): + self.is_save = False + if self.writer is not None: + self.writer.release() + self.writer = None + if self.save_path is not None: + print(f"录像结束：{self.save_path}，总帧数：{self.frame_count}") + else: + print(f"录像结束，总帧数：{self.frame_count}") + self.frame_count = 0 + + def closeEvent(self, event): + self.timer.stop() + if self.writer is not None: + self.writer.release() + self._close_mesh_viewer() + if self.sensor is not None: + self.sensor.close() + event.accept() + +# --------------------------------------------------------------------------- +# 入口 +# --------------------------------------------------------------------------- + +def main(): + script_dir = os.path.dirname(os.path.abspath(__file__)) + log_dir = os.path.join(script_dir, "logs") + os.makedirs(log_dir, exist_ok=True) + log_path = os.path.join(log_dir, f"qt_viewer_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log") + log_file = open(log_path, 'w', encoding='utf-8') + # stdout 到文件的临时重定向（创建 MainWindow 后会加上日志面板） + sys.stdout = log_file + sys.stderr = sys.stdout + + parser = argparse.ArgumentParser() + parser.add_argument("--video", "-v", type=str, default="0") + parser.add_argument("--config", "-c", type=str, default="./config/dd02-ov.json") + parser.add_argument("--cal", type=str, default="") + parser.add_argument("--verbose", "-verbose", type=bool, default=True) + parser.add_argument( + "--cutoff", + type=float, + default=10.0, + help="Low-pass cutoff for force curves (Hz), sample rate 30 Hz", + ) + args = parser.parse_args() + + print("\n按键说明：Q 退出 | R 重置追踪器 | S 开始录像 | E 结束录像 | ESC 全屏") + + # 高 DPI 适配（笔记本高分屏） + QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) + QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) + app = QApplication(sys.argv) + window = MainWindow(args, log_file) + window.showFullScreen() + QApplication.processEvents() + window.adjust_square_images() + + sys.exit(app.exec_()) + +if __name__ == '__main__': + main() diff --git a/setup.py b/setup.py index f72959d..6008341 100644 --- a/setup.py +++ b/setup.py @@ -1,32 +1,13 @@ -from setuptools import setup, find_packages +"""Setuptools entry point; bundle rules are read from pyproject.toml.""" + from pathlib import Path +import sys -config_dir = Path("config") -data_files = [] -if config_dir.exists(): - json_files = [str(p) for p in sorted(config_dir.glob("*.json"))] - if json_files: - data_files.append(("orisys/configs", json_files)) +from setuptools import setup -setup( - name="orisys", - version="0.4.0", - description="Orisys - Optical Tactile Sensor SDK", - author="Your Name", - package_dir={"": "src"}, - packages=find_packages(where="src"), - include_package_data=True, - data_files=data_files, - install_requires=[ - "numpy>=1.24.4", - "opencv-python>=4.10.0.84", - "scipy>=1.10.1", - ], - extras_require={ - "cuda12": ["cupy-cuda12x>=12.3.0"], - "cuda13": ["cupy-cuda13x>=13.0.0"], - # For backward compatibility, 'cuda' defaults to CUDA 12.x - "cuda": ["cupy-cuda12x>=12.3.0"], - }, - python_requires=">=3.8", -) +REPO_ROOT = Path(__file__).resolve().parent +sys.path.insert(0, str(REPO_ROOT / "build")) + +from pyproject_utils import bundle_data_files + +setup(data_files=bundle_data_files(REPO_ROOT)) diff --git a/src/orisys/Stitcher.py b/src/orisys/Stitcher.py index f74e18a..b1e9008 100644 --- a/src/orisys/Stitcher.py +++ b/src/orisys/Stitcher.py @@ -5,7 +5,7 @@ from scipy import ndimage import cv2 from numba import njit -@njit(cache=True) +@njit(cache=False) def _blend_kernel(blend_patch, wsum_patch, cov_patch, src_patch, w_patch): """Fused incremental weighted-average composite (numba-accelerated).""" sh = src_patch.shape[0] diff --git a/src/orisys/__init__.py b/src/orisys/__init__.py index 562c61d..0b7eb55 100644 --- a/src/orisys/__init__.py +++ b/src/orisys/__init__.py @@ -11,6 +11,7 @@ __author__ = "Orisys" from .sensor import Sensor from . import util from . import configs +from . import finger CompoundEyeSensor = Sensor CameraSensor = Sensor @@ -28,6 +29,9 @@ __all__ = [ # util functions 'util', + # finger runtime helpers + 'finger', + # configs utilities functions 'list_config', 'export_config' diff --git a/src/orisys/finger/__init__.py b/src/orisys/finger/__init__.py new file mode 100644 index 0000000..b352e08 --- /dev/null +++ b/src/orisys/finger/__init__.py @@ -0,0 +1,59 @@ +"""Finger-specific UV runtime helpers built on top of the Orisys SDK.""" + +from .exports import ( + DEFAULT_FINGER_EXPORT_DIR, + DEFAULT_FINGER_MAP_CACHE, + FingerExportBundle, + FingerExportPaths, + load_finger_export_bundle, + resolve_export_paths, +) +from .mesh import ( + DEFAULT_SURFACE_COLOR, + UVSurfaceMesh, + build_uv_surface_mesh, + compute_uv_surface_frames, + deform_mesh_vertices, + flow_magnitude_vertex_colors, + lift_uv_deformation_to_3d, + lift_uv_flow_to_3d, + mesh_uv_to_world_scale, + scalar_map_vertex_colors, + sample_field_at_mesh_vertices, +) +from .runtime import ( + FingerGeometry, + FingerRuntime, + TactileProcessor, + open_source, + resolve_finger_geometry, +) +from .viewer import UVMeshViewer, init_uv_mesh_viewer, update_uv_mesh_viewer + +__all__ = [ + "DEFAULT_FINGER_EXPORT_DIR", + "DEFAULT_FINGER_MAP_CACHE", + "DEFAULT_SURFACE_COLOR", + "FingerExportBundle", + "FingerExportPaths", + "FingerGeometry", + "FingerRuntime", + "TactileProcessor", + "UVSurfaceMesh", + "UVMeshViewer", + "build_uv_surface_mesh", + "compute_uv_surface_frames", + "deform_mesh_vertices", + "flow_magnitude_vertex_colors", + "init_uv_mesh_viewer", + "lift_uv_deformation_to_3d", + "lift_uv_flow_to_3d", + "load_finger_export_bundle", + "mesh_uv_to_world_scale", + "open_source", + "resolve_export_paths", + "resolve_finger_geometry", + "sample_field_at_mesh_vertices", + "scalar_map_vertex_colors", + "update_uv_mesh_viewer", +] diff --git a/src/orisys/finger/exports.py b/src/orisys/finger/exports.py new file mode 100644 index 0000000..de6d7cd --- /dev/null +++ b/src/orisys/finger/exports.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from ..unroll import ( + compute_uv_distortion_metrics, + load_exported_data, + recommend_edge_expand, + scaled_map_cache_path, +) + +DEFAULT_FINGER_EXPORT_DIR = "assets/finger/30fps" +DEFAULT_FINGER_MAP_CACHE = f"{DEFAULT_FINGER_EXPORT_DIR}/unroll_map_uv.npz" + + +@dataclass(frozen=True) +class FingerExportPaths: + """Resolved Blender export bundle paths for finger UV pipelines.""" + + export_dir: Path + mesh: Path + intrinsics: Path + pose: Path + + def as_tuple(self) -> tuple[str, str, str]: + return (str(self.mesh), str(self.intrinsics), str(self.pose)) + + +@dataclass(frozen=True) +class FingerExportBundle: + """Loaded Blender export data and UV map-cache settings.""" + + export_paths: FingerExportPaths + data: dict + map_cache: str + edge_expand: float + + +def _candidate_export_dirs(export_dir) -> list[Path]: + raw = Path(export_dir).expanduser() + if raw.is_absolute(): + return [raw] + + here = Path(__file__).resolve() + candidates = [ + Path.cwd() / raw, # repo root / user cwd + here.parents[1] / raw, # packaged wheel: orisys/assets/... + here.parents[3] / raw, # editable install: repo/assets/... + ] + unique: list[Path] = [] + for candidate in candidates: + resolved = candidate.resolve() + if resolved not in unique: + unique.append(resolved) + return unique + + +def resolve_export_paths(export_dir) -> FingerExportPaths: + """Return Blender export paths expected inside one folder.""" + checked: list[str] = [] + for root in _candidate_export_dirs(export_dir): + paths = FingerExportPaths( + export_dir=root, + mesh=root / "mesh.npz", + intrinsics=root / "camera_intrinsics.npz", + pose=root / "world_to_cam.npy", + ) + missing = [name for name, path in vars(paths).items() if name != "export_dir" and not path.is_file()] + if not missing: + return paths + checked.append(f"{root} (missing: {', '.join(missing)})") + + raise FileNotFoundError( + f"Missing export file(s) for finger bundle '{export_dir}'. " + f"Checked: {'; '.join(checked)}" + ) + + +def load_finger_export_bundle( + export_dir, + *, + map_cache=DEFAULT_FINGER_MAP_CACHE, + source_scale=1.0, +) -> FingerExportBundle: + """Resolve export paths, load bundle data, and pick the UV map cache path.""" + export_paths = resolve_export_paths(export_dir) + if Path(map_cache) == Path(DEFAULT_FINGER_MAP_CACHE): + map_cache = str(export_paths.export_dir / "unroll_map_uv.npz") + map_cache = scaled_map_cache_path(map_cache, source_scale) + mesh_path, intrinsics_path, pose_path = export_paths.as_tuple() + data = load_exported_data(mesh_path, intrinsics_path, pose_path) + distortion = compute_uv_distortion_metrics(data) + edge_expand = recommend_edge_expand(distortion) + return FingerExportBundle( + export_paths=export_paths, + data=data, + map_cache=map_cache, + edge_expand=float(edge_expand), + ) diff --git a/src/orisys/finger/mesh.py b/src/orisys/finger/mesh.py new file mode 100644 index 0000000..36e2842 --- /dev/null +++ b/src/orisys/finger/mesh.py @@ -0,0 +1,271 @@ +"""UV-domain finger surface mesh built from world_xyz and tactile field coloring.""" + +from __future__ import annotations + +import numpy as np + +# Default grey-white surface tone (linear RGB, 0-1). +DEFAULT_SURFACE_COLOR = (0.88, 0.88, 0.90) + + +class UVSurfaceMesh: + """Triangle mesh built from a UV flatten grid and world_xyz surface map.""" + + __slots__ = ( + "rest_vertices", + "triangles", + "vertex_ys", + "vertex_xs", + "index_map", + "valid_mask", + "shape", + "world_xyz", + "base_colors", + ) + + def __init__( + self, + rest_vertices, + triangles, + vertex_ys, + vertex_xs, + index_map, + valid_mask, + world_xyz, + base_colors, + ): + self.rest_vertices = rest_vertices + self.triangles = triangles + self.vertex_ys = vertex_ys + self.vertex_xs = vertex_xs + self.index_map = index_map + self.valid_mask = valid_mask + self.shape = valid_mask.shape + self.world_xyz = world_xyz + self.base_colors = base_colors + + +def build_uv_surface_mesh( + world_xyz: np.ndarray, + valid_mask: np.ndarray, + base_color=DEFAULT_SURFACE_COLOR, +) -> UVSurfaceMesh: + """Build a triangle mesh from UV grid world_xyz and valid_mask. + + Each valid UV pixel becomes one vertex. Adjacent valid 2x2 cells form two + triangles. Invalid or non-finite vertices are excluded. + """ + world_xyz = np.asarray(world_xyz, dtype=np.float64) + valid_mask = np.asarray(valid_mask, dtype=bool) + if world_xyz.shape[:2] != valid_mask.shape: + raise ValueError("world_xyz and valid_mask must share the same HxW shape") + + h, w = valid_mask.shape + finite = np.all(np.isfinite(world_xyz), axis=2) + vertex_mask = valid_mask & finite + + index_map = np.full((h, w), -1, dtype=np.int32) + ys, xs = np.nonzero(vertex_mask) + num_vertices = len(ys) + index_map[ys, xs] = np.arange(num_vertices, dtype=np.int32) + + rest_vertices = world_xyz[ys, xs].astype(np.float64) + color = np.asarray(base_color, dtype=np.float64).reshape(1, 3) + base_colors = np.tile(color, (num_vertices, 1)) + + triangles = [] + for y in range(h - 1): + for x in range(w - 1): + i00 = index_map[y, x] + i10 = index_map[y, x + 1] + i01 = index_map[y + 1, x] + i11 = index_map[y + 1, x + 1] + if min(i00, i10, i01, i11) < 0: + continue + triangles.append((i00, i10, i01)) + triangles.append((i10, i11, i01)) + + if not triangles: + raise ValueError("UV surface mesh has no valid triangles.") + + return UVSurfaceMesh( + rest_vertices=rest_vertices, + triangles=np.asarray(triangles, dtype=np.int32), + vertex_ys=ys.astype(np.int32), + vertex_xs=xs.astype(np.int32), + index_map=index_map, + valid_mask=vertex_mask, + world_xyz=world_xyz, + base_colors=base_colors, + ) + + +def compute_uv_surface_frames(world_xyz: np.ndarray, valid_mask: np.ndarray): + """Estimate tangent frame and normals on the UV surface grid.""" + valid_mask = np.asarray(valid_mask, dtype=bool) + filled = np.asarray(world_xyz, dtype=np.float64).copy() + invalid = ~valid_mask | ~np.all(np.isfinite(filled), axis=2) + filled[invalid] = 0.0 + + d_y = np.gradient(filled, axis=0) + d_x = np.gradient(filled, axis=1) + normals = np.cross(d_x, d_y) + norm = np.linalg.norm(normals, axis=2, keepdims=True) + normals = np.divide(normals, norm, out=np.zeros_like(normals), where=norm > 1e-8) + d_x[invalid] = 0.0 + d_y[invalid] = 0.0 + normals[invalid] = 0.0 + return d_x, d_y, normals + + +def mesh_uv_to_world_scale(world_xyz: np.ndarray, valid_mask: np.ndarray) -> float: + """Approximate world-units per UV pixel for deformation scaling.""" + points = world_xyz[valid_mask] + if len(points) == 0: + return 1e-3 + extent = np.nanmax(points, axis=0) - np.nanmin(points, axis=0) + mesh_span = float(np.max(extent)) + h, w = world_xyz.shape[:2] + return mesh_span / max(h, w, 1) + + +def lift_uv_deformation_to_3d( + vshear: np.ndarray, + vnormal: np.ndarray, + tangent_x: np.ndarray, + tangent_y: np.ndarray, + normals: np.ndarray, + normal_gain: float = 1.0, +) -> np.ndarray: + """Lift UV-space shear/normal HHD fields to 3D tangent-plane displacement.""" + tx_norm = np.linalg.norm(tangent_x, axis=2, keepdims=True) + ty_norm = np.linalg.norm(tangent_y, axis=2, keepdims=True) + t1 = np.divide(tangent_x, tx_norm, out=np.zeros_like(tangent_x), where=tx_norm > 1e-8) + t2 = np.divide(tangent_y, ty_norm, out=np.zeros_like(tangent_y), where=ty_norm > 1e-8) + + shear_3d = vshear[:, :, 0:1] * t1 + vshear[:, :, 1:2] * t2 + normal_mag = np.linalg.norm(vnormal, axis=2, keepdims=True) * float(normal_gain) + normal_3d = normal_mag * normals + + normal_component = np.sum(shear_3d * normals, axis=2, keepdims=True) + shear_tangent = shear_3d - normal_component * normals + return shear_tangent + normal_3d + + +def lift_uv_flow_to_3d( + flow: np.ndarray, + tangent_x: np.ndarray, + tangent_y: np.ndarray, + normals: np.ndarray, +) -> np.ndarray: + """Lift raw 2D optical flow on the UV grid to tangent-plane 3D displacement.""" + tx_norm = np.linalg.norm(tangent_x, axis=2, keepdims=True) + ty_norm = np.linalg.norm(tangent_y, axis=2, keepdims=True) + t1 = np.divide(tangent_x, tx_norm, out=np.zeros_like(tangent_x), where=tx_norm > 1e-8) + t2 = np.divide(tangent_y, ty_norm, out=np.zeros_like(tangent_y), where=ty_norm > 1e-8) + flow_3d = flow[:, :, 0:1] * t1 + flow[:, :, 1:2] * t2 + normal_component = np.sum(flow_3d * normals, axis=2, keepdims=True) + return flow_3d - normal_component * normals + + +def sample_field_at_mesh_vertices(field: np.ndarray, mesh: UVSurfaceMesh) -> np.ndarray: + """Sample a per-pixel UV field at mesh vertex locations.""" + sampled = field[mesh.vertex_ys, mesh.vertex_xs] + if sampled.ndim == 1: + return sampled.astype(np.float64) + return sampled.astype(np.float64) + + +def flow_magnitude_vertex_colors( + flow: np.ndarray | None, + mesh: UVSurfaceMesh, + *, + base_color=DEFAULT_SURFACE_COLOR, + max_magnitude: float | None = None, + threshold: float = 0.0, +) -> np.ndarray: + """Color mesh vertices from UV-space optical-flow magnitude. + + Magnitude matches ``orisys.util.draw_arrows``: ``2 * ||flow||``. + Vertices below ``threshold`` keep the base grey-white surface color. + """ + colors = mesh.base_colors.copy() + if flow is None or flow.size == 0: + return colors + + flow = np.asarray(flow, dtype=np.float64) + mag = 2.0 * np.linalg.norm(flow, axis=2) + vertex_mag = mag[mesh.vertex_ys, mesh.vertex_xs] + finite = np.isfinite(vertex_mag) + active = finite & (vertex_mag >= float(threshold)) + if not np.any(active): + return colors + + active_mag = vertex_mag[active] + if max_magnitude is None or max_magnitude <= 0: + max_magnitude = max(float(np.percentile(active_mag, 95)), 1e-6) + + value = np.clip(vertex_mag / max_magnitude, 0.0, 1.0) + heat = np.column_stack( + ( + value, + np.clip(1.0 - np.abs(value - 0.5) * 2.0, 0.0, 1.0), + 1.0 - value, + ) + ).astype(np.float64) + base = np.asarray(base_color, dtype=np.float64) + colors[active] = 0.25 * base + 0.75 * heat[active] + return colors + + +def scalar_map_vertex_colors( + scalar_map: np.ndarray | None, + mesh: UVSurfaceMesh, + *, + base_color=DEFAULT_SURFACE_COLOR, + value_min: float | None = None, + value_max: float | None = None, + threshold: float = 0.0, +) -> np.ndarray: + """Color mesh vertices from a scalar UV map such as depth/divergence.""" + colors = mesh.base_colors.copy() + if scalar_map is None or scalar_map.size == 0: + return colors + + scalar_map = np.asarray(scalar_map, dtype=np.float64) + vertex_values = scalar_map[mesh.vertex_ys, mesh.vertex_xs] + finite = np.isfinite(vertex_values) + active = finite & (vertex_values > float(threshold)) + if not np.any(active): + return colors + + active_values = vertex_values[active] + if value_min is None: + value_min = float(np.percentile(active_values, 5)) + if value_max is None: + value_max = float(np.percentile(active_values, 95)) + if value_max <= value_min: + value_max = value_min + 1e-6 + + value = np.clip((vertex_values - value_min) / (value_max - value_min), 0.0, 1.0) + heat = np.column_stack( + ( + value, + np.clip(1.0 - np.abs(value - 0.5) * 2.0, 0.0, 1.0), + 1.0 - value, + ) + ).astype(np.float64) + base = np.asarray(base_color, dtype=np.float64) + colors[active] = 0.25 * base + 0.75 * heat[active] + return colors + + +def deform_mesh_vertices( + mesh: UVSurfaceMesh, + displacement_map: np.ndarray, + pixel_to_world: float, + deform_scale: float = 1.0, +) -> np.ndarray: + """Apply lifted 3D displacement to rest mesh vertices.""" + disp = sample_field_at_mesh_vertices(displacement_map, mesh) + return mesh.rest_vertices + disp * float(pixel_to_world) * float(deform_scale) diff --git a/src/orisys/finger/runtime.py b/src/orisys/finger/runtime.py new file mode 100644 index 0000000..7aaccc2 --- /dev/null +++ b/src/orisys/finger/runtime.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +from dataclasses import dataclass + +import cv2 +import numpy as np + +from ..configs import load_config +from ..frame_processors import UVUnrollFrameProcessor +from ..sensor import Sensor +from ..unroll import ( + build_or_load_domain, + load_calib_json, + resolve_uv_runtime_sizes, + scale_camera_geometry, +) +from .exports import ( + DEFAULT_FINGER_EXPORT_DIR, + DEFAULT_FINGER_MAP_CACHE, + FingerExportPaths, + load_finger_export_bundle, +) + + +def normalize_video_source(video): + """Accept camera index, numeric string, or filesystem video path.""" + if video is None: + return 0 + if isinstance(video, str): + stripped = video.strip().strip("/\\") + if not stripped: + return 0 + if stripped.isdigit(): + return int(stripped) + return stripped + return video + + +def open_source(video, config): + capture = cv2.VideoCapture(normalize_video_source(video)) + capture.set(cv2.CAP_PROP_FRAME_WIDTH, config.get("camera", {}).get("resolution_raw")[0]) + capture.set(cv2.CAP_PROP_FRAME_HEIGHT, config.get("camera", {}).get("resolution_raw")[1]) + capture.set(cv2.CAP_PROP_FPS, config.get("camera", {}).get("fps", 30)) + capture.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter.fourcc('M', 'J', 'P', 'G')) + return capture + + +class TactileProcessor: + """Narrow tactile-analysis wrapper around ``Sensor`` external-frame mode.""" + + def __init__(self, sensor: Sensor, *, default_valid_mask=None): + self._sensor = sensor + self._default_valid_mask = default_valid_mask + + @property + def raw_sensor(self) -> Sensor: + """Escape hatch for legacy callers that still need the full ``Sensor``.""" + return self._sensor + + @property + def info(self): + return self._sensor.info + + @property + def frame(self): + return self._sensor.frame + + @property + def img(self): + return self._sensor.img + + @property + def flow(self): + return self._sensor.flow + + @property + def vnormal(self): + return self._sensor.vnormal + + @property + def vshear(self): + return self._sensor.vshear + + @property + def depth_map(self): + return self._sensor.depth_map + + @property + def mask_stats(self): + return self._sensor.mask_stats + + @property + def valid_mask(self): + return self._sensor.valid_mask + + def push_frame(self, frame, *, already_gray=False, valid_mask=None): + """Feed one processed tactile frame into the analysis pipeline.""" + mask = self._default_valid_mask if valid_mask is None else valid_mask + return self._sensor.get_img( + frame=frame, + already_processed=True, + already_gray=already_gray, + valid_mask=mask, + ) + + def warmup(self): + return self._sensor.warmup_deformation() + + def compute_deformation(self, *args, **kwargs): + return self._sensor.compute_deformation(*args, **kwargs) + + def compute_contact(self, *args, **kwargs): + return self._sensor.compute_contact(*args, **kwargs) + + def read_info(self, *args): + return self._sensor.read_info(*args) + + def reset(self): + return self._sensor.reset() + + def disconnect(self): + return self._sensor.disconnect() + + +@dataclass(frozen=True) +class FingerGeometry: + """Resolved one-time geometry/bootstrap state for a finger UV runtime.""" + + config: dict + camera_config: dict + export_paths: FingerExportPaths + data: dict + source_K: np.ndarray + dist_coeffs: np.ndarray | None + native_width: int + native_height: int + source_width: int + source_height: int + out_width: int + out_height: int + source_scale: float + map_cache: str + edge_expand: float + + +def resolve_finger_geometry( + *, + export_dir, + config="./config/dd02-ov.json", + map_cache=DEFAULT_FINGER_MAP_CACHE, + source_scale=None, + calib=None, +) -> FingerGeometry: + """Resolve config/export/camera geometry for a UV finger runtime.""" + config = load_config(config) + camera_config = config.get("camera", {}) + out_width, out_height, source_scale = resolve_uv_runtime_sizes( + camera_config, + source_scale=source_scale, + ) + bundle = load_finger_export_bundle( + export_dir, + map_cache=map_cache, + source_scale=source_scale, + ) + + if calib: + source_K, dist_coeffs, img_size = load_calib_json(calib) + native_width, native_height = img_size + else: + source_K = bundle.data["K"] + dist_coeffs = None + native_width = int(bundle.data["width"]) + native_height = int(bundle.data["height"]) + + source_K, source_width, source_height = scale_camera_geometry( + source_K, + native_width, + native_height, + source_scale, + ) + + return FingerGeometry( + config=config, + camera_config=camera_config, + export_paths=bundle.export_paths, + data=bundle.data, + source_K=source_K, + dist_coeffs=dist_coeffs, + native_width=native_width, + native_height=native_height, + source_width=source_width, + source_height=source_height, + out_width=out_width, + out_height=out_height, + source_scale=float(source_scale), + map_cache=bundle.map_cache, + edge_expand=bundle.edge_expand, + ) + + +class FingerRuntime: + """Reusable UV finger runtime composed from capture, UV remap, and tactile analysis.""" + + def __init__( + self, + *, + geometry: FingerGeometry, + domain, + tactile: TactileProcessor, + frame_processor: UVUnrollFrameProcessor, + capture, + ): + self.geometry = geometry + self.domain = domain + self.tactile = tactile + self.sensor = tactile + self.frame_processor = frame_processor + self.capture = capture + + @classmethod + def open( + cls, + *, + export_dir=DEFAULT_FINGER_EXPORT_DIR, + config="./config/dd02-ov.json", + map_cache=DEFAULT_FINGER_MAP_CACHE, + source_scale=None, + calib=None, + video=0, + camera_id=None, + flip_source_vertical=True, + verbose=False, + mirror=False, + ) -> "FingerRuntime": + """Open a live finger runtime from export bundle, config, and video source.""" + if camera_id is not None: + video = camera_id + video = normalize_video_source(video) + geometry = resolve_finger_geometry( + export_dir=export_dir, + config=config, + map_cache=map_cache, + source_scale=source_scale, + calib=calib, + ) + + mesh_path, intrinsics_path, pose_path = geometry.export_paths.as_tuple() + domain = build_or_load_domain( + mesh_path, + intrinsics_path, + pose_path, + geometry.map_cache, + geometry.source_K, + geometry.source_width, + geometry.source_height, + geometry.out_width, + geometry.out_height, + edge_expand=geometry.edge_expand, + include_surface=True, + verbose=verbose, + source_scale=geometry.source_scale, + ) + domain.remap.ensure_fast_maps() + + capture = open_source(video, geometry.config) + if not capture.isOpened(): + raise RuntimeError("Cannot open camera/video source") + + sensor = Sensor("external", config_name=config, verbose=verbose, mirror=mirror) + frame_processor = UVUnrollFrameProcessor( + domain.remap, + camera_matrix=geometry.source_K if geometry.dist_coeffs is not None else None, + dist_coeffs=geometry.dist_coeffs, + source_width=geometry.source_width, + source_height=geometry.source_height, + flip_source_vertical=flip_source_vertical, + ) + tactile = TactileProcessor(sensor, default_valid_mask=frame_processor.valid_mask) + return cls( + geometry=geometry, + domain=domain, + tactile=tactile, + frame_processor=frame_processor, + capture=capture, + ) + + @property + def valid_mask(self): + return self.frame_processor.valid_mask + + def read_source(self): + return self.capture.read() + + def process_frame(self, frame, *, color=False): + """Prepare, unroll, and feed one source frame into the tactile processor.""" + if color: + prepared = self.frame_processor.prepare_source(frame) + unrolled = self.frame_processor.flatten_prepared(prepared) + self.tactile.push_frame(unrolled) + return unrolled + + prepared_gray = self.frame_processor.prepare_source_gray(frame) + unrolled = self.frame_processor.flatten_prepared_gray(prepared_gray) + self.tactile.push_frame(unrolled, already_gray=True) + return unrolled + + def grab_and_process(self, *, color=False): + ok, frame = self.read_source() + if not ok: + return False, None + return True, self.process_frame(frame, color=color) + + def get_img(self, *, color=False): + """Sensor-style alias: capture, unroll, and return the tactile frame or None.""" + ok, unrolled = self.grab_and_process(color=color) + return unrolled if ok else None + + def warmup(self, *, color=False): + ok, unrolled = self.grab_and_process(color=color) + if ok: + self.tactile.warmup() + return ok, unrolled + + def close(self): + if self.capture is not None: + self.capture.release() + self.tactile.disconnect() diff --git a/src/orisys/finger/viewer.py b/src/orisys/finger/viewer.py new file mode 100644 index 0000000..292285f --- /dev/null +++ b/src/orisys/finger/viewer.py @@ -0,0 +1,130 @@ +"""Open3D viewer for UV-domain finger surface meshes.""" + +from __future__ import annotations + +import numpy as np + +from .mesh import ( + UVSurfaceMesh, + flow_magnitude_vertex_colors, + scalar_map_vertex_colors, +) + + +def init_uv_mesh_viewer( + mesh: UVSurfaceMesh, + vertices: np.ndarray | None = None, + vertex_colors: np.ndarray | None = None, + window_name: str = "UV Finger Mesh 3D", +): + """Create Open3D viewer with an unlit triangle mesh.""" + import open3d as o3d + + if vertices is None: + vertices = mesh.rest_vertices + if vertex_colors is None: + vertex_colors = mesh.base_colors + + vis = o3d.visualization.Visualizer() + vis.create_window(window_name=window_name, width=1280, height=720) + + tri_mesh = o3d.geometry.TriangleMesh() + tri_mesh.vertices = o3d.utility.Vector3dVector(vertices.astype(np.float64)) + tri_mesh.triangles = o3d.utility.Vector3iVector(mesh.triangles.astype(np.int32)) + tri_mesh.vertex_colors = o3d.utility.Vector3dVector(vertex_colors.astype(np.float64)) + tri_mesh.compute_vertex_normals() + vis.add_geometry(tri_mesh) + + render_option = vis.get_render_option() + render_option.mesh_show_back_face = True + if hasattr(render_option, "light_on"): + render_option.light_on = False + render_option.background_color = np.array([0.05, 0.05, 0.08]) + vis.poll_events() + vis.update_renderer() + return vis, tri_mesh, o3d + + +def update_uv_mesh_viewer( + vis, + tri_mesh, + o3d, + vertices: np.ndarray, + vertex_colors: np.ndarray, +): + """Update mesh vertices/colors.""" + tri_mesh.vertices = o3d.utility.Vector3dVector(vertices.astype(np.float64)) + tri_mesh.vertex_colors = o3d.utility.Vector3dVector(vertex_colors.astype(np.float64)) + tri_mesh.compute_vertex_normals() + vis.update_geometry(tri_mesh) + + vis.poll_events() + vis.update_renderer() + + +class UVMeshViewer: + """Stateful UV mesh 3D viewer driven by UV-space tactile fields.""" + + def __init__( + self, + mesh: UVSurfaceMesh, + *, + max_magnitude: float | None = None, + threshold: float = 0.0, + color_mode: str = "flow", + value_min: float | None = None, + value_max: float | None = None, + ): + self.mesh = mesh + self.max_magnitude = max_magnitude + self.threshold = threshold + self.color_mode = color_mode + self.value_min = value_min + self.value_max = value_max + self.vis = None + self.tri_mesh = None + self._o3d = None + + def update(self, flow=None, scalar_map=None): + """Update viewer from UV-space maps. Returns False if window closed.""" + if self.color_mode == "depth": + colors = scalar_map_vertex_colors( + scalar_map, + self.mesh, + value_min=self.value_min, + value_max=self.value_max, + threshold=self.threshold, + ) + else: + colors = flow_magnitude_vertex_colors( + flow, + self.mesh, + max_magnitude=self.max_magnitude, + threshold=self.threshold, + ) + vertices = self.mesh.rest_vertices + + if self.vis is None: + self.vis, self.tri_mesh, self._o3d = init_uv_mesh_viewer( + self.mesh, + vertices=vertices, + vertex_colors=colors, + ) + return True + + if not self.vis.poll_events(): + return False + + update_uv_mesh_viewer( + self.vis, + self.tri_mesh, + self._o3d, + vertices, + colors, + ) + return True + + def close(self): + if self.vis is not None: + self.vis.destroy_window() + self.vis = None diff --git a/src/orisys/hhd/solvers/masked_sparse.py b/src/orisys/hhd/solvers/masked_sparse.py index 58c98b2..e00ef0f 100644 --- a/src/orisys/hhd/solvers/masked_sparse.py +++ b/src/orisys/hhd/solvers/masked_sparse.py @@ -17,7 +17,10 @@ class MaskedSparseSolver: if rhs.size == 0: return numpy.zeros_like(f, dtype=numpy.float32) - sol_active = domain.factor(rhs) # (N,) reduced-system potential + # MaskedDomain builds a positive diagonal / negative off-diagonal stencil + # for (-Laplace). Negate the direct solve result so the masked sparse path + # matches the FFT solver convention used elsewhere: Laplace(u) = rhs. + sol_active = -domain.factor(rhs) # (N,) reduced-system potential return domain.scatter(sol_active, f) def solve_pair(self, f0, f1, domain: MaskedDomain): @@ -28,9 +31,9 @@ class MaskedSparseSolver: return empty, empty.copy() try: - sol_pair = domain.factor(rhs_pair) # (N, 2) reduced-system potentials + sol_pair = -domain.factor(rhs_pair) # (N, 2) reduced-system potentials except Exception: - sol_pair = numpy.column_stack( + sol_pair = -numpy.column_stack( [domain.factor(rhs_pair[:, 0]), domain.factor(rhs_pair[:, 1])] ) return domain.scatter_pair(sol_pair, f0) diff --git a/tests/test_finger_runtime.py b/tests/test_finger_runtime.py new file mode 100644 index 0000000..ea06304 --- /dev/null +++ b/tests/test_finger_runtime.py @@ -0,0 +1,180 @@ +from pathlib import Path + +import numpy as np +import pytest + +import orisys +from orisys.finger import TactileProcessor, load_finger_export_bundle, resolve_export_paths +from orisys.finger.exports import DEFAULT_FINGER_MAP_CACHE +from orisys.finger.runtime import resolve_finger_geometry + + +def _touch_bundle(root: Path): + for name in ("mesh.npz", "camera_intrinsics.npz", "world_to_cam.npy"): + (root / name).write_bytes(b"") + + +def test_resolve_export_paths_returns_expected_bundle(tmp_path): + _touch_bundle(tmp_path) + + bundle = resolve_export_paths(tmp_path) + + assert bundle.export_dir == tmp_path.resolve() + assert bundle.mesh.name == "mesh.npz" + assert bundle.intrinsics.name == "camera_intrinsics.npz" + assert bundle.pose.name == "world_to_cam.npy" + + +def test_load_finger_export_bundle_rehomes_default_map_cache(tmp_path, monkeypatch): + _touch_bundle(tmp_path) + + import orisys.finger.exports as exports_mod + + monkeypatch.setattr( + exports_mod, + "load_exported_data", + lambda *args: {"K": np.eye(3), "width": 1280, "height": 720}, + ) + monkeypatch.setattr(exports_mod, "compute_uv_distortion_metrics", lambda data: {"ok": True}) + monkeypatch.setattr(exports_mod, "recommend_edge_expand", lambda _: 1.2) + + bundle = load_finger_export_bundle( + tmp_path, + map_cache=DEFAULT_FINGER_MAP_CACHE, + source_scale=1.0, + ) + + assert Path(bundle.map_cache) == (tmp_path / "unroll_map_uv.npz").resolve() + assert bundle.edge_expand == 1.2 + + +def test_resolve_finger_geometry_uses_config_defaults(tmp_path, monkeypatch): + _touch_bundle(tmp_path) + + import orisys.finger.exports as exports_mod + import orisys.finger.runtime as runtime_mod + + monkeypatch.setattr( + runtime_mod, + "load_config", + lambda _: {"camera": {"resolution_stitch": [280, 300], "source_scale": 0.4}}, + ) + monkeypatch.setattr( + exports_mod, + "load_exported_data", + lambda *args: {"K": np.eye(3), "width": 1280, "height": 720}, + ) + monkeypatch.setattr(exports_mod, "compute_uv_distortion_metrics", lambda data: {"ok": True}) + monkeypatch.setattr(exports_mod, "recommend_edge_expand", lambda _: 1.2) + + geometry = resolve_finger_geometry( + export_dir=tmp_path, + config="dd02-ov", + ) + + assert geometry.out_width == 280 + assert geometry.out_height == 300 + assert geometry.source_scale == 0.4 + assert geometry.source_width == 512 + assert geometry.source_height == 288 + assert Path(geometry.map_cache) == (tmp_path / "unroll_map_uv_s0.4.npz").resolve() + + +def test_default_map_cache_rehomes_to_resolved_export_dir(tmp_path, monkeypatch): + _touch_bundle(tmp_path) + + import orisys.finger.exports as exports_mod + import orisys.finger.runtime as runtime_mod + + monkeypatch.setattr( + runtime_mod, + "load_config", + lambda _: {"camera": {"resolution_stitch": [280, 300], "source_scale": 1.0}}, + ) + monkeypatch.setattr( + exports_mod, + "load_exported_data", + lambda *args: {"K": np.eye(3), "width": 1280, "height": 720}, + ) + monkeypatch.setattr(exports_mod, "compute_uv_distortion_metrics", lambda data: {"ok": True}) + monkeypatch.setattr(exports_mod, "recommend_edge_expand", lambda _: 1.2) + + geometry = resolve_finger_geometry( + export_dir=tmp_path, + config="dd02-ov", + map_cache=DEFAULT_FINGER_MAP_CACHE, + ) + + assert Path(geometry.map_cache) == (tmp_path / "unroll_map_uv.npz").resolve() + + +def test_orisys_root_exports_finger_module(): + assert hasattr(orisys, "finger") + + +def test_tactile_processor_push_frame_uses_default_valid_mask(): + class DummySensor: + info = object() + + def __init__(self): + self.calls = [] + self.frame = None + self.img = None + self.flow = None + self.vnormal = None + self.vshear = None + self.depth_map = None + self.mask_stats = {} + self.valid_mask = None + + def get_img(self, **kwargs): + self.calls.append(kwargs) + self.frame = kwargs["frame"] + self.valid_mask = kwargs["valid_mask"] + return "ok" + + sensor = DummySensor() + default_mask = np.array([[True, False], [False, True]]) + tactile = TactileProcessor(sensor, default_valid_mask=default_mask) + frame = np.zeros((2, 2), dtype=np.uint8) + + result = tactile.push_frame(frame, already_gray=True) + + assert result == "ok" + assert sensor.calls[0]["frame"] is frame + assert sensor.calls[0]["already_processed"] is True + assert sensor.calls[0]["already_gray"] is True + assert sensor.calls[0]["valid_mask"] is default_mask + + +def test_finger_runtime_get_img_returns_none_on_capture_failure(): + from orisys.finger.runtime import FingerRuntime + + runtime = object.__new__(FingerRuntime) + runtime.grab_and_process = lambda *, color=False: (False, None) + assert runtime.get_img() is None + + +def test_finger_runtime_get_img_returns_unrolled_frame_on_success(): + from orisys.finger.runtime import FingerRuntime + + frame = np.zeros((2, 2), dtype=np.uint8) + runtime = object.__new__(FingerRuntime) + runtime.grab_and_process = lambda *, color=False: (True, frame) + assert runtime.get_img(color=True) is frame + + +@pytest.mark.parametrize( + "video,expected", + [ + ("0", 0), + (" 1 ", 1), + (r".\data\dd02\0607_1.mp4" + "\\", r".\data\dd02\0607_1.mp4"), + (0, 0), + ("./clip.mp4", "./clip.mp4"), + ], +) +def test_normalize_video_source(video, expected): + from orisys.finger.runtime import normalize_video_source + + assert normalize_video_source(video) == expected diff --git a/tests/test_masked_poisson_capfft.py b/tests/test_masked_poisson_capfft.py index 11a3976..4c96b7f 100644 --- a/tests/test_masked_poisson_capfft.py +++ b/tests/test_masked_poisson_capfft.py @@ -1,7 +1,9 @@ from pathlib import Path import numpy as np +import pytest +from orisys.backend_availability import CUPY_AVAILABLE from orisys.hhd import PoissonSolver, nHHD @@ -124,6 +126,46 @@ def test_capfft_hhd_forces_track_sparse_reference(): assert np.allclose(vshear[~mask], 0.0, atol=1e-5) +def test_sparse_hhd_normal_sign_tracks_capfft_on_masked_uv_flow(): + shape = (36, 44) + mask = _circle_mask(shape) + flow = _radial_flow(shape, scale=0.03) + flow[~mask] = 0.0 + + sparse_hhd = nHHD( + cuda=False, + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cpu", + masked_poisson_solver="sparse", + ) + capfft_hhd = nHHD( + cuda=False, + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cpu", + masked_poisson_solver="capfft", + masked_poisson_max_boundary=40, + masked_poisson_max_iters=3, + ) + + sparse_hhd.get_force(flow, valid_mask=mask) + capfft_hhd.get_force(flow, valid_mask=mask) + + sparse_vnormal = sparse_hhd.get_normal() + capfft_vnormal = capfft_hhd.get_normal() + + # The masked sparse solver should use the same sign convention as capfft/FFT. + sparse_flat = sparse_vnormal[mask].reshape(-1, 2) + capfft_flat = capfft_vnormal[mask].reshape(-1, 2) + dot = float(np.sum(sparse_flat * capfft_flat)) + assert dot > 0.0 + + denom = max(1.0, float(np.linalg.norm(capfft_flat))) + rel = float(np.linalg.norm(sparse_flat - capfft_flat) / denom) + assert rel < 0.35 + + def test_capfft_uv_mask_fixture_improves_over_raw_fft(): uv_path = REPO_ROOT / "unroll_map_uv.npz" if not uv_path.exists(): @@ -160,3 +202,91 @@ def test_capfft_uv_mask_fixture_improves_over_raw_fft(): err_cap = float(np.linalg.norm((capfft - sparse)[mask])) err_raw = float(np.linalg.norm((raw_fft - sparse)[mask])) assert err_cap < err_raw + + +@pytest.mark.skipif(not CUPY_AVAILABLE, reason="CuPy not available") +def test_capfft_solver_cupy_backend_uses_gpu_bulk_solve_and_tracks_cpu(): + shape = (32, 40) + mask = _circle_mask(shape) + rhs = _smooth_rhs(shape) + rhs_masked = np.where(mask, rhs, 0).astype(np.float32) + + cpu_solver = PoissonSolver( + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cpu", + masked_poisson_solver="capfft", + masked_poisson_max_boundary=40, + masked_poisson_max_iters=3, + ) + cupy_solver = PoissonSolver( + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cupy", + masked_poisson_solver="capfft", + masked_poisson_max_boundary=40, + masked_poisson_max_iters=3, + ) + + cpu_capfft = cpu_solver.solve(rhs_masked, mask=mask) + raw_fft = cpu_solver.solve(rhs_masked, mask=None) + + original_gpu_solve = cupy_solver._rect_solver._solve_gpu_internal + calls = {"count": 0} + + def counted_gpu_solve(f_gpu): + calls["count"] += 1 + return original_gpu_solve(f_gpu) + + cupy_solver._rect_solver._solve_gpu_internal = counted_gpu_solve + cupy_capfft = cupy_solver.solve(rhs_masked, mask=mask) + + assert calls["count"] > 0 + assert np.allclose(cupy_capfft[~mask], 0.0, atol=1e-6) + + err_gpu = float(np.linalg.norm((cupy_capfft - cpu_capfft)[mask])) + err_raw = float(np.linalg.norm((raw_fft - cpu_capfft)[mask])) + assert err_gpu < err_raw + + denom = max(1.0, float(np.linalg.norm(cpu_capfft[mask]))) + rel = err_gpu / denom + assert rel < 1e-2 + + +@pytest.mark.skipif(not CUPY_AVAILABLE, reason="CuPy not available") +def test_capfft_hhd_cupy_backend_tracks_cpu_capfft_reference(): + shape = (36, 44) + mask = _circle_mask(shape) + flow = _radial_flow(shape, scale=0.03) + flow[~mask] = 0.0 + + cpu_hhd = nHHD( + cuda=False, + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cpu", + masked_poisson_solver="capfft", + masked_poisson_max_boundary=40, + masked_poisson_max_iters=3, + ) + cupy_hhd = nHHD( + cuda=False, + grid=shape, + spacings=(1.0, 1.0), + poisson_backend="cupy", + masked_poisson_solver="capfft", + masked_poisson_max_boundary=40, + masked_poisson_max_iters=3, + ) + + _, cpu_forces = cpu_hhd.get_force(flow, valid_mask=mask) + _, cupy_forces = cupy_hhd.get_force(flow, valid_mask=mask) + + denom = max(1.0, float(np.linalg.norm(cpu_forces))) + rel = float(np.linalg.norm(cupy_forces - cpu_forces) / denom) + assert rel < 1e-2 + + vnormal = cupy_hhd.get_normal() + vshear = cupy_hhd.get_shear() + assert np.allclose(vnormal[~mask], 0.0, atol=1e-5) + assert np.allclose(vshear[~mask], 0.0, atol=1e-5) diff --git a/tests/test_pyproject.py b/tests/test_pyproject.py new file mode 100644 index 0000000..f4533f7 --- /dev/null +++ b/tests/test_pyproject.py @@ -0,0 +1,82 @@ +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +@pytest.fixture +def packaging_utils(): + import sys + + build_dir = REPO_ROOT / "build" + if str(build_dir) not in sys.path: + sys.path.insert(0, str(build_dir)) + from pyproject_utils import extras_require, install_requires, project_metadata + + return install_requires, extras_require, project_metadata + + +def test_pyproject_dependencies(packaging_utils): + install_requires, extras_require, project_metadata = packaging_utils + metadata = project_metadata(REPO_ROOT) + + assert metadata["name"] == "orisys" + assert "numpy>=" in install_requires(REPO_ROOT)[0] + assert "cuda12" in extras_require(REPO_ROOT) + assert extras_require(REPO_ROOT)["cuda"] == extras_require(REPO_ROOT)["cuda12"] + + +def test_pyproject_dynamic_version(packaging_utils): + import sys + + build_dir = REPO_ROOT / "build" + if str(build_dir) not in sys.path: + sys.path.insert(0, str(build_dir)) + from pyproject_utils import get_version + + assert get_version(REPO_ROOT) == "0.4.0" + + +def test_pyproject_bundle_manifest(packaging_utils): + import sys + + build_dir = REPO_ROOT / "build" + if str(build_dir) not in sys.path: + sys.path.insert(0, str(build_dir)) + from pyproject_utils import bundle_data_files, bundle_items + + items = bundle_items(REPO_ROOT) + names = {item["name"] for item in items} + assert names == {"configs", "examples", "docs", "assets"} + + data_files = dict(bundle_data_files(REPO_ROOT)) + assert "orisys/configs" in data_files + config_files = data_files["orisys/configs"] + assert config_files + assert all("/" in path and "*" not in path for path in config_files) + assert any(path.endswith(".json") for path in config_files) + + asset_dirs = [dest for dest in data_files if dest.startswith("orisys/assets/")] + assert asset_dirs + assert all("*" not in path for files in data_files.values() for path in files) + assert "examples" not in data_files + + +def test_bundle_explicit_files(): + import sys + + build_dir = REPO_ROOT / "build" + if str(build_dir) not in sys.path: + sys.path.insert(0, str(build_dir)) + from wheel_layout import data_file_entries_for_item, resolve_bundle_source_files + + item = { + "name": "configs", + "source": "config", + "target": "configs", + "files": ["dd02-ov.json"], + } + paths = resolve_bundle_source_files(REPO_ROOT, item) + assert [p.name for p in paths] == ["dd02-ov.json"] + assert data_file_entries_for_item(item) == ["config/dd02-ov.json"] diff --git a/tests/test_uv_geometry.py b/tests/test_uv_geometry.py new file mode 100644 index 0000000..e6e90b0 --- /dev/null +++ b/tests/test_uv_geometry.py @@ -0,0 +1,34 @@ +import numpy as np +import pytest + +from orisys.unroll.geometry import ( + resolve_uv_runtime_sizes, + scale_camera_geometry, + scaled_map_cache_path, +) + + +def test_scale_camera_geometry_halves_intrinsics_and_size(): + k = np.array([[1000.0, 0.0, 640.0], [0.0, 1000.0, 360.0], [0.0, 0.0, 1.0]]) + k_scaled, w, h = scale_camera_geometry(k, 1280, 720, 0.5) + assert w == 640 + assert h == 360 + assert k_scaled[0, 0] == pytest.approx(500.0) + assert k_scaled[1, 1] == pytest.approx(500.0) + assert k_scaled[0, 2] == pytest.approx(320.0) + assert k_scaled[1, 2] == pytest.approx(180.0) + + +def test_resolve_uv_runtime_sizes_from_work4(): + out_w, out_h, scale = resolve_uv_runtime_sizes( + { + "resolution_stitch": [280, 300], + "work4": [280, 300, 4, 0.4], + } + ) + assert (out_w, out_h, scale) == (280, 300, 0.4) + + +def test_scaled_map_cache_path_suffix(): + assert scaled_map_cache_path("unroll_map_uv.npz", 0.4) == "unroll_map_uv_s0.4.npz" + assert scaled_map_cache_path("unroll_map_uv.npz", 1.0) == "unroll_map_uv.npz" diff --git a/tests/test_uv_mesh_viewer.py b/tests/test_uv_mesh_viewer.py new file mode 100644 index 0000000..8aa26bf --- /dev/null +++ b/tests/test_uv_mesh_viewer.py @@ -0,0 +1,80 @@ +import numpy as np +import pytest + +from orisys.finger.mesh import ( + build_uv_surface_mesh, + compute_uv_surface_frames, + flow_magnitude_vertex_colors, + lift_uv_flow_to_3d, + mesh_uv_to_world_scale, + scalar_map_vertex_colors, +) + + +def test_build_uv_surface_mesh_from_grid(): + h, w = 4, 5 + yy, xx = np.mgrid[0:h, 0:w] + world_xyz = np.stack([xx, yy, np.zeros((h, w))], axis=2).astype(np.float64) + valid_mask = np.ones((h, w), dtype=bool) + valid_mask[0, 0] = False + + mesh = build_uv_surface_mesh(world_xyz, valid_mask) + assert mesh.rest_vertices.shape[0] == int(valid_mask.sum()) + assert mesh.triangles.shape[1] == 3 + assert len(mesh.triangles) > 0 + assert mesh.base_colors.shape == (mesh.rest_vertices.shape[0], 3) + assert np.allclose(mesh.base_colors[0], mesh.base_colors[-1]) + + +def test_lift_uv_flow_to_3d_tangent_only(): + flow = np.zeros((3, 3, 2), dtype=np.float64) + flow[1, 1] = (1.0, 0.0) + world_xyz = np.zeros((3, 3, 3), dtype=np.float64) + world_xyz[:, :, 0] = np.arange(3)[None, :] + world_xyz[:, :, 1] = np.arange(3)[:, None] + valid = np.ones((3, 3), dtype=bool) + tx, ty, normals = compute_uv_surface_frames(world_xyz, valid) + lifted = lift_uv_flow_to_3d(flow, tx, ty, normals) + assert lifted.shape == (3, 3, 3) + assert np.linalg.norm(lifted[1, 1]) > 0.0 + + +def test_flow_magnitude_vertex_colors_respects_threshold(): + h, w = 3, 3 + world_xyz = np.zeros((h, w, 3), dtype=np.float64) + valid_mask = np.ones((h, w), dtype=bool) + mesh = build_uv_surface_mesh(world_xyz, valid_mask) + + flow = np.zeros((h, w, 2), dtype=np.float64) + flow[1, 1] = (0.1, 0.0) # mag = 2 * 0.1 = 0.2 + flow[2, 2] = (1.0, 0.0) # mag = 2.0 + + colors = flow_magnitude_vertex_colors(flow, mesh, threshold=0.4) + low_idx = np.where((mesh.vertex_ys == 1) & (mesh.vertex_xs == 1))[0][0] + high_idx = np.where((mesh.vertex_ys == 2) & (mesh.vertex_xs == 2))[0][0] + assert np.allclose(colors[low_idx], mesh.base_colors[low_idx]) + assert not np.allclose(colors[high_idx], mesh.base_colors[high_idx]) + + +def test_scalar_map_vertex_colors_respects_threshold(): + h, w = 3, 3 + world_xyz = np.zeros((h, w, 3), dtype=np.float64) + valid_mask = np.ones((h, w), dtype=bool) + mesh = build_uv_surface_mesh(world_xyz, valid_mask) + + depth_map = np.zeros((h, w), dtype=np.float64) + depth_map[1, 1] = 0.05 + depth_map[2, 2] = 1.0 + + colors = scalar_map_vertex_colors(depth_map, mesh, threshold=0.1) + low_idx = np.where((mesh.vertex_ys == 1) & (mesh.vertex_xs == 1))[0][0] + high_idx = np.where((mesh.vertex_ys == 2) & (mesh.vertex_xs == 2))[0][0] + assert np.allclose(colors[low_idx], mesh.base_colors[low_idx]) + assert not np.allclose(colors[high_idx], mesh.base_colors[high_idx]) + + +def test_mesh_uv_to_world_scale_positive(): + world_xyz = np.random.randn(8, 8, 3).astype(np.float64) + valid = np.ones((8, 8), dtype=bool) + scale = mesh_uv_to_world_scale(world_xyz, valid) + assert scale > 0.0 diff --git a/tools/3d_patch_generation_blender/dd02c_0607.blend b/tools/3d_patch_generation_blender/dd02c_0607.blend index 5e97939..a16183a 100644 Binary files a/tools/3d_patch_generation_blender/dd02c_0607.blend and b/tools/3d_patch_generation_blender/dd02c_0607.blend differ diff --git a/tools/3d_patch_generation_blender/dd02c_0607.blend1 b/tools/3d_patch_generation_blender/dd02c_0607.blend1 index e8e70e4..5e97939 100644 Binary files a/tools/3d_patch_generation_blender/dd02c_0607.blend1 and b/tools/3d_patch_generation_blender/dd02c_0607.blend1 differ diff --git a/tools/dd02_3dviewer/sdk_mapto3d.py b/tools/dd02_3dviewer/sdk_mapto3d.py index 068a6c8..31cd6d7 100644 --- a/tools/dd02_3dviewer/sdk_mapto3d.py +++ b/tools/dd02_3dviewer/sdk_mapto3d.py @@ -6,6 +6,11 @@ import time import cv2 import numpy as np +# UV-unroll mesh viewer (grey-white surface, UV-space field mapping): +# from examples.viz.mesh_viewer import UVMeshViewer, build_uv_surface_mesh +# # or: from orisys.finger import UVMeshViewer, build_uv_surface_mesh +# See examples/finger_pipeline_3dview.py + SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) REPO_ROOT = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "..")) diff --git a/tools/dd02_3dviewer/uv_mesh_viewer.py b/tools/dd02_3dviewer/uv_mesh_viewer.py new file mode 100644 index 0000000..8ca871a --- /dev/null +++ b/tools/dd02_3dviewer/uv_mesh_viewer.py @@ -0,0 +1,4 @@ +"""Backward-compatible re-exports; prefer ``examples/viz/mesh_viewer.py`` or ``orisys.finger``.""" + +from orisys.finger.mesh import * # noqa: F403 +from orisys.finger.viewer import * # noqa: F403