feat(studio): 三维视图接入产品(阶段 4:界面、生命周期与记录隔离)

- What: `Scene3DView` 装配(widget/相机/渲染器/合成器/协调器/适配器),`asset_source`
  只查找并后台加载已编译的 `.oryscene` 包,`scene3d/prefs` 持久化并**修复**显示设置,
  `ui/mesh3d_panel` 用真实容器替换占位标签(无 GL/无包/包被拒/传感器不匹配各显示一行原因),
  `main.py` 在 `QApplication` 之前选定 GL profile,`app_window` 接线:3D 作为采集线程上的
  第二个 sink、`_on_run_tick` 驱动 `tick()`、视图切换与 `closeEvent` 释放、preferences 落盘。
- Why: 阶段 4 的验收要求"三维开关不影响采集/记录/力曲线",并把阶段 1 留下的
  "开/关三维 CSV 一致"这条补上;同时把"没有 context"从"看起来像在工作"变成可显示的终态。
- Verification: app 侧 test_scene3d_view 28 项、test_scene3d_gl_smoke 14 项(真实 GPU)、
  adapters 17 项、skeleton 4 项、stage1 24 项、coordinator 13 项、ui_smoke 86 项;
  shared 侧 contract 26 / assets 33 / package 19 / live 25;`ui_check.py` 566 项检查 0 失败
  (新增 "3D pane" 段 12 项);`locale_check.py` 302 个字面 key 0 缺失、中英各 377 key 对齐;
  记录隔离:同一 96 帧会话挂/不挂 3D 消费者,`forces.csv` 逐字节相同、有效样本数与记录问题数相同;
  生产者预算:合成器放慢到 50 ms 时 `submit_packet` p99 < 1 ms。
- Not in this change: 示意位移(`displacement_sensor_m`)仍未接入(需 SDK lift 与已确认尺度);
  ARM 板卡适配(按用户要求暂缓);包签名/AES-GCM 与长时性能报告(阶段 5);
  仓库中仍未提供 `.oryscene` 包(发布产物,且其单位与变换必须由人声明)。
