Compare commits

..

12 Commits

Author SHA1 Message Date
zhangyijian 75fa98eff0 Merge pull request 'Dev' (#19) from dev into main
Reviewed-on: #19
2026-09-02 21:39:03 +08:00
zhangyijian bc059e3539 Merge pull request 'Hotfix/mems cam open' (#20) from hotfix/mems_cam_open into dev
Reviewed-on: #20
2026-09-02 21:38:48 +08:00
ykcheungab 5be2d9759a hotfix/camera not connected error 2026-09-02 21:37:40 +08:00
ykcheungab 32823a7666 test if optical flow integration error: not significant for Mems 2026-09-01 21:27:10 +08:00
zhangyijian 26168852ef Merge pull request 'fix input flag convention mismatch' (#18) from feature/force_cal into dev
Reviewed-on: #18
2026-08-26 20:27:35 +08:00
ykcheungab 36e0c22de6 fix input flag convention mismatch 2026-08-26 20:26:05 +08:00
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
44 changed files with 2365 additions and 955 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 importlib.util
import shutil
import subprocess
import sys
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]]]:
"""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
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)
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.hhd import nHHD
from orisys.tracker import Tracker
from orisys import capture
from orisys.Stitcher import Stitcher
assert orisys.__version__
assert Sensor is not None
@@ -31,6 +33,8 @@ assert UVSurfaceMesh is not None
assert init_uv_mesh_viewer is not None
assert nHHD 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")
"""
+2 -2
View File
@@ -43,7 +43,7 @@ EXCLUDE_MODULES = [
DATA_DIRS = [
("config", "config"),
("assets", "assets"),
("sensors", "sensors"),
]
README_LINES = [
@@ -56,7 +56,7 @@ README_LINES = [
"",
"Bundled folders:",
" 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).",
]
+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['examples']} example files to orisys/examples/")
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(
NUITKA_WHEEL_DIR,
+3
View File
@@ -16,7 +16,10 @@ from common.paths import repo_path
NUITKA_INCLUDE_PACKAGES = (
"orisys",
"orisys.capture",
"orisys.finger",
"orisys.stitch",
"orisys.stitch.backends",
"orisys.unroll",
"orisys.hhd",
"orisys.hhd.solvers",
+12 -1
View File
@@ -6,12 +6,19 @@ import shutil
from pathlib import Path
# 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 = (
"finger/assets/30fps-0703/mesh.npz",
"finger/assets/30fps-0703/camera_intrinsics.npz",
"finger/assets/30fps-0703/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]:
@@ -61,7 +68,11 @@ def verify_required_assets(repo_root: Path, items: list[dict]) -> None:
continue
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:
joined = ", ".join(missing)
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 文件,文件名格式通常为:
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)
操作系统:Windows(win)
硬件架构:64位(amd64)
@@ -97,7 +97,7 @@ Python 版本:3.10(cp310)
```bash
# 请将文件名替换为您实际下载的文件名
pip install orisys-0.4.3-cp310-cp310-win_amd64.whl
pip install orisys-0.5.0-cp310-cp310-win_amd64.whl
```
### 安装 CuPy(可选)
@@ -139,7 +139,7 @@ python
import orisys
print(orisys.__version__)
```
如果输出版本号(例如 0.4.3),则说明安装成功。
如果输出版本号(例如 0.5.0),则说明安装成功。
## 使用方法
### 运行示例程序
+31 -13
View File
@@ -7,7 +7,7 @@ ORY-BOX:单传感器基础流程
3. 按需计算接触区域
4. 读取并可视化结果
SDK version: 0.4.3
SDK version: 0.5.0
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
@@ -28,12 +28,9 @@ def main():
# cal_path 为标定文件路径;每台设备建议使用独立的标定文件
# 输入既可以是摄像头编号,也可以是视频文件路径
try:
input = int(args.video)
is_camera = True
sensor = orisys.Sensor(input, config_name=args.config, verbose=args.verbose)
except:
is_camera = False
if args.video.isdigit():
sensor = orisys.Sensor(int(args.video), config_name=args.config, verbose=args.verbose)
else:
sensor = orisys.Sensor(args.video, config_name=args.config, verbose=args.verbose)
print("\n按键说明:'q' 退出程序,'r' 重置追踪器。")
@@ -49,8 +46,23 @@ def main():
sensor.compute_contact()
# 步骤 2.4:读取结果
fps, flow, vnormal, img, contour, centroid, depth_map, fnormal, fshearx, fsheary = sensor.read_info(
# 步骤 2.4:读取结果(FNORMAL/FSHEAR* 在有 sensors/box/force_calibration.json 时为标定后牛顿力;
# *_RAW 始终为原始积分量)
(
fps,
flow,
vnormal,
img,
contour,
centroid,
depth_map,
fnormal,
fshearx,
fsheary,
fnormal_raw,
fshearx_raw,
fsheary_raw,
) = sensor.read_info(
sensor.info.FPS,
sensor.info.VRAW,
sensor.info.VNORMAL,
@@ -60,7 +72,10 @@ def main():
sensor.info.DEPTH,
sensor.info.FNORMAL,
sensor.info.FSHEARX,
sensor.info.FSHEARY
sensor.info.FSHEARY,
sensor.info.FNORMAL_RAW,
sensor.info.FSHEARX_RAW,
sensor.info.FSHEARY_RAW,
)
# 步骤 3:可视化显示
@@ -89,10 +104,13 @@ def main():
# 应用伪彩色映射
cv2.imshow("depth", cv2.applyColorMap(div_normalized, cv2.COLORMAP_JET))
# 步骤 3.4:打印关键结果
# 步骤 3.4:打印关键结果(标定力 + 原始力)
print(
f"FPS={fps:.2f}, 法向力={fnormal:.4f}, 切向力X={fshearx:.4f}, "
f"切向力Y={fsheary:.4f}, 深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}"
f"FPS={fps:.2f}, "
f"法向力={fnormal:.4f} N (raw={fnormal_raw:.4f}), "
f"切向力X={fshearx:.4f} N (raw={fshearx_raw:.4f}), "
f"切向力Y={fsheary:.4f} N (raw={fsheary_raw:.4f}), "
f"深度图尺寸={depth_map.shape}, 原始分辨率={sensor.img_size_raw}"
)
# 步骤 3.5:键盘控制
+2 -1
View File
@@ -25,7 +25,7 @@ from orisys.finger import FingerRuntime
def main():
parser = argparse.ArgumentParser(description="Minimum UV-unroll finger tactile pipeline")
parser.add_argument("--camera_id", type=int, default=0)
parser.add_argument("--video", help="Video file path (overrides camera)")
parser.add_argument("--video","-v", help="Video file path (overrides camera)")
parser.add_argument("--config", "-c", default="finger", help="Orisys SDK config name or path")
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")
@@ -90,6 +90,7 @@ def main():
threshold=0.4,
grid_spacing=10,
arrow_scale=1.0,
valid_mask=tactile.valid_mask,
)
cv2.imshow("UV Unrolled", arrows)
+3 -4
View File
@@ -1,12 +1,11 @@
{
"video_source": "D:/repo/orisys-sdk/data/dd02-0703/1.mp4",
"sensor_bundle_dir": "sensors/finger",
"config_path": "sensors/finger/sensor.json",
"config_path": "finger",
"finger_assets_dir": "sensors/finger/assets/30fps-0703",
"csv_save_dir": "./outputs/csv",
"show_log": true,
"show_flow_arrows": false,
"arrow_background": "pure_color",
"arrow_background": "unrolled",
"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 numpy as np
import orisys
from orisys.signal_processing import calibrate_normal_force
from .roi_geometry import (
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
class FrameSnapshot:
"""Normalized per-frame payload consumed by the Qt viewer."""
@@ -46,6 +49,11 @@ class FrameSnapshot:
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):
"""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 numpy as np
import orisys
from orisys.signal_processing import lowpass_filter
from scipy.signal import butter, filtfilt
FORCE_SAMPLE_RATE_HZ = 30.0
PLOT_BUFFER_POINTS = 120
@@ -61,6 +61,32 @@ class ForcePlotState:
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]:
if img is None:
return None
+135 -18
View File
@@ -1,22 +1,139 @@
"""Discover USB cameras for the dashboard.
Thin compatibility shim over ``orisys.capture`` discovery helpers.
"""
"""Discover USB cameras for the dashboard (Windows DSHOW/MSMF, Linux V4L2)."""
from __future__ import annotations
from orisys.capture import (
DEFAULT_MAX_CAMERA_INDEX,
CameraFilter,
CameraInfo,
list_cameras,
pick_default_camera,
)
import contextlib
import os
import sys
from dataclasses import dataclass
__all__ = [
"DEFAULT_MAX_CAMERA_INDEX",
"CameraFilter",
"CameraInfo",
"list_cameras",
"pick_default_camera",
]
import cv2
DEFAULT_MAX_CAMERA_INDEX = 10
@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
+75 -124
View File
@@ -2,19 +2,14 @@
from __future__ import annotations
import json
import math
import sys
import time
from pathlib import Path
from PyQt5.QtCore import Qt, QTimer
from PyQt5.QtGui import QIcon
from PyQt5.QtWidgets import (
QAction,
QActionGroup,
QApplication,
QDockWidget,
QFileDialog,
QMainWindow,
QSplitter,
@@ -66,7 +61,6 @@ class FingerDashboardWindow(QMainWindow):
self._frame_index = 0
self._wall_fps = 0.0
self._wall_fps_alpha = 0.15
self._log_display_mode = "hidden"
self.setWindowTitle(layout.window_title)
self.setStyleSheet(STYLE)
@@ -90,7 +84,7 @@ class FingerDashboardWindow(QMainWindow):
root_layout.setSpacing(layout.root_spacing_px)
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_viewport = MeshViewportWidget(layout=layout, mesh_state=self.mesh_state)
arrows_slot = layout.slot("arrows")
@@ -117,6 +111,7 @@ class FingerDashboardWindow(QMainWindow):
self._preferences = load_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()))
if self.arrow_panel is not None:
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.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)
h_splitter = QSplitter(Qt.Horizontal)
@@ -172,104 +172,18 @@ class FingerDashboardWindow(QMainWindow):
self.h_splitter = 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())
QTimer.singleShot(0, self._apply_panel_layout)
def _set_log_panel_visible(self, visible: bool) -> None:
self._log_panel_visible = bool(visible)
if hasattr(self, "log_dock"):
self.log_dock.setVisible(self._log_panel_visible)
self.log_widget.setVisible(self._log_panel_visible)
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:
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:
self.mesh_viewport.set_free_rotation(enabled)
@@ -282,7 +196,6 @@ class FingerDashboardWindow(QMainWindow):
def _save_preferences(self) -> None:
prefs = self.settings_panel.preference_state()
prefs["log_display_mode"] = self._log_display_mode
self._preferences = prefs
save_preferences(prefs)
# Keep in-memory pipeline config aligned with panel values.
@@ -304,6 +217,7 @@ class FingerDashboardWindow(QMainWindow):
scale = self._layout_scale_factor()
column_height = self._scaled_top_row_height(scale)
charts_slot = self.layout.top_slot("charts")
log_slot = setup.log_panel
for slot in setup.top_row.slots:
if not slot.enabled:
@@ -322,11 +236,27 @@ class FingerDashboardWindow(QMainWindow):
settings_height,
)
self.force_charts.set_panel_size(
self._scaled_px(charts_slot.width_px, scale),
column_height,
)
self.charts_column.setSizes([column_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._scaled_px(charts_slot.width_px, scale),
column_height,
)
self.charts_column.setSizes([column_height, 0])
self.h_splitter.setSizes(
[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()
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()
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_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:
return 1.0
return min(available_width / base_width, available_height / base_height)
@@ -628,6 +570,8 @@ class FingerDashboardWindow(QMainWindow):
camera_id = str(camera_id).strip()
if not camera_id:
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._save_preferences()
self._update_play_enabled_for_source()
@@ -646,41 +590,48 @@ class FingerDashboardWindow(QMainWindow):
self._update_play_enabled_for_source()
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(
self,
"选择配置文件",
"config",
start_dir,
"JSON (*.json);;所有文件 (*)",
)
if path:
self.settings_panel.set_config_path(path)
self.config.config = path
config = normalize_config_preference(path) or default_finger_config()
self.settings_panel.set_config_path(config)
self.config.config = config
self._save_preferences()
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(
self,
"选择传感器 bundle 目录",
self.settings_panel.sensor_bundle_dir() or "sensors/finger",
"选择 finger 资产目录",
start_dir,
)
if path:
bundle_dir = Path(path)
config_path = bundle_dir / "sensor.json"
manifest_path = bundle_dir / "manifest.json"
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)
assets = normalize_assets_preference(path)
self.settings_panel.set_finger_export_dir(assets)
self.config.export_dir = assets
self._save_preferences()
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.
_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:
# 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)
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]:
if not path.is_file():
return {}
@@ -34,14 +133,14 @@ def _read_json(path: Path) -> 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())
if "video_source" not in prefs:
legacy = _read_json(legacy_camera_selection_path())
camera_id = legacy.get("camera_id")
if camera_id is not None and 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:
@@ -49,9 +148,10 @@ def save_preferences(prefs: dict[str, Any]) -> None:
if not isinstance(prefs, dict):
return
path = preference_path()
normalized = _normalize_preferences(prefs)
try:
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:
pass
+31 -1
View File
@@ -25,6 +25,23 @@ class SensorBundlePreset:
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:
path = bundle_root / "manifest.json"
if not path.is_file():
@@ -50,6 +67,14 @@ def _bundle_from_root(bundle_root: Path, *, label: str | None = None) -> SensorB
break
except ValueError:
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(
label=label or "ORY-FINGER",
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:
packaged = _packaged_finger_preset()
if packaged is not None:
return packaged
for root in _repo_root_candidates():
preset = _bundle_from_root(root / "sensors" / "finger", label="ORY-FINGER")
if preset is not None:
@@ -66,7 +94,7 @@ def default_finger_bundle() -> SensorBundlePreset:
return SensorBundlePreset(
label="ORY-FINGER",
bundle_dir="sensors/finger",
config="sensors/finger/sensor.json",
config="finger",
export_dir=DEFAULT_FINGER_EXPORT_DIR,
)
@@ -108,6 +136,8 @@ def sensor_bundle_preset_index(
for idx, preset in enumerate(presets):
if config == preset.config and export_dir == preset.export_dir:
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):
return idx
return -1
+27 -73
View File
@@ -15,15 +15,11 @@ from PyQt5.QtWidgets import (
QWidget,
)
from orisys.finger import DEFAULT_FINGER_EXPORT_DIR
from viz.render_2d import ArrowBackground
from ..layout import DashboardLayout, LAYOUT
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
_PLAY_PAUSE_BUTTON_NAME = "btnPlayPause"
@@ -90,7 +86,7 @@ class SettingsPanelWidget(QWidget):
self.show_log_check = QCheckBox("Show log panel")
self.show_log_check.setChecked(show_log)
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.setChecked(self._mesh_state.flow_arrows_enabled)
@@ -114,6 +110,8 @@ class SettingsPanelWidget(QWidget):
self.camera_combo = QComboBox()
self.camera_combo.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Fixed)
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)
video_row.addWidget(self.camera_combo, 1)
self.btn_refresh_cameras = QPushButton("Refresh")
@@ -138,38 +136,35 @@ class SettingsPanelWidget(QWidget):
self._video_file_path = ""
self._updating_camera_options = False
root.addWidget(self._caption("Sensor bundle"))
bundle_row = QHBoxLayout()
bundle_row.setSpacing(6)
self.sensor_bundle_combo = QComboBox()
self.sensor_bundle_combo.setToolTip("Select a finger sensor bundle")
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()
root.addWidget(self._caption("Config"))
config_row = QHBoxLayout()
config_row.setSpacing(6)
self.config_edit = QLineEdit("finger")
config_row.addWidget(self.config_edit, 1)
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.
self.cal_edit = QLineEdit("")
self.cal_edit.setPlaceholderText("Optional calibration path")
self.cal_edit.hide()
self.finger_assets_edit = QLineEdit("sensors/finger/assets/30fps-0703")
self.finger_assets_edit.hide()
self._apply_bundle_preset(self._bundle_presets[0])
root.addWidget(self._caption("Finger assets"))
finger_assets_row = QHBoxLayout()
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"))
csv_dir_row = QHBoxLayout()
@@ -344,18 +339,16 @@ class SettingsPanelWidget(QWidget):
def set_config_path(self, value: str) -> None:
self.config_edit.setText(value)
self._sync_bundle_combo_from_paths()
def set_calibration_path(self, value: str) -> None:
self.cal_edit.setText(value or "")
def set_finger_export_dir(self, value: str) -> None:
self.finger_assets_edit.setText(value)
self._sync_bundle_combo_from_paths()
def set_finger_export_dir_enabled(self, enabled: bool) -> None:
self.finger_assets_edit.setEnabled(enabled)
self.btn_browse_finger_assets.setEnabled(enabled)
self.sensor_bundle_combo.setEnabled(enabled)
def set_csv_save_dir(self, value: str) -> None:
self.csv_dir_edit.setText(value)
@@ -396,33 +389,6 @@ class SettingsPanelWidget(QWidget):
return
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:
value = self.arrow_background_combo.currentData()
if value:
@@ -449,15 +415,6 @@ class SettingsPanelWidget(QWidget):
def finger_export_dir(self) -> str:
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:
return self.csv_dir_edit.text().strip()
@@ -465,7 +422,6 @@ class SettingsPanelWidget(QWidget):
"""Serialize user-facing settings suitable for preference.json."""
return {
"video_source": self.video_source(),
"sensor_bundle_dir": self.sensor_bundle_dir(),
"config_path": self.config_path(),
"finger_assets_dir": self.finger_export_dir(),
"csv_save_dir": self.csv_save_dir(),
@@ -511,8 +467,6 @@ class SettingsPanelWidget(QWidget):
if isinstance(assets, str) and assets.strip():
self.set_finger_export_dir(assets.strip())
self._sync_bundle_combo_from_paths()
csv_dir = state.get("csv_save_dir")
if isinstance(csv_dir, str) and csv_dir.strip():
self.set_csv_save_dir(csv_dir.strip())
+1 -1
View File
@@ -7,7 +7,7 @@ ORY-MINI:单传感器基础流程
3. 按需计算接触区域
4. 读取并可视化结果
SDK version: 0.4.3
SDK version: 0.5.0
"""
import os
os.environ["OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS"] = "0"
+1 -1
View File
@@ -18,7 +18,7 @@
- Windows 下必须在 if __name__ == '__main__' 中调用 mp.freeze_support()
- 每个传感器进程都会占用独立的 GPU 显存
SDK version: 0.4.3
SDK version: 0.5.0
"""
import os
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
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]:
@@ -63,8 +67,9 @@ def resolve_viz_pose_path(path: str | Path | None = None) -> Path:
here = Path(__file__).resolve()
candidates: list[Path] = []
# Prefer the sensors/finger bundle for the default (and matching relative) path.
if path is None or raw.as_posix() == DEFAULT_FINGER_VIZ_POSE or not raw.is_absolute():
# Prefer the sensors/finger bundle for default / legacy relative paths.
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())
try:
+29 -2
View File
@@ -1,5 +1,32 @@
"""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"
source = "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
dev_install = true
+461
View File
@@ -0,0 +1,461 @@
#!/usr/bin/env python3
"""UART finger controller for Radxa / Dragon Q6A (sensor-side receiver).
Combines ORY-FINGER processing (same path as ``examples/finger_pipeline.py``)
with a line-based UART command interface. A host (or ``uart_send.py``-style
transmitter) sends commands; this script replies and streams forces.
Dependencies (beyond orisys):
pip install pyserial
Default UART (Dragon Q6A):
port=/dev/ttyHS1 baud=115200 8N1
Commands (one line, case-insensitive):
search List cameras (/dev/video*) via orisys.capture.list_cameras
connect [source] Open FingerRuntime (camera index, /dev/videoN, or video file)
run Stream FNORMAL / FSHEARX / FSHEARY until stop (or EOF)
stop Stop force streaming; keep source connected
quit | exit Close runtime and serial, then exit
Replies:
camera,<index>,<path>,<WxH>
OK search <N>
OK connect <source>
OK run
force,<fnormal>,<fshearx>,<fsheary>
OK eof Video file ended (when --no-loop-video)
OK stop
OK quit
ERR <context> <message>
Examples:
# Live camera over UART
python scripts/example_uart.py --port /dev/ttyHS1 --config finger
# Offline debug from a recorded file (no camera needed)
python scripts/example_uart.py --port /dev/ttyHS1 --video /path/to/clip.mp4
# then UART: connect (uses --video) or connect /path/to/clip.mp4
# then UART: run
"""
from __future__ import annotations
import argparse
import os
import shlex
import sys
from pathlib import Path
from typing import Optional, Union
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
try:
import serial
except ImportError as exc: # pragma: no cover
raise SystemExit(
"pyserial is required for this script.\n"
" pip install pyserial\n"
f"Import error: {exc}"
) from exc
import cv2
from orisys.capture import is_live_camera, list_cameras
from orisys.finger import FingerRuntime
Source = Union[int, str]
def _parse_source(token: str) -> Optional[Source]:
"""Accept camera index int, numeric string, V4L path, or filesystem path."""
text = str(token).strip().strip('"').strip("'")
if not text:
return None
if text.isdigit() or (text.startswith("-") and text[1:].isdigit()):
return int(text)
return text
def _is_video_file(source: Source) -> bool:
"""True when source is a filesystem video path (not a live camera)."""
if is_live_camera(source):
return False
if isinstance(source, int):
return False
path = Path(str(source))
return path.is_file() or path.suffix.lower() in {
".mp4",
".avi",
".mkv",
".mov",
".m4v",
".wmv",
".webm",
}
class UartFingerController:
"""Serial command dispatcher around a single FingerRuntime."""
def __init__(
self,
ser: serial.Serial,
*,
config: str = "finger",
default_source: Source = 0,
max_camera_index: int = 10,
loop_video: bool = True,
show: bool = False,
verbose: bool = False,
):
self.ser = ser
self.config = config
self.default_source = default_source
self.max_camera_index = max_camera_index
self.loop_video = loop_video
self.show = show
self.verbose = verbose
self.runtime: Optional[FingerRuntime] = None
self._connected_source: Optional[Source] = None
self._last_search_source: Optional[Source] = None
self._streaming = False
self._should_quit = False
def reply(self, line: str) -> None:
text = line.rstrip("\r\n") + "\n"
self.ser.write(text.encode("utf-8"))
self.ser.flush()
if self.verbose:
print(f"<< {text.rstrip()}", flush=True)
def close_runtime(self) -> None:
self._streaming = False
if self.runtime is not None:
try:
self.runtime.close()
except Exception: # noqa: BLE001
pass
self.runtime = None
self._connected_source = None
if self.show:
try:
cv2.destroyAllWindows()
except Exception: # noqa: BLE001
pass
def _resolve_connect_source(self, args: list[str]) -> Source:
if args:
# Join remaining tokens so unquoted paths with spaces still work.
joined = " ".join(args).strip()
parsed = _parse_source(joined)
if parsed is None:
raise ValueError("empty connect source")
return parsed
# Prefer --video default over last search so offline debug stays on file.
if isinstance(self.default_source, str) and not is_live_camera(self.default_source):
return self.default_source
if self._last_search_source is not None:
return self._last_search_source
return self.default_source
def cmd_search(self, _args: list[str]) -> None:
try:
cameras = list_cameras(max_index=self.max_camera_index)
except Exception as exc: # noqa: BLE001
self.reply(f"ERR search {exc}")
return
self._last_search_source = None
for cam in cameras:
path = (cam.path or f"/dev/video{cam.index}").strip() or f"/dev/video{cam.index}"
size = f"{cam.width}x{cam.height}"
self.reply(f"camera,{cam.index},{path},{size}")
if self._last_search_source is None:
self._last_search_source = cam.index
self.reply(f"OK search {len(cameras)}")
def cmd_connect(self, args: list[str]) -> None:
try:
source = self._resolve_connect_source(args)
except ValueError as exc:
self.reply(f"ERR connect {exc}")
return
if isinstance(source, str) and not is_live_camera(source):
path = Path(source)
if not path.is_file():
self.reply(f"ERR connect file not found: {source}")
return
self._streaming = False
self.close_runtime()
try:
runtime = FingerRuntime.open(
config=self.config,
video=source,
verbose=self.verbose,
)
runtime.warmup(color=False)
except Exception as exc: # noqa: BLE001
self.reply(f"ERR connect {exc}")
return
self.runtime = runtime
self._connected_source = source
kind = "file" if _is_video_file(source) else "camera"
self.reply(f"OK connect {source} ({kind})")
def _rewind_video(self) -> bool:
"""Seek connected file source back to frame 0. Return True on success."""
if self.runtime is None or self._connected_source is None:
return False
if not _is_video_file(self._connected_source):
return False
cap = self.runtime.capture
if cap is None:
return False
try:
return bool(cap.set(cv2.CAP_PROP_POS_FRAMES, 0))
except Exception: # noqa: BLE001
return False
def _poll_command_line(self, *, blocking: bool = True) -> Optional[str]:
"""Read one command line if available; return None on timeout/empty."""
if not blocking and self.ser.in_waiting <= 0:
return None
raw = self.ser.readline()
if not raw:
return None
try:
line = raw.decode("utf-8", errors="replace").strip()
except Exception: # noqa: BLE001
return None
return line or None
def cmd_run(self, _args: list[str]) -> None:
if self.runtime is None:
self.reply("ERR run not connected")
return
self._streaming = True
self.reply("OK run")
tactile = self.runtime.tactile
is_file = self._connected_source is not None and _is_video_file(self._connected_source)
while self._streaming and not self._should_quit:
ok, unrolled = self.runtime.grab_and_process(color=False)
if not ok:
if is_file and self.loop_video and self._rewind_video():
if self.verbose:
print("video EOF: looping", flush=True)
continue
if is_file:
self.reply("OK eof")
else:
self.reply("ERR run grab failed")
self._streaming = False
break
tactile.compute_deformation()
tactile.compute_contact()
fnormal, fshearx, fsheary = tactile.read_info(
tactile.info.FNORMAL,
tactile.info.FSHEARX,
tactile.info.FSHEARY,
)
self.reply(f"force,{float(fnormal):.6f},{float(fshearx):.6f},{float(fsheary):.6f}")
if self.show:
try:
import orisys
arrows = orisys.util.draw_arrows(
unrolled,
tactile.flow,
threshold=0.4,
grid_spacing=10,
arrow_scale=1.0,
valid_mask=tactile.valid_mask,
)
cv2.imshow("UV Unrolled", arrows)
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
self._should_quit = True
break
if key == ord("r"):
tactile.reset()
except Exception as exc: # noqa: BLE001
if self.verbose:
print(f"show error: {exc}", file=sys.stderr)
# Non-blocking command poll so force streaming is not delayed by UART timeout.
pending = self._poll_command_line(blocking=False)
if pending is not None:
if self.verbose:
print(f">> {pending}", flush=True)
if not self._dispatch(pending, allow_run=False):
break
def cmd_stop(self, _args: list[str]) -> None:
self._streaming = False
self.reply("OK stop")
def cmd_quit(self, _args: list[str]) -> None:
self._streaming = False
self._should_quit = True
self.reply("OK quit")
def _dispatch(self, line: str, *, allow_run: bool = True) -> bool:
"""Dispatch one command line. Return False if the main loop should exit."""
try:
parts = shlex.split(line)
except ValueError:
parts = line.split()
if not parts:
return True
cmd = parts[0].lower()
args = parts[1:]
if cmd == "search":
self.cmd_search(args)
elif cmd == "connect":
self.cmd_connect(args)
elif cmd == "run":
if not allow_run:
self.reply("ERR run already running")
else:
self.cmd_run(args)
elif cmd == "stop":
self.cmd_stop(args)
elif cmd in ("quit", "exit"):
self.cmd_quit(args)
return False
else:
self.reply(f"ERR unknown {cmd}")
return not self._should_quit
def serve(self) -> None:
self.reply("OK ready")
while not self._should_quit:
line = self._poll_command_line()
if line is None:
continue
if self.verbose:
print(f">> {line}", flush=True)
if not self._dispatch(line):
break
def open_serial(port: str, baud: int, timeout: float) -> serial.Serial:
try:
return serial.Serial(
port=port,
baudrate=baud,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=timeout,
)
except serial.SerialException as exc:
print(f"Serial error: {exc}", file=sys.stderr)
print("Check:", file=sys.stderr)
print(f" 1. Device node {port} exists", file=sys.stderr)
print(" 2. User has access (sudo or dialout group)", file=sys.stderr)
print(" 3. Wiring / USB-TTL connection", file=sys.stderr)
raise SystemExit(1) from exc
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="UART finger controller (search/connect/run/stop force stream)"
)
parser.add_argument("--port", default="/dev/ttyHS1", help="Serial device (default: /dev/ttyHS1)")
parser.add_argument("--baud", type=int, default=115200, help="Baud rate (default: 115200)")
parser.add_argument(
"--timeout",
type=float,
default=0.1,
help="Serial readline timeout seconds (default: 0.1)",
)
parser.add_argument("--config", "-c", default="finger", help="Orisys config name or path")
parser.add_argument(
"--camera-id",
type=int,
default=0,
help="Default camera index when connect has no argument (ignored if --video is set)",
)
parser.add_argument(
"--video",
default=None,
help="Default video file for connect (offline debug without a camera)",
)
parser.add_argument(
"--max-camera-index",
type=int,
default=10,
help="Highest index probed by search",
)
parser.add_argument(
"--loop-video",
dest="loop_video",
action="store_true",
default=True,
help="Loop video file on EOF during run (default: on)",
)
parser.add_argument(
"--no-loop-video",
dest="loop_video",
action="store_false",
help="Stop run with OK eof when video file ends",
)
parser.add_argument(
"--show",
action="store_true",
help="Show UV unrolled OpenCV window (local debug only)",
)
parser.add_argument("--verbose", action="store_true", help="Echo UART traffic and SDK logs")
args = parser.parse_args(argv)
if args.video:
video_path = Path(args.video)
if not video_path.is_file():
print(f"Video file not found: {args.video}", file=sys.stderr)
return 1
default_source: Source = str(video_path.resolve())
else:
default_source = args.camera_id
ser = open_serial(args.port, args.baud, args.timeout)
controller = UartFingerController(
ser,
config=args.config,
default_source=default_source,
max_camera_index=args.max_camera_index,
loop_video=args.loop_video,
show=args.show,
verbose=args.verbose,
)
print(f"UART finger controller on {args.port} @ {args.baud} bps", flush=True)
print(f"Default source: {default_source}", flush=True)
print("Commands: search | connect [source|file] | run | stop | quit", flush=True)
try:
controller.serve()
except KeyboardInterrupt:
print("\nInterrupted.", flush=True)
try:
controller.reply("OK quit")
except Exception: # noqa: BLE001
pass
finally:
controller.close_runtime()
if ser.is_open:
ser.close()
print("Serial closed.", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+27 -5
View File
@@ -1,8 +1,30 @@
{
"normal_force": {
"model": "linear",
"slope": 3.92685554668795e-05,
"offset": 0.0217596967278535,
"unit": "N"
"schema_version": 1,
"unit": "N",
"channels": {
"normal": {
"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_mode": 1,
"algorithm_list": ["raft", "dis"],
"hhd_downsample": 5
"hhd_downsample": 1
},
"contact": {
"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
from pathlib import Path
from setuptools import setup
from setuptools.command.build_py import build_py as _build_py
REPO_ROOT = Path(__file__).resolve().parent
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.
"""
__version__ = "0.4.3"
__version__ = "0.5.0"
__author__ = "Orisys"
from .sensor import Sensor
+3 -13
View File
@@ -1,14 +1,10 @@
"""Capture subsystem: sources, buffering, reconnect, discovery, multi-camera."""
"""Capture subsystem: OpenCV source, optional background thread, discovery."""
from .buffer import FrameBuffer, FramePacket
from .config import (
MAX_QUEUE_SIZE,
MAX_SIMULTANEOUS_CAPTURES,
CaptureConfig,
CaptureHealth,
CaptureMode,
CaptureStatus,
OverflowPolicy,
ReconnectPolicy,
)
from .discovery import (
@@ -19,31 +15,25 @@ from .discovery import (
list_cameras,
pick_default_camera,
)
from .manager import CaptureManager
from .source import (
CaptureSource,
CaptureWorker,
is_live_camera,
normalize_source,
open_capture,
open_opencv_capture,
)
from .worker import CaptureWorker, open_capture
__all__ = [
"MAX_QUEUE_SIZE",
"MAX_SIMULTANEOUS_CAPTURES",
"DEFAULT_MAX_CAMERA_INDEX",
"DEFAULT_LAPTOP_WEBCAM_PATTERNS",
"CaptureConfig",
"CaptureHealth",
"CaptureMode",
"CaptureStatus",
"OverflowPolicy",
"ReconnectPolicy",
"FrameBuffer",
"FramePacket",
"CaptureSource",
"CaptureWorker",
"CaptureManager",
"CameraInfo",
"CameraFilter",
"list_cameras",
-127
View File
@@ -1,127 +0,0 @@
"""Bounded frame transport for capture workers."""
from __future__ import annotations
import threading
import time
from collections import deque
from dataclasses import dataclass
from typing import Optional
import numpy as np
from .config import CaptureMode, OverflowPolicy
@dataclass
class FramePacket:
"""One captured frame with sequencing metadata."""
frame: np.ndarray
seq: int
timestamp: float
class FrameBuffer:
"""Thread-safe frame buffer supporting latest-only and bounded queue modes.
* ``latest``: always overwrite the single published frame (low latency).
* ``queue``: FIFO up to ``maxsize``; overflow controlled by ``OverflowPolicy``.
"""
def __init__(
self,
*,
mode: CaptureMode = CaptureMode.LATEST,
maxsize: int = 1,
overflow: OverflowPolicy = OverflowPolicy.DROP_OLDEST,
):
self.mode = CaptureMode(mode) if isinstance(mode, str) else mode
self.maxsize = max(1, int(maxsize))
self.overflow = OverflowPolicy(overflow) if isinstance(overflow, str) else overflow
self._lock = threading.Lock()
self._cond = threading.Condition(self._lock)
self._latest: Optional[FramePacket] = None
self._queue: deque[FramePacket] = deque()
self.dropped_frames = 0
self._seq = 0
def clear(self) -> None:
with self._cond:
self._latest = None
self._queue.clear()
def publish(self, frame: np.ndarray, *, timestamp: Optional[float] = None) -> int:
"""Publish a frame. Returns assigned sequence number."""
ts = time.perf_counter() if timestamp is None else float(timestamp)
with self._cond:
self._seq += 1
packet = FramePacket(frame=frame, seq=self._seq, timestamp=ts)
if self.mode == CaptureMode.LATEST:
self._latest = packet
self._cond.notify_all()
return packet.seq
if len(self._queue) >= self.maxsize:
if self.overflow == OverflowPolicy.DROP_NEWEST:
self.dropped_frames += 1
return packet.seq
if self.overflow == OverflowPolicy.BLOCK:
while len(self._queue) >= self.maxsize:
self._cond.wait(timeout=0.05)
# If still full after wait, drop oldest to avoid deadlock if
# consumer stopped; count as dropped.
if len(self._queue) >= self.maxsize:
self._queue.popleft()
self.dropped_frames += 1
break
else:
# drop_oldest
self._queue.popleft()
self.dropped_frames += 1
self._queue.append(packet)
self._latest = packet
self._cond.notify_all()
return packet.seq
def read_latest(self, *, copy: bool = True, wait_s: float = 0.0) -> tuple[bool, Optional[np.ndarray], int]:
"""Return the newest frame without consuming a queue slot."""
deadline = time.perf_counter() + max(0.0, wait_s)
with self._cond:
while self._latest is None:
remaining = deadline - time.perf_counter()
if remaining <= 0:
return False, None, 0
self._cond.wait(timeout=remaining)
packet = self._latest
frame = packet.frame.copy() if copy else packet.frame
return True, frame, packet.seq
def read_next(self, *, copy: bool = True, wait_s: float = 0.0) -> tuple[bool, Optional[np.ndarray], int]:
"""Pop the next queued frame (queue mode) or return latest (latest mode)."""
if self.mode == CaptureMode.LATEST:
return self.read_latest(copy=copy, wait_s=wait_s)
deadline = time.perf_counter() + max(0.0, wait_s)
with self._cond:
while not self._queue:
remaining = deadline - time.perf_counter()
if remaining <= 0:
return False, None, 0
self._cond.wait(timeout=remaining)
packet = self._queue.popleft()
self._cond.notify_all()
frame = packet.frame.copy() if copy else packet.frame
return True, frame, packet.seq
@property
def last_seq(self) -> int:
with self._lock:
return self._seq
@property
def pending(self) -> int:
with self._lock:
if self.mode == CaptureMode.LATEST:
return 0 if self._latest is None else 1
return len(self._queue)
+7 -34
View File
@@ -7,23 +7,6 @@ from enum import Enum
from typing import Any, Optional, Sequence, Tuple, Union
MAX_SIMULTANEOUS_CAPTURES = 10
MAX_QUEUE_SIZE = 10
DEFAULT_QUEUE_SIZE = 1
class CaptureMode(str, Enum):
"""Frame delivery mode for a live capture worker."""
LATEST = "latest"
QUEUE = "queue"
class OverflowPolicy(str, Enum):
"""Behavior when a bounded queue is full."""
DROP_OLDEST = "drop_oldest"
DROP_NEWEST = "drop_newest"
BLOCK = "block"
class CaptureHealth(str, Enum):
@@ -83,11 +66,8 @@ class CaptureConfig:
height: Optional[int] = None
fps: Optional[float] = None
fourcc: Optional[str] = "MJPG"
mode: CaptureMode = CaptureMode.LATEST
queue_size: int = DEFAULT_QUEUE_SIZE
overflow: OverflowPolicy = OverflowPolicy.DROP_OLDEST
background: bool = True
reconnect: ReconnectPolicy = field(default_factory=ReconnectPolicy)
start_worker: bool = True
verbose: bool = False
# Optional discovery filters (used by list/open helpers, not open itself).
name_include: Optional[Sequence[str]] = None
@@ -97,19 +77,8 @@ class CaptureConfig:
min_height: Optional[int] = None
def __post_init__(self) -> None:
if isinstance(self.mode, str):
self.mode = CaptureMode(self.mode)
if isinstance(self.overflow, str):
self.overflow = OverflowPolicy(self.overflow)
if not isinstance(self.reconnect, ReconnectPolicy):
raise TypeError("reconnect must be a ReconnectPolicy")
size = int(self.queue_size)
if size < 1 or size > MAX_QUEUE_SIZE:
raise ValueError(f"queue_size must be in 1..{MAX_QUEUE_SIZE}, got {size}")
self.queue_size = size
if self.mode == CaptureMode.LATEST:
# Latest mode keeps only one published frame regardless of queue_size.
self.queue_size = 1
@classmethod
def from_camera_dict(
@@ -117,12 +86,17 @@ class CaptureConfig:
source: SourceType,
camera: dict[str, Any],
*,
mode: CaptureMode | str = CaptureMode.LATEST,
reconnect: Optional[ReconnectPolicy] = None,
verbose: bool = False,
**kwargs: Any,
) -> "CaptureConfig":
"""Build config from a sensor JSON ``camera`` section."""
# Accept deprecated queue/worker kwargs from older callers without error.
kwargs.pop("mode", None)
kwargs.pop("queue_size", None)
kwargs.pop("overflow", None)
if "start_worker" in kwargs:
kwargs.setdefault("background", bool(kwargs.pop("start_worker")))
res = camera.get("resolution_raw") or [None, None]
width = int(res[0]) if res and res[0] else None
height = int(res[1]) if res and res[1] else None
@@ -132,7 +106,6 @@ class CaptureConfig:
width=width,
height=height,
fps=float(fps) if fps is not None else None,
mode=mode,
reconnect=reconnect or ReconnectPolicy(),
verbose=verbose,
**kwargs,
-122
View File
@@ -1,122 +0,0 @@
"""Manage up to N simultaneous capture sources."""
from __future__ import annotations
from typing import Dict, List, Optional, Sequence
from .config import (
MAX_SIMULTANEOUS_CAPTURES,
CaptureConfig,
CaptureStatus,
SourceType,
)
from .source import normalize_source
from .worker import CaptureWorker, open_capture
class CaptureManager:
"""Own multiple ``CaptureWorker`` instances (default max 10).
Each camera is isolated: one failed USB device does not stop the others.
Hardware USB bandwidth may still limit how many high-res streams work in
practice even when this manager allows up to ``max_sources`` software slots.
"""
def __init__(self, *, max_sources: int = MAX_SIMULTANEOUS_CAPTURES):
if max_sources < 1 or max_sources > MAX_SIMULTANEOUS_CAPTURES:
raise ValueError(
f"max_sources must be in 1..{MAX_SIMULTANEOUS_CAPTURES}, got {max_sources}"
)
self.max_sources = int(max_sources)
self._workers: Dict[str, CaptureWorker] = {}
def __len__(self) -> int:
return len(self._workers)
def __contains__(self, key: SourceType) -> bool:
return self._key(key) in self._workers
@staticmethod
def _key(source: SourceType) -> str:
return str(normalize_source(source))
@property
def sources(self) -> List[str]:
return list(self._workers.keys())
def open(self, config: CaptureConfig, *, factory=None) -> CaptureWorker:
"""Open one capture; raises if capacity exceeded or source already open."""
key = self._key(config.source)
if key in self._workers:
raise RuntimeError(f"Capture already open for source {key!r}")
if len(self._workers) >= self.max_sources:
raise RuntimeError(
f"CaptureManager at capacity ({self.max_sources}). "
"Close a source before opening another."
)
worker = open_capture(config, factory=factory)
self._workers[key] = worker
return worker
def open_many(
self,
configs: Sequence[CaptureConfig],
*,
factory=None,
) -> List[CaptureWorker]:
"""Open several captures atomically (rolls back on failure)."""
if len(configs) + len(self._workers) > self.max_sources:
raise RuntimeError(
f"Opening {len(configs)} sources would exceed capacity "
f"{self.max_sources} (currently {len(self._workers)} open)."
)
opened: List[CaptureWorker] = []
try:
for cfg in configs:
opened.append(self.open(cfg, factory=factory))
except Exception:
for worker in opened:
key = self._key(worker.source.config.source)
worker.close()
self._workers.pop(key, None)
raise
return opened
def get(self, source: SourceType) -> Optional[CaptureWorker]:
return self._workers.get(self._key(source))
def read_latest(self, source: SourceType, *, wait_s: float = 0.0):
worker = self.get(source)
if worker is None:
return False, None
return worker.buffer.read_latest(wait_s=wait_s)[:2]
def read_latest_all(self, *, wait_s: float = 0.0) -> Dict[str, tuple]:
"""Return ``{source_key: (ok, frame)}`` for every open capture."""
out: Dict[str, tuple] = {}
for key, worker in self._workers.items():
ok, frame, _seq = worker.buffer.read_latest(wait_s=wait_s)
out[key] = (ok, frame)
return out
def status_all(self) -> Dict[str, CaptureStatus]:
return {key: worker.source.status for key, worker in self._workers.items()}
def close(self, source: SourceType) -> None:
key = self._key(source)
worker = self._workers.pop(key, None)
if worker is not None:
worker.close()
def close_all(self) -> None:
keys = list(self._workers.keys())
for key in keys:
worker = self._workers.pop(key, None)
if worker is not None:
worker.close()
def __enter__(self) -> "CaptureManager":
return self
def __exit__(self, exc_type, exc, tb) -> None:
self.close_all()
+225 -22
View File
@@ -1,9 +1,10 @@
"""OpenCV-backed capture source with optional reconnect."""
"""OpenCV-backed capture source with optional background thread and reconnect."""
from __future__ import annotations
import os
import sys
import threading
import time
from typing import Any, Optional, Protocol, runtime_checkable
@@ -28,6 +29,7 @@ class VideoCaptureLike(Protocol):
def release(self) -> None: ...
def get(self, propId: int) -> float: ...
def set(self, propId: int, value: float) -> bool: ...
def grab(self) -> bool: ...
def normalize_source(source: SourceType | None) -> SourceType:
@@ -40,7 +42,6 @@ def normalize_source(source: SourceType | None) -> SourceType:
return 0
if stripped.isdigit():
return int(stripped)
# Only strip trailing separators (Windows path quirk); keep leading '/'.
return stripped.rstrip("/\\") or stripped
return source
@@ -69,12 +70,10 @@ def _backend_candidates(preferred: str = "auto") -> list[tuple[str, int]]:
elif preferred in ("v4l2", "v4l") and sys.platform.startswith("linux"):
add("V4L2", "CAP_V4L2")
elif preferred not in ("auto", "any", ""):
# Unknown preference: fall through to platform defaults.
pass
if not candidates:
if sys.platform == "win32":
# Prefer DSHOW to avoid MSMF HW-transform stalls on many USB cameras.
add("DSHOW", "CAP_DSHOW")
add("MSMF", "CAP_MSMF")
elif sys.platform.startswith("linux"):
@@ -101,6 +100,45 @@ def _decode_fourcc(value: float) -> str:
return "".join(chr(b) if 32 <= b <= 126 else "." for b in bytes_)
def _read_stream_meta(cap: VideoCaptureLike) -> tuple[int, int, float, str]:
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0)
fourcc = _decode_fourcc(cap.get(cv2.CAP_PROP_FOURCC) or 0)
return width, height, fps, fourcc
def _apply_stream_props(
cap: VideoCaptureLike,
*,
width: Optional[int],
height: Optional[int],
fps: Optional[float],
fourcc: Optional[str],
) -> None:
"""Apply live-camera stream properties (width/height before fourcc on DSHOW)."""
code = _fourcc_code(fourcc)
if width is not None:
cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width))
if height is not None:
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height))
if code is not None:
cap.set(cv2.CAP_PROP_FOURCC, code)
if fps is not None:
cap.set(cv2.CAP_PROP_FPS, float(fps))
# DSHOW often resets fourcc after size negotiation.
if code is not None:
cap.set(cv2.CAP_PROP_FOURCC, code)
def _warmup_capture(cap: VideoCaptureLike, *, grabs: int = 3) -> None:
for _ in range(max(0, grabs)):
try:
cap.grab()
except Exception:
break
def open_opencv_capture(
source: SourceType,
*,
@@ -115,13 +153,13 @@ def open_opencv_capture(
Returns ``(capture, backend_name)``.
"""
# Avoid MSMF HW transform stalls on Windows camera capture.
if sys.platform == "win32":
os.environ.setdefault("OPENCV_VIDEOIO_MSMF_ENABLE_HW_TRANSFORMS", "0")
source = normalize_source(source)
live = is_live_camera(source)
last_error = "Failed to open video source."
requested_fourcc = (fourcc or "").strip().upper()
backends = _backend_candidates(backend) if live else [("ANY", cv2.CAP_ANY)]
for backend_name, backend_api in backends:
@@ -138,22 +176,28 @@ def open_opencv_capture(
last_error = f"Backend {backend_name} failed to open {source!r}"
continue
if live and apply_stream_props:
code = _fourcc_code(fourcc)
if code is not None:
cap.set(cv2.CAP_PROP_FOURCC, code)
if fps is not None:
cap.set(cv2.CAP_PROP_FPS, float(fps))
if width is not None:
cap.set(cv2.CAP_PROP_FRAME_WIDTH, float(width))
if height is not None:
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, float(height))
_apply_stream_props(
cap,
width=width,
height=height,
fps=fps,
fourcc=fourcc,
)
_warmup_capture(cap)
_, _, _, negotiated = _read_stream_meta(cap)
if requested_fourcc and negotiated.upper() != requested_fourcc:
# Retry once: some drivers need a second fourcc set after warmup.
code = _fourcc_code(fourcc)
if code is not None:
cap.set(cv2.CAP_PROP_FOURCC, code)
_warmup_capture(cap, grabs=2)
return cap, backend_name
raise RuntimeError(last_error)
class CaptureSource:
"""Owns a video capture handle, health state, and reconnect logic.
"""Owns a video capture handle, optional background reader, and reconnect logic.
Optionally accepts an injected ``factory`` for tests::
@@ -172,7 +216,13 @@ class CaptureSource:
self._backend_name = ""
self._status = CaptureStatus(source=normalize_source(config.source))
self._reconnect_attempt = 0
self._lock = __import__("threading").Lock()
self._lock = threading.Lock()
self._frame_lock = threading.Lock()
self._frame_cond = threading.Condition(self._frame_lock)
self._latest_frame: Optional[np.ndarray] = None
self._latest_seq = 0
self._thread: Optional[threading.Thread] = None
self._background_running = False
@staticmethod
def _default_factory(config: CaptureConfig) -> tuple[VideoCaptureLike, str]:
@@ -186,6 +236,15 @@ class CaptureSource:
apply_stream_props=is_live_camera(normalize_source(config.source)),
)
# Compatibility with older CaptureWorker API.
@property
def source(self) -> "CaptureSource":
return self
@property
def running(self) -> bool:
return self._background_running
@property
def status(self) -> CaptureStatus:
with self._lock:
@@ -240,12 +299,27 @@ class CaptureSource:
f"OpenCV negotiated stream: FOURCC='{st.fourcc}', "
f"size={st.width}x{st.height}, fps={st.fps:.2f}, backend={st.backend_name}"
)
requested = (self.config.fourcc or "").strip().upper()
if (
requested
and is_live_camera(normalize_source(self.config.source))
and st.fourcc.upper() != requested
):
print(
f"Warning: requested FOURCC '{self.config.fourcc}' but camera "
f"negotiated '{st.fourcc}'."
)
def release(self) -> None:
self.stop_background()
with self._lock:
self._release_unlocked()
self._status.health = CaptureHealth.CLOSED
def close(self) -> None:
"""Stop background thread and release the capture handle."""
self.release()
def _release_unlocked(self) -> None:
if self._cap is not None:
try:
@@ -253,21 +327,32 @@ class CaptureSource:
except Exception:
pass
self._cap = None
with self._frame_cond:
self._latest_frame = None
self._latest_seq = 0
def _refresh_stream_meta_unlocked(self) -> None:
cap = self._cap
if cap is None:
return
try:
self._status.width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
self._status.height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
self._status.fps = float(cap.get(cv2.CAP_PROP_FPS) or 0.0)
self._status.fourcc = _decode_fourcc(cap.get(cv2.CAP_PROP_FOURCC) or 0)
width, height, fps, fourcc = _read_stream_meta(cap)
self._status.width = width
self._status.height = height
self._status.fps = fps
self._status.fourcc = fourcc
except Exception:
pass
def warmup_frames(self, count: int = 3) -> None:
"""Discard frames so the driver/format stabilizes (sync reads only)."""
if self._background_running:
return
for _ in range(max(0, int(count))):
self.read_raw()
def read_raw(self) -> tuple[bool, Optional[np.ndarray]]:
"""Read one frame from the underlying capture (no reconnect)."""
"""Read one frame directly from the underlying capture (no reconnect)."""
cap = self._cap
if cap is None:
return False, None
@@ -289,6 +374,92 @@ class CaptureSource:
self._status.health = CaptureHealth.CONNECTED
return True, frame
def read_latest(
self,
*,
copy: bool = True,
wait_s: float = 0.0,
) -> tuple[bool, Optional[np.ndarray]]:
"""Return the newest frame from the background thread, or read sync."""
if not self._background_running:
ok, frame = self.read_raw()
if ok and copy and frame is not None:
return ok, frame.copy()
return ok, frame
deadline = time.perf_counter() + max(0.0, wait_s)
with self._frame_cond:
while self._latest_frame is None:
remaining = deadline - time.perf_counter()
if remaining <= 0:
return False, None
self._frame_cond.wait(timeout=remaining)
frame = self._latest_frame.copy() if copy else self._latest_frame
return True, frame
def read(self, *, copy: bool = True, wait_s: float = 1.0) -> tuple[bool, Optional[np.ndarray]]:
"""Compat alias used by finger runtime and older worker callers."""
wait = wait_s if wait_s > 0 else 1.0
return self.read_latest(copy=copy, wait_s=wait)
def start_background(self) -> None:
"""Start a daemon thread that keeps only the latest frame."""
if self._thread is not None:
return
if not self.is_open():
self.open()
self._background_running = True
self._thread = threading.Thread(
target=self._background_loop,
daemon=True,
name="orisys-capture",
)
self._thread.start()
if self.config.verbose:
print("Capture background thread started.")
def start(self) -> None:
"""Compat alias for ``start_background``."""
self.start_background()
def stop_background(self, timeout: float = 1.0) -> None:
self._background_running = False
if self._thread is not None:
self._thread.join(timeout=timeout)
self._thread = None
with self._frame_cond:
self._latest_frame = None
def stop(self, timeout: float = 1.0) -> None:
"""Compat alias for ``stop_background``."""
self.stop_background(timeout=timeout)
def _background_loop(self) -> None:
while self._background_running:
if self.should_reconnect():
if not self.reconnect():
time.sleep(0.05)
if self.status.health == CaptureHealth.FAILED:
time.sleep(0.2)
continue
ok, frame = self.read_raw()
if not ok or frame is None:
if self.should_reconnect():
continue
time.sleep(0.001)
continue
with self._frame_cond:
if self._latest_frame is not None:
with self._lock:
self._status.dropped_frames += 1
self._latest_seq += 1
self._latest_frame = frame
with self._lock:
self._status.last_seq = self._latest_seq
self._frame_cond.notify_all()
def note_published(self, seq: int) -> None:
with self._lock:
self._status.last_seq = int(seq)
@@ -334,7 +505,10 @@ class CaptureSource:
delay = policy.next_backoff(attempt)
if self.config.verbose:
print(f"Reconnecting to {self.config.source!r} (attempt {attempt + 1}, wait {delay:.2f}s)...")
print(
f"Reconnecting to {self.config.source!r} "
f"(attempt {attempt + 1}, wait {delay:.2f}s)..."
)
time.sleep(delay)
try:
self.open()
@@ -347,3 +521,32 @@ class CaptureSource:
else:
self._status.health = CaptureHealth.RECONNECTING
return False
# Backward-compatible alias for one release cycle.
CaptureWorker = CaptureSource
def open_capture(
config: CaptureConfig,
*,
factory=None,
start_background: Optional[bool] = None,
start_worker: Optional[bool] = None,
) -> CaptureSource:
"""Open a capture source and optionally start its background thread.
For live cameras, the background thread starts by default (``config.background``).
File/video sources stay synchronous unless ``start_background=True``.
"""
source = CaptureSource(config, factory=factory)
source.open()
should_start = config.background
if start_background is not None:
should_start = start_background
elif start_worker is not None:
should_start = start_worker
live = is_live_camera(normalize_source(config.source))
if should_start and live:
source.start_background()
return source
-124
View File
@@ -1,124 +0,0 @@
"""Background capture worker that feeds a FrameBuffer."""
from __future__ import annotations
import threading
import time
from typing import Optional
from .buffer import FrameBuffer
from .config import CaptureConfig, CaptureHealth, CaptureMode
from .source import CaptureSource, is_live_camera, normalize_source
class CaptureWorker:
"""One thread per live camera: read → publish → optional reconnect."""
def __init__(
self,
source: CaptureSource,
buffer: Optional[FrameBuffer] = None,
):
self.source = source
self.config: CaptureConfig = source.config
self.buffer = buffer or FrameBuffer(
mode=self.config.mode,
maxsize=self.config.queue_size,
overflow=self.config.overflow,
)
self._thread: Optional[threading.Thread] = None
self._running = False
self._last_consumed_seq = -1
@property
def running(self) -> bool:
return self._running
def start(self) -> None:
if self._thread is not None:
return
if not self.source.is_open():
self.source.open()
self._running = True
self._thread = threading.Thread(target=self._loop, daemon=True, name="orisys-capture")
self._thread.start()
if self.config.verbose:
mode = self.config.mode.value if isinstance(self.config.mode, CaptureMode) else self.config.mode
print(f"Capture worker started (mode={mode}).")
def stop(self, timeout: float = 1.0) -> None:
self._running = False
if self._thread is not None:
self._thread.join(timeout=timeout)
self._thread = None
self.buffer.clear()
def close(self) -> None:
self.stop()
self.source.release()
def read(self, *, copy: bool = True, wait_s: float = 1.0) -> tuple[bool, object]:
"""Read according to configured mode; mirrors Sensor._read_frame semantics."""
if not self._running:
ok, frame = self.source.read_raw()
return ok, frame
if self.config.mode == CaptureMode.QUEUE:
ok, frame, seq = self.buffer.read_next(copy=copy, wait_s=wait_s)
else:
# Startup race: wait briefly for first frame.
wait = wait_s if wait_s > 0 else 1.0
ok, frame, seq = self.buffer.read_latest(copy=copy, wait_s=wait)
if ok:
self._last_consumed_seq = seq
return ok, frame
def _loop(self) -> None:
while self._running:
if self.source.should_reconnect():
if not self.source.reconnect():
# Back off a bit before retrying the should_reconnect check.
time.sleep(0.05)
if self.source.status.health == CaptureHealth.FAILED:
time.sleep(0.2)
continue
ok, frame = self.source.read_raw()
if not ok or frame is None:
if self.source.should_reconnect():
continue
time.sleep(0.001)
continue
before_dropped = self.buffer.dropped_frames
seq = self.buffer.publish(frame)
after_dropped = self.buffer.dropped_frames
if after_dropped > before_dropped:
self.source.note_dropped(after_dropped - before_dropped)
self.source.note_published(seq)
def open_capture(
config: CaptureConfig,
*,
factory=None,
start_worker: Optional[bool] = None,
) -> CaptureWorker:
"""Open a capture source and optionally start its worker thread.
For live cameras, the worker starts by default (``config.start_worker``).
File/video sources stay synchronous unless ``start_worker=True``.
"""
source = CaptureSource(config, factory=factory)
source.open()
buffer = FrameBuffer(
mode=config.mode,
maxsize=config.queue_size,
overflow=config.overflow,
)
worker = CaptureWorker(source, buffer)
should_start = config.start_worker if start_worker is None else start_worker
live = is_live_camera(normalize_source(config.source))
if should_start and live:
worker.start()
return worker
+25 -26
View File
@@ -6,7 +6,7 @@ import numpy as np
from ..capture import (
CaptureConfig,
CaptureWorker,
CaptureSource,
ReconnectPolicy,
is_live_camera,
normalize_source,
@@ -35,23 +35,21 @@ def normalize_video_source(video):
def open_source(video, config, *, auto_reconnect: bool = False, verbose: bool = False):
"""Open a capture worker for finger pipelines (compat: returns worker.cap-like via ``.cap``).
"""Open a capture source for finger pipelines (compat: returns ``.cap``-like handle).
Historically returned a bare ``cv2.VideoCapture``. Callers that need the raw
OpenCV handle can use ``worker.source.cap``; prefer ``worker.read()``.
OpenCV handle can use ``source.cap``; prefer ``source.read()``.
"""
capture_config = CaptureConfig.from_camera_dict(
normalize_video_source(video),
config.get("camera", {}),
reconnect=ReconnectPolicy(enabled=auto_reconnect),
verbose=verbose,
start_worker=False,
background=False,
)
worker = open_capture(capture_config, start_worker=False)
# Preserve prior attribute access: treat returned object like VideoCapture for
# simple .read()/.release()/.isOpened() callers, while exposing capture worker.
cap = worker.source.cap
cap._orisys_worker = worker # type: ignore[attr-defined]
source = open_capture(capture_config, start_background=False)
cap = source.cap
cap._orisys_worker = source # type: ignore[attr-defined]
return cap
@@ -220,7 +218,7 @@ class FingerRuntime:
tactile: TactileProcessor,
frame_processor: FrameUnwarper,
capture,
capture_worker: CaptureWorker | None = None,
capture_source: CaptureSource | None = None,
):
self.geometry = geometry
self.domain = domain
@@ -228,7 +226,9 @@ class FingerRuntime:
self.sensor = tactile
self.frame_processor = frame_processor
self.capture = capture
self.capture_worker = capture_worker
self.capture_source = capture_source
# Deprecated alias.
self.capture_worker = capture_source
@classmethod
def open(
@@ -248,6 +248,7 @@ class FingerRuntime:
capture_mode: str = "latest",
) -> "FingerRuntime":
"""Open a live finger runtime from export bundle, config, and video source."""
del capture_mode # deprecated; latest-frame capture only
if camera_id is not None:
video = camera_id
video = normalize_video_source(video)
@@ -280,19 +281,16 @@ class FingerRuntime:
capture_config = CaptureConfig.from_camera_dict(
video,
geometry.config.get("camera", {}),
mode=capture_mode,
reconnect=ReconnectPolicy(enabled=auto_reconnect),
verbose=verbose,
start_worker=False,
background=False,
)
capture_worker = open_capture(capture_config, start_worker=False)
if not capture_worker.source.is_open():
capture_source = open_capture(capture_config, start_background=False)
if not capture_source.is_open():
raise RuntimeError("Cannot open camera/video source")
# Start latest-frame worker for live cameras so processing speed does not
# stall the device buffer.
if is_live_camera(video):
capture_worker.start()
capture = capture_worker.source.cap
capture_source.start_background()
capture = capture_source.cap
sensor = Sensor("external", config_name=config, verbose=verbose, mirror=mirror)
frame_processor = FrameUnwarper(
@@ -310,7 +308,7 @@ class FingerRuntime:
tactile=tactile,
frame_processor=frame_processor,
capture=capture,
capture_worker=capture_worker,
capture_source=capture_source,
)
@property
@@ -318,10 +316,10 @@ class FingerRuntime:
return self.frame_processor.valid_mask
def read_source(self):
if self.capture_worker is not None and self.capture_worker.running:
return self.capture_worker.read(copy=True, wait_s=1.0)
if self.capture_worker is not None:
return self.capture_worker.source.read_raw()
if self.capture_source is not None and self.capture_source.running:
return self.capture_source.read(copy=True, wait_s=1.0)
if self.capture_source is not None:
return self.capture_source.read_raw()
return self.capture.read()
def process_frame(self, frame, *, color=False):
@@ -355,8 +353,9 @@ class FingerRuntime:
return ok, unrolled
def close(self):
if self.capture_worker is not None:
self.capture_worker.close()
if self.capture_source is not None:
self.capture_source.close()
self.capture_source = None
self.capture_worker = None
self.capture = None
elif self.capture is not None:
+111 -38
View File
@@ -11,6 +11,7 @@ import numpy as np
import cv2
import time
from enum import Enum, auto
from pathlib import Path
from typing import Optional, Tuple
# from threading import Thread, Event
# from queue import Queue, Empty
@@ -22,10 +23,15 @@ from . import configs
from .get_depth import Contact3D
from .backend_availability import (check_cupy_available, check_opencl_available)
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 (
CaptureConfig,
CaptureWorker,
CaptureSource,
ReconnectPolicy,
is_live_camera,
normalize_source,
@@ -70,7 +76,8 @@ class Sensor:
config_name="prototype", cal_path=None, override=False, box=None, mirror=True,
backend='auto', capture_config: Optional[CaptureConfig] = None,
auto_reconnect: bool = False, capture_mode: str = "latest",
capture_queue_size: int = 1, capture_overflow: str = "drop_oldest"):
capture_queue_size: int = 1, capture_overflow: str = "drop_oldest",
**_deprecated_capture_kwargs):
"""
Initialize Sensor with video source and configuration.
@@ -90,9 +97,9 @@ class Sensor:
"cpu": Use CPU for computation
capture_config: Optional ``CaptureConfig`` overriding capture kwargs below.
auto_reconnect: Enable USB disconnect auto-reconnect for live cameras.
capture_mode: ``"latest"`` (low latency, default) or ``"queue"``.
capture_queue_size: Bounded queue size for queue mode (1..10).
capture_overflow: ``drop_oldest`` | ``drop_newest`` | ``block``.
capture_mode: Deprecated; ignored (latest-frame capture only).
capture_queue_size: Deprecated; ignored.
capture_overflow: Deprecated; ignored.
"""
self.CUPY_AVAILABLE = check_cupy_available()
@@ -114,9 +121,6 @@ class Sensor:
self.mirror = mirror
self._capture_config_override = capture_config
self._auto_reconnect = bool(auto_reconnect)
self._capture_mode = capture_mode
self._capture_queue_size = int(capture_queue_size)
self._capture_overflow = capture_overflow
# Load configuration
self.__config_cam = None
@@ -148,6 +152,7 @@ class Sensor:
self.n_boxes = None
self.fnormal = self.fshearx = self.fsheary = self.ftor = 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_cutoff_hz = 0.0
self._lowpass_sample_rate_hz = float(DEFAULT_SAMPLE_RATE_HZ)
@@ -158,6 +163,13 @@ class Sensor:
order=self._lowpass_order,
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)
self.centroid = (0, 0)
@@ -217,7 +229,7 @@ class Sensor:
# self._av_stream = None
# self._av_decoder = None
self.cap = None # compat alias for _capture_worker.source.cap
self._capture_worker: Optional[CaptureWorker] = None
self._capture_source: Optional[CaptureSource] = None
self._is_live_camera = (
is_live_camera(self.vid_src) and not self._external_only
)
@@ -691,9 +703,9 @@ class Sensor:
def disconnect(self):
"""Release resources and stop async threads."""
if self._capture_worker is not None:
self._capture_worker.close()
self._capture_worker = None
if self._capture_source is not None:
self._capture_source.close()
self._capture_source = None
self.cap = None
if self.verbose:
print("Sensor disconnected.")
@@ -790,17 +802,58 @@ class Sensor:
self._update_hhd_mask_cache()
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):
raw = np.asarray(self._force_raw, dtype=np.float64)
if self._lowpass_enabled and self._lowpass_cutoff_hz > 0:
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:
self.fnormal, self.fshearx, self.fsheary = (
float(raw[0]),
float(raw[1]),
float(raw[2]),
)
fn, fx, fy = float(raw[0]), float(raw[1]), float(raw[2])
# Apply sensors/<name>/force_calibration.json when present; otherwise keep raw.
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"):
# """Return current configuration dict."""
@@ -879,25 +932,23 @@ class Sensor:
capture_config = CaptureConfig.from_camera_dict(
self.vid_src,
self.__config_cam,
mode=self._capture_mode,
queue_size=self._capture_queue_size,
overflow=self._capture_overflow,
reconnect=ReconnectPolicy(enabled=self._auto_reconnect),
verbose=self.verbose,
start_worker=False,
background=False,
)
try:
self._capture_worker = open_capture(capture_config, start_worker=False)
self._capture_source = open_capture(capture_config, start_background=False)
except Exception as exc:
raise Exception("Failed to open video source.") from exc
self.cap = self._capture_worker.source.cap
# self._use_pyav = False
self.cap = self._capture_source.cap
if not self._is_source_open():
raise Exception("Failed to open video source.")
self._capture_source.warmup_frames(3)
# Wait for first valid frame
frame = None
while frame is None:
@@ -912,16 +963,24 @@ class Sensor:
if box is not None:
self.stitcher.isstitch = self.isstitch = False
self.stitcher.boxes = box
self.stitch_valid = bool(self.isstitch)
img = self.frame_processor.process(frame)
self.initial_img = self._make_initial_gray(img)
else:
print("Override=True but no box provided.")
else:
self._find_stitch_boxes(frame)
if not self.stitch_valid:
raise RuntimeError(
f"Failed to detect {self._config_units} vision tiles for stitching. "
"Check camera index, lighting, and that MJPG 1080p is negotiated."
)
img = self.frame_processor.process(frame)
self.initial_img = self._make_initial_gray(img)
# For live cameras, start a capture worker to avoid latency buildup.
if self._is_live_camera and self._capture_worker is not None:
self._capture_worker.start()
# For live cameras, start background capture to avoid latency buildup.
if self._is_live_camera and self._capture_source is not None:
self._capture_source.start_background()
def _make_initial_gray(self, img_bgr):
gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY)
@@ -935,6 +994,9 @@ class Sensor:
def _find_stitch_boxes(self, img_src, show=False):
"""Detect vision tile boxes with camera stabilization."""
n_boxes, v_mean0, timeout = 0, 0, 5
boxes = []
best_n_boxes = 0
best_boxes = []
start_time = cv2.getTickCount()
while n_boxes < self._config_units:
@@ -945,12 +1007,18 @@ class Sensor:
if abs(v_mean - v_mean0) < 5:
n_boxes, boxes = self.stitcher.find_stitch_boxes(frame, show=show)
if n_boxes > best_n_boxes:
best_n_boxes = n_boxes
best_boxes = boxes
if (cv2.getTickCount() - start_time) / cv2.getTickFrequency() > timeout:
print("Timeout finding vision units.")
self.stitcher.isstitch = self.isstitch = False
break
v_mean0 = v_mean
if n_boxes < self._config_units and best_n_boxes >= self._config_units:
n_boxes = best_n_boxes
boxes = best_boxes
self.n_boxes = n_boxes
self.boxes = boxes
@@ -968,8 +1036,8 @@ class Sensor:
# return True
def _is_source_open(self):
if self._capture_worker is not None:
return self._capture_worker.source.is_open()
if self._capture_source is not None:
return self._capture_source.is_open()
return False
# Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor).
# if self._use_pyav:
@@ -977,13 +1045,13 @@ class Sensor:
# return self.cap is not None and self.cap.isOpened()
def _read_frame(self):
if self._capture_worker is not None and self._capture_worker.running:
return self._capture_worker.read(copy=True, wait_s=1.0)
if self._capture_source is not None and self._capture_source.running:
return self._capture_source.read_latest(copy=True, wait_s=1.0)
return self._read_frame_raw()
def _read_frame_raw(self):
if self._capture_worker is not None:
return self._capture_worker.source.read_raw()
if self._capture_source is not None:
return self._capture_source.read_raw()
return False, None
# Legacy PyAV / direct-cap fallbacks (unused after orisys.capture refactor).
# if self._use_pyav:
@@ -1003,9 +1071,14 @@ class Sensor:
@property
def capture_status(self):
"""Return capture health snapshot, or None in external-frame mode."""
if self._capture_worker is None:
if self._capture_source is None:
return None
return self._capture_worker.source.status
return self._capture_source.status
@property
def _capture_worker(self):
"""Deprecated alias for ``_capture_source``."""
return self._capture_source
# =========================================================================
+238 -18
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.
``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 dataclasses import dataclass
import numpy as np
from scipy.signal import butter, filtfilt, sosfilt, sosfilt_zi
@@ -18,28 +25,241 @@ DEFAULT_SAMPLE_RATE_HZ = 30.0
DEFAULT_LOWPASS_ORDER = 2
def _load_normal_force_coeffs(sensor_name: str | None = None) -> tuple[float, float]:
if sensor_name:
try:
from .configs import load_force_calibration
def _parse_calibration_point(point) -> tuple[float, float]:
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}")
data = load_force_calibration(sensor_name)
except Exception:
data = None
if isinstance(data, dict):
block = data.get("normal_force", data)
if isinstance(block, dict):
slope = block.get("slope")
offset = block.get("offset")
if slope is not None and offset is not None:
return float(slope), float(offset)
@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:
from .configs import load_force_calibration
data = load_force_calibration(sensor_name)
except Exception:
return None
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)
cal = ForceChannelCalibration.from_channel_dict(
block, unit=unit, default_method="linear"
)
if cal is not None:
return cal
if isinstance(block, dict):
slope = block.get("slope")
offset = block.get("offset")
if slope is not None and offset is not None:
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
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."""
slope, offset = _load_normal_force_coeffs(sensor_name)
return slope * float(raw_value) + offset
"""Convert a raw normal-force reading to Newtons.
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:
+420
View File
@@ -0,0 +1,420 @@
#!/usr/bin/env python3
"""Compare cumulative adaptive DIS vs fixed initial-reference DIS (no integration).
Reproduces drift on static scenes where adaptive flow integrates incremental
displacement: final_flow = queue[-1] + prev_fine_flow (see orisys.tracker).
Usage:
python tests/DIS_no_integrate.py --source /path/to/static_clip.mp4 --frames 500
python tests/DIS_no_integrate.py --source 0 --frames 300 --plot drift.png
python tests/DIS_no_integrate.py --config box --source 0 --no-show
Modes compared on identical grayscale frames:
cumulative Tracker(adaptive=True) — production default
non_adaptive Tracker(adaptive=False) — single ref, warm-started DIS
fixed_ref DIS calc(ref0, frame, None) — initial frame only, no queue
"""
from __future__ import annotations
import argparse
import os
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import cv2
import numpy as np
REPO_ROOT = Path(__file__).resolve().parents[1]
SRC_ROOT = REPO_ROOT / "src"
for path in (REPO_ROOT, SRC_ROOT):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
from orisys.tracker import Tracker # noqa: E402
DEFAULT_CONFIG = "box"
def parse_source(value: str):
text = str(value).strip()
for prefix in ("camera:", "cam:", "webcam:"):
if text.lower().startswith(prefix):
text = text[len(prefix) :].strip()
break
try:
return int(text)
except ValueError:
return text
def load_tracker_config(config_ref: str) -> dict:
"""Load tracker section from an Orisys config stem or JSON path."""
path = Path(config_ref)
if path.suffix.lower() == ".json" and path.is_file():
import json
data = json.loads(path.read_text(encoding="utf-8"))
else:
import orisys
data = orisys.configs.load_config(config_ref)
tracker = dict(data.get("tracker", {}))
tracker.setdefault("algorithm", "dis")
tracker.setdefault("dis_preset", "medium")
tracker.setdefault("dis_mode", 3)
tracker.setdefault("lower_threshold", 12.0)
tracker.setdefault("upper_threshold", 18.0)
tracker.setdefault("loop_closure_interval", 1)
return tracker
def dis_preset_from_name(name: str) -> int:
preset = (name or "medium").lower()
if preset == "ultrafast":
return cv2.DISOPTICAL_FLOW_PRESET_ULTRAFAST
if preset == "fast":
return cv2.DISOPTICAL_FLOW_PRESET_FAST
return cv2.DISOPTICAL_FLOW_PRESET_MEDIUM
def flow_magnitude_stats(flow: np.ndarray) -> tuple[float, float]:
"""Return (mean magnitude, max magnitude) for HxWx2 flow."""
if flow is None:
return float("nan"), float("nan")
mag = np.linalg.norm(flow.astype(np.float32), axis=-1)
return float(mag.mean()), float(mag.max())
@dataclass
class FrameMetrics:
mean_mag: float
max_mag: float
queue_len: int = 0
@dataclass
class ModeSeries:
name: str
mean_mags: list[float] = field(default_factory=list)
max_mags: list[float] = field(default_factory=list)
queue_lens: list[int] = field(default_factory=list)
def append(self, metrics: FrameMetrics) -> None:
self.mean_mags.append(metrics.mean_mag)
self.max_mags.append(metrics.max_mag)
self.queue_lens.append(metrics.queue_len)
def slope(self) -> float:
if len(self.mean_mags) < 2:
return float("nan")
x = np.arange(len(self.mean_mags), dtype=np.float64)
y = np.asarray(self.mean_mags, dtype=np.float64)
if not np.all(np.isfinite(y)):
y = np.nan_to_num(y, nan=0.0)
return float(np.polyfit(x, y, 1)[0])
class FixedRefDIS:
"""DIS from a fixed initial reference frame; no integration or queue."""
def __init__(self, config_tracker: dict):
preset = dis_preset_from_name(config_tracker.get("dis_preset", "medium"))
self._dis = cv2.DISOpticalFlow_create(preset)
self._dis.setFinestScale(int(config_tracker.get("dis_mode", 3)))
self._ref: Optional[np.ndarray] = None
def reset(self) -> None:
self._ref = None
def track(self, gray: np.ndarray) -> np.ndarray:
if self._ref is None:
self._ref = gray.copy()
h, w = gray.shape[:2]
return np.zeros((h, w, 2), dtype=np.float32)
return self._dis.calc(self._ref, gray, None)
def make_tracker(config_tracker: dict, *, adaptive: bool) -> Tracker:
cfg = dict(config_tracker)
return Tracker(
adaptive=adaptive,
cuda=False,
verbose=False,
config_tracker=cfg,
)
def read_gray_frame(cap: cv2.VideoCapture, static_image: Optional[np.ndarray]) -> tuple[bool, Optional[np.ndarray]]:
if static_image is not None:
return True, static_image.copy()
ok, frame = cap.read()
if not ok or frame is None:
return False, None
if frame.ndim == 3:
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
return True, frame
def run_comparison(
source,
*,
config_tracker: dict,
max_frames: int,
reset_every: Optional[int],
show: bool,
) -> dict[str, ModeSeries]:
path = Path(str(source)) if isinstance(source, str) else None
static_image: Optional[np.ndarray] = None
if isinstance(source, str) and path is not None and path.is_file():
img = cv2.imread(str(path), cv2.IMREAD_GRAYSCALE)
if img is not None:
static_image = img
cap = None if static_image is not None else cv2.VideoCapture(source)
if cap is not None and not cap.isOpened():
raise RuntimeError(f"Cannot open source: {source}")
cumulative = make_tracker(config_tracker, adaptive=True)
non_adaptive = make_tracker(config_tracker, adaptive=False)
fixed_ref = FixedRefDIS(config_tracker)
series = {
"cumulative": ModeSeries("cumulative"),
"non_adaptive": ModeSeries("non_adaptive"),
"fixed_ref": ModeSeries("fixed_ref"),
}
window = "DIS compare (cumulative | non_adaptive | fixed_ref)"
if show:
cv2.namedWindow(window, cv2.WINDOW_NORMAL)
try:
for frame_idx in range(max_frames):
ok, gray = read_gray_frame(cap, static_image)
if not ok or gray is None:
break
if reset_every and reset_every > 0 and frame_idx > 0 and frame_idx % reset_every == 0:
cumulative.reset()
flow_c = cumulative.track(gray.copy())
flow_n = non_adaptive.track(gray.copy())
flow_f = fixed_ref.track(gray)
mean_c, max_c = flow_magnitude_stats(flow_c)
mean_n, max_n = flow_magnitude_stats(flow_n)
mean_f, max_f = flow_magnitude_stats(flow_f)
series["cumulative"].append(
FrameMetrics(mean_c, max_c, queue_len=cumulative.getFrameQueueLength())
)
series["non_adaptive"].append(FrameMetrics(mean_n, max_n, queue_len=1))
series["fixed_ref"].append(FrameMetrics(mean_f, max_f, queue_len=1))
if show and frame_idx % 2 == 0:
h, w = gray.shape[:2]
panels = []
for flow, label, mean_val in (
(flow_c, "cumulative", mean_c),
(flow_n, "non_adaptive", mean_n),
(flow_f, "fixed_ref", mean_f),
):
vis = cv2.cvtColor(gray, cv2.COLOR_GRAY2BGR)
step = max(8, min(h, w) // 40)
ys, xs = np.mgrid[step // 2 : h : step, step // 2 : w : step]
for y, x in zip(ys.ravel(), xs.ravel()):
dx, dy = flow[y, x]
x2 = int(round(x + dx * 2))
y2 = int(round(y + dy * 2))
cv2.arrowedLine(vis, (x, y), (x2, y2), (0, 255, 0), 1, tipLength=0.3)
cv2.putText(
vis,
f"{label} m={mean_val:.3f}",
(8, 20),
cv2.FONT_HERSHEY_SIMPLEX,
0.5,
(0, 255, 255),
1,
cv2.LINE_AA,
)
panels.append(vis)
combined = np.hstack(panels)
cv2.imshow(window, combined)
key = cv2.waitKey(1) & 0xFF
if key == ord("q"):
break
finally:
if cap is not None:
cap.release()
if show:
cv2.destroyAllWindows()
return series
def print_summary(series: dict[str, ModeSeries], frames_processed: int) -> None:
print(f"\nProcessed {frames_processed} frame(s)\n")
header = f"{'mode':<14} {'initial':>10} {'final':>10} {'mean':>10} {'slope':>12} {'max_queue':>10}"
print(header)
print("-" * len(header))
for key in ("cumulative", "non_adaptive", "fixed_ref"):
s = series[key]
if not s.mean_mags:
print(f"{key:<14} {'(no data)':>10}")
continue
initial = s.mean_mags[0]
final = s.mean_mags[-1]
mean_all = float(np.mean(s.mean_mags))
slope = s.slope()
max_q = max(s.queue_lens) if s.queue_lens else 0
print(
f"{key:<14} {initial:10.4f} {final:10.4f} {mean_all:10.4f} {slope:12.6f} {max_q:10d}"
)
print("\nInterpretation on a static scene:")
print(" cumulative — rising mean/slope indicates integration drift")
print(" fixed_ref — should stay near a small bias if drift is removed")
def save_plot(series: dict[str, ModeSeries], plot_path: str) -> None:
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(10, 5))
for key, color in (
("cumulative", "C0"),
("non_adaptive", "C1"),
("fixed_ref", "C2"),
):
s = series[key]
if not s.mean_mags:
continue
ax.plot(s.mean_mags, label=key, color=color)
ax.set_xlabel("frame")
ax.set_ylabel("mean flow magnitude")
ax.set_title("DIS drift comparison (static scene expected: flat fixed_ref)")
ax.legend()
ax.grid(True, alpha=0.3)
fig.tight_layout()
path = Path(plot_path)
path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(path, dpi=120)
plt.close(fig)
print(f"Saved plot: {path.resolve()}")
def parse_args():
parser = argparse.ArgumentParser(
description="Compare cumulative vs fixed-reference DIS optical flow drift"
)
parser.add_argument(
"--source",
default="0",
help="Camera index, video path, or image path (default: 0)",
)
parser.add_argument(
"--config",
"-c",
default=DEFAULT_CONFIG,
help=f"Orisys config stem or JSON path for DIS settings (default: {DEFAULT_CONFIG})",
)
parser.add_argument(
"--frames",
type=int,
default=300,
help="Maximum frames to process (default: 300)",
)
parser.add_argument(
"--reset-every",
type=int,
default=0,
help="Reset cumulative tracker every N frames (0 = disabled)",
)
parser.add_argument(
"--plot",
nargs="?",
const="dis_drift.png",
default=None,
help="Save matplotlib plot (optional path, default: dis_drift.png)",
)
parser.add_argument(
"--no-show",
action="store_true",
help="Skip OpenCV preview window",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
source = parse_source(args.source)
config_tracker = load_tracker_config(args.config)
if config_tracker.get("algorithm", "dis") != "dis":
print(f"Warning: config algorithm is {config_tracker.get('algorithm')}; forcing dis for test.")
print("DIS no-integrate comparison")
print(f" source: {source}")
print(f" config: {args.config}")
print(f" dis_preset: {config_tracker.get('dis_preset')} dis_mode: {config_tracker.get('dis_mode')}")
print(f" frames: {args.frames} reset_every: {args.reset_every or 'off'}")
series = run_comparison(
source,
config_tracker=config_tracker,
max_frames=max(args.frames, 1),
reset_every=args.reset_every or None,
show=not args.no_show,
)
n = len(series["cumulative"].mean_mags)
print_summary(series, n)
if args.plot:
save_plot(series, args.plot)
return 0
def test_dis_no_integrate_smoke() -> None:
"""Optional pytest entry: synthetic static frames, no camera required."""
if os.getenv("ORISYS_RUN_DIS_DRIFT") != "1":
import pytest
pytest.skip("Set ORISYS_RUN_DIS_DRIFT=1 to run DIS drift smoke test")
gray = np.full((64, 64), 128, dtype=np.uint8)
# Slight per-frame noise to mimic sensor variation without true motion.
cap_frames = []
rng = np.random.default_rng(0)
for _ in range(60):
noisy = gray.astype(np.int16) + rng.integers(-2, 3, gray.shape, dtype=np.int16)
cap_frames.append(np.clip(noisy, 0, 255).astype(np.uint8))
config_tracker = load_tracker_config(DEFAULT_CONFIG)
cumulative = make_tracker(config_tracker, adaptive=True)
fixed_ref = FixedRefDIS(config_tracker)
cum_mags = []
fix_mags = []
for frame in cap_frames:
fc = cumulative.track(frame.copy())
ff = fixed_ref.track(frame)
cum_mags.append(flow_magnitude_stats(fc)[0])
fix_mags.append(flow_magnitude_stats(ff)[0])
assert len(cum_mags) >= 10
# Fixed ref should not drift more than cumulative on identical static+noise input.
cum_slope = float(np.polyfit(np.arange(len(cum_mags)), cum_mags, 1)[0])
fix_slope = float(np.polyfit(np.arange(len(fix_mags)), fix_mags, 1)[0])
assert abs(fix_slope) <= abs(cum_slope) + 0.05, (
f"fixed_ref slope {fix_slope:.4f} worse than cumulative {cum_slope:.4f}"
)
if __name__ == "__main__":
raise SystemExit(main())
+1 -1
View File
@@ -8,7 +8,7 @@ Flow:
4. Feed the rectified frame into Sensor.get_img(..., already_processed=True)
5. Compute deformation/contact and visualize
SDK version: 0.4.3
SDK version: 0.5.0
"""
from __future__ import annotations
+115 -142
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import threading
import time
import cv2
import numpy as np
import pytest
@@ -14,35 +15,47 @@ from orisys.capture import (
CameraInfo,
CaptureConfig,
CaptureHealth,
CaptureManager,
CaptureMode,
CaptureSource,
FrameBuffer,
OverflowPolicy,
ReconnectPolicy,
list_cameras,
normalize_source,
open_capture,
pick_default_camera,
)
from orisys.capture.source import _apply_stream_props, _warmup_capture
from orisys.capture.discovery import DEFAULT_LAPTOP_WEBCAM_PATTERNS
MJPG = cv2.VideoWriter_fourcc(*"MJPG")
YUY2 = cv2.VideoWriter_fourcc(*"YUY2")
class FakeCap:
"""Deterministic stand-in for cv2.VideoCapture."""
def __init__(self, frames=None, *, fail_after=None, fail_reads=0, open_ok=True):
def __init__(
self,
frames=None,
*,
fail_after=None,
fail_reads=0,
open_ok=True,
reset_fourcc_on_resize=False,
):
self._frames = list(frames or [])
self._idx = 0
self._open = bool(open_ok)
self._fail_after = fail_after
self._fail_reads_remaining = int(fail_reads)
self.read_count = 0
self.grab_count = 0
self.release_count = 0
self.width = 4
self.height = 3
self.fps = 30.0
self.fourcc = 0
self.reset_fourcc_on_resize = reset_fourcc_on_resize
self.set_calls: list[tuple[int, float]] = []
def isOpened(self):
return self._open
@@ -58,7 +71,6 @@ class FakeCap:
self._open = False
return False, None
if self._idx >= len(self._frames):
# Repeat last frame like a live camera holding the stream.
if not self._frames:
return False, None
frame = self._frames[-1]
@@ -67,12 +79,18 @@ class FakeCap:
self._idx += 1
return True, frame.copy()
def grab(self):
self.grab_count += 1
if not self._open:
return False
ok, _ = self.read()
return ok
def release(self):
self.release_count += 1
self._open = False
def get(self, prop_id):
# Values mirror common OpenCV property IDs without importing cv2 constants.
if prop_id in (3,): # CAP_PROP_FRAME_WIDTH
return float(self.width)
if prop_id in (4,): # CAP_PROP_FRAME_HEIGHT
@@ -84,6 +102,19 @@ class FakeCap:
return 0.0
def set(self, prop_id, value):
self.set_calls.append((prop_id, float(value)))
if prop_id in (3,): # CAP_PROP_FRAME_WIDTH
self.width = int(value)
if self.reset_fourcc_on_resize:
self.fourcc = YUY2
if prop_id in (4,): # CAP_PROP_FRAME_HEIGHT
self.height = int(value)
if self.reset_fourcc_on_resize:
self.fourcc = YUY2
if prop_id in (5,): # CAP_PROP_FPS
self.fps = float(value)
if prop_id in (6,): # CAP_PROP_FOURCC
self.fourcc = int(value)
return True
@@ -92,7 +123,6 @@ def _frame(val: int) -> np.ndarray:
def _factory_from_caps(caps):
"""Return a factory that hands out FakeCap instances in order."""
lock = threading.Lock()
state = {"i": 0}
@@ -107,11 +137,6 @@ def _factory_from_caps(caps):
return factory
# ---------------------------------------------------------------------------
# Source normalization / config
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"video,expected",
[
@@ -127,63 +152,42 @@ def test_normalize_source(video, expected):
assert normalize_source(video) == expected
def test_capture_config_queue_bounds():
with pytest.raises(ValueError):
CaptureConfig(source=0, queue_size=0)
with pytest.raises(ValueError):
CaptureConfig(source=0, queue_size=11)
cfg = CaptureConfig(source=0, mode="queue", queue_size=10)
assert cfg.queue_size == 10
latest = CaptureConfig(source=0, mode="latest", queue_size=5)
assert latest.queue_size == 1
def test_capture_config_from_camera_dict_ignores_deprecated_kwargs():
cfg = CaptureConfig.from_camera_dict(
0,
{"resolution_raw": [1920, 1080], "fps": 30},
mode="queue",
queue_size=5,
overflow="drop_newest",
start_worker=False,
)
assert cfg.width == 1920
assert cfg.height == 1080
assert cfg.background is False
# ---------------------------------------------------------------------------
# FrameBuffer
# ---------------------------------------------------------------------------
def test_apply_stream_props_sets_size_before_fourcc():
cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)
_apply_stream_props(cap, width=1920, height=1080, fps=30, fourcc="MJPG")
assert cap.fourcc == MJPG
width_idx = [p for p, _ in cap.set_calls if p == 3]
fourcc_calls = [v for p, v in cap.set_calls if p == 6]
assert width_idx
assert fourcc_calls
assert cap.set_calls.index((3, 1920.0)) < cap.set_calls.index((6, float(MJPG)))
def test_frame_buffer_latest_overwrites():
buf = FrameBuffer(mode=CaptureMode.LATEST)
buf.publish(_frame(1))
buf.publish(_frame(2))
ok, frame, seq = buf.read_latest(copy=True)
assert ok and seq == 2
assert int(frame[0, 0, 0]) == 2
def test_frame_buffer_queue_drop_oldest():
buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=2, overflow=OverflowPolicy.DROP_OLDEST)
buf.publish(_frame(1))
buf.publish(_frame(2))
buf.publish(_frame(3))
assert buf.dropped_frames == 1
ok, f1, _ = buf.read_next(copy=True)
ok2, f2, _ = buf.read_next(copy=True)
assert ok and ok2
assert int(f1[0, 0, 0]) == 2
assert int(f2[0, 0, 0]) == 3
def test_frame_buffer_queue_drop_newest():
buf = FrameBuffer(mode=CaptureMode.QUEUE, maxsize=1, overflow=OverflowPolicy.DROP_NEWEST)
buf.publish(_frame(1))
buf.publish(_frame(2))
assert buf.dropped_frames == 1
ok, frame, _ = buf.read_next(copy=True)
assert ok and int(frame[0, 0, 0]) == 1
# ---------------------------------------------------------------------------
# CaptureSource reconnect
# ---------------------------------------------------------------------------
def test_warmup_capture_grabs():
cap = FakeCap([_frame(1)])
_warmup_capture(cap, grabs=3)
assert cap.grab_count == 3
def test_capture_source_read_and_status():
frames = [_frame(7)]
cap = FakeCap(frames)
src = CaptureSource(
CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False), start_worker=False),
CaptureConfig(source=0, reconnect=ReconnectPolicy(enabled=False)),
factory=lambda cfg: (cap, "FAKE"),
)
src.open()
@@ -209,13 +213,12 @@ def test_auto_reconnect_after_failed_reads():
max_attempts=3,
)
src = CaptureSource(
CaptureConfig(source=0, reconnect=policy, start_worker=False, verbose=False),
CaptureConfig(source=0, reconnect=policy, verbose=False),
factory=factory,
)
src.open()
ok, _ = src.read_raw()
assert ok
# Next read fails and closes first cap.
ok, _ = src.read_raw()
assert not ok
assert src.should_reconnect()
@@ -232,7 +235,6 @@ def test_reconnect_disabled_does_not_reopen():
CaptureConfig(
source=0,
reconnect=ReconnectPolicy(enabled=False, fail_threshold=1),
start_worker=False,
),
factory=lambda cfg: (cap, "FAKE"),
)
@@ -244,7 +246,6 @@ def test_reconnect_disabled_does_not_reopen():
def test_reconnect_attempts_exhausted():
good = FakeCap([_frame(1)], fail_after=1)
state = {"n": 0}
def factory(_cfg):
@@ -262,7 +263,7 @@ def test_reconnect_attempts_exhausted():
max_attempts=2,
)
src = CaptureSource(
CaptureConfig(source=0, reconnect=policy, start_worker=False),
CaptureConfig(source=0, reconnect=policy),
factory=factory,
)
src.open()
@@ -274,37 +275,32 @@ def test_reconnect_attempts_exhausted():
assert src.status.health == CaptureHealth.FAILED
# ---------------------------------------------------------------------------
# CaptureWorker
# ---------------------------------------------------------------------------
def test_worker_latest_mode_publishes_frames():
def test_background_thread_publishes_latest_frame():
caps = [FakeCap([_frame(i) for i in range(5)])]
worker = open_capture(
src = open_capture(
CaptureConfig(
source=0,
mode=CaptureMode.LATEST,
start_worker=True,
background=True,
reconnect=ReconnectPolicy(enabled=False),
),
factory=_factory_from_caps(caps),
start_worker=True,
start_background=True,
)
try:
deadline = time.perf_counter() + 2.0
ok = False
frame = None
while time.perf_counter() < deadline:
ok, frame = worker.read(wait_s=0.05)
ok, frame = src.read(wait_s=0.05)
if ok:
break
assert ok and frame is not None
assert src.running
finally:
worker.close()
src.close()
def test_worker_reconnect_in_background():
def test_background_reconnect():
first = FakeCap([_frame(1)], fail_after=2)
second = FakeCap([_frame(42) for _ in range(20)])
policy = ReconnectPolicy(
@@ -314,19 +310,19 @@ def test_worker_reconnect_in_background():
max_backoff_s=0.02,
stale_timeout_s=None,
)
worker = open_capture(
CaptureConfig(source=0, mode="latest", reconnect=policy, start_worker=True),
src = open_capture(
CaptureConfig(source=0, reconnect=policy, background=True),
factory=_factory_from_caps([first, second]),
start_worker=True,
start_background=True,
)
try:
saw_reconnect = False
saw_new = False
deadline = time.perf_counter() + 3.0
while time.perf_counter() < deadline:
if worker.source.status.reconnect_attempts > 0:
if src.status.reconnect_attempts > 0:
saw_reconnect = True
ok, frame = worker.read(wait_s=0.05)
ok, frame = src.read(wait_s=0.05)
if ok and frame is not None and int(frame[0, 0, 0]) == 42:
saw_new = True
break
@@ -334,65 +330,19 @@ def test_worker_reconnect_in_background():
assert saw_reconnect
assert saw_new
finally:
worker.close()
src.close()
# ---------------------------------------------------------------------------
# CaptureManager
# ---------------------------------------------------------------------------
def test_capture_manager_capacity_and_status():
mgr = CaptureManager(max_sources=2)
caps = [FakeCap([_frame(1)]), FakeCap([_frame(2)]), FakeCap([_frame(3)])]
factory = _factory_from_caps(caps)
try:
mgr.open(CaptureConfig(source=0, start_worker=False), factory=factory)
mgr.open(CaptureConfig(source=1, start_worker=False), factory=factory)
assert len(mgr) == 2
with pytest.raises(RuntimeError):
mgr.open(CaptureConfig(source=2, start_worker=False), factory=factory)
statuses = mgr.status_all()
assert set(statuses) == {"0", "1"}
assert all(s.health == CaptureHealth.CONNECTED for s in statuses.values())
finally:
mgr.close_all()
assert len(mgr) == 0
def test_capture_manager_max_constant():
assert MAX_SIMULTANEOUS_CAPTURES == 10
with pytest.raises(ValueError):
CaptureManager(max_sources=11)
def test_capture_manager_read_latest_all():
mgr = CaptureManager(max_sources=3)
caps = [FakeCap([_frame(10)]), FakeCap([_frame(20)])]
factory = _factory_from_caps(caps)
try:
w0 = mgr.open(
CaptureConfig(source=0, mode="latest", start_worker=True),
factory=factory,
)
w1 = mgr.open(
CaptureConfig(source=1, mode="latest", start_worker=True),
factory=factory,
)
deadline = time.perf_counter() + 2.0
while time.perf_counter() < deadline:
if w0.buffer.last_seq > 0 and w1.buffer.last_seq > 0:
break
time.sleep(0.01)
frames = mgr.read_latest_all()
assert "0" in frames and "1" in frames
finally:
mgr.close_all()
# ---------------------------------------------------------------------------
# Camera discovery filters (no hardware)
# ---------------------------------------------------------------------------
def test_warmup_frames_discards_reads():
cap = FakeCap([_frame(i) for i in range(10)])
src = CaptureSource(
CaptureConfig(source=0),
factory=lambda cfg: (cap, "FAKE"),
)
src.open()
src.warmup_frames(3)
ok, frame = src.read_raw()
assert ok and int(frame[0, 0, 0]) == 3
def test_camera_filter_excludes_laptop_webcams():
@@ -447,10 +397,13 @@ def test_camera_info_label_includes_name():
def test_list_cameras_is_callable():
# Smoke: function import works; may return [] without hardware.
assert callable(list_cameras)
def test_max_simultaneous_captures_constant():
assert MAX_SIMULTANEOUS_CAPTURES == 10
def test_sensor_external_mode_still_works():
from orisys import Sensor
@@ -458,6 +411,26 @@ def test_sensor_external_mode_still_works():
try:
assert sensor._external_only
assert sensor.capture_status is None
assert sensor._capture_worker is None
assert sensor._capture_source is None
finally:
sensor.disconnect()
def test_capture_source_open_applies_stream_props_via_default_factory(monkeypatch):
cap = FakeCap([_frame(1)], reset_fourcc_on_resize=True)
def fake_open(*args, **kwargs):
_apply_stream_props(
cap,
width=kwargs.get("width"),
height=kwargs.get("height"),
fps=kwargs.get("fps"),
fourcc=kwargs.get("fourcc"),
)
_warmup_capture(cap)
return cap, "FAKE"
monkeypatch.setattr("orisys.capture.source.open_opencv_capture", fake_open)
src = CaptureSource(CaptureConfig(source=0, width=1920, height=1080, fourcc="MJPG"))
src.open()
assert cap.fourcc == MJPG
+25 -4
View File
@@ -35,7 +35,7 @@ def test_pyproject_dynamic_version(packaging_utils):
sys.path.insert(0, str(build_dir))
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):
@@ -48,7 +48,7 @@ def test_pyproject_bundle_manifest(packaging_utils):
items = bundle_items(REPO_ROOT)
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))
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 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
sensor_dirs = [dest for dest in data_files if dest.startswith("orisys/sensors")]
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 "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():
import sys
+104 -3
View File
@@ -12,6 +12,8 @@ from orisys.signal_processing import (
NORMAL_FORCE_CAL_SLOPE,
LowpassFilter,
calibrate_normal_force,
calibrate_shear_components,
calibrate_shear_force,
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)
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():
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.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)
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.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._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,
)
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)
finally:
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():
sensor = _external_sensor()
try:
+6 -6
View File
@@ -19,25 +19,25 @@ def wheel_package():
def test_compliant_manylinux_tags(wheel_package):
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(
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(
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):
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(
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(
Path("orisys-0.4.3-cp310-cp310-win_amd64.whl")
Path("orisys-0.5.0-cp310-cp310-win_amd64.whl")
)