orisys-tech/sensor-bench
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update plot to force units

Browse Source
This commit is contained in:
ykcheungab
2026-09-01 16:24:12 +08:00
parent 8cacb12abc
commit e7789c86cb
31 changed files with 767 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/sensor_bench/acquisition.py
+133 -23
View File
@@ -18,6 +18,11 @@ import cv2
import numpy as np
from .force_calibration import FORCE_SAMPLE_RATE_HZ, LOWPASS_CUTOFF_HZ
from .paths import (
bundle_has_force_calibration,
normalize_sensor_config_ref,
resolve_force_bundle_from_config,
)
@dataclass(frozen=True)
@@ -35,6 +40,21 @@ class FixedCameraMode:
return f"{self.width}x{self.height} @{self.fps:g} (DDJX01 / MJPG)"
@dataclass
class ForceReadout:
"""Calibrated and raw force values from one Orisys Sensor tick."""
normal_force: float = 0.0
shear_x: float = 0.0
shear_y: float = 0.0
normal_force_raw: float = 0.0
shear_x_raw: float = 0.0
shear_y_raw: float = 0.0
force_calibrated: bool = False
force_unit: str = "raw"
flow: Optional[np.ndarray] = None
@dataclass
class SdkPacket:
"""One acquisition tick from the Orisys Sensor."""
@@ -51,11 +71,109 @@ class SdkPacket:
normal_force: float = 0.0
shear_x: float = 0.0
shear_y: float = 0.0
normal_force_raw: float = 0.0
shear_x_raw: float = 0.0
shear_y_raw: float = 0.0
force_unit: str = "raw"
force_calibrated: bool = False
session_id: int = 0
mode: Optional[FixedCameraMode] = None
flow: Optional[np.ndarray] = None
def sensor_force_calibrated(sensor, *, config_ref: str | Path | None = None) -> bool:
"""True when Orisys resolved sensors/<name>/force_calibration.json."""
if getattr(sensor, "_force_cal_sensor_name", None) is not None:
return True
bundle = resolve_force_bundle_from_config(config_ref)
return bundle is not None and bundle_has_force_calibration(bundle)
def read_forces_from_sensor(
sensor,
*,
include_flow: bool = True,
config_ref: str | Path | None = None,
) -> ForceReadout:
"""Read FN/FX/FY plus *_RAW from Orisys, with legacy SDK fallback."""
info = sensor.info
fn = fx = fy = 0.0
fn_raw = fx_raw = fy_raw = 0.0
flow = None
read_raw = all(
hasattr(info, name)
for name in ("FNORMAL_RAW", "FSHEARX_RAW", "FSHEARY_RAW")
)
has_flow = include_flow and hasattr(info, "VRAW")
try:
if read_raw and has_flow:
fn, fx, fy, fn_raw, fx_raw, fy_raw, flow = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
info.FNORMAL_RAW,
info.FSHEARX_RAW,
info.FSHEARY_RAW,
info.VRAW,
)
elif read_raw:
fn, fx, fy, fn_raw, fx_raw, fy_raw = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
info.FNORMAL_RAW,
info.FSHEARX_RAW,
info.FSHEARY_RAW,
)
elif has_flow:
fn, fx, fy, flow = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
info.VRAW,
)
fn_raw, fx_raw, fy_raw = fn, fx, fy
else:
fn, fx, fy = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
)
fn_raw, fx_raw, fy_raw = fn, fx, fy
except Exception: # noqa: BLE001
try:
if has_flow:
fn, fx, fy, flow = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
info.VRAW,
)
else:
fn, fx, fy = sensor.read_info(
info.FNORMAL,
info.FSHEARX,
info.FSHEARY,
)
fn_raw, fx_raw, fy_raw = fn, fx, fy
except Exception: # noqa: BLE001
pass
calibrated = sensor_force_calibrated(sensor, config_ref=config_ref)
unit = "N" if calibrated else "raw"
return ForceReadout(
normal_force=float(fn),
shear_x=float(fx),
shear_y=float(fy),
normal_force_raw=float(fn_raw),
shear_x_raw=float(fx_raw),
shear_y_raw=float(fy_raw),
force_calibrated=calibrated,
force_unit=unit,
flow=flow,
)
def orisys_importable() -> tuple[bool, str]:
try:
import orisys # noqa: F401
@@ -255,7 +373,7 @@ class SdkAcquisitionWorker:
try:
self._status(f"Opening Orisys Sensor ({mode.label()})…")
kwargs = {
"config_name": mode.config_ref,
"config_name": normalize_sensor_config_ref(mode.config_ref),
"verbose": False,
"backend": self._backend,
}
@@ -297,24 +415,11 @@ class SdkAcquisitionWorker:
frame_times = [t for t in frame_times if t >= cutoff]
fps = float(len(frame_times)) if frame_times else 0.0
fn = fx = fy = 0.0
flow = None
try:
fn, fx, fy, flow = self._sensor.read_info(
self._sensor.info.FNORMAL,
self._sensor.info.FSHEARX,
self._sensor.info.FSHEARY,
self._sensor.info.VRAW,
)
except Exception: # noqa: BLE001
try:
fn, fx, fy = self._sensor.read_info(
self._sensor.info.FNORMAL,
self._sensor.info.FSHEARX,
self._sensor.info.FSHEARY,
)
except Exception: # noqa: BLE001
pass
forces = read_forces_from_sensor(
self._sensor,
include_flow=True,
config_ref=mode.config_ref,
)
raw_bgr = _ensure_bgr(raw)
proc_bgr = _ensure_bgr(processed)
@@ -328,12 +433,17 @@ class SdkAcquisitionWorker:
height=int(raw_bgr.shape[0]),
processed_width=int(proc_bgr.shape[1]),
processed_height=int(proc_bgr.shape[0]),
normal_force=float(fn),
shear_x=float(fx),
shear_y=float(fy),
normal_force=forces.normal_force,
shear_x=forces.shear_x,
shear_y=forces.shear_y,
normal_force_raw=forces.normal_force_raw,
shear_x_raw=forces.shear_x_raw,
shear_y_raw=forces.shear_y_raw,
force_unit=forces.force_unit,
force_calibrated=forces.force_calibrated,
session_id=session,
mode=mode,
flow=flow,
flow=forces.flow,
)
with self._lock:
self._latest = packet
src/sensor_bench/dd06_acquisition.py
+24 -23
View File
@@ -7,8 +7,16 @@ import time
from pathlib import Path
from typing import Callable, Optional
from .acquisition import FixedCameraMode, SdkPacket, _ensure_bgr, load_fixed_mode, orisys_importable
from .acquisition import (
FixedCameraMode,
SdkPacket,
_ensure_bgr,
load_fixed_mode,
orisys_importable,
read_forces_from_sensor,
)
from .force_calibration import FORCE_SAMPLE_RATE_HZ, LOWPASS_CUTOFF_HZ
from .paths import normalize_sensor_config_ref
from .roi_geometry import (
build_roi_warp,
load_json_config,
@@ -138,6 +146,7 @@ class Dd06AcquisitionWorker:
config = load_json_config(config_ref)
roi_points, roi_image_size = load_roi_polygon(config)
out_size = rectified_output_size(config, roi_points)
config_ref = normalize_sensor_config_ref(config_ref)
self._sensor = orisys.Sensor("external", config_name=config_ref, verbose=False)
self._sensor.configure(
lowpass_cutoff_hz=LOWPASS_CUTOFF_HZ,
@@ -190,24 +199,11 @@ class Dd06AcquisitionWorker:
frame_times = [t for t in frame_times if t >= cutoff]
fps = float(len(frame_times)) if frame_times else 0.0
fn = fx = fy = 0.0
flow = None
try:
fn, fx, fy, flow = self._sensor.read_info(
self._sensor.info.FNORMAL,
self._sensor.info.FSHEARX,
self._sensor.info.FSHEARY,
self._sensor.info.VRAW,
)
except Exception: # noqa: BLE001
try:
fn, fx, fy = self._sensor.read_info(
self._sensor.info.FNORMAL,
self._sensor.info.FSHEARX,
self._sensor.info.FSHEARY,
)
except Exception: # noqa: BLE001
pass
forces = read_forces_from_sensor(
self._sensor,
include_flow=True,
config_ref=self._config_ref,
)
raw_bgr = _ensure_bgr(frame)
# Prefer sensor.img / get_img result for processed visualization.
@@ -225,12 +221,17 @@ class Dd06AcquisitionWorker:
height=int(raw_bgr.shape[0]),
processed_width=int(proc_bgr.shape[1]),
processed_height=int(proc_bgr.shape[0]),
normal_force=float(fn),
shear_x=float(fx),
shear_y=float(fy),
normal_force=forces.normal_force,
shear_x=forces.shear_x,
shear_y=forces.shear_y,
normal_force_raw=forces.normal_force_raw,
shear_x_raw=forces.shear_x_raw,
shear_y_raw=forces.shear_y_raw,
force_unit=forces.force_unit,
force_calibrated=forces.force_calibrated,
session_id=session,
mode=mode,
flow=flow,
flow=forces.flow,
)
with self._lock:
self._latest = packet
src/sensor_bench/exhibition_app.py
+46 -9
View File
@@ -47,12 +47,13 @@ from .acquisition import (
from .camera_discovery import list_cameras, pick_default_camera
from .dd06_acquisition import Dd06AcquisitionWorker
from .force_calibration import (
CALIBRATED_FORCE_Y_MAX,
CALIBRATED_FORCE_Y_MIN,
FORCE_SAMPLE_RATE_HZ,
PLOT_BUFFER_POINTS,
PLOT_WINDOW_SEC,
Y_AXIS_MIN_SPAN,
Y_RANGE_ALPHA,
calibrate_normal_force,
)
from .i18n import DEFAULT_LOCALE, Translator
from .paths import (
@@ -226,6 +227,8 @@ class ExhibitionWindow(QMainWindow):
self._view_mode = "pure_color"
self._sensor_type = "orisys"
self._log_panel_visible = False
self._force_unit = "raw"
self._force_calibrated = False
self._fixed_mode: Optional[FixedCameraMode] = None
self._orisys_config = str(default_ddjx01_config())
self._dd06_config = str(default_dd06_config())
@@ -448,12 +451,23 @@ class ExhibitionWindow(QMainWindow):
self.view_combo.blockSignals(False)
for plot, title_key in zip(self.plots, self.plot_titles):
plot.setTitle(self.t(title_key), color="#9aa0a6", size="12pt")
plot.setTitle(self._plot_title(title_key), color="#9aa0a6", size="12pt")
if self._source_pixmap is None:
self.image_label.setText(self.t("labels.no_preview"))
self.status_label.setText(self.t("status.ready"))
def _plot_title(self, title_key: str) -> str:
return self.t(
"exhibition.plot_with_unit",
title=self.t(title_key),
unit=self._force_unit,
)
def _update_force_plot_titles(self) -> None:
for plot, title_key in zip(self.plots, self.plot_titles):
plot.setTitle(self._plot_title(title_key), color="#9aa0a6", size="12pt")
def _selected_config(self) -> str:
data = self.config_combo.currentData()
if data:
@@ -646,6 +660,9 @@ class ExhibitionWindow(QMainWindow):
return
self.force_history.reset()
self._force_unit = "raw"
self._force_calibrated = False
self._update_force_plot_titles()
self._awaiting_first_frame = True
self._source_pixmap = None
self.image_label.clear()
@@ -794,12 +811,29 @@ class ExhibitionWindow(QMainWindow):
)
self.status_label.setText(info)
self._log(info)
self._log(
self.t(
"exhibition.force_fn_n",
value=calibrate_normal_force(packet.normal_force),
self._force_unit = packet.force_unit
self._force_calibrated = packet.force_calibrated
self._update_force_plot_titles()
if packet.force_calibrated:
self._log(
self.t(
"exhibition.force_calibrated",
fn=packet.normal_force,
fn_raw=packet.normal_force_raw,
fx=packet.shear_x,
fy=packet.shear_y,
unit=packet.force_unit,
)
)
else:
self._log(
self.t(
"exhibition.force_raw",
fn=packet.normal_force,
fx=packet.shear_x,
fy=packet.shear_y,
)
)
)
self._last_packet = packet
self._last_packet_id = packet_id
@@ -870,8 +904,11 @@ class ExhibitionWindow(QMainWindow):
x = state.times
y = state.values[:, i]
curve.setData(x, y)
ymin, ymax = state.y_ranges[i]
self.plots[i].setYRange(ymin, ymax)
if self._force_calibrated:
ymin, ymax = CALIBRATED_FORCE_Y_MIN, CALIBRATED_FORCE_Y_MAX
else:
ymin, ymax = state.y_ranges[i]
self.plots[i].setYRange(ymin, ymax, padding=0)
self.dot_cores[i].setData(x[-1:], y[-1:])
x0, x1 = state.x_range
for plot in self.plots:
src/sensor_bench/force_calibration.py
+2
View File
@@ -11,6 +11,8 @@ FORCE_SAMPLE_RATE_HZ = 30.0
PLOT_BUFFER_POINTS = 120
PLOT_WINDOW_SEC = PLOT_BUFFER_POINTS / FORCE_SAMPLE_RATE_HZ
Y_AXIS_MIN_SPAN = 20000.0
CALIBRATED_FORCE_Y_MIN = -20.0
CALIBRATED_FORCE_Y_MAX = 20.0
Y_RANGE_ALPHA = 0.5
LOWPASS_CUTOFF_HZ = 10.0
src/sensor_bench/paths.py
+174 -16
View File
@@ -8,6 +8,31 @@ from pathlib import Path
NON_SENSOR_CONFIG_STEMS = {"options", "bench_theme"}
# Product bundles with optional force_calibration.json (orisys-sdk layout).
PRODUCT_BUNDLES = ("box", "mini", "finger")
# Legacy flat config stems mapped to a product bundle for force calibration lookup.
LEGACY_STEM_TO_BUNDLE: dict[str, str] = {
"ddjx01": "box",
"ddjx01_480p": "box",
"ddjx01_q6a": "box",
"ddjx01_wuxi": "box",
"box": "box",
"prototype": "box",
"prototype_of3": "box",
"prototype_of3_480p": "box",
"prototype_of3_720p": "box",
"prototype_raft": "box",
"mini": "mini",
"mini-fast": "mini",
"finger": "finger",
"dd02-ov": "finger",
"dd02": "finger",
"dd06-ov": "finger",
"dd06-ov-0810": "finger",
"x-ov": "finger",
}
def is_frozen() -> bool:
return bool(getattr(sys, "frozen", False))
@@ -58,6 +83,11 @@ def local_sensor_configs_dir() -> Path:
return app_root() / "config"
def local_sensor_bundles_dir() -> Path:
"""Per-sensor bundle root: ``sensors/<product>/sensor.json``."""
return app_root() / "sensors"
def _is_sensor_config_json(path: Path) -> bool:
if path.suffix.lower() != ".json":
return False
@@ -76,10 +106,114 @@ def list_local_sensor_config_files() -> list[Path]:
return sorted(p for p in root.glob("*.json") if _is_sensor_config_json(p))
def local_sensor_bundle_sensor_json(name: str) -> Path | None:
"""Return ``sensors/<name>/sensor.json`` beside the app when present."""
path = local_sensor_bundles_dir() / name / "sensor.json"
return path if path.is_file() else None
def local_sensor_bundle_force_calibration(name: str) -> Path | None:
path = local_sensor_bundles_dir() / name / "force_calibration.json"
return path if path.is_file() else None
def list_local_sensor_bundle_names() -> list[str]:
"""Product bundle folder names under ``sensors/`` with a ``sensor.json``."""
root = local_sensor_bundles_dir()
if not root.is_dir():
return []
found: list[str] = []
for name in PRODUCT_BUNDLES:
if local_sensor_bundle_sensor_json(name) is not None:
found.append(name)
return found
def _orisys_bundle_has_force_calibration(name: str) -> bool:
try:
import orisys
return orisys.configs.load_force_calibration(name) is not None
except Exception: # noqa: BLE001
return False
def bundle_has_force_calibration(name: str) -> bool:
"""True when local or packaged Orisys bundle includes force_calibration.json."""
if local_sensor_bundle_force_calibration(name) is not None:
return True
return _orisys_bundle_has_force_calibration(name)
def resolve_force_bundle_from_config(config_ref: str | Path | None) -> str | None:
"""Map a legacy flat config path/stem to a product bundle with calibration."""
if config_ref is None:
return None
text = str(config_ref).strip().replace("\\", "/")
if not text:
return None
path = Path(text)
candidates: list[str] = []
if path.suffix.lower() == ".json":
candidates.append(path.stem.lower())
stem = path.stem.lower() if path.suffix else text.lower()
if stem:
candidates.append(stem)
if text.lower() in PRODUCT_BUNDLES:
candidates.insert(0, text.lower())
parts = [p for p in text.split("/") if p]
if "sensors" in parts:
idx = parts.index("sensors")
if idx + 1 < len(parts):
candidates.insert(0, parts[idx + 1].lower())
seen: set[str] = set()
for candidate in candidates:
if not candidate or candidate in seen:
continue
seen.add(candidate)
if candidate in PRODUCT_BUNDLES and local_sensor_bundle_sensor_json(candidate):
if bundle_has_force_calibration(candidate):
return candidate
mapped = LEGACY_STEM_TO_BUNDLE.get(candidate)
if mapped and bundle_has_force_calibration(mapped):
return mapped
return None
def normalize_sensor_config_ref(config_ref: str | Path) -> str:
"""Prefer product bundle names so Orisys loads ``sensors/<name>/`` calibration."""
text = str(config_ref).strip()
if not text:
return text
path = Path(text)
if path.is_file() and path.name.lower() == "sensor.json":
parent = path.parent.name.lower()
if parent in PRODUCT_BUNDLES:
return parent
bundle = resolve_force_bundle_from_config(text)
if bundle is not None:
local_json = local_sensor_bundle_sensor_json(bundle)
if local_json is not None:
return str(local_json.resolve())
return bundle
if text.lower() in PRODUCT_BUNDLES:
local_json = local_sensor_bundle_sensor_json(text)
if local_json is not None:
return str(local_json.resolve())
return text
return text
def orisys_configs_dir() -> Path | None:
"""Prefer local sensor configs dir when present; else installed package dir."""
local = local_sensor_configs_dir()
if list_local_sensor_config_files():
if list_local_sensor_config_files() or list_local_sensor_bundle_names():
return local
try:
import orisys
@@ -93,37 +227,54 @@ def orisys_configs_dir() -> Path | None:
def list_orisys_configs() -> list[str]:
"""Return selectable sensor configs.
Local ``config/*.json`` files (excluding non-sensor stems such as
``bench_theme``) are listed first as
absolute paths so Start uses the editable copy beside the EXE. Package stems
from the installed Orisys wheel are appended when not already covered.
Product bundles under ``sensors/<name>/sensor.json`` and Orisys bundle
names are listed before legacy flat ``config/*.json`` files so force
calibration resolves correctly.
"""
entries: list[str] = []
seen_stems: set[str] = set()
for path in list_local_sensor_config_files():
stem = path.stem.lower()
if stem in seen_stems:
continue
for name in list_local_sensor_bundle_names():
path = local_sensor_bundle_sensor_json(name)
assert path is not None
entries.append(str(path.resolve()))
seen_stems.add(stem)
seen_stems.add(name.lower())
try:
import orisys
for stem in orisys.configs.list_config():
key = str(stem).strip()
if not key:
continue
if key.lower() in NON_SENSOR_CONFIG_STEMS:
for name in orisys.configs.list_sensor_bundles():
key = str(name).strip()
if not key or key.lower() in NON_SENSOR_CONFIG_STEMS:
continue
if key.lower() in seen_stems:
continue
entries.append(key)
seen_stems.add(key.lower())
for stem in orisys.configs.list_config():
key = str(stem).strip()
if not key or key.lower() in NON_SENSOR_CONFIG_STEMS:
continue
if key.lower() in seen_stems:
continue
if key.lower() in PRODUCT_BUNDLES:
continue
entries.append(key)
seen_stems.add(key.lower())
except Exception: # noqa: BLE001
pass
for path in list_local_sensor_config_files():
stem = path.stem.lower()
if stem in seen_stems:
continue
mapped = LEGACY_STEM_TO_BUNDLE.get(stem)
if mapped and mapped.lower() in seen_stems:
continue
entries.append(str(path.resolve()))
seen_stems.add(stem)
return entries
@@ -133,6 +284,11 @@ def config_display_name(value: str) -> str:
if not text:
return ""
path = Path(text)
parts = [p for p in path.as_posix().split("/") if p]
if "sensors" in parts:
idx = parts.index("sensors")
if idx + 1 < len(parts):
return parts[idx + 1]
if path.suffix.lower() == ".json" or "/" in text or "\\" in text:
return path.stem
return text
@@ -143,6 +299,8 @@ def default_config_name(sensor_type: str = "orisys") -> str:
kind = str(sensor_type or "orisys").strip().lower()
if kind in ("dd06", "dd06-ov", "roi", "polygon_roi"):
return "dd06-ov"
if list_local_sensor_bundle_names() or _orisys_bundle_has_force_calibration("box"):
return "box"
return "ddjx01"
@@ -157,7 +315,7 @@ def _pick_default_config(preferred_stem: str) -> str:
def default_ddjx01_config() -> str:
"""Prefer local editable config path, else package stem."""
"""Prefer product bundle (box), else local editable config path."""
return _pick_default_config(default_config_name("orisys"))