orisys-app/apps/studio/tests/test_scene3d_view.py

"""Stage 4 acceptance: the 3D pane in the product.

Four things are being checked here, all of them about integration rather than about
mathematics (the camera, the sampler and the composer have their own files):

* **the pane degrades into a sentence.** No GL context, no package, a refused
  package, the wrong sensor type: each one is a line of text in the operator's
  language, and every one of them leaves the two-dimensional application alone.
* **the pane is one state with two controls.** The View menu entry and the checkbox
  inside the pane are the same flag, and toggling them repeatedly does not drift or
  leak.
* **the record is not affected.** The 3D sink runs on the acquisition thread next to
  the record sink, so the same session is run twice — once with a consumer attached
  to the pane, once without — and the CSV bytes and the valid-sample count must be
  identical. This is the stage-1 note ("开/关三维的 CSV 行序一致") being paid off.
* **the saved settings are repaired, not trusted.** A preferences file is editable by
  hand; it must not be able to make the window fail to open or make a stale file
  relabel a live measurement.
"""

from __future__ import annotations

import itertools
import shutil
from dataclasses import replace
from pathlib import Path

import numpy as np
import pytest

import ui_check
from orisys_app_shared.scene3d import synthetic
from orisys_app_shared.scene3d.contracts import (
    ColorScale,
    DisplaySettings,
    FieldDescriptor,
    UNAVAILABLE_DESCRIPTOR,
)
from orisys_app_shared.scene3d.package import write_package
from orisys_studio.scene3d import prefs as scene3d_prefs
from orisys_studio.scene3d.asset_source import (
    PACKAGE_SUFFIX,
    SceneSource,
    find_package,
    load_package,
    no_asset_diagnostic,
)
from tests.test_ui_smoke import _finish, _running_window

#: Scratch root for the two-session comparison. One subdirectory per run, because
#: the point of the test is that two recordings are identical, not that one file
#: can be overwritten.
RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record_3d"


@pytest.fixture
def next_record_root():
    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
    RECORD_SCRATCH.mkdir(parents=True, exist_ok=True)
    counter = itertools.count()

    def _next() -> Path:
        path = RECORD_SCRATCH / f"run{next(counter)}"
        path.mkdir(parents=True, exist_ok=True)
        return path

    yield _next
    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)


@pytest.fixture
def write_poc(tmp_path):
    """A real package on disk, produced by the shipping writer."""

    def _write(name: str = "finger-display-1.0.0") -> "object":
        path = tmp_path / f"{name}{PACKAGE_SUFFIX}"
        write_package(
            synthetic.poc_scene(),
            path,
            sensor_model="finger",
            hardware_revisions=("*",),
        )
        return path

    return _write


def _pump(qt_app, predicate, *, timeout_s: float = 10.0) -> bool:
    """Let Qt deliver queued signals until ``predicate`` holds."""
    import time

    deadline = time.perf_counter() + timeout_s
    while time.perf_counter() < deadline:
        qt_app.processEvents()
        if predicate():
            return True
        time.sleep(0.01)
    return False


# ------------------------------------------------------------------- the pane
def test_a_box_sensor_reports_that_it_has_no_3d_scene(qt_app):
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("box")
        assert panel.showing_message
        assert panel.view is None, "a box sensor must not build a GL widget at all"
        assert "3D" in panel.message.text() or "三维" in panel.message.text()
    finally:
        panel.deleteLater()


def test_a_missing_package_names_where_it_looked(tmp_path, qt_app):
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("finger", package_override=tmp_path / "nope.oryscene")
        assert panel.showing_message
        diagnostic = panel.last_diagnostic
        assert diagnostic is not None and diagnostic.code == "asset_rejected"
        assert "directory" in diagnostic.details
        assert "compile_scene.py" in diagnostic.details["compile"]
        assert panel.message.text(), "the pane showed an empty line instead of the reason"
    finally:
        panel.deleteLater()


def test_the_view_is_built_once_and_only_for_a_finger(qt_app, write_poc):
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("box")
        assert panel.ensure_view() is None
        panel.set_sensor("finger", package_override=write_poc())
        first = panel.ensure_view()
        second = panel.ensure_view()
        assert first is not None and first is second, "the pane rebuilt its GL widget"
        assert panel.scene_path() is not None
    finally:
        panel.close_view()
        panel.deleteLater()


