feat(studio-3d): 设备几何身份接入 —— 同尺寸不同映射/不同表面被拒绝,资产加载只采用最新请求

- What:
  1. **设备侧自己说几何**(新增 `shared/.../acquisition/geometry.py`):`describe_sensor_geometry()`
     从运行时真正加载的文件推导——`manifest.json` 的 `assets_dir`、`sensor.json` 的
     `camera.resolution_stitch`/`source_scale`、以及据此命名的 unroll map(`unroll_map_uv_s0.4.npz`)。
     产出 `DeviceGeometry(expected, verified, detail)`:`field_domain_id` + `shape_hw`(映射身份,含
     flip/scale/映射内容摘要)与 **无单位** 的 `source_geometry_id`(地图里 `world_xyz` 的内容摘要)。
     米制 `geometry_id` 需要"1 源单位 = N 米"这句只有人能做的声明,设备侧**故意留空**。两类失败区分对待:
     定位不到 bundle / 没写 `assets_dir` → `verified=False`(可显示,但不声明已校验);bundle 变了
     (map 缺失、与 `resolution_stitch` 不符)→ 直接拒绝并指名两个形状/文件。
  2. **契约**:`ExpectedGeometry += source_geometry_id`;`SceneAssets += source_geometry_id,
     compatible_sensor_model, compatible_hardware_revisions`;诊断码 `geometry_unverified`(info);
     `digest.source_geometry_digest(world_xyz)`;`legacy.convert()` 写入 `source_geometry_id`;
     `package.py` 的 manifest 增 `source_geometry` 块(数组名与单位写明),读取时把 `compatible` 与
     `source_geometry` 一并还原(旧包读到空串)。`validate_assets()` 的 expected 段重写:**只比较已声明的
     字段**,`source_geometry_id` 不符即拒绝,包内未记录该身份 → 返回 `geometry_unverified` 警告而不是
     静默通过;顺带修掉 `expected.sensor_model != assets.parts[0].role` 这行毫无意义的比较,改成真正的
     型号/硬件版本比对(`*` = 任意版本)。
  3. **接线**:`app_window._sync_3d_source()` 传 `expected_provider=`(一个闭包,指向当前传感器配置);
     `SceneAssetLoader` 在**工作线程**上解析它(`_resolve_and_load`);`view.use_scene(assets, source_id,
     expected)` 改为 `validate_assets(assets, expected)` 并把警告送到面板,同时记录 `device_verified`;
     `_make_snapshot()` 用**设备**的身份构造 adapter(设备没声明的字段才回退资产),使
     "帧与场景匹配" 第一次成为独立检查而不是自证;`describe()` 记录 `scene.device_verified` 与设备身份。
  4. **资产加载请求身份(R7-a)**:`SceneAssetLoader` 增请求号与 pending 槽——忙时不再丢弃新请求,而是
     保留最新一个并在前一个结束后启动;完成信号带请求号(`loaded(assets, 描述, id, device)` /
     `failed(diag, id)`),`Mesh3dPanel` 丢弃过期完成(否则旧文件的场景会落回新选择上)。失败后可重试。
  5. **工具与包**:`compile_scene.py`/`validate_scene.py` 报告 `geometry_id`/`field_domain_id`/
     `source_geometry_id`/兼容信息;README 写明无单位身份是什么、为什么设备侧用它;**重编发布包**
     `apps/studio/config/sensors/finger/scene/finger-display-1.0.0.oryscene`(geometry_id 仍为
     `1a2e0e92…`、1032 顶点 / 1974 三角形,新增 `source_geometry_id=46a03a5f…`,因此 asset_hash 变为 `ea4beab6…`)。
- Why: 审查报告 P1/R3:主窗口只传 sensor_type 与 package_override,`Mesh3dPanel._expected` 永远是 None,
  `validate_assets(assets, None)` 跳过全部设备比对,而 adapter 的 geometry_id/field_domain_id/shape_hw
  全部来自**已选中的资产**——于是收到的数据被贴上"资产自己期望的身份"。结果:shape 相同但 UV 映射、
  flip_v、source_scale 或相机配置不同时无法区分。P1/R7:`SceneAssetLoader.load()` 在忙时直接 return,
  既没保留最新请求也没有 generation,`_on_loaded` 又用当前 `_source` 标记已加载路径,快速切换配置时旧
  文件的任务会回到新选择上。