This commit is contained in:
ykcheungab
2026-09-14 15:47:32 +08:00
parent ea1027e5a6
commit ac4dcd643f
20 changed files with 2718 additions and 83 deletions
+38 -2
View File
@@ -156,8 +156,13 @@
"plots": "Force Charts",
"cal": "Calibration",
"mesh3d": "3D mesh",
"mesh3d_empty": "Finger 3D view placeholder —\nOpen3D mesh coming later",
"mesh3d_empty_tip": "3D mesh preview for finger sensors"
"mesh3d_empty": "3D surface view",
"mesh3d_empty_tip": "3D surface view for finger sensors (coloured by the live deformation field)",
"mesh3d_no_scene": "No 3D scene for this sensor type",
"mesh3d_no_asset": "No compiled 3D scene installed.\nA scene package (*.oryscene) is looked for in:\n{directory}\nBuild one with tools/scene_assets/compile_scene.py — the names of the files it needs are in its --help. Only the 3D view is affected; measuring, recording and the charts work as usual.",
"mesh3d_loading": "Loading 3D scene {name}…",
"mesh3d_rejected": "The 3D scene was refused: {message}\nOnly the 3D view is affected; measuring, recording and the charts work as usual.",
"mesh3d_unavailable_gl_reason": "3D is not available on this machine: {reason}\nMeasuring, recording and the charts work as usual."
},
"chart": {
"fn": "FN",
@@ -367,5 +372,36 @@
"config_missing": "Sensor config not found for {type}",
"units_raw_fallback": "Calibration unavailable — showing raw units",
"reveal_failed": "Could not reveal config: {error}"
},
"scene3d": {
"enable": "Show the 3D view",
"preset": "View",
"housing": "Housing",
"deform": "Exaggerate",
"deform_tip": "Display-only magnification of the measured deformation. It changes the picture and nothing else: the recorded data, the calibration and the camera are not touched.",
"scale": "Colour scale",
"scale_range": "Range",
"scale_range_tip": "Bounds of the colour scale in fixed mode. In automatic mode the range comes from the valid samples of the frame on screen, so a quantity whose size differs by orders of magnitude is still readable.",
"scale_fixed": "Fixed",
"scale_auto": "Automatic",
"max_fps": "Display rate",
"max_fps_tip": "How many frames per second the 3D view draws. A display budget only: the camera and the recording are unaffected, and a dropped frame here is never a dropped sample.",
"reset_view": "Reset view",
"legend": "{quantity} [{unit}] · {quality}",
"legend_none": "No measurement on screen — the surface model is shown uncoloured",
"quality_calibrated": "calibrated",
"quality_approximate": "relative deformation, illustration",
"quality_display_only": "display only",
"state_disabled": "3D view is off",
"state_loading": "Loading the scene…",
"state_ready": "Ready — waiting for the first frame",
"state_live": "Live",
"state_no_data": "No valid measurement in this frame — showing the surface model",
"state_suspended": "Suspended (the view is hidden)",
"state_unavailable": "3D is unavailable",
"severity_info": "Note",
"severity_warning": "Warning",
"severity_error": "Error",
"stats": "shown {displayed} · dropped {dropped} · compose {compose} ms · paint {paint} ms"
}
}
+38 -2
View File
@@ -156,8 +156,13 @@
"plots": "力曲线图",
"cal": "标定",
"mesh3d": "三维网格",
"mesh3d_empty": "手指三维视图占位 —\nOpen3D 网格稍后加入",
"mesh3d_empty_tip": "手指传感器的三维网格预览"
"mesh3d_empty": "三维表面视图",
"mesh3d_empty_tip": "手指传感器的三维表面视图(按实时形变场着色)",
"mesh3d_no_scene": "此传感器类型没有三维场景",
"mesh3d_no_asset": "本机未安装已编译的三维场景。\n场景包(*.oryscene)的查找位置:\n{directory}\n可用 tools/scene_assets/compile_scene.py 生成,它需要哪些文件见 --help。只影响三维视图:采集、记录与力曲线照常工作。",
"mesh3d_loading": "正在加载三维场景 {name}…",
"mesh3d_rejected": "三维场景被拒绝:{message}\n只影响三维视图:采集、记录与力曲线照常工作。",
"mesh3d_unavailable_gl_reason": "本机无法使用三维显示:{reason}\n采集、记录与力曲线照常工作。"
},
"chart": {
"fn": "FN",
@@ -367,5 +372,36 @@
"config_missing": "未找到传感器配置: {type}",
"units_raw_fallback": "无标定数据 — 显示原始单位",
"reveal_failed": "无法打开配置位置: {error}"
},
"scene3d": {
"enable": "显示三维视图",
"preset": "观察方向",
"housing": "显示外壳",
"deform": "放大形变",
"deform_tip": "只放大画面上的形变显示。它只改图像:记录的数据、标定与算法相机都不受影响。",
"scale": "色标",
"scale_range": "范围",
"scale_range_tip": "固定模式下色标的上下限。自动模式的范围来自当前帧的有效样本,因此量级差别很大的量也都能看清。",
"scale_fixed": "固定",
"scale_auto": "自动",
"max_fps": "显示帧率",
"max_fps_tip": "三维视图每秒绘制多少帧。这只是显示预算:摄像头与记录不受影响,这里丢的帧永远不会变成丢的样本。",
"reset_view": "复位视角",
"legend": "{quantity} [{unit}] · {quality}",
"legend_none": "屏幕上没有测量值 — 仅显示不着色的表面模型",
"quality_calibrated": "已标定",
"quality_approximate": "相对形变示意",
"quality_display_only": "仅显示",
"state_disabled": "三维视图已关闭",
"state_loading": "正在加载场景…",
"state_ready": "就绪 — 等待第一帧",
"state_live": "实时",
"state_no_data": "本帧没有有效测量 — 显示表面模型",
"state_suspended": "已挂起(视图被隐藏)",
"state_unavailable": "三维不可用",
"severity_info": "提示",
"severity_warning": "警告",
"severity_error": "错误",
"stats": "已显示 {displayed} · 丢弃 {dropped} · 合成 {compose} ms · 绘制 {paint} ms"
}
}
@@ -9,6 +9,14 @@
"show_cal_panel": false,
"show_finger_3d": true,
"show_fps": false,
"gl_profile": "desktop33",
"scene3d_package": "",
"scene3d_display": {
"deform_enabled": false,
"display_gain": 0.0,
"housing_visible": true,
"max_fps": 30
},
"sensor_type": "box",
"config_name": "",
"units": "newton",
+84 -1
View File
@@ -39,6 +39,7 @@ from orisys_app_shared.record import RecordConfig
from orisys_app_shared.record.sample import to_record_sample
from orisys_studio.recording import RecordingCoordinator, record_config_from_prefs
from orisys_studio.replay import ReplayController
from orisys_studio.scene3d.prefs import settings_to_prefs
from orisys_app_shared.i18n import set_locale, t
from orisys_app_shared.logging import report_error
from orisys_app_shared.paths import logo_path, sensor_json_path
@@ -69,7 +70,11 @@ RUNNABLE_TYPES = frozenset({"box", "mini", "finger"})
ARROW_ROTATE_TYPES = frozenset({"box", "mini"})
DEFAULT_SPLITTER_SIZES = (340, 480, 400)
MIN_PANEL_WIDTHS = {
"mesh3d": 240,
# The 3D pane is the only one whose content is a real control panel rather than a
# label or a picture, and its minimum is measured under the application
# stylesheet in both languages: the widest case is English at 241 px
# (``tests/test_scene3d_view.py`` pins it), so 240 would have been 1 px short.
"mesh3d": 260,
"arrow": 300,
"plots": 340,
"cal": 300,
@@ -252,6 +257,9 @@ class StudioMainWindow(QMainWindow):
self.content_splitter.addWidget(self.cal_panel)
self.content_splitter.setChildrenCollapsible(False)
root.addWidget(self.content_splitter, 1)
# The pane's own switch is the same state as 视图 ▸ 三维网格. The window owns
# the state; the checkbox is a second view of it (see `_on_3d_pane_toggled`).
self.mesh3d_panel.enabledChanged.connect(self._on_3d_pane_toggled)
# Playback drives the *chart cursor* (the playhead) instead of a slider
# under the panels: the two views share one layout, and the transport is
@@ -277,6 +285,11 @@ class StudioMainWindow(QMainWindow):
self._wire_sensor_picker()
self._wire_replay()
self._restore_prefs_into_ui()
# The 3D pane's sensor and its persisted display settings, before anything
# asks it to draw: the pane resolves its asset from the sensor, and restoring
# the settings later would make the first frame use the defaults.
self._sync_3d_source()
self._restore_3d_settings()
# Visibility and the nav highlight come from the restored view; the
# stored splitter sizes still win at startup (``apply_sizes=False``).
self._apply_view_preset(self._view_preset, apply_sizes=False, persist=False)
@@ -325,6 +338,11 @@ class StudioMainWindow(QMainWindow):
# Both workers publish into the same record channel.
for worker in (self._box_worker, self._finger_worker):
worker.add_sink(self._on_packet_recorded)
# The 3D pane is a second sink on the same thread. It is registered even
# while the pane is off, because a sink that is added and removed as the
# pane is toggled is a sink whose cost depends on the click order; the
# view's own enabled flag is what makes it do nothing.
worker.add_sink(self._on_packet_for_3d)
# ------------------------------------------------------- experiment record
@property
@@ -1206,6 +1224,11 @@ class StudioMainWindow(QMainWindow):
show_arrow = preset != "calibration"
show_cal = self._show_cal if preset == "custom" else preset == "calibration"
self._show_cal = show_cal
if show_3d:
# Built on first use only: until the pane is actually shown, nothing
# imports GL and no scene is read from disk.
self.mesh3d_panel.ensure_view()
self.mesh3d_panel.set_enabled(show_3d)
self.mesh3d_panel.setVisible(show_3d)
self.arrow_panel.setVisible(show_arrow)
self.force_plots.setVisible(True)
@@ -1915,6 +1938,49 @@ class StudioMainWindow(QMainWindow):
self._apply_panel_visibility()
self._persist_prefs()
# ------------------------------------------------------------------- 3D pane
def _on_3d_pane_toggled(self, enabled: bool) -> None:
"""The pane's own switch. Same state as the View menu entry, one owner.
The early return is what keeps the two controls from driving each other: the
window synchronises the pane after every visibility change, and this handler
is the way back.
"""
if bool(enabled) == bool(self._show_finger_3d):
return
self._show_finger_3d = bool(enabled)
self.prefs["show_finger_3d"] = self._show_finger_3d
self._apply_panel_visibility()
self._persist_prefs()
def _sync_3d_source(self) -> None:
"""Tell the 3D pane which sensor it is looking at. No IO, no scene read."""
self.mesh3d_panel.set_sensor(
self.sensor_picker.sensor_type(),
package_override=self.prefs.get("scene3d_package"),
)
def _restore_3d_settings(self) -> None:
"""Persisted display settings, repaired by ``scene3d.prefs`` on the way in."""
from orisys_studio.scene3d.prefs import settings_from_prefs
self.mesh3d_panel.set_display_settings(
settings_from_prefs(self.prefs.get("scene3d_display"))
)
def _on_packet_for_3d(self, packet) -> None:
"""Runs on the **acquisition thread**.
The pane copies what it needs into a read-only frame and offers it to a
capacity-one mailbox; it never touches the recording, the session store or the
camera. The view is the one that decides it is switched off, so this stays a
reference read and a method call.
"""
view = self.mesh3d_panel.view
if view is None:
return
view.submit_packet(packet)
def _on_show_fps_toggled(self, visible: bool) -> None:
self._show_fps = bool(visible)
self.prefs["show_fps"] = self._show_fps
@@ -1929,6 +1995,9 @@ class StudioMainWindow(QMainWindow):
sensor_type=str(sensor_type),
config_ref=self._resolve_config_ref(),
)
# The pane follows the sensor: a box sensor has no 3D scene, and a finger's
# missing package must not be reported while looking at a box.
self._sync_3d_source()
self._apply_panel_visibility()
self._apply_arrow_rotation_for_sensor()
self._update_run_enabled()
@@ -2210,6 +2279,7 @@ class StudioMainWindow(QMainWindow):
# it is wired even if the worker was installed by something other than
# ``_ensure_workers`` (``add_sink`` is idempotent).
self._active_worker.add_sink(self._on_packet_recorded)
self._active_worker.add_sink(self._on_packet_for_3d)
# The filters for this session, before it opens the device: the chain runs
# in the SDK, so this is the one moment the choice can be handed over.
# The acquisition thread applies it and records what the device reports.
@@ -2326,6 +2396,11 @@ class StudioMainWindow(QMainWindow):
logger.info("%s", message)
def _on_run_tick(self) -> None:
# The 3D pane advances on this timer rather than one of its own: the window's
# refresh rate is then the only one that exists, and hiding or disabling the
# pane cannot leave a timer behind. It starts at most one compose on the
# pane's own worker and takes at most one finished frame.
self.mesh3d_panel.tick()
if self._active_worker is None:
return
# Record bookkeeping happens before anything that can return early: a
@@ -2625,6 +2700,9 @@ class StudioMainWindow(QMainWindow):
self.prefs.setdefault("plot_shear_raw_max", DEFAULT_PLOT_SHEAR_RAW_MAX)
self.prefs["show_cal_panel"] = bool(self._show_cal)
self.prefs["show_finger_3d"] = bool(self._show_finger_3d)
self.prefs["scene3d_display"] = settings_to_prefs(
self.mesh3d_panel.display_settings()
)
self.prefs["show_fps"] = bool(self._show_fps)
self.prefs["settings_expanded"] = self.toolbar.settings_expanded()
self.prefs["view_preset"] = self._view_preset
@@ -2640,10 +2718,15 @@ class StudioMainWindow(QMainWindow):
for worker in (self._box_worker, self._finger_worker):
if worker is not None and hasattr(worker, "remove_sink"):
worker.remove_sink(self._on_packet_recorded)
worker.remove_sink(self._on_packet_for_3d)
# Blocking: on the way out we must not leave a device handle or a
# half-written experiment behind.
self._stop_run(reason="user_stop", wait=True)
self._stop_recording(wait=True)
# The pane releases its GL objects from its own context and stops its worker.
# Last, because the window is still referring to it above; idempotent, because
# closing twice must not raise (Qt can deliver closeEvent more than once).
self.mesh3d_panel.close_view()
self._persist_prefs()
logger.info("Studio shell closed; preferences saved")
super().closeEvent(event)
+36 -1
View File
@@ -21,6 +21,35 @@ from orisys_studio.app_window import StudioMainWindow
logger = logging.getLogger(__name__)
#: The GL profiles the 3D view can be started with. One per process: the surface
#: format is a start-up decision, so it is read from the preferences here rather
#: than switched at runtime (design document §11.1).
GL_PROFILES = ("desktop33", "gles30")
DEFAULT_GL_PROFILE = "desktop33"
def _configure_gl(prefs: dict) -> str:
"""Select the default GL surface format. Must run before ``QApplication``.
Optional and failure-tolerant on purpose: a machine without PyOpenGL, or with a
preference that names a profile this build does not know, still gets the whole
two-dimensional application. The 3D pane reports its own state; a start-up that
refused to run because a picture might not appear would be the worse bug.
"""
profile = str(prefs.get("gl_profile") or DEFAULT_GL_PROFILE)
if profile not in GL_PROFILES:
logger.warning("Unknown gl_profile %r; using %s", profile, DEFAULT_GL_PROFILE)
profile = DEFAULT_GL_PROFILE
try:
from orisys_studio.scene3d.gl_backend import configure_surface_format
configure_surface_format(profile) # type: ignore[arg-type]
except Exception: # noqa: BLE001 - 3D is optional, the window is not
logger.exception("Could not configure the OpenGL surface format")
return profile
logger.info("OpenGL surface format requested: profile=%s", profile)
return profile
def main() -> int:
init_logging("orisys-studio")
@@ -41,6 +70,7 @@ def main() -> int:
language = str(prefs.get("language") or "zh_CN")
theme = str(prefs.get("theme") or "dark")
set_locale(language)
gl_profile = _configure_gl(prefs)
app = QApplication(sys.argv)
app.setApplicationName("Orisys Studio")
@@ -49,7 +79,12 @@ def main() -> int:
window = StudioMainWindow(prefs=prefs)
window.show()
logger.info("Studio shell started (theme=%s, language=%s)", theme, language)
logger.info(
"Studio shell started (theme=%s, language=%s, gl_profile=%s)",
theme,
language,
gl_profile,
)
code = int(app.exec_())
# Leave the process immediately.
@@ -74,12 +74,41 @@ def _shape_of(array: Optional[np.ndarray]) -> str:
return "none" if array is None else "x".join(str(int(v)) for v in array.shape[:2])
def _intersect_device_mask(
packet: SdkPacket, mask: np.ndarray, shape_hw: tuple[int, int]
) -> np.ndarray:
"""AND in the device's own validity, when it describes this grid.
A device mask of the wrong shape is ignored rather than resized: a mask from
another grid says nothing about these pixels, and stretching it would mark
samples valid that the device never claimed.
"""
explicit = getattr(packet, "valid_mask", None)
if explicit is None:
return freeze(mask, dtype=np.bool_)
explicit_array = np.asarray(explicit, dtype=bool)
if tuple(explicit_array.shape) == tuple(shape_hw):
return freeze(np.asarray(mask, dtype=bool) & explicit_array, dtype=np.bool_)
return freeze(mask, dtype=np.bool_)
class FingerFrameAdapter:
"""The finger runtime's field, in the scene's UV domain.
``map_x``/``map_y`` are the unroll map's source coordinates. When they are
supplied the field is remapped through them; when they are not, the field is
used directly and only if its shape already *is* the domain.
**The live flow is already unrolled, measured rather than assumed.** The
pipeline computes optical flow on the frame the unroll stage produced
(``FrameUnwarper`` output, ``camera.resolution_stitch`` = 280x300), and
``Sensor.flow`` is that tracker's field; the scene's UV domain is the same
grid (300, 280) with the same valid mask. So the app passes **no** remap here:
remapping an already-unrolled field through ``map_x``/``map_y`` would be a
double remap — plausible-looking and wrong. The parameters exist for a source
that really does deliver source-domain flow; the app must not use them for
this one.
The device's own ``valid_mask`` is intersected in, because outside the UV
island the tracker is looking at black border: those samples are not
measurements, and colouring them would invent a signal at the edge of the
model.
"""
def __init__(
@@ -143,6 +172,9 @@ class FingerFrameAdapter:
domain=f"{self.shape_hw[0]}x{self.shape_hw[1]}",
)
return None
# Inside the UV island is the only place a flow value means anything: the
# tracker is looking at masked black outside it.
mask = _intersect_device_mask(packet, mask, self.shape_hw)
self.last_diagnostic = None
return SurfaceFieldFrame(
identity=identity,
@@ -195,13 +227,10 @@ class BoxFrameAdapter:
array = np.asarray(field, dtype=np.float32)
if array.ndim != 2 or tuple(array.shape) != self.shape_hw:
continue
mask = np.isfinite(array)
explicit = getattr(packet, "valid_mask", None)
if explicit is not None:
explicit_array = np.asarray(explicit, dtype=bool)
if tuple(explicit_array.shape) == self.shape_hw:
# The device's own mask wins: it knows where its surface is.
mask = mask & explicit_array
# The device's own mask wins: it knows where its surface is.
mask = _intersect_device_mask(
packet, np.isfinite(array), self.shape_hw
)
self.last_diagnostic = None
return SurfaceFieldFrame(
identity=identity,
@@ -0,0 +1,181 @@
"""Where the application's 3D scene comes from, and how it fails when it is absent.
Stage 4 in ``docs/architecture/studio-3d-stages.md``.
The runtime loads a **compiled** package. It does not import the Blender export at
start-up, and above all it does not invent the two facts only a person can supply:
the unit scale (how many metres one source unit is) and the sensor-to-source rigid
transform. ``legacy.LegacyFingerImporter`` refuses to guess them, and a 3D pane that
quietly assumed "one source unit is a millimetre" would draw a model that looks
right and is wrong by whatever the real factor is. So the search below looks for
artifacts that already carry those decisions, and its failure message is the
instructions for producing one.
Search order (first hit wins):
1. ``prefs["scene3d_package"]`` — an explicit path from the settings file, which is
also how a developer points the app at a freshly compiled package;
2. ``<config>/sensors/<sensor>/scene/*.oryscene`` — a package shipped beside the
sensor configuration, which is where a release puts it;
3. nothing: the caller reports ``no_asset`` and shows the compile command.
Loading reads a file and validates a package: it must not run on the GUI thread.
:class:`SceneAssetLoader` does that work on one worker thread and hands the result
back through a signal, because a 200 ms freeze while the operator presses Run is
exactly the kind of thing this feature is not allowed to cause.
"""
from __future__ import annotations
import logging
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Optional
from PyQt5.QtCore import QObject, pyqtSignal
from orisys_app_shared.paths import config_dir
from orisys_app_shared.scene3d.contracts import (
Diagnostic,
ExpectedGeometry,
SceneAssets,
diagnostic,
error,
)
logger = logging.getLogger(__name__)
#: The container suffix written by ``tools/scene_assets/compile_scene.py``.
PACKAGE_SUFFIX = ".oryscene"
#: The command that produces a package, quoted in the failure message so the pane
#: tells the operator what to do instead of only what is missing.
COMPILE_HINT = (
"python tools/scene_assets/compile_scene.py "
"--legacy-dir config/sensors/finger/assets/30fps-0703 "
"--map config/sensors/finger/assets/30fps-0703/unroll_map_uv_s0.4.npz "
"--meters-per-source-unit <stated> --output "
"config/sensors/finger/scene/finger-display-1.0.0.oryscene"
)
@dataclass(frozen=True)
class SceneSource:
"""A package the app can load, and where it came from."""
path: Path
kind: str # "package"
directory: Path
def describe(self) -> str:
return f"{self.kind}: {self.path}"
def scene_dir(sensor_type: str, *, root: Optional[Path] = None) -> Path:
"""Where a compiled package for this sensor belongs."""
base = Path(root) if root is not None else config_dir()
return base / "sensors" / str(sensor_type) / "scene"
def find_package(
sensor_type: str,
*,
override: str | Path | None = None,
root: Optional[Path] = None,
) -> Optional[SceneSource]:
"""The package to load, or ``None``. Never raises: absence is a state."""
if override:
path = Path(str(override)).expanduser()
if path.is_file():
return SceneSource(path=path, kind="package", directory=path.parent)
logger.warning("the configured 3D package does not exist: %s", path)
directory = scene_dir(sensor_type, root=root)
if directory.is_dir():
candidates = sorted(p for p in directory.glob(f"*{PACKAGE_SUFFIX}") if p.is_file())
if candidates:
# A single package is the normal case; several means a release shipped
# more than one version, and the name order picks the last one, which is
# also what a person reading the directory would do.
return SceneSource(path=candidates[-1], kind="package", directory=directory)
return None
def no_asset_diagnostic(sensor_type: str, *, override: str | Path | None = None) -> Diagnostic:
"""What to say when there is no package: where it looked, and how to make one.
The message stays in English like every other diagnostic (it is also logged);
the operator-facing wording is the panel's own ``panels.mesh3d_no_asset`` label,
and the compile command travels in ``details`` for support to quote exactly.
"""
searched = scene_dir(sensor_type)
details = {"directory": str(searched), "compile": COMPILE_HINT}
if override:
details["configured"] = str(override)
return diagnostic(
"asset_rejected",
f"no compiled scene package for this sensor: looked in {searched} "
f"for *{PACKAGE_SUFFIX}",
**details,
)
def load_package(
source: SceneSource, expected: Optional[ExpectedGeometry] = None
) -> SceneAssets:
"""Read and validate a package. Runs on a worker thread; raises ``ThreeDError``."""
from orisys_app_shared.scene3d.package import PackagedAssetLoader
if not source.path.is_file():
raise error("asset_rejected", f"the scene package is missing: {source.path}")
return PackagedAssetLoader().load(source.path, expected)
class SceneAssetLoader(QObject):
"""Loads one scene off the GUI thread and reports the result as a signal.
One load is kept: the pane only ever needs the newest, and a second request while
one is running is a request for the same file.
"""
loaded = pyqtSignal(object, str) # SceneAssets, source description
failed = pyqtSignal(object) # Diagnostic
def __init__(self, parent: Optional[QObject] = None) -> None:
super().__init__(parent)
self._executor: Optional[ThreadPoolExecutor] = None
self._inflight = False
def load(self, source: SceneSource, expected: Optional[ExpectedGeometry] = None) -> None:
if self._inflight:
return
if self._executor is None:
self._executor = ThreadPoolExecutor(
max_workers=1, thread_name_prefix="scene3d-assets"
)
self._inflight = True
future = self._executor.submit(load_package, source, expected)
future.add_done_callback(
lambda done: self._finished(done, source)
)
def _finished(self, future, source: SceneSource) -> None:
self._inflight = False
try:
assets = future.result()
except Exception as exc: # noqa: BLE001 - a bad package is a diagnostic
item = getattr(exc, "diagnostic", None) or diagnostic(
"asset_rejected", f"{type(exc).__name__}: {exc}"
)
logger.warning("3D scene %s was rejected: %s", source.path, item.message)
# Emitted from this worker thread; the receiver is a Qt object on the
# GUI thread, so the connection is queued (see the view's `_on_loaded`).
self.failed.emit(item)
return
logger.info("3D scene loaded from %s", source.path)
self.loaded.emit(assets, source.describe())
def close(self) -> None:
if self._executor is not None:
self._executor.shutdown(wait=False)
self._executor = None
@@ -206,6 +206,51 @@ def shader_source(profile: Profile, program: str) -> tuple[str, str]:
return header + _VERTEX_BODY, header + body
def _valid_scalar_range(scalar, valid) -> Optional[tuple[float, float]]:
"""The range of the *valid* samples, or ``None`` when there are none.
Only valid and finite samples count. Including the invalid ones would make the
colour range follow whatever garbage sits outside the UV island, and including
NaN would produce a range of NaN and a picture with no colours at all.
"""
import numpy as np
values = np.asarray(scalar, dtype=np.float64).reshape(-1)
mask = np.asarray(valid, dtype=bool).reshape(-1)
if values.size != mask.size:
return None
usable = values[mask & np.isfinite(values)]
if usable.size == 0:
return None
return float(usable.min()), float(usable.max())
def _scale_bounds(scale, automatic: Optional[tuple[float, float]]) -> tuple[float, float]:
"""The colour range to draw with: the operator's, or the data's own.
``automatic`` mode exists because the quantities differ in magnitude by orders
(a flow magnitude in pixels, a deformation response around 1, a calibrated
displacement in metres): a single fixed range would leave one of them as a flat
colour. An automatic range with no valid samples falls back to a fixed 0..1 so
the shader always receives a usable interval — never NaN, and never a division
by a span of zero.
"""
if scale is None:
fallback = automatic or (0.0, 1.0)
elif str(getattr(scale, "mode", "fixed")) == "automatic" and automatic is not None:
fallback = automatic
else:
fallback = (float(scale.lower), float(scale.upper))
lower, upper = float(fallback[0]), float(fallback[1])
if not (lower == lower) or not (upper == upper): # NaN
return 0.0, 1.0
if upper <= lower:
# A constant field still has to be visible rather than all one clipped colour.
pad = max(abs(lower) * 0.5, 0.5)
return lower - pad, lower + pad
return lower, upper
def lut_rgba(lut_id: str = "sequential") -> np.ndarray:
"""A 256x1 RGBA8 lookup table.
@@ -458,6 +503,9 @@ class GLRenderer:
self._parts: dict[str, _PartBuffers] = {}
self._lut: int = 0
self._lut_id: str = ""
#: Range of the valid samples of the frame currently uploaded, for the
#: automatic colour scale. ``None`` until a frame carries valid data.
self._scalar_range: Optional[tuple[float, float]] = None
self._sensing_part_id: str = ""
self._estimated_gpu_bytes: int = 0
self._initialized = False
@@ -547,6 +595,7 @@ class GLRenderer:
GL.glBufferSubData(GL.GL_ARRAY_BUFFER, 0, rows.nbytes, rows)
GL.glBindBuffer(GL.GL_ARRAY_BUFFER, 0)
self._last_upload_ms = (time.perf_counter() - started) * 1000.0
self._scalar_range = _valid_scalar_range(frame.scalar, frame.valid)
def draw(
self,
@@ -570,10 +619,7 @@ class GLRenderer:
view = np.asarray(camera.view, dtype=np.float32).reshape(4, 4)
projection = np.asarray(camera.projection, dtype=np.float32).reshape(4, 4)
scale = settings.color_scale
lower = float(scale.lower) if scale is not None else 0.0
upper = float(scale.upper) if scale is not None else 1.0
if upper <= lower:
upper = lower + 1.0
lower, upper = _scale_bounds(scale, self._scalar_range)
self._ensure_lut("diverging" if (scale and scale.descriptor.signed) else "sequential")
for part in self._parts.values():
if part.role == "housing" and not settings.housing_visible:
+345 -13
View File
@@ -1,37 +1,73 @@
"""``ThreeDPanel``: the controls for the three-dimensional view.
"""``ThreeDPanel``: the controls and the status line for the three-dimensional view.
Stage 4 in ``docs/architecture/studio-3d-stages.md``.
UI only. It reads and writes :class:`DisplaySettings`, camera presets and
visibility, and it shows what the scene says about itself — quantity, unit,
quality, frame number, "no live data". It does not open assets, read the SDK, or
call GL, and it never presents an ``approximate`` displacement as millimetres.
call GL.
The measurement colour bar is hidden (not left showing the previous range)
whenever the frame's descriptor is ``unavailable`` — the reference model is not a
measurement and must not look like one.
Two rules it exists to enforce in the interface:
* the measurement colour bar is **hidden**, not left showing the previous range,
whenever the frame's descriptor is ``unavailable``: the reference model is not a
measurement and must not look like one;
* an ``approximate`` displacement is labelled "relative deformation, ×N". There is
no code path here that can print a millimetre figure for it.
"""
from __future__ import annotations
from typing import Optional
from PyQt5.QtWidgets import QWidget
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtWidgets import (
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFormLayout,
QFrame,
QHBoxLayout,
QLabel,
QPushButton,
QVBoxLayout,
QWidget,
)
from orisys_app_shared.i18n import t
from orisys_app_shared.scene3d.contracts import (
ColorScale,
Diagnostic,
DisplaySettings,
FieldDescriptor,
RenderStats,
SceneAssets,
SceneState,
)
from orisys_studio.scene3d.coordinator import ThreeDCoordinator
from orisys_studio.scene3d.prefs import FPS_CHOICES
from orisys_studio.scene3d.widget import Sensor3DWidget
_STAGE = "stage 4 (product integration)"
#: Width of one control in the pane. Chosen against
#: ``MIN_PANEL_WIDTHS["mesh3d"]`` (240), not against how it looks on a wide screen:
#: the pane is a splitter child that the operator can drag narrow.
FIELD_WIDTH = 88
SPIN_WIDTH = 68
#: Display limits offered in the panel. ``max_fps`` is a *display* budget; it never
#: reaches the camera or the recording. Defined once, in ``prefs``, because the
#: saved-settings repair has to accept exactly the values this combo offers.
__all__ = ["FPS_CHOICES", "ThreeDPanel"]
class ThreeDPanel(QWidget):
class ThreeDPanel(QFrame):
"""Toggle, presets, reset, housing visibility, gain, colour scale, status."""
enabledChanged = pyqtSignal(bool)
settingsChanged = pyqtSignal(object) # DisplaySettings
resetRequested = pyqtSignal()
presetRequested = pyqtSignal(str)
def __init__(
self,
coordinator: ThreeDCoordinator,
@@ -39,16 +75,312 @@ class ThreeDPanel(QWidget):
parent: Optional[QWidget] = None,
) -> None:
super().__init__(parent)
self.setObjectName("contentPanel")
self._coordinator = coordinator
self._widget = widget
self._assets: Optional[SceneAssets] = None
self._descriptor: Optional[FieldDescriptor] = None
self._state: SceneState = "disabled"
self._syncing = False
def set_diagnostic(self, diagnostic: Optional[Diagnostic]) -> None:
"""Show one diagnostic line; the 2D views stay usable behind it."""
raise NotImplementedError(f"{_STAGE}: show a diagnostic")
root = QVBoxLayout(self)
root.setContentsMargins(8, 6, 8, 8)
root.setSpacing(6)
self.enable_check = QCheckBox()
self.enable_check.toggled.connect(self._on_enabled_toggled)
root.addWidget(self.enable_check)
form = QFormLayout()
form.setContentsMargins(0, 0, 0, 0)
self.preset_label = QLabel()
self.preset_combo = QComboBox()
# The pane is 240 px wide at its narrowest (``MIN_PANEL_WIDTHS["mesh3d"]``),
# so the fields are compact and the panel's own minimum stays below that.
# Without this the layout's minimum is set by two spin boxes side by side,
# the splitter cannot honour the panel width, and the picture is squeezed out.
self.preset_combo.setMaximumWidth(FIELD_WIDTH)
self.preset_combo.currentIndexChanged.connect(self._on_preset_chosen)
form.addRow(self.preset_label, self.preset_combo)
self.housing_check = QCheckBox()
# Checked to match ``DisplaySettings.housing_visible``'s default: the panel's
# starting state and the settings it reports must be the same thing.
self.housing_check.setChecked(True)
self.housing_check.toggled.connect(self._emit_settings)
form.addRow(self.housing_check)
self.deform_check = QCheckBox()
self.deform_check.toggled.connect(self._on_deform_toggled)
self.gain_spin = QDoubleSpinBox()
self.gain_spin.setRange(0.0, 500.0)
self.gain_spin.setSingleStep(1.0)
self.gain_spin.setDecimals(0)
self.gain_spin.setSuffix(" ×")
self.gain_spin.setValue(0.0)
self.gain_spin.setMaximumWidth(SPIN_WIDTH)
self.gain_spin.valueChanged.connect(self._emit_settings)
gain_row = QHBoxLayout()
gain_row.addWidget(self.deform_check)
gain_row.addWidget(self.gain_spin)
gain_row.addStretch(1)
form.addRow(gain_row)
self.scale_label = QLabel()
self.scale_combo = QComboBox()
self.scale_combo.addItem("fixed", "fixed")
self.scale_combo.addItem("automatic", "automatic")
# Automatic by default: the quantities differ by orders of magnitude (pixels,
# a response around 1, metres), so one fixed range would show one of them as
# a single flat colour. The fixed bounds are still there for a comparison
# across frames, which is the reason to prefer them when they are wanted.
self.scale_combo.setCurrentIndex(1)
self.scale_combo.setMaximumWidth(FIELD_WIDTH)
self.scale_combo.currentIndexChanged.connect(self._on_scale_mode_changed)
form.addRow(self.scale_label, self.scale_combo)
# The two bounds get their own row rather than sharing the mode's: three
# controls side by side made this row the widest thing in the pane, and a pane
# whose content demands more than ``MIN_PANEL_WIDTHS["mesh3d"]`` clips its own
# controls when the operator drags the splitter to the declared minimum.
self.range_label = QLabel()
self.lower_spin = QDoubleSpinBox()
self.lower_spin.setRange(-1e6, 1e6)
self.lower_spin.setDecimals(3)
self.lower_spin.setMaximumWidth(SPIN_WIDTH)
self.lower_spin.valueChanged.connect(self._emit_settings)
self.upper_spin = QDoubleSpinBox()
self.upper_spin.setRange(-1e6, 1e6)
self.upper_spin.setDecimals(3)
self.upper_spin.setValue(1.0)
self.upper_spin.setMaximumWidth(SPIN_WIDTH)
self.upper_spin.valueChanged.connect(self._emit_settings)
range_row = QHBoxLayout()
range_row.addWidget(self.lower_spin)
range_row.addWidget(self.upper_spin)
range_row.addStretch(1)
form.addRow(self.range_label, range_row)
self.fps_label = QLabel()
self.fps_combo = QComboBox()
for value in FPS_CHOICES:
self.fps_combo.addItem(f"{value} fps", value)
self.fps_combo.setCurrentIndex(FPS_CHOICES.index(30))
self.fps_combo.setMaximumWidth(FIELD_WIDTH)
self.fps_combo.currentIndexChanged.connect(self._emit_settings)
form.addRow(self.fps_label, self.fps_combo)
root.addLayout(form)
buttons = QHBoxLayout()
self.reset_button = QPushButton()
self.reset_button.clicked.connect(self.resetRequested.emit)
buttons.addWidget(self.reset_button)
buttons.addStretch(1)
root.addLayout(buttons)
self.legend_label = QLabel()
self.legend_label.setObjectName("panelMeta")
self.legend_label.setWordWrap(True)
self.legend_label.setMinimumWidth(1)
root.addWidget(self.legend_label)
self.state_label = QLabel()
self.state_label.setObjectName("hintText")
self.state_label.setWordWrap(True)
self.state_label.setMinimumWidth(1)
root.addWidget(self.state_label)
self.stats_label = QLabel()
self.stats_label.setObjectName("panelMeta")
self.stats_label.setWordWrap(True)
self.stats_label.setMinimumWidth(1)
root.addWidget(self.stats_label)
root.addStretch(1)
self.retranslate()
self.set_scene(None)
self.set_state("disabled")
# ------------------------------------------------------------------ inputs
def set_scene(self, assets: Optional[SceneAssets]) -> None:
"""Fill the preset list from the asset: its default view is the asset's."""
self._assets = assets
self._syncing = True
try:
self.preset_combo.clear()
if assets is not None:
for preset in assets.camera_presets:
self.preset_combo.addItem(preset.name, preset.name)
finally:
self._syncing = False
self.preset_combo.setEnabled(bool(assets is not None and self.preset_combo.count()))
self.reset_button.setEnabled(assets is not None)
def set_descriptor(self, descriptor: Optional[FieldDescriptor]) -> None:
"""The measurement currently on screen — or none, which hides the bar."""
self._descriptor = descriptor
if descriptor is None or descriptor.quantity == "unavailable":
# No measurement: do not leave the previous range on display.
self.legend_label.setText(t("scene3d.legend_none"))
self.legend_label.setVisible(True)
self.scale_combo.setEnabled(False)
self.lower_spin.setEnabled(False)
self.upper_spin.setEnabled(False)
return
self.scale_combo.setEnabled(True)
fixed = self.scale_combo.currentData() == "fixed"
self.lower_spin.setEnabled(fixed)
self.upper_spin.setEnabled(fixed)
self.legend_label.setText(
t(
"scene3d.legend",
quantity=descriptor.quantity,
unit=descriptor.unit,
quality=t(f"scene3d.quality_{descriptor.quality}"),
)
)
self.legend_label.setVisible(True)
def set_state(self, state: SceneState) -> None:
raise NotImplementedError(f"{_STAGE}: reflect the scene state")
self._state = state
self.state_label.setText(t(f"scene3d.state_{state}"))
self.state_label.setVisible(state != "live")
def set_diagnostic(self, diagnostic: Optional[Diagnostic]) -> None:
"""One diagnostic line. The 2D views stay usable behind it."""
if diagnostic is None:
self._legend_reset_note()
return
prefix = t(f"scene3d.severity_{diagnostic.severity}")
self.legend_label.setText(f"{prefix}: {diagnostic.message}")
self.legend_label.setVisible(True)
def set_render_stats(self, stats: RenderStats) -> None:
"""Display-side statistics only: never shown as recording health."""
raise NotImplementedError(f"{_STAGE}: show render statistics")
self.stats_label.setText(
t(
"scene3d.stats",
displayed=stats.displayed_frames,
dropped=stats.superseded_frames,
compose=f"{stats.compose_ms:.1f}",
paint=f"{stats.paint_ms:.1f}",
)
)
def set_enabled(self, enabled: bool) -> None:
self._syncing = True
try:
self.enable_check.setChecked(bool(enabled))
finally:
self._syncing = False
for widget in (
self.preset_combo,
self.housing_check,
self.deform_check,
self.gain_spin,
self.scale_combo,
self.fps_combo,
self.reset_button,
):
widget.setEnabled(bool(enabled))
self.set_descriptor(self._descriptor if enabled else None)
# ----------------------------------------------------------------- outputs
def apply_settings(self, settings: DisplaySettings) -> None:
"""Move the controls to these settings and tell the view about them.
Used when a session's settings are restored from the preferences file. The
values have already been repaired (``scene3d.prefs``); anything that did not
fit a control is repaired again here by Qt's own ranges, which is the only
thing that keeps a hand-edited preference from being applied literally.
"""
self._syncing = True
try:
self.deform_check.setChecked(bool(settings.deform_enabled))
self.gain_spin.setValue(float(settings.display_gain))
self.housing_check.setChecked(bool(settings.housing_visible))
scale = settings.color_scale
if scale is not None:
index = self.scale_combo.findData(str(scale.mode))
if index >= 0:
self.scale_combo.setCurrentIndex(index)
self.lower_spin.setValue(float(scale.lower))
self.upper_spin.setValue(float(scale.upper))
index = self.fps_combo.findData(int(settings.max_fps))
if index >= 0:
self.fps_combo.setCurrentIndex(index)
finally:
self._syncing = False
self.gain_spin.setEnabled(self.deform_check.isChecked())
self._emit_settings()
def settings(self) -> DisplaySettings:
descriptor = self._descriptor or FieldDescriptor(
"unavailable", "1", "display_only", False
)
lower = float(self.lower_spin.value())
upper = float(self.upper_spin.value())
if upper <= lower:
upper = lower + 1.0
return DisplaySettings(
display_gain=float(self.gain_spin.value()),
deform_enabled=self.deform_check.isChecked(),
color_scale=ColorScale(
descriptor=descriptor,
lower=lower,
upper=upper,
lut_id="diverging" if descriptor.signed else "sequential",
mode=str(self.scale_combo.currentData()),
),
housing_visible=self.housing_check.isChecked(),
max_fps=int(self.fps_combo.currentData()),
)
# ------------------------------------------------------------------ signals
def _on_enabled_toggled(self, enabled: bool) -> None:
if self._syncing:
return
self.set_enabled(enabled)
self.enabledChanged.emit(bool(enabled))
def _on_preset_chosen(self, index: int) -> None:
if self._syncing or index < 0:
return
name = self.preset_combo.itemData(index)
if name:
self.presetRequested.emit(str(name))
def _on_deform_toggled(self, enabled: bool) -> None:
self.gain_spin.setEnabled(bool(enabled) and self.enable_check.isChecked())
self._emit_settings()
def _on_scale_mode_changed(self, index: int) -> None:
"""Only a fixed scale has bounds to type in; automatic ones come from data."""
self.set_descriptor(self._descriptor)
self._emit_settings()
def _emit_settings(self) -> None:
if self._syncing:
return
self.settingsChanged.emit(self.settings())
def _legend_reset_note(self) -> None:
self.set_descriptor(self._descriptor)
# --------------------------------------------------------------- i18n / UI
def retranslate(self) -> None:
self.enable_check.setText(t("scene3d.enable"))
self.preset_label.setText(t("scene3d.preset"))
self.housing_check.setText(t("scene3d.housing"))
self.deform_check.setText(t("scene3d.deform"))
self.deform_check.setToolTip(t("scene3d.deform_tip"))
self.scale_label.setText(t("scene3d.scale"))
self.range_label.setText(t("scene3d.scale_range"))
self.range_label.setToolTip(t("scene3d.scale_range_tip"))
self.scale_combo.setItemText(0, t("scene3d.scale_fixed"))
self.scale_combo.setItemText(1, t("scene3d.scale_auto"))
self.fps_label.setText(t("scene3d.max_fps"))
self.fps_label.setToolTip(t("scene3d.max_fps_tip"))
self.reset_button.setText(t("scene3d.reset_view"))
self.state_label.setText(t(f"scene3d.state_{self._state}"))
self.set_descriptor(self._descriptor)
@@ -0,0 +1,137 @@
"""Display settings as a preference value: read, write, and repair.
Stage 4 in ``docs/architecture/studio-3d-stages.md``.
The 3D pane's display settings are the operator's, so they are persisted between
sessions — and a preferences file is a file a person can edit, hand-edit, or carry
over from an older version. So this module's real job is not serialisation, it is
**repair**: a settings block that names a colour range of the wrong type, a frame
rate the build does not offer, or a gain that is not a number has to become a
usable setting rather than a start-up exception. Nothing in the 3D view is worth a
window that will not open.
One rule is not negotiable here: ``deform_enabled`` and ``display_gain`` describe a
*picture*, so restoring them cannot change a measurement. The saved block carries
the descriptor it was captured with, but the restored settings do **not** inherit
it — the colour scale's descriptor comes from the live frame every time, so a stale
file cannot relabel a measurement as another quantity.
"""
from __future__ import annotations
from typing import Any, Mapping, Optional
from orisys_app_shared.scene3d.contracts import (
ColorScale,
DisplaySettings,
FieldDescriptor,
UNAVAILABLE_DESCRIPTOR,
)
#: Frame rates the pane offers. A saved value outside this set is repaired to the
#: default rather than passed to the coordinator as a budget it never promised.
FPS_CHOICES = (15, 30, 60)
DEFAULT_MAX_FPS = 30
MIN_GAIN = 0.0
MAX_GAIN = 500.0
SCALE_MODES = ("fixed", "automatic")
#: The largest magnitude a colour bound may carry. Beyond this the LUT is useless
#: and the number is certainly a typo ("1e308" in a hand-edited file).
MAX_SCALE_BOUND = 1e12
def _number(value: Any, fallback: float) -> float:
try:
number = float(value)
except (TypeError, ValueError):
return fallback
if number != number or number in (float("inf"), float("-inf")): # NaN / inf
return fallback
return number
def _clamp_bound(value: float) -> float:
"""A colour bound inside the interval a spin box and a LUT can actually use."""
return min(max(float(value), -MAX_SCALE_BOUND), MAX_SCALE_BOUND)
def _flag(value: Any, fallback: bool) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
return bool(value)
if isinstance(value, str):
return value.strip().lower() in ("1", "true", "yes", "on")
return fallback
def settings_to_prefs(settings: DisplaySettings) -> dict[str, Any]:
"""A JSON-safe block. The descriptor is recorded, never trusted back."""
scale = settings.color_scale
block: dict[str, Any] = {
"deform_enabled": bool(settings.deform_enabled),
"display_gain": float(settings.display_gain),
"housing_visible": bool(settings.housing_visible),
"max_fps": int(settings.max_fps),
}
if scale is not None:
block["color_scale"] = {
"mode": str(scale.mode),
"lower": float(scale.lower),
"upper": float(scale.upper),
# Recorded for support: if a saved file is in front of us, this says
# what the picture was showing when it was written.
"quantity": str(scale.descriptor.quantity),
"unit": str(scale.descriptor.unit),
"quality": str(scale.descriptor.quality),
}
return block
def settings_from_prefs(
raw: Optional[Mapping[str, Any]],
*,
descriptor: Optional[FieldDescriptor] = None,
) -> DisplaySettings:
"""Rebuild the display settings, repairing anything unusable.
``descriptor`` is the field the colour scale is for; when it is not given, the
restored block carries no colour scale at all (the panel then labels the bar from
the live frame). It is never read back out of the saved file.
"""
if not isinstance(raw, Mapping):
return DisplaySettings(max_fps=DEFAULT_MAX_FPS)
gain = min(max(_number(raw.get("display_gain"), 0.0), MIN_GAIN), MAX_GAIN)
max_fps = int(_number(raw.get("max_fps"), DEFAULT_MAX_FPS))
if max_fps not in FPS_CHOICES:
max_fps = DEFAULT_MAX_FPS
scale: Optional[ColorScale] = None
saved = raw.get("color_scale")
if isinstance(saved, Mapping):
mode = str(saved.get("mode") or "fixed")
if mode not in SCALE_MODES:
mode = "fixed"
lower = _clamp_bound(_number(saved.get("lower"), 0.0))
upper = _clamp_bound(_number(saved.get("upper"), 1.0))
if upper <= lower:
# Zero-width or inverted: widened, and the widened pair is clamped again so
# the repair cannot itself leave the allowed interval (a bound sitting
# exactly at the limit would otherwise come back as limit + 1).
upper = _clamp_bound(lower + 1.0)
if upper <= lower:
lower = _clamp_bound(lower - 1.0)
upper = _clamp_bound(lower + 1.0)
scale = ColorScale(
descriptor=descriptor or UNAVAILABLE_DESCRIPTOR,
lower=lower,
upper=upper,
lut_id=str(saved.get("lut_id") or "sequential"),
mode=mode, # type: ignore[arg-type]
)
return DisplaySettings(
display_gain=gain,
deform_enabled=_flag(raw.get("deform_enabled"), False),
color_scale=scale,
housing_visible=_flag(raw.get("housing_visible"), True),
max_fps=max_fps,
)
@@ -0,0 +1,412 @@
"""``Scene3DView``: everything the application has to assemble for a 3D pane.
Stage 4 in ``docs/architecture/studio-3d-stages.md``. The application's job should
be "put this widget in the window and hand it packets", so the assembly — widget,
camera, renderer, composer, coordinator, timer, adapters — lives here instead of in
``app_window``.
Three properties this class is responsible for, all of them about not breaking the
application:
* **it always constructs.** With no PyOpenGL, no GPU, or no asset it reports itself
unavailable and emits a diagnostic; the two-dimensional views are untouched. A 3D
feature that can stop acquisition from starting is worse than no 3D feature.
* **the producer side is O(1).** ``submit_packet`` runs on the acquisition thread and
only adapts one packet and offers it to a capacity-one mailbox: no IO, no GL, no
Qt, and no way to make the recording wait.
* **it is display-only.** Nothing here writes to the session store, the CSV or the
camera; the same packets reach the record channel through their own sinks.
The adapter for each sensor is chosen by the asset, not by a global: a scene whose
sensing geometry is a finger's is fed the finger's flow, and a scene whose device
description disagrees with the asset is refused with both shapes named.
"""
from __future__ import annotations
import logging
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Optional
from PyQt5.QtCore import pyqtSignal
from PyQt5.QtWidgets import QVBoxLayout, QWidget
from orisys_app_shared.scene3d.composer import SceneComposer
from orisys_app_shared.scene3d.contracts import (
Diagnostic,
DisplaySettings,
ExpectedGeometry,
FieldDescriptor,
FrameIdentity,
RenderStats,
SceneAssets,
SceneState,
diagnostic,
validate_assets,
)
from orisys_studio.scene3d.adapters import BoxFrameAdapter, FingerFrameAdapter
from orisys_studio.scene3d.camera import OrbitCamera
from orisys_studio.scene3d.coordinator import ThreeDCoordinator
from orisys_studio.scene3d.panel import ThreeDPanel
from orisys_studio.scene3d.widget import Sensor3DWidget
logger = logging.getLogger(__name__)
#: Display settings when the panel is not built (no GL). Kept here so ``settings()``
#: has one answer rather than two.
DEFAULT_SETTINGS = DisplaySettings()
#: Smallest useful picture height, in logical pixels. Below this the controls win the
#: layout and the GL widget is allocated a height of zero.
MIN_VIEW_HEIGHT = 120
def gl_available() -> tuple[bool, str]:
"""Whether a Qt OpenGL widget and a GL binding can be built at all.
Imported lazily and guarded: this is what lets the application construct the 3D
pane on a machine where PyOpenGL is not installed.
"""
try:
from PyQt5.QtWidgets import QOpenGLWidget # noqa: F401
except Exception as exc: # noqa: BLE001
return False, f"PyQt5 OpenGL widgets are unavailable: {exc}"
try:
import OpenGL # noqa: F401
except Exception as exc: # noqa: BLE001
return False, f"PyOpenGL is not installed: {exc}"
return True, ""
class Scene3DView(QWidget):
"""One 3D pane: the GL widget on top, its controls underneath."""
stateChanged = pyqtSignal(object) # SceneState
diagnosticRaised = pyqtSignal(object) # Diagnostic
renderStatsChanged = pyqtSignal(object) # RenderStats
descriptorChanged = pyqtSignal(object) # Optional[FieldDescriptor]
enabledChanged = pyqtSignal(bool)
def __init__(self, parent: Optional[QWidget] = None) -> None:
super().__init__(parent)
self._available, self._reason = gl_available()
self._widget: Optional[Sensor3DWidget] = None
self._panel: Optional[ThreeDPanel] = None
self._coordinator: Optional[ThreeDCoordinator] = None
self._camera = OrbitCamera()
self._executor: Optional[ThreadPoolExecutor] = None
self._assets: Optional[SceneAssets] = None
self._expected: Optional[ExpectedGeometry] = None
self._snapshot: Optional[Callable[[Any, FrameIdentity], Any]] = None
self._source_id = ""
self._enabled = False
self._last_diagnostic: Optional[Diagnostic] = None
self._last_descriptor: Optional[FieldDescriptor] = None
self._last_stats: Optional[RenderStats] = None
self._layout = QVBoxLayout(self)
self._layout.setContentsMargins(0, 0, 0, 0)
if self._available:
try:
self._build()
except Exception as exc: # noqa: BLE001 - never let 3D stop the app
logger.exception("Could not build the 3D view")
self._fail(f"the 3D view could not be built: {type(exc).__name__}: {exc}")
# ------------------------------------------------------------------- build
def _build(self) -> None:
from orisys_studio.scene3d.gl_backend import GLRenderer
self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="scene3d")
self._widget = Sensor3DWidget(GLRenderer(), self._camera)
self._coordinator = ThreeDCoordinator(
self._widget, SceneComposer(), self._executor, max_fps=30
)
self._panel = ThreeDPanel(self._coordinator, self._widget)
# The picture must not be squeezed out of existence by the controls above it:
# a GL widget with zero height draws nothing and looks like a broken view.
self._widget.setMinimumHeight(MIN_VIEW_HEIGHT)
self._layout.addWidget(self._widget, 1)
self._layout.addWidget(self._panel, 0)
self._widget.capabilitiesChanged.connect(self._on_capabilities)
self._widget.diagnosticRaised.connect(self._on_diagnostic)
self._widget.renderStatsChanged.connect(self._on_stats)
self._widget.descriptorChanged.connect(self._on_descriptor)
self._panel.enabledChanged.connect(self.set_enabled)
self._panel.settingsChanged.connect(self.set_settings)
self._panel.resetRequested.connect(self.reset_view)
self._panel.presetRequested.connect(self.apply_preset)
def _fail(self, message: str) -> None:
self._available = False
self._reason = message
self._on_diagnostic(diagnostic("gl_unsupported", message))
# ------------------------------------------------------------------ assets
def use_scene(
self,
assets: SceneAssets,
source_id: str,
expected: Optional[ExpectedGeometry] = None,
) -> bool:
"""Adopt a validated scene. Returns whether the view accepted it.
``expected`` is the device's own description. It is optional for the finger
(whose adapter needs nothing but the packet) and required for a box, where it
is the only statement of what the device's field means — an adapter built
without it would be guessing which grid to trust.
"""
if not self._available or self._coordinator is None:
return False
try:
validate_assets(assets, None)
except Exception as exc: # noqa: BLE001
self._on_diagnostic(
getattr(exc, "diagnostic", None)
or diagnostic("asset_rejected", f"{type(exc).__name__}: {exc}")
)
return False
snapshot = self._make_snapshot(assets, expected)
if snapshot is None:
return False
self._assets = assets
self._expected = expected
self._snapshot = snapshot
self._source_id = str(source_id)
self._last_descriptor = None
self._coordinator.activate(assets, self._source_id)
if self._panel is not None:
self._panel.set_scene(assets)
return True
def _make_snapshot(
self, assets: SceneAssets, expected: Optional[ExpectedGeometry]
) -> Optional[Callable[[Any, FrameIdentity], Any]]:
"""Bind the adapter that fits this scene, or refuse with a diagnostic.
The two adapters have different call signatures because they need different
things (contracts' own protocols): the finger adapter takes the frame's cost
from the packet, while the box adapter needs the device description to know
whether the field's grid is the one the scene was built for. Binding the call
here keeps the acquisition path free of both variants.
"""
asset_id = str(getattr(assets, "asset_id", "") or "")
common = {
"geometry_id": assets.geometry_id,
"field_domain_id": assets.field_domain_id,
"shape_hw": tuple(assets.shape_hw),
}
if asset_id.startswith(("box", "mini")):
if expected is None:
self._on_diagnostic(
diagnostic(
"asset_rejected",
"a box scene needs the device description to place its field",
)
)
return None
adapter = BoxFrameAdapter(**common)
return lambda packet, identity: adapter.snapshot(packet, identity, expected)
# The finger's flow arrives already unrolled on the scene's UV domain
# (measured: apps/studio/tests/test_scene3d_adapters.py), so no remap is
# passed here. Remapping it again would index the flow at source pixels.
adapter = FingerFrameAdapter(**common)
return lambda packet, identity: adapter.snapshot(packet, None, identity)
# --------------------------------------------------------------- producer
def submit_packet(self, packet: Any) -> None:
"""Called from the acquisition thread. Must stay O(1) and Qt-free."""
coordinator = self._coordinator
snapshot = self._snapshot
if coordinator is None or snapshot is None or not self._enabled:
return
try:
identity = self.identity_for(packet)
if identity is None:
return
frame = snapshot(packet, identity)
if frame is not None:
coordinator.submit(frame)
except Exception: # noqa: BLE001 - a display sink must never break capture
logger.exception("the 3D sink failed for one packet")
def identity_for(self, packet: Any) -> Optional[FrameIdentity]:
"""The identity a field frame built from this packet would carry."""
if self._coordinator is None or self._assets is None:
return None
return FrameIdentity(
source_id=self._source_id,
session_id=int(getattr(packet, "session_id", 0) or 0),
frame_id=int(getattr(packet, "frame_id", 0) or 0),
t_rel_s=float(getattr(packet, "mono_t", 0.0) or 0.0),
generation=self._coordinator.generation,
)
# ------------------------------------------------------------------- view
def set_enabled(self, enabled: bool) -> None:
"""Turn the pane's work on or off. Idempotent, and never touches the record."""
enabled = bool(enabled) and self._available
self._enabled = enabled
if self._coordinator is not None:
self._coordinator.set_visible(enabled)
if self._panel is not None:
self._panel.set_enabled(enabled)
self.enabledChanged.emit(enabled)
def settings(self) -> DisplaySettings:
if self._panel is not None:
return self._panel.settings()
return DEFAULT_SETTINGS
def restore_settings(self, settings: DisplaySettings) -> None:
"""Adopt persisted display settings.
Two entry points because the panel and the coordinator hold the same values:
the panel's controls have to *show* the restored setting, and the coordinator
has to *use* it. Going through the panel's own signal keeps one path.
"""
self.set_settings(settings)
if self._panel is not None:
self._panel.apply_settings(settings)
def set_settings(self, settings: DisplaySettings) -> None:
if self._coordinator is not None:
self._coordinator.set_settings(settings)
def apply_preset(self, name: str) -> None:
if self._assets is None:
return
for preset in self._assets.camera_presets:
if preset.name == str(name):
try:
self._camera.set_preset(preset)
except ValueError as exc:
logger.warning("ignoring preset %s: %s", name, exc)
return
if self._widget is not None:
self._widget.update()
return
def reset_view(self) -> None:
if self._assets is not None and self._widget is not None:
self._camera.reset(self._assets.bounds_sensor_m)
self._widget.update()
# ------------------------------------------------------------------ state
@property
def available(self) -> bool:
return self._available
@property
def reason(self) -> str:
return self._reason
@property
def enabled(self) -> bool:
return self._enabled
@property
def assets(self) -> Optional[SceneAssets]:
return self._assets
@property
def coordinator(self) -> Optional[ThreeDCoordinator]:
return self._coordinator
@property
def gl_widget(self) -> Optional[Sensor3DWidget]:
return self._widget
@property
def panel(self) -> Optional[ThreeDPanel]:
return self._panel
@property
def last_diagnostic(self) -> Optional[Diagnostic]:
return self._last_diagnostic
@property
def last_descriptor(self) -> Optional[FieldDescriptor]:
return self._last_descriptor
def state(self) -> SceneState:
if not self._available:
return "unavailable"
if self._coordinator is None:
return "disabled"
return self._coordinator.state
def render_stats(self) -> Optional[RenderStats]:
return self._last_stats
def close(self) -> None:
"""Release the GPU resources, stop the worker. Idempotent."""
if self._coordinator is not None:
self._coordinator.close()
if self._widget is not None:
try:
self._widget.release_resources()
except Exception: # noqa: BLE001 - teardown must not raise on the way out
logger.exception("releasing the 3D GPU resources failed")
if self._executor is not None:
self._executor.shutdown(wait=False)
self._executor = None
self._enabled = False
# ------------------------------------------------------------------- ticks
def tick(self) -> None:
"""Advance the display one step. Called by the application's run timer.
Deliberately **not** a timer of its own: the window already has a 16 ms
timer while a session runs, and a component that started threads or timers
by itself would be a component whose cost the window cannot see. Nothing
here reaches the acquisition thread — it starts at most one compose on the
view's own worker and takes at most one finished frame.
"""
if self._coordinator is None or not self._enabled:
return
self._coordinator.tick()
state = self._coordinator.state
if self._panel is not None:
self._panel.set_state(state)
self.stateChanged.emit(state)
def _on_capabilities(self, capabilities: Any) -> None:
if self._panel is not None and getattr(capabilities, "software_renderer", False):
self._panel.set_diagnostic(
diagnostic(
"gl_unsupported",
"the OpenGL context is a software rasteriser: expect a slow view",
)
)
def _on_descriptor(self, descriptor: Any) -> None:
"""The frame on screen changed what it claims to be.
Skipped when it is the same descriptor as last time: this runs once per
displayed frame, and the label and the settings only change when the claim
does. When it does change, the display settings are re-issued so the colour
scale's descriptor — hence the choice between a sequential and a diverging
ramp — follows the live field rather than the one that was on screen when the
operator last touched a control.
"""
if descriptor == self._last_descriptor:
return
self._last_descriptor = descriptor
if self._panel is not None:
self._panel.set_descriptor(descriptor)
self.set_settings(self._panel.settings())
self.descriptorChanged.emit(descriptor)
def _on_diagnostic(self, item: Diagnostic) -> None:
self._last_diagnostic = item
if self._panel is not None:
self._panel.set_diagnostic(item)
self.diagnosticRaised.emit(item)
def _on_stats(self, stats: RenderStats) -> None:
self._last_stats = stats
if self._panel is not None:
self._panel.set_render_stats(stats)
self.renderStatsChanged.emit(stats)
@@ -51,6 +51,10 @@ class Sensor3DWidget(QOpenGLWidget):
capabilitiesChanged = pyqtSignal(object) # GLCapabilities
diagnosticRaised = pyqtSignal(object) # Diagnostic
renderStatsChanged = pyqtSignal(object) # RenderStats, rate-limited
#: What the frame on screen claims to be — quantity, unit, quality. The panel
#: labels the colour bar from this and nothing else, so an unmeasured surface
#: cannot inherit the previous frame's range.
descriptorChanged = pyqtSignal(object) # Optional[FieldDescriptor]
def __init__(
self,
@@ -93,6 +97,7 @@ class Sensor3DWidget(QOpenGLWidget):
"""Keep the newest CPU frame and ask for a repaint. Never requires a context."""
self._frame = frame
self._needs_surface_upload = True
self.descriptorChanged.emit(frame.scalar_descriptor)
self.update()
def set_settings(self, settings: DisplaySettings) -> None:
@@ -224,6 +229,27 @@ class Sensor3DWidget(QOpenGLWidget):
)
)
def paintEvent(self, event: QEvent) -> None: # noqa: N802 - Qt name
"""Qt's own paint, plus the one check Qt never reports as an error.
When the platform cannot create a GL context (the offscreen plugin, a broken
driver, a board without EGL) ``initializeGL`` is simply never called: there is
no exception to catch, and the widget would sit there looking like a working
view that has no data yet. Qt does answer the question, though — ``context()``
stays ``None`` — so the first paint asks it and turns "no context" into the
state the rest of the application already knows how to show.
"""
super().paintEvent(event)
if self._unavailable or self._capabilities is not None:
return
if self.context() is None:
self._fail(
diagnostic(
"gl_unsupported",
"Qt could not create an OpenGL context for the 3D view",
)
)
def resizeGL(self, w: int, h: int) -> None: # pragma: no cover - needs a context
"""Nothing to cache: the projection is rebuilt from the live viewport."""
self._camera.aspect = max(float(w), 1.0) / max(float(h), 1.0)
+251 -16
View File
@@ -1,35 +1,270 @@
"""Placeholder panel for finger 3D mesh view."""
"""``Mesh3dPanel``: the 3D pane's frame, and what it shows when 3D cannot run.
Stage 4 in ``docs/architecture/studio-3d-stages.md``. The splitter slot has to exist
and has to be safe to show in every configuration, so this class answers one
question — *is there a picture, and if not, why not* — and never raises.
The pane degrades in one direction only. No PyOpenGL, no GL context, no compiled
scene package, a rejected package: each one replaces the picture with a line of text
in the operator's language, and the two-dimensional views, the charts, the recording
and the camera keep running exactly as before. There is no path here that can stop
acquisition, and no path that reports a 3D problem as a recording problem.
"""
from __future__ import annotations
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import QFrame, QLabel, QVBoxLayout
from pathlib import Path
from typing import Optional
from PyQt5.QtCore import Qt, pyqtSignal
from PyQt5.QtWidgets import QFrame, QLabel, QStackedLayout, QVBoxLayout, QWidget
from orisys_app_shared.i18n import t
from orisys_app_shared.scene3d.contracts import Diagnostic, ExpectedGeometry
from orisys_studio.scene3d.asset_source import (
SceneAssetLoader,
SceneSource,
find_package,
no_asset_diagnostic,
)
#: Shown when the pane is on but this sensor has no 3D scene at all (box, mini).
_NO_SCENE = "no_scene"
#: Shown when the sensor could have a scene but no package is installed.
_NO_ASSET = "no_asset"
#: Shown when GL cannot run here.
_UNAVAILABLE = "unavailable"
#: Shown while a package is being read (off the GUI thread).
_LOADING = "loading"
#: Shown when a package exists but was refused.
_REJECTED = "rejected"
#: Diagnostics that mean "this view cannot draw here", as opposed to "this one frame
#: was refused". The first kind replaces the picture; the second must not.
_TERMINAL_CODES = frozenset({"gl_unsupported", "shader_compile_failed"})
#: One locale key per state, so ``retranslate`` can rebuild whatever is on screen.
_MESSAGE_KEYS = {
_NO_SCENE: "panels.mesh3d_no_scene",
_NO_ASSET: "panels.mesh3d_no_asset",
_UNAVAILABLE: "panels.mesh3d_unavailable_gl_reason",
_LOADING: "panels.mesh3d_loading",
_REJECTED: "panels.mesh3d_rejected",
}
class Mesh3dPanel(QFrame):
def __init__(self, parent=None) -> None:
"""Hosts the 3D view, or says why there is none."""
#: Emitted when the operator turns the pane on. The window wires the packet sink
#: from this, so no packet is adapted while the pane is off.
enabledChanged = pyqtSignal(bool)
def __init__(self, parent: Optional[QWidget] = None) -> None:
super().__init__(parent)
self.setObjectName("contentPanel")
layout = QVBoxLayout(self)
self._view: Optional[QWidget] = None
self._loader: Optional[SceneAssetLoader] = None
self._source: Optional[SceneSource] = None
self._sensor_type = ""
self._expected: Optional[ExpectedGeometry] = None
self._enabled = False
self._loaded = False
self._reported: Optional[Diagnostic] = None
self._pending_settings = None
self._message_kind = _UNAVAILABLE
self._message_args: dict[str, str] = {"reason": ""}
header = QVBoxLayout()
header.setSpacing(2)
root = QVBoxLayout(self)
root.setContentsMargins(0, 0, 0, 0)
self.title = QLabel()
self.title.setObjectName("panelTitle")
header.addWidget(self.title)
layout.addLayout(header)
root.addWidget(self.title)
self.body = QLabel()
self.body.setObjectName("hintText")
self.body.setAlignment(Qt.AlignCenter)
self.body.setWordWrap(True)
layout.addWidget(self.body, 1)
self._stack = QStackedLayout()
root.addLayout(self._stack, 1)
self.message = QLabel()
self.message.setObjectName("hintText")
self.message.setAlignment(Qt.AlignCenter)
self.message.setWordWrap(True)
self._stack.addWidget(self.message)
self.retranslate()
# ------------------------------------------------------------------ scene
def set_sensor(
self,
sensor_type: str,
*,
expected: Optional[ExpectedGeometry] = None,
package_override: str | Path | None = None,
) -> None:
"""Tell the pane which sensor it is looking at. Cheap; no IO.
Called when the sensor changes, so a box sensor reports "no 3D scene for this
sensor" rather than the finger's missing-package message.
"""
sensor_type = str(sensor_type)
if sensor_type == self._sensor_type and self._source is not None:
# Same sensor, already resolved: re-resolving would drop a loaded scene
# every time the window re-applied its panel visibility.
self._expected = expected if expected is not None else self._expected
return
self._sensor_type = sensor_type
self._expected = expected
self._loaded = False
if self._sensor_type != "finger":
self._show_message(_NO_SCENE)
return
self._source = find_package(self._sensor_type, override=package_override)
if self._source is None:
self._reported = no_asset_diagnostic(self._sensor_type, override=package_override)
self._show_message(
_NO_ASSET, directory=str(self._reported.details["directory"])
)
return
self._show_message(_LOADING, name=self._source.path.name)
def ensure_view(self) -> Optional[QWidget]:
"""Build the 3D view on first use, so a headless machine never imports GL."""
if self._view is not None or self._sensor_type != "finger":
return self._view
from orisys_studio.scene3d.view import Scene3DView
view = Scene3DView(self)
self._view = view
self._stack.addWidget(view)
view.diagnosticRaised.connect(self._on_diagnostic)
view.enabledChanged.connect(self._on_view_enabled)
if self._pending_settings is not None and hasattr(view, "restore_settings"):
view.restore_settings(self._pending_settings)
if not view.available:
self._show_message(_UNAVAILABLE, reason=view.reason)
return view
self._load_scene()
return view
def _load_scene(self) -> None:
if self._loaded or self._source is None or self._view is None:
return
self._loaded = True
view = self._view
if self._loader is None:
self._loader = SceneAssetLoader(self)
self._loader.loaded.connect(self._on_loaded)
self._loader.failed.connect(self._on_load_failed)
self._loader.load(self._source, self._expected)
# -------------------------------------------------------------- settings
def set_display_settings(self, settings) -> None:
"""Persisted display settings, applied now if the view exists.
They are held until then because the view — and with it the only widget that
can show a colour range or an exaggeration — is built on first use.
"""
self._pending_settings = settings
if self._view is not None and hasattr(self._view, "restore_settings"):
self._view.restore_settings(settings)
def display_settings(self):
"""The settings currently in force, for the preferences file."""
from orisys_studio.scene3d.prefs import settings_from_prefs
if self._view is not None and hasattr(self._view, "settings"):
return self._view.settings()
if self._pending_settings is not None:
return self._pending_settings
return settings_from_prefs(None)
# ------------------------------------------------------------------ events
def _on_loaded(self, assets, description: str) -> None:
view = self._view
if view is None:
return
if not view.use_scene(assets, source_id=description, expected=self._expected):
return
self._stack.setCurrentWidget(view)
# Adopting a scene is not the same as showing one: the pane stays off until
# the operator (or the window, restoring a preference) turns it on.
view.set_enabled(self._enabled)
def _on_load_failed(self, item: Diagnostic) -> None:
self._reported = item
self._show_message(_REJECTED, message=item.message)
def _on_diagnostic(self, item: Diagnostic) -> None:
"""A 3D problem. The pane reports it; nothing else changes state.
Two kinds of diagnostic arrive here and they mean different things. A frame
that was refused (``geometry_mismatch``, ``field_domain_mismatch``) is a
transient: the view keeps drawing the reference model and must stay on screen.
A view that cannot draw at all (``gl_unsupported``, ``shader_compile_failed``)
is terminal: showing an empty GL widget instead would read as "no data yet".
"""
self._reported = item
if item.code in _TERMINAL_CODES:
self._show_message(_UNAVAILABLE, reason=item.message)
def _on_view_enabled(self, enabled: bool) -> None:
self.enabledChanged.emit(bool(enabled))
# ------------------------------------------------------------------ inputs
@property
def view(self) -> Optional[QWidget]:
return self._view
@property
def last_diagnostic(self) -> Optional[Diagnostic]:
return self._reported
@property
def showing_message(self) -> bool:
return self._stack.currentWidget() is self.message
def enabled(self) -> bool:
return bool(self._enabled)
def set_enabled(self, enabled: bool) -> None:
self._enabled = bool(enabled)
if self._view is not None:
self._view.set_enabled(self._enabled)
def is_available(self) -> bool:
"""Whether a 3D picture is possible here at all (GL and an asset)."""
return self._view is not None and bool(getattr(self._view, "available", False))
def scene_path(self) -> Optional[Path]:
return None if self._source is None else self._source.path
def tick(self) -> None:
"""Advance the pane. Called by the window's run timer, never by itself."""
if self._view is not None and hasattr(self._view, "tick"):
self._view.tick()
def close_view(self) -> None:
"""Release GL and the loader. Idempotent, and safe to call twice."""
if self._view is not None and hasattr(self._view, "close"):
self._view.close()
if self._loader is not None:
self._loader.close()
self._loader = None
# --------------------------------------------------------------- i18n / UI
def _show_message(self, kind: str, **args: str) -> None:
"""Show one state line. Every state is a sentence, never a blank pane."""
self._message_kind = str(kind)
self._message_args = {str(key): str(value) for key, value in args.items()}
self._render_message()
def _render_message(self) -> None:
text = t(_MESSAGE_KEYS[self._message_kind], **self._message_args)
self.message.setText(text)
self.message.setToolTip(text)
self.message.setProperty("mesh3dState", self._message_kind)
self._stack.setCurrentWidget(self.message)
def retranslate(self) -> None:
self.title.setText(t("panels.mesh3d"))
self.body.setText(t("panels.mesh3d_empty"))
self.body.setToolTip(t("panels.mesh3d_empty_tip"))
self.setToolTip(t("panels.mesh3d_empty_tip"))
self._render_message()
+107 -2
View File
@@ -158,11 +158,116 @@ def test_the_finger_adapter_can_be_built_from_a_domain(scene):
assert domain.source_hw == (288, 512), domain.source_hw
def test_the_shipped_map_cannot_be_applied_to_the_shipped_flow():
"""Why the app must not remap the finger's flow: the map is domain→*source*.
The unroll map is the same size as the UV domain, but its values are pixel
coordinates in the 512x288 source image — not in the 300x280 domain. The live
flow is already computed on the unrolled frame, so feeding it through this map
would be a double remap: it would index the flow at source-pixel coordinates.
This test pins the reason the app passes no map, so a later "improvement" has to
argue with it.
Note the asymmetry that makes the mistake look plausible: in x the map exceeds
the domain's column count (so it would read out of range), while in y it stays
below the row count (so it would silently read the wrong rows). Half of a
double remap crashes and half of it lies.
"""
map_path = _run_map()
if map_path is None:
pytest.skip("the shipped unroll map is not present")
from orisys_app_shared.scene3d.legacy import load_legacy_domain
domain = load_legacy_domain(map_path)
assert domain.map_x is not None and domain.map_y is not None
rows, columns = domain.shape_hw
source_rows, source_columns = domain.source_hw
assert (rows, columns) == (300, 280) and (source_rows, source_columns) == (288, 512)
# Bounded by the source, not by the domain: that is what makes them source pixels.
assert 0.0 <= float(domain.map_x[domain.map_x >= 0].max()) <= source_columns - 1
assert 0.0 <= float(domain.map_y[domain.map_y >= 0].max()) <= source_rows - 1
assert float(domain.map_x.max()) > columns - 1, (
"x would index past the last column of a 280-wide flow field"
)
assert float(domain.map_y.max()) > rows - 1 or source_rows <= rows, (
"y stays inside the 300-row field: the y half of the mistake is silent"
)
def test_the_live_flow_is_already_the_scene_domain():
"""The measurement behind that decision, taken from the shipping config.
``camera.resolution_stitch`` is what ``Sensor`` sizes its flow buffer from and
what the unroll stage outputs, so the flow grid *is* the UV domain. If a future
config changes that, this test fails and the adapter wiring has to be revisited
rather than the numbers quietly not matching.
"""
import json
from pathlib import Path
config = Path(__file__).resolve().parents[1] / "config" / "sensors" / "finger"
map_path = _run_map()
if not (config / "sensor.json").is_file() or map_path is None:
pytest.skip("the finger config and the unroll map are needed")
camera = json.loads((config / "sensor.json").read_text(encoding="utf-8"))["camera"]
resolution = camera["resolution_stitch"]
# Sensor builds `flow` as (resolution_stitch[1], resolution_stitch[0]).
flow_hw = (int(resolution[1]), int(resolution[0]))
from orisys_app_shared.scene3d.legacy import load_legacy_domain
domain = load_legacy_domain(map_path)
assert flow_hw == tuple(domain.shape_hw) == (300, 280)
def test_the_finger_adapter_does_not_colour_outside_the_uv_island(scene):
"""The device mask is what says where the surface is; outside it there is none."""
adapter = FingerFrameAdapter(
geometry_id=scene.geometry_id,
field_domain_id=scene.field_domain_id,
shape_hw=scene.shape_hw,
)
flow = np.zeros((*scene.shape_hw, 2), dtype=np.float32)
flow[..., 0] = 1.0
mask = np.zeros(scene.shape_hw, dtype=bool)
mask[1, 1] = True
frame = adapter.snapshot(
FakePacket(flow=flow, valid_mask=mask), runtime=None, identity=_identity()
)
assert frame is not None
valid = np.asarray(frame.valid)
assert valid.sum() == 1 and bool(valid[1, 1])
assert np.asarray(frame.scalar)[0, 0] == 1.0, (
"the scalar is kept as measured; only validity says it is not a measurement"
)
def test_a_device_mask_from_another_grid_is_ignored_not_stretched(scene):
adapter = FingerFrameAdapter(
geometry_id=scene.geometry_id,
field_domain_id=scene.field_domain_id,
shape_hw=scene.shape_hw,
)
flow = np.ones((*scene.shape_hw, 2), dtype=np.float32)
wrong = np.zeros((7, 7), dtype=bool)
frame = adapter.snapshot(
FakePacket(flow=flow, valid_mask=wrong), runtime=None, identity=_identity()
)
assert frame is not None
assert np.asarray(frame.valid).all()
def _run_map():
from pathlib import Path
path = Path(r"D:\orisys_ws\orisys-sdk\sensors\finger\assets\30fps-0703\unroll_map_uv_s0.4.npz")
return path if path.is_file() else None
root = Path(__file__).resolve().parents[1] / "config" / "sensors" / "finger" / "assets"
for name in ("30fps-0703", "visualization"):
path = root / name / "unroll_map_uv_s0.4.npz"
if path.is_file():
return path
for path in sorted(root.glob("*/unroll_map_uv*.npz")):
return path
return None
# -------------------------------------------------------------------- box adapter
+181
View File
@@ -377,3 +377,184 @@ def test_the_probe_reports_this_machine_truthfully():
assert context["profile_used"] == "desktop33"
assert not context["software_renderer"], "the dev machine has a real GPU"
assert report["renderer_initialized"] and report["released"]
# ------------------------------------------------- stage 4: the assembled view
class _Packet:
"""The parts of an ``SdkPacket`` the finger adapter and the identity need."""
def __init__(self, frame_id: int, shape_hw: tuple[int, int], value: float = 0.5) -> None:
self.session_id = 1
self.frame_id = int(frame_id)
self.mono_t = float(frame_id) / 30.0
self.flow = np.full((*shape_hw, 2), float(value), dtype=np.float32)
self.raw_frame = np.zeros((4, 4, 3), dtype=np.uint8)
@pytest.fixture(scope="module")
def assembled_view(qt_app):
"""A real ``Scene3DView``: pane, coordinator, composer, widget, real context."""
from orisys_studio.scene3d.gl_backend import configure_surface_format
from orisys_studio.scene3d.view import Scene3DView
configure_surface_format("desktop33")
view = Scene3DView()
if not view.available:
pytest.skip(f"no GL view here: {view.reason}")
assets = synthetic.poc_scene()
if not view.use_scene(assets, source_id="poc-live"):
pytest.skip("the synthetic scene was refused")
view.resize(360, 300)
view.show()
view.set_enabled(True)
renderer = view.gl_widget._renderer
painted = _pump(qt_app, view.gl_widget, lambda: bool(renderer.part_ids), timeout_s=8.0)
if not painted:
view.close()
pytest.skip(f"no usable GL context here: {view.gl_widget.diagnostic()}")
yield view, renderer, assets
view.close()
view.close()
qt_app.processEvents()
def test_the_assembled_view_draws_what_the_acquisition_thread_hands_it(assembled_view, qt_app):
"""The whole stage-4 path: packet in, pixels out, and no recording involved."""
view, renderer, assets = assembled_view
widget = view.gl_widget
for index in range(30):
view.submit_packet(_Packet(index, assets.shape_hw, value=0.4 + index * 0.05))
view.tick()
qt_app.processEvents()
assert _pump(qt_app, widget, lambda: view.state() in ("live", "no_data"), timeout_s=5.0)
assert view.state() in ("live", "no_data"), view.state()
stats = widget._last_stats_at and None # stats are published on a timer
del stats
assert view.last_descriptor is not None, "the pane never learned what it is showing"
assert view.last_descriptor.quantity == "flow_magnitude"
assert view.last_descriptor.quality == "approximate"
drawn = _drawn(_render(widget, qt_app))
assert drawn.size, "the live frame was not drawn"
def test_the_producer_side_never_waits_for_the_gpu(assembled_view, qt_app):
"""The measured budget: the acquisition thread's own cost, with a slow composer.
This is the stage-3 property asserted through the *application's* entry point, so
it includes the adapter (the flow magnitude over the whole grid) and not just the
mailbox. The composer is slowed to 50 ms, which is far below any real frame rate:
frames must drop in the display and nowhere else.
"""
import time
view, _, assets = assembled_view
coordinator = view.coordinator
real_composer = coordinator._composer
class Slow:
def compose(self, *args, **kwargs):
time.sleep(0.05)
return real_composer.compose(*args, **kwargs)
def rest_frame(self, *args, **kwargs):
return real_composer.rest_frame(*args, **kwargs)
coordinator._composer = Slow()
try:
samples: list[float] = []
for index in range(200):
packet = _Packet(index, assets.shape_hw, value=0.5)
started = time.perf_counter()
view.submit_packet(packet)
samples.append((time.perf_counter() - started) * 1000.0)
view.tick()
ordered = sorted(samples)
p99 = ordered[int(len(ordered) * 0.99) - 1]
assert p99 < 1.0, f"submit p99 was {p99:.2f} ms with a 50 ms composer"
assert max(samples) < 5.0, f"submit peaked at {max(samples):.2f} ms"
assert coordinator.superseded_frames > 0, (
"a 50 ms composer produced no display drops: the frame rate limiter or the "
"mailbox is not doing its job"
)
finally:
coordinator._composer = real_composer
# Drain the in-flight compose that the slow composer left behind.
deadline = time.perf_counter() + 1.0
while time.perf_counter() < deadline and coordinator._inflight is not None:
coordinator.tick()
time.sleep(0.01)
def test_an_automatic_colour_scale_follows_the_frame(assembled_view, qt_app):
"""A range taken from the data, and a fixed range that ignores it."""
view, renderer, assets = assembled_view
widget = view.gl_widget
assets_frame = view.assets
assert assets_frame is not None
count = int(assets_frame.sensing_part().positions_sensor_m.shape[0])
def frame_with(values: np.ndarray) -> RenderFrame:
return RenderFrame(
identity=FrameIdentity("poc-live", 1, 1, 0.0, generation=view.coordinator.generation),
asset_hash=assets.asset_hash,
geometry_id=assets.geometry_id,
sensing_part_id=assets_frame.sensing_part().part_id,
positions_sensor_m=assets_frame.sensing_part().positions_sensor_m,
normals_sensor=assets_frame.sensing_part().normals_sensor,
scalar=freeze(np.asarray(values, dtype=np.float32), dtype=np.float32),
valid=freeze(np.ones(count, dtype=np.bool_), dtype=np.bool_),
scalar_descriptor=FieldDescriptor("flow_magnitude", "px", "approximate", False),
)
descriptor = FieldDescriptor("flow_magnitude", "px", "approximate", False)
auto = DisplaySettings(
housing_visible=False,
color_scale=ColorScale(
descriptor=descriptor, lower=0.0, upper=1.0, lut_id="sequential", mode="automatic"
),
)
fixed_wide = DisplaySettings(
housing_visible=False,
color_scale=ColorScale(
descriptor=descriptor, lower=0.0, upper=1000.0, lut_id="sequential", mode="fixed"
),
)
ramp = np.linspace(3.0, 7.0, count)
widget.set_settings(fixed_wide)
widget.present(frame_with(ramp))
assert renderer._scalar_range is not None and renderer._scalar_range[0] > 2.0
flat = len(np.unique(_drawn(_render(widget, qt_app)), axis=0))
widget.set_settings(auto)
widget.present(frame_with(ramp))
spread = len(np.unique(_drawn(_render(widget, qt_app)), axis=0))
assert spread > flat, (
"an automatic range did not use the frame's own values "
f"(distinct colours: automatic {spread}, fixed 0..1000 {flat})"
)
def test_the_pane_releases_its_context_and_survives_a_second_close(qt_app):
"""`closeEvent` calls this; a second close must be a no-op, not a crash."""
from orisys_studio.scene3d.gl_backend import configure_surface_format
from orisys_studio.scene3d.view import Scene3DView
configure_surface_format("desktop33")
view = Scene3DView()
if not view.available:
pytest.skip("no GL view here")
if not view.use_scene(synthetic.poc_scene(), source_id="poc-close"):
pytest.skip("the synthetic scene was refused")
view.resize(200, 160)
view.show()
view.set_enabled(True)
renderer = view.gl_widget._renderer
if not _pump(qt_app, view.gl_widget, lambda: bool(renderer.part_ids), timeout_s=8.0):
view.close()
pytest.skip("no usable GL context here")
view.close()
assert not renderer.is_initialized
assert not renderer.part_ids, "closing the pane left GPU buffers behind"
view.close()
view.close()
qt_app.processEvents()
+48 -17
View File
@@ -88,19 +88,25 @@ def test_the_widget_holds_cpu_state_and_touches_no_context(qt_app):
widget.deleteLater()
def test_the_remaining_stub_says_which_stage_fills_it_in(qt_app):
"""What is implemented is exercised elsewhere; what is not says so.
def test_no_scene3d_entry_point_is_still_a_stub(qt_app):
"""The stage-0 floor, kept as a gate: nothing left says "not implemented".
Stages 1 to 3 landed the camera, the dispatcher, the renderer, the sampler, the
composer, the schedule and the adapters. ``ThreeDPanel`` is the one piece left,
and it must still fail loudly with the stage that owns it rather than quietly
doing nothing.
Every stage of the plan replaces stubs with the real thing, and a stub that
survives its stage is worse than a missing feature: it looks implemented. This
fails if a ``raise NotImplementedError`` comes back into either scene3d package,
and it exercises the last entry points that used to be stubs (stage 4's panel)
so "implemented" is also checked by calling them.
"""
from pathlib import Path
import orisys_studio.scene3d
from orisys_app_shared.scene3d.contracts import (
CameraPreset,
DisplaySettings,
FieldDescriptor,
FrameIdentity,
RenderStats,
diagnostic,
)
from orisys_studio.scene3d.camera import OrbitCamera
from orisys_studio.scene3d.coordinator import ThreeDCoordinator
@@ -108,9 +114,20 @@ def test_the_remaining_stub_says_which_stage_fills_it_in(qt_app):
from orisys_studio.scene3d.panel import ThreeDPanel
from orisys_studio.scene3d.widget import Sensor3DWidget
roots = [Path(orisys_studio.scene3d.__file__).resolve().parent]
import orisys_app_shared.scene3d as shared_scene3d
roots.append(Path(shared_scene3d.__file__).resolve().parent)
for root in roots:
for path in sorted(root.rglob("*.py")):
text = path.read_text(encoding="utf-8")
assert "NotImplementedError" not in text or "__init__" in path.name, path
if path.name == "__init__.py":
# The package docstring may mention the word; code may not.
assert "raise NotImplementedError" not in text, path
camera = OrbitCamera()
renderer = GLRenderer()
widget = Sensor3DWidget(renderer, camera)
widget = Sensor3DWidget(GLRenderer(), camera)
coordinator = ThreeDCoordinator(widget, composer=object(), executor=None, max_fps=30) # type: ignore[arg-type]
panel = ThreeDPanel(coordinator, widget)
try:
@@ -126,14 +143,28 @@ def test_the_remaining_stub_says_which_stage_fills_it_in(qt_app):
assert coordinator.superseded_frames == 0
coordinator.close() # idempotent, and safe before any scene is active
for call, stage in (
(lambda: panel.set_state("ready"), "stage 4"),
(lambda: panel.set_render_stats(RenderStats()), "stage 4"),
(lambda: panel.set_diagnostic(None), "stage 4"),
):
with pytest.raises(NotImplementedError) as excinfo:
call()
assert stage in str(excinfo.value), (stage, str(excinfo.value))
del DisplaySettings, FrameIdentity
# Stage 4: the panel is the operator's view of all of the above.
panel.set_state("ready")
assert panel.state_label.text()
panel.set_descriptor(FieldDescriptor("flow_magnitude", "px", "approximate", False))
assert "flow_magnitude" in panel.legend_label.text()
panel.set_descriptor(None)
panel.set_render_stats(RenderStats(displayed_frames=3, superseded_frames=1))
assert "3" in panel.stats_label.text()
panel.set_diagnostic(None)
panel.set_diagnostic(diagnostic("field_unavailable", "no field in this frame"))
assert "no field in this frame" in panel.legend_label.text()
panel.set_scene(None)
panel.set_enabled(False)
settings = panel.settings()
assert settings.max_fps == 30
assert settings.display_gain == 0.0
assert settings.housing_visible is True, (
"the panel's resting state and DisplaySettings' default must agree"
)
assert settings.color_scale is not None
assert settings.color_scale.mode == "automatic"
assert DisplaySettings(max_fps=30).max_fps == settings.max_fps
del FrameIdentity
finally:
widget.deleteLater()
+535
View File
@@ -0,0 +1,535 @@
"""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 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 = []
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 _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()
+91
View File
@@ -2792,6 +2792,97 @@ def main() -> int:
app.processEvents()
stuck.release()
section("3D pane")
from orisys_studio.app_window import MIN_PANEL_WIDTHS as _MIN_PANEL_WIDTHS
pane_window = StudioMainWindow(
prefs={
"language": "zh_CN",
"theme": "dark",
"sensor_type": "finger",
"source_mode": "camera",
"show_finger_3d": True,
}
)
pane_window.resize(1600, 900)
pane_window.show()
app.processEvents()
pane = pane_window.mesh3d_panel
try:
# A box sensor has no 3D scene, and the pane says so rather than showing an
# empty picture. The view already built for the finger sensor is kept (it is
# reused when the sensor goes back), so "no view" is checked on a fresh pane
# and "the message is what is on screen" on this one.
from orisys_studio.ui.mesh3d_panel import Mesh3dPanel as _Mesh3dPanel
fresh = _Mesh3dPanel()
try:
fresh.set_sensor("box")
check(
fresh.view is None,
"a 3D view was built for a sensor type that has no scene",
)
check(fresh.showing_message, "a box sensor's 3D pane is not showing a message")
check(fresh.message.text().strip() != "", "the 3D pane showed an empty line")
finally:
fresh.deleteLater()
pane.set_sensor("box")
check(pane.showing_message, "the 3D pane kept showing a picture for a box sensor")
# A finger with no package names where the package belongs and how to build one.
pane.set_sensor("finger")
diagnostic = pane.last_diagnostic
check(
diagnostic is not None and diagnostic.code == "asset_rejected",
f"a missing 3D scene was not reported as an asset problem: {diagnostic}",
)
if diagnostic is not None:
check(
"compile_scene.py" in diagnostic.details.get("compile", ""),
"the missing-scene report does not say how to build one",
)
check(
str(diagnostic.details.get("directory", "")) in pane.message.text()
or pane.message.text().strip(),
"the pane does not name the directory it looked in",
)
# The pane's controls must fit the width the window reserves for it.
view = pane.ensure_view()
if view is not None:
wanted = view.minimumSizeHint().width()
check(
wanted <= _MIN_PANEL_WIDTHS["mesh3d"],
f"the 3D pane needs {wanted} px, the window reserves "
f"{_MIN_PANEL_WIDTHS['mesh3d']}",
)
# One state, two controls: toggling either one twenty times must not drift,
# rebuild the view, or leave the menu checkbox disagreeing with the pane.
first_view = pane.view
for _ in range(20):
pane_window.act_finger_3d.setChecked(not pane_window.act_finger_3d.isChecked())
check(
pane_window.act_finger_3d.isChecked() == bool(pane_window._show_finger_3d),
"the View menu entry and the window's 3D flag disagree after 20 toggles",
)
check(
pane.enabled() == bool(pane_window._show_finger_3d),
"the 3D pane's own state drifted from the window's after 20 toggles",
)
check(
pane.view is first_view,
"toggling the 3D pane rebuilt the view (a per-toggle allocation)",
)
pane.close_view()
pane.close_view()
check(True, "closing the 3D pane twice is safe")
finally:
pane_window.close()
app.processEvents()
section("replay")
replay_root = APP_ROOT / "outputs" / "_replay_check"
_shutil.rmtree(replay_root, ignore_errors=True)