def test_the_pane_fits_the_width_the_window_reserves_for_it(qt_app):
    """A pane whose content demands more than its declared minimum clips itself."""
    from orisys_studio.app_window import MIN_PANEL_WIDTHS
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        wanted = view.minimumSizeHint().width()
        assert wanted <= MIN_PANEL_WIDTHS["mesh3d"], (
            f"the 3D pane needs {wanted} px but the window reserves only "
            f"{MIN_PANEL_WIDTHS['mesh3d']}"
        )
        assert view.gl_widget is None or view.gl_widget.minimumHeight() >= 100, (
            "the picture must keep a usable height when the pane is short"
        )
    finally:
        view.close()


def test_a_platform_without_a_context_says_so_instead_of_showing_an_empty_view(qt_app):
    """The offscreen plugin cannot create a context; that must not look like "no data".

    Qt reports this by never calling ``initializeGL``, so the widget asks Qt the one
    question that has an answer — is there a context at all — on its first paint.
    """
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    panel = Mesh3dPanel()
    try:
        panel.set_sensor("finger")
        view = panel.ensure_view()
        if view is None:
            pytest.skip("the pane did not build a view here")
        # The pane has to be a shown window for Qt to lay out and paint its child: a
        # child of a hidden parent is never painted, which is why the widget only
        # learns that it has no context once something is actually on screen.
        panel.resize(240, 400)
        panel.show()
        view.show()
        qt_app.processEvents()
        widget = view.gl_widget
        for _ in range(5):
            widget.repaint()
            qt_app.processEvents()
        if widget.capabilities() is not None:
            pytest.skip("this platform has a real GL context")
        assert widget.unavailable, "a context-less widget claimed to be usable"
        assert panel.showing_message, "the pane kept showing an empty GL widget"
        assert panel.message.text().strip(), "the pane showed an empty line"
    finally:
        panel.close_view()
        panel.deleteLater()


# --------------------------------------------------------- one state, two controls
def test_the_pane_and_the_view_menu_are_the_same_flag(make_window):
    window = make_window(sensor_type="finger")
    try:
        assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
        for _ in range(20):
            window.act_finger_3d.setChecked(not window.act_finger_3d.isChecked())
            assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
            assert window.act_finger_3d.isChecked() == bool(window._show_finger_3d)
        # The pane's own checkbox is the other way in, and it drives the same flag.
        view = window.mesh3d_panel.view
        if view is not None and view.panel is not None:
            view.panel.enable_check.setChecked(not window._show_finger_3d)
            assert window.act_finger_3d.isChecked() == bool(window._show_finger_3d)
            assert window.mesh3d_panel.enabled() == bool(window._show_finger_3d)
    finally:
        window.close()


def test_panel_controls_do_not_touch_the_algorithm_side(make_window):
    """The observation camera is display state; nothing else may move with it."""
    window = make_window(sensor_type="finger")
    try:
        before = {
            "units": window.toolbar.units(),
            "image_view_mode": window.arrow_panel.view_mode(),
            "arrow_rotation_deg": window.arrow_panel.rotation_deg(),
            "view_preset": window._view_preset,
            "sensor_type": window.sensor_picker.sensor_type(),
            "splitter": list(window.content_splitter.sizes()),
            "samples": window.session_store.count,
        }
        view = window.mesh3d_panel.view
        if view is None:
            pytest.skip("no 3D view on this platform")
        camera = view._camera
        start = np.asarray(camera.matrices((240, 180)).view, dtype=float)
        camera.orbit((30.0, 10.0), (240, 180))
        camera.pan((5.0, 5.0), (240, 180))
        camera.zoom(2.0)
        moved = np.asarray(camera.matrices((240, 180)).view, dtype=float)
        assert not np.allclose(start, moved), "the camera did not react to the mouse"
        # Still a rigid transform: no mirroring, no scale change.
        rotation = moved[:3, :3]
        assert np.linalg.det(rotation) == pytest.approx(1.0, abs=1e-6)
        assert rotation @ rotation.T == pytest.approx(np.eye(3), abs=1e-6)

        after = {
            "units": window.toolbar.units(),
            "image_view_mode": window.arrow_panel.view_mode(),
            "arrow_rotation_deg": window.arrow_panel.rotation_deg(),
            "view_preset": window._view_preset,
            "sensor_type": window.sensor_picker.sensor_type(),
            "splitter": list(window.content_splitter.sizes()),
            "samples": window.session_store.count,
        }
        assert after == before
    finally:
        window.close()