- Verification:
  · 证据探针 `.repro/probe_stage6_device.py` → `probe_stage6_device.json`(真实 bundle + 真实包):
    设备描述 `verified=true`、domain `feb7f71e…`、source geometry `46a03a5f…`、shape 300x280、
    米制 id 未声明、map = `unroll_map_uv_s0.4.npz`;发布包通过该描述;**同尺寸不同映射**(只改 map 一个
    像素的坐标)→ domain 变化、shape 不变、包被 `field_domain_mismatch` 拒绝;设备什么都不声明 →
    接受 + `geometry_unverified`。
  · 新增 `shared/tests/test_scene3d_device_geometry.py` 9 项:描述来源、发布包通过、同尺寸不同映射拒绝、
    同 map 不同表面拒绝、定位不到 bundle → unverified、`resolution_stitch` 与 map 不符拒绝、map 缺失
    指名文件、无 `assets_dir` → unverified、map 命名规则。
  · `test_scene3d_contracts.py` +4(未声明身份 → 警告而非匹配、source geometry 不符拒绝/缺记录 → 警告、
    型号与硬件版本、`source_geometry_digest` 的稳定性与形状校验);`test_scene3d_package.py` +2
    (`source_geometry` 往返、旧包读成"未声明");`test_scene3d_view.py` +6(adapter 身份来自设备、
    设备 domain 不符 → 拒绝且不上屏、无描述 → `geometry_unverified` 且 `device_verified=false`、
    **最新请求才运行**、面板丢弃过期完成、失败后可重试)。
  · 全量:apps/studio 0 失败(`--junitxml`:231→236 项);shared 0 失败(含 device_geometry 9 项);
    `validate_scene.py` 对重编包全项 ok。
- Not in this change: 生产侧仍不声明米制 `geometry_id`(那句单位声明只有人能写;要设备端复算它需要把
  `meters_per_source_unit` 与刚体变换记进配置,属可选加强项,已在本提交的 `DeviceGeometry.detail` 里留位);
  `ExpectedGeometry.mapping_digest` 未加——`field_domain_id` 本身就是由映射内容摘要派生的,再加一个字段
  只是同一个事实的第二个名字;`docs/architecture/studio-3d-contracts.pyi` 与本阶段新契约的同步放在阶段 6
  的文档提交里;保护包密钥注入(R4)、限速与真实统计、二维后端统一均未动。
