ykcheungab
417 lines
16 KiB
Python
417 lines
16 KiB
Python
"""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}
|