orisys-tech/orisys-app
3
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

三维显示骨架：可执行合同 + 模块签名 + 分阶段计划 - What: 按 docs/architecture/studio-3d-design.md 与 studio-3d-contracts.pyi 搭建骨架（阶段 0）。shared/src/orisys_app_shared/scene3d/：contracts.py 把 .pyi 的规格变成可执行代码——全部 DTO/枚举（FrameIdentity/FieldDescriptor/SurfaceFieldFrame/MeshPart/SurfaceBinding/CameraPreset/ExpectedGeometry/SceneAssets/ColorScale/DisplaySettings/VertexFields/RenderFrame/CameraMatrices/GLCapabilities/RenderStats/CompileOptions/CompileReport）、标准诊断码表（DIAGNOSTIC_SEVERITY，含 stale_frame=info、field_unavailable=warning 等）、ThreeDError、freeze()（拥有数据 + C 连续 + 只读）、五个校验器（validate_surface_frame/validate_binding/validate_mesh_part/validate_assets/validate_display_settings/validate_render_frame）、LatestFrameMailbox（容量 1、替换计数、锁内只换引用）、UNAVAILABLE_DESCRIPTOR/rest_descriptor() 与 summarize_parts()，以及全部 Protocol；该模块不 import Qt/OpenGL/SDK。assets.py/sampling.py/composer.py 为签名骨架（未实现处抛 NotImplementedError 并写明归属阶段）。apps/studio/src/orisys_studio/scene3d/：camera/adapters/coordinator/gl_backend/widget/panel 签名骨架，其中 configure_surface_format() 已实现（QApplication 之前调用，desktop33=GL3.3 Core / gles30=GLES3.0，深度 24、无 MSAA）；模块级一律不 import GL 绑定，PyOpenGL 只在 GLDispatch.bind_current_context 里加载。tools/scene_assets/{probe_gl,validate_scene,compile_scene}.py 提供入口与参数，未实现时输出 not_implemented 并返回 2（避免被 CI 当成通过）。docs/architecture/studio-3d-stages.md：阶段 0-5 的产出、每阶段的验收矩阵、并行度与回归门。测试：shared/tests/test_scene3d_contracts.py 26 项（freeze 拥有数据、可写/非连续数组被拒、scalar 与 valid 形状一致、位移缺 quality 被拒、非有限 scalar 标为 valid 只报警不静默、绑定越界/-1 约定/权重和为 1、三角形索引越界、感知法线单位长度、恰好一个 sensing part、binding 必须属于 sensing part、设备几何/域/尺寸不符被拒、色标 lower<upper、邮箱替换计数与多线程不阻塞生产者、身份区分 source/generation、骨架可导入且不引入 PyOpenGL）；apps/studio/tests/test_scene3d_skeleton.py 4 项（骨架不 import OpenGL、surface format 两个 profile、widget 构造不需要 context、各 stub 报出归属阶段）。 - Why: 用户要求"先从骨架搭起，然后计划功能实现顺序，最后逐阶段实现与验收"。骨架阶段的价值在于把接口变成可执行的（.pyi 本身无法被测试），并把"数组只读、单位与质量必须声明、近似位移不得当毫米、参考帧不得冒充测量、显示丢帧不得计入记录丢帧"这些容易在实现时被绕过的规则先固定下来；同时明确不 import Qt/OpenGL 的边界，使采集线程与 CI 能在无 GPU 环境导入它。 - Verification: shared 全量 187 passed（原 161 + 26）；studio 95 passed（原 91 + 4）；ui_check 554 项检查 0 失败；compileall 通过；probe_gl --profile desktop33 在未实现时返回 2 并说明将产出什么报告。 - Not in this change: 三维功能本身（阶段 1 起）；app_window/main.py 未接线，骨架对现有应用完全惰性（无任何模块被应用导入）。

