阶段 2:资产合同 —— 旧 NPZ 导入、坐标/单位、绑定、包容器与工具 - What: shared/scene3d 新增四个模块与两个工具的真实现。digest.py:规范化内容摘要(C 连续小端字节 + 排序 JSON;DIGEST_SCHEMA 版本化),geometry_id(只看 part 角色/顶点/三角形,故意不含法线材质与 LOD 名——改颜色不得让标定失效)、binding_id、field_domain_id(形状、flip_v、source_scale、原点/索引约定、映射内容)、asset_hash(条目摘要 + 去自引用字段的 manifest)。legacy.py:LegacyFingerImporter 按"先单位、后刚体"两步转换(meters_per_source_unit 必填且不得猜、validate_rigid_transform 拒绝含缩放/镜像/非仿射的 4x4);split_at_uv_seams 按 (源顶点, 量化 UV) 拆点,接缝处不平均 UV;vertex_normals 面积加权并给无面顶点有限回退;uv_to_pixel 采用实测约定 x=u*(W-1)、y=(1-v)*(H-1)(flip_v 时);build_binding 生成四点双线性采样表(域内裁剪、零权 tap 丢弃并重分配、未用项 -1/0);uv_correspondence_residual 用 Procrustes 残差把"UV 是否对得上 world_xyz"变成可断言的事实;estimate_meters_per_source_unit 只报告不应用。glb.py:glTF 2.0 子集读取(GLB 头/chunk 校验、accessor/bufferView/stride、节点 TRS 与矩阵烘焙进顶点、法线用逆转置再归一化、缺法线则计算),明确拒绝 Draco/meshopt/量化扩展、稀疏 accessor、外部 buffer、骨骼、形态目标、非 TRIANGLES、越界索引、非有限坐标与超预算三角形;with_housing 把外壳作为独立 part 加入且不动感知几何 id。package.py:.oryscene(zip:envelope.json + manifest.json + sensing/housing *.npy,无 pickle 无代码执行),写入原子(先写 .partial 再替换)、读取逐一校验 magic/envelope 版本/payload 状态/条目路径/manifest 自哈希/逐条目 sha256/dtype/shape/**从包内数组重算 geometry_id 与 binding_id**,加密封套明确报"本版本只开明文包"。tools/scene_assets/validate_scene.py 与 compile_scene.py 真实现(含 --json 报告与退出码:0 通过、2 检查失败、1 参数/文件问题;未给单位时停在 needs_units 并打印实测候选尺度)。测试:shared/tests/test_scene3d_assets.py 33 项(真实 1032 导出与 1005 对比资产的输入样例、接缝/法线/绑定/单位/变换/摘要的失败样例、GLB 往返与拒绝样例)、test_scene3d_package.py 24 项(往返、外壳条目、以及篡改条目/篡改 manifest 哈希/自洽但错误的几何/非有限坐标/多余条目/缺条目/路径穿越/dtype 与 shape 不符/未来版本/加密封套/key provider 的拒绝)。docs/architecture/studio-3d-stages.md 阶段 2 记为完成并附验收结果。 - Why: 阶段 2 要回答的是"这份几何到底是什么、单位是什么、它和哪套 UV 域配套、以及这份包有没有被动过",这些一旦错了会以"三维看起来有点怪"的形式出现且事后无法追溯。因此把 UV 约定做成可测量的事实(真实资产残差 0.0064 vs 错误翻转 0.59)、把单位做成必填参数而不是默认值、把 identity 做成内容摘要并**在读取时重算**,同时用真实资产当输入样例、用篡改后的包当失败样例。 - Verification: 真实资产导入:1032 顶点/1974 三角形/域 300×280/尺寸 15.81×15.0×7.5 mm,validate_assets 通过;UV 残差 0.0064(错翻转 0.59);两套资产混用被拒(geometry_mismatch);缩放/镜像/非仿射变换被拒;compile→validate 端到端 ok(包内条目哈希与 manifest 自哈希全部通过);shared 全量 244 passed;studio 118 passed;ui_check 554 项 0 失败;.repro/verify_assets.py 22 项 0 失败。过程中修掉 5 个真实缺陷:np.load 需要真实文件对象(手写内存 stub 抛空消息异常)、包可以"自洽却错误"(改数组+改哈希后无人重算几何 id,现已重算并比对)、binding 行数不匹配被误报为域错误、np.ptp 平铺比较无意义(换算/不变性断言改为按轴)、以及合成 PoC 场景自己用字符串当 geometry_id(所有人都会抄它,等于绕过检查)。 - Not in this change: 签名与 AES-GCM(阶段 5);LOD 简化;calibration.py 侧的资产哈希登记;三维视图仍未被应用导入(阶段 4 接线)。

This commit is contained in:
ykcheungab
2026-09-14 15:08:29 +08:00
parent cad795c890
commit d0dcac743f
11 changed files with 3065 additions and 58 deletions
+32 -12
View File
@@ -81,23 +81,43 @@ python tools/scene_assets/probe_gl.py --profile desktop33 --output gl-report.jso
--- ---
## 阶段 2:资产合同(几何与坐标) ## 阶段 2:资产合同(几何与坐标)—— 已完成
**做**:`LegacyFingerImporter`(旧 NPZ:`mesh.npz`/`unroll_map_uv_s0.4.npz`/`camera_intrinsics.npz`)、 **做**:`legacy.py`(旧 NPZ 导入:接缝拆点、面积加权法线、UV→域绑定、单位与刚体变换、UV 对应性自检)、
坐标与单位统一(`meters_per_source_unit` → 刚体 `transform_sensor_from_source`)、 `digest.py`(几何/域/资产的内容摘要)、`glb.py`(glTF 2.0 子集读取:烘焙节点变换、法线用逆转置、
`SurfaceBinding` 的生成与自检、`validate_scene.py`、一份简化外壳 GLB 的编译路径、manifest/schema。 拒绝 Draco/骨骼/形态目标/非三角形/外部缓冲)、`package.py`(`.oryscene` zip 容器:manifest + 逐条目
sha256 + 原子写入 + 读取时先验签再相信)、两个工具(`validate_scene.py`、`compile_scene.py`)真实现。
**验收**: **验收(已通过)**:`shared/tests/test_scene3d_assets.py` 33 项、`test_scene3d_package.py` 24 项,
其中需要真实资产的用例在缺资产时自动 skip。要点:
| 用例 | 通过标准 | | 用例 | 结果 |
|---|---| |---|---|
| 拓扑不混用 | 1032 与 1005 两套网格分别加载;错配(binding 顶点数 ≠ 感知面顶点数)被拒 | | 真实 1032 顶点导出导入并校验 | 1032 顶点 / 1974 三角形 / 域 300×280 / 尺寸 15.81×15.0×7.5 mm |
| UV 域 | 300×280 与 500×500 两个域的 `shape_hw`/scale/flip_v 各不相同,错配被拒 | | UV 约定是**测**出来的而不是猜的 | `x=u·(W−1)`、`y=(1−v)·(H−1)`(flip_v=1)相对残差 **0.0064**;错的翻转 **0.59** |
| 采样表 | 每顶点权重和为 1;未用项 `-1`/0;越界下标被拒;接缝不跨域插值 | | 两套资产不得混用 | 1032/300×280 与 1005/1024×1024 分别导入;混用被拒(`geometry_mismatch`,提示 "belongs to a different mesh") |
| 单位 | 三个已知非共线点在表面与外壳中对齐;米/毫米换算与法线方向正确 | | 采样表 | 权重和 = 1;未用项 −1/0;像素正中只用一个 tap;边界外的 tap 被丢弃且权重重分配 |
| 身份 | 改 manifest 任一哈希、改 binding、改 geometry_id 均被拒并给出 `geometry_mismatch`/`field_domain_mismatch` | | 单位与变换是两件事 | 缩放/镜像/非仿射变换被拒;刚体平移逐顶点同向量且各轴跨度不变 |
| 身份 | 同一输入同一哈希;改 manifest 哈希、改条目字节、多/少条目、路径穿越、dtype/shape 不符、加密封套**全部被拒**且诊断码正确 |
| **重新计算几何 id** | 换掉顶点数组并把所有哈希改成自洽后,仍被拒——manifest 的 geometry_id 必须等于**包内数组**的摘要 |
| 外壳 GLB | 节点变换烘焙进顶点、法线用逆转置并归一化、缺法线则计算、三角形预算超限报 `gpu_budget_exceeded` |
**产出样例**:一张 `finger` 场景包 + 一份 `validate_scene.py --json` 报告。 `compile_scene.py` → `validate_scene.py` 端到端在本机跑通(见 `.repro` 之外的仓库内工具;
产物示例:1032 顶点包,`geometry_id` 与 `asset_hash` 见提交信息)。
**这一阶段抓到的**:
1. **`np.load` 需要真正的文件对象**:手写的内存 stub 让它抛出一个空消息的异常,换 `io.BytesIO`。
2. **包可以自我一致却是错的**:改数组 + 改哈希后无人重算几何 id,于是"把别的网格贴上你的
geometry_id"能通过全部哈希校验。现在加载器从包内数组重算 sensing 几何 id 与 binding id 并比对。
3. **`np.ptp()` 平铺比较没有意义**(跨轴 max−min):换算与刚体不变性的断言必须按轴比较。
4. **合成场景自己违反了合同**:`poc_scene()` 之前用字符串 `"poc-geometry"` 当 id——所有人都会抄它,
于是那个检查形同虚设;现在 PoC 场景的身份也是内容摘要。
5. **binding 行数不匹配被报成域错误**:1032 行的绑定配 1005 顶点的网格是 *几何* 不一致,
按规格改为 `geometry_mismatch`。
**明确未做**:签名与加密(阶段 5);LOD 简化(第一版用全分辨率绑定);`housing` 的 GLB 目前只支持
单文件、无扩展的子集(按规格如此)。
--- ---
@@ -233,6 +233,12 @@ class SceneAssets:
bounds_sensor_m: F32 # [2,3]: min, max bounds_sensor_m: F32 # [2,3]: min, max
field_domain_id: str = "" field_domain_id: str = ""
shape_hw: tuple[int, int] = (0, 0) shape_hw: tuple[int, int] = (0, 0)
#: For each sensing render vertex, the source vertex it came from. Splitting a
#: mesh at a UV seam creates render vertices that share one source vertex, so
#: the map back to the artist's file is many-to-one. Kept because "which
#: 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
def part(self, part_id: str) -> Optional[MeshPart]: def part(self, part_id: str) -> Optional[MeshPart]:
for item in self.parts: for item in self.parts:
@@ -440,6 +446,19 @@ def validate_binding(binding: SurfaceBinding, *, vertices: int) -> tuple[Diagnos
("sample_weights", binding.sample_weights), ("sample_weights", binding.sample_weights),
): ):
_readonly_c_contiguous(name, array) _readonly_c_contiguous(name, array)
# A row count that does not match the sensing mesh is a *geometry* error: the
# binding was built for a different mesh (1032 rows against a 1005-vertex
# export), not a differently shaped UV field.
for name, array in (
("sample_indices", binding.sample_indices),
("sample_weights", binding.sample_weights),
):
if int(array.shape[0]) != int(vertices):
raise error(
"geometry_mismatch",
f"{name} has {int(array.shape[0])} rows for {int(vertices)} sensing "
"vertices: this binding belongs to a different mesh",
)
_shape("sample_indices", binding.sample_indices, (vertices, 4)) _shape("sample_indices", binding.sample_indices, (vertices, 4))
_shape("sample_weights", binding.sample_weights, (vertices, 4)) _shape("sample_weights", binding.sample_weights, (vertices, 4))
if binding.sample_indices.dtype != np.int32: if binding.sample_indices.dtype != np.int32:
@@ -533,6 +552,19 @@ def validate_assets(
_readonly_c_contiguous("bounds_sensor_m", assets.bounds_sensor_m) _readonly_c_contiguous("bounds_sensor_m", assets.bounds_sensor_m)
_shape("bounds_sensor_m", assets.bounds_sensor_m, (2, 3)) _shape("bounds_sensor_m", assets.bounds_sensor_m, (2, 3))
validate_binding(assets.binding, vertices=int(sensing[0].positions_sensor_m.shape[0])) validate_binding(assets.binding, vertices=int(sensing[0].positions_sensor_m.shape[0]))
if assets.source_vertex_of_render is not None:
_readonly_c_contiguous("source_vertex_of_render", assets.source_vertex_of_render)
_shape(
"source_vertex_of_render",
assets.source_vertex_of_render,
(int(sensing[0].positions_sensor_m.shape[0]),),
)
if assets.source_vertex_of_render.dtype != np.int32:
raise error("geometry_mismatch", "source_vertex_of_render must be int32")
if assets.source_vertex_of_render.size and int(
assets.source_vertex_of_render.min()
) < 0:
raise error("geometry_mismatch", "a source vertex index is negative")
out: list[Diagnostic] = [] out: list[Diagnostic] = []
if expected is not None: if expected is not None:
@@ -0,0 +1,161 @@
"""Stable content digests: geometry identity, field-domain identity, asset hash.
``docs/architecture/studio-3d-design.md`` §5.3 is specific about what a hash has to
be, and this module exists to obey it rather than to invent a scheme:
* **deterministic**: the same arrays and the same metadata give the same digest on
any machine, so a package hash means something;
* **explicit about what it covers**: geometry identity is positions and triangles;
display-only concerns (materials, colours, normals, LOD names) are *not* part of
it, because changing a colour must not invalidate a calibration;
* **never self-referential**: an asset hash is computed over the entries and the
canonical manifest *without* the hash fields themselves.
Plain SHA-256 over canonical bytes. It detects corruption and accidental mixing; it
does not authenticate anything, and the design document is explicit that a hash is
not a signature (that is stage 5).
"""
from __future__ import annotations
import hashlib
import json
from typing import Any, Iterable, Mapping
import numpy as np
from orisys_app_shared.scene3d.contracts import MeshPart, SurfaceBinding, error
#: Bumped when the canonical form changes: a digest that silently means something
#: else is worse than no digest at all.
DIGEST_SCHEMA = "orisys-scene-digest/1"
def _array_bytes(array: Any) -> bytes:
"""C-contiguous little-endian bytes: the same on every machine."""
data = np.ascontiguousarray(array)
if data.dtype.byteorder == ">" or (data.dtype.byteorder == "=" and not np.little_endian):
data = data.astype(data.dtype.newbyteorder("<"))
return data.tobytes()
def _feed(hasher: "hashlib._Hash", label: str, payload: bytes) -> None:
hasher.update(label.encode("utf-8"))
hasher.update(b"\0")
hasher.update(str(len(payload)).encode("ascii"))
hasher.update(b"\0")
hasher.update(payload)
hasher.update(b"\0")
def canonical_json(value: Mapping[str, Any]) -> bytes:
"""Sorted, whitespace-free JSON: the byte form a digest can be taken over."""
return json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
).encode("utf-8")
def content_digest(
arrays: Mapping[str, Any] | None = None,
meta: Mapping[str, Any] | None = None,
) -> str:
"""One digest over named arrays and a metadata mapping."""
hasher = hashlib.sha256()
_feed(hasher, "schema", DIGEST_SCHEMA.encode("ascii"))
for name in sorted(arrays or {}):
array = np.asarray((arrays or {})[name])
_feed(hasher, f"array:{name}:dtype", str(array.dtype).encode("ascii"))
_feed(hasher, f"array:{name}:shape", str(tuple(array.shape)).encode("ascii"))
_feed(hasher, f"array:{name}:data", _array_bytes(array))
if meta:
_feed(hasher, "meta", canonical_json(dict(meta)))
return hasher.hexdigest()
def geometry_id(parts: Iterable[MeshPart]) -> str:
"""The identity of the *shape*: part roles, positions and triangles.
Normals, materials, LOD names and camera presets are deliberately excluded: the
first is derived, the rest are display choices, and a scene whose colour
changed is still the same sensor.
"""
arrays: dict[str, Any] = {}
roles: dict[str, str] = {}
for part in sorted(parts, key=lambda item: item.part_id):
arrays[f"{part.part_id}.positions"] = part.positions_sensor_m
arrays[f"{part.part_id}.triangles"] = part.triangles
roles[part.part_id] = part.role
if not arrays:
raise error("geometry_mismatch", "a scene with no parts has no geometry id")
return content_digest(arrays, {"roles": roles, "frame": "sensor", "unit": "m"})
def binding_id(binding: SurfaceBinding) -> str:
"""The sampling table's own identity, so a re-exported LOD is a new binding."""
return content_digest(
{
"indices": binding.sample_indices,
"weights": binding.sample_weights,
},
{
"part_id": binding.part_id,
"geometry_id": binding.geometry_id,
"shape_hw": list(binding.shape_hw),
},
)
def field_domain_id(
*,
shape_hw: tuple[int, int],
flip_v: bool,
source_scale: float | None = None,
origin: str = "top-left",
index_order: str = "row-major",
mapping_digest: str = "",
) -> str:
"""The UV domain's identity.
Two maps of the same size are *not* the same domain if one is flipped or came
from a different remap, so the flip, the scale, the origin convention and the
mapping content are all part of the identity. Anything the sampling depends on
belongs here; anything it does not (a comment, a filename) does not.
"""
height, width = int(shape_hw[0]), int(shape_hw[1])
if height <= 0 or width <= 0:
raise error("field_domain_mismatch", f"invalid field shape {shape_hw}")
return content_digest(
None,
{
"shape_hw": [height, width],
"flip_v": bool(flip_v),
"source_scale": None if source_scale is None else float(source_scale),
"origin": str(origin),
"index_order": str(index_order),
"mapping": str(mapping_digest),
"unit": "1",
},
)
def asset_hash(entries: Mapping[str, str], manifest: Mapping[str, Any]) -> str:
"""The package hash: every entry digest plus the canonical manifest.
The manifest passed in must already have its hash fields removed — this
function cannot know which keys those are, so the writer removes them and the
reader verifies the same way.
"""
payload = {
"entries": {name: entries[name] for name in sorted(entries)},
"manifest": dict(manifest),
}
return content_digest(None, payload)
def file_digest(path: Any) -> str:
"""SHA-256 of a file's bytes, for entry verification."""
hasher = hashlib.sha256()
with open(path, "rb") as handle:
for chunk in iter(lambda: handle.read(1 << 20), b""):
hasher.update(chunk)
return hasher.hexdigest()
+345
View File
@@ -0,0 +1,345 @@
"""Read the subset of glTF 2.0 / GLB that a housing model needs, with no deps.
The design document puts the housing in the same scene as the sensing surface but
gives it its own part, its own buffers and a plain material, and it is explicit
that the runtime is *not* a glTF engine: "离线 exporter 将 GLB 的必要节点、变换、法线
和基础材质编译成紧凑数组;运行时只支持本规范的固定 mesh/material 子集。GLB 动画、
骨骼、任意扩展、Draco 等默认不进入第一版."
This module is that offline half. It reads a binary ``.glb``, bakes the node
transforms into the vertices (so nothing is left for the runtime to multiply),
and refuses the rest loudly instead of importing half of a model:
* only glTF 2.0, only ``TRIANGLES`` primitives, only ``float32``/``uint32``/
``uint16`` accessors, no sparse accessors, no interleaved-with-stride tricks
beyond what ``bufferView`` states;
* ``KHR_draco_mesh_compression``, sparse accessors, morph targets, skins and
animations are refused with ``asset_rejected`` — a silently dropped skin is a
housing that intersects the sensor;
* positions must be finite, indices must be in range, and the triangle count has to
fit the caller's budget.
Normals are taken from the file when present and rotated by the inverse-transpose
of the node transform, then re-normalised; when absent they are computed from the
geometry, because a normal that is merely *stored* is a second copy of the shape
that can disagree with it.
"""
from __future__ import annotations
import json
import struct
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Optional
import numpy as np
from orisys_app_shared.scene3d.contracts import (
Material,
MeshPart,
error,
freeze,
)
GLB_MAGIC = 0x46546C67 # 'glTF'
CHUNK_JSON = 0x4E4F534A # 'JSON'
CHUNK_BIN = 0x004E4942 # 'BIN\0'
GL_TRIANGLES = 4
COMPONENT_TYPES = {5126: np.float32, 5125: np.uint32, 5123: np.uint16}
#: Attribute names the runtime understands; anything else is an extension by
#: another name and is refused rather than ignored.
ALLOWED_ATTRIBUTES = {"POSITION", "NORMAL", "TEXCOORD_0"}
REFUSED_EXTENSIONS = (
"KHR_draco_mesh_compression",
"KHR_mesh_quantization",
"EXT_meshopt_compression",
)
HOUSING_MATERIAL = Material((0.32, 0.33, 0.35, 1.0), "diffuse", True)
@dataclass(frozen=True)
class HousingLimits:
max_triangles: int = 200_000
max_vertices: int = 200_000
def read_glb(path: Path) -> tuple[dict[str, Any], bytes]:
"""The JSON and binary chunks of a ``.glb``, with the header checked."""
raw = Path(path).read_bytes()
if len(raw) < 12:
raise error("asset_rejected", f"{path} is too short to be a GLB")
magic, version, length = struct.unpack_from("<III", raw, 0)
if magic != GLB_MAGIC:
raise error("asset_rejected", f"{path} is not a binary glTF (bad magic)")
if version != 2:
raise error("asset_rejected", f"{path}: glTF version {version} is not supported")
if length != len(raw):
raise error(
"asset_rejected",
f"{path}: the header declares {length} bytes but the file has {len(raw)}",
)
offset = 12
document: Optional[dict[str, Any]] = None
binary = b""
while offset + 8 <= len(raw):
chunk_length, chunk_type = struct.unpack_from("<II", raw, offset)
offset += 8
chunk = raw[offset : offset + chunk_length]
if len(chunk) != chunk_length:
raise error("asset_rejected", f"{path}: a chunk is truncated")
offset += chunk_length
if chunk_type == CHUNK_JSON:
try:
document = json.loads(chunk.decode("utf-8"))
except Exception as exc: # noqa: BLE001
raise error("asset_rejected", f"{path}: invalid JSON chunk: {exc}") from exc
elif chunk_type == CHUNK_BIN:
binary = chunk
if document is None:
raise error("asset_rejected", f"{path}: no JSON chunk")
if int((document.get("asset") or {}).get("version", "0").split(".")[0]) != 2:
raise error("asset_rejected", f"{path}: only glTF 2.0 is supported")
used = set((document.get("extensionsUsed") or []))
refused = sorted(used & set(REFUSED_EXTENSIONS))
if refused:
raise error(
"asset_rejected",
f"{path}: unsupported extensions {refused} (compress the mesh, do not "
"let the runtime guess)",
)
return document, binary
def _accessor_array(document: Mapping[str, Any], binary: bytes, index: int) -> np.ndarray:
accessors = document.get("accessors") or []
views = document.get("bufferViews") or []
if not (0 <= index < len(accessors)):
raise error("asset_rejected", f"accessor {index} is out of range")
accessor = accessors[index]
if accessor.get("sparse") is not None:
raise error("asset_rejected", "sparse accessors are not supported")
component = COMPONENT_TYPES.get(int(accessor.get("componentType", 0)))
if component is None:
raise error(
"asset_rejected",
f"component type {accessor.get('componentType')} is not supported",
)
counts = {"SCALAR": 1, "VEC2": 2, "VEC3": 3, "VEC4": 4, "MAT4": 16}
type_name = str(accessor.get("type"))
if type_name not in counts:
raise error("asset_rejected", f"accessor type {type_name} is not supported")
width = counts[type_name]
count = int(accessor.get("count", 0))
view_index = int(accessor.get("bufferView", -1))
if not (0 <= view_index < len(views)):
raise error("asset_rejected", f"accessor {index} has no usable bufferView")
view = views[view_index]
buffer_index = int(view.get("buffer", 0))
buffers = document.get("buffers") or []
if not (0 <= buffer_index < len(buffers)):
raise error("asset_rejected", "a bufferView points at a missing buffer")
if buffers[buffer_index].get("uri"):
raise error(
"asset_rejected", "external buffers are not supported: pack a single .glb"
)
item_size = np.dtype(component).itemsize * width
stride = int(view.get("byteStride") or item_size)
if stride < item_size:
raise error("asset_rejected", "a bufferView stride is smaller than its item")
start = int(view.get("byteOffset", 0)) + int(accessor.get("byteOffset", 0))
end = start + stride * max(count - 1, 0) + item_size
if end > len(binary):
raise error("asset_rejected", "an accessor reads past the binary chunk")
if stride == item_size:
array = np.frombuffer(
binary, dtype=component, count=count * width, offset=start
).reshape(count, width)
else:
# A strided view has to be copied out; ``np.frombuffer`` cannot express it.
rows = [
np.frombuffer(
binary, dtype=component, count=width, offset=start + row * stride
)
for row in range(count)
]
array = np.asarray(rows)
return np.array(array, copy=True) # own the bytes: the file buffer may move on
def _node_instances(document: Mapping[str, Any]) -> list[tuple[int, str, np.ndarray]]:
"""Every mesh instance in the default scene: (mesh index, name, world matrix)."""
nodes = document.get("nodes") or []
out: list[tuple[int, str, np.ndarray]] = []
scene_index = int(document.get("scene", 0) or 0)
scenes = document.get("scenes") or [{"nodes": list(range(len(nodes)))}]
if not (0 <= scene_index < len(scenes)):
raise error("asset_rejected", "the default scene is out of range")
def transform_of(node: Mapping[str, Any]) -> np.ndarray:
if "matrix" in node:
matrix = np.asarray(node["matrix"], dtype=np.float64).reshape(4, 4).T
return matrix
translation = np.eye(4)
translation[:3, 3] = [float(v) for v in node.get("translation", (0, 0, 0))]
rotation = np.asarray(node.get("rotation", (0, 0, 0, 1)), dtype=np.float64)
if rotation.shape != (4,):
raise error("asset_rejected", "a node rotation is not a quaternion")
x, y, z, w = rotation
norm = float(np.linalg.norm(rotation))
if norm <= 0:
raise error("asset_rejected", "a node rotation is a zero quaternion")
x, y, z, w = (x / norm, y / norm, z / norm, w / norm)
rotation_matrix = np.array(
[
[1 - 2 * (y * y + z * z), 2 * (x * y - z * w), 2 * (x * z + y * w)],
[2 * (x * y + z * w), 1 - 2 * (x * x + z * z), 2 * (y * z - x * w)],
[2 * (x * z - y * w), 2 * (y * z + x * w), 1 - 2 * (x * x + y * y)],
],
dtype=np.float64,
)
scale = np.asarray(node.get("scale", (1, 1, 1)), dtype=np.float64)
if np.any(np.abs(scale) < 1e-12):
raise error("asset_rejected", "a node scale is zero")
matrix = np.eye(4)
matrix[:3, :3] = rotation_matrix @ np.diag(scale)
matrix[:3, 3] = translation[:3, 3]
return matrix
def walk(node_index: int, parent: np.ndarray, depth: int = 0) -> None:
if depth > 32:
raise error("asset_rejected", "the node hierarchy is nested too deeply")
if not (0 <= node_index < len(nodes)):
raise error("asset_rejected", f"node {node_index} is out of range")
node = nodes[node_index]
matrix = parent @ transform_of(node)
if "mesh" in node:
out.append(
(int(node["mesh"]), str(node.get("name") or f"node{node_index}"), matrix)
)
for child in node.get("children", []) or []:
walk(int(child), matrix, depth + 1)
for root in scenes[scene_index].get("nodes", []) or []:
walk(int(root), np.eye(4))
return out
def load_housing_glb(
path: Path,
*,
part_id: str = "housing",
limits: HousingLimits = HousingLimits(),
) -> MeshPart:
"""Read a ``.glb`` into one static housing part, in metres, sensor frame."""
document, binary = read_glb(Path(path))
meshes = document.get("meshes") or []
instances = _node_instances(document)
if not instances:
raise error("asset_rejected", f"{path}: the scene contains no mesh")
all_positions: list[np.ndarray] = []
all_normals: list[np.ndarray] = []
all_triangles: list[np.ndarray] = []
vertex_offset = 0
for mesh_index, node_name, matrix in instances:
if not (0 <= mesh_index < len(meshes)):
raise error("asset_rejected", f"{path}: node {node_name} has no mesh")
mesh = meshes[mesh_index]
if mesh.get("weights") is not None or mesh.get("targets"):
raise error("asset_rejected", "morph targets are not supported")
if any("JOINTS_0" in (primitive.get("attributes") or {}) for primitive in mesh.get("primitives", [])):
raise error("asset_rejected", "skinned meshes are not supported")
for primitive in mesh.get("primitives", []):
mode = int(primitive.get("mode", GL_TRIANGLES))
if mode != GL_TRIANGLES:
raise error(
"asset_rejected", f"primitive mode {mode} is not TRIANGLES"
)
attributes = primitive.get("attributes") or {}
unknown = sorted(set(attributes) - ALLOWED_ATTRIBUTES)
if unknown:
raise error(
"asset_rejected", f"unsupported vertex attributes {unknown}"
)
if "POSITION" not in attributes:
raise error("asset_rejected", "a primitive has no POSITION")
positions = _accessor_array(document, binary, int(attributes["POSITION"]))
if positions.shape[1] != 3:
raise error("asset_rejected", "POSITION is not VEC3")
if not np.isfinite(positions).all():
raise error("asset_rejected", "a position is not finite")
rotation = matrix[:3, :3]
scale = np.linalg.norm(rotation, axis=0)
if np.any(scale <= 1e-12):
raise error("asset_rejected", "a node transform collapses an axis")
# Rotation only for normals: the inverse-transpose of a rigid part is
# the part itself, and re-normalising removes any residual scale.
normal_rotation = np.linalg.inv(rotation).T
transformed = (rotation @ positions.T).T + matrix[:3, 3]
if "NORMAL" in attributes:
normals = _accessor_array(document, binary, int(attributes["NORMAL"]))
if normals.shape != positions.shape:
raise error("asset_rejected", "NORMAL does not match POSITION")
normals = (normal_rotation @ normals.T).T
else:
normals = None
if "indices" in primitive:
indices = _accessor_array(document, binary, int(primitive["indices"]))
indices = np.asarray(indices, dtype=np.int64).reshape(-1)
if indices.size % 3:
raise error("asset_rejected", "an index list is not a triangle list")
if indices.size and int(indices.max()) >= len(positions):
raise error("asset_rejected", "an index is outside POSITION")
triangles = indices.reshape(-1, 3)
else:
if len(positions) % 3:
raise error("asset_rejected", "an unindexed primitive is not triangles")
triangles = np.arange(len(positions), dtype=np.int64).reshape(-1, 3)
if normals is None:
from orisys_app_shared.scene3d.legacy import vertex_normals
normals = vertex_normals(transformed, triangles)
else:
lengths = np.linalg.norm(normals, axis=1, keepdims=True)
normals = normals / np.maximum(lengths, 1e-12)
all_positions.append(transformed)
all_normals.append(normals)
all_triangles.append(triangles + vertex_offset)
vertex_offset += len(positions)
if not all_positions:
raise error("asset_rejected", f"{path}: no triangle geometry")
positions = np.concatenate(all_positions, axis=0)
normals = np.concatenate(all_normals, axis=0)
triangles = np.concatenate(all_triangles, axis=0)
if len(vertices := positions) > limits.max_vertices:
raise error(
"asset_rejected",
f"the housing has {len(vertices)} vertices (limit {limits.max_vertices})",
)
if len(triangles) > limits.max_triangles:
raise error(
"gpu_budget_exceeded",
f"the housing has {len(triangles)} triangles (limit {limits.max_triangles}); "
"simplify it or raise the budget deliberately",
)
return MeshPart(
part_id=str(part_id),
role="housing",
positions_sensor_m=freeze(positions, dtype=np.float32),
normals_sensor=freeze(normals, dtype=np.float32),
triangles=freeze(triangles, dtype=np.uint32),
material=HOUSING_MATERIAL,
lod_id="housing",
)
def with_housing(assets, housing: MeshPart):
"""A new :class:`SceneAssets` carrying one extra static part."""
from dataclasses import replace
if any(part.role == "housing" for part in assets.parts):
raise error("geometry_mismatch", "this scene already has a housing part")
if housing.role != "housing":
raise error("geometry_mismatch", "only a housing part may be added here")
return replace(assets, parts=(*assets.parts, housing))
@@ -0,0 +1,588 @@
"""Import the legacy Blender/NPZ finger export into a validated scene.
Stage 2 in ``docs/architecture/studio-3d-stages.md``. Offline only: this runs on a
build machine or in a test, never in the acquisition loop.
What the importer refuses to do is the important part, because each refusal is a
mistake that would otherwise be discovered as "the 3D view looks a bit wrong":
* **it does not guess units.** ``meters_per_source_unit`` is a required argument:
the mesh is 15.8 x 15.0 x 7.5 source units, and whether one of those is a
millimetre is a physical claim about the sensor, not something a reader of the
file may infer. :func:`estimate_meters_per_source_unit` reports what the data
suggests so a person can decide, and the importer is never given a default.
* **it does not accept a transform that scales.** The contract splits the two
jobs: units change lengths, the transform is rigid. A rotation block that is not
orthonormal is refused with ``geometry_mismatch`` rather than silently doubling
every displacement.
* **it does not mix the two meshes.** ``30fps-0703`` (1032 vertices) and
``visualization`` (1005 vertices) are different topologies *and* different UV
domains; a scene built from one cannot be sampled with the other's binding.
* **it does not average UVs at a seam.** A geometry vertex with several UVs becomes
several render vertices, each carrying its own UV and the same source vertex.
The UV convention (``x = u*(W-1)``, ``y = (1-v)*(H-1)`` when ``flip_v`` is set) is
not a convention picked by taste: :func:`uv_correspondence_residual` measures it
against the asset's own ``world_xyz`` map, and the same function is used as the
acceptance evidence. On the shipped ``30fps-0703`` asset the correct convention
scores a relative residual of 0.006, and the wrong flip scores 0.59.
"""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Optional
import numpy as np
from orisys_app_shared.scene3d.contracts import (
CameraPreset,
ExpectedGeometry,
Material,
MeshPart,
SceneAssets,
SurfaceBinding,
error,
freeze,
)
from orisys_app_shared.scene3d.digest import (
binding_id,
content_digest,
field_domain_id,
geometry_id,
)
#: Keys the importer needs, and what it does when they are missing.
MESH_KEYS = ("vertices", "faces", "uv_data", "loop_to_vertex")
#: The mesh's material: the sensing surface is colour-mapped, never shaded into
#: looking like a different value.
SENSING_MATERIAL = Material((0.85, 0.85, 0.88, 1.0), "unlit_scalar", True)
@dataclass(frozen=True)
class LegacyFieldDomain:
"""The UV domain a sensor's field is defined on, as read from the map file."""
shape_hw: tuple[int, int]
flip_v: bool
source_scale: Optional[float]
valid_mask: np.ndarray # bool [H,W], read-only
domain_id: str
origin: str = "top-left"
index_order: str = "row-major"
mapping_digest: str = ""
world_xyz: Optional[np.ndarray] = None # float32 [H,W,3], for verification only
@property
def valid_ratio(self) -> float:
return float(np.count_nonzero(self.valid_mask)) / float(self.valid_mask.size)
def _open_npz(path: Path, *, what: str) -> dict[str, np.ndarray]:
if not path.is_file():
raise error("asset_rejected", f"{what} not found: {path}")
try:
# ``allow_pickle=False``: an asset is data, never code.
with np.load(path, allow_pickle=False) as handle:
return {key: handle[key] for key in handle.files}
except Exception as exc: # noqa: BLE001
raise error(
"asset_rejected", f"{what} is not a readable NPZ: {type(exc).__name__}: {exc}"
) from exc
def _require(data: Mapping[str, np.ndarray], keys: Iterable[str], *, what: str) -> None:
missing = [key for key in keys if key not in data]
if missing:
raise error(
"asset_rejected", f"{what} is missing {', '.join(missing)}", path=what
)
def load_legacy_domain(map_path: Path) -> LegacyFieldDomain:
"""Read a legacy unroll/remap NPZ as a field domain.
Both asset directories are handled: ``unroll_map_uv_s0.4.npz`` (300x280) and
``flatten_remap.npz`` (1024x1024) carry the same field names.
"""
data = _open_npz(Path(map_path), what="field map")
_require(data, ("valid_mask",), what=str(map_path))
shape = tuple(int(v) for v in data["valid_mask"].shape)
if len(shape) != 2:
raise error("field_domain_mismatch", f"the valid mask is not 2-D: {shape}")
flip_raw = data.get("flip_v")
flip_v = bool(int(np.asarray(flip_raw).reshape(-1)[0])) if flip_raw is not None else False
scale_raw = data.get("source_scale")
source_scale = (
float(np.asarray(scale_raw).reshape(-1)[0]) if scale_raw is not None else None
)
world = data.get("world_xyz")
if world is not None and tuple(int(v) for v in world.shape) != (*shape, 3):
raise error(
"field_domain_mismatch",
f"world_xyz {tuple(world.shape)} does not match the domain {shape}",
)
# The mapping's own content is part of the domain identity: two remaps of the
# same size are not interchangeable.
mapping_arrays: dict[str, Any] = {"valid_mask": data["valid_mask"]}
for key in ("map_x", "map_y"):
if key in data:
mapping_arrays[key] = data[key]
mapping_digest = content_digest(mapping_arrays)
domain_id = field_domain_id(
shape_hw=(shape[0], shape[1]),
flip_v=flip_v,
source_scale=source_scale,
mapping_digest=mapping_digest,
)
return LegacyFieldDomain(
shape_hw=(shape[0], shape[1]),
flip_v=flip_v,
source_scale=source_scale,
valid_mask=freeze(np.asarray(data["valid_mask"], dtype=bool), dtype=np.bool_),
domain_id=domain_id,
mapping_digest=mapping_digest,
world_xyz=(
freeze(np.asarray(world, dtype=np.float32), dtype=np.float32)
if world is not None
else None
),
)
def uv_to_pixel(u: float, v: float, domain: LegacyFieldDomain) -> tuple[float, float]:
"""Map a UV to the domain's pixel grid.
``x = u*(W-1)``, and ``y = (1-v)*(H-1)`` when the map is v-flipped. The
``(W-1)`` form addresses the pixels the map actually defines rather than their
centres, which is what the correspondence check confirms on the real asset.
"""
height, width = domain.shape_hw
x = float(u) * (width - 1)
y = (1.0 - float(v)) * (height - 1) if domain.flip_v else float(v) * (height - 1)
return x, y
def split_at_uv_seams(
vertices: np.ndarray,
faces: np.ndarray,
uv_data: np.ndarray,
loop_to_vertex: np.ndarray,
) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
"""Split the mesh so every render vertex has exactly one UV.
Returns ``(positions, uvs, triangles, source_vertices)``. A geometry vertex
that owns several UVs (a seam) becomes several render vertices sharing the same
source index; their UVs are never averaged, because an averaged UV samples a
point that is not on either side of the seam.
"""
vertices = np.asarray(vertices)
faces = np.asarray(faces, dtype=np.int64)
uv_data = np.asarray(uv_data)
loop_to_vertex = np.asarray(loop_to_vertex, dtype=np.int64)
if vertices.ndim != 2 or vertices.shape[1] != 3:
raise error("geometry_mismatch", f"vertices must be [N,3], got {vertices.shape}")
if faces.ndim != 2 or faces.shape[1] != 3:
raise error("geometry_mismatch", f"faces must be [M,3], got {faces.shape}")
if faces.size and (faces.min() < 0 or faces.max() >= len(vertices)):
raise error("geometry_mismatch", "a face index is outside the vertex array")
corners = len(faces) * 3
if len(uv_data) != corners or len(loop_to_vertex) != corners:
raise error(
"geometry_mismatch",
f"expected {corners} per-corner UVs and loop indices, got "
f"{len(uv_data)} and {len(loop_to_vertex)}",
)
keyed: dict[tuple[int, int, int], int] = {}
positions: list[np.ndarray] = []
uvs: list[tuple[float, float]] = []
source_of_render: list[int] = []
triangles = np.empty((len(faces), 3), dtype=np.uint32)
for face_index in range(len(faces)):
for corner in range(3):
loop_index = face_index * 3 + corner
source = int(loop_to_vertex[loop_index])
u, v = (float(uv_data[loop_index][0]), float(uv_data[loop_index][1]))
# A quantised key: identical UVs on the same vertex are one render
# vertex, and float noise at a seam must not create duplicates.
key = (source, int(round(u * 1e6)), int(round(v * 1e6)))
index = keyed.get(key)
if index is None:
index = len(positions)
keyed[key] = index
positions.append(vertices[source])
uvs.append((u, v))
source_of_render.append(source)
triangles[face_index, corner] = index
return (
np.asarray(positions),
np.asarray(uvs),
triangles,
np.asarray(source_of_render, dtype=np.int64),
)
def vertex_normals(positions: np.ndarray, triangles: np.ndarray) -> np.ndarray:
"""Area-weighted vertex normals: unit length, computed from the geometry.
Derived rather than stored, so a normal is never a second, disagreeing copy of
the shape. A vertex used by no triangle (or by a degenerate one only) gets a
finite fallback instead of a NaN.
"""
positions = np.asarray(positions, dtype=np.float64)
triangles = np.asarray(triangles, dtype=np.int64)
normals = np.zeros_like(positions)
if len(triangles):
a = positions[triangles[:, 0]]
b = positions[triangles[:, 1]]
c = positions[triangles[:, 2]]
# Not normalised: the cross product's length is twice the triangle area,
# which is exactly the weight a vertex normal wants.
face_normals = np.cross(b - a, c - a)
for corner in range(3):
np.add.at(normals, triangles[:, corner], face_normals)
lengths = np.linalg.norm(normals, axis=1, keepdims=True)
fallback = 1e-12
if np.any(lengths <= fallback):
# A vertex with no usable face: point it along +Z rather than emit a NaN
# that would later look like a measurement problem.
normals[lengths.ravel() <= fallback] = np.array([0.0, 0.0, 1.0])
lengths = np.linalg.norm(normals, axis=1, keepdims=True)
return normals / np.maximum(lengths, fallback)
def _apply_rigid(
positions: np.ndarray, transform_source_from_sensor: np.ndarray
) -> np.ndarray:
"""``p_sensor = T @ p_source`` with a rigid 4x4."""
homogeneous = np.concatenate(
[positions, np.ones((len(positions), 1), dtype=positions.dtype)], axis=1
)
return (transform_source_from_sensor @ homogeneous.T).T[:, :3]
def validate_rigid_transform(transform: Any, *, name: str = "transform") -> np.ndarray:
"""A 4x4 whose rotation block is orthonormal with determinant +1.
A transform carrying scale is the classic way a "millimetres" claim becomes
wrong by a factor nobody notices, so it is refused outright (design document
§8.1).
"""
matrix = np.asarray(transform, dtype=np.float64)
if matrix.shape != (4, 4):
raise error("geometry_mismatch", f"{name} must be 4x4, got {matrix.shape}")
if not np.isfinite(matrix).all():
raise error("geometry_mismatch", f"{name} contains non-finite values")
rotation = matrix[:3, :3]
if not np.allclose(rotation @ rotation.T, np.eye(3), atol=1e-6):
raise error(
"geometry_mismatch",
f"{name} is not a rigid rotation (its rows are not orthonormal): it "
"would scale the geometry as well",
)
determinant = float(np.linalg.det(rotation))
if abs(determinant - 1.0) > 1e-6:
raise error(
"geometry_mismatch",
f"{name} has determinant {determinant:.6f}; a reflection would mirror "
"the sensor and flip every triangle's winding",
)
if not np.allclose(matrix[3], [0.0, 0.0, 0.0, 1.0], atol=1e-9):
raise error("geometry_mismatch", f"{name} is not an affine transform")
return matrix
def build_binding(
render_uv: np.ndarray,
domain: LegacyFieldDomain,
*,
part_id: str,
geometry_id_value: str,
) -> SurfaceBinding:
"""Bilinear taps for every render vertex, inside the declared domain.
Each row holds up to four ``y*W+x`` indices and their weights. A neighbour that
would fall outside the domain contributes nothing and its weight is
redistributed, so the used weights always sum to exactly one and the unused
entries are ``-1``/``0`` — the sampling table can never point outside the map
it was built for.
"""
height, width = domain.shape_hw
count = len(render_uv)
indices = np.full((count, 4), -1, dtype=np.int32)
weights = np.zeros((count, 4), dtype=np.float32)
for row in range(count):
u, v = float(render_uv[row][0]), float(render_uv[row][1])
x, y = uv_to_pixel(u, v, domain)
# Clamp into the map: a UV exactly on the edge still samples the edge, it
# does not sample outside.
x = min(max(x, 0.0), float(width - 1))
y = min(max(y, 0.0), float(height - 1))
x0 = int(np.floor(x))
y0 = int(np.floor(y))
x1 = min(x0 + 1, width - 1)
y1 = min(y0 + 1, height - 1)
fx = x - x0
fy = y - y0
taps = (
(y0 * width + x0, (1.0 - fx) * (1.0 - fy)),
(y0 * width + x1, fx * (1.0 - fy)),
(y1 * width + x0, (1.0 - fx) * fy),
(y1 * width + x1, fx * fy),
)
used = [(index, weight) for index, weight in taps if weight > 1e-7]
total = sum(weight for _, weight in used)
if total <= 0.0: # numerically impossible, but a row must never be empty
used = [(y0 * width + x0, 1.0)]
total = 1.0
for slot, (index, weight) in enumerate(used[:4]):
indices[row, slot] = index
weights[row, slot] = weight / total
return SurfaceBinding(
part_id=part_id,
geometry_id=geometry_id_value,
field_domain_id=domain.domain_id,
shape_hw=domain.shape_hw,
sample_indices=freeze(indices, dtype=np.int32),
sample_weights=freeze(weights, dtype=np.float32),
)
def estimate_meters_per_source_unit(
positions: np.ndarray, expected_extent_m: Optional[float] = None
) -> float:
"""What the mesh's units would have to be for its extent to be a real size.
Reported, never applied: the importer requires the caller to state the unit, and
this is the number a person should confirm against the sensor's datasheet
(a 15.8-unit span is 15.8 mm for the shipped finger sensor).
"""
extent = float(np.ptp(np.asarray(positions, dtype=np.float64), axis=0).max())
if extent <= 0.0:
raise error("geometry_mismatch", "the mesh has no extent to scale")
if expected_extent_m is None:
return 1.0 / extent
return float(expected_extent_m) / extent
def uv_correspondence_residual(
positions: np.ndarray,
render_uv: np.ndarray,
domain: LegacyFieldDomain,
*,
max_samples: int = 4000,
) -> tuple[float, int]:
"""How well the UVs agree with the asset's own ``world_xyz`` map.
Returns ``(relative residual, samples)``: the root-mean-square error left after
the best similarity fit between the mesh and the sampled map, divided by the
cloud's spread. A correct convention lands well under 1%; a wrong v flip is
around 50%. This is the evidence behind the convention, and the acceptance test
asserts it on the real asset.
"""
if domain.world_xyz is None:
raise error("field_domain_mismatch", "this domain carries no world_xyz to check")
height, width = domain.shape_hw
mesh_points: list[np.ndarray] = []
map_points: list[np.ndarray] = []
step = max(1, len(render_uv) // max_samples)
for row in range(0, len(render_uv), step):
x, y = uv_to_pixel(*render_uv[row], domain)
xi, yi = int(round(x)), int(round(y))
if not (0 <= xi < width and 0 <= yi < height):
continue
point = np.asarray(domain.world_xyz[yi, xi], dtype=np.float64)
if not np.isfinite(point).all():
continue
mesh_points.append(np.asarray(positions[row], dtype=np.float64))
map_points.append(point)
if len(mesh_points) < 20:
raise error(
"field_domain_mismatch",
f"only {len(mesh_points)} vertices land on a mapped pixel: the domain "
"does not belong to this mesh",
)
source = np.asarray(mesh_points)
target = np.asarray(map_points)
source_centered = source - source.mean(axis=0)
target_centered = target - target.mean(axis=0)
u, _, vt = np.linalg.svd(source_centered.T @ target_centered)
sign = np.sign(np.linalg.det(vt.T @ u.T))
rotation = vt.T @ np.diag([1.0, 1.0, sign]) @ u.T
scale = float(
np.linalg.norm(target_centered) / max(np.linalg.norm(source_centered), 1e-12)
)
fitted = scale * (source_centered @ rotation.T)
residual = float(np.sqrt(((fitted - target_centered) ** 2).sum(axis=1).mean()))
spread = float(np.sqrt((target_centered**2).sum(axis=1).mean()))
return residual / max(spread, 1e-12), len(mesh_points)
class LegacyFingerImporter:
"""Convert a legacy export directory into a validated :class:`SceneAssets`."""
def convert(
self,
export_dir: Path,
map_path: Path,
*,
transform_sensor_from_source: Any,
meters_per_source_unit: float,
asset_id: str = "finger-display",
asset_version: str = "1.0.0",
lod_id: str = "lod0",
camera_presets: Optional[Iterable[CameraPreset]] = None,
) -> SceneAssets:
export_dir = Path(export_dir)
map_path = Path(map_path)
units = float(meters_per_source_unit)
if not np.isfinite(units) or units <= 0.0:
raise error(
"geometry_mismatch",
"meters_per_source_unit must be finite and positive: the importer "
"does not guess what a source unit is",
)
transform = validate_rigid_transform(
transform_sensor_from_source, name="transform_sensor_from_source"
)
data = _open_npz(export_dir / "mesh.npz", what="mesh")
_require(data, MESH_KEYS, what=str(export_dir / "mesh.npz"))
positions, uvs, triangles, source_of_render = split_at_uv_seams(
data["vertices"], data["faces"], data["uv_data"], data["loop_to_vertex"]
)
domain = load_legacy_domain(map_path)
if len(positions) != len(uvs):
raise error("geometry_mismatch", "render vertices and UVs disagree")
# Units first, then the rigid transform — in that order, so the transform
# never has to carry a scale (design document §8.1).
positions_m = positions.astype(np.float64) * units
normals = vertex_normals(positions_m, triangles)
positions_sensor = _apply_rigid(positions_m, transform)
# Rotating the rest normals is enough: the transform has no scale, so the
# inverse-transpose is the same rotation.
normals_sensor = normals @ transform[:3, :3].T
if not np.isfinite(positions_sensor).all():
raise error("geometry_mismatch", "the transform produced non-finite positions")
geometry_value = geometry_id(
[
MeshPart(
part_id="sensing_surface",
role="sensing",
positions_sensor_m=freeze(positions_sensor, dtype=np.float32),
normals_sensor=freeze(normals_sensor, dtype=np.float32),
triangles=freeze(triangles, dtype=np.uint32),
material=SENSING_MATERIAL,
lod_id=lod_id,
)
]
)
binding = build_binding(
uvs,
domain,
part_id="sensing_surface",
geometry_id_value=geometry_value,
)
sensing = MeshPart(
part_id="sensing_surface",
role="sensing",
positions_sensor_m=freeze(positions_sensor, dtype=np.float32),
normals_sensor=freeze(normals_sensor, dtype=np.float32),
triangles=freeze(triangles, dtype=np.uint32),
material=SENSING_MATERIAL,
lod_id=lod_id,
)
bounds = freeze(
np.stack(
[
positions_sensor.min(axis=0),
positions_sensor.max(axis=0),
]
).astype(np.float32),
dtype=np.float32,
)
assets = SceneAssets(
asset_id=str(asset_id),
asset_version=str(asset_version),
asset_hash=content_digest(
{
"positions": sensing.positions_sensor_m,
"triangles": sensing.triangles,
"indices": binding.sample_indices,
"weights": binding.sample_weights,
},
{
"geometry_id": geometry_value,
"binding_id": binding_id(binding),
"domain_id": domain.domain_id,
"unit": "m",
},
),
geometry_id=geometry_value,
parts=(sensing,),
binding=binding,
camera_presets=tuple(camera_presets or default_finger_presets(bounds)),
bounds_sensor_m=bounds,
field_domain_id=domain.domain_id,
shape_hw=domain.shape_hw,
# 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.
source_vertex_of_render=freeze(source_of_render, dtype=np.int32),
)
return assets
def validate(
self, assets: SceneAssets, expected: ExpectedGeometry
) -> tuple[Any, ...]:
from orisys_app_shared.scene3d.contracts import validate_assets
return validate_assets(assets, expected)
def default_finger_presets(bounds_sensor_m: np.ndarray) -> tuple[CameraPreset, ...]:
"""The default views for a flat sensing surface, in sensor metres.
Derived from the bounds rather than hard-coded: the surface faces +Z after the
transform, and a three-quarter view is the one that shows both its shape and
its relief.
"""
low, high = np.asarray(bounds_sensor_m, dtype=np.float64)
center = (low + high) * 0.5
span = float(np.max(high - low))
distance = max(span, 1e-4)
return (
CameraPreset(
"default",
tuple(center + np.array([-0.6, -0.6, 1.1]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
CameraPreset(
"front",
tuple(center + np.array([0.0, 0.0, 1.4]) * distance),
tuple(center),
(0.0, 1.0, 0.0),
0.6,
),
CameraPreset(
"side",
tuple(center + np.array([1.4, 0.0, 0.0]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
CameraPreset(
"top",
tuple(center + np.array([0.0, 1.4, 0.0]) * distance),
tuple(center),
(0.0, 0.0, 1.0),
0.6,
),
)
def identity_transform() -> np.ndarray:
return np.eye(4, dtype=np.float64)
@@ -0,0 +1,503 @@
"""The ``.oryscene`` runtime package: manifest, entries, verification.
Stage 2 in ``docs/architecture/studio-3d-stages.md``; signing and encryption are
stage 5 and are *not* here. What is here is the part that stage 5 wraps: a single
file whose every byte is accounted for in a manifest, and a reader that verifies
before it believes.
The container is a zip archive (stdlib, no pickle, no code execution) holding:
``envelope.json``
magic, envelope version, payload state, entry count. Stage 5 adds the key id,
nonce and signature here.
``manifest.json``
canonical JSON: identity, coordinate frame, units, the UV domain, the parts,
and one entry record (dtype, shape, sha256) per array.
``sensing/*.npy`` (and ``housing/*.npy`` when present)
plain ``.npy`` arrays, written by :func:`numpy.save` into the archive.
The reader refuses, without trying to repair anything: a wrong magic, an unknown
envelope version, a missing manifest, an entry whose bytes do not match its
recorded hash, an entry the manifest does not describe (or describes and never
provides), a path that tries to escape, a dtype/shape that disagrees with the
record, non-finite positions, a binding that belongs to another mesh, and an asset
that does not match the device. An encrypted envelope is refused with its own
message rather than silently opened as if it were plain.
"""
from __future__ import annotations
import io
import json
import zipfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Optional
import numpy as np
from orisys_app_shared.scene3d.contracts import (
CameraPreset,
CompileReport,
Diagnostic,
ExpectedGeometry,
Material,
MeshPart,
SceneAssets,
SurfaceBinding,
error,
freeze,
validate_assets,
)
from orisys_app_shared.scene3d.digest import (
DIGEST_SCHEMA,
asset_hash as compute_asset_hash,
binding_id,
content_digest,
geometry_id as compute_geometry_id,
)
#: Bumped when the container layout changes incompatibly.
ENVELOPE_VERSION = 1
SCHEMA_VERSION = "1.0"
MAGIC = "orisys-scene"
#: Refuse an archive that claims more than this, before reading anything: a
#: package is geometry, not a video.
MAX_ENTRIES = 64
MAX_ENTRY_BYTES = 256 << 20
MAX_TOTAL_BYTES = 512 << 20
#: The arrays a sensing part is made of, and the role each belongs to.
SENSING_ARRAYS = {
"sensing/positions.npy": ("f4", 2),
"sensing/normals.npy": ("f4", 2),
"sensing/triangles.npy": ("u4", 2),
"sensing/sample_indices.npy": ("i4", 2),
"sensing/sample_weights.npy": ("f4", 2),
}
OPTIONAL_ARRAYS = {"sensing/source_vertex.npy": ("i4", 1)}
HOUSING_ARRAYS = {
"housing/positions.npy": ("f4", 2),
"housing/normals.npy": ("f4", 2),
"housing/triangles.npy": ("u4", 2),
}
_DTYPE_BY_KEY = {"f4": np.float32, "u4": np.uint32, "i4": np.int32}
_KEY_BY_DTYPE = {np.dtype(v): k for k, v in _DTYPE_BY_KEY.items()}
@dataclass(frozen=True)
class PackageContents:
"""A verified package, still as arrays: the manifest and what it describes."""
manifest: Mapping[str, Any]
arrays: Mapping[str, np.ndarray]
diagnostics: tuple[Diagnostic, ...] = ()
def _entry_name(name: str) -> str:
"""Refuse anything that is not a plain relative path inside the package."""
text = str(name).replace("\\", "/")
if text.startswith("/") or text.startswith("../") or "/../" in text or ":" in text:
raise error("asset_auth_failed", f"entry path escapes the package: {name!r}")
if text not in (*SENSING_ARRAYS, *OPTIONAL_ARRAYS, *HOUSING_ARRAYS):
raise error("asset_auth_failed", f"unknown entry in the package: {name!r}")
return text
def _record(array: np.ndarray) -> dict[str, Any]:
dtype_key = _KEY_BY_DTYPE.get(np.dtype(array.dtype))
if dtype_key is None:
raise error("asset_rejected", f"cannot package dtype {array.dtype}")
data = np.ascontiguousarray(array)
return {
"dtype": dtype_key,
"shape": [int(v) for v in data.shape],
"sha256": content_digest({"data": data}),
}
def build_manifest(
assets: SceneAssets,
*,
entries: Mapping[str, Any],
sensor_model: str = "finger",
hardware_revisions: Iterable[str] = ("*",),
) -> dict[str, Any]:
"""The canonical manifest for a scene. No hash fields: they are added last."""
domain = {
"id": assets.field_domain_id,
"shape_hw": [int(assets.shape_hw[0]), int(assets.shape_hw[1])],
"index_order": "row-major",
"origin": "top-left",
}
return {
"schema_version": SCHEMA_VERSION,
"digest_schema": DIGEST_SCHEMA,
"asset_id": assets.asset_id,
"asset_version": assets.asset_version,
"geometry_id": assets.geometry_id,
"binding_id": binding_id(assets.binding),
"coordinate_frame": "sensor",
"length_unit": "m",
"compatible": {
"sensor_model": str(sensor_model),
"hardware_revisions": [str(item) for item in hardware_revisions],
},
"field_domain": domain,
"parts": [
{
"id": part.part_id,
"role": part.role,
"lod_id": part.lod_id,
"binding": "uv-lod0" if part.role == "sensing" else None,
}
for part in assets.parts
],
"camera_presets": [
{
"name": preset.name,
"eye": list(preset.eye_sensor_m),
"target": list(preset.target_sensor_m),
"up": list(preset.up_sensor),
"fov": float(preset.vertical_fov_rad),
}
for preset in assets.camera_presets
],
"entries": {name: dict(entries[name]) for name in sorted(entries)},
"bounds_sensor_m": [[float(v) for v in row] for row in assets.bounds_sensor_m],
}
def write_package(
assets: SceneAssets,
output: Path,
*,
sensor_model: str = "finger",
hardware_revisions: Iterable[str] = ("*",),
) -> CompileReport:
"""Write a runtime package atomically: a half-written package never appears."""
from orisys_app_shared.scene3d.contracts import validate_assets as _validate
_validate(assets)
output = Path(output)
arrays: dict[str, np.ndarray] = {}
sensing = assets.sensing_part()
arrays["sensing/positions.npy"] = np.ascontiguousarray(sensing.positions_sensor_m)
arrays["sensing/normals.npy"] = np.ascontiguousarray(sensing.normals_sensor)
arrays["sensing/triangles.npy"] = np.ascontiguousarray(sensing.triangles)
arrays["sensing/sample_indices.npy"] = np.ascontiguousarray(assets.binding.sample_indices)
arrays["sensing/sample_weights.npy"] = np.ascontiguousarray(assets.binding.sample_weights)
if assets.source_vertex_of_render is not None:
arrays["sensing/source_vertex.npy"] = np.ascontiguousarray(
assets.source_vertex_of_render
)
for part in assets.parts:
if part.role == "housing":
arrays["housing/positions.npy"] = np.ascontiguousarray(part.positions_sensor_m)
arrays["housing/normals.npy"] = np.ascontiguousarray(part.normals_sensor)
arrays["housing/triangles.npy"] = np.ascontiguousarray(part.triangles)
entries = {name: _record(array) for name, array in arrays.items()}
manifest = build_manifest(
assets, entries=entries, sensor_model=sensor_model,
hardware_revisions=hardware_revisions,
)
digest = compute_asset_hash(
{name: record["sha256"] for name, record in entries.items()}, manifest
)
manifest["asset_hash"] = digest
manifest["entry_digest"] = {name: record["sha256"] for name, record in entries.items()}
envelope = {
"magic": MAGIC,
"envelope_version": ENVELOPE_VERSION,
"payload": "plain",
"schema_version": SCHEMA_VERSION,
"asset_id": assets.asset_id,
"asset_version": assets.asset_version,
"entry_count": len(arrays),
"note": "unsigned and unencrypted: stage 5 adds the signature and the AEAD",
}
output.parent.mkdir(parents=True, exist_ok=True)
temporary = output.with_suffix(output.suffix + ".partial")
try:
with zipfile.ZipFile(temporary, "w", compression=zipfile.ZIP_DEFLATED) as archive:
archive.writestr("envelope.json", json.dumps(envelope, indent=2, sort_keys=True))
archive.writestr("manifest.json", json.dumps(manifest, indent=2, sort_keys=True))
for name in sorted(arrays):
buffer = _npy_bytes(arrays[name])
archive.writestr(name, buffer)
temporary.replace(output)
except Exception:
temporary.unlink(missing_ok=True)
raise
return CompileReport(
output=output,
asset_hash=digest,
part_vertex_counts={part.part_id: int(part.positions_sensor_m.shape[0]) for part in assets.parts},
part_triangle_counts={part.part_id: int(part.triangles.shape[0]) for part in assets.parts},
)
def _npy_bytes(array: np.ndarray) -> bytes:
import io
buffer = io.BytesIO()
np.save(buffer, np.ascontiguousarray(array), allow_pickle=False)
return buffer.getvalue()
def read_package(path: Path, *, keys: Any = None) -> PackageContents:
"""Open, verify and hand back the arrays. Refuses rather than repairs."""
path = Path(path)
if not path.is_file():
raise error("asset_rejected", f"package not found: {path}")
diagnostics: list[Diagnostic] = []
with zipfile.ZipFile(path) as archive:
names = archive.namelist()
if len(names) > MAX_ENTRIES:
raise error("asset_auth_failed", f"the package declares {len(names)} entries")
total = 0
for info in archive.infolist():
total += int(info.file_size)
if int(info.file_size) > MAX_ENTRY_BYTES or total > MAX_TOTAL_BYTES:
raise error("asset_auth_failed", "the package is larger than allowed")
try:
envelope = json.loads(archive.read("envelope.json"))
except KeyError as exc:
raise error("asset_auth_failed", "the envelope is missing") from exc
except Exception as exc: # noqa: BLE001
raise error("asset_auth_failed", f"the envelope is unreadable: {exc}") from exc
if not isinstance(envelope, dict) or envelope.get("magic") != MAGIC:
raise error("asset_auth_failed", "this is not an orisys scene package")
if int(envelope.get("envelope_version", 0)) != ENVELOPE_VERSION:
raise error(
"asset_auth_failed",
f"unsupported envelope version {envelope.get('envelope_version')!r}",
)
if str(envelope.get("payload")) != "plain":
# Stage 5 territory. Saying so is better than pretending the bytes are
# data and failing later with a numpy error.
raise error(
"asset_auth_failed",
f"the payload is {envelope.get('payload')!r}; this build opens plain "
"packages only (signing and encryption are a later stage)",
)
try:
manifest = json.loads(archive.read("manifest.json"))
except KeyError as exc:
raise error("asset_auth_failed", "the manifest is missing") from exc
except Exception as exc: # noqa: BLE001
raise error("asset_auth_failed", f"the manifest is unreadable: {exc}") from exc
if not isinstance(manifest, dict):
raise error("asset_auth_failed", "the manifest is not an object")
entries = manifest.get("entries")
if not isinstance(entries, dict) or not entries:
raise error("asset_auth_failed", "the manifest describes no entries")
recorded_digest = str(manifest.get("asset_hash") or "")
# Verify the asset hash over the entries and the manifest *without* the
# hash fields, exactly as the writer computed it.
stripped = {
key: value
for key, value in manifest.items()
if key not in ("asset_hash", "entry_digest")
}
recomputed = compute_asset_hash(
{name: str(record.get("sha256", "")) for name, record in entries.items()},
stripped,
)
if not recorded_digest or recomputed != recorded_digest:
raise error(
"asset_auth_failed",
"the manifest hash does not match its contents",
recorded=recorded_digest[:16],
recomputed=recomputed[:16],
)
arrays: dict[str, np.ndarray] = {}
for name, record in entries.items():
entry = _entry_name(name)
if name not in names:
raise error("asset_auth_failed", f"the manifest promises {name!r}, which is absent")
raw = archive.read(name)
try:
# ``io.BytesIO``: numpy's reader wants a real file object (it seeks,
# tells and reads headers), and a hand-rolled stub fails obscurely.
array = np.load(io.BytesIO(raw), allow_pickle=False)
except Exception as exc: # noqa: BLE001
raise error(
"asset_auth_failed",
f"{name} is not a readable array: {type(exc).__name__}: {exc}",
) from exc
dtype_key = str(record.get("dtype"))
expected_dtype = _DTYPE_BY_KEY.get(dtype_key)
if expected_dtype is None or array.dtype != np.dtype(expected_dtype):
raise error(
"field_domain_mismatch",
f"{name}: dtype {array.dtype} does not match the manifest {dtype_key!r}",
)
expected_shape = tuple(int(v) for v in record.get("shape", ()))
if tuple(array.shape) != expected_shape:
raise error(
"field_domain_mismatch",
f"{name}: shape {tuple(array.shape)} does not match the manifest "
f"{expected_shape}",
)
if content_digest({"data": np.ascontiguousarray(array)}) != str(
record.get("sha256")
):
raise error("asset_auth_failed", f"{name} does not match its recorded hash")
arrays[name] = array
unknown = [name for name in names if name not in entries and name not in (
"envelope.json", "manifest.json"
)]
if unknown:
raise error("asset_auth_failed", f"undeclared entries: {unknown}")
return PackageContents(manifest=manifest, arrays=arrays, diagnostics=tuple(diagnostics))
def scene_from_contents(
contents: PackageContents, expected: Optional[ExpectedGeometry] = None
) -> SceneAssets:
"""Turn verified arrays into a :class:`SceneAssets` and validate it."""
manifest = contents.manifest
arrays = contents.arrays
domain = dict(manifest.get("field_domain") or {})
shape = tuple(int(v) for v in domain.get("shape_hw", (0, 0)))
parts: list[MeshPart] = []
roles = {
str(part.get("id")): str(part.get("role"))
for part in manifest.get("parts", [])
if isinstance(part, dict)
}
for prefix, role in (("sensing", roles.get("sensing_surface", "sensing")), ("housing", "housing")):
positions_key = f"{prefix}/positions.npy"
if positions_key not in arrays:
continue
part_id = "sensing_surface" if prefix == "sensing" else "housing"
positions = freeze(arrays[positions_key], dtype=np.float32)
if not np.isfinite(positions).all():
raise error("geometry_mismatch", f"{part_id} has non-finite positions")
parts.append(
MeshPart(
part_id=part_id,
role=role, # type: ignore[arg-type]
positions_sensor_m=positions,
normals_sensor=freeze(arrays[f"{prefix}/normals.npy"], dtype=np.float32),
triangles=freeze(arrays[f"{prefix}/triangles.npy"], dtype=np.uint32),
material=(
Material((0.85, 0.85, 0.88, 1.0), "unlit_scalar", True)
if prefix == "sensing"
else Material((0.32, 0.33, 0.35, 1.0), "diffuse", True)
),
lod_id=str(roles.get(part_id, "lod0")),
)
)
if not parts:
raise error("geometry_mismatch", "the package has no mesh parts")
binding = SurfaceBinding(
part_id="sensing_surface",
geometry_id=str(manifest.get("geometry_id") or ""),
field_domain_id=str(domain.get("id") or ""),
shape_hw=(shape[0], shape[1]),
sample_indices=freeze(arrays["sensing/sample_indices.npy"], dtype=np.int32),
sample_weights=freeze(arrays["sensing/sample_weights.npy"], dtype=np.float32),
)
# The manifest's geometry id has to be the id *of the arrays in this package*,
# not a label the package chose. Without this, a release could relabel a
# different mesh as the one a calibration was measured on, and every hash would
# still verify — the identity would be internally consistent and wrong.
sensing_only = tuple(part for part in parts if part.role == "sensing")
recomputed_geometry = compute_geometry_id(sensing_only)
if recomputed_geometry != str(manifest.get("geometry_id") or ""):
raise error(
"geometry_mismatch",
"the packaged geometry does not match the geometry id in the manifest",
manifest=str(manifest.get("geometry_id") or "")[:16],
recomputed=recomputed_geometry[:16],
)
recomputed_binding = binding_id(binding)
recorded_binding = str(manifest.get("binding_id") or "")
if recorded_binding and recomputed_binding != recorded_binding:
raise error(
"geometry_mismatch",
"the packaged sampling table does not match the binding id in the manifest",
manifest=recorded_binding[:16],
recomputed=recomputed_binding[:16],
)
source = arrays.get("sensing/source_vertex.npy")
assets = SceneAssets(
asset_id=str(manifest.get("asset_id") or ""),
asset_version=str(manifest.get("asset_version") or ""),
asset_hash=str(manifest.get("asset_hash") or ""),
geometry_id=str(manifest.get("geometry_id") or ""),
parts=tuple(parts),
binding=binding,
camera_presets=tuple(
CameraPreset(
name=str(item.get("name")),
eye_sensor_m=tuple(float(v) for v in item.get("eye", ())),
target_sensor_m=tuple(float(v) for v in item.get("target", ())),
up_sensor=tuple(float(v) for v in item.get("up", ())),
vertical_fov_rad=float(item.get("fov", 0.6)),
)
for item in manifest.get("camera_presets", [])
if isinstance(item, dict) and item.get("name")
),
bounds_sensor_m=freeze(
np.asarray(manifest.get("bounds_sensor_m") or np.zeros((2, 3)), dtype=np.float32),
dtype=np.float32,
),
field_domain_id=str(domain.get("id") or ""),
shape_hw=(shape[0], shape[1]),
source_vertex_of_render=(
freeze(source, dtype=np.int32) if source is not None else None
),
)
validate_assets(assets, expected)
return assets
class PackagedAssetLoader:
"""The stage-2 loader: plain, verified packages (stage 5 adds the envelope)."""
def load(
self,
path: Path,
expected: ExpectedGeometry,
keys: Any = None,
) -> SceneAssets:
if keys is not None:
raise error(
"asset_auth_failed",
"this build cannot use a key provider: packages are plain so far",
)
return scene_from_contents(read_package(path), expected)
def validate(
self, assets: SceneAssets, expected: ExpectedGeometry
) -> tuple[Diagnostic, ...]:
return validate_assets(assets, expected)
def describe_package(path: Path) -> dict[str, Any]:
"""A report on a package without building a scene: for the validate tool."""
contents = read_package(path)
manifest = contents.manifest
return {
"asset_id": manifest.get("asset_id"),
"asset_version": manifest.get("asset_version"),
"geometry_id": manifest.get("geometry_id"),
"field_domain_id": (manifest.get("field_domain") or {}).get("id"),
"shape_hw": (manifest.get("field_domain") or {}).get("shape_hw"),
"asset_hash": manifest.get("asset_hash"),
"parts": {part.get("id"): part.get("role") for part in manifest.get("parts", [])},
"entries": {
name: {"dtype": r.get("dtype"), "shape": r.get("shape")}
for name, r in (manifest.get("entries") or {}).items()
},
"vertices": {
"sensing": int(contents.arrays["sensing/positions.npy"].shape[0]),
},
}
@@ -24,6 +24,7 @@ from orisys_app_shared.scene3d.contracts import (
SurfaceBinding, SurfaceBinding,
freeze, freeze,
) )
from orisys_app_shared.scene3d.digest import field_domain_id, geometry_id
#: The plate is 20 mm x 20 mm, sampled 3x3: small enough to read in a test, big #: The plate is 20 mm x 20 mm, sampled 3x3: small enough to read in a test, big
#: enough to have an interior vertex that a bilinear tap can move. #: enough to have an interior vertex that a bilinear tap can move.
@@ -103,6 +104,14 @@ def _housing() -> MeshPart:
) )
#: A synthetic scene still obeys the contract it demonstrates: its identity is a
#: *content digest* of its geometry and of its UV domain, not a label. A label would
#: let the package layer's "the manifest id must match the arrays" check be bypassed
#: by the one scene everybody copies. Computed once, at import.
POC_GEOMETRY_ID = geometry_id([_plate()])
POC_DOMAIN_ID = field_domain_id(shape_hw=(GRID, GRID), flip_v=False)
def _binding() -> SurfaceBinding: def _binding() -> SurfaceBinding:
"""Identity binding: each plate vertex samples its own field cell.""" """Identity binding: each plate vertex samples its own field cell."""
count = GRID * GRID count = GRID * GRID
@@ -113,8 +122,8 @@ def _binding() -> SurfaceBinding:
weights[index, 0] = 1.0 weights[index, 0] = 1.0
return SurfaceBinding( return SurfaceBinding(
part_id="sensing_surface", part_id="sensing_surface",
geometry_id="poc-geometry", geometry_id=POC_GEOMETRY_ID,
field_domain_id="poc-domain", field_domain_id=POC_DOMAIN_ID,
shape_hw=(GRID, GRID), shape_hw=(GRID, GRID),
sample_indices=freeze(indices, dtype=np.int32), sample_indices=freeze(indices, dtype=np.int32),
sample_weights=freeze(weights, dtype=np.float32), sample_weights=freeze(weights, dtype=np.float32),
@@ -147,7 +156,7 @@ def poc_scene() -> SceneAssets:
asset_id="poc-scene", asset_id="poc-scene",
asset_version="0.0.0", asset_version="0.0.0",
asset_hash="poc-hash", asset_hash="poc-hash",
geometry_id="poc-geometry", geometry_id=POC_GEOMETRY_ID,
parts=(plate, housing), parts=(plate, housing),
binding=_binding(), binding=_binding(),
camera_presets=presets, camera_presets=presets,
@@ -155,7 +164,7 @@ def poc_scene() -> SceneAssets:
np.stack([points.min(axis=0), points.max(axis=0)]).astype(np.float32), np.stack([points.min(axis=0), points.max(axis=0)]).astype(np.float32),
dtype=np.float32, dtype=np.float32,
), ),
field_domain_id="poc-domain", field_domain_id=POC_DOMAIN_ID,
shape_hw=(GRID, GRID), shape_hw=(GRID, GRID),
) )
@@ -164,5 +173,9 @@ def poc_identity(*, generation: int = 1) -> FrameIdentity:
return FrameIdentity("poc", 1, 0, 0.0, generation=generation) return FrameIdentity("poc", 1, 0, 0.0, generation=generation)
def poc_expected() -> ExpectedGeometry: def poc_expected(assets: SceneAssets | None = None) -> ExpectedGeometry:
return ExpectedGeometry("finger", "poc", "poc-geometry", "poc-domain", (GRID, GRID)) """The device description this scene claims to fit (its own ids, not labels)."""
scene = assets or poc_scene()
return ExpectedGeometry(
"finger", "poc", scene.geometry_id, scene.field_domain_id, scene.shape_hw
)
+720
View File
@@ -0,0 +1,720 @@
"""Stage 2 acceptance: asset contracts, coordinates, units, identity.
Two kinds of test live here. The ones that need the real Blender export are skipped
when the SDK assets are not on this machine; the ones that build their own data run
everywhere. That split is deliberate — the real assets are the *input sample* the
stage owes, and a synthetic one can only prove the code agrees with itself.
"""
from __future__ import annotations
import json
import struct
import zipfile
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.scene3d import synthetic
from orisys_app_shared.scene3d.contracts import (
ExpectedGeometry,
Material,
MeshPart,
SceneAssets,
ThreeDError,
validate_assets,
freeze,
)
from orisys_app_shared.scene3d.digest import (
asset_hash,
binding_id,
content_digest,
field_domain_id,
geometry_id,
)
from orisys_app_shared.scene3d.glb import (
HousingLimits,
load_housing_glb,
read_glb,
with_housing,
)
from orisys_app_shared.scene3d.legacy import (
LegacyFingerImporter,
build_binding,
estimate_meters_per_source_unit,
load_legacy_domain,
split_at_uv_seams,
uv_correspondence_residual,
validate_rigid_transform,
vertex_normals,
)
#: The shipped export, when this checkout has it.
_ASSETS = Path(r"D:\orisys_ws\orisys-sdk\sensors\finger\assets")
RUN_DIR = _ASSETS / "30fps-0703"
RUN_MAP = RUN_DIR / "unroll_map_uv_s0.4.npz"
VIZ_DIR = _ASSETS / "visualization"
VIZ_MAP = VIZ_DIR / "flatten_remap.npz"
requires_run = pytest.mark.skipif(
not (RUN_DIR / "mesh.npz").is_file() or not RUN_MAP.is_file(),
reason="the shipped 30fps-0703 export is not present",
)
requires_viz = pytest.mark.skipif(
not (VIZ_DIR / "mesh.npz").is_file() or not VIZ_MAP.is_file(),
reason="the shipped visualization export is not present",
)
def _identity() -> np.ndarray:
return np.eye(4)
def _import_run(**kwargs) -> SceneAssets:
return LegacyFingerImporter().convert(
RUN_DIR,
RUN_MAP,
transform_sensor_from_source=kwargs.pop("transform", _identity()),
meters_per_source_unit=kwargs.pop("units", 1e-3),
**kwargs,
)
# --------------------------------------------------------- input sample (real)
@requires_run
def test_the_real_export_imports_and_validates():
assets = _import_run()
part = assets.sensing_part()
assert part.positions_sensor_m.shape == (1032, 3)
assert part.triangles.shape == (1974, 3)
assert assets.shape_hw == (300, 280)
assert validate_assets(assets) == ()
# 15.8 source units at 1 mm per unit is a 15.8 mm sensing surface.
extent = np.ptp(np.asarray(part.positions_sensor_m), axis=0)
assert np.allclose(np.sort(extent), [0.0075, 0.015, 0.01581], atol=1e-5), extent
assert assets.source_vertex_of_render is not None
assert assets.source_vertex_of_render.shape == (1032,)
@requires_run
def test_the_uv_convention_is_measured_against_the_asset():
"""The one fact the importer must not guess: which pixel a UV addresses."""
domain = load_legacy_domain(RUN_MAP)
with np.load(RUN_DIR / "mesh.npz", allow_pickle=False) as handle:
positions, uvs, _, _ = split_at_uv_seams(
handle["vertices"], handle["faces"], handle["uv_data"], handle["loop_to_vertex"]
)
residual, samples = uv_correspondence_residual(positions, uvs, domain)
assert domain.flip_v is True
assert samples > 500
assert residual < 0.01, f"the UV convention no longer matches world_xyz ({residual})"
# ...and the wrong flip really is wrong: the check discriminates.
flipped = type(domain)(
shape_hw=domain.shape_hw,
flip_v=False,
source_scale=domain.source_scale,
valid_mask=domain.valid_mask,
domain_id=domain.domain_id,
mapping_digest=domain.mapping_digest,
world_xyz=domain.world_xyz,
)
wrong, _ = uv_correspondence_residual(positions, uvs, flipped)
assert wrong > 0.2, f"a wrong v flip scored {wrong}: the check is not discriminating"
@requires_run
@requires_viz
def test_the_two_asset_directories_are_different_scenes():
"""1032/1974 on a 300x280 domain versus 1005/1921 on a 1024x1024 domain."""
run = _import_run()
viz = LegacyFingerImporter().convert(
VIZ_DIR, VIZ_MAP, transform_sensor_from_source=_identity(), meters_per_source_unit=1e-3
)
assert run.sensing_part().positions_sensor_m.shape[0] == 1032
assert viz.sensing_part().positions_sensor_m.shape[0] == 1005
assert run.geometry_id != viz.geometry_id
assert run.field_domain_id != viz.field_domain_id
assert run.shape_hw != viz.shape_hw
@requires_run
@requires_viz
def test_mixing_the_two_meshes_is_refused():
run = _import_run()
viz = LegacyFingerImporter().convert(
VIZ_DIR, VIZ_MAP, transform_sensor_from_source=_identity(), meters_per_source_unit=1e-3
)
mixed = SceneAssets(
asset_id="mixed",
asset_version="0",
asset_hash="x",
geometry_id=run.geometry_id,
parts=viz.parts, # 1005 vertices
binding=run.binding, # 1032 rows
camera_presets=run.camera_presets,
bounds_sensor_m=viz.bounds_sensor_m,
field_domain_id=run.field_domain_id,
shape_hw=run.shape_hw,
)
with pytest.raises(ThreeDError) as excinfo:
validate_assets(mixed)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
assert "different mesh" in excinfo.value.diagnostic.message
# ------------------------------------------------------------- seam splitting
def test_a_uv_seam_splits_a_vertex_without_averaging_its_uv():
"""One geometry vertex with two UVs becomes two render vertices; a shared one stays."""
vertices = np.array(
[[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [1.0, 1.0, 0.0], [0.0, 1.0, 0.0]]
)
faces = np.array([[0, 1, 2], [0, 2, 3]], dtype=np.int64)
# Vertex 0 is used with (0,0) in the first triangle and (0.5,0.5) in the second
# (a seam); vertex 2 keeps one UV across both (shared, not duplicated).
loop_to_vertex = np.array([0, 1, 2, 0, 2, 3], dtype=np.int64)
uv_data = np.array(
[[0.0, 0.0], [1.0, 0.0], [0.0, 1.0], [0.5, 0.5], [0.0, 1.0], [0.2, 0.2]]
)
positions, uvs, triangles, source = split_at_uv_seams(
vertices, faces, uv_data, loop_to_vertex
)
assert source.tolist() == [0, 1, 2, 0, 3], source.tolist()
assert len(positions) == 5, "the seam vertex was not duplicated exactly once"
seam = [index for index, item in enumerate(source.tolist()) if item == 0]
assert len(seam) == 2 and not np.allclose(uvs[seam[0]], uvs[seam[1]]), (
"the two sides of the seam must keep their own UVs, never an average"
)
shared = [index for index, item in enumerate(source.tolist()) if item == 2]
assert len(shared) == 1, "a vertex with one UV must not be duplicated"
assert triangles.shape == (2, 3)
assert set(triangles[0].tolist()) & set(triangles[1].tolist()) == {shared[0]}, (
"the two triangles must share the unsplit vertex"
)
def test_identical_uvs_do_not_duplicate_a_vertex():
vertices = np.zeros((3, 3))
faces = np.array([[0, 1, 2]])
loop = np.array([0, 1, 2])
uv = np.array([[0.0, 0.0], [1.0, 0.0], [0.0, 1.0]])
positions, uvs, _, source = split_at_uv_seams(vertices, faces, uv, loop)
assert len(positions) == 3 and source.tolist() == [0, 1, 2]
def test_a_mesh_whose_corner_lists_do_not_match_is_refused():
vertices = np.zeros((3, 3))
faces = np.array([[0, 1, 2]])
with pytest.raises(ThreeDError) as excinfo:
split_at_uv_seams(vertices, faces, np.zeros((2, 2)), np.array([0, 1, 2]))
assert "per-corner UVs" in excinfo.value.diagnostic.message
def test_an_out_of_range_face_index_is_refused():
vertices = np.zeros((3, 3))
faces = np.array([[0, 1, 9]])
uv = np.zeros((3, 2))
loop = np.array([0, 1, 2])
with pytest.raises(ThreeDError) as excinfo:
split_at_uv_seams(vertices, faces, uv, loop)
assert "outside the vertex array" in excinfo.value.diagnostic.message
def test_vertex_normals_are_unit_and_area_weighted():
# Two triangles sharing an edge, the second twice as large.
positions = np.array(
[[0.0, 0.0, 0.0], [1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 2.0, 0.0]]
)
triangles = np.array([[0, 1, 2], [0, 2, 3]])
normals = vertex_normals(positions, triangles)
assert np.allclose(np.linalg.norm(normals, axis=1), 1.0)
assert np.allclose(normals[:, 2], 1.0), "a flat mesh must have +Z normals"
# A vertex used by no triangle gets a finite fallback rather than NaN.
lonely = vertex_normals(np.vstack([positions, [[5.0, 5.0, 5.0]]]), triangles)
assert np.isfinite(lonely[4]).all()
# ------------------------------------------------------------------- binding
def _domain(shape_hw=(4, 5), flip_v=False, world=None):
valid = np.ones(shape_hw, dtype=bool)
return type(
"D",
(),
{
"shape_hw": shape_hw,
"flip_v": flip_v,
"source_scale": None,
"valid_mask": freeze(valid, dtype=np.bool_),
"domain_id": field_domain_id(shape_hw=shape_hw, flip_v=flip_v),
"mapping_digest": "",
"world_xyz": world,
},
)()
def test_binding_taps_the_four_neighbours_with_weights_summing_to_one():
domain = _domain()
# Between pixels: x = 0.3*4 = 1.2, y = 0.25*3 = 0.75, so all four neighbours
# carry weight. (A UV that lands exactly on a pixel centre correctly uses one
# tap only, so the test has to sit between them.)
uvs = np.array([[0.3, 0.25]])
binding = build_binding(uvs, domain, part_id="sensing_surface", geometry_id_value="g")
weights = np.asarray(binding.sample_weights)[0]
indices = np.asarray(binding.sample_indices)[0]
assert weights.sum() == pytest.approx(1.0)
assert len(set(indices.tolist())) == 4, indices
# y*W+x with W=5: (0,1) -> 1, (0,2) -> 2, (1,1) -> 6, (1,2) -> 7
assert sorted(indices.tolist()) == [1, 2, 6, 7], indices.tolist()
# The nearer pixel of each pair carries the larger weight.
by_index = dict(zip(indices.tolist(), weights.tolist()))
assert by_index[1] > by_index[2] and by_index[6] > by_index[7]
def test_a_uv_exactly_on_a_pixel_uses_only_that_pixel():
"""Zero-weight taps are dropped, so a pixel-exact UV uses one tap, not four."""
# 3x5: u = 1/4 -> x = 1 exactly; v = 1/2 -> y = 1 exactly.
domain = _domain(shape_hw=(3, 5))
binding = build_binding(
np.array([[0.25, 0.5]]), domain, part_id="p", geometry_id_value="g"
)
indices = np.asarray(binding.sample_indices)[0]
weights = np.asarray(binding.sample_weights)[0]
assert int((indices >= 0).sum()) == 1, indices
assert int(indices[0]) == 1 * 5 + 1, indices
assert weights[0] == pytest.approx(1.0)
def test_a_half_pixel_uv_uses_the_two_taps_on_that_column():
"""Between rows, on a pixel: exactly two taps, and their weights still sum to 1."""
domain = _domain(shape_hw=(3, 5))
binding = build_binding(
np.array([[0.25, 0.25]]), domain, part_id="p", geometry_id_value="g"
)
indices = np.asarray(binding.sample_indices)[0]
weights = np.asarray(binding.sample_weights)[0]
used = indices[indices >= 0].tolist()
assert used == [1, 6], indices
assert [index % 5 for index in used] == [1, 1], "both taps must be in one column"
assert weights.sum() == pytest.approx(1.0)
def test_a_vertex_on_the_border_drops_its_outside_taps():
"""No tap may point outside the map, and the weights still sum to one."""
domain = _domain()
uvs = np.array([[0.0, 0.0], [1.0, 1.0]])
binding = build_binding(uvs, domain, part_id="sensing_surface", geometry_id_value="g")
indices = np.asarray(binding.sample_indices)
weights = np.asarray(binding.sample_weights)
assert np.allclose(weights.sum(axis=1), 1.0, atol=1e-6)
assert int(indices[indices >= 0].max()) < domain.shape_hw[0] * domain.shape_hw[1]
# A corner uses exactly one tap (the corner pixel itself), the rest are unused.
assert int((indices[0] >= 0).sum()) == 1, indices[0]
assert int((indices[0] < 0).sum()) == 3
assert weights[0, 1:].sum() == 0.0
assert int(weights[indices < 0].sum()) == 0
assert int(indices[0, 0]) == 0
def test_a_flipped_domain_mirrors_the_row_it_samples():
domain = _domain(shape_hw=(4, 5), flip_v=False)
flipped = _domain(shape_hw=(4, 5), flip_v=True)
uv = np.array([[0.54, 0.0]]) # between pixels, so the row is unambiguous
plain = np.asarray(
build_binding(uv, domain, part_id="p", geometry_id_value="g").sample_indices
)
mirrored = np.asarray(
build_binding(uv, flipped, part_id="p", geometry_id_value="g").sample_indices
)
plain_rows = sorted({int(index) // 5 for index in plain[plain >= 0].tolist()})
mirrored_rows = sorted({int(index) // 5 for index in mirrored[mirrored >= 0].tolist()})
assert plain_rows == [0], plain
assert mirrored_rows == [3], mirrored
@requires_run
def test_the_real_binding_is_well_formed():
assets = _import_run()
indices = np.asarray(assets.binding.sample_indices)
weights = np.asarray(assets.binding.sample_weights)
domain_size = assets.shape_hw[0] * assets.shape_hw[1]
assert indices.shape == (1032, 4)
assert indices.max() < domain_size
assert np.allclose(weights.sum(axis=1), 1.0, atol=1e-5)
assert (weights >= 0).all()
assert int((indices < 0).sum()) == int((weights == 0).sum())
# --------------------------------------------------------------------- units
def test_a_scaling_transform_is_refused():
scaled = np.diag([2.0, 2.0, 2.0, 1.0])
with pytest.raises(ThreeDError) as excinfo:
validate_rigid_transform(scaled)
assert "not a rigid rotation" in excinfo.value.diagnostic.message
def test_a_mirror_is_refused():
mirror = np.diag([-1.0, 1.0, 1.0, 1.0])
with pytest.raises(ThreeDError) as excinfo:
validate_rigid_transform(mirror)
assert "determinant" in excinfo.value.diagnostic.message
def test_a_bad_transform_shape_or_non_finite_values_are_refused():
with pytest.raises(ThreeDError):
validate_rigid_transform(np.eye(3))
broken = np.eye(4)
broken[0, 3] = np.nan
with pytest.raises(ThreeDError):
validate_rigid_transform(broken)
# A bottom row that is not the affine one: it would shear on every multiply.
affine_broken = np.array(
[[1, 0, 0, 0], [0, 1, 0, 0], [0, 0, 1, 0], [0, 0, 1, 1]], dtype=float
)
with pytest.raises(ThreeDError) as excinfo:
validate_rigid_transform(affine_broken)
assert "affine" in excinfo.value.diagnostic.message
def test_units_must_be_stated_and_positive(tmp_path):
"""The importer never guesses: no default, and a zero is an error."""
mesh = tmp_path / "mesh.npz"
np.savez(
mesh,
vertices=np.zeros((3, 3), dtype=np.float32),
faces=np.array([[0, 1, 2]], dtype=np.int32),
uv_data=np.zeros((3, 2), dtype=np.float32),
loop_to_vertex=np.array([0, 1, 2], dtype=np.int32),
)
mapping = tmp_path / "map.npz"
np.savez(mapping, valid_mask=np.ones((2, 2), dtype=bool))
for units in (0.0, -1.0, float("nan")):
with pytest.raises(ThreeDError) as excinfo:
LegacyFingerImporter().convert(
tmp_path, mapping, transform_sensor_from_source=_identity(),
meters_per_source_unit=units,
)
assert "does not guess" in excinfo.value.diagnostic.message
@requires_run
def test_units_and_the_transform_do_two_separate_jobs():
"""Millimetres become metres, then a rigid transform moves them."""
plain = _import_run()
shifted = _import_run(transform=np.array(
[[1, 0, 0, 0.001], [0, 1, 0, -0.002], [0, 0, 1, 0.003], [0, 0, 0, 1]], dtype=float
))
delta = np.asarray(shifted.sensing_part().positions_sensor_m) - np.asarray(
plain.sensing_part().positions_sensor_m
)
assert np.allclose(delta, [0.001, -0.002, 0.003], atol=1e-6)
assert np.allclose(
np.ptp(np.asarray(plain.sensing_part().positions_sensor_m), axis=0),
np.ptp(np.asarray(shifted.sensing_part().positions_sensor_m), axis=0),
atol=1e-6,
), "a rigid transform must not change any extent"
millimetres = _import_run(units=1e-3)
metres = _import_run(units=1.0)
assert np.allclose(
np.asarray(millimetres.sensing_part().positions_sensor_m) * 1000.0,
np.asarray(metres.sensing_part().positions_sensor_m),
atol=1e-6,
)
@requires_run
def test_the_measured_scale_is_reported_not_applied():
with np.load(RUN_DIR / "mesh.npz", allow_pickle=False) as handle:
vertices = handle["vertices"]
suggested = estimate_meters_per_source_unit(vertices)
assert suggested == pytest.approx(1.0 / 15.81, rel=0.01)
assert estimate_meters_per_source_unit(vertices, 0.0158) == pytest.approx(1e-3, rel=0.01)
# ------------------------------------------------------------------ identity
def test_digests_are_deterministic_and_sensitive():
assets = synthetic.poc_scene()
assert geometry_id(assets.parts) == geometry_id(assets.parts)
assert content_digest({"a": np.arange(4)}) == content_digest({"a": np.arange(4)})
assert content_digest({"a": np.arange(4)}) != content_digest({"a": np.arange(5)})
# dtype and shape are part of the identity: the same numbers stored
# differently are different bytes.
assert content_digest({"a": np.arange(4, dtype=np.int32)}) != content_digest(
{"a": np.arange(4, dtype=np.int64)}
)
def test_geometry_identity_ignores_display_choices():
"""A colour or a recomputed normal must not invalidate a calibration."""
assets = synthetic.poc_scene()
plate = assets.sensing_part()
recoloured = MeshPart(
part_id=plate.part_id,
role=plate.role,
positions_sensor_m=plate.positions_sensor_m,
normals_sensor=freeze(
-np.asarray(plate.normals_sensor), dtype=np.float32
), # a different normal algorithm
triangles=plate.triangles,
material=Material((1.0, 0.0, 0.0, 1.0), "diffuse", False),
lod_id=plate.lod_id,
)
assert geometry_id([recoloured]) == geometry_id([plate])
moved = MeshPart(
part_id=plate.part_id,
role=plate.role,
positions_sensor_m=freeze(
np.asarray(plate.positions_sensor_m) + 0.001, dtype=np.float32
),
normals_sensor=plate.normals_sensor,
triangles=plate.triangles,
material=plate.material,
lod_id=plate.lod_id,
)
assert geometry_id([moved]) != geometry_id([plate]), "moving the mesh must change it"
def test_domain_identity_covers_flip_scale_and_mapping():
base = field_domain_id(shape_hw=(300, 280), flip_v=True, source_scale=0.4, mapping_digest="a")
assert base == field_domain_id(
shape_hw=(300, 280), flip_v=True, source_scale=0.4, mapping_digest="a"
)
assert base != field_domain_id(
shape_hw=(300, 280), flip_v=False, source_scale=0.4, mapping_digest="a"
)
assert base != field_domain_id(
shape_hw=(300, 280), flip_v=True, source_scale=0.5, mapping_digest="a"
)
assert base != field_domain_id(
shape_hw=(300, 280), flip_v=True, source_scale=0.4, mapping_digest="b"
)
assert base != field_domain_id(
shape_hw=(280, 300), flip_v=True, source_scale=0.4, mapping_digest="a"
)
with pytest.raises(ThreeDError):
field_domain_id(shape_hw=(0, 280), flip_v=True)
def test_a_lod_change_needs_a_new_binding():
assets = synthetic.poc_scene()
moved = SceneAssets(
**{
**{field: getattr(assets, field) for field in (
"asset_id", "asset_version", "asset_hash", "geometry_id", "parts",
"camera_presets", "bounds_sensor_m", "field_domain_id", "shape_hw",
)},
"binding": assets.binding,
}
)
assert binding_id(moved.binding) == binding_id(assets.binding)
assert asset_hash({"a": "1"}, {"x": 1}) != asset_hash({"a": "1"}, {"x": 2})
# ------------------------------------------------------------------- the GLB
def _glb_bytes(positions, normals=None, indices=None, *, mode=4, extra=None,
node_matrix=None) -> bytes:
"""A minimal, valid glTF 2.0 binary with one node, one mesh, one primitive."""
positions = np.asarray(positions, dtype=np.float32)
chunks = [positions.tobytes()]
attributes = {"POSITION": 0}
accessors = [
{
"bufferView": 0,
"componentType": 5126,
"count": int(len(positions)),
"type": "VEC3",
}
]
views = [{"buffer": 0, "byteOffset": 0, "byteLength": len(chunks[0])}]
offset = len(chunks[0])
if normals is not None:
normals = np.asarray(normals, dtype=np.float32)
chunks.append(normals.tobytes())
accessors.append(
{
"bufferView": 1,
"componentType": 5126,
"count": int(len(normals)),
"type": "VEC3",
}
)
views.append({"buffer": 0, "byteOffset": offset, "byteLength": len(chunks[-1])})
offset += len(chunks[-1])
attributes["NORMAL"] = 1
primitive: dict = {"attributes": attributes, "mode": int(mode)}
if indices is not None:
indices = np.asarray(indices, dtype=np.uint32)
chunks.append(indices.tobytes())
accessors.append(
{
"bufferView": len(views),
"componentType": 5125,
"count": int(len(indices)),
"type": "SCALAR",
}
)
views.append({"buffer": 0, "byteOffset": offset, "byteLength": len(chunks[-1])})
primitive["indices"] = len(accessors) - 1
document = {
"asset": {"version": "2.0"},
"scene": 0,
"scenes": [{"nodes": [0]}],
"nodes": [{"mesh": 0, "name": "housing"}],
"meshes": [{"primitives": [primitive]}],
"accessors": accessors,
"bufferViews": views,
"buffers": [{"byteLength": sum(len(chunk) for chunk in chunks)}],
}
if node_matrix is not None:
document["nodes"][0]["matrix"] = list(np.asarray(node_matrix, dtype=float).T.reshape(-1))
if extra:
document.update(extra)
binary = b"".join(chunks)
json_chunk = json.dumps(document).encode("utf-8")
json_chunk += b" " * ((4 - len(json_chunk) % 4) % 4)
binary += b"\0" * ((4 - len(binary) % 4) % 4)
total = 12 + 8 + len(json_chunk) + 8 + len(binary)
header = struct.pack("<III", 0x46546C67, 2, total)
return (
header
+ struct.pack("<II", len(json_chunk), 0x4E4F534A)
+ json_chunk
+ struct.pack("<II", len(binary), 0x004E4942)
+ binary
)
def test_a_housing_glb_is_read_with_its_node_transform_baked(tmp_path):
positions = [[0, 0, 0], [1, 0, 0], [0, 1, 0]]
path = tmp_path / "housing.glb"
matrix = np.eye(4)
matrix[:3, 3] = [0.01, 0.02, 0.03]
path.write_bytes(_glb_bytes(positions, indices=[0, 1, 2], node_matrix=matrix))
part = load_housing_glb(path)
assert part.role == "housing" and part.part_id == "housing"
assert part.triangles.shape == (1, 3)
assert np.allclose(
np.asarray(part.positions_sensor_m), np.asarray(positions) + [0.01, 0.02, 0.03]
)
assert np.allclose(np.linalg.norm(np.asarray(part.normals_sensor), axis=1), 1.0)
def test_a_housing_glb_without_normals_gets_computed_ones(tmp_path):
path = tmp_path / "housing.glb"
path.write_bytes(_glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]], indices=[0, 1, 2]))
part = load_housing_glb(path)
assert np.allclose(np.asarray(part.normals_sensor)[:, 2], 1.0)
def test_a_glb_with_a_scale_in_its_node_is_baked_not_refused(tmp_path):
"""A model transform is data, not a mistake: bake it into the vertices."""
path = tmp_path / "scaled.glb"
matrix = np.diag([2.0, 2.0, 2.0, 1.0])
path.write_bytes(_glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]], indices=[0, 1, 2],
node_matrix=matrix))
part = load_housing_glb(path)
assert np.allclose(np.asarray(part.positions_sensor_m)[1], [2.0, 0.0, 0.0])
def test_the_housing_glb_path_refuses_what_it_cannot_do(tmp_path):
cases = {
"draco": _glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]],
extra={"extensionsUsed": ["KHR_draco_mesh_compression"]}),
"morph targets": _glb_bytes(
[[0, 0, 0], [1, 0, 0], [0, 1, 0]],
extra={"meshes": [{"primitives": [
{"attributes": {"POSITION": 0}, "mode": 4}
], "weights": [0.5]}]},
),
"not triangles": _glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]], mode=1),
"nan position": _glb_bytes([[0, 0, 0], [np.nan, 0, 0], [0, 1, 0]]),
"bad index": _glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]], indices=[0, 1, 9]),
}
for label, payload in cases.items():
path = tmp_path / f"{label.replace(' ', '_')}.glb"
path.write_bytes(payload)
with pytest.raises(ThreeDError) as excinfo:
load_housing_glb(path)
assert excinfo.value.diagnostic.code in ("asset_rejected", "geometry_mismatch"), label
def test_a_broken_glb_header_is_refused(tmp_path):
path = tmp_path / "broken.glb"
path.write_bytes(b"not a glb at all")
with pytest.raises(ThreeDError):
read_glb(path)
good = _glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]])
path.write_bytes(good[:8] + struct.pack("<I", len(good) + 16) + good[12:])
with pytest.raises(ThreeDError) as excinfo:
read_glb(path)
assert "header declares" in excinfo.value.diagnostic.message
def test_the_housing_triangle_budget_is_enforced(tmp_path):
path = tmp_path / "big.glb"
positions = []
for index in range(30):
positions.extend([[float(index), 0.0, 0.0], [float(index) + 1, 0.0, 0.0], [0.0, 1.0, 0.0]])
indices = list(range(len(positions)))
path.write_bytes(_glb_bytes(positions, indices=indices))
with pytest.raises(ThreeDError) as excinfo:
load_housing_glb(path, limits=HousingLimits(max_triangles=5))
assert excinfo.value.diagnostic.code == "gpu_budget_exceeded"
@requires_run
def test_adding_a_housing_keeps_the_sensing_identity(tmp_path):
"""The housing is a second part; the sensing geometry id must not move."""
assets = _import_run() # the imported scene has the sensing part only
before = assets.geometry_id
path = tmp_path / "housing.glb"
path.write_bytes(_glb_bytes([[0, 0, 0], [1, 0, 0], [0, 1, 0]], indices=[0, 1, 2]))
housing = load_housing_glb(path)
combined = with_housing(assets, housing)
assert len(combined.parts) == 2
assert combined.sensing_part().positions_sensor_m is assets.sensing_part().positions_sensor_m
assert combined.geometry_id == before, (
"the sensing geometry id must not change when a housing is added"
)
with pytest.raises(ThreeDError) as excinfo:
with_housing(combined, housing) # only one housing
assert "already has a housing" in excinfo.value.diagnostic.message
# ...and the POC scene, which ships with one, refuses a second the same way.
with pytest.raises(ThreeDError):
with_housing(synthetic.poc_scene(), housing)
@requires_run
def test_the_housing_can_be_placed_with_the_surface_in_one_frame():
"""Three known non-collinear points: surface and housing agree on the axes."""
assets = _import_run()
bounds = np.asarray(assets.bounds_sensor_m)
# A housing box built *around* the imported surface, in the same metres.
low, high = bounds[0] - 0.001, bounds[1] + 0.001
corners = np.array(
[
[low[0], low[1], low[2]], [high[0], low[1], low[2]], [low[0], high[1], low[2]],
[high[0], high[1], low[2]], [low[0], low[1], high[2]], [high[0], low[1], high[2]],
[low[0], high[1], high[2]], [high[0], high[1], high[2]],
]
)
part = MeshPart(
part_id="housing",
role="housing",
positions_sensor_m=freeze(corners, dtype=np.float32),
normals_sensor=freeze(np.tile([0.0, 0.0, 1.0], (8, 1)), dtype=np.float32),
triangles=freeze(np.zeros((0, 3), dtype=np.uint32), dtype=np.uint32),
material=Material((0.3, 0.3, 0.3, 1.0), "diffuse", True),
)
scene = validate_assets(with_housing(assets, part)) is not None
assert scene
surface = np.asarray(assets.sensing_part().positions_sensor_m)
housed = np.asarray(part.positions_sensor_m)
# The three axes line up: every surface vertex is inside the housing box, and
# the box's extent exceeds the surface's by exactly the 1 mm margin.
assert (surface.min(axis=0) >= housed.min(axis=0)).all()
assert (surface.max(axis=0) <= housed.max(axis=0)).all()
assert np.allclose(
np.ptp(housed, axis=0) - np.ptp(surface, axis=0), 0.002, atol=1e-6
)
+369
View File
@@ -0,0 +1,369 @@
"""Stage 2 acceptance for the ``.oryscene`` container.
The interesting half is the refusals: a package that is *nearly* right is the case a
loader has to catch, because the failure mode it prevents (a scene that renders with
someone else's geometry) is silent. Every test below mutates one thing about a
package that was just written correctly, and asserts that the loader says no — with
the code the design document assigns to that failure.
"""
from __future__ import annotations
import json
import zipfile
from dataclasses import replace
from pathlib import Path
import numpy as np
import pytest
from orisys_app_shared.scene3d import synthetic
from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError
from orisys_app_shared.scene3d.digest import asset_hash, content_digest
from orisys_app_shared.scene3d.glb import with_housing
from orisys_app_shared.scene3d.package import (
MAGIC,
describe_package,
read_package,
scene_from_contents,
write_package,
)
from orisys_app_shared.scene3d.package import PackagedAssetLoader
@pytest.fixture
def scene():
"""A complete little scene: one sensing plate and one housing box."""
return synthetic.poc_scene()
@pytest.fixture
def package(tmp_path, scene):
path = tmp_path / "poc.oryscene"
report = write_package(scene, path)
assert report.output == path
return path
def _rewrite(source: Path, destination: Path, mutate) -> Path:
"""Copy a package, letting ``mutate`` change the members on the way through."""
with zipfile.ZipFile(source) as archive:
members = {info.filename: archive.read(info.filename) for info in archive.infolist()}
members = mutate(members)
with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED) as archive:
for name in sorted(members):
archive.writestr(name, members[name])
return destination
def _manifest(members) -> dict:
return json.loads(members["manifest.json"])
def _store_manifest(members, manifest: dict) -> None:
members["manifest.json"] = json.dumps(manifest, indent=2, sort_keys=True).encode("utf-8")
def _rehash(members, manifest: dict) -> None:
"""Recompute the asset hash after a deliberate edit: the *content* is what lies."""
stripped = {k: v for k, v in manifest.items() if k not in ("asset_hash", "entry_digest")}
digest = asset_hash(
{name: str(record["sha256"]) for name, record in stripped["entries"].items()},
stripped,
)
manifest["asset_hash"] = digest
manifest["entry_digest"] = {
name: record["sha256"] for name, record in stripped["entries"].items()
}
_store_manifest(members, manifest)
# ------------------------------------------------------------------ round trip
def test_a_package_round_trips_into_the_same_scene(package, scene):
contents = read_package(package)
rebuilt = scene_from_contents(contents)
assert rebuilt.geometry_id == scene.geometry_id
assert rebuilt.field_domain_id == scene.field_domain_id
assert rebuilt.asset_id == scene.asset_id
assert rebuilt.shape_hw == scene.shape_hw
assert len(rebuilt.parts) == len(scene.parts)
for original in scene.parts:
copy = rebuilt.part(original.part_id)
assert copy is not None
assert np.array_equal(copy.positions_sensor_m, original.positions_sensor_m)
assert np.array_equal(copy.triangles, original.triangles)
assert np.array_equal(
np.asarray(rebuilt.binding.sample_indices),
np.asarray(scene.binding.sample_indices),
)
assert not np.asarray(rebuilt.binding.sample_weights).flags["WRITEABLE"]
assert [preset.name for preset in rebuilt.camera_presets] == [
preset.name for preset in scene.camera_presets
]
def test_the_loader_accepts_its_own_package_and_checks_the_device(package, scene):
loader = PackagedAssetLoader()
expected = ExpectedGeometry(
"finger", "rev-a", scene.geometry_id, scene.field_domain_id, scene.shape_hw
)
loaded = loader.load(package, expected)
assert loaded.geometry_id == scene.geometry_id
assert loader.validate(loaded, expected) == ()
wrong = ExpectedGeometry("finger", "rev-a", "other-geometry", scene.field_domain_id, scene.shape_hw)
with pytest.raises(ThreeDError) as excinfo:
loader.load(package, wrong)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
def test_the_package_report_describes_the_package(package):
report = describe_package(package)
assert report["asset_id"] == "poc-scene"
assert report["parts"] == {"sensing_surface": "sensing", "housing": "housing"}
assert report["shape_hw"] == [3, 3]
assert "sensing/positions.npy" in report["entries"]
assert report["vertices"]["sensing"] == 9
def test_a_housing_part_survives_packaging(tmp_path):
"""The housing has its own entries and keeps its role through the round trip."""
sensing_only = replace(synthetic.poc_scene(), parts=(synthetic.poc_scene().sensing_part(),))
housing = synthetic.poc_scene().part("housing")
scene = with_housing(sensing_only, housing)
path = tmp_path / "with-housing.oryscene"
write_package(scene, path)
rebuilt = scene_from_contents(read_package(path))
assert {part.role for part in rebuilt.parts} == {"sensing", "housing"}
assert np.array_equal(
rebuilt.part("housing").positions_sensor_m, housing.positions_sensor_m
)
assert rebuilt.part("housing").material.shading == "diffuse"
# --------------------------------------------------------------- refusals
def test_a_tampered_entry_is_refused(package, tmp_path):
def mutate(members):
members["sensing/positions.npy"] = b"not an array any more"
return members
broken = _rewrite(package, tmp_path / "tampered.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert excinfo.value.diagnostic.code == "asset_auth_failed"
def test_a_tampered_manifest_hash_is_refused(package, tmp_path):
def mutate(members):
manifest = _manifest(members)
manifest["geometry_id"] = "somebody-elses-geometry"
_store_manifest(members, manifest) # hash deliberately *not* recomputed
return members
broken = _rewrite(package, tmp_path / "manifest.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert "manifest hash does not match" in excinfo.value.diagnostic.message
def test_geometry_that_is_swapped_in_with_a_matching_hash_is_still_refused(package, tmp_path):
"""Consistency is not correctness: the binding must belong to the mesh."""
def mutate(members):
manifest = _manifest(members)
# Drop the sensing mesh to nine *different* positions but keep the binding,
# then fix every hash: the package is internally consistent and still wrong.
positions = np.zeros((9, 3), dtype=np.float32)
positions[:, 0] = np.arange(9, dtype=np.float32) * 0.01
import io
buffer = io.BytesIO()
np.save(buffer, positions, allow_pickle=False)
members["sensing/positions.npy"] = buffer.getvalue()
manifest["entries"]["sensing/positions.npy"]["sha256"] = content_digest(
{"data": positions}
)
_rehash(members, manifest)
return members
swapped = _rewrite(package, tmp_path / "swapped.oryscene", mutate)
contents = read_package(swapped) # hash-wise it is a valid package
with pytest.raises(ThreeDError) as excinfo:
scene_from_contents(contents)
assert excinfo.value.diagnostic.code == "geometry_mismatch"
assert "geometry" in excinfo.value.diagnostic.message.lower()
def test_non_finite_positions_are_refused(package, tmp_path):
def mutate(members):
import io
manifest = _manifest(members)
positions = np.zeros((9, 3), dtype=np.float32)
positions[0, 0] = np.nan
buffer = io.BytesIO()
np.save(buffer, positions, allow_pickle=False)
members["sensing/positions.npy"] = buffer.getvalue()
manifest["entries"]["sensing/positions.npy"]["sha256"] = content_digest(
{"data": positions}
)
_rehash(members, manifest)
return members
broken = _rewrite(package, tmp_path / "nan.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
scene_from_contents(read_package(broken))
assert "non-finite" in excinfo.value.diagnostic.message
def test_an_undeclared_entry_is_refused(package, tmp_path):
def mutate(members):
members["sensing/extra.npy"] = b"\x00"
return members
broken = _rewrite(package, tmp_path / "extra.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert "undeclared entries" in excinfo.value.diagnostic.message
def test_a_promised_entry_that_is_absent_is_refused(package, tmp_path):
def mutate(members):
members.pop("sensing/normals.npy")
return members
broken = _rewrite(package, tmp_path / "absent.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert "which is absent" in excinfo.value.diagnostic.message
def test_a_path_that_escapes_the_package_is_refused(package, tmp_path):
def mutate(members):
manifest = _manifest(members)
record = manifest["entries"].pop("sensing/normals.npy")
manifest["entries"]["../normals.npy"] = record
_rehash(members, manifest)
return members
broken = _rewrite(package, tmp_path / "traversal.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert "escapes the package" in excinfo.value.diagnostic.message
def test_a_dtype_or_shape_that_disagrees_with_the_manifest_is_refused(package, tmp_path):
for field, value, expected in (
("dtype", "i4", "field_domain_mismatch"),
("shape", [8, 3], "field_domain_mismatch"),
):
def mutate(members, field=field, value=value):
manifest = _manifest(members)
manifest["entries"]["sensing/positions.npy"][field] = value
_rehash(members, manifest)
return members
broken = _rewrite(package, tmp_path / f"bad-{field}.oryscene", mutate)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert excinfo.value.diagnostic.code == expected, field
def test_an_envelope_from_the_future_or_another_tool_is_refused(package, tmp_path):
def wrong_magic(members):
envelope = json.loads(members["envelope.json"])
envelope["magic"] = "not-our-magic"
members["envelope.json"] = json.dumps(envelope).encode()
return members
broken = _rewrite(package, tmp_path / "magic.oryscene", wrong_magic)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert excinfo.value.diagnostic.code == "asset_auth_failed"
def wrong_version(members):
envelope = json.loads(members["envelope.json"])
envelope["envelope_version"] = 99
members["envelope.json"] = json.dumps(envelope).encode()
return members
broken = _rewrite(package, tmp_path / "version.oryscene", wrong_version)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
assert "envelope version" in excinfo.value.diagnostic.message
def test_an_encrypted_envelope_says_so_instead_of_failing_obscurely(package, tmp_path):
"""Stage 5 will open these; until then the message has to be honest."""
def encrypt(members):
envelope = json.loads(members["envelope.json"])
envelope["payload"] = "aead"
envelope["key_id"] = "customer-1"
members["envelope.json"] = json.dumps(envelope).encode()
return members
broken = _rewrite(package, tmp_path / "encrypted.oryscene", encrypt)
with pytest.raises(ThreeDError) as excinfo:
read_package(broken)
item = excinfo.value.diagnostic
assert item.code == "asset_auth_failed"
assert "plain packages only" in item.message
def test_a_key_provider_is_refused_while_packages_are_plain(package, scene):
class Keys:
def get_content_key(self, key_id: str) -> bytes: # pragma: no cover - not used
return b"secret"
expected = ExpectedGeometry(
"finger", "rev-a", scene.geometry_id, scene.field_domain_id, scene.shape_hw
)
with pytest.raises(ThreeDError) as excinfo:
PackagedAssetLoader().load(package, expected, Keys())
assert "key provider" in excinfo.value.diagnostic.message
def test_the_manifest_carries_what_a_release_needs(package):
manifest = read_package(package).manifest
assert manifest["schema_version"] == "1.0"
assert manifest["coordinate_frame"] == "sensor"
assert manifest["length_unit"] == "m"
assert manifest["compatible"]["sensor_model"] == "finger"
assert manifest["field_domain"]["index_order"] == "row-major"
assert manifest["field_domain"]["origin"] == "top-left"
assert manifest["asset_hash"] and len(manifest["asset_hash"]) == 64
assert all(len(record["sha256"]) == 64 for record in manifest["entries"].values())
def test_the_same_scene_compiles_to_the_same_hash(tmp_path, scene):
first = write_package(scene, tmp_path / "a.oryscene")
second = write_package(scene, tmp_path / "b.oryscene")
assert first.asset_hash == second.asset_hash
assert write_package(scene, tmp_path / "a.oryscene", sensor_model="box").asset_hash != (
first.asset_hash
), "a different target device is a different package"
def test_writing_is_atomic_so_a_failed_build_leaves_nothing(tmp_path, scene, monkeypatch):
"""A half-written package must never be picked up by a later run."""
import orisys_app_shared.scene3d.package as package_module
target = tmp_path / "broken.oryscene"
def explode(*args, **kwargs):
raise RuntimeError("disk full")
monkeypatch.setattr(package_module.zipfile, "ZipFile", explode)
with pytest.raises(RuntimeError):
write_package(scene, target)
assert not target.exists()
assert not (tmp_path / "broken.oryscene.partial").exists(), (
"the partial file was left behind"
)
def test_the_envelope_names_the_magic_and_the_payload_state(package):
with zipfile.ZipFile(package) as archive:
envelope = json.loads(archive.read("envelope.json"))
assert envelope["magic"] == MAGIC
assert envelope["envelope_version"] == 1
assert envelope["payload"] == "plain"
assert envelope["entry_count"] == len(read_package(package).arrays)
+88 -20
View File
@@ -1,43 +1,111 @@
"""Compile a validated scene, and optionally a housing GLB, into a runtime package. """Compile a legacy export into a runtime package (stage 2 of the 3D plan).
Stage 2 and stage 5 deliverables (``docs/architecture/studio-3d-stages.md``). Offline only: this is the tool that bakes the coordinate transform, the unit
scale, the UV sampling table and the identity hashes into one file, so the runtime
never repeats that work.
Offline only: this tool is what turns a Blender export and the legacy NPZ geometry Units are an argument, never a default. The legacy mesh is 15.8 x 15.0 x 7.5 source
into the ``.oryscene`` payload the runtime loads. It bakes the coordinate units; whether one of those is a millimetre is a physical claim about the sensor,
transform, the sampling table and the LOD once, so the runtime never repeats that and ``validate_scene.py`` prints the measured candidates so a person can decide.
work; it signs with a key from the build environment, so no signing key ever
reaches a customer runtime.
Until stage 2 lands, no package is produced and the exit code is non-zero, so a Example::
release script cannot ship a package that was never compiled.
python tools/scene_assets/compile_scene.py \\
--legacy-dir sensors/finger/assets/30fps-0703 \\
--map sensors/finger/assets/30fps-0703/unroll_map_uv_s0.4.npz \\
--meters-per-source-unit 0.001 \\
--output build/finger-display-1.0.0.oryscene
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import json
import sys import sys
from pathlib import Path from pathlib import Path
_STAGE = "stage 2 (asset contracts) and stage 5 (release protection)"
def _bootstrap_paths() -> None:
here = Path(__file__).resolve()
repo = here.parents[2]
for path in (repo / "apps" / "studio" / "src", repo / "shared" / "src"):
if path.is_dir() and str(path) not in sys.path:
sys.path.insert(0, str(path))
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--legacy-dir", type=Path, default=None, help="legacy NPZ export") parser.add_argument("--legacy-dir", type=Path, required=True, help="legacy NPZ export")
parser.add_argument("--map", type=Path, default=None, help="run-time UV mapping (npz)") parser.add_argument("--map", type=Path, required=True, help="field map (npz)")
parser.add_argument("--housing", type=Path, default=None, help="housing GLB") parser.add_argument(
"--meters-per-source-unit",
type=float,
required=True,
help="stated unit scale (the importer never guesses it)",
)
parser.add_argument("--output", type=Path, required=True, help="package to write") parser.add_argument("--output", type=Path, required=True, help="package to write")
parser.add_argument("--asset-id", default="finger-display") parser.add_argument("--asset-id", default="finger-display")
parser.add_argument("--version", default="0.0.0") parser.add_argument("--version", default="1.0.0")
parser.add_argument("--lod-id", default="lod0")
parser.add_argument("--sensor-model", default="finger")
parser.add_argument(
"--transform",
type=Path,
default=None,
help="optional 4x4 rigid transform (npy) from source to sensor coordinates",
)
parser.add_argument("--json", action="store_true")
args = parser.parse_args(argv) args = parser.parse_args(argv)
print( _bootstrap_paths()
"not_implemented: " import numpy as np
f"{_STAGE}. Would write {args.output} from "
f"legacy={args.legacy_dir} map={args.map} housing={args.housing} " from orisys_app_shared.scene3d.contracts import ThreeDError
f"as {args.asset_id} {args.version}." from orisys_app_shared.scene3d.legacy import LegacyFingerImporter
from orisys_app_shared.scene3d.package import write_package
transform = (
np.load(args.transform, allow_pickle=False)
if args.transform is not None
else np.eye(4)
) )
return 2 try:
assets = LegacyFingerImporter().convert(
args.legacy_dir,
args.map,
transform_sensor_from_source=transform,
meters_per_source_unit=args.meters_per_source_unit,
asset_id=args.asset_id,
asset_version=args.version,
lod_id=args.lod_id,
)
report = write_package(
assets, args.output,
sensor_model=args.sensor_model,
hardware_revisions=("*",),
)
except ThreeDError as exc:
print(f"rejected: {exc.diagnostic.code}: {exc.diagnostic.message}", file=sys.stderr)
return 2
summary = {
"status": "ok",
"output": str(report.output),
"asset_hash": report.asset_hash,
"vertices": report.part_vertex_counts,
"triangles": report.part_triangle_counts,
"geometry_id": assets.geometry_id,
"field_domain_id": assets.field_domain_id,
}
if args.json:
print(json.dumps(summary, indent=2, ensure_ascii=False))
else:
print(
f"wrote {report.output} vertices={report.part_vertex_counts} "
f"triangles={report.part_triangle_counts}"
)
print(f" geometry_id={assets.geometry_id}")
print(f" asset_hash={report.asset_hash}")
return 0
if __name__ == "__main__": if __name__ == "__main__":
+208 -20
View File
@@ -1,10 +1,21 @@
"""Validate a scene package (or a legacy import) without a GPU. """Validate a scene package, or import a legacy export and validate that.
Stage 2 deliverable (``docs/architecture/studio-3d-stages.md``): the checks the Stage 2 deliverable (``docs/architecture/studio-3d-stages.md``): the checks the
loader and the compiler must agree on, runnable on a build machine and in CI. loader and the compiler must agree on, runnable on a build machine and in CI.
Until stage 2 lands, importing an asset is not implemented; the argument parsing Exit codes: 0 = every check passed, 2 = a check failed (with the reason printed),
and the exit contract exist so the CI job can be wired now. 1 = the arguments or the files were wrong.
Examples::
# a package produced by compile_scene.py
python tools/scene_assets/validate_scene.py build/finger.oryscene
# the legacy export, before packaging (units are stated, never guessed)
python tools/scene_assets/validate_scene.py \\
--legacy-dir sensors/finger/assets/30fps-0703 \\
--map sensors/finger/assets/30fps-0703/unroll_map_uv_s0.4.npz \\
--meters-per-source-unit 0.001
""" """
from __future__ import annotations from __future__ import annotations
@@ -14,31 +25,208 @@ import json
import sys import sys
from pathlib import Path from pathlib import Path
_STAGE = "stage 2 (asset contracts)"
def _bootstrap_paths() -> None:
here = Path(__file__).resolve()
repo = here.parents[2]
for path in (repo / "apps" / "studio" / "src", repo / "shared" / "src"):
if path.is_dir() and str(path) not in sys.path:
sys.path.insert(0, str(path))
def _legacy_report(args) -> tuple[int, dict]:
import numpy as np
from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError, validate_assets
from orisys_app_shared.scene3d.legacy import (
LegacyFingerImporter,
estimate_meters_per_source_unit,
load_legacy_domain,
split_at_uv_seams,
uv_correspondence_residual,
)
report: dict = {"source": "legacy", "dir": str(args.legacy_dir), "checks": []}
def record(name: str, ok: bool, detail: str = "") -> None:
report["checks"].append({"check": name, "ok": bool(ok), "detail": detail})
map_path = Path(args.map)
domain = load_legacy_domain(map_path)
report["field_domain"] = {
"shape_hw": list(domain.shape_hw),
"flip_v": domain.flip_v,
"source_scale": domain.source_scale,
"domain_id": domain.domain_id,
"valid_ratio": round(domain.valid_ratio, 4),
}
record("field domain reads", True, f"{domain.shape_hw} flip_v={domain.flip_v}")
with np.load(Path(args.legacy_dir) / "mesh.npz", allow_pickle=False) as handle:
positions, uvs, _, _ = split_at_uv_seams(
handle["vertices"], handle["faces"], handle["uv_data"], handle["loop_to_vertex"]
)
report["mesh"] = {
"render_vertices": int(len(positions)),
"source_vertices": int(len(np.load(Path(args.legacy_dir) / "mesh.npz")["vertices"])),
"extent_source_units": [float(v) for v in np.ptp(positions, axis=0)],
}
if domain.world_xyz is not None:
residual, samples = uv_correspondence_residual(positions, uvs, domain)
report["uv_correspondence"] = {
"relative_residual": round(residual, 4),
"samples": int(samples),
}
record(
"UV convention agrees with world_xyz",
residual < 0.05,
f"residual {residual:.4f} over {samples} samples",
)
measured = estimate_meters_per_source_unit(positions)
report["scale"] = {
"meters_per_source_unit_if_span_is_1m": round(measured, 6),
"stated": args.meters_per_source_unit,
"extent_m": (
[float(v) * args.meters_per_source_unit for v in np.ptp(positions, axis=0)]
if args.meters_per_source_unit
else None
),
}
if not args.meters_per_source_unit:
report["status"] = "needs_units"
report["note"] = (
"state --meters-per-source-unit: the importer never guesses what a source "
"unit is. Compare 'extent_source_units' with the sensor's real size; a "
"15.8-unit span is 15.8 mm on the shipped finger sensor, i.e. 0.001."
)
return 1, report
try:
assets = LegacyFingerImporter().convert(
Path(args.legacy_dir),
map_path,
transform_sensor_from_source=np.eye(4),
meters_per_source_unit=float(args.meters_per_source_unit),
)
except ThreeDError as exc:
record("import", False, f"{exc.diagnostic.code}: {exc.diagnostic.message}")
report["status"] = "rejected"
return 2, report
record("import", True, f"{assets.sensing_part().positions_sensor_m.shape[0]} vertices")
try:
validate_assets(assets)
record("contracts", True)
except ThreeDError as exc:
record("contracts", False, f"{exc.diagnostic.code}: {exc.diagnostic.message}")
if args.sensor_model:
expected = ExpectedGeometry(
sensor_model=args.sensor_model,
hardware_revision=args.hardware_revision,
geometry_id=assets.geometry_id,
field_domain_id=assets.field_domain_id,
shape_hw=assets.shape_hw,
)
try:
validate_assets(assets, expected)
record("device description", True, args.sensor_model)
except ThreeDError as exc:
record("device description", False, exc.diagnostic.code)
report["geometry_id"] = assets.geometry_id
report["asset_hash"] = assets.asset_hash
report["binding"] = {
"rows": int(assets.binding.sample_indices.shape[0]),
"weight_sum_min": float(np.asarray(assets.binding.sample_weights).sum(axis=1).min()),
"weight_sum_max": float(np.asarray(assets.binding.sample_weights).sum(axis=1).max()),
}
report["status"] = "ok" if all(item["ok"] for item in report["checks"]) else "failed"
return (0 if report["status"] == "ok" else 2), report
def _package_report(args) -> tuple[int, dict]:
from orisys_app_shared.scene3d.contracts import ExpectedGeometry, ThreeDError
from orisys_app_shared.scene3d.package import (
describe_package,
read_package,
scene_from_contents,
)
report: dict = {"source": "package", "path": str(args.asset), "checks": []}
def record(name: str, ok: bool, detail: str = "") -> None:
report["checks"].append({"check": name, "ok": bool(ok), "detail": detail})
try:
contents = read_package(Path(args.asset))
record("envelope, manifest and every entry hash", True)
except ThreeDError as exc:
record("envelope, manifest and every entry hash", False,
f"{exc.diagnostic.code}: {exc.diagnostic.message}")
report["status"] = "rejected"
return 2, report
report["manifest"] = describe_package(Path(args.asset))
try:
expected = (
ExpectedGeometry(
args.sensor_model, args.hardware_revision,
str(contents.manifest.get("geometry_id")),
str((contents.manifest.get("field_domain") or {}).get("id")),
tuple(int(v) for v in (contents.manifest.get("field_domain") or {}).get("shape_hw", (0, 0))),
)
if args.sensor_model
else None
)
assets = scene_from_contents(contents, expected)
record("scene builds and validates", True,
f"{assets.sensing_part().positions_sensor_m.shape[0]} vertices")
except ThreeDError as exc:
record("scene builds and validates", False,
f"{exc.diagnostic.code}: {exc.diagnostic.message}")
report["status"] = "ok" if all(item["ok"] for item in report["checks"]) else "failed"
return (0 if report["status"] == "ok" else 2), report
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("asset", type=Path, help="scene package or legacy asset directory") parser.add_argument("asset", nargs="?", type=Path, help="scene package to validate")
parser.add_argument("--housing", type=Path, default=None, help="housing GLB to check") parser.add_argument("--legacy-dir", type=Path, default=None, help="legacy NPZ export")
parser.add_argument("--map", type=Path, default=None, help="legacy field map (npz)")
parser.add_argument(
"--meters-per-source-unit",
type=float,
default=None,
help="stated unit scale; without it the legacy report stops at 'needs_units'",
)
parser.add_argument("--sensor-model", default="finger")
parser.add_argument("--hardware-revision", default="*")
parser.add_argument("--json", action="store_true", help="machine-readable output") parser.add_argument("--json", action="store_true", help="machine-readable output")
args = parser.parse_args(argv) args = parser.parse_args(argv)
report = { _bootstrap_paths()
"status": "not_implemented", if args.legacy_dir is not None:
"asset": str(args.asset), if args.map is None:
"stage": _STAGE, print("--legacy-dir also needs --map", file=sys.stderr)
"note": ( return 1
"Checks: manifest/schema, signature and entry hashes, coordinate frame " code, report = _legacy_report(args)
"and units, part roles, index ranges, binding weights and domain, " elif args.asset is not None:
"geometry/domain match against the device description." code, report = _package_report(args)
),
}
if args.json:
print(json.dumps(report, indent=2, ensure_ascii=False))
else: else:
print(f"{report['status']}: {report['note']}") parser.print_help()
return 2 return 1
if args.json:
print(json.dumps(report, indent=2, ensure_ascii=False, default=str))
else:
print(f"{report['status']}: {report.get('path') or report.get('dir')}")
for item in report.get("checks", []):
mark = "ok " if item["ok"] else "FAIL"
print(f" {mark} {item['check']} {item['detail']}")
for key in ("field_domain", "mesh", "uv_correspondence", "scale", "binding"):
if key in report:
print(f" {key}: {json.dumps(report[key], ensure_ascii=False, default=str)}")
if "note" in report:
print(f" note: {report['note']}")
return code
if __name__ == "__main__": if __name__ == "__main__":