# ------------------------------------------------------------------ record isolation
class _RecordingView:
    """A stand-in for the pane's view that records what it was handed.

    Used to attach a *real* consumer to the real sink path on a machine with no GL:
    what is being tested is whether the window's second sink can change the record,
    not whether the adapter can compute a flow magnitude.
    """

    def __init__(self) -> None:
        self.packets: list = []
        self.coordinator = None

    def submit_packet(self, packet) -> None:
        # Deliberately more work than the real one does per packet, so a leak into the
        # record path would show up as a difference rather than as a timing artefact.
        self.packets.append(packet)
        frame = np.asarray(packet.raw_frame)
        float(np.asarray(frame, dtype=np.float64).sum())

    def render_stats(self):
        return None


def _run_session(make_window, record_root, *, with_3d: bool):
    """One full session with the 3D sink attached, or with no pane at all."""
    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: record_root
    view = None
    if with_3d:
        view = _RecordingView()
        window.mesh3d_panel._view = view
    else:
        window.mesh3d_panel._view = None
    window.toolbar.btn_record.setChecked(True)
    for index in range(96):
        stuck.emit(ui_check._fake_packet(index, mono_t=index / 30.0))
        window._on_run_tick()
    window.toolbar.btn_record.setChecked(False)
    report = window._record_last_report
    rows = ""
    if report is not None and (report.directory / "forces.csv").is_file():
        rows = (report.directory / "forces.csv").read_bytes().decode("utf-8")
    counts = (window.session_store.count, len(list(window.session_store.iter_valid("raw"))))
    _finish(window, stuck)
    window.close()
    return rows, counts, report, view


def test_the_3d_consumer_cannot_change_the_record(make_window, next_record_root):
    """The acceptance row: identical CSV and identical valid samples, 3D on or off."""
    with_3d_csv, with_3d_counts, with_3d_report, view = _run_session(
        make_window, next_record_root(), with_3d=True
    )
    without_csv, without_counts, without_report, _ = _run_session(
        make_window, next_record_root(), with_3d=False
    )
    assert view is not None and len(view.packets) == 96, (
        "the pane's sink did not receive the session's packets"
    )
    assert with_3d_csv, "the recording produced no forces.csv"
    assert with_3d_csv == without_csv, (
        "the record changed when a 3D consumer was attached to the same packets"
    )
    assert with_3d_counts == without_counts, (with_3d_counts, without_counts)
    assert with_3d_report is not None and without_report is not None
    assert len(with_3d_report.problems) == len(without_report.problems)
    header = with_3d_csv.splitlines()[0]
    assert "t_rel_s" in header and len(with_3d_csv.splitlines()) > 10


def test_the_3d_sink_is_registered_and_inert_when_the_pane_is_off(make_window, next_record_root):
    """A sink that has to be added and removed around a toggle is a state machine."""
    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: next_record_root()
    try:
        assert window._on_packet_recorded in stuck.sinks()
        assert window._on_packet_for_3d in stuck.sinks(), (
            "the 3D sink must be registered with the worker, not attached per toggle"
        )
        window.mesh3d_panel._view = None
        for index in range(10):
            stuck.emit(ui_check._fake_packet(index))
        assert window.session_store.count == 10
        assert window.mesh3d_panel.view is None
    finally:
        _finish(window, stuck)
        window.close()


def test_repeated_closes_are_safe(make_window):
    window = make_window(sensor_type="finger")
    try:
        window.mesh3d_panel.ensure_view()
        window.mesh3d_panel.close_view()
        window.mesh3d_panel.close_view()
        window.close()
        window.close()
    finally:
        pass


# ------------------------------------------------------------------ preferences
def test_the_display_settings_round_trip():
    settings = DisplaySettings(
        display_gain=12.0,
        deform_enabled=True,
        housing_visible=False,
        max_fps=60,
        color_scale=ColorScale(
            descriptor=FieldDescriptor("flow_magnitude", "px", "approximate", False),
            lower=0.25,
            upper=4.5,
            lut_id="sequential",
            mode="fixed",
        ),
    )
    restored = scene3d_prefs.settings_from_prefs(scene3d_prefs.settings_to_prefs(settings))
    assert restored.display_gain == 12.0
    assert restored.deform_enabled is True
    assert restored.housing_visible is False
    assert restored.max_fps == 60
    assert restored.color_scale is not None
    assert (restored.color_scale.lower, restored.color_scale.upper) == (0.25, 4.5)
    assert restored.color_scale.mode == "fixed"


