organize settings to dedicated sensors folder, json + assets
This commit is contained in:
+129
-3
@@ -1,4 +1,7 @@
|
||||
import json
|
||||
from contextlib import ExitStack
|
||||
from dataclasses import dataclass
|
||||
from importlib.resources import as_file, files
|
||||
from pathlib import Path
|
||||
|
||||
# Product-facing names that map to on-disk config stems.
|
||||
@@ -8,6 +11,18 @@ CONFIG_ALIASES = {
|
||||
"dd02": "dd02-ov",
|
||||
}
|
||||
|
||||
PRODUCT_BUNDLES = ("box", "mini", "finger")
|
||||
_RESOURCE_STACK = ExitStack()
|
||||
|
||||
|
||||
def _resource_path(*parts: str) -> Path | None:
|
||||
"""Resolve a package resource to a filesystem path for wheel-safe loaders."""
|
||||
try:
|
||||
resource = files("orisys").joinpath(*parts)
|
||||
return _RESOURCE_STACK.enter_context(as_file(resource))
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _candidate_config_dirs() -> list[Path]:
|
||||
"""
|
||||
@@ -28,6 +43,26 @@ def _candidate_config_dirs() -> list[Path]:
|
||||
return unique
|
||||
|
||||
|
||||
def _candidate_sensor_roots() -> list[Path]:
|
||||
"""Candidate roots that contain per-sensor bundle folders."""
|
||||
here = Path(__file__).resolve().parent
|
||||
candidates = []
|
||||
packaged = _resource_path("sensors")
|
||||
if packaged is not None:
|
||||
candidates.append(packaged) # packaged wheel layout: orisys/sensors/
|
||||
candidates.extend([
|
||||
here / "sensors",
|
||||
here.parent.parent / "sensors", # editable install: repo_root/sensors/
|
||||
here.parent / "sensors",
|
||||
Path.cwd() / "sensors",
|
||||
])
|
||||
unique: list[Path] = []
|
||||
for path in candidates:
|
||||
if path not in unique:
|
||||
unique.append(path)
|
||||
return unique
|
||||
|
||||
|
||||
def _get_config_dir() -> Path:
|
||||
for candidate in _candidate_config_dirs():
|
||||
if candidate.exists() and any(candidate.glob("*.json")):
|
||||
@@ -50,14 +85,96 @@ def _load_json(path: Path) -> dict:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SensorBundle:
|
||||
"""Resolved paths for one additive per-sensor package folder."""
|
||||
|
||||
name: str
|
||||
root: Path
|
||||
sensor_json: Path
|
||||
force_calibration_json: Path | None = None
|
||||
manifest_json: Path | None = None
|
||||
assets_dir: Path | None = None
|
||||
|
||||
|
||||
def resolve_sensor_bundle(name: str) -> SensorBundle | None:
|
||||
"""Resolve ``sensors/<name>/`` when present; otherwise return ``None``."""
|
||||
requested = CONFIG_ALIASES.get(str(name), str(name))
|
||||
for root in _candidate_sensor_roots():
|
||||
bundle_root = root / requested
|
||||
sensor_json = bundle_root / "sensor.json"
|
||||
if not sensor_json.is_file():
|
||||
continue
|
||||
force_path = bundle_root / "force_calibration.json"
|
||||
manifest_path = bundle_root / "manifest.json"
|
||||
assets_dir = None
|
||||
if manifest_path.is_file():
|
||||
try:
|
||||
manifest = _load_json(manifest_path)
|
||||
except Exception:
|
||||
manifest = {}
|
||||
rel_assets = manifest.get("assets_dir")
|
||||
if isinstance(rel_assets, str) and rel_assets.strip():
|
||||
candidate = (bundle_root / rel_assets).resolve()
|
||||
if candidate.is_dir():
|
||||
assets_dir = candidate
|
||||
return SensorBundle(
|
||||
name=requested,
|
||||
root=bundle_root.resolve(),
|
||||
sensor_json=sensor_json.resolve(),
|
||||
force_calibration_json=force_path.resolve() if force_path.is_file() else None,
|
||||
manifest_json=manifest_path.resolve() if manifest_path.is_file() else None,
|
||||
assets_dir=assets_dir,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def resolve_sensor_config_path(name: str) -> Path | None:
|
||||
"""Return ``sensor.json`` for a resolved sensor bundle."""
|
||||
bundle = resolve_sensor_bundle(name)
|
||||
return bundle.sensor_json if bundle is not None else None
|
||||
|
||||
|
||||
def resolve_sensor_asset_dir(name: str) -> Path | None:
|
||||
"""Return the default asset directory for a resolved sensor bundle."""
|
||||
bundle = resolve_sensor_bundle(name)
|
||||
return bundle.assets_dir if bundle is not None else None
|
||||
|
||||
|
||||
def resolve_sensor_resource(name: str, *relative_parts: str) -> Path | None:
|
||||
"""Return a path inside one sensor bundle when present."""
|
||||
bundle = resolve_sensor_bundle(name)
|
||||
if bundle is None:
|
||||
return None
|
||||
return bundle.root.joinpath(*relative_parts)
|
||||
|
||||
|
||||
def list_sensor_bundles() -> list[str]:
|
||||
"""Return product names that have an additive ``sensors/<name>/`` folder."""
|
||||
found: list[str] = []
|
||||
for name in PRODUCT_BUNDLES:
|
||||
if resolve_sensor_bundle(name) is not None:
|
||||
found.append(name)
|
||||
return found
|
||||
|
||||
|
||||
def load_force_calibration(name: str) -> dict | None:
|
||||
"""Load ``force_calibration.json`` from a sensor bundle when available."""
|
||||
bundle = resolve_sensor_bundle(name)
|
||||
if bundle is None or bundle.force_calibration_json is None:
|
||||
return None
|
||||
return _load_json(bundle.force_calibration_json)
|
||||
|
||||
|
||||
def load_config(config_name_or_path):
|
||||
"""
|
||||
Load sensor configuration from a JSON file path or config name.
|
||||
|
||||
Config lookup order:
|
||||
1. Exact file path (absolute or relative)
|
||||
2. File stem under the config directory (e.g. ``box`` -> ``config/box.json``)
|
||||
3. JSON field ``"name"`` matching the requested name (keeps ``prototype`` compatible)
|
||||
2. Additive sensor bundle ``sensors/<name>/sensor.json``
|
||||
3. File stem under the config directory (e.g. ``box`` -> ``config/box.json``)
|
||||
4. JSON field ``"name"`` matching the requested name (keeps ``prototype`` compatible)
|
||||
"""
|
||||
config_path = Path(config_name_or_path)
|
||||
|
||||
@@ -75,6 +192,11 @@ def load_config(config_name_or_path):
|
||||
if alias and alias not in candidates:
|
||||
candidates.append(alias)
|
||||
|
||||
for name in candidates:
|
||||
bundle = resolve_sensor_bundle(name)
|
||||
if bundle is not None:
|
||||
return _load_json(bundle.sensor_json)
|
||||
|
||||
for name in candidates:
|
||||
direct = CONFIG_DIR / f"{name}.json"
|
||||
if direct.exists():
|
||||
@@ -103,7 +225,11 @@ def list_config():
|
||||
"""
|
||||
List all available JSON configuration files by file stem.
|
||||
"""
|
||||
return [f.stem for f in _config_files()]
|
||||
stems = [f.stem for f in _config_files()]
|
||||
for name in list_sensor_bundles():
|
||||
if name not in stems:
|
||||
stems.append(name)
|
||||
return sorted(stems)
|
||||
|
||||
|
||||
def export_config(config_name=None, path="./config/exported_config.json"):
|
||||
|
||||
@@ -4,6 +4,7 @@ import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ..configs import resolve_sensor_asset_dir
|
||||
from ..unroll import (
|
||||
compute_uv_distortion_metrics,
|
||||
load_exported_data,
|
||||
@@ -11,7 +12,7 @@ from ..unroll import (
|
||||
scaled_map_cache_path,
|
||||
)
|
||||
|
||||
DEFAULT_FINGER_EXPORT_DIR = "assets/finger/30fps-0703"
|
||||
DEFAULT_FINGER_EXPORT_DIR = "sensors/finger/assets/30fps-0703"
|
||||
DEFAULT_FINGER_MAP_CACHE = f"{DEFAULT_FINGER_EXPORT_DIR}/unroll_map_uv.npz"
|
||||
|
||||
|
||||
@@ -44,11 +45,15 @@ def _candidate_export_dirs(export_dir) -> list[Path]:
|
||||
return [raw]
|
||||
|
||||
here = Path(__file__).resolve()
|
||||
candidates = [
|
||||
candidates = []
|
||||
bundled_assets = resolve_sensor_asset_dir("finger")
|
||||
if bundled_assets is not None and raw.as_posix() == DEFAULT_FINGER_EXPORT_DIR:
|
||||
candidates.append(bundled_assets)
|
||||
candidates.extend([
|
||||
Path.cwd() / raw,
|
||||
here.parents[1] / raw, # packaged wheel: orisys/assets/...
|
||||
here.parents[1] / raw, # packaged wheel filesystem layout
|
||||
here.parents[3] / raw, # editable install: repo/assets/...
|
||||
]
|
||||
])
|
||||
if getattr(sys, "frozen", False):
|
||||
candidates.insert(0, Path(sys.executable).resolve().parent / raw)
|
||||
unique: list[Path] = []
|
||||
|
||||
@@ -10,6 +10,7 @@ import numpy as np
|
||||
from scipy.signal import butter, filtfilt, sosfilt, sosfilt_zi
|
||||
|
||||
# Linear calibration from raw normal-force units to Newtons (ORY-BOX / ORY-MINI).
|
||||
# Prefer sensors/<name>/force_calibration.json when available.
|
||||
NORMAL_FORCE_CAL_SLOPE = 3.92685554668795e-05
|
||||
NORMAL_FORCE_CAL_OFFSET = 0.0217596967278535
|
||||
|
||||
@@ -17,9 +18,28 @@ DEFAULT_SAMPLE_RATE_HZ = 30.0
|
||||
DEFAULT_LOWPASS_ORDER = 2
|
||||
|
||||
|
||||
def calibrate_normal_force(raw_value: float) -> float:
|
||||
def _load_normal_force_coeffs(sensor_name: str | None = None) -> tuple[float, float]:
|
||||
if sensor_name:
|
||||
try:
|
||||
from .configs import load_force_calibration
|
||||
|
||||
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)
|
||||
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."""
|
||||
return NORMAL_FORCE_CAL_SLOPE * float(raw_value) + NORMAL_FORCE_CAL_OFFSET
|
||||
slope, offset = _load_normal_force_coeffs(sensor_name)
|
||||
return slope * float(raw_value) + offset
|
||||
|
||||
|
||||
def _nyquist_hz(sample_rate_hz: float) -> float:
|
||||
|
||||
Reference in New Issue
Block a user