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

fixed calibration data not loading due to path

Browse Source
This commit is contained in:
ykcheungab
2026-09-20 11:00:33 +08:00
parent 14fbfec62e
commit 1cf78c9d71
5 changed files with 448 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/orisys/configs.py
+139 -17
View File
@@ -1,5 +1,6 @@
import json
import shutil
import sys
from contextlib import ExitStack
from dataclasses import dataclass
from importlib.resources import as_file, files
@@ -25,29 +26,116 @@ def _resource_path(*parts: str) -> Path | None:
return None
def _candidate_config_dirs() -> list[Path]:
def _unique_paths(paths) -> list[Path]:
"""Ordered, de-duplicated paths (Windows comparison is case-insensitive)."""
unique: list[Path] = []
seen: set[str] = set()
for path in paths:
try:
key = str(path.resolve()).casefold()
except OSError: # pragma: no cover - defensive: unresolvable path
key = str(path).casefold()
if key in seen:
continue
seen.add(key)
unique.append(path)
return unique
def _is_sdk_runtime_root(path: Path) -> bool:
"""Whether ``path`` carries the runnable SDK layout.
The anchor is the **folder layout**, never the folder's name: a distribution
folder may be called anything (a release checkout, a renamed copy, an
unpacked zip), but it always ships ``examples/`` beside ``sensors/``. That is
what distinguishes a run root from an arbitrary directory, and it is why a
renamed root keeps working while ``dist/**`` (``sensors/`` without
``examples/``) is not mistaken for one.
"""
Candidate config directories for editable installs and packaged wheels.
try:
if not path.is_dir():
return False
return (path / "examples").is_dir() and (path / "sensors").is_dir()
except OSError: # pragma: no cover - defensive: unreadable path
return False
def _runtime_roots() -> list[Path]:
"""Runnable SDK roots for this process, nearest to the launcher first.
Evaluated once at import time, so a process that changes directory later
keeps the answer it started with.
"""
anchors: list[Path] = []
def add(path: Path) -> None:
if path not in anchors:
anchors.append(path)
cwd = Path.cwd()
add(cwd)
# A script run from a subdirectory (``<root>/examples/foo.py``, or a caller
# that launched from ``<root>/scripts``) still names its root through argv[0].
script_dir: Path | None = None
try:
argv0 = sys.argv[0] if sys.argv else ""
if argv0:
candidate = Path(argv0).expanduser()
script_dir = candidate.resolve().parent if candidate.is_file() else None
except (OSError, ValueError): # pragma: no cover - defensive
script_dir = None
# Walk up from both, so the root is found however deep the launcher sits.
for start in (cwd, script_dir):
if start is None:
continue
for candidate in (start, *start.parents):
add(candidate)
# Roots that put themselves on sys.path (``python <root>/examples/foo.py``
# with the root as cwd, ``PYTHONPATH=<root>``, and similar).
for entry in list(sys.path):
if not entry:
continue
try:
add(Path(entry).expanduser())
except (OSError, ValueError): # pragma: no cover - defensive
continue
return [path for path in _unique_paths(anchors) if _is_sdk_runtime_root(path)]
def _candidate_config_dirs() -> list[Path]:
"""Candidate config directories, the running SDK root first.
A run root's ``config/`` outranks the packaged copy so a bundle's
``sensor.json`` and its ``force_calibration.json`` come from the same place.
"""
here = Path(__file__).resolve().parent
candidates = [
return _unique_paths([
*(root / "config" for root in _runtime_roots()),
here / "configs", # packaged wheel layout: orisys/configs/*.json
here.parent.parent / "config", # editable install from repo: repo_root/config/*.json
here.parent / "config", # fallback
Path.cwd() / "config", # script launched from repo root
]
unique: list[Path] = []
for path in candidates:
if path not in unique:
unique.append(path)
return unique
])
def _candidate_sensor_roots() -> list[Path]:
"""Candidate roots that contain per-sensor bundle folders."""
"""Candidate roots that contain per-sensor bundle folders.
Lookup order:
1. the ``sensors/`` of every runnable SDK root found at startup, nearest
first -- so the bundle under the directory that is actually running wins,
whatever that directory is called;
2. the packaged resource ``orisys/sensors`` (wheel installs);
3. the module-relative candidates (editable checkout, packaged tree);
4. the current working directory, last, as a safety net.
"""
here = Path(__file__).resolve().parent
candidates = []
candidates = [root / "sensors" for root in _runtime_roots()]
packaged = _resource_path("sensors")
if packaged is not None:
candidates.append(packaged) # packaged wheel layout: orisys/sensors/
@@ -57,11 +145,7 @@ def _candidate_sensor_roots() -> list[Path]:
here.parent / "sensors",
Path.cwd() / "sensors",
])
unique: list[Path] = []
for path in candidates:
if path not in unique:
unique.append(path)
return unique
return _unique_paths(candidates)
def _get_config_dir() -> Path:
@@ -130,6 +214,44 @@ def resolve_sensor_bundle(name: str) -> SensorBundle | None:
return None
def describe_bundle_resolution(name: str) -> dict:
"""Explain how ``name`` resolved, so a missing calibration stops being silent.
A bundle can resolve and still carry no ``force_calibration.json``: the
lookup stops at the first root that has ``<name>/sensor.json`` and never
asks whether that root also has the coefficients. Reporting which bundle was
found -- and which roots were passed over -- is what makes that case
distinguishable from "no bundle at all", which is otherwise what the caller
sees.
Returns ``name``, the ``searched`` roots in order, and the first root that
supplied ``<name>/sensor.json`` together with whether it carried a
``force_calibration.json``.
"""
requested = CONFIG_ALIASES.get(str(name), str(name))
searched = _candidate_sensor_roots()
for root in searched:
sensor_json = root / requested / "sensor.json"
if not sensor_json.is_file():
continue
return {
"name": requested,
"searched": searched,
"found_root": sensor_json.parent,
"sensor_json": sensor_json,
"force_calibration_json": (
sensor_json.parent / "force_calibration.json"
).is_file(),
}
return {
"name": requested,
"searched": searched,
"found_root": None,
"sensor_json": None,
"force_calibration_json": False,
}
def resolve_sensor_config_path(name: str) -> Path | None:
"""Return ``sensor.json`` for a resolved sensor bundle."""
bundle = resolve_sensor_bundle(name)
src/orisys/sensor.py
+60 -4
View File
@@ -217,6 +217,9 @@ class Sensor:
self._force_raw = (0.0, 0.0, 0.0)
self.calibration: Optional[cal_mod.Calibration] = None
self._force_cal_sensor_name: Optional[str] = None
#: Why no calibration was applied, when a bundle existed but carried no
#: usable coefficients. ``None`` means the question never came up.
self._force_cal_attempts: Optional[list[dict]] = None
self._init_force_calibration()
# ---------------------------------------------------------------
@@ -330,10 +333,11 @@ class Sensor:
if self.verbose:
if self.calibration is not None and self.calibration.source_path is not None:
print(f"Force calibration: {self.calibration.source_path}")
elif self._force_cal_sensor_name:
print(
f"Force calibration: sensors/{self._force_cal_sensor_name}/force_calibration.json"
)
elif self._force_cal_attempts is not None:
# A bundle was resolved for this sensor type, so "file missing"
# would be the wrong story: name the bundle and the roots that
# were searched instead.
print(self._force_calibration_note())
else:
print("Force calibration: raw (no force_calibration.json)")
@@ -1258,22 +1262,71 @@ class Sensor:
ordered.append(name)
return ordered
def _force_calibration_note(self) -> str:
"""Explain a bundle that resolved but carried no usable coefficients.
The old message ("raw (no force_calibration.json)") reads as "the file is
missing", which is exactly the wrong story when the lookup found a
bundle from another root and stopped there without ever asking whether
that root had coefficients.
"""
attempts = self._force_cal_attempts or []
found = [entry for entry in attempts if entry.get("found_root") is not None]
with_cal = [entry for entry in found if entry.get("force_calibration_json")]
chosen = (with_cal or found or attempts or [None])[0]
searched: list = []
for entry in attempts:
for root in entry.get("searched") or ():
if root not in searched:
searched.append(root)
lines: list[str] = []
if chosen is None:
lines.append("Force calibration: raw (no sensor bundle resolved)")
else:
root = chosen["found_root"]
if chosen.get("force_calibration_json"):
lines.append(
f"Force calibration: raw (a bundle was found at {root}, but its "
"force_calibration.json could not be loaded)"
)
else:
lines.append(
f"Force calibration: raw (bundle found at {root} has no "
"force_calibration.json)"
)
if searched:
lines.append(
" roots searched: " + "; ".join(str(root) for root in searched)
)
return "\n".join(lines)
def _init_force_calibration(self) -> None:
"""Load force calibration from ``cal_path`` or the product bundle."""
self.calibration = None
self._force_cal_sensor_name = None
self._force_cal_attempts = None
if self.cal_path is not None and str(self.cal_path).strip():
# Explicit path: missing/invalid raises (no silent fallback).
self.calibration = cal_mod.load(self.cal_path, strict=True)
return
attempts: list[dict] = []
for name in self._bundle_name_candidates():
cal = cal_mod.load(sensor_type=name)
if cal is not None:
self.calibration = cal
self._force_cal_sensor_name = name
self._force_cal_attempts = None
return
try:
attempts.append(configs.describe_bundle_resolution(name))
except Exception: # noqa: BLE001 - diagnostics must never break loading
continue
# Deliberately NOT setting ``_force_cal_sensor_name``: callers read a
# non-None value there as "this sensor is calibrated", and a bundle
# without coefficients is not that.
self._force_cal_attempts = attempts or None
@property
def force_calibrated(self) -> bool:
@@ -1292,11 +1345,14 @@ class Sensor:
if cal is None:
self.calibration = None
self._force_cal_sensor_name = None
self._force_cal_attempts = None
return
if isinstance(cal, cal_mod.Calibration):
self.calibration = cal
self._force_cal_attempts = None
return
self.calibration = cal_mod.load(cal, strict=True)
self._force_cal_attempts = None
def _apply_force_filter(self):
pre_filter = np.asarray(self._force_raw, dtype=np.float64)