def test_a_saved_descriptor_is_never_restored_as_a_measurement():
    """The colour bar describes the live field, not the last one that was on screen."""
    block = scene3d_prefs.settings_to_prefs(
        DisplaySettings(
            color_scale=ColorScale(
                descriptor=FieldDescriptor("displacement", "m", "calibrated", True),
                lower=0.0,
                upper=1.0,
                lut_id="diverging",
                mode="fixed",
            )
        )
    )
    restored = scene3d_prefs.settings_from_prefs(block)
    assert restored.color_scale is not None
    assert restored.color_scale.descriptor == UNAVAILABLE_DESCRIPTOR

    given = FieldDescriptor("flow_magnitude", "px", "approximate", False)
    restored_with = scene3d_prefs.settings_from_prefs(block, descriptor=given)
    assert restored_with.color_scale.descriptor == given


@pytest.mark.parametrize(
    "block",
    [
        None,
        {},
        {"display_gain": "lots", "max_fps": "fast", "deform_enabled": "maybe"},
        {"display_gain": -5.0, "max_fps": 17},
        {"display_gain": 1e9, "max_fps": 30},
        {"color_scale": {"mode": "sideways", "lower": "a", "upper": "b"}},
        {"color_scale": {"mode": "fixed", "lower": 5.0, "upper": 5.0}},
        {"color_scale": {"mode": "fixed", "lower": float("nan"), "upper": float("inf")}},
        {"color_scale": {"mode": "fixed", "lower": 1e300, "upper": 1e301}},
    ],
)
def test_a_hand_edited_settings_block_is_repaired(block):
    settings = scene3d_prefs.settings_from_prefs(block)
    assert isinstance(settings, DisplaySettings)
    assert scene3d_prefs.MIN_GAIN <= settings.display_gain <= scene3d_prefs.MAX_GAIN
    assert settings.max_fps in scene3d_prefs.FPS_CHOICES
    if settings.color_scale is not None:
        lower, upper = settings.color_scale.lower, settings.color_scale.upper
        assert lower == lower and upper == upper, "a NaN survived the repair"
        assert upper > lower
        assert abs(lower) <= scene3d_prefs.MAX_SCALE_BOUND
        assert abs(upper) <= scene3d_prefs.MAX_SCALE_BOUND
        assert settings.color_scale.mode in scene3d_prefs.SCALE_MODES


def test_the_panel_offers_exactly_the_rates_the_repair_accepts(qt_app):
    from orisys_studio.scene3d.panel import FPS_CHOICES

    assert tuple(FPS_CHOICES) == tuple(scene3d_prefs.FPS_CHOICES)


# ------------------------------------------------------------------- asset source
def test_the_package_search_order_is_override_then_sensor_directory(tmp_path, write_poc):
    explicit = write_poc("explicit")
    sensor_dir = tmp_path / "sensors" / "finger" / "scene"
    sensor_dir.mkdir(parents=True)
    shipped = sensor_dir / f"shipped{PACKAGE_SUFFIX}"
    shipped.write_bytes(explicit.read_bytes())

    found = find_package("finger", override=explicit, root=tmp_path)
    assert found is not None and found.path == explicit
    found_dir = find_package("finger", root=tmp_path)
    assert found_dir is not None and found_dir.path == shipped
    assert find_package("box", root=tmp_path) is None
    assert find_package("finger", override=tmp_path / "gone.oryscene", root=tmp_path) is not None, (
        "a stale override must fall through to the shipped package, not disable 3D"
    )


def test_a_package_is_loaded_off_the_gui_thread(qt_app, write_poc):
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    path = write_poc()
    loader = SceneAssetLoader()
    loaded: list = []
    failed: list = []
    loader.loaded.connect(lambda assets, description: loaded.append((assets, description)))
    loader.failed.connect(failed.append)
    try:
        loader.load(SceneSource(path=path, kind="package", directory=path.parent))
        assert _pump(qt_app, lambda: bool(loaded or failed)), "the loader never answered"
    finally:
        loader.close()
    assert not failed, [item.message for item in failed]
    assets, description = loaded[0]
    assert assets.geometry_id == synthetic.poc_scene().geometry_id
    assert str(path) in description
    # A second load while one is in flight is the same request, not a queue.
    loader2 = SceneAssetLoader()
    try:
        loader2._inflight = True
        loader2.load(SceneSource(path=path, kind="package", directory=path.parent))
        assert loader2._executor is None
    finally:
        loader2.close()


def test_a_corrupt_package_is_reported_not_raised(qt_app, tmp_path):
    from orisys_studio.scene3d.asset_source import SceneAssetLoader

    broken = tmp_path / f"broken{PACKAGE_SUFFIX}"
    broken.write_bytes(b"not a zip archive at all")
    loader = SceneAssetLoader()
    failed: list = []
    loader.failed.connect(failed.append)
    try:
        loader.load(SceneSource(path=broken, kind="package", directory=tmp_path))
        assert _pump(qt_app, lambda: bool(failed)), "a corrupt package produced no answer"
    finally:
        loader.close()
    assert failed[0].code in ("asset_rejected", "asset_auth_failed")
    assert failed[0].message


