From 474dbf5a64cc4f9afda5abaa0c06d1d74bdd797c Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Wed, 21 Oct 2026 11:04:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=BA=E6=AF=8F=E4=B8=AA=20Sensor=20?= =?UTF-8?q?=E7=BB=91=E5=AE=9A=E5=8F=AA=E8=AF=BB=20SensorContext?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 构造时一次绑定 SN、配置、资产和当前生效的标定：只接受显式给出的 SN，不做猜测；标定路径只取当前生效的标定，清除后为 None；换标定时整体替换。帧身份复用 stream-contract 的 FrameIdentity。load_config 改为复用 resolve_config_source，使记录的配置来源与实际读取的文件一致，并补一致性测试。已发布 Sensor 数值和资产包解析不变。回归 275 项、lint 通过。 --- docs/stream-contract.zh.md | 44 ++- src/orisys/configs.py | 69 +++-- src/orisys/sensor.py | 54 ++++ src/orisys/sensors/__init__.py | 11 + src/orisys/sensors/context.py | 353 ++++++++++++++++++++++++ tests/test_config_source_consistency.py | 97 +++++++ tests/test_sensor_context.py | 292 ++++++++++++++++++++ 7 files changed, 899 insertions(+), 21 deletions(-) create mode 100644 src/orisys/sensors/context.py create mode 100644 tests/test_config_source_consistency.py create mode 100644 tests/test_sensor_context.py diff --git a/docs/stream-contract.zh.md b/docs/stream-contract.zh.md index b8b9472..5fa3f7f 100644 --- a/docs/stream-contract.zh.md +++ b/docs/stream-contract.zh.md @@ -163,14 +163,54 @@ assert measurement.stamp == stamp 未知字段被拒绝，是为了让不同版本的记录不会静默错配。采用政策 A：字段变更（包括新增字段）一律升 `STREAM_CONTRACT_VERSION`，同时把新版本加入读取方的 `SUPPORTED_CONTRACT_VERSIONS`；同版本内不新增字段。 +## Sensor 上下文（只读绑定） + +`Sensor.context` 返回一个 frozen `SensorContext`（`from orisys.sensors import SensorContext`），在构造时为本实例绑定一次以下内容，并在显式换标定时整体替换： + +| 字段 | 说明 | +|---|---| +| `sensor_id` | 仅显式提供的设备 SN（`device_id`）；未显式绑定为 `None`，绝不由枚举顺序、文件夹名或相机编号猜测 | +| `product_family` / `processing_backend` | 由配置/资产包确定的产品家族与图像处理后端 | +| `config_name` / `config_source` / `config_fingerprint` | 配置名、实际读取的 JSON 文件、加载后配置的规范化 SHA-256 | +| `bundle_root` / `manifest_digest` / `assets` | 资产包根目录、manifest 内容摘要、manifest 声明的资产角色与包内相对路径（legacy manifest 与 descriptor manifest 都支持；运行时实际加载路径可能不同） | +| `calibration_path` / `calibration_fingerprint` | 当前**正在生效**的力标定文件与内容指纹；未标定或 `set_calibration(None)` 后为 `None`/`"none"` | +| `diagnostics` | 绑定过程中可容忍的资产/配置说明，不改变 Sensor 的构造结果 | + +规则： + +- 帧身份/时钟字段不在这里重复：`context.frame_identity(session_id=..., frame_id=..., source_frame_id=..., acquisition_id=..., sync_group_id=...)` 直接构造 T03 的 `FrameIdentity`。没有显式 SN 时它抛 `ValueError`，不伪造身份。 +- 每个 Sensor 实例独立绑定，没有模块级共享缓存；创建或使用另一个 Sensor 不会改变本实例的上下文。 +- 显式 `Sensor.set_calibration()` 换标定时原子替换上下文：SN/配置/资产绑定保持不变，标定路径与指纹更新；`set_calibration(None)` 后 `calibration_path` 立即为 `None`、指纹为 `"none"`，不保留构造时请求过的旧路径。 +- 绑定异常只写入 `diagnostics`，不把原本可构造的 Sensor 变成构造失败。 +- `assets`/`manifest_digest` 是 **manifest 声明的绑定**，不是运行时实际加载的路径：Finger 的导出目录由 `finger/exports.py` 在 cwd、打包目录、可编辑安装目录等候选中另行查找，实际加载路径由 T05 的图像块绑定；需要复现实验时以 T05/记录的身份为准。 +- 资产取应用包内相对路径，不复制、不上传、不按顺序推断；配置指纹取实际加载后的配置内容，而不是文件名。 +- 本任务（T04）不改变 Sensor 的数值与既有资产包解析行为，也不把上下文接入流入口；T05/T08 消费该上下文。 + +```python +from orisys import configs +from orisys.sensors import build_sensor_context +from orisys.stream_contract import FrameIdentity + +context = build_sensor_context( + config_name="box", + config=configs.load_config("box"), + device_id="BOX-001", + bundle_candidates=["box"], +) +identity: FrameIdentity = context.frame_identity( + session_id="run-7", frame_id=12, source_frame_id=41, +) +assert identity.sensor_id == "BOX-001" +``` + ## 不包含什么 - 不实现 SensorStream 入口、实时最新选帧、完整回放和背压（T08）。 -- 不实现图像块拆分、测量负载、ForceSession 重做、零点与输出滤波（T04–T07）。 +- 不实现图像块拆分、测量负载、ForceSession 重做、零点与输出滤波（T05–T07）。 - 不实现批次、配组、多流调度（后续多流计划）。 - 不实现录制/回放格式、Studio 迁移和离线重新滤波（T11、T13–T15）。 - 不修改现有 `Sensor`、`StreamPipeline` 和当前 `ForceSession`；现有 `ForceSession.FrameStamp` 在 T06 重做时被本合同的帧头替换。 ## 验证 -`tests/test_stream_contract.py` 覆盖 T03 的四项验收：逐路身份与时钟、组字段可缺但不半组、同 PTS 帧可区分、`source_dt_s` 与事件 JSON 往返；同时覆盖 frozen 校验、严格版本/未知字段拒绝（政策 A）、`SUPPORTED_CONTRACT_VERSIONS`、事件可哈希、`kind` 的 lower_snake 格式和非法负载拒绝。类型本身不启动传感器、流水线或会话。 +`tests/test_stream_contract.py` 覆盖 T03 的四项验收：逐路身份与时钟、组字段可缺但不半组、同 PTS 帧可区分、`source_dt_s` 与事件 JSON 往返；同时覆盖 frozen 校验、严格版本/未知字段拒绝（政策 A）、`SUPPORTED_CONTRACT_VERSIONS`、事件可哈希、`kind` 的 lower_snake 格式和非法负载拒绝。`tests/test_sensor_context.py` 覆盖 T04 的只读 SensorContext：显式 SN/配置/资产/标定路径绑定、不猜 SN、实例隔离、换标定原子替换、绑定失败只报 diagnostics，以及 config source/资产包绑定。类型与上下文本身不启动传感器、流水线或会话。 diff --git a/src/orisys/configs.py b/src/orisys/configs.py index 8adba21..0305fb6 100644 --- a/src/orisys/configs.py +++ b/src/orisys/configs.py @@ -283,6 +283,48 @@ def resolve_sensor_config_path(name: str) -> Path | None: return bundle.sensor_json if bundle is not None else None +def resolve_config_source(config_name_or_path) -> Path | None: + """Return the JSON file :func:`load_config` would read for this name/path. + + Uses the same lookup order as ``load_config``: exact file, additive sensor + bundle, config directory stem, then the legacy ``"name"`` field scan. Returns + ``None`` when the name has no file (never inventing a path), and follows + aliases exactly like loading does so a recorded path is the one that was + actually read. + """ + config_path = Path(config_name_or_path) + if config_path.exists() and config_path.is_file(): + return config_path.resolve() + + if config_path.is_absolute(): + return None + + requested = str(config_name_or_path) + candidates = [requested] + alias = CONFIG_ALIASES.get(requested) + 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 bundle.sensor_json + + for name in candidates: + direct = CONFIG_DIR / f"{name}.json" + if direct.exists(): + return direct.resolve() + + for candidate in _config_files(): + try: + data = _load_json(candidate) + except Exception: # noqa: BLE001, S112 - unreadable configs are skipped during the scan + continue + if data.get("name") in candidates: + return candidate.resolve() + return 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) @@ -402,6 +444,10 @@ def load_config(config_name_or_path): 2. Additive sensor bundle ``sensors//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) + + The lookup itself lives in :func:`resolve_config_source`, so a caller that + records which file was read (for example ``SensorContext.config_source``) + cannot drift from what was actually loaded here. """ config_path = Path(config_name_or_path) @@ -413,31 +459,16 @@ def load_config(config_name_or_path): if config_path.is_absolute(): raise FileNotFoundError(f"Config file not found at absolute path: {config_path}") + source = resolve_config_source(config_name_or_path) + if source is not None: + return _load_json(source) + requested = str(config_name_or_path) candidates = [requested] alias = CONFIG_ALIASES.get(requested) 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(): - return _load_json(direct) - - # Backward-compatible alias resolution: support old names via JSON "name" field. - for candidate in _config_files(): - try: - data = _load_json(candidate) - except Exception: # noqa: S112 - skip unreadable config during alias scan - continue - if data.get("name") in candidates: - return data - available = [f.stem for f in _config_files()] available_str = ", ".join(available) if available else "none" raise FileNotFoundError( diff --git a/src/orisys/sensor.py b/src/orisys/sensor.py index 02d6af1..c13657f 100644 --- a/src/orisys/sensor.py +++ b/src/orisys/sensor.py @@ -11,6 +11,7 @@ import time from dataclasses import dataclass from enum import Enum, auto from pathlib import Path +from typing import TYPE_CHECKING import cv2 import numpy as np @@ -41,6 +42,9 @@ from .capture import ( open_capture, ) +if TYPE_CHECKING: + from .sensors.context import SensorContext + @dataclass(frozen=True) class FrameMeasurement: @@ -229,6 +233,10 @@ class Sensor: self.bind_calibration_context(build_context( self._config, self.filters_state(), preprocessing={"mirror": self.mirror}, )) + # One read-only binding of SN/config/assets/cal_path per instance. The + # later read-only SensorContext contract (T04) is additive; a binding + # problem is reported on the context, never by changing construction. + self._bind_sensor_context() if self.verbose: if self.calibration is not None and self.calibration.source_path is not None: @@ -1435,6 +1443,48 @@ class Sensor: ), } + def _bind_sensor_context(self) -> None: + """Bind this instance's read-only SN/config/assets/cal_path context. + + Called once during construction and again only when the force + calibration is explicitly replaced. The binding is local to this + instance; no state is cached or shared between sensors, and an + unexpected binding problem is reported on the context instead of + changing whether a published Sensor can be constructed. Partially + initialised Sensor doubles (tests build some with ``__new__``) are + tolerated: their missing fields bind as ``None``. + """ + from .sensors.context import build_sensor_context, fallback_sensor_context + + config_name = getattr(self, "config_name", "") + config = getattr(self, "_config", None) + device_id = getattr(self, "device_id", None) + calibration = getattr(self, "calibration", None) + try: + bundle_candidates = self._bundle_name_candidates() + except Exception: # noqa: BLE001 - a test double may not carry the config + bundle_candidates = [] + try: + self._sensor_context = build_sensor_context( + config_name=config_name, + config=config if isinstance(config, dict) else {}, + device_id=device_id, + calibration=calibration, + bundle_candidates=bundle_candidates, + ) + except Exception as exc: # noqa: BLE001 - context binding must not break a Sensor + self._sensor_context = fallback_sensor_context( + config_name=config_name, + config=config, + device_id=device_id, + reason=exc, + ) + + @property + def context(self) -> "SensorContext": + """Read-only SN/config/assets/calibration binding of this Sensor instance.""" + return self._sensor_context + @property def force_calibrated(self) -> bool: """True when at least one channel curve is loaded.""" @@ -1475,6 +1525,7 @@ class Sensor: self._calibration_context = None self._force_readout.reset() self._apply_force_filter() + self._bind_sensor_context() return if isinstance(cal, cal_mod.Calibration) or getattr(cal, "schema_version", None) == 3: selected = cal @@ -1509,6 +1560,9 @@ class Sensor: if not self._uses_calibration_model(): self._force_readout.reset() self._apply_force_filter() + # The context is a value object: an explicit calibration swap atomically + # replaces it (SN/config/assets stay the same; cal path/fingerprint update). + self._bind_sensor_context() def _apply_force_filter(self): convert = self.calibration.convert if self.calibration is not None else None diff --git a/src/orisys/sensors/__init__.py b/src/orisys/sensors/__init__.py index 4c3c38d..77e4531 100644 --- a/src/orisys/sensors/__init__.py +++ b/src/orisys/sensors/__init__.py @@ -53,6 +53,12 @@ from .capabilities import ( report, unavailable, ) +from .context import ( + SensorAssetBinding, + SensorContext, + build_sensor_context, + fallback_sensor_context, +) from .descriptor import ( BACKEND_COMPOUND_EYE, BACKEND_SINGLE_CAMERA, @@ -100,6 +106,11 @@ __all__ = [ "Availability", "AxisAvailability", "Issue", + # read-only per-instance binding (T04) + "SensorContext", + "SensorAssetBinding", + "build_sensor_context", + "fallback_sensor_context", # descriptor "BundleDescriptor", "DescriptorError", diff --git a/src/orisys/sensors/context.py b/src/orisys/sensors/context.py new file mode 100644 index 0000000..efe9ab6 --- /dev/null +++ b/src/orisys/sensors/context.py @@ -0,0 +1,353 @@ +"""Read-only ``SensorContext``: the SN, configuration, assets and calibration path +bound once for one Sensor instance. + +Stage-1 architecture (K11.1, sensor-context task T04) makes one object own what +must not be re-guessed per frame: which physical sensor this is, which +``sensor.json``/configuration it runs, which bundle assets were selected, and +which force calibration was chosen. The context is a frozen value object; a +``Sensor`` binds it once during construction (and atomically replaces it only +when its calibration is explicitly swapped), so two Sensor instances cannot +share or overwrite each other's identity. + +Identity fields are not duplicated here: :meth:`SensorContext.frame_identity` +builds the frozen :class:`~orisys.stream_contract.FrameIdentity` from stage 1, +so the per-frame stream contract remains the single home for frame identity and +clock fields. + +The context is plain data (paths as strings, hashes, tuples): no Qt, no device +handle, no mutable mapping. ``build_sensor_context`` is tolerant by design - a +bundle that cannot be described is reported in ``diagnostics`` instead of +changing a published Sensor's construction behaviour. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .. import configs as sdk_configs +from ..stream_contract import FrameIdentity +from .descriptor import parse_descriptor +from .legacy import backend_from_config + +__all__ = [ + "SensorAssetBinding", + "SensorContext", + "build_sensor_context", + "fallback_sensor_context", +] + +#: Manifest keys that describe the bundle itself rather than an asset file. +_MANIFEST_META_KEYS = frozenset({"name", "sensor_json", "version", "schema_version"}) + + +def _json_digest(value: Any) -> tuple[str, str | None]: + """Canonical SHA-256 of a JSON value plus an optional diagnostic.""" + try: + text = json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False) + except (TypeError, ValueError) as exc: # a config that is not plain JSON + return ( + hashlib.sha256(str(value).encode("utf-8")).hexdigest(), + f"config is not plain JSON for fingerprinting: {exc}", + ) + return hashlib.sha256(text.encode("utf-8")).hexdigest(), None + + +def _file_digest(path: Path) -> str | None: + try: + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + except OSError: + return None + + +def _relative_posix(path: Path, root: Path) -> str | None: + try: + return path.resolve().relative_to(root.resolve()).as_posix() + except (OSError, ValueError): + return None + + +def _manifest_assets( + bundle: sdk_configs.SensorBundle, + manifest_raw: Mapping[str, Any] | None, + descriptor, + diagnostics: list[str], +) -> tuple[SensorAssetBinding, ...]: + if manifest_raw is None: + return () + + if descriptor is not None: + # Descriptor-shaped manifest: bind the declared computational assets. + bindings = [] + for declared in (descriptor.computational_assets, descriptor.assets): + for role, value in sorted(declared.items()): + rel = _relative_posix(Path(value), bundle.root) + if rel is None: + diagnostics.append(f"asset {role!r} escapes the bundle folder") + continue + bindings.append(SensorAssetBinding(str(role), rel)) + return tuple(bindings) + + # Legacy manifest (for example ``sensors/finger/manifest.json``): bind the + # declared asset directory and any single-file entries it names. + bindings = [] + for role, value in sorted(manifest_raw.items()): + if ( + role in _MANIFEST_META_KEYS + or not isinstance(value, str) + or not value.strip() + ): + continue + candidate = (bundle.root / Path(value)).resolve() + rel = _relative_posix(candidate, bundle.root) + if rel is None: + diagnostics.append(f"manifest entry {role!r} escapes the bundle folder") + continue + bindings.append(SensorAssetBinding(str(role), rel)) + return tuple(bindings) + + +@dataclass(frozen=True) +class SensorAssetBinding: + """One bundle asset the instance was bound to: a role and a bundle-relative path.""" + + role: str + path: str + + +@dataclass(frozen=True) +class SensorContext: + """The immutable identity/config/assets/calibration binding of one Sensor instance. + + ``sensor_id`` is only the **explicitly supplied** SN (``device_id``); it is + ``None`` when the caller did not bind one. The context never guesses an SN + from enumeration order, bundle names or reserved camera indices - a frame + that needs a stream identity therefore fails loudly instead of being + attributed to the wrong device. ``calibration_path`` and + ``calibration_fingerprint`` describe the calibration currently in effect: + ``set_calibration(None)`` leaves them ``None``/``"none"`` even when the + Sensor was constructed with an explicit path. + """ + + sensor_id: str | None + product_family: str | None + processing_backend: str + config_name: str + config_source: str | None + config_fingerprint: str + bundle_root: str | None + manifest_digest: str | None + assets: tuple[SensorAssetBinding, ...] + calibration_path: str | None + calibration_fingerprint: str + diagnostics: tuple[str, ...] = field(default_factory=tuple) + + def frame_identity( + self, + *, + session_id: str, + frame_id: int, + source_frame_id: int | None, + acquisition_id: str | None = None, + sync_group_id: str | None = None, + ) -> FrameIdentity: + """Build this stream's frozen frame identity from the bound SN. + + The capture clock is not part of ``SensorContext``: it belongs to the + frame header and is produced with the frame. This method reuses + :class:`~orisys.stream_contract.FrameIdentity` rather than defining a + second identity shape. + """ + if self.sensor_id is None: + raise ValueError( + "SensorContext has no explicit sensor_id (SN); construct the Sensor " + "with device_id (or bind one) before building a frame identity" + ) + return FrameIdentity( + sensor_id=self.sensor_id, + session_id=session_id, + frame_id=frame_id, + source_frame_id=source_frame_id, + acquisition_id=acquisition_id, + sync_group_id=sync_group_id, + ) + + def as_dict(self) -> dict[str, Any]: + """JSON-serialisable snapshot for records and diagnostics.""" + return { + "sensor_id": self.sensor_id, + "product_family": self.product_family, + "processing_backend": self.processing_backend, + "config_name": self.config_name, + "config_source": self.config_source, + "config_fingerprint": self.config_fingerprint, + "bundle_root": self.bundle_root, + "manifest_digest": self.manifest_digest, + "assets": [ + {"role": asset.role, "path": asset.path} for asset in self.assets + ], + "calibration_path": self.calibration_path, + "calibration_fingerprint": self.calibration_fingerprint, + "diagnostics": list(self.diagnostics), + } + + +def build_sensor_context( + *, + config_name: str | Path, + config: Mapping[str, Any], + device_id: str | None = None, + calibration: Any = None, + bundle_candidates: tuple[str, ...] | list[str] = (), +) -> SensorContext: + """Bind one Sensor's SN, configuration identity, assets and calibration path. + + Every lookup is explicit and local to this call; nothing is shared between + instances. Optional bundle description problems are collected in + ``diagnostics`` and never raised, so adding a context cannot change whether a + published Sensor can be constructed. + """ + diagnostics: list[str] = [] + cfg = config if isinstance(config, Mapping) else {} + config_fingerprint, digest_problem = _json_digest(dict(cfg)) + if digest_problem: + diagnostics.append(digest_problem) + + config_source: str | None = None + try: + source = sdk_configs.resolve_config_source(config_name) + except Exception as exc: # noqa: BLE001 - diagnostics only + source = None + diagnostics.append(f"config source could not be resolved: {exc}") + if source is not None: + config_source = str(Path(source).resolve()) + + names = [str(candidate) for candidate in bundle_candidates] + product_family = next( + (name for name in names if name in sdk_configs.PRODUCT_BUNDLES), None + ) + bundle = None + for name in names: + try: + bundle = sdk_configs.resolve_sensor_bundle(name) + except Exception as exc: # noqa: BLE001 - diagnostics only + diagnostics.append(f"bundle {name!r} could not be resolved: {exc}") + continue + if bundle is not None: + break + + processing_backend = backend_from_config(cfg) + bundle_root: str | None = None + manifest_digest: str | None = None + assets: tuple[SensorAssetBinding, ...] = () + if bundle is not None: + bundle_root = str(bundle.root) + if product_family is None and bundle.name in sdk_configs.PRODUCT_BUNDLES: + product_family = bundle.name + manifest_raw: Mapping[str, Any] | None = None + descriptor = None + if bundle.manifest_json is not None: + manifest_digest = _file_digest(bundle.manifest_json) + if manifest_digest is None: + diagnostics.append("bundle manifest.json is unreadable") + try: + loaded = json.loads( + Path(bundle.manifest_json).read_text(encoding="utf-8-sig") + ) + except (OSError, json.JSONDecodeError) as exc: + diagnostics.append(f"bundle manifest.json could not be read: {exc}") + else: + if isinstance(loaded, Mapping): + manifest_raw = loaded + if loaded.get("schema_version") is not None: + try: + descriptor = parse_descriptor(bundle.root) + except Exception as exc: # noqa: BLE001 - reported, never fatal + diagnostics.append( + f"bundle manifest.json is not a valid descriptor: {exc}" + ) + # Legacy manifests have no backend field; the config rule that + # already drives create_frame_processor stays in force. + else: + diagnostics.append("bundle manifest.json is not a JSON object") + if descriptor is not None: + processing_backend = descriptor.processing_backend + assets = _manifest_assets(bundle, manifest_raw, descriptor, diagnostics) + else: + diagnostics.append("no additive sensor bundle resolved for this config") + + calibration_path: str | None = None + calibration_fingerprint = "none" + if calibration is not None: + source_path = getattr(calibration, "source_path", None) + if source_path is not None: + calibration_path = str(Path(source_path).resolve()) + fingerprint = getattr(calibration, "fingerprint", None) + if isinstance(fingerprint, str) and fingerprint: + calibration_fingerprint = fingerprint + else: + calibration_fingerprint = "unknown" + diagnostics.append( + f"{type(calibration).__name__} exposes no content fingerprint" + ) + + sensor_id = None + if device_id is not None: + text = str(device_id).strip() + sensor_id = text or None + if sensor_id is None: + diagnostics.append("device_id was empty; no SN is bound") + + return SensorContext( + sensor_id=sensor_id, + product_family=product_family, + processing_backend=processing_backend, + config_name=str(config_name), + config_source=config_source, + config_fingerprint=config_fingerprint, + bundle_root=bundle_root, + manifest_digest=manifest_digest, + assets=assets, + calibration_path=calibration_path, + calibration_fingerprint=calibration_fingerprint, + diagnostics=tuple(diagnostics), + ) + + +def fallback_sensor_context( + *, + config_name: str | Path, + config: Mapping[str, Any] | None, + device_id: str | None, + reason: BaseException, +) -> SensorContext: + """A minimal context used when binding itself fails unexpectedly. + + The Sensor keeps working; the failure and its cause are visible on the + context instead of crashing a published construction path. + """ + config_fingerprint, _ = _json_digest( + dict(config) if isinstance(config, Mapping) else {} + ) + backend = ( + backend_from_config(config) if isinstance(config, Mapping) else "single_camera" + ) + sensor_id = str(device_id).strip() if device_id is not None else None + return SensorContext( + sensor_id=sensor_id or None, + product_family=None, + processing_backend=backend, + config_name=str(config_name), + config_source=None, + config_fingerprint=config_fingerprint, + bundle_root=None, + manifest_digest=None, + assets=(), + calibration_path=None, + calibration_fingerprint="none", + diagnostics=(f"context binding failed: {type(reason).__name__}: {reason}",), + ) diff --git a/tests/test_config_source_consistency.py b/tests/test_config_source_consistency.py new file mode 100644 index 0000000..7d1bdb6 --- /dev/null +++ b/tests/test_config_source_consistency.py @@ -0,0 +1,97 @@ +"""``resolve_config_source`` and ``load_config`` must agree on what is read. + +``SensorContext.config_source`` promises the JSON file that was actually loaded. +That promise only holds while the recorded lookup and the loading lookup are the +same implementation, so this module checks it against every installed config +name, the alias table, an explicit path and the legacy ``"name"`` field, and +pins the preserved error types/messages for a missing or non-JSON input. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from orisys import configs + +REPO_ROOT = Path(__file__).resolve().parents[1] +SENSORS = REPO_ROOT / "sensors" + + +def _read(path: str | Path) -> dict: + return json.loads(Path(path).read_text(encoding="utf-8-sig")) + + +def test_every_listed_name_resolves_to_the_file_load_config_reads() -> None: + names = configs.list_config() + assert names + for name in names: + source = configs.resolve_config_source(name) + assert source is not None, name + assert _read(source) == configs.load_config(name), name + # The recorded source can be loaded directly with the same result. + assert configs.load_config(source) == configs.load_config(name), name + + +def test_aliases_resolve_like_loading() -> None: + aliases = dict(configs.CONFIG_ALIASES) + assert aliases + for alias, canonical in aliases.items(): + source = configs.resolve_config_source(alias) + assert source is not None, alias + assert _read(source) == configs.load_config(alias), alias + canonical_source = configs.resolve_config_source(canonical) + if canonical_source is not None: + assert source == canonical_source, alias + + +def test_explicit_paths_resolve_to_themselves(tmp_path: Path) -> None: + bundle_config = SENSORS / "box" / "sensor.json" + assert configs.resolve_config_source(bundle_config) == bundle_config.resolve() + assert configs.load_config(bundle_config) == configs.load_config("box") + + copied = tmp_path / "copied-config.json" + copied.write_text(bundle_config.read_text(encoding="utf-8"), encoding="utf-8") + assert configs.resolve_config_source(copied) == copied.resolve() + assert configs.load_config(copied) == configs.load_config(bundle_config) + + +def test_legacy_name_field_scan_agrees_with_loading() -> None: + """A config's JSON ``"name"`` may differ from its file stem (``prototype``).""" + found = False + for name in configs.list_config(): + source = configs.resolve_config_source(name) + if source is None: + continue + field = _read(source).get("name") + if ( + isinstance(field, str) + and field + and field not in configs.list_config() + ): + resolved = configs.resolve_config_source(field) + assert resolved is not None, field + assert _read(resolved) == configs.load_config(field), field + found = True + assert found, "expected at least one legacy name field in the installed configs" + + +def test_missing_and_non_json_inputs_keep_their_error_contract(tmp_path: Path) -> None: + assert configs.resolve_config_source("no-such-config-name") is None + with pytest.raises(FileNotFoundError, match="Input config file not found"): + configs.load_config("no-such-config-name") + + absolute_missing = tmp_path / "missing.json" + assert configs.resolve_config_source(absolute_missing) is None + with pytest.raises(FileNotFoundError, match="not found at absolute path"): + configs.load_config(absolute_missing) + + not_json = tmp_path / "config.txt" + not_json.write_text("{}", encoding="utf-8") + # The source lookup names the existing file; loading still refuses a non-JSON + # file with the historical ValueError instead of silently reading it. + assert configs.resolve_config_source(not_json) == not_json.resolve() + with pytest.raises(ValueError, match="must be a JSON file"): + configs.load_config(not_json) diff --git a/tests/test_sensor_context.py b/tests/test_sensor_context.py new file mode 100644 index 0000000..2a4f61c --- /dev/null +++ b/tests/test_sensor_context.py @@ -0,0 +1,292 @@ +"""Read-only SensorContext (T04): SN/config/assets/cal_path bound once per instance. + +The context is the stage-1 answer to "which physical sensor is this, which +configuration and assets does it run, and which force calibration was selected". +These tests protect the two properties the task names: the binding happens once +from explicit inputs (no SN guessed from enumeration order), and two Sensor +instances do not share or overwrite each other's context. The identity itself is +built through the frozen T03 ``FrameIdentity``, not a second shape. +""" + +from __future__ import annotations + +import dataclasses +import json +from pathlib import Path + +import pytest + +from orisys import Sensor, configs +from orisys.sensors import SensorContext, build_sensor_context +from orisys.stream_contract import FrameIdentity + +# The shipped box calibration has no declared identity; the device tests add one, +# exactly like an operator-managed /.json file. +BOX_CALIBRATION = Path(configs.resolve_sensor_bundle("box").force_calibration_json) +BOX_CONFIG = Path(configs.resolve_sensor_bundle("box").sensor_json) + + +def _device_calibration(root: Path, sn: str, *, family: str = "box") -> Path: + """Write a device calibration whose declared SN is ``sn``.""" + document = json.loads(BOX_CALIBRATION.read_text(encoding="utf-8-sig")) + document["identity"] = {"sensor_type": family, "sn": sn} + path = root / family / f"{sn}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(document), encoding="utf-8") + return path + + +def _external_sensor(**kwargs) -> Sensor: + options = {"verbose": False, "backend": "cpu"} + options.update(kwargs) + return Sensor("external", **options) + + +# ------------------------------------------------------------ explicit binding +def test_context_binds_the_explicit_sn_config_and_calibration_path( + tmp_path: Path, +) -> None: + root = tmp_path / "calibrations" + calibration = _device_calibration(root, "BOX-001") + sensor = _external_sensor( + config_name="box", calibration_dir=root, device_id="BOX-001" + ) + try: + context = sensor.context + assert isinstance(context, SensorContext) + assert context.sensor_id == "BOX-001" + assert sensor.device_id == "BOX-001" + assert context.product_family == "box" + assert context.config_name == "box" + assert context.config_source == str(BOX_CONFIG.resolve()) + assert len(context.config_fingerprint) == 64 + assert context.bundle_root == str(BOX_CONFIG.parent.resolve()) + assert context.calibration_path == str(calibration.resolve()) + assert context.calibration_fingerprint == sensor.calibration.fingerprint + assert context.diagnostics == () + # The published attributes still say what they said before T04. + assert sensor.calibration.source_path == calibration.resolve() + assert sensor.force_unit == "N" + finally: + sensor.disconnect() + + +def test_context_never_guesses_an_sn_from_enumeration_order(tmp_path: Path) -> None: + """No explicit SN means no identity; the stream header refuses to invent one.""" + first = _external_sensor(config_name="box") + second = _external_sensor(config_name="box") + try: + assert first.context.sensor_id is None + assert second.context.sensor_id is None + # Same configuration and bundle, so the non-identity binding is stable. + assert first.context.config_fingerprint == second.context.config_fingerprint + assert first.context.bundle_root == second.context.bundle_root + with pytest.raises(ValueError, match="explicit sensor_id"): + first.context.frame_identity( + session_id="run-1", frame_id=0, source_frame_id=None + ) + finally: + first.disconnect() + second.disconnect() + + +def test_two_instances_keep_separate_contexts(tmp_path: Path) -> None: + root_one = tmp_path / "one" + root_two = tmp_path / "two" + _device_calibration(root_one, "BOX-001") + _device_calibration(root_two, "BOX-002") + first = _external_sensor( + config_name="box", calibration_dir=root_one, device_id="BOX-001" + ) + second = _external_sensor( + config_name="box", calibration_dir=root_two, device_id="BOX-002" + ) + try: + first_context = first.context + first_identity = first_context.frame_identity( + session_id="run-1", frame_id=0, source_frame_id=None + ) + second_context = second.context + second_identity = second_context.frame_identity( + session_id="run-1", frame_id=0, source_frame_id=None + ) + assert first_context != second_context + assert first_identity != second_identity + assert first_identity.sensor_id == "BOX-001" + assert second_identity.sensor_id == "BOX-002" + # Building the second Sensor did not mutate the first context. + assert first.context == first_context + assert first.context.calibration_path != second_context.calibration_path + assert isinstance(first_identity, FrameIdentity) + finally: + first.disconnect() + second.disconnect() + + +def test_context_is_frozen_read_only_value_data(tmp_path: Path) -> None: + root = tmp_path / "calibrations" + _device_calibration(root, "BOX-001") + sensor = _external_sensor( + config_name="box", calibration_dir=root, device_id="BOX-001" + ) + try: + context = sensor.context + with pytest.raises(dataclasses.FrozenInstanceError): + context.sensor_id = "BOX-999" # type: ignore[misc] + payload = context.as_dict() + assert payload["sensor_id"] == "BOX-001" + assert isinstance(payload["assets"], list) + finally: + sensor.disconnect() + + +def test_calibration_swap_replaces_the_context_atomically(tmp_path: Path) -> None: + root = tmp_path / "calibrations" + first_cal = _device_calibration(root, "BOX-001") + # The same device SN, a different calibration document: a live save always + # keeps the device identity, which is what a real recalibration does. + variant = root / "box" / "BOX-001-variant.json" + document = json.loads(BOX_CALIBRATION.read_text(encoding="utf-8-sig")) + document["identity"] = {"sensor_type": "box", "sn": "BOX-001"} + # A real recalibration is a different document; changing one point keeps the + # curve valid while making the content fingerprint different. + document["channels"]["normal"]["points"][1][1] = 0.3771 + variant.write_text(json.dumps(document), encoding="utf-8") + sensor = _external_sensor( + config_name="box", calibration_dir=root, device_id="BOX-001" + ) + try: + before = sensor.context + assert before.calibration_path == str(first_cal.resolve()) + + # The published live-swap entry stays available; the calibration identity + # must still match the bound SN. + sensor.set_calibration(variant) + after = sensor.context + assert after is not before + assert after.sensor_id == before.sensor_id == "BOX-001" + assert after.config_fingerprint == before.config_fingerprint + assert after.bundle_root == before.bundle_root + assert after.calibration_path == str(variant.resolve()) + assert after.calibration_fingerprint == sensor.calibration.fingerprint + assert after.calibration_fingerprint != before.calibration_fingerprint + + sensor.set_calibration(None) + assert sensor.context.calibration_path is None + assert sensor.context.calibration_fingerprint == "none" + assert sensor.force_unit == "raw" + finally: + sensor.disconnect() + + +def test_clearing_a_calibration_loaded_from_a_path_reports_no_path() -> None: + """The context describes the calibration in effect, not the path ever requested. + + Constructing with an explicit ``cal_path`` then clearing the calibration must + leave no calibration path behind: the old path would otherwise be recorded as + if that calibration still shaped the measurements. + """ + sensor = _external_sensor(config_name="box", cal_path=str(BOX_CALIBRATION)) + try: + assert sensor.context.calibration_path == str(BOX_CALIBRATION.resolve()) + assert sensor.force_unit == "N" + + sensor.set_calibration(None) + assert sensor.calibration is None + assert sensor.context.calibration_path is None + assert sensor.context.calibration_fingerprint == "none" + assert sensor.force_unit == "raw" + finally: + sensor.disconnect() + + +def test_binding_failure_is_reported_without_breaking_construction( + tmp_path: Path, monkeypatch +) -> None: + """A context problem must not turn a previously constructible Sensor into an error.""" + import orisys.sensors.context as context_module + + def broken(**kwargs): + raise RuntimeError("bundle exploded") + + monkeypatch.setattr(context_module, "build_sensor_context", broken) + sensor = _external_sensor(config_name="box") + try: + assert sensor.context.sensor_id is None + assert sensor.context.config_name == "box" + assert sensor.context.diagnostics == ( + "context binding failed: RuntimeError: bundle exploded", + ) + assert sensor.context.calibration_fingerprint == "none" + finally: + sensor.disconnect() + + +# --------------------------------------------------- configuration by path/name +def test_config_binding_follows_the_actual_loaded_file(tmp_path: Path) -> None: + custom = tmp_path / "custom-box.json" + custom.write_text(BOX_CONFIG.read_text(encoding="utf-8"), encoding="utf-8") + sensor = _external_sensor(config_name=str(custom)) + try: + context = sensor.context + assert context.config_source == str(custom.resolve()) + assert context.config_name == str(custom) + assert context.product_family is None + assert context.bundle_root is None + assert context.assets == () + assert any("no additive sensor bundle" in line for line in context.diagnostics) + finally: + sensor.disconnect() + + # The recorded source is the file load_config() actually reads. + assert configs.resolve_config_source("box") == BOX_CONFIG.resolve() + assert configs.resolve_config_source(custom) == custom.resolve() + assert configs.resolve_config_source("no-such-config-name") is None + + +def test_same_configuration_has_the_same_config_fingerprint(tmp_path: Path) -> None: + root = tmp_path / "calibrations" + _device_calibration(root, "BOX-001") + first = _external_sensor(config_name="box") + second = _external_sensor( + config_name="box", calibration_dir=root, device_id="BOX-001" + ) + try: + assert first.context.config_fingerprint == second.context.config_fingerprint + assert ( + first.context.processing_backend + == second.context.processing_backend + == "compound_eye" + ) + finally: + first.disconnect() + second.disconnect() + + +# --------------------------------------------------------------- asset binding +def test_bundle_assets_are_bound_as_relative_paths_once() -> None: + finger_config = configs.load_config("finger") + context = build_sensor_context( + config_name="finger", + config=finger_config, + device_id="FINGER-001", + bundle_candidates=["finger"], + ) + assert context.sensor_id == "FINGER-001" + assert context.product_family == "finger" + assert context.bundle_root is not None + assert context.manifest_digest is not None and len(context.manifest_digest) == 64 + roles = {asset.role for asset in context.assets} + assert "assets_dir" in roles + root = Path(context.bundle_root) + for asset in context.assets: + assert not Path(asset.path).is_absolute() + assert (root / asset.path).parent.is_dir() + + # Two bindings of the same bundle describe the same immutable assets. + again = build_sensor_context( + config_name="finger", config=finger_config, bundle_candidates=["finger"] + ) + assert again.assets == context.assets + assert again.manifest_digest == context.manifest_digest + assert again.config_fingerprint == context.config_fingerprint