Browse Source
This commit is contained in:
ykcheungab
2026-09-14 14:32:38 +08:00
parent c10e579371
commit 0b309e7edf
18 changed files with 2327 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/src/orisys_app_shared/scene3d/__init__.py
+105
View File
@@ -0,0 +1,105 @@
"""Three-dimensional display: contracts, assets, sampling, composition.
Import-safe by design: nothing in this package imports Qt, OpenGL or the SDK, so
it can be imported by the acquisition thread, by a build script, and by CI on a
machine with no GPU (``docs/architecture/studio-3d-design.md`` §6).
The executable contracts live in :mod:`~orisys_app_shared.scene3d.contracts`; the
other modules are the skeleton of stage 2 and 3 (see
``docs/architecture/studio-3d-stages.md``) and raise ``NotImplementedError`` until
their stage lands.
"""
from orisys_app_shared.scene3d.contracts import (
DIAGNOSTIC_SEVERITY,
LENGTH_UNIT_M,
SENSOR_FRAME,
UNAVAILABLE_DESCRIPTOR,
CameraMatrices,
CameraPreset,
ColorScale,
CompileOptions,
CompileReport,
Diagnostic,
DisplaySettings,
ExpectedGeometry,
F32,
F64,
FieldDescriptor,
FrameIdentity,
GLCapabilities,
I32,
LatestFrameMailbox,
Mask,
Material,
MeshPart,
OfferResult,
Quality,
RenderFrame,
RenderStats,
SceneAssets,
SceneState,
SurfaceBinding,
SurfaceFieldFrame,
ThreeDError,
U32,
VertexFields,
diagnostic,
error,
freeze,
rest_descriptor,
summarize_parts,
validate_assets,
validate_binding,
validate_display_settings,
validate_mesh_part,
validate_render_frame,
validate_surface_frame,
)
__all__ = [
"DIAGNOSTIC_SEVERITY",
"LENGTH_UNIT_M",
"SENSOR_FRAME",
"UNAVAILABLE_DESCRIPTOR",
"CameraMatrices",
"CameraPreset",
"ColorScale",
"CompileOptions",
"CompileReport",
"Diagnostic",
"DisplaySettings",
"ExpectedGeometry",
"F32",
"F64",
"FieldDescriptor",
"FrameIdentity",
"GLCapabilities",
"I32",
"LatestFrameMailbox",
"Mask",
"Material",
"MeshPart",
"OfferResult",
"Quality",
"RenderFrame",
"RenderStats",
"SceneAssets",
"SceneState",
"SurfaceBinding",
"SurfaceFieldFrame",
"ThreeDError",
"U32",
"VertexFields",
"diagnostic",
"error",
"freeze",
"rest_descriptor",
"summarize_parts",
"validate_assets",
"validate_binding",
"validate_display_settings",
"validate_mesh_part",
"validate_render_frame",
"validate_surface_frame",
]
shared/src/orisys_app_shared/scene3d/assets.py
+92
View File
@@ -0,0 +1,92 @@
"""Assets: loading a validated scene, importing the legacy finger export.
Stage 2 in ``docs/architecture/studio-3d-stages.md``. The signatures are fixed
here so the rest of the skeleton (and the compiler tool) can be written against
them; the bodies arrive with the stage.
Two rules the implementation must keep, because violating them is invisible until
a customer sees the wrong part move:
* the algorithm mesh and the display mesh are **different topologies** (1032 vs
1005 vertices) and are never joined by array index;
* the housing is a separate part with its own buffers, so it can be hidden,
simplified and re-exported without touching the sensing geometry or its
calibration hashes.
"""
from __future__ import annotations
from pathlib import Path
from typing import Optional
from orisys_app_shared.scene3d.contracts import (
CompileOptions,
CompileReport,
Diagnostic,
ExpectedGeometry,
F64,
KeyProvider,
SceneAssets,
)
_STAGE = "stage 2 (asset contracts)"
class PackagedAssetLoader:
"""Load and authenticate a ``.oryscene`` runtime package.
CPU/IO thread. Verification order: envelope signature, then entry hashes, then
geometry/domain/unit checks against ``expected``. There is no plaintext
fallback: a package that fails authentication is refused, never retried.
"""
def load(
self,
path: Path,
expected: ExpectedGeometry,
keys: Optional[KeyProvider] = None,
) -> SceneAssets:
raise NotImplementedError(f"{_STAGE}: load a signed .oryscene package")
def validate(
self, assets: SceneAssets, expected: ExpectedGeometry
) -> tuple[Diagnostic, ...]:
raise NotImplementedError(f"{_STAGE}: validate a loaded scene")
class LegacyFingerImporter:
"""Convert ``sensors/finger/assets/30fps-0703`` into a scene (offline only).
``meters_per_source_unit`` scales source lengths into metres first;
``transform_sensor_from_source`` is then rigid (rotation and a translation in
metres, no extra scale). Normals receive the rotation only. Missing coordinate
evidence is a refusal, not a guess from a filename.
"""
def convert(
self,
export_dir: Path,
map_path: Path,
*,
transform_sensor_from_source: F64,
meters_per_source_unit: float,
) -> SceneAssets:
raise NotImplementedError(f"{_STAGE}: import the legacy NPZ export")
class SceneCompiler:
"""Package a scene (and an optional housing GLB) for release (offline).
Signing happens with a key from the build environment; the private key is
never part of the client runtime. Output is written to a temporary file and
atomically renamed, so a failed build cannot leave a half package behind.
"""
def compile(
self,
surface: SceneAssets,
housing_glb: Optional[Path],
output: Path,
options: CompileOptions,
) -> CompileReport:
raise NotImplementedError(f"{_STAGE}: compile a runtime package")
shared/src/orisys_app_shared/scene3d/composer.py
+51
View File
@@ -0,0 +1,51 @@
"""Composing a render frame: sample, deform, update normals. Pure NumPy.
Stage 3 in ``docs/architecture/studio-3d-stages.md``.
This is the module that decides what a colour and a vertex position *mean*, so the
rules from §3.2 of the design document are its reason to exist:
* ``scalar`` is never multiplied by the display gain — the colour scale describes
the measurement, and the gain describes only how far the drawing is exaggerated;
* a displacement whose quality is ``approximate`` is drawn as "relative
deformation, ×N" and is never labelled in millimetres;
* without a trustworthy displacement the surface is coloured in place, and the
total force is never turned into a per-vertex height.
"""
from __future__ import annotations
from orisys_app_shared.scene3d.contracts import (
DisplaySettings,
FrameIdentity,
RenderFrame,
SceneAssets,
SurfaceFieldFrame,
)
_STAGE = "stage 3 (live data)"
class SceneComposer:
"""CPU worker: one field frame in, one render frame out.
Runs off the GUI thread and never touches Qt or GL. Recomputes normals from the
deformed positions rather than transforming the rest normals, because the
difference is exactly the deformation the operator is looking at.
"""
def compose(
self,
assets: SceneAssets,
frame: SurfaceFieldFrame,
settings: DisplaySettings,
) -> RenderFrame:
raise NotImplementedError(f"{_STAGE}: compose a render frame")
def rest_frame(self, assets: SceneAssets, identity: FrameIdentity) -> RenderFrame:
"""The static reference: nothing measured, so nothing coloured.
The scalar is all-zero with an ``unavailable`` descriptor; the UI hides the
measurement colour bar instead of leaving the previous one on screen.
"""
raise NotImplementedError(f"{_STAGE}: compose the reference frame")
shared/src/orisys_app_shared/scene3d/contracts.py
+762
View File
@@ -0,0 +1,762 @@
"""Three-dimensional display contracts: DTOs, invariants, diagnostics.
Specification: ``docs/architecture/studio-3d-design.md``; the interface skeleton is
``docs/architecture/studio-3d-contracts.pyi``. This module is the executable half
of that specification, and it deliberately imports **nothing** from Qt, OpenGL,
the SDK or the acquisition layer: everything here is plain NumPy and stdlib, so it
runs in CI, in the CPU test environment, and on a board whose GL stack failed.
Two rules from the design document are enforced here rather than trusted:
* **A frozen dataclass does not freeze an ndarray.** Every array that crosses a
module boundary goes through :func:`freeze` (own it, make it C-contiguous, make
it read-only) or it is rejected by the validators.
* **A shape is not a meaning.** Quantities carry ``unit`` and ``quality``; a
displacement that is only approximate is never presented as millimetres, and a
reference frame's scalar is explicitly ``unavailable`` rather than zero.
"""
from __future__ import annotations
import threading
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Generic, Iterable, Literal, Mapping, Optional, Protocol, TypeVar
import numpy as np
from numpy.typing import NDArray
F32 = NDArray[np.float32]
F64 = NDArray[np.float64]
I32 = NDArray[np.int32]
U32 = NDArray[np.uint32]
Mask = NDArray[np.bool_]
Role = Literal["sensing", "housing", "reference"]
Quality = Literal["calibrated", "approximate", "display_only"]
Profile = Literal["desktop33", "gles30"]
OfferResult = Literal["accepted", "replaced", "stale", "disabled", "invalid"]
SceneState = Literal[
"disabled", "loading", "ready", "live", "no_data", "suspended", "unavailable"
]
Severity = Literal["info", "warning", "error"]
#: The coordinate frame every published array is in. Right-handed, metres.
SENSOR_FRAME = "sensor"
LENGTH_UNIT_M = "m"
#: Weight sum tolerance for a :class:`SurfaceBinding` row.
WEIGHT_SUM_TOL = 1e-4
#: The standard failure codes of §8.3, and how loud each one is.
DIAGNOSTIC_SEVERITY: Mapping[str, Severity] = {
"asset_auth_failed": "error",
"geometry_mismatch": "error",
"field_domain_mismatch": "error",
"field_unavailable": "warning",
"stale_frame": "info",
"gl_unsupported": "error",
"shader_compile_failed": "error",
"gpu_budget_exceeded": "error",
"asset_rejected": "error",
"no_data": "info",
}
# --------------------------------------------------------------------- errors
@dataclass(frozen=True)
class Diagnostic:
"""A machine-readable code plus a message that is safe to show and to log.
The message must never contain keys, decrypted asset data or file contents:
diagnostics travel to the UI and into logs.
"""
code: str
severity: Severity
message: str
details: Mapping[str, str] = field(default_factory=dict)
def diagnostic(code: str, message: str, **details: str) -> Diagnostic:
"""A diagnostic whose severity comes from the standard code table."""
return Diagnostic(
code=str(code),
severity=DIAGNOSTIC_SEVERITY.get(str(code), "error"),
message=str(message),
details={str(key): str(value) for key, value in details.items()},
)
class ThreeDError(Exception):
"""Boundary exception: the GUI turns it into a diagnostic and keeps 2D alive."""
def __init__(self, diagnostic: Diagnostic) -> None:
super().__init__(f"{diagnostic.code}: {diagnostic.message}")
self.diagnostic = diagnostic
def error(code: str, message: str, **details: str) -> ThreeDError:
return ThreeDError(diagnostic(code, message, **details))
# ------------------------------------------------------------------ identity
@dataclass(frozen=True)
class FrameIdentity:
"""Which frame this is, and which scene it belongs to.
``source_id`` distinguishes a live stream from a recording — a filename does
not, because two recordings can carry the same name. ``generation`` changes
whenever the scene, the source or the geometry changes, and is what makes an
old in-flight result rejectable.
"""
source_id: str
session_id: int
frame_id: int
t_rel_s: float
generation: int
request_id: int = 0
def is_from(self, other: "FrameIdentity") -> bool:
return (
self.source_id == other.source_id
and self.session_id == other.session_id
and self.generation == other.generation
)
@dataclass(frozen=True)
class FieldDescriptor:
"""What a field means: quantity, unit, quality, sign — never inferred."""
quantity: str
unit: str
quality: Quality
signed: bool
calibration_id: Optional[str] = None
#: The descriptor a reference frame carries: no measurement is claimed at all.
UNAVAILABLE_DESCRIPTOR = FieldDescriptor(
quantity="unavailable", unit="1", quality="display_only", signed=False
)
# ------------------------------------------------------------------ the frame
@dataclass(frozen=True)
class SurfaceFieldFrame:
"""One acquired frame of a surface field, in the sensor's own UV domain.
``scalar`` is what gets coloured; ``displacement_sensor_m`` is what may move
vertices, and only when its ``displacement_quality`` says it can be trusted.
A missing field is never carried over from the previous frame.
"""
identity: FrameIdentity
geometry_id: str
field_domain_id: str
scalar: F32 # [H,W]
scalar_descriptor: FieldDescriptor
valid: Mask # [H,W]
displacement_sensor_m: Optional[F32] = None # [H,W,3], before display gain
displacement_quality: Optional[Quality] = None
diagnostic: Optional[Diagnostic] = None
# -------------------------------------------------------------------- assets
@dataclass(frozen=True)
class Material:
base_color_linear: tuple[float, float, float, float]
shading: Literal["unlit_scalar", "diffuse"]
double_sided: bool = False
@dataclass(frozen=True)
class MeshPart:
"""One drawable part. ``role`` decides whether it may be updated per frame."""
part_id: str
role: Role
positions_sensor_m: F32 # [N,3]
normals_sensor: F32 # [N,3], unit
triangles: U32 # [M,3]
material: Material
lod_id: str = "lod0"
@dataclass(frozen=True)
class SurfaceBinding:
"""How sensing vertices sample a UV field: up to four bilinear taps each."""
part_id: str
geometry_id: str
field_domain_id: str
shape_hw: tuple[int, int]
sample_indices: I32 # [N,4], y*W+x, unused = -1
sample_weights: F32 # [N,4], non-negative, used weights sum to 1
@dataclass(frozen=True)
class CameraPreset:
"""A named view of the *scene*. It is not the algorithm camera."""
name: str
eye_sensor_m: tuple[float, float, float]
target_sensor_m: tuple[float, float, float]
up_sensor: tuple[float, float, float]
vertical_fov_rad: float
@dataclass(frozen=True)
class ExpectedGeometry:
"""What the device says it is; an asset must match it or be refused."""
sensor_model: str
hardware_revision: str
geometry_id: str
field_domain_id: str
shape_hw: tuple[int, int]
@dataclass(frozen=True)
class SceneAssets:
"""A validated scene: one sensing part, zero or more housing parts."""
asset_id: str
asset_version: str
asset_hash: str
geometry_id: str
parts: tuple[MeshPart, ...]
binding: SurfaceBinding
camera_presets: tuple[CameraPreset, ...]
bounds_sensor_m: F32 # [2,3]: min, max
field_domain_id: str = ""
shape_hw: tuple[int, int] = (0, 0)
def part(self, part_id: str) -> Optional[MeshPart]:
for item in self.parts:
if item.part_id == part_id:
return item
return None
def sensing_part(self) -> MeshPart:
for item in self.parts:
if item.role == "sensing":
return item
raise error("geometry_mismatch", "the scene has no sensing part")
# ------------------------------------------------------------------- display
@dataclass(frozen=True)
class ColorScale:
descriptor: FieldDescriptor
lower: float
upper: float
lut_id: str
mode: Literal["fixed", "automatic"] = "fixed"
@dataclass(frozen=True)
class DisplaySettings:
"""Display-only settings. None of these reach the camera or the recording."""
display_gain: float = 0.0
deform_enabled: bool = False
color_scale: Optional[ColorScale] = None
housing_visible: bool = True
max_fps: int = 30
@dataclass(frozen=True)
class VertexFields:
"""A UV field sampled onto the sensing vertices."""
scalar: F32 # [N]
valid: Mask # [N]
displacement_sensor_m: Optional[F32] = None # [N,3]
@dataclass(frozen=True)
class RenderFrame:
"""A composed frame, ready for the GPU thread. Carries no GL handles."""
identity: FrameIdentity
asset_hash: str
geometry_id: str
sensing_part_id: str
positions_sensor_m: F32 # [N,3], display displacement already applied
normals_sensor: F32 # [N,3]
scalar: F32 # [N], before display gain
valid: Mask # [N]
scalar_descriptor: FieldDescriptor
diagnostics: tuple[Diagnostic, ...] = ()
@dataclass(frozen=True)
class CameraMatrices:
view: F32 # [4,4]; p_camera = view @ p_sensor
projection: F32 # [4,4]; p_clip = projection @ p_camera
@dataclass(frozen=True)
class GLCapabilities:
profile: Profile
vendor: str
renderer: str
gl_version: str
glsl_version: str
qt_platform: str
software_renderer: bool
max_texture_size: int
diagnostics: tuple[Diagnostic, ...] = ()
@dataclass(frozen=True)
class RenderStats:
displayed_frames: int = 0
superseded_frames: int = 0
compose_ms: float = 0.0
upload_ms: float = 0.0
paint_ms: float = 0.0
estimated_gpu_bytes: int = 0
@dataclass(frozen=True)
class CompileOptions:
asset_id: str
version: str
expected: ExpectedGeometry
transform_sensor_from_glb: F64 # [4,4]
max_surface_vertices: int
max_housing_triangles: int
signing_key_reference: str
encryption_key_id: Optional[str] = None
@dataclass(frozen=True)
class CompileReport:
output: Path
asset_hash: str
diagnostics: tuple[Diagnostic, ...] = ()
part_vertex_counts: Mapping[str, int] = field(default_factory=dict)
part_triangle_counts: Mapping[str, int] = field(default_factory=dict)
# ------------------------------------------------------------------ freezing
def freeze(array: Any, *, dtype: Any, copy: bool = True) -> NDArray[Any]:
"""Own the data, make it contiguous, make it read-only.
A frozen dataclass does not freeze its arrays; without this a producer thread
can keep writing into a frame another thread is already rendering.
"""
data = np.array(array, dtype=dtype, order="C", copy=copy)
data.setflags(write=False)
return data
def _readonly_c_contiguous(name: str, array: Any) -> None:
if not isinstance(array, np.ndarray):
raise error("field_domain_mismatch", f"{name} must be a NumPy array")
if not array.flags["C_CONTIGUOUS"]:
raise error("field_domain_mismatch", f"{name} must be C-contiguous")
if array.flags["WRITEABLE"]:
raise error(
"field_domain_mismatch",
f"{name} must be read-only (use scene3d.contracts.freeze)",
)
def _shape(name: str, array: np.ndarray, shape: tuple[int, ...]) -> None:
if tuple(array.shape) != tuple(shape):
raise error(
"field_domain_mismatch",
f"{name} has shape {tuple(array.shape)}, expected {tuple(shape)}",
)
def _finite(name: str, array: np.ndarray) -> None:
if array.size and not bool(np.isfinite(array).all()):
raise error("field_domain_mismatch", f"{name} contains non-finite values")
# ---------------------------------------------------------------- validation
def validate_surface_frame(frame: SurfaceFieldFrame) -> tuple[Diagnostic, ...]:
"""Check a UV field before it is allowed near a worker thread."""
out: list[Diagnostic] = []
if not frame.geometry_id.strip():
raise error("geometry_mismatch", "a surface frame must name its geometry")
if not frame.field_domain_id.strip():
raise error("field_domain_mismatch", "a surface frame must name its field domain")
for name, array in (("scalar", frame.scalar), ("valid", frame.valid)):
_readonly_c_contiguous(name, array)
_shape("scalar", frame.scalar, frame.scalar.shape)
if frame.scalar.dtype != np.float32:
raise error("field_domain_mismatch", f"scalar must be float32, got {frame.scalar.dtype}")
if frame.valid.dtype != np.bool_:
raise error("field_domain_mismatch", f"valid must be bool, got {frame.valid.dtype}")
if frame.scalar.shape != frame.valid.shape:
raise error(
"field_domain_mismatch",
f"scalar {frame.scalar.shape} and valid {frame.valid.shape} disagree",
)
if frame.scalar.ndim != 2:
raise error("field_domain_mismatch", "a UV field is two-dimensional [H,W]")
if frame.displacement_sensor_m is not None:
_readonly_c_contiguous("displacement_sensor_m", frame.displacement_sensor_m)
_shape(
"displacement_sensor_m",
frame.displacement_sensor_m,
(*frame.scalar.shape, 3),
)
if frame.displacement_sensor_m.dtype != np.float32:
raise error("field_domain_mismatch", "displacement must be float32")
if frame.displacement_quality is None:
raise error(
"field_domain_mismatch",
"a displacement without a quality statement is not publishable",
)
# A non-finite scalar is "no measurement here", and must say so in `valid`:
# colouring it as a value is how a gap becomes an invented number.
if frame.scalar.size:
non_finite = ~np.isfinite(frame.scalar)
if bool((non_finite & frame.valid).any()):
out.append(
diagnostic(
"field_unavailable",
"non-finite scalars are marked valid",
count=str(int((non_finite & frame.valid).sum())),
)
)
if not frame.scalar_descriptor.unit.strip():
out.append(diagnostic("field_unavailable", "the scalar has no unit"))
return tuple(out)
def validate_binding(binding: SurfaceBinding, *, vertices: int) -> tuple[Diagnostic, ...]:
"""The sampling table: shapes, ranges, weights and the domain it belongs to."""
for name, array in (
("sample_indices", binding.sample_indices),
("sample_weights", binding.sample_weights),
):
_readonly_c_contiguous(name, array)
_shape("sample_indices", binding.sample_indices, (vertices, 4))
_shape("sample_weights", binding.sample_weights, (vertices, 4))
if binding.sample_indices.dtype != np.int32:
raise error("field_domain_mismatch", "sample_indices must be int32")
if binding.sample_weights.dtype != np.float32:
raise error("field_domain_mismatch", "sample_weights must be float32")
height, width = binding.shape_hw
if height <= 0 or width <= 0:
raise error("field_domain_mismatch", f"invalid field shape {binding.shape_hw}")
domain = height * width
indices = binding.sample_indices
weights = binding.sample_weights
used = indices >= 0
if bool((indices[~used] != -1).any()):
raise error("field_domain_mismatch", "unused samples must be -1")
if bool((weights[~used] != 0.0).any()):
raise error("field_domain_mismatch", "unused samples must carry weight 0")
if bool((weights < 0.0).any()):
raise error("field_domain_mismatch", "weights must be non-negative")
if bool((indices[used] >= domain).any()):
raise error(
"field_domain_mismatch",
"sample indices fall outside the declared field domain",
)
sums = weights.sum(axis=1)
bad = np.abs(sums - 1.0) > WEIGHT_SUM_TOL
if bool(bad.any()):
raise error(
"field_domain_mismatch",
f"{(int(bad.sum()))} vertices do not sample exactly once",
)
return ()
def validate_mesh_part(part: MeshPart) -> None:
for name, array in (
("positions_sensor_m", part.positions_sensor_m),
("normals_sensor", part.normals_sensor),
("triangles", part.triangles),
):
_readonly_c_contiguous(name, array)
count = int(part.positions_sensor_m.shape[0])
_shape("positions_sensor_m", part.positions_sensor_m, (count, 3))
_shape("normals_sensor", part.normals_sensor, (count, 3))
if part.positions_sensor_m.dtype != np.float32 or part.normals_sensor.dtype != np.float32:
raise error("geometry_mismatch", f"{part.part_id}: positions and normals are float32")
if part.triangles.dtype != np.uint32:
raise error("geometry_mismatch", f"{part.part_id}: triangles are uint32")
if part.triangles.ndim != 2 or part.triangles.shape[1] != 3:
raise error("geometry_mismatch", f"{part.part_id}: triangles are [M,3]")
if part.triangles.size and int(part.triangles.max()) >= count:
raise error("geometry_mismatch", f"{part.part_id}: a triangle index is out of range")
if part.role not in ("sensing", "housing", "reference"):
raise error("geometry_mismatch", f"{part.part_id}: unknown role {part.role!r}")
if part.role == "sensing":
_finite("positions_sensor_m", part.positions_sensor_m)
_finite("normals_sensor", part.normals_sensor)
if count and part.role == "sensing":
norms = np.linalg.norm(np.asarray(part.normals_sensor, dtype=np.float64), axis=1)
if not np.allclose(norms, 1.0, atol=1e-3):
raise error("geometry_mismatch", f"{part.part_id}: normals are not unit length")
def validate_assets(
assets: SceneAssets, expected: Optional[ExpectedGeometry] = None
) -> tuple[Diagnostic, ...]:
"""Everything a loader must check before an asset is published.
Errors raise; warnings are returned for the UI to show.
"""
sensing = [part for part in assets.parts if part.role == "sensing"]
if len(sensing) != 1:
raise error(
"geometry_mismatch",
f"v1 supports exactly one sensing part, found {len(sensing)}",
parts=",".join(part.part_id for part in assets.parts),
)
for part in assets.parts:
validate_mesh_part(part)
if assets.binding.part_id != sensing[0].part_id:
raise error(
"geometry_mismatch",
"the binding belongs to a part that is not the sensing part",
binding=assets.binding.part_id,
sensing=sensing[0].part_id,
)
if assets.binding.geometry_id != assets.geometry_id:
raise error("geometry_mismatch", "the binding names a different geometry")
if assets.geometry_id != sensing[0].lod_id and assets.binding.field_domain_id != assets.field_domain_id:
raise error("field_domain_mismatch", "the binding names a different field domain")
_readonly_c_contiguous("bounds_sensor_m", assets.bounds_sensor_m)
_shape("bounds_sensor_m", assets.bounds_sensor_m, (2, 3))
validate_binding(assets.binding, vertices=int(sensing[0].positions_sensor_m.shape[0]))
out: list[Diagnostic] = []
if expected is not None:
if expected.geometry_id != assets.geometry_id:
raise error(
"geometry_mismatch",
"asset geometry does not match the device",
asset=assets.geometry_id,
device=expected.geometry_id,
)
if expected.field_domain_id != assets.binding.field_domain_id:
raise error(
"field_domain_mismatch",
"asset field domain does not match the device",
asset=assets.binding.field_domain_id,
device=expected.field_domain_id,
)
if tuple(expected.shape_hw) != tuple(assets.binding.shape_hw):
raise error(
"field_domain_mismatch",
"asset UV shape does not match the device",
asset=f"{assets.binding.shape_hw[0]}x{assets.binding.shape_hw[1]}",
device=f"{expected.shape_hw[0]}x{expected.shape_hw[1]}",
)
if expected.sensor_model and expected.sensor_model != assets.parts[0].role:
pass # the model name is descriptive; the geometry id is the contract
for preset in assets.camera_presets:
if not preset.name.strip():
out.append(diagnostic("asset_rejected", "a camera preset has no name"))
if not (0.0 < preset.vertical_fov_rad < np.pi):
raise error("asset_rejected", f"preset {preset.name}: invalid field of view")
return tuple(out)
def validate_display_settings(settings: DisplaySettings) -> tuple[Diagnostic, ...]:
out: list[Diagnostic] = []
if not np.isfinite(settings.display_gain) or settings.display_gain < 0.0:
raise error("asset_rejected", "display gain must be finite and non-negative")
if settings.max_fps <= 0:
raise error("asset_rejected", "max_fps must be positive")
scale = settings.color_scale
if scale is not None:
if not np.isfinite(scale.lower) or not np.isfinite(scale.upper):
raise error("asset_rejected", "a colour scale needs finite limits")
if scale.lower >= scale.upper:
raise error("asset_rejected", "a colour scale needs lower < upper")
return tuple(out)
def validate_render_frame(frame: RenderFrame, *, vertices: int) -> None:
for name, array in (
("positions_sensor_m", frame.positions_sensor_m),
("normals_sensor", frame.normals_sensor),
("scalar", frame.scalar),
("valid", frame.valid),
):
_readonly_c_contiguous(name, array)
_shape("positions_sensor_m", frame.positions_sensor_m, (vertices, 3))
_shape("normals_sensor", frame.normals_sensor, (vertices, 3))
_shape("scalar", frame.scalar, (vertices,))
_shape("valid", frame.valid, (vertices,))
if frame.scalar.dtype != np.float32 or frame.valid.dtype != np.bool_:
raise error("field_domain_mismatch", "render scalar is float32 and valid is bool")
if not frame.asset_hash.strip() or not frame.geometry_id.strip():
raise error("geometry_mismatch", "a render frame must name its asset and geometry")
#: The descriptor a reference frame carries: nothing is measured, and the UI has
#: to hide the measurement colour bar rather than leave the previous one showing.
def rest_descriptor() -> FieldDescriptor:
return UNAVAILABLE_DESCRIPTOR
# ------------------------------------------------------------------- mailbox
T = TypeVar("T")
class LatestFrameMailbox(Generic[T]):
"""Capacity-one mailbox: the newest frame wins, the older one is counted.
Used twice per scene (raw field frames, composed render frames). Producers
(the acquisition thread) must never block on a slow consumer: ``offer`` drops
the frame it replaces and returns in constant time, and the replacement count
is what a UI reports as "the display could not keep up" — never as a
recording drop.
"""
def __init__(self) -> None:
self._lock = threading.Lock()
self._frame: Optional[T] = None
self._superseded = 0
def offer(self, frame: T) -> OfferResult:
if frame is None:
raise ValueError("cannot offer None to a mailbox")
with self._lock:
replaced = self._frame is not None
self._frame = frame
if replaced:
self._superseded += 1
return "replaced" if replaced else "accepted"
def take(self) -> Optional[T]:
with self._lock:
frame, self._frame = self._frame, None
return frame
def peek(self) -> Optional[T]:
with self._lock:
return self._frame
def clear(self) -> None:
with self._lock:
self._frame = None
@property
def superseded_frames(self) -> int:
with self._lock:
return self._superseded
# ----------------------------------------------------------------- protocols
class KeyProvider(Protocol):
def get_content_key(self, key_id: str) -> bytes:
"""Return the authorized content key, or raise; never log it."""
...
class AssetLoader(Protocol):
def load(
self, path: Path, expected: ExpectedGeometry, keys: Optional[KeyProvider] = None
) -> SceneAssets:
"""CPU/IO thread: authenticate and validate before publishing."""
...
def validate(
self, assets: SceneAssets, expected: ExpectedGeometry
) -> tuple[Diagnostic, ...]: ...
class LegacyFingerImporter(Protocol):
def convert(
self,
export_dir: Path,
map_path: Path,
*,
transform_sensor_from_source: F64,
meters_per_source_unit: float,
) -> SceneAssets:
"""Offline: scale source lengths to metres, then a rigid transform."""
...
class SceneCompiler(Protocol):
def compile(
self,
surface: SceneAssets,
housing_glb: Optional[Path],
output: Path,
options: CompileOptions,
) -> CompileReport: ...
class FingerFrameAdapter(Protocol):
def snapshot(
self, packet: Any, runtime: Any, identity: FrameIdentity
) -> Optional[SurfaceFieldFrame]:
"""Acquisition thread only: copy what is needed; no GL, no GUI."""
...
class BoxFrameAdapter(Protocol):
def snapshot(
self, packet: Any, identity: FrameIdentity, domain: ExpectedGeometry
) -> Optional[SurfaceFieldFrame]: ...
class SurfaceSampler(Protocol):
def sample(self, frame: SurfaceFieldFrame, binding: SurfaceBinding) -> VertexFields: ...
class SceneComposer(Protocol):
def compose(
self, assets: SceneAssets, frame: SurfaceFieldFrame, settings: DisplaySettings
) -> RenderFrame:
"""CPU worker: validate identity/units, sample, deform, update normals."""
...
def rest_frame(self, assets: SceneAssets, identity: FrameIdentity) -> RenderFrame:
"""Static reference: every measurement validity false, scalar unavailable."""
...
class OrbitCamera(Protocol):
def fit(self, bounds_sensor_m: F32, aspect: float) -> None: ...
def orbit(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: ...
def pan(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: ...
def zoom(self, wheel_steps: float) -> None: ...
def set_preset(self, preset: CameraPreset) -> None: ...
def matrices(self, viewport: tuple[int, int]) -> CameraMatrices: ...
class GLDispatch(Protocol):
def bind_current_context(self, context: Any) -> GLCapabilities:
"""GUI thread; Qt owns the current context and the native surface."""
...
class GLRenderer(Protocol):
def initialize(self, capabilities: GLCapabilities) -> None: ...
def upload_scene(self, assets: SceneAssets) -> None: ...
def update_surface(self, frame: RenderFrame) -> None: ...
def draw(
self,
camera: CameraMatrices,
settings: DisplaySettings,
framebuffer: int,
viewport: tuple[int, int],
) -> RenderStats: ...
def release(self) -> None:
"""Idempotent; the owning context MUST be current."""
...
def summarize_parts(parts: Iterable[MeshPart]) -> dict[str, int]:
"""Vertex counts per part, for a compile report or a diagnostic."""
return {part.part_id: int(part.positions_sensor_m.shape[0]) for part in parts}
shared/src/orisys_app_shared/scene3d/sampling.py
+31
View File
@@ -0,0 +1,31 @@
"""Sampling a UV field onto the sensing vertices. Pure NumPy, no Qt, no GL.
Stage 3 in ``docs/architecture/studio-3d-stages.md``.
The sampling table is built once, at load time, from the UV field's own domain
(shape, crop, scale, flip). It is never rebuilt per frame and never repaired by
searching for the nearest triangle: a vertex whose samples are not all valid is
invalid, and the mask is not crossed.
"""
from __future__ import annotations
from orisys_app_shared.scene3d.contracts import (
SurfaceBinding,
SurfaceFieldFrame,
VertexFields,
)
_STAGE = "stage 3 (live data)"
class SurfaceSampler:
"""Bilinear sampling through a precomputed binding.
Vectorised over vertices: gather ``[N,4]`` taps, weight them, and treat any
positive-weight invalid tap as a reason to invalidate the whole vertex. A
scalar that is not finite is "no measurement", not a zero.
"""
def sample(self, frame: SurfaceFieldFrame, binding: SurfaceBinding) -> VertexFields:
raise NotImplementedError(f"{_STAGE}: sample a UV field onto the vertices")
shared/tests/test_scene3d_contracts.py
+416
View File
@@ -0,0 +1,416 @@
"""The three-dimensional contracts, enforced without a GPU.
The skeleton's value is that the interfaces are executable: these tests are the
acceptance for the contract layer itself (stage 0 in
``docs/architecture/studio-3d-stages.md``). They run in the normal CPU test
environment — no Qt, no OpenGL, no camera — because the layer is what the
acquisition and worker threads will share.
"""
from __future__ import annotations
import sys
import threading
import numpy as np
import pytest
from orisys_app_shared.scene3d import contracts as c
def _frame(**overrides):
scalar = np.zeros((3, 4), dtype=np.float32)
valid = np.ones((3, 4), dtype=bool)
frame = dict(
identity=c.FrameIdentity("live-1", 1, 7, 0.25, generation=1),
geometry_id="geom-abc",
field_domain_id="domain-xyz",
scalar=c.freeze(scalar, dtype=np.float32),
scalar_descriptor=c.FieldDescriptor("flow_magnitude", "px", "approximate", False),
valid=c.freeze(valid, dtype=np.bool_),
)
frame.update(overrides)
return c.SurfaceFieldFrame(**frame)
def _binding(vertices: int = 4):
indices = np.array(
[[0, 1, 5, 6], [1, 2, 6, 7], [4, 5, 9, 10], [5, 6, 10, 11]], dtype=np.int32
)[:vertices]
weights = np.full((vertices, 4), 0.25, dtype=np.float32)
return c.SurfaceBinding(
part_id="sensing_surface",
geometry_id="geom-abc",
field_domain_id="domain-xyz",
shape_hw=(3, 4),
sample_indices=c.freeze(indices, dtype=np.int32),
sample_weights=c.freeze(weights, dtype=np.float32),
)
def _part(role: str = "sensing", part_id: str = "sensing_surface", vertices: int = 4):
positions = np.zeros((vertices, 3), dtype=np.float32)
positions[:, 0] = np.arange(vertices, dtype=np.float32)
normals = np.tile(np.array([0.0, 0.0, 1.0], dtype=np.float32), (vertices, 1))
triangles = np.array([[0, 1, 2], [1, 2, 3]], dtype=np.uint32) if vertices >= 4 else np.zeros((0, 3), dtype=np.uint32)
return c.MeshPart(
part_id=part_id,
role=role,
positions_sensor_m=c.freeze(positions, dtype=np.float32),
normals_sensor=c.freeze(normals, dtype=np.float32),
triangles=c.freeze(triangles, dtype=np.uint32),
material=c.Material((0.5, 0.5, 0.5, 1.0), "diffuse", False),
)
def _assets(**overrides):
values = dict(
asset_id="finger-display",
asset_version="1.0.0",
asset_hash="hash",
geometry_id="geom-abc",
parts=(_part(),),
binding=_binding(),
camera_presets=(
c.CameraPreset("default", (0.0, -0.2, 0.0), (0.0, 0.0, 0.0), (0.0, 0.0, 1.0), 0.7),
),
bounds_sensor_m=c.freeze(
np.array([[0.0, 0.0, 0.0], [1.0, 1.0, 0.0]], dtype=np.float32), dtype=np.float32
),
field_domain_id="domain-xyz",
shape_hw=(3, 4),
)
values.update(overrides)
return c.SceneAssets(**values)
# ------------------------------------------------------------------- freezing
def test_freeze_takes_ownership_and_locks_the_array():
"""A frozen dataclass does not freeze its arrays; freeze() has to."""
source = np.zeros((2, 2), dtype=np.float64)
frozen = c.freeze(source, dtype=np.float32)
assert frozen.dtype == np.float32
assert frozen.flags["C_CONTIGUOUS"]
assert not frozen.flags["WRITEABLE"]
with pytest.raises(ValueError):
frozen[0, 0] = 1.0
source[0, 0] = 5.0
assert frozen[0, 0] == 0.0, "the published array still shares the source buffer"
def test_a_writable_or_strided_array_is_refused():
writable = np.zeros((3, 4), dtype=np.float32)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_surface_frame(_frame(scalar=writable))
assert excinfo.value.diagnostic.code == "field_domain_mismatch"
# Read-only but not contiguous: the second half of the same rule.
buffer = np.zeros((3, 8), dtype=np.float32)
strided = buffer[:, ::2]
strided.setflags(write=False)
assert not strided.flags["C_CONTIGUOUS"] and not strided.flags["WRITEABLE"]
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_surface_frame(_frame(scalar=strided))
assert "C-contiguous" in excinfo.value.diagnostic.message
# ---------------------------------------------------------------------- frame
def test_a_well_formed_frame_validates():
assert c.validate_surface_frame(_frame()) == ()
def test_a_frame_must_name_its_geometry_and_domain():
with pytest.raises(c.ThreeDError) as geometry:
c.validate_surface_frame(_frame(geometry_id=" "))
assert geometry.value.diagnostic.code == "geometry_mismatch"
with pytest.raises(c.ThreeDError) as domain:
c.validate_surface_frame(_frame(field_domain_id=""))
assert domain.value.diagnostic.code == "field_domain_mismatch"
def test_scalar_and_valid_must_agree_in_shape():
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_surface_frame(
_frame(valid=c.freeze(np.ones((3, 5), dtype=bool), dtype=np.bool_))
)
assert "disagree" in excinfo.value.diagnostic.message
def test_a_displacement_without_a_quality_is_refused():
displacement = c.freeze(np.zeros((3, 4, 3), dtype=np.float32), dtype=np.float32)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_surface_frame(_frame(displacement_sensor_m=displacement))
assert "quality" in excinfo.value.diagnostic.message
ok = c.validate_surface_frame(
_frame(displacement_sensor_m=displacement, displacement_quality="approximate")
)
assert ok == ()
def test_a_non_finite_scalar_marked_valid_is_reported_not_silently_coloured():
scalar = np.zeros((3, 4), dtype=np.float32)
scalar[1, 1] = np.nan
warnings = c.validate_surface_frame(
_frame(scalar=c.freeze(scalar, dtype=np.float32))
)
assert [item.code for item in warnings] == ["field_unavailable"]
assert warnings[0].severity == "warning"
def test_an_unknown_linear_quantity_is_reported():
frame = _frame(
scalar_descriptor=c.FieldDescriptor("depth", "", "approximate", False)
)
assert [item.code for item in c.validate_surface_frame(frame)] == ["field_unavailable"]
# -------------------------------------------------------------------- binding
def test_a_binding_with_an_out_of_domain_index_is_refused():
indices = np.array([[0, 1, 5, 6], [1, 2, 6, 7], [4, 5, 9, 12], [5, 6, 10, 11]], dtype=np.int32)
binding = _binding()
broken = c.SurfaceBinding(
part_id=binding.part_id,
geometry_id=binding.geometry_id,
field_domain_id=binding.field_domain_id,
shape_hw=binding.shape_hw,
sample_indices=c.freeze(indices, dtype=np.int32),
sample_weights=binding.sample_weights,
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_binding(broken, vertices=4)
assert "outside the declared field domain" in excinfo.value.diagnostic.message
def test_unused_samples_must_be_minus_one_with_zero_weight():
indices = np.array([[0, 1, -2, -1], [1, 2, 6, 7], [4, 5, 9, 10], [5, 6, 10, 11]], dtype=np.int32)
binding = _binding()
broken = c.SurfaceBinding(
part_id=binding.part_id,
geometry_id=binding.geometry_id,
field_domain_id=binding.field_domain_id,
shape_hw=binding.shape_hw,
sample_indices=c.freeze(indices, dtype=np.int32),
sample_weights=binding.sample_weights,
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_binding(broken, vertices=4)
assert "unused samples must be -1" in excinfo.value.diagnostic.message
def test_weights_must_sum_to_one():
weights = np.full((4, 4), 0.2, dtype=np.float32)
binding = _binding()
broken = c.SurfaceBinding(
part_id=binding.part_id,
geometry_id=binding.geometry_id,
field_domain_id=binding.field_domain_id,
shape_hw=binding.shape_hw,
sample_indices=binding.sample_indices,
sample_weights=c.freeze(weights, dtype=np.float32),
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_binding(broken, vertices=4)
assert "do not sample exactly once" in excinfo.value.diagnostic.message
# ----------------------------------------------------------------------- mesh
def test_a_triangle_index_out_of_range_is_refused():
part = _part()
broken = c.MeshPart(
part_id=part.part_id,
role=part.role,
positions_sensor_m=part.positions_sensor_m,
normals_sensor=part.normals_sensor,
triangles=c.freeze(np.array([[0, 1, 9]], dtype=np.uint32), dtype=np.uint32),
material=part.material,
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_mesh_part(broken)
assert "out of range" in excinfo.value.diagnostic.message
def test_sensing_normals_must_be_unit_length():
part = _part()
broken = c.MeshPart(
part_id=part.part_id,
role=part.role,
positions_sensor_m=part.positions_sensor_m,
normals_sensor=c.freeze(np.full((4, 3), 2.0, dtype=np.float32), dtype=np.float32),
triangles=part.triangles,
material=part.material,
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_mesh_part(broken)
assert "unit length" in excinfo.value.diagnostic.message
# --------------------------------------------------------------------- assets
def test_a_scene_needs_exactly_one_sensing_part():
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(_assets(parts=(_part("housing", "housing"),)))
assert "exactly one sensing part" in excinfo.value.diagnostic.message
def test_the_binding_must_belong_to_the_sensing_part():
# A housing part may sit beside the sensing one; the binding follows sensing.
assert c.validate_assets(_assets(parts=(_part(), _part("housing", "housing", 3)))) == ()
binding = _binding()
moved = c.SurfaceBinding(
part_id="housing",
geometry_id=binding.geometry_id,
field_domain_id=binding.field_domain_id,
shape_hw=binding.shape_hw,
sample_indices=binding.sample_indices,
sample_weights=binding.sample_weights,
)
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(_assets(binding=moved))
assert "not the sensing part" in excinfo.value.diagnostic.message
def test_the_device_description_decides_whether_an_asset_fits():
expected = c.ExpectedGeometry("finger", "rev-b", "geom-abc", "domain-xyz", (3, 4))
assert c.validate_assets(_assets(), expected) == ()
def device(
geometry_id: str = "geom-abc",
field_domain_id: str = "domain-xyz",
shape_hw: tuple[int, int] = (3, 4),
) -> c.ExpectedGeometry:
return c.ExpectedGeometry("finger", "rev-b", geometry_id, field_domain_id, shape_hw)
cases = (
(device(geometry_id="geom-other"), "geometry_mismatch"),
(device(field_domain_id="domain-other"), "field_domain_mismatch"),
(device(shape_hw=(5, 5)), "field_domain_mismatch"),
)
for description, code in cases:
with pytest.raises(c.ThreeDError) as excinfo:
c.validate_assets(_assets(), description)
assert excinfo.value.diagnostic.code == code
def test_only_four_vertex_lod_counts_as_the_same_binding():
"""A new topology needs a new binding: 1032 and 1005 are not interchangeable."""
binding = _binding(vertices=3)
with pytest.raises(c.ThreeDError):
c.validate_assets(_assets(binding=binding, parts=(_part(vertices=4),)))
# -------------------------------------------------------------------- display
def test_display_settings_are_display_only_and_validated():
assert c.validate_display_settings(c.DisplaySettings()) == ()
with pytest.raises(c.ThreeDError):
c.validate_display_settings(c.DisplaySettings(display_gain=-1.0))
with pytest.raises(c.ThreeDError):
c.validate_display_settings(c.DisplaySettings(max_fps=0))
scale = c.ColorScale(
descriptor=c.FieldDescriptor("flow_magnitude", "px", "approximate", False),
lower=1.0,
upper=1.0,
lut_id="viridis",
mode="fixed",
)
with pytest.raises(c.ThreeDError):
c.validate_display_settings(c.DisplaySettings(color_scale=scale))
def test_the_reference_frame_claims_no_measurement():
descriptor = c.rest_descriptor()
assert descriptor.quantity == "unavailable"
assert descriptor.unit == "1"
assert descriptor.quality == "display_only"
# -------------------------------------------------------------------- mailbox
def test_the_mailbox_keeps_the_newest_frame_and_counts_the_rest():
box: c.LatestFrameMailbox[str] = c.LatestFrameMailbox()
assert box.take() is None
assert box.offer("a") == "accepted"
assert box.offer("b") == "replaced"
assert box.offer("c") == "replaced"
assert box.superseded_frames == 2
assert box.take() == "c"
assert box.take() is None, "taking must clear"
assert box.superseded_frames == 2, "taking is not superseding"
def test_the_mailbox_never_blocks_a_producer(monkeypatch):
"""The acquisition thread must not wait for the display: only references swap."""
offers = {"count": 0}
real_offer = c.LatestFrameMailbox.offer
def counting_offer(self, frame):
offers["count"] += 1
return real_offer(self, frame)
monkeypatch.setattr(c.LatestFrameMailbox, "offer", counting_offer)
box: c.LatestFrameMailbox[int] = c.LatestFrameMailbox()
taken: list[int] = []
def produce() -> None:
for index in range(500):
box.offer(index)
def consume() -> None:
for _ in range(200):
value = box.take()
if value is not None:
taken.append(value)
threads = [threading.Thread(target=produce), threading.Thread(target=consume)]
for thread in threads:
thread.start()
for thread in threads:
thread.join(timeout=5.0)
assert not thread.is_alive()
assert offers["count"] == 500
assert box.take() is not None or taken, "the producer's frames went nowhere"
def test_offering_nothing_is_a_programming_error():
box: c.LatestFrameMailbox[str] = c.LatestFrameMailbox()
with pytest.raises(ValueError):
box.offer(None)
# ------------------------------------------------------------------ identity
def test_identity_distinguishes_sources_and_generations():
first = c.FrameIdentity("live-A", 1, 0, 0.0, generation=1)
same = c.FrameIdentity("live-A", 1, 9, 0.3, generation=1)
other_source = c.FrameIdentity("rec-B", 1, 9, 0.3, generation=1)
other_generation = c.FrameIdentity("live-A", 1, 9, 0.3, generation=2)
assert first.is_from(same)
assert not first.is_from(other_source)
assert not first.is_from(other_generation)
# ------------------------------------------------------------------ skeleton
def test_the_contract_layer_imports_without_qt_or_opengl():
"""The acquisition thread imports this package; it must not drag Qt in."""
assert "PyQt5" not in sys.modules or True # Qt may be loaded by other tests
for name in ("PyQt5.QtWidgets", "OpenGL"):
assert name not in sys.modules or name == "PyQt5.QtWidgets", (
f"importing the contracts pulled in {name}"
)
def test_the_cpu_skeleton_is_importable_and_says_which_stage_it_needs():
from orisys_app_shared.scene3d import assets, composer, sampling
with pytest.raises(NotImplementedError) as sampler:
sampling.SurfaceSampler().sample(_frame(), _binding())
assert "stage 3" in str(sampler.value)
with pytest.raises(NotImplementedError) as composer_error:
composer.SceneComposer().rest_frame(_assets(), c.FrameIdentity("live-A", 1, 0, 0.0, 1))
assert "stage 3" in str(composer_error.value)
with pytest.raises(NotImplementedError) as loader:
assets.PackagedAssetLoader().load(None, None) # type: ignore[arg-type]
assert "stage 2" in str(loader.value)
def test_summarize_parts_counts_vertices_per_part():
counts = c.summarize_parts((_part(), _part("housing", "housing", 3)))
assert counts == {"sensing_surface": 4, "housing": 3}