This commit is contained in:
ykcheungab
2026-09-14 21:10:53 +08:00
parent 7d1fca4c93
commit 3c0424b90c
17 changed files with 981 additions and 40 deletions
@@ -0,0 +1,191 @@
"""What the acquisition side can say about the geometry it is really running.
The 3D asset carries an identity of its own, but an identity the *asset* stamps onto
the incoming frames proves nothing: if the adapter is built from the asset, then
"the frame matches the scene" is a sentence the asset wrote about itself. The device
side has to make its own statement, and this module derives it from the same files the
runtime loads — the sensor bundle (``manifest.json``, ``sensor.json``) and the unroll
map those name. A map that changed, a different ``source_scale``, a different
resolution or a different bundle then becomes a *mismatch* instead of a picture that
looks plausible.
Two identities are produced, and the difference matters:
* ``field_domain_id`` — the UV mapping the tracker's field actually lives on (shape,
flip, scale, origin and the map's own content digest). The device can always state
this, and it is what catches "same 300x280, different mapping".
* ``source_geometry_id`` — the unit-free identity of the surface behind that mapping.
The metre-scaled ``geometry_id`` needs the operator's "one source unit is N metres"
statement, which no device can make; the unit-free digest needs nothing but the map
file, so this is the geometry check a device *can* perform.
``geometry_id`` is deliberately left empty here. A consumer that sees an empty id must
report ``geometry_unverified`` rather than treat it as a match (see
``contracts.validate_assets``).
"""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Optional
from orisys_app_shared.scene3d.contracts import ExpectedGeometry, error
from orisys_app_shared.scene3d.digest import source_geometry_digest
from orisys_app_shared.scene3d.legacy import load_legacy_domain
logger = logging.getLogger(__name__)
#: The map file name the SDK derives from ``camera.source_scale`` (see the SDK's
#: ``scaled_map_cache_path``): ``unroll_map_uv_s0.4.npz`` for 0.4, and the unscaled
#: ``unroll_map_uv.npz`` for 1.0.
_MAP_STEM = "unroll_map_uv"
_MAP_SUFFIX = ".npz"
@dataclass(frozen=True)
class DeviceGeometry:
"""The device's own description, and whether it is strong enough to verify."""
expected: ExpectedGeometry
#: True when at least one identity was stated (so the asset *was* compared).
verified: bool
#: Paths and identities, for the log and for a recording. Never contains data.
detail: Mapping[str, str]
def map_path_for(assets_dir: Path, source_scale: Optional[float]) -> Path:
"""The unroll map the SDK would load for this scale, inside ``assets_dir``."""
if source_scale is None or abs(float(source_scale) - 1.0) < 1e-9:
return Path(assets_dir) / f"{_MAP_STEM}{_MAP_SUFFIX}"
return Path(assets_dir) / f"{_MAP_STEM}_s{float(source_scale):g}{_MAP_SUFFIX}"
def _read_json(path: Path) -> dict:
try:
data = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return {}
except (OSError, json.JSONDecodeError) as exc:
raise error(
"asset_rejected", f"{path.name} is not readable JSON: {type(exc).__name__}"
) from exc
return data if isinstance(data, dict) else {}
def _bundle_root(config_ref: str | Path) -> Optional[Path]:
path = Path(str(config_ref))
if path.is_file():
return path.parent
if path.is_dir():
return path
return None
def _unverified(sensor_type: str, reason: str, **detail: str) -> DeviceGeometry:
"""Nothing to compare: the pane may show the scene, but not as a verified match."""
logger.warning("no device geometry description for %s: %s", sensor_type, reason)
return DeviceGeometry(
expected=ExpectedGeometry(
sensor_model=str(sensor_type),
hardware_revision="",
geometry_id="",
field_domain_id="",
shape_hw=(0, 0),
source_geometry_id="",
),
verified=False,
detail={"reason": reason, **detail},
)
def describe_sensor_geometry(
sensor_type: str,
config_ref: str | Path,
*,
hardware_revision: str = "",
) -> DeviceGeometry:
"""Derive the device's description of its own geometry.
Runs off the GUI thread (it digests the map once). The two failure kinds are
deliberately different:
* **nothing to describe** (the bundle cannot be located, or names no ``assets_dir``):
an *unverified* description with no identities, so the pane shows the surface and
says it was not checked — a static reference is not a lie;
* **a bundle that changed** (the map it names is missing, or its grid disagrees with
``resolution_stitch``): raises with a diagnostic naming both, because that is
exactly the configuration change the check exists to catch.
"""
root = _bundle_root(config_ref)
if root is None:
return _unverified(
sensor_type, f"the sensor bundle could not be located: {config_ref}"
)
manifest = _read_json(root / "manifest.json")
sensor = _read_json(root / "sensor.json")
assets_dir = manifest.get("assets_dir")
if not assets_dir:
return _unverified(
sensor_type,
f"the sensor bundle {root.name} does not name its 'assets_dir'",
bundle=str(root),
)
assets_path = (root / str(assets_dir)).resolve()
camera = dict(sensor.get("camera") or {})
stitch = camera.get("resolution_stitch") or ()
scale = camera.get("source_scale")
try:
shape_hw = (int(stitch[1]), int(stitch[0]))
except (IndexError, TypeError, ValueError):
return _unverified(
sensor_type,
f"the sensor bundle {root.name} does not state camera.resolution_stitch",
bundle=str(root),
)
map_path = map_path_for(assets_path, scale)
if not map_path.is_file():
raise error(
"asset_rejected",
f"the unroll map for this sensor is missing: {map_path}",
directory=str(assets_path),
)
domain = load_legacy_domain(map_path)
if tuple(domain.shape_hw) != shape_hw:
raise error(
"field_domain_mismatch",
"the unroll map and camera.resolution_stitch disagree about the field grid",
domain=f"{domain.shape_hw[0]}x{domain.shape_hw[1]}",
config=f"{shape_hw[0]}x{shape_hw[1]}",
)
source_geometry_id = (
source_geometry_digest(domain.world_xyz) if domain.world_xyz is not None else ""
)
expected = ExpectedGeometry(
sensor_model=str(sensor_type),
hardware_revision=str(hardware_revision or ""),
# The metre-scaled id needs the unit statement; the device states what it can.
geometry_id="",
field_domain_id=domain.domain_id,
shape_hw=domain.shape_hw,
source_geometry_id=source_geometry_id,
)
detail = {
"bundle": str(root),
"assets_dir": str(assets_path),
"map": str(map_path),
"field_domain_id": domain.domain_id,
"source_geometry_id": source_geometry_id,
"shape_hw": f"{domain.shape_hw[0]}x{domain.shape_hw[1]}",
"source_scale": "" if scale is None else f"{float(scale):g}",
}
verified = bool(domain.domain_id) and bool(source_geometry_id)
if not verified:
logger.warning(
"the device description for %s states no geometry identity: %s",
sensor_type,
detail,
)
return DeviceGeometry(expected=expected, verified=verified, detail=detail)
@@ -61,6 +61,10 @@ DIAGNOSTIC_SEVERITY: Mapping[str, Severity] = {
#: A producer made no statement about where its array came from. The consumer
#: must not assume it is fresh (see ``SdkPacket.flow_valid``).
"field_unverified": "warning",
#: Device and asset could not be compared on one of the identities (the package
#: predates the record, or the device cannot state it). Informational: the pane may
#: show the geometry, but it must not claim the measurement was checked against it.
"geometry_unverified": "info",
"stale_frame": "info",
"gl_unsupported": "error",
"shader_compile_failed": "error",
@@ -228,13 +232,23 @@ class CameraPreset:
@dataclass(frozen=True)
class ExpectedGeometry:
"""What the device says it is; an asset must match it or be refused."""
"""What the device says it is; an asset must match it or be refused.
``geometry_id`` is the metre-scaled identity of the shape, and deriving it needs the
operator's unit statement — so a device that cannot make that statement leaves it
empty and states ``source_geometry_id`` instead: the unit-free identity of the
surface it is actually running, which it *can* compute from the map file it loads.
An empty field is "unstated", never "matches": a comparison skips it and the loader
reports :func:`geometry_unverified` rather than claiming a match nobody made.
"""
sensor_model: str
hardware_revision: str
geometry_id: str
field_domain_id: str
shape_hw: tuple[int, int]
#: Unit-free identity of the source geometry behind the field domain.
source_geometry_id: str = ""
@dataclass(frozen=True)
@@ -462,6 +476,12 @@ class SceneAssets:
#: triangle is this?" is a question an operator will ask, and re-deriving it
#: from UVs alone is a guess.
source_vertex_of_render: Optional[I32] = None
#: Unit-free identity of the source geometry this asset was built from, and what the
#: package says it is compatible with. The first is what a device can verify about
#: the surface; the second catches a package released for another sensor.
source_geometry_id: str = ""
compatible_sensor_model: str = ""
compatible_hardware_revisions: tuple[str, ...] = ()
#: Where this package may be displayed (stage 5). ``None`` means "not locked",
#: which is what every package written before the lock existed carries.
platform_lock: Optional[PlatformLock] = None
@@ -857,29 +877,75 @@ def validate_assets(
f"the platform lock is malformed: {exc.diagnostic.message}",
) from exc
if expected is not None:
if expected.geometry_id != assets.geometry_id:
# Only *stated* fields are compared: an empty one is "the device cannot say",
# which is a reason to report "unverified", never a reason to pass silently.
if expected.geometry_id and expected.geometry_id != assets.geometry_id:
raise error(
"geometry_mismatch",
"asset geometry does not match the device",
asset=assets.geometry_id,
device=expected.geometry_id,
)
if expected.field_domain_id != assets.binding.field_domain_id:
if expected.field_domain_id and expected.field_domain_id != assets.binding.field_domain_id:
raise error(
"field_domain_mismatch",
"asset field domain does not match the device",
asset=assets.binding.field_domain_id,
device=expected.field_domain_id,
)
if tuple(expected.shape_hw) != tuple(assets.binding.shape_hw):
if (
tuple(expected.shape_hw) != (0, 0)
and tuple(expected.shape_hw) != tuple(assets.binding.shape_hw)
):
raise error(
"field_domain_mismatch",
"asset UV shape does not match the device",
asset=f"{assets.binding.shape_hw[0]}x{assets.binding.shape_hw[1]}",
device=f"{expected.shape_hw[0]}x{expected.shape_hw[1]}",
)
if expected.sensor_model and expected.sensor_model != assets.parts[0].role:
pass # the model name is descriptive; the geometry id is the contract
if expected.source_geometry_id:
if not assets.source_geometry_id:
out.append(
diagnostic(
"geometry_unverified",
"the package records no source geometry, so the device's own "
"geometry could not be compared with it",
device=expected.source_geometry_id[:16],
)
)
elif expected.source_geometry_id != assets.source_geometry_id:
raise error(
"geometry_mismatch",
"the asset was built from a different source geometry than the "
"device is running",
asset=assets.source_geometry_id[:16],
device=expected.source_geometry_id[:16],
)
if expected.sensor_model and assets.compatible_sensor_model:
if expected.sensor_model.casefold() != assets.compatible_sensor_model.casefold():
raise error(
"geometry_mismatch",
"the package was released for another sensor model",
asset=assets.compatible_sensor_model,
device=expected.sensor_model,
)
if expected.hardware_revision and assets.compatible_hardware_revisions:
revisions = tuple(str(item) for item in assets.compatible_hardware_revisions)
if "*" not in revisions and expected.hardware_revision not in revisions:
raise error(
"geometry_mismatch",
"the package was not released for this hardware revision",
asset=",".join(revisions),
device=expected.hardware_revision,
)
if not expected.geometry_id and not expected.source_geometry_id:
out.append(
diagnostic(
"geometry_unverified",
"the device states no geometry identity: the asset is shown as a "
"reference, not as verified against this device",
)
)
for preset in assets.camera_presets:
if not preset.name.strip():
out.append(diagnostic("asset_rejected", "a camera preset has no name"))
@@ -121,6 +121,26 @@ def binding_id(binding: SurfaceBinding) -> str:
)
def source_geometry_digest(world_xyz: Any) -> str:
"""The identity of a source geometry, in the source's own units.
:func:`geometry_id` is the identity of the *sensor-space* shape, and computing it
needs a statement only a person can make (how many metres one source unit is). A
device cannot make that statement, so it could never verify such an id — and
"the device stamped the asset's own id onto the incoming frames" is not a check at
all. This digest needs nothing but the map file the runtime already loads, so the
acquisition side really can compare it against the asset it is about to draw.
"""
array = np.asarray(world_xyz)
if array.ndim != 3 or array.shape[-1] != 3:
raise error(
"geometry_mismatch", f"world_xyz must be [H,W,3], got {tuple(array.shape)}"
)
return content_digest(
{"world_xyz": array}, {"frame": "source", "unit": "source", "arrays": "world_xyz"}
)
def field_domain_id(
*,
shape_hw: tuple[int, int],
@@ -51,6 +51,7 @@ from orisys_app_shared.scene3d.digest import (
content_digest,
field_domain_id,
geometry_id,
source_geometry_digest,
)
#: Keys the importer needs, and what it does when they are missing.
@@ -561,6 +562,13 @@ class LegacyFingerImporter:
bounds_sensor_m=bounds,
field_domain_id=domain.domain_id,
shape_hw=domain.shape_hw,
# Unit-free, computed from the map's own geometry: the only geometry identity
# an acquisition-side description can reproduce without the unit statement.
source_geometry_id=(
source_geometry_digest(domain.world_xyz)
if domain.world_xyz is not None
else ""
),
# The map back to the artist's file: many render vertices may share one
# source vertex, and an operator asking "which triangle is this?"
# deserves an answer rather than a guess.
@@ -172,6 +172,19 @@ def build_manifest(
"sensor_model": str(sensor_model),
"hardware_revisions": [str(item) for item in hardware_revisions],
},
# The unit-free source-geometry identity, so an acquisition-side description can
# verify it (see ``ExpectedGeometry.source_geometry_id``). Absent on a package
# built before this field existed, and then reported as "unverified" rather than
# silently treated as matching.
"source_geometry": (
{
"id": assets.source_geometry_id,
"arrays": ["world_xyz"],
"unit": "source",
}
if assets.source_geometry_id
else None
),
"field_domain": domain,
"parts": [
{
@@ -671,6 +684,8 @@ def scene_from_contents(
recomputed=recomputed_binding[:16],
)
source = arrays.get("sensing/source_vertex.npy")
compatible = dict(manifest.get("compatible") or {})
source_geometry = dict(manifest.get("source_geometry") or {})
assets = SceneAssets(
asset_id=str(manifest.get("asset_id") or ""),
asset_version=str(manifest.get("asset_version") or ""),
@@ -699,6 +714,11 @@ def scene_from_contents(
freeze(source, dtype=np.int32) if source is not None else None
),
platform_lock=PlatformLock.from_dict(manifest.get("platform_lock")),
source_geometry_id=str(source_geometry.get("id") or ""),
compatible_sensor_model=str(compatible.get("sensor_model") or ""),
compatible_hardware_revisions=tuple(
str(item) for item in (compatible.get("hardware_revisions") or ())
),
)
validate_assets(assets, expected)
return assets
+69
View File
@@ -293,6 +293,75 @@ def test_the_device_description_decides_whether_an_asset_fits():
assert excinfo.value.diagnostic.code == code
def test_an_unstated_device_identity_is_reported_not_matched():
"""An empty field is "the device cannot say", never "it matches"."""
device = c.ExpectedGeometry("finger", "", "", "domain-xyz", (3, 4), source_geometry_id="")
warnings = c.validate_assets(_assets(), device)
assert [item.code for item in warnings] == ["geometry_unverified"]
assert device.source_geometry_id == ""
def test_the_source_geometry_must_match_too():
"""Same size, same mapping, different surface: still not the same geometry."""
device = c.ExpectedGeometry(
"finger", "", "", "domain-xyz", (3, 4), source_geometry_id="src-1"
)
assert c.validate_assets(_assets(source_geometry_id="src-1"), device) == ()
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(_assets(source_geometry_id="src-2"), device)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
assert excinfo.value.diagnostic.details["device"] == "src-1"
# A package that predates the record: unverified, and said so.
warnings = c.validate_assets(_assets(), device)
assert [item.code for item in warnings] == ["geometry_unverified"]
def test_a_package_released_for_another_sensor_or_revision_is_refused():
device = c.ExpectedGeometry("finger", "rev-b", "geom-abc", "domain-xyz", (3, 4))
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(_assets(compatible_sensor_model="box"), device)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(
_assets(
compatible_sensor_model="finger",
compatible_hardware_revisions=("rev-a",),
),
device,
)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
# ``*`` is "any revision", so it never refuses.
assert c.validate_assets(
_assets(
compatible_sensor_model="finger",
compatible_hardware_revisions=("*",),
),
device,
) == ()
# An unstated revision on the device side is not a mismatch either.
assert c.validate_assets(
_assets(
compatible_sensor_model="finger",
compatible_hardware_revisions=("rev-a",),
),
c.ExpectedGeometry("finger", "", "geom-abc", "domain-xyz", (3, 4)),
) == ()
def test_the_source_geometry_digest_is_stable_and_shape_checked():
from orisys_app_shared.scene3d.digest import source_geometry_digest
world = np.arange(3 * 4 * 3, dtype=np.float32).reshape(3, 4, 3)
assert source_geometry_digest(world) == source_geometry_digest(world.copy())
moved = world + 1.0
assert source_geometry_digest(moved) != source_geometry_digest(world)
with pytest.raises(c.ThreeDError) as excinfo:
source_geometry_digest(np.zeros((3, 4), dtype=np.float32))
assert excinfo.value.diagnostic.code == "geometry_mismatch"
def test_only_four_vertex_lod_counts_as_the_same_binding():
"""A new topology needs a new binding: 1032 and 1005 are not interchangeable."""
binding = _binding(vertices=3)
@@ -0,0 +1,162 @@
"""Stage 6 acceptance for the *device's own* geometry description.
The review's R3 finding was that the internal-consistency check was never wired to the
acquisition side: the adapter was built from the asset, so "the frame matches the scene"
was a sentence the asset wrote about itself. These tests are about the other half — the
description the acquisition side derives from the files the runtime actually loads, and
what happens when those files and the package disagree.
The bundle used here is the shipped one (``apps/studio/config/sensors/finger``); the
tests that need it skip when it is absent, because the module is also imported by runs
that only have the SDK.
"""
from __future__ import annotations
import json
import shutil
from dataclasses import replace
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.acquisition.geometry import (
describe_sensor_geometry,
map_path_for,
)
from orisys_app_shared.scene3d.contracts import ThreeDError
from orisys_app_shared.scene3d.package import PackagedAssetLoader
BUNDLE = (
Path(__file__).resolve().parents[2]
/ "apps"
/ "studio"
/ "config"
/ "sensors"
/ "finger"
)
PACKAGE = BUNDLE / "scene" / "finger-display-1.0.0.oryscene"
def _require_bundle() -> None:
if not (BUNDLE / "sensor.json").is_file():
pytest.skip("the shipped finger bundle is not present")
def _require_package() -> None:
_require_bundle()
if not PACKAGE.is_file():
pytest.skip("the compiled finger package is not present")
def test_the_map_name_follows_the_scale_the_sdk_would_use(tmp_path):
assert map_path_for(tmp_path, 0.4).name == "unroll_map_uv_s0.4.npz"
assert map_path_for(tmp_path, 1.0).name == "unroll_map_uv.npz"
assert map_path_for(tmp_path, None).name == "unroll_map_uv.npz"
def test_the_device_description_comes_from_the_files_the_runtime_loads():
_require_bundle()
device = describe_sensor_geometry("finger", BUNDLE / "sensor.json")
assert device.verified, device.detail
assert device.expected.sensor_model == "finger"
# The metre-scaled identity needs the operator's unit statement; a device that
# cannot make it must say nothing rather than echo the asset's id back.
assert device.expected.geometry_id == ""
assert device.expected.field_domain_id
assert device.expected.source_geometry_id
assert tuple(device.expected.shape_hw) == (300, 280)
assert Path(device.detail["map"]).name == "unroll_map_uv_s0.4.npz"
assert device.detail["shape_hw"] == "300x280"
def test_the_shipped_package_passes_the_shipped_device_description():
_require_package()
device = describe_sensor_geometry("finger", BUNDLE / "sensor.json")
assets = PackagedAssetLoader().load(PACKAGE, device.expected)
assert assets.source_geometry_id == device.expected.source_geometry_id
assert assets.compatible_sensor_model == "finger"
def test_a_same_size_different_mapping_is_refused(tmp_path):
"""The risk the review named: 300x280 is not one mapping."""
_require_package()
fake = tmp_path / "finger"
shutil.copytree(BUNDLE, fake)
map_path = fake / "assets" / "30fps-0703" / "unroll_map_uv_s0.4.npz"
with np.load(map_path, allow_pickle=False) as handle:
data = {key: handle[key] for key in handle.files}
# The same grid, the same valid area — a different mapping.
data["map_x"] = np.asarray(data["map_x"], dtype=np.float32).copy()
data["map_x"][0, 0] = float(data["map_x"][0, 0]) + 3.0
np.savez(map_path, **data)
shipped = describe_sensor_geometry("finger", BUNDLE / "sensor.json")
other = describe_sensor_geometry("finger", fake / "sensor.json")
assert tuple(other.expected.shape_hw) == tuple(shipped.expected.shape_hw)
assert other.expected.field_domain_id != shipped.expected.field_domain_id, (
"two different mappings of the same size produced the same domain id"
)
with pytest.raises(ThreeDError) as excinfo:
PackagedAssetLoader().load(PACKAGE, other.expected)
assert excinfo.value.diagnostic.code == "field_domain_mismatch"
def test_a_changed_mesh_behind_the_same_map_is_refused(tmp_path):
"""The other half: the map is identical, the surface behind it is not."""
_require_package()
device = describe_sensor_geometry("finger", BUNDLE / "sensor.json")
other_geometry = replace(device.expected, source_geometry_id="somebody-elses-surface")
with pytest.raises(ThreeDError) as excinfo:
PackagedAssetLoader().load(PACKAGE, other_geometry)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
def test_a_bundle_that_cannot_be_described_is_unverified_not_a_match(tmp_path):
_require_bundle()
unknown = tmp_path / "nowhere"
device = describe_sensor_geometry("finger", unknown)
assert device.verified is False
assert device.expected.geometry_id == ""
assert device.expected.source_geometry_id == ""
assert "could not be located" in device.detail["reason"]
def test_a_bundle_whose_map_disagrees_with_the_config_is_refused(tmp_path):
"""``resolution_stitch`` and the map are two statements of one grid."""
_require_bundle()
fake = tmp_path / "finger"
shutil.copytree(BUNDLE, fake)
config = json.loads((fake / "sensor.json").read_text(encoding="utf-8"))
config["camera"]["resolution_stitch"] = [280, 299]
(fake / "sensor.json").write_text(
json.dumps(config, ensure_ascii=False), encoding="utf-8"
)
with pytest.raises(ThreeDError) as excinfo:
describe_sensor_geometry("finger", fake / "sensor.json")
assert excinfo.value.diagnostic.code == "field_domain_mismatch"
def test_a_missing_map_names_the_file_it_looked_for(tmp_path):
_require_bundle()
fake = tmp_path / "finger"
(fake / "assets" / "30fps-0703").mkdir(parents=True)
(fake / "manifest.json").write_text(
json.dumps({"assets_dir": "assets/30fps-0703"}), encoding="utf-8"
)
shutil.copy(BUNDLE / "sensor.json", fake / "sensor.json")
with pytest.raises(ThreeDError) as excinfo:
describe_sensor_geometry("finger", fake / "sensor.json")
assert excinfo.value.diagnostic.code == "asset_rejected"
assert "unroll_map_uv_s0.4.npz" in excinfo.value.diagnostic.message
def test_a_bundle_without_an_assets_dir_is_unverified(tmp_path):
fake = tmp_path / "finger"
fake.mkdir()
(fake / "manifest.json").write_text(json.dumps({"name": "finger"}), encoding="utf-8")
(fake / "sensor.json").write_text(json.dumps({"camera": {}}), encoding="utf-8")
device = describe_sensor_geometry("finger", fake / "sensor.json")
assert device.verified is False
assert "assets_dir" in device.detail["reason"]
+28
View File
@@ -416,3 +416,31 @@ def test_the_envelope_names_the_magic_and_the_payload_state(package):
assert envelope["payload"] == "plain"
assert envelope["entry_count"] == len(read_package(package).arrays)
assert "signature" not in envelope, "a plain package must not look signed"
# ------------------------------------------- the device-verifiable geometry record
def test_the_source_geometry_identity_round_trips(tmp_path, scene):
"""The record a device can reproduce without the operator's unit statement."""
from dataclasses import replace
marked = replace(scene, source_geometry_id="src-46a03a5f")
path = tmp_path / "marked.oryscene"
write_package(marked, path)
manifest = read_package(path).manifest
assert manifest["source_geometry"]["id"] == "src-46a03a5f"
assert manifest["source_geometry"]["unit"] == "source"
rebuilt = scene_from_contents(read_package(path))
assert rebuilt.source_geometry_id == "src-46a03a5f"
assert rebuilt.compatible_sensor_model == "finger"
assert rebuilt.compatible_hardware_revisions == ("*",)
def test_a_package_without_the_record_reads_as_unstated(tmp_path, scene):
"""An older package must not look as if its geometry had been checked."""
path = tmp_path / "plain.oryscene"
write_package(scene, path) # the synthetic scene states no source geometry
manifest = read_package(path).manifest
assert manifest["source_geometry"] is None
rebuilt = scene_from_contents(read_package(path))
assert rebuilt.source_geometry_id == ""
assert rebuilt.compatible_sensor_model == "finger"