ykcheungab
1259 lines
50 KiB
Python
1259 lines
50 KiB
Python
"""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 math
|
||
import time
|
||
from concurrent.futures import Future
|
||
from dataclasses import replace
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pytest
|
||
import ui_check
|
||
from orisys_app_shared.i18n import t
|
||
from orisys_app_shared.scene3d import synthetic
|
||
from orisys_app_shared.scene3d.contracts import (
|
||
UNAVAILABLE_DESCRIPTOR,
|
||
ColorScale,
|
||
DisplaySettings,
|
||
FieldDescriptor,
|
||
)
|
||
from orisys_app_shared.scene3d.package import write_package
|
||
from orisys_studio.app_window import MIN_PANEL_WIDTHS
|
||
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 PySide6.QtCore import Signal
|
||
from PySide6.QtWidgets import QWidget
|
||
from tests.conftest import (
|
||
finish_run as _finish,
|
||
)
|
||
from tests.conftest import (
|
||
running_window as _running_window,
|
||
)
|
||
from tests.conftest import (
|
||
show_window as _show,
|
||
)
|
||
|
||
|
||
@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, monkeypatch):
|
||
"""With no package anywhere, the pane names the directory it looked in.
|
||
|
||
The search is stubbed rather than relied on: the repository may or may not carry
|
||
a compiled package (the application's own config directory is where one belongs),
|
||
and a test that only passes on a machine without one is a test that fails the
|
||
first time somebody installs the feature.
|
||
"""
|
||
import orisys_studio.ui.mesh3d_panel as panel_module
|
||
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel
|
||
|
||
monkeypatch.setattr(panel_module, "find_package", lambda *a, **k: None)
|
||
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(
|
||
write_poc, 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. The
|
||
scene comes from a package written for this test, so the case does not depend on
|
||
whether the repository happens to carry a compiled one.
|
||
"""
|
||
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel
|
||
|
||
panel = Mesh3dPanel()
|
||
try:
|
||
panel.set_sensor("finger", package_override=write_poc())
|
||
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()
|
||
panel.set_enabled(True)
|
||
assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0), (
|
||
"the test package never loaded"
|
||
)
|
||
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_an_analysis_packet_ticks_the_3d_pane(make_window, monkeypatch):
|
||
"""Replay has no run timer: submit must flush the compose so the frame is presented."""
|
||
window = make_window(sensor_type="finger")
|
||
try:
|
||
flushes: list[int] = []
|
||
submitted: list[object] = []
|
||
policies: list[object] = []
|
||
|
||
class _View:
|
||
def set_overlay_policy(self, policy) -> None:
|
||
policies.append(policy)
|
||
|
||
def submit_packet(self, packet) -> None:
|
||
submitted.append(packet)
|
||
|
||
def flush(self) -> None:
|
||
flushes.append(1)
|
||
|
||
# A full Scene3DView needs GL; only the submit/flush seam is under test here.
|
||
window.mesh3d_panel._view = _View() # noqa: SLF001
|
||
monkeypatch.setattr(window, "_connect_analysis_display", lambda panel: True)
|
||
window.show_analysis_packet(object())
|
||
assert submitted, "the analysis packet never reached the 3D producer"
|
||
assert policies, "the overlay policy was not handed to the 3D view"
|
||
assert flushes, "the pane was not flushed after submit"
|
||
finally:
|
||
window.mesh3d_panel._view = None # noqa: SLF001
|
||
window.close()
|
||
|
||
|
||
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(QWidget):
|
||
"""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. It implements the container's
|
||
whole contract (see ``Mesh3dPanel.attach_view``) rather than the one method a
|
||
particular test happens to call, so a change to that contract shows up here. It is
|
||
a real ``QWidget`` because the container puts it in its stack.
|
||
"""
|
||
|
||
diagnosticRaised = Signal(object)
|
||
enabledChanged = Signal(bool)
|
||
|
||
def __init__(self, parent=None) -> None:
|
||
super().__init__(parent)
|
||
self.packets: list = []
|
||
self.coordinator = None
|
||
self.assets = None
|
||
self.scenes_used: list = []
|
||
self.accept_scene = True
|
||
self.available = True
|
||
self.reason = ""
|
||
self.source_kind = "live"
|
||
self.sessions = 0
|
||
self.reference_only = 0
|
||
self.closed = 0
|
||
|
||
# the sink the acquisition thread calls
|
||
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())
|
||
|
||
# the container's contract
|
||
def use_scene(self, assets, source_id: str, expected=None) -> bool:
|
||
"""The seam ``Mesh3dPanel._on_loaded`` adopts a scene through."""
|
||
self.scenes_used.append((assets, source_id, expected))
|
||
if not self.accept_scene:
|
||
return False
|
||
self.assets = assets
|
||
return True
|
||
|
||
def set_enabled(self, enabled: bool) -> None:
|
||
self.enabled_flag = bool(enabled)
|
||
|
||
def set_source_kind(self, kind: str) -> None:
|
||
self.source_kind = str(kind)
|
||
|
||
def begin_session(self) -> None:
|
||
self.sessions += 1
|
||
|
||
def show_reference_only(self) -> None:
|
||
self.reference_only += 1
|
||
|
||
def tick(self) -> None:
|
||
self.ticks = getattr(self, "ticks", 0) + 1
|
||
|
||
def close(self) -> None:
|
||
self.closed += 1
|
||
|
||
def settings(self):
|
||
return DisplaySettings()
|
||
|
||
def restore_settings(self, settings) -> None:
|
||
self.settings_value = settings
|
||
|
||
def describe(self) -> dict:
|
||
return {"scene": {"asset_id": "double"}, "gl": {}, "display": {}, "stream": {}}
|
||
|
||
def render_totals(self) -> dict:
|
||
return {"shown": 0, "dropped_display": 0, "stale": 0, "invalid": 0}
|
||
|
||
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.attach_view(view)
|
||
# 3D is submitted on the GUI deliver path (same tick as 2D), not as an
|
||
# acquisition-thread sink — with or without a pane attached.
|
||
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_live_3d_is_not_an_acquisition_thread_sink(make_window, next_record_root):
|
||
"""3D shares the GUI deliver path with 2D; only the record channel is a sink."""
|
||
import orisys_studio.app_window as app_module
|
||
|
||
window, stuck = _running_window(make_window)
|
||
app_module.recordings_dir = lambda: next_record_root()
|
||
try:
|
||
assert stuck.sinks() == (window._on_packet_recorded,), (
|
||
"live 3D must not register a second acquisition-thread sink"
|
||
)
|
||
assert window.mesh3d_panel.view is None, "a box sensor must not build a view"
|
||
for index in range(10):
|
||
stuck.emit(ui_check._fake_packet(index))
|
||
window._on_run_tick()
|
||
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,
|
||
flow_arrows_enabled=False,
|
||
contact_arrow_enabled=True,
|
||
flow_on_px=0.55,
|
||
flow_off_px=0.35,
|
||
flow_grid_spacing=12,
|
||
max_arrows=250,
|
||
depth_scale_adaptive=True,
|
||
depth_threshold=0.025,
|
||
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.flow_arrows_enabled is False
|
||
assert restored.contact_arrow_enabled is True
|
||
assert restored.flow_on_px == 0.55
|
||
assert restored.flow_off_px == 0.35
|
||
assert restored.flow_grid_spacing == 12
|
||
assert restored.max_arrows == 250
|
||
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"
|
||
assert restored.depth_scale_adaptive is True
|
||
assert restored.depth_threshold == 0.025
|
||
assert scene3d_prefs.settings_from_prefs({}).depth_scale_adaptive is False
|
||
assert scene3d_prefs.settings_from_prefs({"depth_threshold": -1}).depth_threshold == 0.0
|
||
|
||
|
||
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
|
||
|
||
|
||
def test_a_saved_field_choice_survives_and_a_bad_one_is_repaired():
|
||
"""Which quantity colours the surface is a preference; an unknown one is repaired."""
|
||
block = scene3d_prefs.settings_to_prefs(DisplaySettings(field="flow"))
|
||
assert block["field"] == "flow"
|
||
assert scene3d_prefs.settings_from_prefs(block).field == "flow"
|
||
|
||
# The default is the depth response (the operator chose it for the pane), and a
|
||
# hand-edited or older file that names something else gets that default.
|
||
assert DisplaySettings().field == scene3d_prefs.DEFAULT_FIELD == "depth"
|
||
for bad in ({"field": "displacement"}, {"field": 3}, {"field": None}, {}):
|
||
assert scene3d_prefs.settings_from_prefs(bad).field == "depth", bad
|
||
|
||
|
||
@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
|
||
assert settings.field in scene3d_prefs.FIELD_CHOICES
|
||
if settings.color_scale is not None:
|
||
lower, upper = settings.color_scale.lower, settings.color_scale.upper
|
||
# Not a tautology: ``float("nan")`` is in the parametrisation above, and this
|
||
# is the assertion that the repair actually replaced it.
|
||
assert not math.isnan(lower) and not math.isnan(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)))
|
||
# Same arity caveat as below: this branch should never fire, and under PySide6 a
|
||
# builtin slot would fail *silently* if it ever did.
|
||
loader.failed.connect(lambda item, request_id: failed.append(item))
|
||
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 request while one is in flight is not dropped: it is the one the
|
||
# operator asked for *last*, so it waits for the first to finish and then runs.
|
||
loader2 = SceneAssetLoader()
|
||
try:
|
||
loader2._inflight = True
|
||
first = loader2.load(SceneSource(path=path, kind="package", directory=path.parent))
|
||
second = loader2.load(SceneSource(path=path, kind="package", directory=path.parent))
|
||
assert second > first
|
||
assert loader2._executor is None, "a queued request must not open a second worker"
|
||
assert loader2._pending is not None, "the newer request was dropped"
|
||
# The request tuple ends with the release keys; the id is the element before it.
|
||
assert second in loader2._pending
|
||
finally:
|
||
loader2.close()
|
||
|
||
|
||
class _ManualExecutor:
|
||
"""Futures that only run when the test says so, like a loaded thread pool."""
|
||
|
||
def __init__(self) -> None:
|
||
self.pending: list = []
|
||
|
||
def submit(self, function, *args, **kwargs):
|
||
future: Future = Future()
|
||
self.pending.append((future, function, args, kwargs))
|
||
return future
|
||
|
||
def run_next(self) -> None:
|
||
future, function, args, kwargs = self.pending.pop(0)
|
||
if future.set_running_or_notify_cancel():
|
||
try:
|
||
future.set_result(function(*args, **kwargs))
|
||
except BaseException as exc: # noqa: BLE001 - mirrored into the future
|
||
future.set_exception(exc)
|
||
|
||
def shutdown(self, wait: bool = True) -> None: # noqa: ARG002 - executor contract
|
||
self.pending.clear()
|
||
|
||
|
||
def test_the_newest_load_is_the_one_that_runs(qt_app, write_poc):
|
||
"""R7: requests carry an id, and a queue keeps the newest rather than the first."""
|
||
from orisys_studio.scene3d.asset_source import SceneAssetLoader
|
||
|
||
path = write_poc()
|
||
loader = SceneAssetLoader()
|
||
loader._executor = _ManualExecutor()
|
||
answers: list = []
|
||
loader.loaded.connect(
|
||
lambda assets, description, request_id, device: answers.append(request_id)
|
||
)
|
||
loader.failed.connect(lambda item, request_id: answers.append(("failed", request_id)))
|
||
try:
|
||
first = loader.load(SceneSource(path=path, kind="package", directory=path.parent))
|
||
second = loader.load(SceneSource(path=path, kind="package", directory=path.parent))
|
||
assert answers == [], "a request was answered before its worker ran"
|
||
loader._executor.run_next()
|
||
assert answers == [first], "the pending request did not start after the first"
|
||
assert loader._pending is None
|
||
loader._executor.run_next()
|
||
assert answers == [first, second]
|
||
finally:
|
||
loader.close()
|
||
|
||
|
||
def test_the_pane_ignores_a_completion_for_a_superseded_load(qt_app):
|
||
"""The other half: an old answer must not mark the *new* selection as loaded."""
|
||
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel
|
||
|
||
panel = Mesh3dPanel()
|
||
view = _RecordingView()
|
||
panel.attach_view(view)
|
||
assets = synthetic.poc_scene()
|
||
panel._load_request = 2
|
||
panel._on_loaded(assets, "old.oryscene", 1, None)
|
||
assert view.assets is None, "a completion for an older request was adopted"
|
||
panel._on_loaded(assets, "new.oryscene", 2, None)
|
||
assert view.assets is assets
|
||
assert view.scenes_used[-1][1] == "new.oryscene"
|
||
|
||
|
||
def test_a_failed_load_can_be_retried(qt_app, tmp_path, write_poc):
|
||
"""A rejection is not a dead end: the next request still runs."""
|
||
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")
|
||
good = write_poc()
|
||
loader = SceneAssetLoader()
|
||
loader._executor = _ManualExecutor()
|
||
answers: list = []
|
||
loader.loaded.connect(lambda assets, description, request_id, device: answers.append(request_id))
|
||
loader.failed.connect(lambda item, request_id: answers.append(("failed", request_id)))
|
||
try:
|
||
bad = loader.load(SceneSource(path=broken, kind="package", directory=tmp_path))
|
||
loader._executor.run_next()
|
||
assert answers == [("failed", bad)]
|
||
again = loader.load(SceneSource(path=good, kind="package", directory=good.parent))
|
||
loader._executor.run_next()
|
||
assert answers[-1] == again
|
||
finally:
|
||
loader.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 = []
|
||
# ``failed`` carries (diagnostic, request id). PySide6 adapts to a *Python* callable's
|
||
# arity but not to a builtin's, so ``failed.append`` raises TypeError inside the emit,
|
||
# Qt swallows it, and the failure signal looks like it was never sent.
|
||
loader.failed.connect(lambda item, request_id: failed.append(item))
|
||
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 6: a field that is not this frame's
|
||
def _stale_scene_view(qt_app):
|
||
"""A real view with the synthetic scene, or a skip when GL is not buildable."""
|
||
from orisys_studio.scene3d.view import Scene3DView
|
||
|
||
view = Scene3DView()
|
||
if not view.available:
|
||
view.close()
|
||
pytest.skip("no GL binding here")
|
||
if not view.use_scene(synthetic.poc_scene(), source_id="poc-live"):
|
||
view.close()
|
||
pytest.skip("the synthetic scene was refused")
|
||
view.set_enabled(True)
|
||
return view
|
||
|
||
|
||
def test_a_packet_that_cannot_claim_its_field_marks_the_measurement_stale(qt_app):
|
||
"""R2 through the view: the refusal used to be dropped on the floor.
|
||
|
||
``snapshot()`` returning ``None`` left the pane claiming ``live`` about a picture
|
||
nothing was producing. Now the reason travels to the coordinator *and* back out
|
||
to the pane.
|
||
"""
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow_valid = False
|
||
packet.invalid_reason = "compute: RuntimeError: boom"
|
||
view.submit_packet(packet)
|
||
|
||
coordinator = view.coordinator
|
||
assert coordinator is not None
|
||
assert coordinator.state == "stale", "a refused field kept reporting live"
|
||
assert coordinator.last_diagnostic is not None
|
||
assert coordinator.last_diagnostic.code == "field_stale"
|
||
|
||
view.tick()
|
||
assert view.last_diagnostic is not None
|
||
assert view.last_diagnostic.code == "field_stale", (
|
||
"the pane never learned why the measurement stopped"
|
||
)
|
||
assert view.panel is not None
|
||
assert view.panel.state_label.text() not in ("", "scene3d.state_stale"), (
|
||
"the stale state has no operator-facing sentence"
|
||
)
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_an_unstated_field_is_refused_rather_than_assumed(qt_app):
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow_valid = None
|
||
view.submit_packet(packet)
|
||
coordinator = view.coordinator
|
||
assert coordinator is not None
|
||
assert coordinator.state == "stale"
|
||
assert coordinator.last_diagnostic is not None
|
||
assert coordinator.last_diagnostic.code == "field_unverified"
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_a_fresh_field_reaches_the_screen_and_clears_the_stale_state(qt_app):
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
shape = tuple(view.assets.shape_hw)
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
|
||
view.submit_packet(packet)
|
||
assert view._snapshot_diagnostic() is None, "a stated, fresh field was refused"
|
||
coordinator = view.coordinator
|
||
for _ in range(60):
|
||
view.tick()
|
||
if coordinator is not None and coordinator.state == "live":
|
||
break
|
||
time.sleep(0.01)
|
||
assert coordinator is not None and coordinator.state == "live", coordinator.state
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
# ---------------------------------- stage 6: the adapter speaks for the *device* (R3)
|
||
def test_the_adapter_identity_comes_from_the_device_not_the_asset(qt_app):
|
||
"""R3: built from the asset, "the frame matches the scene" is self-certified."""
|
||
from orisys_app_shared.scene3d.contracts import ExpectedGeometry
|
||
from orisys_studio.scene3d.view import Scene3DView
|
||
|
||
view = Scene3DView()
|
||
try:
|
||
if not view.available:
|
||
pytest.skip("no GL binding here")
|
||
assets = replace(synthetic.poc_scene(), source_geometry_id="device-src")
|
||
device = ExpectedGeometry(
|
||
"finger",
|
||
"rev",
|
||
"",
|
||
assets.field_domain_id,
|
||
assets.shape_hw,
|
||
source_geometry_id="device-src",
|
||
)
|
||
assert view.use_scene(assets, source_id="poc", expected=device)
|
||
described = view.describe()["scene"]
|
||
assert described["device_verified"] is True, described
|
||
assert described["device"]["source_geometry_id"] == "device-src"
|
||
assert described["source_geometry_id"] == "device-src"
|
||
|
||
adapter = view._adapter
|
||
assert adapter is not None
|
||
# The device stated no metre-scaled id, so the asset's is used *and* the pane
|
||
# is only verified on the identities the device did state.
|
||
assert adapter.geometry_id == assets.geometry_id
|
||
assert adapter.field_domain_id == assets.field_domain_id
|
||
assert adapter.shape_hw == tuple(assets.shape_hw)
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_a_device_whose_domain_disagrees_refuses_the_scene(qt_app):
|
||
from orisys_app_shared.scene3d.contracts import ExpectedGeometry
|
||
from orisys_studio.scene3d.view import Scene3DView
|
||
|
||
view = Scene3DView()
|
||
try:
|
||
if not view.available:
|
||
pytest.skip("no GL binding here")
|
||
assets = synthetic.poc_scene()
|
||
device = ExpectedGeometry(
|
||
"finger", "", "", "somebody-elses-domain", assets.shape_hw
|
||
)
|
||
assert view.use_scene(assets, source_id="poc", expected=device) is False
|
||
assert view.assets is None
|
||
assert view.last_diagnostic is not None
|
||
assert view.last_diagnostic.code == "field_domain_mismatch"
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_a_scene_without_a_device_description_is_shown_as_unverified(qt_app):
|
||
"""Expected unknown: show the surface, but never claim it was checked."""
|
||
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")
|
||
described = view.describe()["scene"]
|
||
assert described["device_verified"] is False
|
||
assert view.last_diagnostic is not None
|
||
assert view.last_diagnostic.code == "geometry_unverified"
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
# -------------------------------------------- stage 9: producer budget and real costs
|
||
def test_the_producer_does_not_build_a_snapshot_nobody_can_show(qt_app):
|
||
"""R5: the adapter's copy is O(H×W), and a waiting frame makes it wasted work.
|
||
|
||
The mailbox keeps the newest frame, so a second snapshot taken while one is waiting
|
||
is dropped unwatched. Skipping it loses nothing — it is the same frame the display
|
||
would have shown — and keeps the acquisition thread free.
|
||
"""
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
shape = tuple(view.assets.shape_hw)
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
|
||
view.submit_packet(packet)
|
||
assert view.render_totals()["skipped_producer"] == 0
|
||
view.submit_packet(packet)
|
||
assert view.render_totals()["skipped_producer"] == 1, (
|
||
"a snapshot was built while one was already waiting"
|
||
)
|
||
view.tick() # the timer takes it
|
||
view.submit_packet(packet)
|
||
assert view.render_totals()["skipped_producer"] == 1, (
|
||
"the budget must clear as soon as the waiting frame is taken"
|
||
)
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_the_totals_report_a_real_compose_cost_and_how_many_frames_were_drawn(qt_app):
|
||
"""R5: "compose 0.0 ms" was hard-coded, and every repaint counted as a frame."""
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
shape = tuple(view.assets.shape_hw)
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
|
||
view.submit_packet(packet)
|
||
for _ in range(40):
|
||
view.tick()
|
||
if view.state() == "live":
|
||
break
|
||
time.sleep(0.01)
|
||
totals = view.render_totals()
|
||
assert totals["shown"] >= 1
|
||
# The compose really happened, so the frame carries its own measured cost.
|
||
assert totals["compose_ms"] > 0.0, totals
|
||
assert totals["latency_ms"] >= totals["compose_ms"] - 0.5, totals
|
||
stats = view.render_stats()
|
||
assert stats is None or stats.displayed_frames >= 1
|
||
finally:
|
||
view.close()
|
||
def test_a_source_with_no_displacement_cannot_offer_exaggeration(qt_app):
|
||
"""R6: the control must describe an input that exists."""
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
panel = view.panel
|
||
assert panel is not None
|
||
capabilities = view.capabilities()
|
||
assert capabilities is not None
|
||
assert capabilities.displacement_available is False, (
|
||
"the finger adapter publishes no displacement: the capability says so"
|
||
)
|
||
assert not panel.deform_check.isEnabled(), (
|
||
"exaggeration is offered for a source that has no deformation field"
|
||
)
|
||
assert not panel.gain_spin.isEnabled()
|
||
assert t("scene3d.deform_unavailable") in panel.deform_check.toolTip()
|
||
# And the setting itself is forced off, so a persisted gain cannot silently
|
||
# apply to a source with nothing to enlarge.
|
||
panel.deform_check.setChecked(True)
|
||
assert view.settings().deform_enabled is False
|
||
|
||
described = view.describe()
|
||
assert described["capabilities"]["displacement_available"] is False
|
||
assert described["capabilities"]["quantity"] == "flow_magnitude"
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_the_legend_shows_the_range_the_shader_actually_used(qt_app):
|
||
"""R6: the spin boxes are disabled in automatic mode; the legend must not be."""
|
||
from orisys_app_shared.scene3d.contracts import RenderStats
|
||
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
panel = view.panel
|
||
assert panel is not None
|
||
from orisys_app_shared.scene3d.contracts import FieldDescriptor
|
||
|
||
panel.set_descriptor(FieldDescriptor("flow_magnitude", "px", "approximate", False))
|
||
# The real path: the widget emits stats, the view forwards them to the panel.
|
||
view._on_stats(
|
||
RenderStats(
|
||
displayed_frames=9,
|
||
scale_lower=0.25,
|
||
scale_upper=7.5,
|
||
scale_mode="automatic",
|
||
)
|
||
)
|
||
legend = panel.legend_label.text()
|
||
assert "0.250" in legend and "7.500" in legend, legend
|
||
assert t("scene3d.legend_auto") in legend, legend
|
||
assert panel.scale_bar.isVisible() or panel.scale_bar.parent() is not None
|
||
assert "flow_magnitude" not in legend
|
||
described = view.describe()
|
||
assert described["effective_scale"]["lower"] == 0.25
|
||
assert described["effective_scale"]["upper"] == 7.5
|
||
assert described["effective_scale"]["mode"] == "automatic"
|
||
finally:
|
||
view.close()
|
||
|
||
|
||
def test_reset_goes_back_to_the_asset_default_view_not_to_a_zeroed_camera(qt_app):
|
||
"""The first picture and "reset" must be the same view, or they disagree."""
|
||
view = _stale_scene_view(qt_app)
|
||
try:
|
||
assets = view.assets
|
||
assert assets is not None and assets.camera_presets
|
||
preset = assets.camera_presets[0]
|
||
view.apply_preset(preset.name)
|
||
expected = view._camera.matrices((320, 240)).view.copy()
|
||
# Move away from it the way the mouse does.
|
||
view._camera.orbit((30.0, 12.0), (320, 240))
|
||
view._camera.zoom(1.5)
|
||
assert not np.allclose(
|
||
view._camera.matrices((320, 240)).view, expected
|
||
), "the camera never moved"
|
||
view.reset_view()
|
||
assert np.allclose(view._camera.matrices((320, 240)).view, expected), (
|
||
"reset did not return to the asset's own default view"
|
||
)
|
||
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.attach_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_the_replay_view_shows_the_pane_and_says_where_the_field_comes_from(
|
||
make_window, replay_recording, qt_app
|
||
):
|
||
"""History browsing gets the pane, fed by the recompute pass — and says so.
|
||
|
||
A recording stores the raw camera image and the forces, not a per-frame
|
||
deformation field, so the pane in replay is either the surface model (no field) or
|
||
what a recompute pass produces from the video. Those are two different claims and
|
||
the label has to distinguish them.
|
||
"""
|
||
window = _show(make_window(sensor_type="finger"), (1440, 880))
|
||
try:
|
||
assert window.replay_ctl.load_recording(Path(replay_recording))
|
||
assert window._view_preset == "replay"
|
||
pane = window.mesh3d_panel
|
||
assert pane.isVisible(), "the 3D pane is hidden in the replay view"
|
||
assert pane.enabled(), "the pane is visible but not switched on"
|
||
assert window.content_splitter.sizes()[0] >= MIN_PANEL_WIDTHS["mesh3d"], (
|
||
"the replay layout left the pane at zero width"
|
||
)
|
||
view = pane.view
|
||
if view is None or view.panel is None:
|
||
pytest.skip("no 3D view on this platform")
|
||
# Wait for the scene to load, then read what the pane claims.
|
||
assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0), (
|
||
"the scene never loaded: " + str(pane.last_diagnostic)
|
||
)
|
||
label = view.panel.source_label.text()
|
||
assert "回放" in label or "Replay" in label, label
|
||
|
||
# A running pass means the pixels are recomputed, and the label must say so.
|
||
window._recompute_running = True
|
||
window._apply_panel_visibility()
|
||
label = view.panel.source_label.text()
|
||
assert "重算" in label or "Recomputed" in label, label
|
||
|
||
# Stopping the pass must clear the picture, not leave the last frame up.
|
||
window._recompute_running = False
|
||
pane.show_reference_only()
|
||
qt_app.processEvents()
|
||
assert view.coordinator.state in ("no_data", "ready", "disabled")
|
||
assert "没有测量值" in view.panel.legend_label.text() or (
|
||
"No measurement" in view.panel.legend_label.text()
|
||
), view.panel.legend_label.text()
|
||
window._apply_panel_visibility()
|
||
label = view.panel.source_label.text()
|
||
assert "回放" in label or "Replay" in label, label
|
||
finally:
|
||
window.close()
|
||
|
||
|
||
def test_the_live_view_still_says_live(make_window, qt_app):
|
||
window = make_window(sensor_type="finger")
|
||
try:
|
||
window._apply_panel_visibility()
|
||
view = window.mesh3d_panel.view
|
||
if view is None or view.panel is None:
|
||
pytest.skip("no 3D view on this platform")
|
||
label = view.panel.source_label.text()
|
||
assert "实时" in label or "Live" in label, label
|
||
assert view.panel.source_label.toolTip()
|
||
finally:
|
||
window.close()
|
||
|
||
|
||
def test_every_run_gets_a_new_generation(make_window, qt_app):
|
||
"""The window tells the pane when a session starts (see the audit's frozen-pane bug).
|
||
|
||
Without this the pane accepts the first run's frames and refuses every later run's,
|
||
so the picture stays on the first run's last frame forever.
|
||
"""
|
||
import ui_check
|
||
from orisys_app_shared.scene3d.contracts import FrameIdentity
|
||
|
||
window = _show(make_window(sensor_type="finger"), (1440, 880))
|
||
try:
|
||
view = window.mesh3d_panel.view
|
||
if view is None or view.coordinator is None:
|
||
pytest.skip("no 3D view on this platform")
|
||
assert _pump(qt_app, lambda: view.assets is not None, timeout_s=15.0)
|
||
|
||
stuck = ui_check._StuckWorker()
|
||
window._ensure_workers = lambda: None
|
||
window._box_worker = stuck
|
||
window._finger_worker = stuck
|
||
try:
|
||
generations = []
|
||
for _ in range(2):
|
||
window._start_run()
|
||
window._run_timer.stop()
|
||
generations.append(view.coordinator.generation)
|
||
window._stop_run(reason="user_stop")
|
||
assert generations[0] < generations[1], generations
|
||
|
||
# And the frames of the second session are accepted, not refused as stale.
|
||
identity = FrameIdentity(
|
||
source_id=view._source_id,
|
||
session_id=window._session_counter,
|
||
frame_id=1,
|
||
t_rel_s=0.0,
|
||
generation=view.coordinator.generation,
|
||
)
|
||
shape = tuple(view.assets.shape_hw)
|
||
packet = ui_check._fake_packet(1)
|
||
packet.flow = np.full((*shape, 2), 0.5, dtype=np.float32)
|
||
view.submit_packet(packet)
|
||
assert view._snapshot_diagnostic() is None, (
|
||
"the adapter refused the second session's packet"
|
||
)
|
||
assert view.coordinator.stale_frames == 0, (
|
||
"the second session's frames were refused as stale"
|
||
)
|
||
del identity
|
||
finally:
|
||
stuck.release()
|
||
finally:
|
||
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()
|