orisys-app/shared/tests/test_scene3d_contracts.py

"""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
from pathlib import Path

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_an_unstated_device_identity_is_reported_not_matched():
    """An empty field is "the device cannot say", never "it matches"."""
    device = c.ExpectedGeometry("finger", "", "", "domain-xyz", (3, 4), source_geometry_id="")
    warnings = c.validate_assets(_assets(), device)
    assert [item.code for item in warnings] == ["geometry_unverified"]
    assert device.source_geometry_id == ""


def test_the_source_geometry_must_match_too():
    """Same size, same mapping, different surface: still not the same geometry."""
    device = c.ExpectedGeometry(
        "finger", "", "", "domain-xyz", (3, 4), source_geometry_id="src-1"
    )
    assert c.validate_assets(_assets(source_geometry_id="src-1"), device) == ()
    with pytest.raises(c.ThreeDError) as excinfo:
        c.validate_assets(_assets(source_geometry_id="src-2"), device)
    assert excinfo.value.diagnostic.code == "geometry_mismatch"
    assert excinfo.value.diagnostic.details["device"] == "src-1"

    # A package that predates the record: unverified, and said so.
    warnings = c.validate_assets(_assets(), device)
    assert [item.code for item in warnings] == ["geometry_unverified"]


def test_a_package_released_for_another_sensor_or_revision_is_refused():
    device = c.ExpectedGeometry("finger", "rev-b", "geom-abc", "domain-xyz", (3, 4))
    with pytest.raises(c.ThreeDError) as excinfo:
        c.validate_assets(_assets(compatible_sensor_model="box"), device)
    assert excinfo.value.diagnostic.code == "geometry_mismatch"

    with pytest.raises(c.ThreeDError) as excinfo:
        c.validate_assets(
            _assets(
                compatible_sensor_model="finger",
                compatible_hardware_revisions=("rev-a",),
            ),
            device,
        )
    assert excinfo.value.diagnostic.code == "geometry_mismatch"
    # ``*`` is "any revision", so it never refuses.
    assert c.validate_assets(
        _assets(
            compatible_sensor_model="finger",
            compatible_hardware_revisions=("*",),
        ),
        device,
    ) == ()
    # An unstated revision on the device side is not a mismatch either.
    assert c.validate_assets(
        _assets(
            compatible_sensor_model="finger",
            compatible_hardware_revisions=("rev-a",),
        ),
        c.ExpectedGeometry("finger", "", "geom-abc", "domain-xyz", (3, 4)),
    ) == ()


def test_the_source_geometry_digest_is_stable_and_shape_checked():
    from orisys_app_shared.scene3d.digest import source_geometry_digest

    world = np.arange(3 * 4 * 3, dtype=np.float32).reshape(3, 4, 3)
    assert source_geometry_digest(world) == source_geometry_digest(world.copy())
    moved = world + 1.0
    assert source_geometry_digest(moved) != source_geometry_digest(world)
    with pytest.raises(c.ThreeDError) as excinfo:
        source_geometry_digest(np.zeros((3, 4), dtype=np.float32))
    assert excinfo.value.diagnostic.code == "geometry_mismatch"


def test_only_four_vertex_lod_counts_as_the_same_binding():
    """A new topology needs a new binding: 1032 and 1005 are not interchangeable."""
    binding = _binding(vertices=3)
    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_layer_is_importable_and_works_without_a_gpu():
    """The acquisition thread imports this package; it must not drag Qt or GL in.

    Stages 2 and 3 filled in the CPU pieces, so this checks they are reachable and
    usable here — the sampler and the composer run on a machine with no GPU, which
    is the whole reason the contracts layer is separate from the view.
    """
    from orisys_app_shared.scene3d import assets, composer, sampling, synthetic

    scene = synthetic.poc_scene()
    frame = _frame()
    fields = sampling.sample_field(frame, _binding())
    assert np.asarray(fields.scalar).shape == (4,)
    composed = composer.SceneComposer().compose(
        scene, c.SurfaceFieldFrame(
            identity=frame.identity,
            geometry_id=scene.geometry_id,
            field_domain_id=scene.field_domain_id,
            scalar=c.freeze(np.zeros(scene.shape_hw, dtype=np.float32), dtype=np.float32),
            scalar_descriptor=c.FieldDescriptor("flow_magnitude", "px", "approximate", False),
            valid=c.freeze(np.ones(scene.shape_hw, dtype=bool), dtype=np.bool_),
        ),
        c.DisplaySettings(),
    )
    assert composed.scalar.shape == (9,)
    # The loading facade is real too (stage 2), and the compiler refuses what it
    # cannot honestly do: asking for encryption without key material, or without the
    # crypto backend, must not produce a plain package that looks protected.
    assert callable(assets.PackagedAssetLoader().load)
    with pytest.raises(c.ThreeDError) as excinfo:
        assets.SceneCompiler().compile(
            scene,
            None,
            Path("unused.oryscene"),
            c.CompileOptions(
                asset_id="x",
                version="0",
                expected=synthetic.poc_expected(),
                transform_sensor_from_glb=np.eye(4),
                max_surface_vertices=0,
                max_housing_triangles=1000,
                signing_key_reference="build-key",
                encryption_key_id="customer-1",
            ),
        )
    assert "key" in excinfo.value.diagnostic.message
    assert excinfo.value.diagnostic.code == "asset_auth_failed"


def test_summarize_parts_counts_vertices_per_part():
    counts = c.summarize_parts((_part(), _part("housing", "housing", 3)))
    assert counts == {"sensing_surface": 4, "housing": 3}