diff --git a/sensors/box/force_calibration.json b/sensors/box/force_calibration.json index 015fc07..beece03 100644 --- a/sensors/box/force_calibration.json +++ b/sensors/box/force_calibration.json @@ -1,5 +1,5 @@ { - "schema_version": 1, + "schema_version": 2, "unit": "N", "channels": { "normal": { @@ -14,8 +14,22 @@ [62000.0, 2.377] ] }, - "shear": { + "shear_x": { "method": "piecewise_linear", + "raw_key": "fshearx", + "input": "components", + "signed": true, + "extrapolate": true, + "clamp_output_min": 0.0, + "points": [ + [1000.0, 0.0], + [36000.0, 1.19], + [68000.0, 2.06] + ] + }, + "shear_y": { + "method": "piecewise_linear", + "raw_key": "fsheary", "input": "components", "signed": true, "extrapolate": true, diff --git a/src/orisys/configs.py b/src/orisys/configs.py index d3f43d9..7d0d0aa 100644 --- a/src/orisys/configs.py +++ b/src/orisys/configs.py @@ -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 (``/examples/foo.py``, or a caller + # that launched from ``/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 /examples/foo.py`` + # with the root as cwd, ``PYTHONPATH=``, 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 ``/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 ``/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) diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 7ec6b54..8473e01 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -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) diff --git a/tests/test_calibration.py b/tests/test_calibration.py index 71302a8..8d1951c 100644 --- a/tests/test_calibration.py +++ b/tests/test_calibration.py @@ -243,8 +243,84 @@ def test_missing_cal_path_raises(): def test_parse_shipped_box_file_directly(): cal = parse(json.loads(_box_cal_path().read_text(encoding="utf-8")), source_path=_box_cal_path()) - assert cal.schema_version == 1 + assert cal.schema_version == 2 assert cal.convert(10000.0, 36000.0, -36000.0) == pytest.approx((0.377, 1.19, -1.19)) + # The shipped file names both shear axes outright: no shared block, no clone. + assert set(cal.channels) == {"fn", "fx", "fy"} + fx, fy = cal.get("FX"), cal.get("FY") + assert fx is not None and fy is not None and fx is not fy + for curve in (fx, fy): + assert curve.signed is True + assert curve.extrapolate is True + assert curve.clamp_output_min == pytest.approx(0.0) + assert len(curve.points) == 3 + assert fx.raw_key == "fshearx" + assert fy.raw_key == "fsheary" + + +def test_the_shipped_box_shear_axes_can_drift_apart(tmp_path: Path): + """The whole point of the split: FX and FY are independent curves. + + A document with the same shape as the shipped one but different per-axis + points must apply each axis's own curve. Re-collapsing the axes into one + shared ``shear`` block makes this fail. + """ + doc = { + "schema_version": 2, + "unit": "N", + "channels": { + "normal": { + "method": "piecewise_linear", + "points": [[0.0, 0.0], [1000.0, 1.0]], + }, + "shear_x": { + "method": "piecewise_linear", + "signed": True, + "input": "components", + "extrapolate": True, + "clamp_output_min": 0.0, + "points": [[0.0, 0.0], [500.0, 2.0]], + }, + "shear_y": { + "method": "piecewise_linear", + "signed": True, + "input": "components", + "extrapolate": True, + "clamp_output_min": 0.0, + "points": [[0.0, 0.0], [500.0, 3.0]], + }, + }, + } + path = tmp_path / "split.json" + path.write_text(json.dumps(doc), encoding="utf-8") + cal = load(path) + assert cal is not None + assert cal.convert(0.0, 500.0, 500.0)[1:] == pytest.approx((2.0, 3.0)) + assert cal.convert(0.0, -500.0, 500.0)[1:] == pytest.approx((-2.0, 3.0)) + + +def test_out_of_range_readings_extend_instead_of_being_dropped(): + """Outside the calibrated range the nearest segment is extended. + + This pins the "extend, never ignore, never reuse a previous coefficient" + behaviour of the shipped box file, so flipping ``extrapolate`` has to be a + deliberate edit that fails this test first. + """ + cal = load(sensor_type="box") + assert cal is not None + fx, fy, fn = cal.get("FX"), cal.get("FY"), cal.get("FN") + assert fx is not None and fy is not None and fn is not None + # 136000 = 2 x the top raw break (68000 -> 2.06): the last segment's slope is + # applied again, so the value keeps growing past the calibrated range. + (r0, n0), (r1, n1) = fx.points[-2], fx.points[-1] + extended = n1 + (n1 - n0) / (r1 - r0) * (136000.0 - r1) + assert extended > n1 + assert fx.convert_component(136000.0) == pytest.approx(extended) + assert fy.convert_component(136000.0) == pytest.approx(extended) + # A negative beyond the range keeps its sign. + assert fx.convert_component(-136000.0) == pytest.approx(-extended) + # Below the first point the normal channel floors at clamp_output_min. + assert fn.convert(500.0) == pytest.approx(0.0) # ------------------------------------------------------- filter fingerprint diff --git a/tests/test_signal_processing.py b/tests/test_signal_processing.py index b2e3a59..0a735dd 100644 --- a/tests/test_signal_processing.py +++ b/tests/test_signal_processing.py @@ -2,11 +2,13 @@ from __future__ import annotations +import json + import numpy as np import pytest from scipy.signal import butter, filtfilt -from orisys import Sensor +from orisys import Sensor, configs from orisys.signal_processing import ( NORMAL_FORCE_CAL_OFFSET, NORMAL_FORCE_CAL_SLOPE, @@ -263,3 +265,156 @@ def test_configure_hhd_downsample_invalidates_solver_cache(): sensor.configure(hhd_downsample="nope") finally: sensor.disconnect() + + +# ------------------------------------------------- bundle resolution anchors +# +# The bundle is chosen by the *running* root's folder layout, not by the root's +# name and not by whichever checkout happens to be importable. These build small +# real trees, because ``_runtime_roots`` asks the filesystem whether a directory +# carries ``examples/`` beside ``sensors/``. +_CAL_DOC = { + "schema_version": 2, + "unit": "N", + "channels": { + "normal": {"method": "piecewise_linear", "points": [[0.0, 0.0], [1000.0, 1.0]]}, + "shear_x": { + "method": "piecewise_linear", + "signed": True, + "points": [[0.0, 0.0], [500.0, 2.0]], + }, + "shear_y": { + "method": "piecewise_linear", + "signed": True, + "points": [[0.0, 0.0], [500.0, 3.0]], + }, + }, +} + + +def _make_root(base, *, name="anything", sensor_json=True, calibration=True): + """A runnable SDK root: ``examples/`` beside ``sensors//``.""" + root = base / name + (root / "examples").mkdir(parents=True, exist_ok=True) + if sensor_json: + (root / "examples" / "basic_pipeline.py").write_text("# demo\n", encoding="utf-8") + bundle = root / "sensors" / "mini" + bundle.mkdir(parents=True, exist_ok=True) + if sensor_json: + (bundle / "sensor.json").write_text( + json.dumps({"name": "prototype"}), encoding="utf-8" + ) + if calibration: + (bundle / "force_calibration.json").write_text( + json.dumps(_CAL_DOC), encoding="utf-8" + ) + return root + + +def test_runtime_root_anchor_keys_on_layout_not_on_the_folder_name(tmp_path): + named = _make_root(tmp_path, name="some-random-name") + assert configs._is_sdk_runtime_root(named) is True + + no_examples = tmp_path / "dist-like" + (no_examples / "sensors" / "mini").mkdir(parents=True) + assert configs._is_sdk_runtime_root(no_examples) is False + + neither = tmp_path / "plain" + neither.mkdir() + assert configs._is_sdk_runtime_root(neither) is False + + +def test_runtime_roots_walk_up_from_a_subdirectory(tmp_path, monkeypatch): + root = _make_root(tmp_path, name="renamed-release") + subdir = root / "examples" + monkeypatch.setattr(configs.Path, "cwd", classmethod(lambda cls: subdir)) + roots = [str(path) for path in configs._runtime_roots()] + assert str(root) in roots + + +def test_the_running_root_bundle_beats_a_calibration_less_checkout( + tmp_path, monkeypatch +): + """The reported bug: a nearer bundle without coefficients must not win.""" + running = _make_root(tmp_path, name="running-root") + checkout = _make_root(tmp_path, name="importable-checkout", calibration=False) + monkeypatch.setattr(configs, "_runtime_roots", lambda: [running, checkout]) + + bundle = configs.resolve_sensor_bundle("mini") + assert bundle is not None + assert bundle.root == (running / "sensors" / "mini").resolve() + assert bundle.force_calibration_json == ( + running / "sensors" / "mini" / "force_calibration.json" + ).resolve() + + # The whole point: the running root's document is the one that reaches the + # sensor, split shear and all. + sensor = Sensor("external", config_name="mini", verbose=False, backend="cpu") + try: + assert sensor.force_calibrated is True + assert sensor.force_unit == "N" + assert sensor.calibration.source_path == ( + running / "sensors" / "mini" / "force_calibration.json" + ).resolve() + finally: + sensor.disconnect() + + +def test_a_nearer_bundle_wins_even_without_coefficients(tmp_path, monkeypatch): + """Nearest bundle wins, so a calibration-less nearer one hides a farther one. + + That is the chosen precedence (the running root is authoritative), and the + diagnostic exists precisely so this stops being silent: this is the shape of + the original report, where a checkout's ``mini`` has only ``sensor.json``. + """ + checkout = _make_root(tmp_path, name="checkout", calibration=False) + running = _make_root(tmp_path, name="run-root") + monkeypatch.setattr(configs, "_runtime_roots", lambda: [checkout, running]) + bundle = configs.resolve_sensor_bundle("mini") + assert bundle is not None + assert bundle.root == (checkout / "sensors" / "mini").resolve() + assert bundle.force_calibration_json is None + info = configs.describe_bundle_resolution("mini") + assert info["found_root"] == (checkout / "sensors" / "mini").resolve() + assert info["force_calibration_json"] is False + + +def test_describe_bundle_resolution_names_the_bundle_and_the_searched_roots( + tmp_path, monkeypatch +): + checkout = _make_root(tmp_path, name="checkout", calibration=False) + running = _make_root(tmp_path, name="run-root") + roots = [checkout / "sensors", running / "sensors"] + # Patched at the root list, so no real bundle on this machine can satisfy + # the lookup while the "nothing found" case is being asserted. + monkeypatch.setattr(configs, "_candidate_sensor_roots", lambda: list(roots)) + + info = configs.describe_bundle_resolution("mini") + assert info["found_root"] == (checkout / "sensors" / "mini").resolve() + assert info["force_calibration_json"] is False + assert [str(path) for path in info["searched"]] == [str(p) for p in roots] + + (checkout / "sensors" / "mini" / "sensor.json").unlink() + (running / "sensors" / "mini" / "sensor.json").unlink() + missing = configs.describe_bundle_resolution("mini") + assert missing["found_root"] is None + assert missing["sensor_json"] is None + assert missing["force_calibration_json"] is False + + +def test_the_calibration_note_names_the_bundle_that_was_found( + tmp_path, monkeypatch +): + checkout = _make_root(tmp_path, name="checkout", calibration=False) + monkeypatch.setattr(configs, "_runtime_roots", lambda: [checkout]) + sensor = Sensor("external", config_name="mini", verbose=False, backend="cpu") + try: + assert sensor.calibration is None + # Still None: callers read a non-None value here as "calibrated". + assert sensor._force_cal_sensor_name is None + note = sensor._force_calibration_note() + assert "has no force_calibration.json" in note + assert str((checkout / "sensors" / "mini").resolve()) in note + assert "roots searched" in note + finally: + sensor.disconnect()