split example visualizaiton pipeline

This commit is contained in:
ykcheungab
2026-06-18 13:47:16 +08:00
parent fbee1c37ed
commit f19f092b64
42 changed files with 3067 additions and 702 deletions
+1 -1
View File
@@ -9,7 +9,7 @@ __pycache__/
/dist_obfuscated/
/dist_pyd/
/dist_single_pyd/
/dist_nuitka/
/nuitka_wheel/
/build_nuitka/
/release/
/SDK/
+1 -1
View File
@@ -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
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+2 -2
View File
@@ -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
+22 -31
View File
@@ -1,50 +1,41 @@
# Examples
This folder contains user-facing example scripts for the Orisys SDK.
User-facing demo scripts for the Orisys SDK. Run from the repository root.
## Files
## Layout
- `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
- `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
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
python examples/finger_plot_data.py -v 0 -c ./config/dd02-ov.json
```
## Requirements
- Python environment with the SDK dependencies installed
- a supported camera or test video
- valid JSON config in `./config/`
- SDK installed (`pip install -e .`)
- camera or test video
- JSON config under `./config/`
Some examples also require:
Plot examples additionally require:
- `pyqtgraph`
- `PyQt5`
- multiple cameras for multi-sensor demos
```bash
pip install pyqt5 pyqtgraph
```
## 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.
- Press `q` to quit OpenCV windows where applicable.
- Plot examples: `s` starts recording, `e` stops recording.
+40 -161
View File
@@ -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)
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:
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}"
)
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,
try:
video = args.camera_id
if args.video:
video = args.video
runtime = FingerRuntime.open(
config=args.config,
video=video,
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)
except RuntimeError as exc:
print(str(exc), 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,
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"output={geometry.out_width}x{geometry.out_height} "
)
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()
+189
View File
@@ -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()
+101
View File
@@ -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()
+52 -214
View File
@@ -6,137 +6,35 @@
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
_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, draw_magnitude_map, lowpass_filter
def open_sensor(video: str, config: str, cal: str, verbose: bool):
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))
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
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="配置文件名称或路径")
@@ -144,114 +42,54 @@ def main():
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' 结束录像。")
try:
while True:
sensor.get_img() # 获取并拼接复眼图像
sensor.get_img()
recorder.write_if_recording(sensor.frame)
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)
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)
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}")
# 显示形变矢量场
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)
cv2.imshow("流场幅值图", draw_magnitude_map(flow))
# 更新实时曲线
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) # 更新曲线
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)
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
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("当前没有可用帧,无法确定录像尺寸。")
recorder.start(sensor.frame)
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() # 断开传感器连接并释放资源
recorder.stop()
finally:
sensor.disconnect()
if __name__ == '__main__':
if __name__ == "__main__":
main()
+20
View File
@@ -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",
]
+29
View File
@@ -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)
+69
View File
@@ -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()))
+4
View File
@@ -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
+57
View File
@@ -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
+71
View File
@@ -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
+27
View File
@@ -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)
+69
View File
@@ -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
-228
View File
@@ -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()
+858
View File
@@ -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()
+10 -29
View File
@@ -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))
+1 -1
View File
@@ -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]
+4
View File
@@ -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'
+59
View File
@@ -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",
]
+100
View File
@@ -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),
)
+271
View File
@@ -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)
+328
View File
@@ -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()
+130
View File
@@ -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
+6 -3
View File
@@ -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)
+180
View File
@@ -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
+130
View File
@@ -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)
+82
View File
@@ -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"]
+34
View File
@@ -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"
+80
View File
@@ -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
Binary file not shown.
Binary file not shown.
+5
View File
@@ -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, "..", ".."))
+4
View File
@@ -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