Compare commits

...

6 Commits

Author SHA1 Message Date
zhangyijian a387b8b693 Merge pull request 'Feature/force cal' (#17) from feature/force_cal into dev
Reviewed-on: #17
2026-08-26 20:22:27 +08:00
ykcheungab 8a4be5f8f2 update to 0.5.0 2026-08-26 20:20:37 +08:00
ykcheungab 68910e63c1 fix finger gui cannot choose camera 2026-08-26 20:19:55 +08:00
ykcheungab 577e3fe528 update pip install .copy manifest for developments 2026-08-26 20:04:40 +08:00
ykcheungab 32d4164471 update examples to fix path issues 2026-08-26 19:38:47 +08:00
ykcheungab d59e1cb7fb fix path errors 2026-08-26 19:22:37 +08:00
34 changed files with 1056 additions and 307 deletions
+8
View File
@@ -0,0 +1,8 @@
# Non-Python files needed for sdist -> wheel builds from an unpacked sdist.
include pyproject.toml
include setup.py
include README.md
recursive-include config *.json
recursive-include sensors *
recursive-include build/common *.py
recursive-include build/wheel *.py
+36 -1
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import ast import ast
import importlib.util import importlib.util
import shutil
import subprocess import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
@@ -102,7 +103,11 @@ def bundle_items(repo_root: Path) -> list[dict]:
def bundle_data_files(repo_root: Path) -> list[tuple[str, list[str]]]: def bundle_data_files(repo_root: Path) -> list[tuple[str, list[str]]]:
"""Map bundle items marked dev_install=true to setuptools data_files entries.""" """Map bundle items marked ``dev_install=true`` to install destinations.
Kept for tests / tooling. Prefer :func:`stage_dev_bundle_into` for pip
installs so files land inside ``site-packages/orisys/``.
"""
resolve_bundle_source_files = _load_wheel_layout_module().resolve_bundle_source_files resolve_bundle_source_files = _load_wheel_layout_module().resolve_bundle_source_files
grouped: dict[str, list[str]] = {} grouped: dict[str, list[str]] = {}
@@ -123,3 +128,33 @@ def bundle_data_files(repo_root: Path) -> list[tuple[str, list[str]]]:
grouped.setdefault(dest, []).append(rel_src) grouped.setdefault(dest, []).append(rel_src)
return [(dest, sorted(files)) for dest, files in sorted(grouped.items())] return [(dest, sorted(files)) for dest, files in sorted(grouped.items())]
def stage_dev_bundle_into(repo_root: Path, pkg_dir: Path) -> list[str]:
"""Copy ``dev_install`` bundle files under ``pkg_dir`` (the orisys package).
Returns package-relative paths that were written (e.g. ``sensors/box/...``).
"""
resolve_bundle_source_files = _load_wheel_layout_module().resolve_bundle_source_files
packaged: list[str] = []
for item in bundle_items(repo_root):
if not item.get("dev_install", False):
continue
src_dir = repo_root / item["source"]
dst_dir = pkg_dir / item["target"]
dst_dir.mkdir(parents=True, exist_ok=True)
for src_file in resolve_bundle_source_files(repo_root, item):
if item.get("preserve_tree", False):
rel = src_file.relative_to(src_dir)
dst_file = dst_dir / rel
else:
rel = Path(src_file.name)
dst_file = dst_dir / src_file.name
dst_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src_file, dst_file)
packaged.append(f"{item['target']}/{rel.as_posix()}")
return packaged
+4
View File
@@ -17,6 +17,8 @@ from orisys.finger.mesh import UVSurfaceMesh
from orisys.finger.viewer import init_uv_mesh_viewer from orisys.finger.viewer import init_uv_mesh_viewer
from orisys.hhd import nHHD from orisys.hhd import nHHD
from orisys.tracker import Tracker from orisys.tracker import Tracker
from orisys import capture
from orisys.Stitcher import Stitcher
assert orisys.__version__ assert orisys.__version__
assert Sensor is not None assert Sensor is not None
@@ -31,6 +33,8 @@ assert UVSurfaceMesh is not None
assert init_uv_mesh_viewer is not None assert init_uv_mesh_viewer is not None
assert nHHD is not None assert nHHD is not None
assert Tracker is not None assert Tracker is not None
assert capture is not None
assert Stitcher is not None
print(f"orisys {orisys.__version__} public API imports OK") print(f"orisys {orisys.__version__} public API imports OK")
""" """
+2 -2
View File
@@ -43,7 +43,7 @@ EXCLUDE_MODULES = [
DATA_DIRS = [ DATA_DIRS = [
("config", "config"), ("config", "config"),
("assets", "assets"), ("sensors", "sensors"),
] ]
README_LINES = [ README_LINES = [
@@ -56,7 +56,7 @@ README_LINES = [
"", "",
"Bundled folders:", "Bundled folders:",
" config/ sensor JSON configs", " config/ sensor JSON configs",
" assets/ finger mesh and visualization data", " sensors/ per-sensor bundles (finger mesh and visualization data)",
"", "",
"CPU-only build (no CUDA/CuPy).", "CPU-only build (no CUDA/CuPy).",
] ]
+1 -1
View File
@@ -114,7 +114,7 @@ def build(*, target: str = "internal", verify: bool | None = None) -> Path:
print(f"Copied {copied_counts['configs']} JSON config files") 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['examples']} example files to orisys/examples/")
print(f"Copied {copied_counts['docs']} doc files to orisys/docs/") print(f"Copied {copied_counts['docs']} doc files to orisys/docs/")
print(f"Copied {copied_counts['assets']} finger asset files to orisys/assets/") print(f"Copied {copied_counts['sensors']} sensor bundle files to orisys/sensors/")
write_wheel_setup( write_wheel_setup(
NUITKA_WHEEL_DIR, NUITKA_WHEEL_DIR,
+3
View File
@@ -16,7 +16,10 @@ from common.paths import repo_path
NUITKA_INCLUDE_PACKAGES = ( NUITKA_INCLUDE_PACKAGES = (
"orisys", "orisys",
"orisys.capture",
"orisys.finger", "orisys.finger",
"orisys.stitch",
"orisys.stitch.backends",
"orisys.unroll", "orisys.unroll",
"orisys.hhd", "orisys.hhd",
"orisys.hhd.solvers", "orisys.hhd.solvers",
+12 -1
View File
@@ -6,12 +6,19 @@ import shutil
from pathlib import Path from pathlib import Path
# Required files that must ship with the wheel for ORY-FINGER runtime/GUI. # Required files that must ship with the wheel for ORY-FINGER runtime/GUI.
# Accept both the flat layout and a nested assets/finger/ checkout tree.
REQUIRED_ASSET_RELPATHS = ( REQUIRED_ASSET_RELPATHS = (
"finger/assets/30fps-0703/mesh.npz", "finger/assets/30fps-0703/mesh.npz",
"finger/assets/30fps-0703/camera_intrinsics.npz", "finger/assets/30fps-0703/camera_intrinsics.npz",
"finger/assets/30fps-0703/world_to_cam.npy", "finger/assets/30fps-0703/world_to_cam.npy",
"finger/assets/visualization/world_to_cam.npy", "finger/assets/visualization/world_to_cam.npy",
) )
REQUIRED_ASSET_RELPATH_ALTERNATES = {
"finger/assets/30fps-0703/mesh.npz": "finger/assets/finger/30fps-0703/mesh.npz",
"finger/assets/30fps-0703/camera_intrinsics.npz": "finger/assets/finger/30fps-0703/camera_intrinsics.npz",
"finger/assets/30fps-0703/world_to_cam.npy": "finger/assets/finger/30fps-0703/world_to_cam.npy",
"finger/assets/visualization/world_to_cam.npy": "finger/assets/finger/visualization/world_to_cam.npy",
}
def resolve_bundle_source_files(repo_root: Path, item: dict) -> list[Path]: def resolve_bundle_source_files(repo_root: Path, item: dict) -> list[Path]:
@@ -61,7 +68,11 @@ def verify_required_assets(repo_root: Path, items: list[dict]) -> None:
continue continue
resolved.add(rel) resolved.add(rel)
missing = [rel for rel in REQUIRED_ASSET_RELPATHS if rel not in resolved] missing = []
for rel in REQUIRED_ASSET_RELPATHS:
alt = REQUIRED_ASSET_RELPATH_ALTERNATES.get(rel)
if rel not in resolved and (alt is None or alt not in resolved):
missing.append(rel)
if missing: if missing:
joined = ", ".join(missing) joined = ", ".join(missing)
raise FileNotFoundError( raise FileNotFoundError(
+5 -5
View File
@@ -5,7 +5,7 @@ description: 安装并运行 Orisys SDK 的基础步骤
# 快速开始 # 快速开始
版本:0.4.3 版本:0.5.0
## 简介 ## 简介
本文件介绍如何上手和运行叠动视触觉传感器配套的开发套件。本套件是配合叠动视触觉传感器进行图像处理,计算形变、力以及接触信息。它可以从摄像头或视频文件中读取图像,处理后打印和展示处理后信息。 本文件介绍如何上手和运行叠动视触觉传感器配套的开发套件。本套件是配合叠动视触觉传感器进行图像处理,计算形变、力以及接触信息。它可以从摄像头或视频文件中读取图像,处理后打印和展示处理后信息。
@@ -84,9 +84,9 @@ pip install --upgrade orisys
### 使用 .whl 文件安装(离线或指定版本) ### 使用 .whl 文件安装(离线或指定版本)
从官方渠道获取 .whl 文件,文件名格式通常为: 从官方渠道获取 .whl 文件,文件名格式通常为:
orisys-{版本}-{Python版本}-{Python版本}-{操作系统}_{硬件架构}.whl orisys-{版本}-{Python版本}-{Python版本}-{操作系统}_{硬件架构}.whl
例如:orisys-0.4.3-cp310-cp310-win_amd64.whl 表示: 例如:orisys-0.5.0-cp310-cp310-win_amd64.whl 表示:
版本:0.4.3 版本:0.5.0
Python 版本:3.10(cp310) Python 版本:3.10(cp310)
操作系统:Windows(win) 操作系统:Windows(win)
硬件架构:64位(amd64) 硬件架构:64位(amd64)
@@ -97,7 +97,7 @@ Python 版本:3.10(cp310)
```bash ```bash
# 请将文件名替换为您实际下载的文件名 # 请将文件名替换为您实际下载的文件名
pip install orisys-0.4.3-cp310-cp310-win_amd64.whl pip install orisys-0.5.0-cp310-cp310-win_amd64.whl
``` ```
### 安装 CuPy(可选) ### 安装 CuPy(可选)
@@ -139,7 +139,7 @@ python
import orisys import orisys
print(orisys.__version__) print(orisys.__version__)
``` ```
如果输出版本号(例如 0.4.3),则说明安装成功。 如果输出版本号(例如 0.5.0),则说明安装成功。
## 使用方法 ## 使用方法
### 运行示例程序 ### 运行示例程序
+28 -7
View File
@@ -7,7 +7,7 @@ ORY-BOX:单传感器基础流程
3. 按需计算接触区域 3. 按需计算接触区域
4. 读取并可视化结果 4. 读取并可视化结果
SDK version: 0.4.3 SDK version: 0.5.0
""" """
import os import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
@@ -49,8 +49,23 @@ def main():
sensor.compute_contact() sensor.compute_contact()
# 步骤 2.4:读取结果 # 步骤 2.4:读取结果(FNORMAL/FSHEAR* 在有 sensors/box/force_calibration.json 时为标定后牛顿力;
fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info( # *_RAW 始终为原始积分量)
(
fps,
flow,
vnormal,
img,
contour,
centroid,
depth_map,
fnormal,
fshearx,
fsheary,
fnormal_raw,
fshearx_raw,
fsheary_raw,
) = sensor.read_info(
sensor.info.FPS, sensor.info.FPS,
sensor.info.VRAW, sensor.info.VRAW,
sensor.info.VNORMAL, sensor.info.VNORMAL,
@@ -60,7 +75,10 @@ def main():
sensor.info.DEPTH, sensor.info.DEPTH,
sensor.info.FNORMAL, sensor.info.FNORMAL,
sensor.info.FSHEARX, sensor.info.FSHEARX,
sensor.info.FSHEARY sensor.info.FSHEARY,
sensor.info.FNORMAL_RAW,
sensor.info.FSHEARX_RAW,
sensor.info.FSHEARY_RAW,
) )
# 步骤 3:可视化显示 # 步骤 3:可视化显示
@@ -89,10 +107,13 @@ def main():
# 应用伪彩色映射 # 应用伪彩色映射
cv2.imshow("depth", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET)) cv2.imshow("depth", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET))
# 步骤 3.4:打印关键结果 # 步骤 3.4:打印关键结果(标定力 + 原始力)
print( print(
f"FPS={fps:.2f}, 法向力={fnormal:.4f}, 切向力X={fshearx:.4f}, " f"FPS={fps:.2f}, "
f"切向力Y={fsheary:.4f}, 深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}" f"法向力={fnormal:.4f} N (raw={fnormal_raw:.4f}), "
f"切向力X={fshearx:.4f} N (raw={fshearx_raw:.4f}), "
f"切向力Y={fsheary:.4f} N (raw={fsheary_raw:.4f}), "
f"深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}"
) )
# 步骤 3.5:键盘控制 # 步骤 3.5:键盘控制
+1
View File
@@ -90,6 +90,7 @@ def main():
threshold=0.4, threshold=0.4,
grid_spacing=10, grid_spacing=10,
arrow_scale=1.0, arrow_scale=1.0,
valid_mask=tactile.valid_mask,
) )
cv2.imshow("UV Unrolled", arrows) cv2.imshow("UV Unrolled", arrows)
+3 -4
View File
@@ -1,12 +1,11 @@
{ {
"video_source": "D:/repo/orisys-sdk/data/dd02-0703/1.mp4", "video_source": "D:/repo/orisys-sdk/data/dd02-0703/1.mp4",
"sensor_bundle_dir": "sensors/finger", "config_path": "finger",
"config_path": "sensors/finger/sensor.json",
"finger_assets_dir": "sensors/finger/assets/30fps-0703", "finger_assets_dir": "sensors/finger/assets/30fps-0703",
"csv_save_dir": "./outputs/csv", "csv_save_dir": "./outputs/csv",
"show_log": true, "show_log": true,
"show_flow_arrows": false, "show_flow_arrows": false,
"arrow_background": "pure_color", "arrow_background": "unrolled",
"free_rotate": true, "free_rotate": true,
"log_display_mode": "hidden" "sensor_bundle_dir": "sensors/finger"
} }
+9 -1
View File
@@ -11,7 +11,6 @@ from typing import Any, Optional
import cv2 import cv2
import numpy as np import numpy as np
import orisys import orisys
from orisys.signal_processing import calibrate_normal_force
from .roi_geometry import ( from .roi_geometry import (
build_roi_warp, build_roi_warp,
@@ -23,6 +22,10 @@ from .roi_geometry import (
) )
NORMAL_FORCE_CAL_SLOPE = 3.92685554668795e-05
NORMAL_FORCE_CAL_OFFSET = 0.0217596967278535
@dataclass @dataclass
class FrameSnapshot: class FrameSnapshot:
"""Normalized per-frame payload consumed by the Qt viewer.""" """Normalized per-frame payload consumed by the Qt viewer."""
@@ -46,6 +49,11 @@ class FrameSnapshot:
record_frame: Optional[np.ndarray] = None record_frame: Optional[np.ndarray] = None
def calibrate_normal_force(raw_value: float) -> float:
"""Convert raw normal-force reading to Newtons using y = mx + c."""
return NORMAL_FORCE_CAL_SLOPE * float(raw_value) + NORMAL_FORCE_CAL_OFFSET
class SensorAdapter(ABC): class SensorAdapter(ABC):
"""Common interface between Qt and concrete sensor backends.""" """Common interface between Qt and concrete sensor backends."""
+27 -1
View File
@@ -13,7 +13,7 @@ from typing import Literal, Optional, Sequence, Tuple
import cv2 import cv2
import numpy as np import numpy as np
import orisys import orisys
from orisys.signal_processing import lowpass_filter from scipy.signal import butter, filtfilt
FORCE_SAMPLE_RATE_HZ = 30.0 FORCE_SAMPLE_RATE_HZ = 30.0
PLOT_BUFFER_POINTS = 120 PLOT_BUFFER_POINTS = 120
@@ -61,6 +61,32 @@ class ForcePlotState:
y_ranges: Tuple[Tuple[float, float], Tuple[float, float], Tuple[float, float]] y_ranges: Tuple[Tuple[float, float], Tuple[float, float], Tuple[float, float]]
def lowpass_filter(
data,
cutoff_hz: float,
sample_rate_hz: float = FORCE_SAMPLE_RATE_HZ,
order: int = 2,
) -> np.ndarray:
"""Zero-phase Butterworth low-pass 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)
def ensure_bgr(img: Optional[np.ndarray]) -> Optional[np.ndarray]: def ensure_bgr(img: Optional[np.ndarray]) -> Optional[np.ndarray]:
if img is None: if img is None:
return None return None
+135 -18
View File
@@ -1,22 +1,139 @@
"""Discover USB cameras for the dashboard. """Discover USB cameras for the dashboard (Windows DSHOW/MSMF, Linux V4L2)."""
Thin compatibility shim over ``orisys.capture`` discovery helpers.
"""
from __future__ import annotations from __future__ import annotations
from orisys.capture import ( import contextlib
DEFAULT_MAX_CAMERA_INDEX, import os
CameraFilter, import sys
CameraInfo, from dataclasses import dataclass
list_cameras,
pick_default_camera,
)
__all__ = [ import cv2
"DEFAULT_MAX_CAMERA_INDEX",
"CameraFilter", DEFAULT_MAX_CAMERA_INDEX = 10
"CameraInfo",
"list_cameras",
"pick_default_camera", @dataclass(frozen=True)
] class CameraInfo:
index: int
backend_name: str
backend_api: int
width: int
height: int
@property
def source(self) -> str:
return str(self.index)
def label(self) -> str:
return f"{self.index} ({self.width}x{self.height}) [{self.backend_name}]"
@contextlib.contextmanager
def _suppress_opencv_noise():
"""Hide OpenCV VideoCapture probe warnings (stderr + OpenCV log)."""
prev_level = None
set_level = None
try:
logging = getattr(getattr(cv2, "utils", None), "logging", None)
if logging is not None:
set_level = getattr(logging, "setLogLevel", None)
get_level = getattr(logging, "getLogLevel", None)
silent = getattr(logging, "LOG_LEVEL_SILENT", 0)
if set_level is not None:
if get_level is not None:
prev_level = get_level()
set_level(silent)
except Exception:
prev_level = None
devnull = open(os.devnull, "w")
try:
with contextlib.redirect_stderr(devnull):
stderr_fd = sys.stderr.fileno()
saved_fd = os.dup(stderr_fd)
try:
os.dup2(devnull.fileno(), stderr_fd)
yield
finally:
os.dup2(saved_fd, stderr_fd)
os.close(saved_fd)
finally:
devnull.close()
if set_level is not None and prev_level is not None:
try:
set_level(prev_level)
except Exception:
pass
def _backend_candidates() -> list[tuple[str, int]]:
candidates: list[tuple[str, int]] = []
if sys.platform == "win32":
if hasattr(cv2, "CAP_DSHOW"):
candidates.append(("DSHOW", cv2.CAP_DSHOW))
if hasattr(cv2, "CAP_MSMF"):
candidates.append(("MSMF", cv2.CAP_MSMF))
elif sys.platform.startswith("linux"):
# Prefer v4l2 for USB camera indices on Linux.
if hasattr(cv2, "CAP_V4L2"):
candidates.append(("V4L2", cv2.CAP_V4L2))
if not candidates:
candidates.append(("ANY", cv2.CAP_ANY))
return candidates
def _try_open(index: int, backend_name: str, backend_api: int) -> CameraInfo | None:
cap = cv2.VideoCapture(index, backend_api)
if not cap.isOpened():
cap.release()
return None
ok, frame = cap.read()
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
if not ok or frame is None:
if width <= 0 or height <= 0:
cap.release()
return None
else:
h, w = frame.shape[:2]
width = width or int(w)
height = height or int(h)
cap.release()
return CameraInfo(
index=index,
backend_name=backend_name,
backend_api=backend_api,
width=width,
height=height,
)
def list_cameras(max_index: int = DEFAULT_MAX_CAMERA_INDEX) -> list[CameraInfo]:
"""Probe indices ``0 .. max_index`` and return working cameras."""
found: list[CameraInfo] = []
seen_indices: set[int] = set()
backends = _backend_candidates()
with _suppress_opencv_noise():
for index in range(0, max_index + 1):
for backend_name, backend_api in backends:
info = _try_open(index, backend_name, backend_api)
if info is None:
continue
if index in seen_indices:
break
found.append(info)
seen_indices.add(index)
break
return found
def pick_default_camera(cameras: list[CameraInfo], preferred: str | None = None) -> str | None:
if not cameras:
return None
preferred = (preferred or "").strip()
if preferred:
for cam in cameras:
if cam.source == preferred:
return cam.source
return cameras[0].source
+71 -120
View File
@@ -2,19 +2,14 @@
from __future__ import annotations from __future__ import annotations
import json
import math import math
import sys import sys
import time import time
from pathlib import Path
from PyQt5.QtCore import Qt, QTimer from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtGui import QIcon from PyQt5.QtGui import QIcon
from PyQt5.QtWidgets import ( from PyQt5.QtWidgets import (
QAction,
QActionGroup,
QApplication, QApplication,
QDockWidget,
QFileDialog, QFileDialog,
QMainWindow, QMainWindow,
QSplitter, QSplitter,
@@ -66,7 +61,6 @@ class FingerDashboardWindow(QMainWindow):
self._frame_index = 0 self._frame_index = 0
self._wall_fps = 0.0 self._wall_fps = 0.0
self._wall_fps_alpha = 0.15 self._wall_fps_alpha = 0.15
self._log_display_mode = "hidden"
self.setWindowTitle(layout.window_title) self.setWindowTitle(layout.window_title)
self.setStyleSheet(STYLE) self.setStyleSheet(STYLE)
@@ -90,7 +84,7 @@ class FingerDashboardWindow(QMainWindow):
root_layout.setSpacing(layout.root_spacing_px) root_layout.setSpacing(layout.root_spacing_px)
setup = layout.panel_setup setup = layout.panel_setup
self._log_panel_visible = False self._log_panel_visible = setup.log_panel.enabled
self.mesh_state = MeshViewState.from_layout(layout) self.mesh_state = MeshViewState.from_layout(layout)
self.mesh_viewport = MeshViewportWidget(layout=layout, mesh_state=self.mesh_state) self.mesh_viewport = MeshViewportWidget(layout=layout, mesh_state=self.mesh_state)
arrows_slot = layout.slot("arrows") arrows_slot = layout.slot("arrows")
@@ -117,6 +111,7 @@ class FingerDashboardWindow(QMainWindow):
self._preferences = load_preferences() self._preferences = load_preferences()
self.settings_panel.apply_preference_state(self._preferences) self.settings_panel.apply_preference_state(self._preferences)
self._log_panel_visible = bool(self.settings_panel.show_log_check.isChecked())
self.mesh_viewport.set_free_rotation(bool(self.settings_panel.free_rotate_check.isChecked())) self.mesh_viewport.set_free_rotation(bool(self.settings_panel.free_rotate_check.isChecked()))
if self.arrow_panel is not None: if self.arrow_panel is not None:
self.arrow_panel.set_background(self.settings_panel.arrow_background()) self.arrow_panel.set_background(self.settings_panel.arrow_background())
@@ -157,6 +152,11 @@ class FingerDashboardWindow(QMainWindow):
self.charts_column = QSplitter(Qt.Vertical) self.charts_column = QSplitter(Qt.Vertical)
self.charts_column.addWidget(self.force_charts.widget) self.charts_column.addWidget(self.force_charts.widget)
if setup.log_panel.enabled:
self.charts_column.addWidget(self.log_widget)
self.log_widget.setVisible(self._log_panel_visible)
self.charts_column.setStretchFactor(0, setup.charts_log_stretch)
self.charts_column.setStretchFactor(1, setup.log_stretch)
self.charts_column.setChildrenCollapsible(False) self.charts_column.setChildrenCollapsible(False)
h_splitter = QSplitter(Qt.Horizontal) h_splitter = QSplitter(Qt.Horizontal)
@@ -172,104 +172,18 @@ class FingerDashboardWindow(QMainWindow):
self.h_splitter = h_splitter self.h_splitter = h_splitter
root_layout.addWidget(h_splitter) root_layout.addWidget(h_splitter)
self._setup_log_dock()
self._apply_log_display_mode(self._preferences.get("log_display_mode", "hidden"))
self.resize(*self._target_window_size()) self.resize(*self._target_window_size())
QTimer.singleShot(0, self._apply_panel_layout) QTimer.singleShot(0, self._apply_panel_layout)
def _set_log_panel_visible(self, visible: bool) -> None: def _set_log_panel_visible(self, visible: bool) -> None:
self._log_panel_visible = bool(visible) self._log_panel_visible = bool(visible)
if hasattr(self, "log_dock"): self.log_widget.setVisible(self._log_panel_visible)
self.log_dock.setVisible(self._log_panel_visible)
QTimer.singleShot(0, self._apply_panel_layout) QTimer.singleShot(0, self._apply_panel_layout)
def _setup_log_dock(self) -> None:
self.log_dock = QDockWidget("Log", self)
self.log_dock.setObjectName("logDock")
self.log_dock.setWidget(self.log_widget)
self.log_dock.setAllowedAreas(Qt.BottomDockWidgetArea | Qt.RightDockWidgetArea)
self.addDockWidget(Qt.BottomDockWidgetArea, self.log_dock)
self.log_dock.hide()
self.log_dock.visibilityChanged.connect(self._on_log_dock_visibility_changed)
view_menu = self.menuBar().addMenu("View")
self.log_toggle_action = QAction("Log Panel", self)
self.log_toggle_action.setCheckable(True)
self.log_toggle_action.triggered.connect(self._on_log_toggle_action)
view_menu.addAction(self.log_toggle_action)
log_content_menu = view_menu.addMenu("Log Content")
self.log_content_actions = QActionGroup(self)
self.log_content_actions.setExclusive(False)
for label, mode in (
("Runtime log", "runtime"),
("Bundle info", "bundle"),
("Performance/profile", "profile"),
):
action = QAction(label, self)
action.setCheckable(True)
action.setData(mode)
action.triggered.connect(lambda _checked=False, m=mode: self._apply_log_display_mode(m, save=True))
self.log_content_actions.addAction(action)
log_content_menu.addAction(action)
def _on_log_toggle_action(self, checked: bool) -> None:
if checked:
mode = self._log_display_mode if self._log_display_mode != "hidden" else "runtime"
self._apply_log_display_mode(mode, save=True)
else:
self._apply_log_display_mode("hidden", save=True)
def _on_log_dock_visibility_changed(self, visible: bool) -> None:
self._log_panel_visible = bool(visible)
if hasattr(self, "log_toggle_action"):
self.log_toggle_action.blockSignals(True)
self.log_toggle_action.setChecked(visible)
self.log_toggle_action.blockSignals(False)
if not visible and self._log_display_mode != "hidden":
self._log_display_mode = "hidden"
if hasattr(self, "log_content_actions"):
for action in self.log_content_actions.actions():
action.blockSignals(True)
action.setChecked(False)
action.blockSignals(False)
self._save_preferences()
def _apply_log_display_mode(self, mode: str | None, *, save: bool = False) -> None:
mode = mode or "hidden"
self._log_display_mode = mode
if hasattr(self, "log_toggle_action"):
self.log_toggle_action.blockSignals(True)
self.log_toggle_action.setChecked(mode != "hidden")
self.log_toggle_action.blockSignals(False)
if hasattr(self, "log_content_actions"):
for action in self.log_content_actions.actions():
action.blockSignals(True)
action.setChecked(action.data() == mode)
action.blockSignals(False)
if mode == "hidden":
self._set_log_panel_visible(False)
if save:
self._save_preferences()
return
if mode == "bundle":
self.log_widget.setPlainText(
"Sensor bundle\n"
f"Config: {self.settings_panel.config_path()}\n"
f"Assets: {self.settings_panel.finger_export_dir()}\n"
)
elif mode == "profile":
self.log_widget.setPlainText(
"Performance/profile\n"
"Run with --profile to print per-stage timings in the runtime log.\n"
)
self._set_log_panel_visible(True)
if save:
self._save_preferences()
def _on_log_visibility_changed(self, visible: bool) -> None: def _on_log_visibility_changed(self, visible: bool) -> None:
self._apply_log_display_mode("runtime" if visible else "hidden", save=True) self._set_log_panel_visible(visible)
self._save_preferences()
def _on_free_rotation_changed(self, enabled: bool) -> None: def _on_free_rotation_changed(self, enabled: bool) -> None:
self.mesh_viewport.set_free_rotation(enabled) self.mesh_viewport.set_free_rotation(enabled)
@@ -282,7 +196,6 @@ class FingerDashboardWindow(QMainWindow):
def _save_preferences(self) -> None: def _save_preferences(self) -> None:
prefs = self.settings_panel.preference_state() prefs = self.settings_panel.preference_state()
prefs["log_display_mode"] = self._log_display_mode
self._preferences = prefs self._preferences = prefs
save_preferences(prefs) save_preferences(prefs)
# Keep in-memory pipeline config aligned with panel values. # Keep in-memory pipeline config aligned with panel values.
@@ -304,6 +217,7 @@ class FingerDashboardWindow(QMainWindow):
scale = self._layout_scale_factor() scale = self._layout_scale_factor()
column_height = self._scaled_top_row_height(scale) column_height = self._scaled_top_row_height(scale)
charts_slot = self.layout.top_slot("charts") charts_slot = self.layout.top_slot("charts")
log_slot = setup.log_panel
for slot in setup.top_row.slots: for slot in setup.top_row.slots:
if not slot.enabled: if not slot.enabled:
@@ -322,11 +236,27 @@ class FingerDashboardWindow(QMainWindow):
settings_height, settings_height,
) )
if (
self._log_panel_visible
and log_slot.enabled
and log_slot.height_px > 0
):
self.log_widget.show()
log_height = self._scaled_px(log_slot.height_px, scale)
charts_height = self._scaled_px(charts_slot.height_px, scale)
self.log_widget.setFixedHeight(log_height)
self.force_charts.set_panel_size(
self._scaled_px(charts_slot.width_px, scale),
charts_height,
)
self.charts_column.setSizes([charts_height, log_height])
else:
self.log_widget.hide()
self.force_charts.set_panel_size( self.force_charts.set_panel_size(
self._scaled_px(charts_slot.width_px, scale), self._scaled_px(charts_slot.width_px, scale),
column_height, column_height,
) )
self.charts_column.setSizes([column_height]) self.charts_column.setSizes([column_height, 0])
self.h_splitter.setSizes( self.h_splitter.setSizes(
[self._scaled_px(slot.width_px, scale) for slot in setup.top_row.slots if slot.enabled] [self._scaled_px(slot.width_px, scale) for slot in setup.top_row.slots if slot.enabled]
@@ -349,6 +279,12 @@ class FingerDashboardWindow(QMainWindow):
width += margins.left() + margins.right() width += margins.left() + margins.right()
height = layout.charts_column_height_px() height = layout.charts_column_height_px()
if (
self._log_panel_visible
and layout.panel_setup.log_panel.enabled
and layout.panel_setup.log_panel.height_px > 0
):
height += self.charts_column.handleWidth()
height += margins.top() + margins.bottom() height += margins.top() + margins.bottom()
return max(layout.window_width, width), max(layout.window_height, height) return max(layout.window_width, width), max(layout.window_height, height)
@@ -380,6 +316,12 @@ class FingerDashboardWindow(QMainWindow):
base_width += max(len(enabled_slots) - 1, 0) * self.h_splitter.handleWidth() base_width += max(len(enabled_slots) - 1, 0) * self.h_splitter.handleWidth()
base_height = self.layout.charts_column_height_px() base_height = self.layout.charts_column_height_px()
if (
self._log_panel_visible
and self.layout.panel_setup.log_panel.enabled
and self.layout.panel_setup.log_panel.height_px > 0
):
base_height += self.charts_column.handleWidth()
if base_width <= 0 or base_height <= 0: if base_width <= 0 or base_height <= 0:
return 1.0 return 1.0
return min(available_width / base_width, available_height / base_height) return min(available_width / base_width, available_height / base_height)
@@ -628,6 +570,8 @@ class FingerDashboardWindow(QMainWindow):
camera_id = str(camera_id).strip() camera_id = str(camera_id).strip()
if not camera_id: if not camera_id:
return return
# Explicit camera pick clears sticky file mode and persists the live source.
self.settings_panel.set_video_source(camera_id)
self.config.video = camera_id self.config.video = camera_id
self._save_preferences() self._save_preferences()
self._update_play_enabled_for_source() self._update_play_enabled_for_source()
@@ -646,41 +590,48 @@ class FingerDashboardWindow(QMainWindow):
self._update_play_enabled_for_source() self._update_play_enabled_for_source()
def _select_config(self) -> None: def _select_config(self) -> None:
from .preferences import (
default_finger_config,
normalize_config_preference,
packaged_finger_bundle_root,
)
start = packaged_finger_bundle_root()
start_dir = str(start) if start is not None else "sensors/finger"
path, _ = QFileDialog.getOpenFileName( path, _ = QFileDialog.getOpenFileName(
self, self,
"选择配置文件", "选择配置文件",
"config", start_dir,
"JSON (*.json);;所有文件 (*)", "JSON (*.json);;所有文件 (*)",
) )
if path: if path:
self.settings_panel.set_config_path(path) config = normalize_config_preference(path) or default_finger_config()
self.config.config = path self.settings_panel.set_config_path(config)
self.config.config = config
self._save_preferences() self._save_preferences()
def _select_finger_assets(self) -> None: def _select_finger_assets(self) -> None:
from .preferences import (
default_finger_assets_dir,
normalize_assets_preference,
packaged_finger_assets_root,
)
start = packaged_finger_assets_root()
start_dir = (
str(start)
if start is not None
else (self.settings_panel.finger_export_dir() or default_finger_assets_dir())
)
path = QFileDialog.getExistingDirectory( path = QFileDialog.getExistingDirectory(
self, self,
"选择传感器 bundle 目录", "选择 finger 资产目录",
self.settings_panel.sensor_bundle_dir() or "sensors/finger", start_dir,
) )
if path: if path:
bundle_dir = Path(path) assets = normalize_assets_preference(path)
config_path = bundle_dir / "sensor.json" self.settings_panel.set_finger_export_dir(assets)
manifest_path = bundle_dir / "manifest.json" self.config.export_dir = assets
assets_dir = bundle_dir / "assets" / "30fps-0703"
if manifest_path.is_file():
try:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
except Exception:
manifest = {}
rel_assets = manifest.get("assets_dir")
if isinstance(rel_assets, str) and rel_assets.strip():
assets_dir = bundle_dir / rel_assets
if config_path.is_file():
self.settings_panel.set_config_path(str(config_path))
self.config.config = str(config_path)
self.settings_panel.set_finger_export_dir(str(assets_dir))
self.config.export_dir = str(assets_dir)
self._save_preferences() self._save_preferences()
def _select_csv_dir(self) -> None: def _select_csv_dir(self) -> None:
+103 -3
View File
@@ -13,6 +13,17 @@ LEGACY_CAMERA_SELECTION_FILE = "camera_selection.json"
# Keep preference.json at examples/gui/ for backward compatibility. # Keep preference.json at examples/gui/ for backward compatibility.
_GUI_DIR = Path(__file__).resolve().parents[1] _GUI_DIR = Path(__file__).resolve().parents[1]
# Old lab / repo-relative paths that predate packaged sensors/<name>/.
_LEGACY_CONFIG_PATHS = {
"dd02",
"dd02-ov",
"config/dd02-ov.json",
"./config/dd02-ov.json",
"config/finger.json",
"./config/finger.json",
"sensors/finger/sensor.json",
}
def preference_path() -> Path: def preference_path() -> Path:
# Keep preference file at examples/gui/preference.json # Keep preference file at examples/gui/preference.json
@@ -23,6 +34,94 @@ def legacy_camera_selection_path() -> Path:
return resource_path(LEGACY_CAMERA_SELECTION_FILE) return resource_path(LEGACY_CAMERA_SELECTION_FILE)
def default_finger_config() -> str:
"""Public product config stem resolved via packaged ``sensors/finger``."""
return "finger"
def default_finger_assets_dir() -> str:
try:
from orisys.finger import DEFAULT_FINGER_EXPORT_DIR
return DEFAULT_FINGER_EXPORT_DIR
except ImportError:
return "sensors/finger/assets/30fps-0703"
def packaged_finger_bundle_root() -> Path | None:
"""Filesystem root of the installed ``sensors/finger`` bundle when present."""
try:
from orisys.configs import resolve_sensor_bundle
except ImportError:
return None
bundle = resolve_sensor_bundle("finger")
return bundle.root if bundle is not None else None
def packaged_finger_assets_root() -> Path | None:
try:
from orisys.configs import resolve_sensor_asset_dir
except ImportError:
return None
return resolve_sensor_asset_dir("finger")
def normalize_config_preference(value: str | None) -> str:
"""Map legacy config paths to the packaged ``finger`` sensor name."""
if value is None or not str(value).strip():
return default_finger_config()
text = str(value).strip().replace("\\", "/")
if text in _LEGACY_CONFIG_PATHS or text.endswith("/dd02-ov.json"):
return default_finger_config()
root = packaged_finger_bundle_root()
if root is not None:
try:
if Path(value).resolve() == (root / "sensor.json").resolve():
return default_finger_config()
except OSError:
pass
return text
def normalize_assets_preference(value: str | None) -> str:
"""Prefer packaged ``sensors/finger/assets/...`` when saved paths are stale."""
default = default_finger_assets_dir()
if value is None or not str(value).strip():
return default
text = str(value).strip()
posix = text.replace("\\", "/")
if posix == default or posix.endswith("/sensors/finger/assets/30fps-0703"):
return default
path = Path(text)
if path.is_dir():
packaged = packaged_finger_assets_root()
if packaged is not None:
try:
if path.resolve() == packaged.resolve():
return default
except OSError:
pass
# Keep a user-selected custom directory that still exists.
if "sensors/finger" in posix or path.name == "30fps-0703":
return text if path.is_absolute() else default
return text
# Missing old absolute/repo paths (assets/finger, ./config era) -> packaged default.
return default
def _normalize_preferences(prefs: dict[str, Any]) -> dict[str, Any]:
prefs = dict(prefs)
if "config_path" in prefs or prefs:
prefs["config_path"] = normalize_config_preference(prefs.get("config_path"))
if "finger_assets_dir" in prefs or prefs:
prefs["finger_assets_dir"] = normalize_assets_preference(prefs.get("finger_assets_dir"))
prefs.setdefault("sensor_bundle_dir", "sensors/finger")
return prefs
def _read_json(path: Path) -> dict[str, Any]: def _read_json(path: Path) -> dict[str, Any]:
if not path.is_file(): if not path.is_file():
return {} return {}
@@ -34,14 +133,14 @@ def _read_json(path: Path) -> dict[str, Any]:
def load_preferences() -> dict[str, Any]: def load_preferences() -> dict[str, Any]:
"""Load preferences, migrating camera_id from legacy file when needed.""" """Load preferences, migrating camera_id / legacy config paths when needed."""
prefs = _read_json(preference_path()) prefs = _read_json(preference_path())
if "video_source" not in prefs: if "video_source" not in prefs:
legacy = _read_json(legacy_camera_selection_path()) legacy = _read_json(legacy_camera_selection_path())
camera_id = legacy.get("camera_id") camera_id = legacy.get("camera_id")
if camera_id is not None and str(camera_id).strip(): if camera_id is not None and str(camera_id).strip():
prefs["video_source"] = str(camera_id).strip() prefs["video_source"] = str(camera_id).strip()
return prefs return _normalize_preferences(prefs)
def save_preferences(prefs: dict[str, Any]) -> None: def save_preferences(prefs: dict[str, Any]) -> None:
@@ -49,9 +148,10 @@ def save_preferences(prefs: dict[str, Any]) -> None:
if not isinstance(prefs, dict): if not isinstance(prefs, dict):
return return
path = preference_path() path = preference_path()
normalized = _normalize_preferences(prefs)
try: try:
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(prefs, indent=2) + "\n", encoding="utf-8") path.write_text(json.dumps(normalized, indent=2) + "\n", encoding="utf-8")
except OSError: except OSError:
pass pass
+31 -1
View File
@@ -25,6 +25,23 @@ class SensorBundlePreset:
export_dir: str export_dir: str
def _packaged_finger_preset() -> SensorBundlePreset | None:
"""Use logical names that resolve through the installed wheel sensors/ layout."""
try:
from orisys.configs import resolve_sensor_bundle
except ImportError:
return None
bundle = resolve_sensor_bundle("finger")
if bundle is None:
return None
return SensorBundlePreset(
label="ORY-FINGER",
bundle_dir="sensors/finger",
config="finger",
export_dir=DEFAULT_FINGER_EXPORT_DIR,
)
def _load_manifest(bundle_root: Path) -> dict: def _load_manifest(bundle_root: Path) -> dict:
path = bundle_root / "manifest.json" path = bundle_root / "manifest.json"
if not path.is_file(): if not path.is_file():
@@ -50,6 +67,14 @@ def _bundle_from_root(bundle_root: Path, *, label: str | None = None) -> SensorB
break break
except ValueError: except ValueError:
continue continue
# Packaged / checkout sensors/finger: keep resolvable product names, not absolute paths.
if rel_bundle.replace("\\", "/").endswith("sensors/finger") or bundle_root.name == "finger":
return SensorBundlePreset(
label=label or "ORY-FINGER",
bundle_dir="sensors/finger",
config="finger",
export_dir=DEFAULT_FINGER_EXPORT_DIR,
)
return SensorBundlePreset( return SensorBundlePreset(
label=label or "ORY-FINGER", label=label or "ORY-FINGER",
bundle_dir=rel_bundle, bundle_dir=rel_bundle,
@@ -59,6 +84,9 @@ def _bundle_from_root(bundle_root: Path, *, label: str | None = None) -> SensorB
def default_finger_bundle() -> SensorBundlePreset: def default_finger_bundle() -> SensorBundlePreset:
packaged = _packaged_finger_preset()
if packaged is not None:
return packaged
for root in _repo_root_candidates(): for root in _repo_root_candidates():
preset = _bundle_from_root(root / "sensors" / "finger", label="ORY-FINGER") preset = _bundle_from_root(root / "sensors" / "finger", label="ORY-FINGER")
if preset is not None: if preset is not None:
@@ -66,7 +94,7 @@ def default_finger_bundle() -> SensorBundlePreset:
return SensorBundlePreset( return SensorBundlePreset(
label="ORY-FINGER", label="ORY-FINGER",
bundle_dir="sensors/finger", bundle_dir="sensors/finger",
config="sensors/finger/sensor.json", config="finger",
export_dir=DEFAULT_FINGER_EXPORT_DIR, export_dir=DEFAULT_FINGER_EXPORT_DIR,
) )
@@ -108,6 +136,8 @@ def sensor_bundle_preset_index(
for idx, preset in enumerate(presets): for idx, preset in enumerate(presets):
if config == preset.config and export_dir == preset.export_dir: if config == preset.config and export_dir == preset.export_dir:
return idx return idx
if config in ("finger", "dd02", "dd02-ov") and preset.config == "finger":
return idx
if config.endswith("/sensor.json") and export_dir.endswith("/" + Path(preset.export_dir).name): if config.endswith("/sensor.json") and export_dir.endswith("/" + Path(preset.export_dir).name):
return idx return idx
return -1 return -1
+27 -73
View File
@@ -15,15 +15,11 @@ from PyQt5.QtWidgets import (
QWidget, QWidget,
) )
from orisys.finger import DEFAULT_FINGER_EXPORT_DIR
from viz.render_2d import ArrowBackground from viz.render_2d import ArrowBackground
from ..layout import DashboardLayout, LAYOUT from ..layout import DashboardLayout, LAYOUT
from ..mesh_view_state import MeshViewState from ..mesh_view_state import MeshViewState
from ..presets import (
SensorBundlePreset,
discover_sensor_bundle_presets,
sensor_bundle_preset_index,
)
from ..transport_icons import reset_icon, transport_icon from ..transport_icons import reset_icon, transport_icon
_PLAY_PAUSE_BUTTON_NAME = "btnPlayPause" _PLAY_PAUSE_BUTTON_NAME = "btnPlayPause"
@@ -90,7 +86,7 @@ class SettingsPanelWidget(QWidget):
self.show_log_check = QCheckBox("Show log panel") self.show_log_check = QCheckBox("Show log panel")
self.show_log_check.setChecked(show_log) self.show_log_check.setChecked(show_log)
self.show_log_check.toggled.connect(self.log_visibility_changed.emit) self.show_log_check.toggled.connect(self.log_visibility_changed.emit)
self.show_log_check.hide() root.addWidget(self.show_log_check)
self.show_flow_arrows_check = QCheckBox("Show displacement arrows") self.show_flow_arrows_check = QCheckBox("Show displacement arrows")
self.show_flow_arrows_check.setChecked(self._mesh_state.flow_arrows_enabled) self.show_flow_arrows_check.setChecked(self._mesh_state.flow_arrows_enabled)
@@ -114,6 +110,8 @@ class SettingsPanelWidget(QWidget):
self.camera_combo = QComboBox() self.camera_combo = QComboBox()
self.camera_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed) self.camera_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
self.camera_combo.setToolTip("Select an available camera") self.camera_combo.setToolTip("Select an available camera")
# activated fires on user pick even when the index is unchanged (file → live override).
self.camera_combo.activated.connect(self._on_camera_combo_changed)
self.camera_combo.currentIndexChanged.connect(self._on_camera_combo_changed) self.camera_combo.currentIndexChanged.connect(self._on_camera_combo_changed)
video_row.addWidget(self.camera_combo, 1) video_row.addWidget(self.camera_combo, 1)
self.btn_refresh_cameras = QPushButton("Refresh") self.btn_refresh_cameras = QPushButton("Refresh")
@@ -138,38 +136,35 @@ class SettingsPanelWidget(QWidget):
self._video_file_path = "" self._video_file_path = ""
self._updating_camera_options = False self._updating_camera_options = False
root.addWidget(self._caption("Sensor bundle")) root.addWidget(self._caption("Config"))
bundle_row = QHBoxLayout() config_row = QHBoxLayout()
bundle_row.setSpacing(6) config_row.setSpacing(6)
self.sensor_bundle_combo = QComboBox() self.config_edit = QLineEdit("finger")
self.sensor_bundle_combo.setToolTip("Select a finger sensor bundle") config_row.addWidget(self.config_edit, 1)
self._bundle_presets = discover_sensor_bundle_presets()
for preset in self._bundle_presets:
self.sensor_bundle_combo.addItem(preset.label, preset)
self.sensor_bundle_combo.addItem("Custom…", None)
self._updating_bundle_combo = False
self.sensor_bundle_combo.currentIndexChanged.connect(self._on_bundle_combo_changed)
bundle_row.addWidget(self.sensor_bundle_combo, 1)
self.btn_browse_finger_assets = QPushButton("…")
self.btn_browse_finger_assets.setFixedWidth(32)
self.btn_browse_finger_assets.setToolTip("Browse sensor bundle folder")
self.btn_browse_finger_assets.clicked.connect(self.browse_finger_assets_requested.emit)
bundle_row.addWidget(self.btn_browse_finger_assets)
root.addLayout(bundle_row)
self.config_edit = QLineEdit("sensors/finger/sensor.json")
self.config_edit.hide()
self.btn_browse_config = QPushButton("…") self.btn_browse_config = QPushButton("…")
self.btn_browse_config.hide() self.btn_browse_config.setFixedWidth(32)
self.btn_browse_config.setToolTip("Browse config JSON")
self.btn_browse_config.clicked.connect(self.browse_config_requested.emit)
config_row.addWidget(self.btn_browse_config)
root.addLayout(config_row)
# Optional calibration path is kept for future re-enablement but hidden for now. # Optional calibration path is kept for future re-enablement but hidden for now.
self.cal_edit = QLineEdit("") self.cal_edit = QLineEdit("")
self.cal_edit.setPlaceholderText("Optional calibration path") self.cal_edit.setPlaceholderText("Optional calibration path")
self.cal_edit.hide() self.cal_edit.hide()
self.finger_assets_edit = QLineEdit("sensors/finger/assets/30fps-0703") root.addWidget(self._caption("Finger assets"))
self.finger_assets_edit.hide() finger_assets_row = QHBoxLayout()
self._apply_bundle_preset(self._bundle_presets[0]) finger_assets_row.setSpacing(6)
self.finger_assets_edit = QLineEdit(DEFAULT_FINGER_EXPORT_DIR)
self.finger_assets_edit.setPlaceholderText("Blender export bundle directory")
finger_assets_row.addWidget(self.finger_assets_edit, 1)
self.btn_browse_finger_assets = QPushButton("…")
self.btn_browse_finger_assets.setFixedWidth(32)
self.btn_browse_finger_assets.setToolTip("Browse finger asset directory")
self.btn_browse_finger_assets.clicked.connect(self.browse_finger_assets_requested.emit)
finger_assets_row.addWidget(self.btn_browse_finger_assets)
root.addLayout(finger_assets_row)
root.addWidget(self._caption("CSV save directory")) root.addWidget(self._caption("CSV save directory"))
csv_dir_row = QHBoxLayout() csv_dir_row = QHBoxLayout()
@@ -344,18 +339,16 @@ class SettingsPanelWidget(QWidget):
def set_config_path(self, value: str) -> None: def set_config_path(self, value: str) -> None:
self.config_edit.setText(value) self.config_edit.setText(value)
self._sync_bundle_combo_from_paths()
def set_calibration_path(self, value: str) -> None: def set_calibration_path(self, value: str) -> None:
self.cal_edit.setText(value or "") self.cal_edit.setText(value or "")
def set_finger_export_dir(self, value: str) -> None: def set_finger_export_dir(self, value: str) -> None:
self.finger_assets_edit.setText(value) self.finger_assets_edit.setText(value)
self._sync_bundle_combo_from_paths()
def set_finger_export_dir_enabled(self, enabled: bool) -> None: def set_finger_export_dir_enabled(self, enabled: bool) -> None:
self.finger_assets_edit.setEnabled(enabled)
self.btn_browse_finger_assets.setEnabled(enabled) self.btn_browse_finger_assets.setEnabled(enabled)
self.sensor_bundle_combo.setEnabled(enabled)
def set_csv_save_dir(self, value: str) -> None: def set_csv_save_dir(self, value: str) -> None:
self.csv_dir_edit.setText(value) self.csv_dir_edit.setText(value)
@@ -396,33 +389,6 @@ class SettingsPanelWidget(QWidget):
return return
self.csv_record_toggled.emit(not self._csv_recording) self.csv_record_toggled.emit(not self._csv_recording)
def _apply_bundle_preset(self, preset: SensorBundlePreset) -> None:
self.config_edit.setText(preset.config)
self.finger_assets_edit.setText(preset.export_dir)
def _on_bundle_combo_changed(self, _index: int) -> None:
if self._updating_bundle_combo:
return
preset = self.sensor_bundle_combo.currentData()
if isinstance(preset, SensorBundlePreset):
self._apply_bundle_preset(preset)
def _sync_bundle_combo_from_paths(self) -> None:
idx = sensor_bundle_preset_index(
self.config_edit.text(),
self.finger_assets_edit.text(),
self._bundle_presets,
)
if idx < 0:
idx = self.sensor_bundle_combo.findData(None)
if idx < 0:
return
self._updating_bundle_combo = True
self.sensor_bundle_combo.blockSignals(True)
self.sensor_bundle_combo.setCurrentIndex(idx)
self.sensor_bundle_combo.blockSignals(False)
self._updating_bundle_combo = False
def _emit_arrow_background_changed(self, _index: int) -> None: def _emit_arrow_background_changed(self, _index: int) -> None:
value = self.arrow_background_combo.currentData() value = self.arrow_background_combo.currentData()
if value: if value:
@@ -449,15 +415,6 @@ class SettingsPanelWidget(QWidget):
def finger_export_dir(self) -> str: def finger_export_dir(self) -> str:
return self.finger_assets_edit.text().strip() return self.finger_assets_edit.text().strip()
def sensor_bundle_dir(self) -> str:
preset = self.sensor_bundle_combo.currentData()
if isinstance(preset, SensorBundlePreset):
return preset.bundle_dir
config = self.config_path()
if config.endswith("/sensor.json") or config.endswith("\\sensor.json"):
return config.rsplit("/", 1)[0].rsplit("\\", 1)[0]
return ""
def csv_save_dir(self) -> str: def csv_save_dir(self) -> str:
return self.csv_dir_edit.text().strip() return self.csv_dir_edit.text().strip()
@@ -465,7 +422,6 @@ class SettingsPanelWidget(QWidget):
"""Serialize user-facing settings suitable for preference.json.""" """Serialize user-facing settings suitable for preference.json."""
return { return {
"video_source": self.video_source(), "video_source": self.video_source(),
"sensor_bundle_dir": self.sensor_bundle_dir(),
"config_path": self.config_path(), "config_path": self.config_path(),
"finger_assets_dir": self.finger_export_dir(), "finger_assets_dir": self.finger_export_dir(),
"csv_save_dir": self.csv_save_dir(), "csv_save_dir": self.csv_save_dir(),
@@ -511,8 +467,6 @@ class SettingsPanelWidget(QWidget):
if isinstance(assets, str) and assets.strip(): if isinstance(assets, str) and assets.strip():
self.set_finger_export_dir(assets.strip()) self.set_finger_export_dir(assets.strip())
self._sync_bundle_combo_from_paths()
csv_dir = state.get("csv_save_dir") csv_dir = state.get("csv_save_dir")
if isinstance(csv_dir, str) and csv_dir.strip(): if isinstance(csv_dir, str) and csv_dir.strip():
self.set_csv_save_dir(csv_dir.strip()) self.set_csv_save_dir(csv_dir.strip())
+1 -1
View File
@@ -7,7 +7,7 @@ ORY-MINI:单传感器基础流程
3. 按需计算接触区域 3. 按需计算接触区域
4. 读取并可视化结果 4. 读取并可视化结果
SDK version: 0.4.3 SDK version: 0.5.0
""" """
import os import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
+1 -1
View File
@@ -18,7 +18,7 @@
- Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support() - Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support()
- 每个传感器进程都会占用独立的 GPU 显存 - 每个传感器进程都会占用独立的 GPU 显存
SDK version: 0.4.3 SDK version: 0.5.0
""" """
import os import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0" os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
+7 -2
View File
@@ -7,6 +7,10 @@ from pathlib import Path
import numpy as np import numpy as np
DEFAULT_FINGER_VIZ_POSE = "sensors/finger/assets/visualization/world_to_cam.npy" DEFAULT_FINGER_VIZ_POSE = "sensors/finger/assets/visualization/world_to_cam.npy"
_LEGACY_FINGER_VIZ_POSES = {
"assets/finger/visualization/world_to_cam.npy",
DEFAULT_FINGER_VIZ_POSE,
}
def _finger_bundle_viz_pose_candidates() -> list[Path]: def _finger_bundle_viz_pose_candidates() -> list[Path]:
@@ -63,8 +67,9 @@ def resolve_viz_pose_path(path: str | Path | None = None) -> Path:
here = Path(__file__).resolve() here = Path(__file__).resolve()
candidates: list[Path] = [] candidates: list[Path] = []
# Prefer the sensors/finger bundle for the default (and matching relative) path. # Prefer the sensors/finger bundle for default / legacy relative paths.
if path is None or raw.as_posix() == DEFAULT_FINGER_VIZ_POSE or not raw.is_absolute(): posix = raw.as_posix()
if path is None or posix in _LEGACY_FINGER_VIZ_POSES or not raw.is_absolute():
candidates.extend(_finger_bundle_viz_pose_candidates()) candidates.extend(_finger_bundle_viz_pose_candidates())
try: try:
+29 -2
View File
@@ -1,5 +1,32 @@
"""Signal filtering helpers for example force charts.""" """Signal filtering helpers for example force charts."""
from orisys.signal_processing import lowpass_filter from __future__ import annotations
__all__ = ["lowpass_filter"] import numpy as np
from scipy.signal import butter, filtfilt
def lowpass_filter(data, cutoff_hz=5, 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")
taplen = max(len(a), len(b))
# scipy.signal.filtfilt requires len(x) > padlen (default 3 * taplen for recent scipy).
padlen = 3 * taplen
if y.size <= padlen:
return y.copy()
try:
return filtfilt(b, a, y)
except ValueError:
return y.copy()
+10 -1
View File
@@ -79,7 +79,16 @@ dev_install = true
name = "sensors" name = "sensors"
source = "sensors" source = "sensors"
target = "sensors" target = "sensors"
patterns = ["**/*.json", "finger/assets/30fps*/*", "finger/assets/visualization/*"] # JSON bundles (sensor.json, force_calibration.json) plus finger mesh/viz assets.
# Prefer the flat layout sensors/finger/assets/{30fps-*,visualization}/;
# also accept a nested assets/finger/ tree if present in a checkout.
patterns = [
"**/*.json",
"finger/assets/30fps*/*",
"finger/assets/visualization/*",
"finger/assets/finger/30fps*/*",
"finger/assets/finger/visualization/*",
]
preserve_tree = true preserve_tree = true
dev_install = true dev_install = true
+27 -5
View File
@@ -1,8 +1,30 @@
{ {
"normal_force": { "schema_version": 1,
"model": "linear", "unit": "N",
"slope": 3.92685554668795e-05, "channels": {
"offset": 0.0217596967278535, "normal": {
"unit": "N" "method": "piecewise_linear",
"raw_key": "fnormal",
"extrapolate": true,
"clamp_output_min": 0.0,
"points": [
[1000.0, 0.0],
[10000.0, 0.377],
[30000.0, 1.377],
[62000.0, 2.377]
]
},
"shear": {
"method": "piecewise_linear",
"input": "components",
"signed": true,
"extrapolate": true,
"clamp_output_min": 0.0,
"points": [
[1000.0, 0.0],
[36000.0, 1.19],
[68000.0, 2.06]
]
}
} }
} }
+1 -1
View File
@@ -22,7 +22,7 @@
"dis_preset": "medium", "dis_preset": "medium",
"dis_mode": 1, "dis_mode": 1,
"algorithm_list": ["raft", "dis"], "algorithm_list": ["raft", "dis"],
"hhd_downsample": 5 "hhd_downsample": 1
}, },
"contact": { "contact": {
"min_contour_area": 3000, "min_contour_area": 3000,
-8
View File
@@ -1,8 +0,0 @@
{
"normal_force": {
"model": "linear",
"slope": 3.92685554668795e-05,
"offset": 0.0217596967278535,
"unit": "N"
}
}
+33 -4
View File
@@ -1,13 +1,42 @@
"""Setuptools entry point; bundle rules are read from pyproject.toml.""" """Setuptools entry point; bundle rules are read from pyproject.toml.
``dev_install`` bundle items (configs/sensors) are staged into the ``orisys``
package during build so ``pip install .`` places them at
``site-packages/orisys/...`` (importlib / configs resolvers), not under
``sys.prefix/orisys`` via deprecated ``data_files``.
"""
from __future__ import annotations
from pathlib import Path
import sys import sys
from pathlib import Path
from setuptools import setup from setuptools import setup
from setuptools.command.build_py import build_py as _build_py
REPO_ROOT = Path(__file__).resolve().parent REPO_ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(REPO_ROOT / "build" / "common")) sys.path.insert(0, str(REPO_ROOT / "build" / "common"))
from pyproject_utils import bundle_data_files from pyproject_utils import stage_dev_bundle_into # noqa: E402
setup(data_files=bundle_data_files(REPO_ROOT))
class build_py(_build_py):
def run(self) -> None:
super().run()
# After pure modules are copied, stage configs/sensors into build/lib/orisys.
pkg_dir = Path(self.build_lib) / "orisys"
pkg_dir.mkdir(parents=True, exist_ok=True)
stage_dev_bundle_into(REPO_ROOT, pkg_dir)
setup(
cmdclass={"build_py": build_py},
# Declared so wheels record the staged files; actual copy happens in build_py.
package_data={
"orisys": [
"configs/*.json",
"sensors/**/*",
],
},
include_package_data=True,
)
+1 -1
View File
@@ -5,7 +5,7 @@ A Python library for processing and analyzing optical tactile sensor data,
including image stitching, optical flow tracking, and deformation decomposition. including image stitching, optical flow tracking, and deformation decomposition.
""" """
__version__ = "0.4.3" __version__ = "0.5.0"
__author__ = "Orisys" __author__ = "Orisys"
from .sensor import Sensor from .sensor import Sensor
+62 -7
View File
@@ -11,6 +11,7 @@ import numpy as np
import cv2 import cv2
import time import time
from enum import Enum, auto from enum import Enum, auto
from pathlib import Path
from typing import Optional, Tuple from typing import Optional, Tuple
# from threading import Thread, Event # from threading import Thread, Event
# from queue import Queue, Empty # from queue import Queue, Empty
@@ -22,7 +23,12 @@ from . import configs
from .get_depth import Contact3D from .get_depth import Contact3D
from .backend_availability import (check_cupy_available, check_opencl_available) from .backend_availability import (check_cupy_available, check_opencl_available)
from .frame_processors import create_frame_processor from .frame_processors import create_frame_processor
from .signal_processing import DEFAULT_SAMPLE_RATE_HZ, LowpassFilter from .signal_processing import (
DEFAULT_SAMPLE_RATE_HZ,
LowpassFilter,
calibrate_normal_force,
calibrate_shear_components,
)
from .capture import ( from .capture import (
CaptureConfig, CaptureConfig,
CaptureWorker, CaptureWorker,
@@ -148,6 +154,7 @@ class Sensor:
self.n_boxes = None self.n_boxes = None
self.fnormal = self.fshearx = self.fsheary = self.ftor = 0.0 self.fnormal = self.fshearx = self.fsheary = self.ftor = 0.0
self._force_raw = (0.0, 0.0, 0.0) self._force_raw = (0.0, 0.0, 0.0)
self._force_cal_sensor_name = self._resolve_force_calibration_sensor_name()
self._lowpass_enabled = False self._lowpass_enabled = False
self._lowpass_cutoff_hz = 0.0 self._lowpass_cutoff_hz = 0.0
self._lowpass_sample_rate_hz = float(DEFAULT_SAMPLE_RATE_HZ) self._lowpass_sample_rate_hz = float(DEFAULT_SAMPLE_RATE_HZ)
@@ -158,6 +165,13 @@ class Sensor:
order=self._lowpass_order, order=self._lowpass_order,
n_channels=3, n_channels=3,
) )
if self.verbose:
if self._force_cal_sensor_name:
print(
f"Force calibration: sensors/{self._force_cal_sensor_name}/force_calibration.json"
)
else:
print("Force calibration: raw (no force_calibration.json)")
# Contact Data Variables (public) # Contact Data Variables (public)
self.centroid = (0, 0) self.centroid = (0, 0)
@@ -790,17 +804,58 @@ class Sensor:
self._update_hhd_mask_cache() self._update_hhd_mask_cache()
return self return self
def _resolve_force_calibration_sensor_name(self) -> Optional[str]:
"""Return sensors/<name> when force_calibration.json exists; else None.
Lookup order:
1. ``config_name`` stem / product alias (e.g. ``box``)
2. ``sensors/<name>/...`` path segment in ``config_name``
Missing JSON means ``FNORMAL/FSHEAR*`` stay raw (no hardcoded fallback).
"""
text = str(self.config_name or "").strip().replace("\\", "/")
if not text:
return None
candidates: list[str] = []
stem = Path(text).stem if text else ""
if stem:
candidates.append(stem)
parts = [p for p in text.split("/") if p]
if "sensors" in parts:
idx = parts.index("sensors")
if idx + 1 < len(parts):
candidates.append(parts[idx + 1])
for name in configs.PRODUCT_BUNDLES:
if text == name or text.endswith(f"/sensors/{name}") or text.endswith(f"/sensors/{name}/sensor.json"):
candidates.append(name)
seen: set[str] = set()
for candidate in candidates:
alias = configs.CONFIG_ALIASES.get(candidate, candidate)
for name in (alias, candidate):
if name in seen:
continue
seen.add(name)
if configs.load_force_calibration(name) is not None:
return name
return None
def _apply_force_filter(self): def _apply_force_filter(self):
raw = np.asarray(self._force_raw, dtype=np.float64) raw = np.asarray(self._force_raw, dtype=np.float64)
if self._lowpass_enabled and self._lowpass_cutoff_hz > 0: if self._lowpass_enabled and self._lowpass_cutoff_hz > 0:
filtered = self._force_filter.process(raw) filtered = self._force_filter.process(raw)
self.fnormal, self.fshearx, self.fsheary = (float(v) for v in filtered) fn, fx, fy = (float(v) for v in filtered)
else: else:
self.fnormal, self.fshearx, self.fsheary = ( fn, fx, fy = float(raw[0]), float(raw[1]), float(raw[2])
float(raw[0]),
float(raw[1]), # Apply sensors/<name>/force_calibration.json when present; otherwise keep raw.
float(raw[2]), name = self._force_cal_sensor_name
) if name is not None:
fn = float(calibrate_normal_force(fn, sensor_name=name))
fx, fy = calibrate_shear_components(fx, fy, sensor_name=name)
fx, fy = float(fx), float(fy)
self.fnormal, self.fshearx, self.fsheary = fn, fx, fy
# def export_config(self, path="./config/exported_config.json"): # def export_config(self, path="./config/exported_config.json"):
# """Return current configuration dict.""" # """Return current configuration dict."""
+229 -9
View File
@@ -1,11 +1,18 @@
"""Signal helpers for force filtering and normal-force calibration. """Signal helpers for force filtering and force calibration.
Batch ``lowpass_filter`` is the zero-phase Butterworth used by example charts. Batch ``lowpass_filter`` is the zero-phase Butterworth used by example charts.
``LowpassFilter`` is a causal, stateful filter for per-frame ``Sensor`` outputs. ``LowpassFilter`` is a causal, stateful filter for per-frame ``Sensor`` outputs.
Force calibration resolution order (when register support arrives):
1. Sensor register / EEPROM calibration blob
2. ``sensors/<name>/force_calibration.json``
3. Hardcoded linear defaults below
""" """
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass
import numpy as np import numpy as np
from scipy.signal import butter, filtfilt, sosfilt, sosfilt_zi from scipy.signal import butter, filtfilt, sosfilt, sosfilt_zi
@@ -18,28 +25,241 @@ DEFAULT_SAMPLE_RATE_HZ = 30.0
DEFAULT_LOWPASS_ORDER = 2 DEFAULT_LOWPASS_ORDER = 2
def _load_normal_force_coeffs(sensor_name: str | None = None) -> tuple[float, float]: def _parse_calibration_point(point) -> tuple[float, float]:
if sensor_name: if isinstance(point, dict):
raw = point.get("fnormal", point.get("raw", point.get("x")))
force_n = point.get("force_n", point.get("n", point.get("y")))
if raw is None or force_n is None:
raise ValueError(f"invalid calibration point: {point!r}")
return float(raw), float(force_n)
if isinstance(point, (list, tuple)) and len(point) >= 2:
return float(point[0]), float(point[1])
raise ValueError(f"invalid calibration point: {point!r}")
@dataclass(frozen=True)
class ForceChannelCalibration:
"""One force channel: linear coeffs and/or piecewise-linear breakpoints."""
method: str = "linear"
slope: float | None = None
offset: float | None = None
points: tuple[tuple[float, float], ...] = ()
extrapolate: bool = True
clamp_output_min: float | None = 0.0
input_mode: str = "magnitude"
signed: bool = False
unit: str = "N"
@classmethod
def linear(cls, slope: float, offset: float, *, unit: str = "N") -> ForceChannelCalibration:
return cls(method="linear", slope=float(slope), offset=float(offset), unit=unit)
@classmethod
def from_channel_dict(
cls,
block: dict,
*,
unit: str = "N",
default_method: str = "piecewise_linear",
) -> ForceChannelCalibration | None:
if not isinstance(block, dict):
return None
method = str(block.get("method", block.get("model", default_method))).lower()
if method in ("linear", "affine"):
slope = block.get("slope")
offset = block.get("offset")
if slope is None or offset is None:
return None
return cls.linear(float(slope), float(offset), unit=str(block.get("unit", unit)))
if method != "piecewise_linear":
return None
raw_points = block.get("points", [])
if not isinstance(raw_points, (list, tuple)) or len(raw_points) < 2:
return None
try:
points = tuple(sorted(_parse_calibration_point(p) for p in raw_points))
except (TypeError, ValueError):
return None
if len(points) < 2:
return None
return cls(
method="piecewise_linear",
points=points,
extrapolate=bool(block.get("extrapolate", True)),
clamp_output_min=block.get("clamp_output_min", 0.0),
input_mode=str(block.get("input", "magnitude")).lower(),
signed=bool(block.get("signed", False)),
unit=str(block.get("unit", unit)),
)
def convert(self, raw: float) -> float:
raw = float(raw)
if self.method == "linear":
slope = NORMAL_FORCE_CAL_SLOPE if self.slope is None else float(self.slope)
offset = NORMAL_FORCE_CAL_OFFSET if self.offset is None else float(self.offset)
return slope * raw + offset
points = self.points
if len(points) < 2:
return raw
if raw <= points[0][0]:
segment = (points[0], points[1])
if not self.extrapolate:
return self._clamp(points[0][1])
elif raw >= points[-1][0]:
segment = (points[-2], points[-1])
if not self.extrapolate:
return self._clamp(points[-1][1])
else:
segment = None
for left, right in zip(points, points[1:]):
if left[0] <= raw <= right[0]:
segment = (left, right)
break
if segment is None:
return self._clamp(points[-1][1])
(raw0, n0), (raw1, n1) = segment
if raw1 == raw0:
value = n0
else:
ratio = (raw - raw0) / (raw1 - raw0)
value = n0 + ratio * (n1 - n0)
return self._clamp(value)
def convert_component(self, raw: float) -> float:
if not self.signed:
return self.convert(raw)
raw = float(raw)
sign = -1.0 if raw < 0.0 else 1.0
return sign * self.convert(abs(raw))
def _clamp(self, value: float) -> float:
if self.clamp_output_min is not None:
return max(float(self.clamp_output_min), float(value))
return float(value)
_DEFAULT_NORMAL = ForceChannelCalibration.linear(
NORMAL_FORCE_CAL_SLOPE,
NORMAL_FORCE_CAL_OFFSET,
)
def _load_force_calibration_data(sensor_name: str | None) -> dict | None:
if not sensor_name:
return None
try: try:
from .configs import load_force_calibration from .configs import load_force_calibration
data = load_force_calibration(sensor_name) data = load_force_calibration(sensor_name)
except Exception: except Exception:
data = None return None
if isinstance(data, dict): return data if isinstance(data, dict) else None
def _channel_from_data(data: dict | None, channel: str) -> ForceChannelCalibration | None:
if not isinstance(data, dict):
return None
unit = str(data.get("unit", "N"))
channels = data.get("channels")
if isinstance(channels, dict):
block = channels.get(channel)
cal = ForceChannelCalibration.from_channel_dict(block, unit=unit)
if cal is not None:
return cal
# Legacy single-channel file: {"normal_force": {"model": "linear", ...}}
if channel == "normal":
block = data.get("normal_force", data) block = data.get("normal_force", data)
cal = ForceChannelCalibration.from_channel_dict(
block, unit=unit, default_method="linear"
)
if cal is not None:
return cal
if isinstance(block, dict): if isinstance(block, dict):
slope = block.get("slope") slope = block.get("slope")
offset = block.get("offset") offset = block.get("offset")
if slope is not None and offset is not None: if slope is not None and offset is not None:
return float(slope), float(offset) return ForceChannelCalibration.linear(
float(slope),
float(offset),
unit=str(block.get("unit", unit)),
)
return None
def _normal_calibration(sensor_name: str | None = None) -> ForceChannelCalibration:
cal = _channel_from_data(_load_force_calibration_data(sensor_name), "normal")
return cal if cal is not None else _DEFAULT_NORMAL
def _shear_calibration(sensor_name: str | None = None) -> ForceChannelCalibration | None:
return _channel_from_data(_load_force_calibration_data(sensor_name), "shear")
def _load_normal_force_coeffs(sensor_name: str | None = None) -> tuple[float, float]:
"""Legacy helper: linear slope/offset only (piecewise returns hardcoded defaults)."""
cal = _normal_calibration(sensor_name)
if cal.method == "linear" and cal.slope is not None and cal.offset is not None:
return float(cal.slope), float(cal.offset)
return NORMAL_FORCE_CAL_SLOPE, NORMAL_FORCE_CAL_OFFSET return NORMAL_FORCE_CAL_SLOPE, NORMAL_FORCE_CAL_OFFSET
def calibrate_normal_force(raw_value: float, *, sensor_name: str | None = None) -> float: def calibrate_normal_force(raw_value: float, *, sensor_name: str | None = None) -> float:
"""Convert a raw normal-force reading to Newtons using y = mx + c.""" """Convert a raw normal-force reading to Newtons.
slope, offset = _load_normal_force_coeffs(sensor_name)
return slope * float(raw_value) + offset Uses ``sensors/<name>/force_calibration.json`` when present (piecewise or linear).
Falls back to hardcoded linear defaults when missing.
"""
return _normal_calibration(sensor_name).convert(raw_value)
def calibrate_shear_force(raw_value: float, *, sensor_name: str | None = None) -> float:
"""Convert a raw shear magnitude to Newtons using the shear channel curve.
Returns ``float(raw_value)`` unchanged when no shear calibration is available.
"""
cal = _shear_calibration(sensor_name)
if cal is None:
return float(raw_value)
return cal.convert(raw_value)
def calibrate_shear_components(
shear_x: float,
shear_y: float,
*,
sensor_name: str | None = None,
) -> tuple[float, float]:
"""Convert raw shear components to Newtons.
When the shear channel uses ``input: components`` and ``signed: true``, each
axis is calibrated on absolute magnitude then re-signed. Otherwise the
magnitude is calibrated and components are scaled proportionally.
"""
sx = float(shear_x)
sy = float(shear_y)
cal = _shear_calibration(sensor_name)
if cal is None:
return sx, sy
if cal.input_mode == "components":
return cal.convert_component(sx), cal.convert_component(sy)
magnitude = (sx * sx + sy * sy) ** 0.5
if magnitude <= 0.0:
return 0.0, 0.0
calibrated = cal.convert(magnitude)
scale = calibrated / magnitude
return sx * scale, sy * scale
def _nyquist_hz(sample_rate_hz: float) -> float: def _nyquist_hz(sample_rate_hz: float) -> float:
+1 -1
View File
@@ -8,7 +8,7 @@ Flow:
4. Feed the rectified frame into Sensor.get_img(..., already_processed=True) 4. Feed the rectified frame into Sensor.get_img(..., already_processed=True)
5. Compute deformation/contact and visualize 5. Compute deformation/contact and visualize
SDK version: 0.4.3 SDK version: 0.5.0
""" """
from __future__ import annotations from __future__ import annotations
+25 -4
View File
@@ -35,7 +35,7 @@ def test_pyproject_dynamic_version(packaging_utils):
sys.path.insert(0, str(build_dir)) sys.path.insert(0, str(build_dir))
from common.pyproject_utils import get_version from common.pyproject_utils import get_version
assert get_version(REPO_ROOT) == "0.4.3" assert get_version(REPO_ROOT) == "0.5.0"
def test_pyproject_bundle_manifest(packaging_utils): def test_pyproject_bundle_manifest(packaging_utils):
@@ -48,7 +48,7 @@ def test_pyproject_bundle_manifest(packaging_utils):
items = bundle_items(REPO_ROOT) items = bundle_items(REPO_ROOT)
names = {item["name"] for item in items} names = {item["name"] for item in items}
assert names == {"configs", "examples", "docs", "assets"} assert names == {"configs", "examples", "docs", "sensors"}
data_files = dict(bundle_data_files(REPO_ROOT)) data_files = dict(bundle_data_files(REPO_ROOT))
assert "orisys/configs" in data_files assert "orisys/configs" in data_files
@@ -57,12 +57,33 @@ def test_pyproject_bundle_manifest(packaging_utils):
assert all("/" in path and "*" not in path for path in 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) assert any(path.endswith(".json") for path in config_files)
asset_dirs = [dest for dest in data_files if dest.startswith("orisys/assets/")] sensor_dirs = [dest for dest in data_files if dest.startswith("orisys/sensors")]
assert asset_dirs assert sensor_dirs
all_sensor_files = [path for files in (data_files[d] for d in sensor_dirs) for path in files]
assert any(path.endswith("force_calibration.json") for path in all_sensor_files)
assert any("finger" in path and path.endswith(".json") for path in all_sensor_files)
# Finger mesh/viz assets (flat or nested assets/finger/ checkout).
assert any("finger/assets" in dest or any("finger/assets" in p for p in data_files[dest]) for dest in sensor_dirs)
assert all("*" not in path for files in data_files.values() for path in files) assert all("*" not in path for files in data_files.values() for path in files)
assert "examples" not in data_files assert "examples" not in data_files
def test_stage_dev_bundle_into_writes_sensors_under_package(tmp_path):
import sys
build_dir = REPO_ROOT / "build"
if str(build_dir) not in sys.path:
sys.path.insert(0, str(build_dir))
from common.pyproject_utils import stage_dev_bundle_into
pkg = tmp_path / "orisys"
packaged = stage_dev_bundle_into(REPO_ROOT, pkg)
assert any(p.replace("\\", "/").endswith("sensors/box/force_calibration.json") for p in packaged)
assert (pkg / "sensors" / "box" / "force_calibration.json").is_file()
assert (pkg / "configs").is_dir()
assert any((pkg / "configs").glob("*.json"))
def test_bundle_explicit_files(): def test_bundle_explicit_files():
import sys import sys
+104 -3
View File
@@ -12,6 +12,8 @@ from orisys.signal_processing import (
NORMAL_FORCE_CAL_SLOPE, NORMAL_FORCE_CAL_SLOPE,
LowpassFilter, LowpassFilter,
calibrate_normal_force, calibrate_normal_force,
calibrate_shear_components,
calibrate_shear_force,
lowpass_filter, lowpass_filter,
) )
@@ -73,6 +75,46 @@ def test_calibrate_normal_force_preserves_linear_curve():
assert calibrate_normal_force(0.0) == pytest.approx(NORMAL_FORCE_CAL_OFFSET) assert calibrate_normal_force(0.0) == pytest.approx(NORMAL_FORCE_CAL_OFFSET)
def test_calibrate_normal_force_mini_uses_legacy_linear_file():
raw = 10000.0
expected = NORMAL_FORCE_CAL_SLOPE * raw + NORMAL_FORCE_CAL_OFFSET
assert calibrate_normal_force(raw, sensor_name="mini") == pytest.approx(expected)
def test_calibrate_normal_force_box_uses_piecewise_curve():
assert calibrate_normal_force(10000.0, sensor_name="box") == pytest.approx(0.377)
assert calibrate_normal_force(30000.0, sensor_name="box") == pytest.approx(1.377)
# Mid-segment interpolation between 10000→0.377 and 30000→1.377
mid = calibrate_normal_force(20000.0, sensor_name="box")
assert mid == pytest.approx(0.877)
# Below first breakpoint with extrapolate=true
assert calibrate_normal_force(0.0, sensor_name="box") == pytest.approx(0.0)
def test_calibrate_shear_force_box_piecewise():
assert calibrate_shear_force(1000.0, sensor_name="box") == pytest.approx(0.0)
assert calibrate_shear_force(36000.0, sensor_name="box") == pytest.approx(1.19)
assert calibrate_shear_force(68000.0, sensor_name="box") == pytest.approx(2.06)
def test_calibrate_shear_components_box_signed():
sx, sy = calibrate_shear_components(36000.0, -36000.0, sensor_name="box")
assert sx == pytest.approx(1.19)
assert sy == pytest.approx(-1.19)
def test_calibrate_shear_missing_file_returns_raw():
assert calibrate_shear_force(1234.5, sensor_name="no-such-sensor") == pytest.approx(1234.5)
sx, sy = calibrate_shear_components(1.0, -2.0, sensor_name="no-such-sensor")
assert (sx, sy) == pytest.approx((1.0, -2.0))
def test_calibrate_normal_force_unknown_sensor_falls_back_to_defaults():
raw = 5000.0
expected = NORMAL_FORCE_CAL_SLOPE * raw + NORMAL_FORCE_CAL_OFFSET
assert calibrate_normal_force(raw, sensor_name="no-such-sensor") == pytest.approx(expected)
def _external_sensor(): def _external_sensor():
return Sensor("external", config_name="box", verbose=False, backend="cpu") return Sensor("external", config_name="box", verbose=False, backend="cpu")
@@ -87,7 +129,16 @@ def test_configure_lowpass_filters_read_info_and_keeps_raw():
sensor.info.FSHEARX, sensor.info.FSHEARX,
sensor.info.FSHEARY, sensor.info.FSHEARY,
) )
assert (fn, fx, fy) == pytest.approx((1000.0, -200.0, 50.0)) fn_raw, fx_raw, fy_raw = sensor.read_info(
sensor.info.FNORMAL_RAW,
sensor.info.FSHEARX_RAW,
sensor.info.FSHEARY_RAW,
)
assert (fn_raw, fx_raw, fy_raw) == pytest.approx((1000.0, -200.0, 50.0))
assert fn == pytest.approx(calibrate_normal_force(1000.0, sensor_name="box"))
assert (fx, fy) == pytest.approx(
calibrate_shear_components(-200.0, 50.0, sensor_name="box")
)
sensor.configure(lowpass_cutoff_hz=5.0, lowpass_sample_rate_hz=30.0) sensor.configure(lowpass_cutoff_hz=5.0, lowpass_sample_rate_hz=30.0)
noisy = [(1000.0 + (200.0 if i % 2 else -200.0), -200.0, 50.0) for i in range(40)] noisy = [(1000.0 + (200.0 if i % 2 else -200.0), -200.0, 50.0) for i in range(40)]
@@ -124,7 +175,10 @@ def test_configure_lowpass_disabled_or_zero_cutoff_returns_raw():
sensor.info.FSHEARX, sensor.info.FSHEARX,
sensor.info.FSHEARY, sensor.info.FSHEARY,
) )
assert (fn, fx, fy) == pytest.approx((111.0, 222.0, 333.0)) assert fn == pytest.approx(calibrate_normal_force(111.0, sensor_name="box"))
assert (fx, fy) == pytest.approx(
calibrate_shear_components(222.0, 333.0, sensor_name="box")
)
sensor.configure(lowpass_cutoff_hz=5.0, lowpass_enabled=False) sensor.configure(lowpass_cutoff_hz=5.0, lowpass_enabled=False)
sensor._force_raw = (7.0, 8.0, 9.0) sensor._force_raw = (7.0, 8.0, 9.0)
@@ -135,12 +189,59 @@ def test_configure_lowpass_disabled_or_zero_cutoff_returns_raw():
sensor.info.FSHEARY, sensor.info.FSHEARY,
) )
fn_raw = sensor.read_info(sensor.info.FNORMAL_RAW)[0] fn_raw = sensor.read_info(sensor.info.FNORMAL_RAW)[0]
assert (fn, fx, fy) == pytest.approx((7.0, 8.0, 9.0)) assert fn == pytest.approx(calibrate_normal_force(7.0, sensor_name="box"))
assert (fx, fy) == pytest.approx(
calibrate_shear_components(8.0, 9.0, sensor_name="box")
)
assert fn_raw == pytest.approx(7.0) assert fn_raw == pytest.approx(7.0)
finally: finally:
sensor.disconnect() sensor.disconnect()
def test_box_read_info_applies_force_calibration_json():
sensor = _external_sensor()
try:
assert sensor._force_cal_sensor_name == "box"
sensor._force_raw = (10000.0, 36000.0, -36000.0)
sensor._apply_force_filter()
fn, fx, fy = sensor.read_info(
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY,
)
fn_raw, fx_raw, fy_raw = sensor.read_info(
sensor.info.FNORMAL_RAW,
sensor.info.FSHEARX_RAW,
sensor.info.FSHEARY_RAW,
)
assert (fn_raw, fx_raw, fy_raw) == pytest.approx((10000.0, 36000.0, -36000.0))
assert fn == pytest.approx(0.377)
assert fx == pytest.approx(1.19)
assert fy == pytest.approx(-1.19)
finally:
sensor.disconnect()
def test_read_info_force_stays_raw_without_calibration_json(monkeypatch):
monkeypatch.setattr(
"orisys.sensor.configs.load_force_calibration",
lambda _name: None,
)
sensor = Sensor("external", config_name="box", verbose=False, backend="cpu")
try:
assert sensor._force_cal_sensor_name is None
sensor._force_raw = (111.0, -22.0, 33.0)
sensor._apply_force_filter()
fn, fx, fy = sensor.read_info(
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY,
)
assert (fn, fx, fy) == pytest.approx((111.0, -22.0, 33.0))
finally:
sensor.disconnect()
def test_configure_hhd_downsample_invalidates_solver_cache(): def test_configure_hhd_downsample_invalidates_solver_cache():
sensor = _external_sensor() sensor = _external_sensor()
try: try:
+6 -6
View File
@@ -19,25 +19,25 @@ def wheel_package():
def test_compliant_manylinux_tags(wheel_package): def test_compliant_manylinux_tags(wheel_package):
assert wheel_package.is_compliant_manylinux_wheel( assert wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-manylinux_2_17_x86_64.whl") Path("orisys-0.5.0-cp310-cp310-manylinux_2_17_x86_64.whl")
) )
assert wheel_package.is_compliant_manylinux_wheel( assert wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-manylinux2014_x86_64.whl") Path("orisys-0.5.0-cp310-cp310-manylinux2014_x86_64.whl")
) )
assert wheel_package.is_compliant_manylinux_wheel( assert wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-manylinux_2_28_x86_64.whl") Path("orisys-0.5.0-cp310-cp310-manylinux_2_28_x86_64.whl")
) )
def test_rejects_illegal_manylinux_tags(wheel_package): def test_rejects_illegal_manylinux_tags(wheel_package):
assert not wheel_package.is_compliant_manylinux_wheel( assert not wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-manylinux_x86_64.whl") Path("orisys-0.5.0-cp310-cp310-manylinux_x86_64.whl")
) )
assert not wheel_package.is_compliant_manylinux_wheel( assert not wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-linux_x86_64.whl") Path("orisys-0.5.0-cp310-cp310-linux_x86_64.whl")
) )
assert not wheel_package.is_compliant_manylinux_wheel( assert not wheel_package.is_compliant_manylinux_wheel(
Path("orisys-0.4.3-cp310-cp310-win_amd64.whl") Path("orisys-0.5.0-cp310-cp310-win_amd64.whl")
) )