feat: 为每个 Sensor 绑定只读 SensorContext

构造时一次绑定 SN、配置、资产和当前生效的标定:只接受显式给出的 SN,不做猜测;标定路径只取当前生效的标定,清除后为 None;换标定时整体替换。帧身份复用 stream-contract 的 FrameIdentity。load_config 改为复用 resolve_config_source,使记录的配置来源与实际读取的文件一致,并补一致性测试。已发布 Sensor 数值和资产包解析不变。回归 275 项、lint 通过。
This commit is contained in:
ykcheungab
2026-10-21 11:04:02 +08:00
parent f6f3c45285
commit 474dbf5a64
7 changed files with 899 additions and 21 deletions
+42 -2
View File
@@ -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/资产包绑定。类型与上下文本身不启动传感器、流水线或会话。
+50 -19
View File
@@ -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/<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)
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(
+54
View File
@@ -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
+11
View File
@@ -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",
+353
View File
@@ -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}",),
)
+97
View File
@@ -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)
+292
View File
@@ -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 <product>/<sn>.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