def test_loading_a_missing_package_raises_a_diagnostic(tmp_path):
    from orisys_app_shared.scene3d.contracts import ThreeDError

    with pytest.raises(ThreeDError) as excinfo:
        load_package(SceneSource(path=tmp_path / "gone.oryscene", kind="package", directory=tmp_path))
    assert excinfo.value.diagnostic.code == "asset_rejected"


def test_the_missing_asset_message_is_not_localised_but_the_pane_is(tmp_path):
    """Diagnostics are logged and quoted in support tickets: they stay English."""
    diagnostic = no_asset_diagnostic("finger")
    assert diagnostic.code == "asset_rejected"
    assert PACKAGE_SUFFIX in diagnostic.message
    assert "compile_scene.py" in diagnostic.details["compile"]


def test_a_finger_scene_uses_the_finger_adapter(tmp_path, write_poc, qt_app):
    """The adapter is chosen by the asset, and a box scene without a device refuses."""
    from orisys_studio.scene3d.view import Scene3DView

    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        assert view.use_scene(synthetic.poc_scene(), source_id="poc")
        assert view.assets is not None
    finally:
        view.close()


# ------------------------------------------------------- stage 5: the record block
def test_a_recording_says_what_the_3d_pane_was(
    make_window, next_record_root, write_poc, qt_app, monkeypatch
):
    """``session.json`` carries the pane's state, and an event carries the totals."""
    import json

    import orisys_studio.app_window as app_module

    window, stuck = _running_window(make_window)
    app_module.recordings_dir = lambda: next_record_root()
    try:
        # A real view on a machine with no GL is not possible; the stand-in gives the
        # metadata assembly something to describe (an enabled pane with a scene).
        view = _RecordingView()
        view.assets = None
        window.mesh3d_panel._view = view
        window.toolbar.btn_record.setChecked(True)
        for index in range(20):
            stuck.emit(ui_check._fake_packet(index, mono_t=index / 30.0))
        window._on_run_tick()
        window.toolbar.btn_record.setChecked(False)
        report = window._record_last_report
        assert report is not None and report.completed or report is not None
        if report is None or not report.completed:
            pytest.skip("the writer had not finished on this machine")
        meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8"))
        block = meta["visualization3d"]
        assert block["block_version"] == 1
        assert block["enabled"] is False, (
            "the pane was not switched on, so the record must say so"
        )
        # The pane has no scene here (no package installed), so the block names the
        # reason instead of pretending to describe a picture.
        assert block.get("reason")
        events = (report.directory / "events.csv").read_text(encoding="utf-8")
        assert "visualization3d" in events, (
            "the display totals were not recorded as an event"
        )
    finally:
        _finish(window, stuck)
        window.close()


def test_a_platform_locked_scene_is_refused_in_the_application(qt_app):
    """A lock violation is a *pane* failure: the view refuses the scene and says why."""
    from orisys_app_shared.scene3d.contracts import PlatformInfo, PlatformLock
    from orisys_studio.scene3d.view import Scene3DView
    from orisys_studio.ui.mesh3d_panel import Mesh3dPanel

    here = PlatformInfo.here()
    elsewhere = "linux" if here.os_family != "linux" else "windows"
    locked = replace(
        synthetic.poc_scene(),
        platform_lock=PlatformLock(os_family=(elsewhere,), machine=("aarch64",)),
    )
    view = Scene3DView()
    try:
        if not view.available:
            pytest.skip("no GL binding here")
        reported: list = []
        view.diagnosticRaised.connect(reported.append)
        assert view.use_scene(locked, source_id="locked") is False
        assert view.assets is None, "a refused scene must not remain adopted"
        assert any(item.code == "platform_mismatch" for item in reported), reported
    finally:
        view.close()

    panel = Mesh3dPanel()
    try:
        # The pane shows the reason rather than an empty picture: `platform_mismatch`
        # is one of the terminal diagnostics.
        panel._on_diagnostic(
            __import__(
                "orisys_app_shared.scene3d.contracts", fromlist=["diagnostic"]
            ).diagnostic("platform_mismatch", "this package is not released for this OS")
        )
        assert panel.showing_message
        assert "not released" in panel.message.text()
    finally:
        panel.deleteLater()