ykcheungab
273 lines
10 KiB
Python
273 lines
10 KiB
Python
"""Stage 3 acceptance for the frame adapters.
|
|
|
|
The adapters are the boundary between the acquisition thread's arrays and the
|
|
scene, so the tests are about *what they publish and what they refuse*: a field in
|
|
the right grid, read-only, with the right unit and quality — and nothing at all when
|
|
the grid does not match, with a diagnostic that says which shapes were seen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Optional
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys_app_shared.scene3d import synthetic
|
|
from orisys_app_shared.scene3d.contracts import (
|
|
ExpectedGeometry,
|
|
FrameIdentity,
|
|
ThreeDError,
|
|
validate_surface_frame,
|
|
)
|
|
from orisys_studio.scene3d.adapters import (
|
|
DEPTH_RESPONSE,
|
|
FLOW_MAGNITUDE,
|
|
BoxFrameAdapter,
|
|
FingerFrameAdapter,
|
|
flow_magnitude,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FakePacket:
|
|
"""The parts of ``SdkPacket`` the adapters look at."""
|
|
|
|
flow: Optional[np.ndarray] = None
|
|
depth: Optional[np.ndarray] = None
|
|
valid_mask: Optional[np.ndarray] = None
|
|
frame_id: int = 1
|
|
|
|
|
|
def _identity(generation: int = 1) -> FrameIdentity:
|
|
return FrameIdentity("live", 1, 1, 0.1, generation=generation)
|
|
|
|
|
|
@pytest.fixture
|
|
def scene():
|
|
return synthetic.poc_scene()
|
|
|
|
|
|
# ------------------------------------------------------------------- flow helper
|
|
def test_the_flow_magnitude_handles_both_shapes():
|
|
flow = np.zeros((2, 3, 2), dtype=np.float32)
|
|
flow[..., 0] = 3.0
|
|
flow[..., 1] = 4.0
|
|
assert np.allclose(flow_magnitude(FakePacket(flow=flow)), 5.0)
|
|
planar = np.arange(4, dtype=np.float32).reshape(2, 2)
|
|
assert np.array_equal(flow_magnitude(FakePacket(flow=planar)), planar)
|
|
assert flow_magnitude(FakePacket()) is None
|
|
assert flow_magnitude(FakePacket(flow=np.zeros((2, 2, 3), dtype=np.float32))) is None
|
|
|
|
|
|
# ----------------------------------------------------------------- finger adapter
|
|
def test_the_finger_adapter_publishes_a_field_already_in_the_domain(scene):
|
|
adapter = FingerFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
flow = np.zeros((*scene.shape_hw, 2), dtype=np.float32)
|
|
flow[0, 0, 0] = 3.0
|
|
frame = adapter.snapshot(FakePacket(flow=flow), runtime=None, identity=_identity())
|
|
assert frame is not None
|
|
assert validate_surface_frame(frame) == ()
|
|
assert frame.geometry_id == scene.geometry_id
|
|
assert frame.field_domain_id == scene.field_domain_id
|
|
assert frame.scalar_descriptor == FLOW_MAGNITUDE
|
|
assert frame.scalar_descriptor.unit == "px"
|
|
assert frame.scalar_descriptor.quality == "approximate"
|
|
assert np.asarray(frame.scalar)[0, 0] == 3.0
|
|
assert np.asarray(frame.valid).all()
|
|
assert not np.asarray(frame.scalar).flags["WRITEABLE"]
|
|
assert frame.displacement_sensor_m is None, (
|
|
"an uncalibrated displacement must not be invented"
|
|
)
|
|
|
|
|
|
def test_the_finger_adapter_remaps_through_the_unroll_map():
|
|
map_x = np.array([[0, 1, 2], [0, 1, 2], [0, 1, 2]], dtype=np.float32)
|
|
map_y = np.array([[2, 2, 2], [1, 1, 1], [0, 0, 0]], dtype=np.float32)
|
|
adapter = FingerFrameAdapter(
|
|
geometry_id="g", field_domain_id="d", shape_hw=(3, 3),
|
|
map_x=map_x, map_y=map_y,
|
|
)
|
|
flow = np.zeros((3, 3, 2), dtype=np.float32)
|
|
flow[..., 0] = 2.0
|
|
frame = adapter.snapshot(FakePacket(flow=flow), runtime=None, identity=_identity())
|
|
assert frame is not None
|
|
assert np.allclose(np.asarray(frame.scalar), 2.0)
|
|
assert np.asarray(frame.valid).all()
|
|
|
|
|
|
def test_a_remap_with_a_hole_is_invalid_there_and_does_not_read_as_zero():
|
|
map_x = np.array([[0, -1], [0, 0]], dtype=np.float32)
|
|
map_y = np.array([[0, -1], [0, 0]], dtype=np.float32)
|
|
adapter = FingerFrameAdapter(
|
|
geometry_id="g", field_domain_id="d", shape_hw=(2, 2),
|
|
map_x=map_x, map_y=map_y,
|
|
)
|
|
flow = np.full((2, 2, 2), 1.0, dtype=np.float32)
|
|
frame = adapter.snapshot(FakePacket(flow=flow), runtime=None, identity=_identity())
|
|
assert frame is not None
|
|
assert np.asarray(frame.valid).tolist() == [[True, False], [True, True]]
|
|
assert np.isnan(np.asarray(frame.scalar)[0, 1])
|
|
|
|
|
|
def test_the_finger_adapter_refuses_a_grid_it_cannot_place(scene):
|
|
"""No remap and the wrong grid: say which shapes were seen, and publish nothing."""
|
|
adapter = FingerFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
flow = np.zeros((40, 30, 2), dtype=np.float32)
|
|
frame = adapter.snapshot(FakePacket(flow=flow), runtime=None, identity=_identity())
|
|
assert frame is None
|
|
item = adapter.last_diagnostic
|
|
assert item is not None and item.code == "field_domain_mismatch"
|
|
assert item.details["flow"] == "40x30"
|
|
assert item.details["domain"] == f"{scene.shape_hw[0]}x{scene.shape_hw[1]}"
|
|
|
|
|
|
def test_the_finger_adapter_without_flow_says_so(scene):
|
|
adapter = FingerFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
assert adapter.snapshot(FakePacket(), runtime=None, identity=_identity()) is None
|
|
assert adapter.last_diagnostic is not None
|
|
assert adapter.last_diagnostic.code == "field_unavailable"
|
|
|
|
|
|
def test_the_finger_adapter_can_be_built_from_a_domain(scene):
|
|
from orisys_app_shared.scene3d.legacy import load_legacy_domain
|
|
|
|
map_path = _run_map()
|
|
if map_path is None:
|
|
pytest.skip("the shipped unroll map is not present")
|
|
domain = load_legacy_domain(map_path)
|
|
adapter = FingerFrameAdapter.from_domain(domain, geometry_id="geom")
|
|
assert adapter.shape_hw == domain.shape_hw
|
|
assert adapter.field_domain_id == domain.domain_id
|
|
assert adapter.map_x is not None and adapter.map_y is not None, (
|
|
"the domain must carry the unroll map for the adapter to use"
|
|
)
|
|
assert domain.source_hw == (288, 512), domain.source_hw
|
|
|
|
|
|
def _run_map():
|
|
from pathlib import Path
|
|
|
|
path = Path(r"D:\orisys_ws\orisys-sdk\sensors\finger\assets\30fps-0703\unroll_map_uv_s0.4.npz")
|
|
return path if path.is_file() else None
|
|
|
|
|
|
# -------------------------------------------------------------------- box adapter
|
|
def test_the_box_adapter_prefers_the_depth_response(scene):
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
depth = np.full(scene.shape_hw, 0.25, dtype=np.float32)
|
|
flow = np.ones((*scene.shape_hw, 2), dtype=np.float32)
|
|
frame = adapter.snapshot(
|
|
FakePacket(depth=depth, flow=flow),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
|
|
)
|
|
assert frame is not None
|
|
assert frame.scalar_descriptor == DEPTH_RESPONSE
|
|
assert frame.scalar_descriptor.unit == "1", "a deformation response is not a length"
|
|
assert np.allclose(np.asarray(frame.scalar), 0.25)
|
|
|
|
|
|
def test_the_box_adapter_falls_back_to_the_flow_magnitude(scene):
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
flow = np.zeros((*scene.shape_hw, 2), dtype=np.float32)
|
|
flow[..., 0] = 1.0
|
|
frame = adapter.snapshot(
|
|
FakePacket(flow=flow),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
|
|
)
|
|
assert frame is not None and frame.scalar_descriptor == FLOW_MAGNITUDE
|
|
assert np.allclose(np.asarray(frame.scalar), 1.0)
|
|
|
|
|
|
def test_the_device_mask_wins_over_the_finiteness_test(scene):
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
depth = np.ones(scene.shape_hw, dtype=np.float32)
|
|
mask = np.ones(scene.shape_hw, dtype=bool)
|
|
mask[0, 0] = False
|
|
frame = adapter.snapshot(
|
|
FakePacket(depth=depth, valid_mask=mask),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
|
|
)
|
|
assert frame is not None
|
|
assert not bool(np.asarray(frame.valid)[0, 0])
|
|
assert np.asarray(frame.valid).sum() == mask.sum()
|
|
|
|
|
|
def test_the_box_adapter_refuses_a_device_with_another_domain(scene):
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
frame = adapter.snapshot(
|
|
FakePacket(depth=np.ones(scene.shape_hw, dtype=np.float32)),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, (9, 9)),
|
|
)
|
|
assert frame is None
|
|
assert adapter.last_diagnostic is not None
|
|
assert adapter.last_diagnostic.code == "field_domain_mismatch"
|
|
|
|
|
|
def test_the_box_adapter_with_nothing_usable_says_what_it_saw(scene):
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
frame = adapter.snapshot(
|
|
FakePacket(depth=np.ones((5, 5), dtype=np.float32)),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
|
|
)
|
|
assert frame is None
|
|
assert adapter.last_diagnostic is not None
|
|
assert adapter.last_diagnostic.details["depth"] == "5x5"
|
|
|
|
|
|
def test_a_published_frame_must_not_share_memory_with_the_packet(scene):
|
|
"""The producer keeps writing its arrays; the frame must not alias them."""
|
|
adapter = BoxFrameAdapter(
|
|
geometry_id=scene.geometry_id,
|
|
field_domain_id=scene.field_domain_id,
|
|
shape_hw=scene.shape_hw,
|
|
)
|
|
depth = np.zeros(scene.shape_hw, dtype=np.float32)
|
|
frame = adapter.snapshot(
|
|
FakePacket(depth=depth),
|
|
_identity(),
|
|
ExpectedGeometry("box", "rev", scene.geometry_id, scene.field_domain_id, scene.shape_hw),
|
|
)
|
|
assert frame is not None
|
|
assert not frame.scalar.flags["WRITEABLE"]
|
|
with pytest.raises(ValueError):
|
|
np.asarray(frame.scalar)[0, 0] = 5.0
|