From 0b309e7edfe10d5fd50b2a524c4d4d484f06c0e2 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 14 Sep 2026 14:32:38 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=89=E7=BB=B4=E6=98=BE=E7=A4=BA=E9=AA=A8?= =?UTF-8?q?=E6=9E=B6=EF=BC=9A=E5=8F=AF=E6=89=A7=E8=A1=8C=E5=90=88=E5=90=8C?= =?UTF-8?q?=20+=20=E6=A8=A1=E5=9D=97=E7=AD=BE=E5=90=8D=20+=20=E5=88=86?= =?UTF-8?q?=E9=98=B6=E6=AE=B5=E8=AE=A1=E5=88=92=20-=20What:=20=E6=8C=89=20?= =?UTF-8?q?docs/architecture/studio-3d-design.md=20=E4=B8=8E=20studio-3d-c?= =?UTF-8?q?ontracts.pyi=20=E6=90=AD=E5=BB=BA=E9=AA=A8=E6=9E=B6=EF=BC=88?= =?UTF-8?q?=E9=98=B6=E6=AE=B5=200=EF=BC=89=E3=80=82shared/src/orisys=5Fapp?= =?UTF-8?q?=5Fshared/scene3d/=EF=BC=9Acontracts.py=20=E6=8A=8A=20.pyi=20?= =?UTF-8?q?=E7=9A=84=E8=A7=84=E6=A0=BC=E5=8F=98=E6=88=90=E5=8F=AF=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E4=BB=A3=E7=A0=81=E2=80=94=E2=80=94=E5=85=A8=E9=83=A8?= =?UTF-8?q?=20DTO/=E6=9E=9A=E4=B8=BE=EF=BC=88FrameIdentity/FieldDescriptor?= =?UTF-8?q?/SurfaceFieldFrame/MeshPart/SurfaceBinding/CameraPreset/Expecte?= =?UTF-8?q?dGeometry/SceneAssets/ColorScale/DisplaySettings/VertexFields/R?= =?UTF-8?q?enderFrame/CameraMatrices/GLCapabilities/RenderStats/CompileOpt?= =?UTF-8?q?ions/CompileReport=EF=BC=89=E3=80=81=E6=A0=87=E5=87=86=E8=AF=8A?= =?UTF-8?q?=E6=96=AD=E7=A0=81=E8=A1=A8=EF=BC=88DIAGNOSTIC=5FSEVERITY?= =?UTF-8?q?=EF=BC=8C=E5=90=AB=20stale=5Fframe=3Dinfo=E3=80=81field=5Funava?= =?UTF-8?q?ilable=3Dwarning=20=E7=AD=89=EF=BC=89=E3=80=81ThreeDError?= =?UTF-8?q?=E3=80=81freeze()=EF=BC=88=E6=8B=A5=E6=9C=89=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=20+=20C=20=E8=BF=9E=E7=BB=AD=20+=20=E5=8F=AA=E8=AF=BB=EF=BC=89?= =?UTF-8?q?=E3=80=81=E4=BA=94=E4=B8=AA=E6=A0=A1=E9=AA=8C=E5=99=A8=EF=BC=88?= =?UTF-8?q?validate=5Fsurface=5Fframe/validate=5Fbinding/validate=5Fmesh?= =?UTF-8?q?=5Fpart/validate=5Fassets/validate=5Fdisplay=5Fsettings/validat?= =?UTF-8?q?e=5Frender=5Fframe=EF=BC=89=E3=80=81LatestFrameMailbox=EF=BC=88?= =?UTF-8?q?=E5=AE=B9=E9=87=8F=201=E3=80=81=E6=9B=BF=E6=8D=A2=E8=AE=A1?= =?UTF-8?q?=E6=95=B0=E3=80=81=E9=94=81=E5=86=85=E5=8F=AA=E6=8D=A2=E5=BC=95?= =?UTF-8?q?=E7=94=A8=EF=BC=89=E3=80=81UNAVAILABLE=5FDESCRIPTOR/rest=5Fdesc?= =?UTF-8?q?riptor()=20=E4=B8=8E=20summarize=5Fparts()=EF=BC=8C=E4=BB=A5?= =?UTF-8?q?=E5=8F=8A=E5=85=A8=E9=83=A8=20Protocol=EF=BC=9B=E8=AF=A5?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=E4=B8=8D=20import=20Qt/OpenGL/SDK=E3=80=82as?= =?UTF-8?q?sets.py/sampling.py/composer.py=20=E4=B8=BA=E7=AD=BE=E5=90=8D?= =?UTF-8?q?=E9=AA=A8=E6=9E=B6=EF=BC=88=E6=9C=AA=E5=AE=9E=E7=8E=B0=E5=A4=84?= =?UTF-8?q?=E6=8A=9B=20NotImplementedError=20=E5=B9=B6=E5=86=99=E6=98=8E?= =?UTF-8?q?=E5=BD=92=E5=B1=9E=E9=98=B6=E6=AE=B5=EF=BC=89=E3=80=82apps/stud?= =?UTF-8?q?io/src/orisys=5Fstudio/scene3d/=EF=BC=9Acamera/adapters/coordin?= =?UTF-8?q?ator/gl=5Fbackend/widget/panel=20=E7=AD=BE=E5=90=8D=E9=AA=A8?= =?UTF-8?q?=E6=9E=B6=EF=BC=8C=E5=85=B6=E4=B8=AD=20configure=5Fsurface=5Ffo?= =?UTF-8?q?rmat()=20=E5=B7=B2=E5=AE=9E=E7=8E=B0=EF=BC=88QApplication=20?= =?UTF-8?q?=E4=B9=8B=E5=89=8D=E8=B0=83=E7=94=A8=EF=BC=8Cdesktop33=3DGL3.3?= =?UTF-8?q?=20Core=20/=20gles30=3DGLES3.0=EF=BC=8C=E6=B7=B1=E5=BA=A6=2024?= =?UTF-8?q?=E3=80=81=E6=97=A0=20MSAA=EF=BC=89=EF=BC=9B=E6=A8=A1=E5=9D=97?= =?UTF-8?q?=E7=BA=A7=E4=B8=80=E5=BE=8B=E4=B8=8D=20import=20GL=20=E7=BB=91?= =?UTF-8?q?=E5=AE=9A=EF=BC=8CPyOpenGL=20=E5=8F=AA=E5=9C=A8=20GLDispatch.bi?= =?UTF-8?q?nd=5Fcurrent=5Fcontext=20=E9=87=8C=E5=8A=A0=E8=BD=BD=E3=80=82to?= =?UTF-8?q?ols/scene=5Fassets/{probe=5Fgl,validate=5Fscene,compile=5Fscene?= =?UTF-8?q?}.py=20=E6=8F=90=E4=BE=9B=E5=85=A5=E5=8F=A3=E4=B8=8E=E5=8F=82?= =?UTF-8?q?=E6=95=B0=EF=BC=8C=E6=9C=AA=E5=AE=9E=E7=8E=B0=E6=97=B6=E8=BE=93?= =?UTF-8?q?=E5=87=BA=20not=5Fimplemented=20=E5=B9=B6=E8=BF=94=E5=9B=9E=202?= =?UTF-8?q?=EF=BC=88=E9=81=BF=E5=85=8D=E8=A2=AB=20CI=20=E5=BD=93=E6=88=90?= =?UTF-8?q?=E9=80=9A=E8=BF=87=EF=BC=89=E3=80=82docs/architecture/studio-3d?= =?UTF-8?q?-stages.md=EF=BC=9A=E9=98=B6=E6=AE=B5=200-5=20=E7=9A=84?= =?UTF-8?q?=E4=BA=A7=E5=87=BA=E3=80=81=E6=AF=8F=E9=98=B6=E6=AE=B5=E7=9A=84?= =?UTF-8?q?=E9=AA=8C=E6=94=B6=E7=9F=A9=E9=98=B5=E3=80=81=E5=B9=B6=E8=A1=8C?= =?UTF-8?q?=E5=BA=A6=E4=B8=8E=E5=9B=9E=E5=BD=92=E9=97=A8=E3=80=82=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=EF=BC=9Ashared/tests/test=5Fscene3d=5Fcontracts.py=20?= =?UTF-8?q?26=20=E9=A1=B9=EF=BC=88freeze=20=E6=8B=A5=E6=9C=89=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E3=80=81=E5=8F=AF=E5=86=99/=E9=9D=9E=E8=BF=9E?= =?UTF-8?q?=E7=BB=AD=E6=95=B0=E7=BB=84=E8=A2=AB=E6=8B=92=E3=80=81scalar=20?= =?UTF-8?q?=E4=B8=8E=20valid=20=E5=BD=A2=E7=8A=B6=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E3=80=81=E4=BD=8D=E7=A7=BB=E7=BC=BA=20quality=20=E8=A2=AB?= =?UTF-8?q?=E6=8B=92=E3=80=81=E9=9D=9E=E6=9C=89=E9=99=90=20scalar=20?= =?UTF-8?q?=E6=A0=87=E4=B8=BA=20valid=20=E5=8F=AA=E6=8A=A5=E8=AD=A6?= =?UTF-8?q?=E4=B8=8D=E9=9D=99=E9=BB=98=E3=80=81=E7=BB=91=E5=AE=9A=E8=B6=8A?= =?UTF-8?q?=E7=95=8C/-1=20=E7=BA=A6=E5=AE=9A/=E6=9D=83=E9=87=8D=E5=92=8C?= =?UTF-8?q?=E4=B8=BA=201=E3=80=81=E4=B8=89=E8=A7=92=E5=BD=A2=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E8=B6=8A=E7=95=8C=E3=80=81=E6=84=9F=E7=9F=A5=E6=B3=95?= =?UTF-8?q?=E7=BA=BF=E5=8D=95=E4=BD=8D=E9=95=BF=E5=BA=A6=E3=80=81=E6=81=B0?= =?UTF-8?q?=E5=A5=BD=E4=B8=80=E4=B8=AA=20sensing=20part=E3=80=81binding=20?= =?UTF-8?q?=E5=BF=85=E9=A1=BB=E5=B1=9E=E4=BA=8E=20sensing=20part=E3=80=81?= =?UTF-8?q?=E8=AE=BE=E5=A4=87=E5=87=A0=E4=BD=95/=E5=9F=9F/=E5=B0=BA?= =?UTF-8?q?=E5=AF=B8=E4=B8=8D=E7=AC=A6=E8=A2=AB=E6=8B=92=E3=80=81=E8=89=B2?= =?UTF-8?q?=E6=A0=87=20lower Optional[SurfaceFieldFrame]: + raise NotImplementedError(f"{_STAGE}: snapshot the finger UV field") + + +class BoxFrameAdapter: + """Box/mini: a scalar field only. + + There is no verified spatial scale for these sensors yet, so no displacement is + invented: a millimetre height field that was never calibrated is worse than no + height field at all. + """ + + def snapshot( + self, packet: SdkPacket, identity: FrameIdentity, domain: ExpectedGeometry + ) -> Optional[SurfaceFieldFrame]: + raise NotImplementedError(f"{_STAGE}: snapshot the box scalar field") diff --git a/apps/studio/src/orisys_studio/scene3d/camera.py b/apps/studio/src/orisys_studio/scene3d/camera.py new file mode 100644 index 0000000..6be49be --- /dev/null +++ b/apps/studio/src/orisys_studio/scene3d/camera.py @@ -0,0 +1,50 @@ +"""The observation camera: orbit, pan, zoom, presets. Pure maths. + +Stage 1 gets a minimal ``fit``/``orbit`` so a static scene can be looked at; the +full interaction (pan, zoom limits, presets, reset) is stage 4 in +``docs/architecture/studio-3d-stages.md``. + +It is deliberately *not* the algorithm camera. It changes ``view`` and +``projection`` only: ``window_to_world`` for the pixel→surface mapping, the +calibration and the force axes are untouched by anything a user does here. The +default view comes from the asset's ``CameraPreset``, so it is versioned with the +geometry rather than with a user's preference file. +""" + +from __future__ import annotations + +from orisys_app_shared.scene3d.contracts import ( + CameraMatrices, + CameraPreset, + F32, +) + +_STAGE = "stage 1 (fit/orbit) and stage 4 (full interaction)" + + +class OrbitCamera: + """A target, a distance and two angles. Everything else is derived.""" + + def fit(self, bounds_sensor_m: F32, aspect: float) -> None: + """Frame the whole scene: set the target, distance, near and far planes.""" + raise NotImplementedError(f"{_STAGE}: fit the camera to the scene bounds") + + def orbit(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: + """Rotate about the target. The model data does not move.""" + raise NotImplementedError(f"{_STAGE}: orbit") + + def pan(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: + """Slide the target in the view plane, scaled by distance.""" + raise NotImplementedError(f"{_STAGE}: pan") + + def zoom(self, wheel_steps: float) -> None: + """Exponential zoom, clamped so the surface cannot be entered or lost.""" + raise NotImplementedError(f"{_STAGE}: zoom") + + def set_preset(self, preset: CameraPreset) -> None: + """Front / side / top / isometric / asset default.""" + raise NotImplementedError(f"{_STAGE}: apply a camera preset") + + def matrices(self, viewport: tuple[int, int]) -> CameraMatrices: + """``p_camera = view @ p_sensor``, ``p_clip = projection @ p_camera``.""" + raise NotImplementedError(f"{_STAGE}: build the view and projection matrices") diff --git a/apps/studio/src/orisys_studio/scene3d/coordinator.py b/apps/studio/src/orisys_studio/scene3d/coordinator.py new file mode 100644 index 0000000..88dfced --- /dev/null +++ b/apps/studio/src/orisys_studio/scene3d/coordinator.py @@ -0,0 +1,89 @@ +"""Frame schedule and state machine for the three-dimensional view. + +Stage 3 (mailboxes, compose worker) and stage 4 (Qt timer, visibility, settings) in +``docs/architecture/studio-3d-stages.md``. + +The whole point of this class is that **the acquisition loop never waits for the +display**. Producers call :meth:`submit`, which validates cheaply and drops the +previous un-consumed frame; a GUI timer calls :meth:`tick`, which picks up the +newest composed frame and repaints. Display drops are counted here and never in +the recording channel, and no CPU task is submitted while one is still running. +""" + +from __future__ import annotations + +from concurrent.futures import Executor +from typing import Optional + +from orisys_app_shared.scene3d.contracts import ( + DisplaySettings, + LatestFrameMailbox, + OfferResult, + RenderFrame, + SceneAssets, + SceneState, + SurfaceFieldFrame, +) +from orisys_studio.scene3d.widget import Sensor3DWidget + +_STAGE = "stage 3 (schedule) and stage 4 (Qt integration)" + + +class ThreeDCoordinator: + """Owns the two mailboxes, the compose worker and the scene generation.""" + + def __init__( + self, + widget: Sensor3DWidget, + composer: object, + executor: Executor, + max_fps: int = 30, + ) -> None: + self._widget = widget + self._composer = composer + self._executor = executor + self._max_fps = int(max_fps) + self._incoming: LatestFrameMailbox[SurfaceFieldFrame] = LatestFrameMailbox() + self._composed: LatestFrameMailbox[RenderFrame] = LatestFrameMailbox() + self._state: SceneState = "disabled" + self._generation = 0 + self._assets: Optional[SceneAssets] = None + self._visible = False + + def activate(self, assets: SceneAssets, source_id: str) -> int: + """Adopt a scene and a data source; returns the new generation.""" + raise NotImplementedError(f"{_STAGE}: activate a scene") + + def submit(self, frame: SurfaceFieldFrame) -> OfferResult: + """Offer a field frame from a producer thread. Never blocks on the GUI.""" + raise NotImplementedError(f"{_STAGE}: accept or reject an incoming frame") + + def tick(self) -> None: + """GUI timer: take the newest composed frame, present it, schedule work.""" + raise NotImplementedError(f"{_STAGE}: pump the mailboxes") + + def set_visible(self, visible: bool) -> None: + """Hiding suspends composing, uploading and rendering — not acquisition.""" + raise NotImplementedError(f"{_STAGE}: suspend or resume the view") + + def set_settings(self, settings: DisplaySettings) -> None: + raise NotImplementedError(f"{_STAGE}: apply display settings") + + def close(self) -> None: + """Cancel work, stop the timer, clear both mailboxes. Idempotent.""" + raise NotImplementedError(f"{_STAGE}: shut the view down") + + @property + def state(self) -> SceneState: + return self._state + + @property + def generation(self) -> int: + return self._generation + + @property + def superseded_frames(self) -> int: + """Frames the display never showed. Not a recording drop.""" + return ( + self._incoming.superseded_frames + self._composed.superseded_frames + ) diff --git a/apps/studio/src/orisys_studio/scene3d/gl_backend.py b/apps/studio/src/orisys_studio/scene3d/gl_backend.py new file mode 100644 index 0000000..6db58c5 --- /dev/null +++ b/apps/studio/src/orisys_studio/scene3d/gl_backend.py @@ -0,0 +1,99 @@ +"""OpenGL: surface format, capability probing, shaders and GPU resources. + +Stage 1 (platform PoC) in ``docs/architecture/studio-3d-stages.md``. + +Import discipline, and why it is not negotiable: + +* :func:`configure_surface_format` runs **before** ``QApplication`` exists and + only touches ``QSurfaceFormat``; +* the PyOpenGL import happens inside :meth:`GLDispatch.bind_current_context`, + after Qt has made a real context current, so the binding is chosen for the + platform Qt actually used (GLX or EGL) instead of being guessed; +* ``GLRenderer.release`` is idempotent and requires the owning context to be + current — resources are freed by the context that made them. + +The probe is what decides the profile on a board. A missing GL, a software +renderer, or a failed shader compile means the three-dimensional view reports +itself unavailable and the two-dimensional application carries on unchanged. +""" + +from __future__ import annotations + +from typing import Any + +from orisys_app_shared.scene3d.contracts import ( + CameraMatrices, + DisplaySettings, + GLCapabilities, + Profile, + RenderFrame, + RenderStats, + SceneAssets, +) + +_STAGE = "stage 1 (platform PoC)" + + +def configure_surface_format(profile: Profile = "desktop33") -> None: + """Select the default GL surface format. Call before ``QApplication``. + + One profile per process: the two profiles are two different start-up + configurations, not something to switch at runtime (design document §11.1). + + Multisampling stays off in v1: it multiplies the cost of the one thing this + component must not slow down (the acquisition loop feeds the scene through a + capacity-one mailbox), and the design document defers anti-aliasing work. + """ + from PyQt5.QtGui import QSurfaceFormat + + fmt = QSurfaceFormat() + if profile == "gles30": + fmt.setRenderableType(QSurfaceFormat.OpenGLES) + fmt.setVersion(3, 0) + elif profile == "desktop33": + fmt.setRenderableType(QSurfaceFormat.OpenGL) + fmt.setVersion(3, 3) + fmt.setProfile(QSurfaceFormat.CoreProfile) + else: # pragma: no cover - a typo must not reach a board silently + raise ValueError(f"unknown GL profile: {profile!r}") + fmt.setDepthBufferSize(24) + fmt.setStencilBufferSize(8) + fmt.setSwapBehavior(QSurfaceFormat.DoubleBuffer) + fmt.setSamples(0) + QSurfaceFormat.setDefaultFormat(fmt) + + +class GLDispatch: + """Bind the GL entry points for the context Qt actually created.""" + + def bind_current_context(self, context: Any) -> GLCapabilities: + raise NotImplementedError(f"{_STAGE}: probe the context and bind PyOpenGL") + + +class GLRenderer: + """Shaders, buffers and draw calls for one scene. + + Two parts, two materials, one draw call per part: the sensing surface is + re-uploaded per frame, the housing is uploaded once and then left alone. + """ + + def initialize(self, capabilities: GLCapabilities) -> None: + raise NotImplementedError(f"{_STAGE}: compile the shaders for this profile") + + def upload_scene(self, assets: SceneAssets) -> None: + raise NotImplementedError(f"{_STAGE}: upload static buffers and the LUT") + + def update_surface(self, frame: RenderFrame) -> None: + raise NotImplementedError(f"{_STAGE}: update the dynamic surface buffers") + + def draw( + self, + camera: CameraMatrices, + settings: DisplaySettings, + framebuffer: int, + viewport: tuple[int, int], + ) -> RenderStats: + raise NotImplementedError(f"{_STAGE}: draw one frame") + + def release(self) -> None: + raise NotImplementedError(f"{_STAGE}: release GPU resources") diff --git a/apps/studio/src/orisys_studio/scene3d/panel.py b/apps/studio/src/orisys_studio/scene3d/panel.py new file mode 100644 index 0000000..4761756 --- /dev/null +++ b/apps/studio/src/orisys_studio/scene3d/panel.py @@ -0,0 +1,54 @@ +"""``ThreeDPanel``: the controls 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. + +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. +""" + +from __future__ import annotations + +from typing import Optional + +from PyQt5.QtWidgets import QWidget + +from orisys_app_shared.scene3d.contracts import ( + Diagnostic, + RenderStats, + SceneState, +) +from orisys_studio.scene3d.coordinator import ThreeDCoordinator +from orisys_studio.scene3d.widget import Sensor3DWidget + +_STAGE = "stage 4 (product integration)" + + +class ThreeDPanel(QWidget): + """Toggle, presets, reset, housing visibility, gain, colour scale, status.""" + + def __init__( + self, + coordinator: ThreeDCoordinator, + widget: Sensor3DWidget, + parent: Optional[QWidget] = None, + ) -> None: + super().__init__(parent) + self._coordinator = coordinator + self._widget = widget + + 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") + + def set_state(self, state: SceneState) -> None: + raise NotImplementedError(f"{_STAGE}: reflect the scene state") + + def set_render_stats(self, stats: RenderStats) -> None: + """Display-side statistics only: never shown as recording health.""" + raise NotImplementedError(f"{_STAGE}: show render statistics") diff --git a/apps/studio/src/orisys_studio/scene3d/widget.py b/apps/studio/src/orisys_studio/scene3d/widget.py new file mode 100644 index 0000000..9fba946 --- /dev/null +++ b/apps/studio/src/orisys_studio/scene3d/widget.py @@ -0,0 +1,84 @@ +"""``Sensor3DWidget``: the Qt widget that owns the GL context and nothing else. + +Stage 1 (a static scene, resize, close) and stage 4 (product integration) in +``docs/architecture/studio-3d-stages.md``. + +Qt owns the context and the framebuffer, so the widget must never bind target 0: +it draws into the FBO Qt has already set up for the current ``paintGL`` call. The +widget holds CPU-side data (the latest :class:`RenderFrame`) and schedules the +upload; it does no composing, no asset IO and no SDK access, so a slow render can +only drop *display* frames. +""" + +from __future__ import annotations + +from typing import Optional + +from PyQt5.QtCore import pyqtSignal +from PyQt5.QtWidgets import QOpenGLWidget, QWidget + +from orisys_app_shared.scene3d.contracts import ( + Diagnostic, + DisplaySettings, + GLCapabilities, + RenderFrame, + RenderStats, + SceneAssets, +) +from orisys_studio.scene3d.camera import OrbitCamera +from orisys_studio.scene3d.gl_backend import GLRenderer + +_STAGE = "stage 1 (static scene) and stage 4 (product integration)" + + +class Sensor3DWidget(QOpenGLWidget): + """Renders one scene. Signals carry state, never per-frame arrays.""" + + capabilitiesChanged = pyqtSignal(object) # GLCapabilities + diagnosticRaised = pyqtSignal(object) # Diagnostic + renderStatsChanged = pyqtSignal(object) # RenderStats, rate-limited + + def __init__( + self, + renderer: GLRenderer, + camera: OrbitCamera, + parent: Optional[QWidget] = None, + ) -> None: + super().__init__(parent) + self._renderer = renderer + self._camera = camera + self._assets: Optional[SceneAssets] = None + self._frame: Optional[RenderFrame] = None + self._settings = DisplaySettings() + self._capabilities: Optional[GLCapabilities] = None + self._needs_scene_upload = False + self._needs_surface_upload = False + + def set_scene(self, assets: SceneAssets) -> None: + """Retain the asset and schedule its upload for the next current context.""" + self._assets = assets + self._needs_scene_upload = True + self.update() + + def present(self, frame: RenderFrame) -> None: + """Keep the newest CPU frame and ask for a repaint. Never requires a context.""" + self._frame = frame + self._needs_surface_upload = True + self.update() + + def set_settings(self, settings: DisplaySettings) -> None: + self._settings = settings + self.update() + + # ------------------------------------------------------------ Qt overrides + def initializeGL(self) -> None: # pragma: no cover - needs a real context + raise NotImplementedError(f"{_STAGE}: bind the backend and connect teardown") + + def paintGL(self) -> None: # pragma: no cover - needs a real context + raise NotImplementedError(f"{_STAGE}: upload what changed and draw") + + def resizeGL(self, w: int, h: int) -> None: # pragma: no cover + raise NotImplementedError(f"{_STAGE}: calibrate DPR and update the projection") + + def release_resources(self) -> None: # pragma: no cover - needs a real context + raise NotImplementedError(f"{_STAGE}: makeCurrent, release, doneCurrent") diff --git a/apps/studio/tests/test_scene3d_skeleton.py b/apps/studio/tests/test_scene3d_skeleton.py new file mode 100644 index 0000000..1e64a46 --- /dev/null +++ b/apps/studio/tests/test_scene3d_skeleton.py @@ -0,0 +1,141 @@ +"""The studio-side three-dimensional skeleton: importable, inert, Qt-safe. + +Acceptance for stage 0 (``docs/architecture/studio-3d-stages.md``): the module +tree exists, it can be imported without PyOpenGL, and the one piece that must run +before ``QApplication`` — the surface format — selects a complete profile. + +Nothing here renders. These tests are the floor a later stage must keep: the GL +binding is imported only once a real context exists, and the stubs say which stage +fills them in. +""" + +from __future__ import annotations + +import sys + +import pytest + + +@pytest.fixture +def restore_surface_format(): + """The default surface format is process-wide; put it back afterwards.""" + from PyQt5.QtGui import QSurfaceFormat + + previous = QSurfaceFormat.defaultFormat() + try: + yield + finally: + QSurfaceFormat.setDefaultFormat(previous) + + +def test_the_skeleton_does_not_import_a_gl_binding(qt_app): + """PyOpenGL must be imported for the platform Qt chose, not at import time. + + The binding decides GLX versus EGL when it first loads, so importing it while + merely constructing the module tree would freeze a guess. The test fails if a + future change adds a module-level ``import OpenGL``. + """ + for name in [name for name in sys.modules if name == "OpenGL" or name.startswith("OpenGL.")]: + del sys.modules[name] + + import orisys_studio.scene3d.camera # noqa: F401 + import orisys_studio.scene3d.coordinator # noqa: F401 + import orisys_studio.scene3d.gl_backend # noqa: F401 + import orisys_studio.scene3d.panel # noqa: F401 + import orisys_studio.scene3d.widget # noqa: F401 + + loaded = [name for name in sys.modules if name == "OpenGL" or name.startswith("OpenGL.")] + assert loaded == [], f"importing the skeleton pulled in {loaded}" + + +def test_configure_surface_format_selects_a_complete_profile(qt_app, restore_surface_format): + from PyQt5.QtGui import QSurfaceFormat + + from orisys_studio.scene3d.gl_backend import configure_surface_format + + configure_surface_format("desktop33") + desktop = QSurfaceFormat.defaultFormat() + assert desktop.renderableType() == QSurfaceFormat.OpenGL + assert (desktop.majorVersion(), desktop.minorVersion()) == (3, 3) + assert desktop.profile() == QSurfaceFormat.CoreProfile + assert desktop.depthBufferSize() == 24 + assert desktop.samples() == 0, "v1 does not pay for multisampling" + + configure_surface_format("gles30") + embedded = QSurfaceFormat.defaultFormat() + assert embedded.renderableType() == QSurfaceFormat.OpenGLES + assert (embedded.majorVersion(), embedded.minorVersion()) == (3, 0) + + with pytest.raises(ValueError): + configure_surface_format("vulkan") # type: ignore[arg-type] + + +def test_the_widget_holds_cpu_state_and_touches_no_context(qt_app): + """Constructing the widget must not need a current context or a GPU.""" + from orisys_app_shared.scene3d.contracts import DisplaySettings, FrameIdentity + from orisys_studio.scene3d.camera import OrbitCamera + from orisys_studio.scene3d.gl_backend import GLRenderer + from orisys_studio.scene3d.widget import Sensor3DWidget + + widget = Sensor3DWidget(GLRenderer(), OrbitCamera()) + try: + assert isinstance(widget, Sensor3DWidget) + widget.set_settings(DisplaySettings(display_gain=3.0)) + assert widget._settings.display_gain == 3.0 + # Nothing has been presented yet, and no context was needed to get here. + assert widget._frame is None + finally: + widget.deleteLater() + + +def test_the_stubs_name_the_stage_that_fills_them_in(qt_app): + from orisys_app_shared.scene3d.contracts import ( + CameraPreset, + DisplaySettings, + FrameIdentity, + GLCapabilities, + RenderFrame, + SceneAssets, + ) + from orisys_studio.scene3d.camera import OrbitCamera + from orisys_studio.scene3d.coordinator import ThreeDCoordinator + from orisys_studio.scene3d.gl_backend import GLDispatch, GLRenderer + from orisys_studio.scene3d.widget import Sensor3DWidget + + camera = OrbitCamera() + renderer = GLRenderer() + widget = Sensor3DWidget(renderer, camera) + coordinator = ThreeDCoordinator(widget, composer=object(), executor=None, max_fps=30) # type: ignore[arg-type] + try: + assert coordinator.state == "disabled" + assert coordinator.generation == 0 + assert coordinator.superseded_frames == 0 # the mailbox is real, not a stub + + with pytest.raises(NotImplementedError) as camera_error: + camera.set_preset(CameraPreset("default", (0.0, 0.0, 1.0), (0.0, 0.0, 0.0), (0.0, 1.0, 0.0), 0.7)) + assert "stage 1" in str(camera_error.value) + + with pytest.raises(NotImplementedError) as gl_error: + GLDispatch().bind_current_context(object()) + assert "stage 1" in str(gl_error.value) + + with pytest.raises(NotImplementedError) as activate_error: + coordinator.activate(SceneAssets, "live-A") # type: ignore[arg-type] + assert "stage 3" in str(activate_error.value) + + with pytest.raises(NotImplementedError) as settings_error: + coordinator.set_settings(DisplaySettings()) + assert "stage 3" in str(settings_error.value) + + with pytest.raises(NotImplementedError) as initialize_error: + renderer.initialize(GLCapabilities("desktop33", "", "", "", "", "", False, 0)) + assert "stage 1" in str(initialize_error.value) + + with pytest.raises(NotImplementedError) as update_error: + renderer.update_surface(None) # type: ignore[arg-type] + assert "stage 1" in str(update_error.value) + + assert FrameIdentity("live-A", 1, 0, 0.0, 1).generation == 1 + assert RenderFrame is not None and SceneAssets is not None + finally: + widget.deleteLater() diff --git a/docs/architecture/studio-3d-stages.md b/docs/architecture/studio-3d-stages.md new file mode 100644 index 0000000..2ea2b56 --- /dev/null +++ b/docs/architecture/studio-3d-stages.md @@ -0,0 +1,135 @@ +# Studio 三维显示：实施阶段与验收 + +配套文档：`studio-3d-design.md`（规格）、`studio-3d-contracts.pyi`（接口骨架）。 +本文只回答一件事：**按什么顺序做，每一阶段做完拿什么验收**。 + +阶段 0 已经完成（骨架，见下）；阶段 1 起逐段实现。每个阶段结束时必须同时交出： +实现、公共接口单测、输入样例、失败样例、兼容说明、平台诊断报告（规格 §13）。 + +原则（写下来是因为后面每阶段都会被诱惑违反一次）： + +- **骨架已经能跑测试，就不要为了快而绕过合同**：`scene3d.contracts` 的校验是唯一入口， + 适配器/合成器/加载器都不许自己再定义一遍形状和单位。 +- **二维永远可用**：三维持有任何问题（资产、GL、性能）都只降级三维视图，不阻塞采集、记录、力曲线。 +- **不在没有证据时造物理量**：近似位移不写成毫米，未标定的空间尺度不造高度场， + 壳体的顶点不参与测量统计。 + +--- + +## 阶段 0：骨架（已完成） + +产出： + +- `shared/src/orisys_app_shared/scene3d/`：`contracts.py`（全部 DTO、枚举、诊断码、 + `freeze`、五个校验器、`LatestFrameMailbox`、协议）＋ `assets.py`/`sampling.py`/`composer.py` 签名骨架。 +- `apps/studio/src/orisys_studio/scene3d/`：`camera.py`/`adapters.py`/`coordinator.py`/ + `gl_backend.py`/`widget.py`/`panel.py` 签名骨架；`configure_surface_format()` 已实现。 +- `tools/scene_assets/`：`probe_gl.py`/`validate_scene.py`/`compile_scene.py` 入口（未实现时返回 2）。 +- 测试：`shared/tests/test_scene3d_contracts.py`（26 项）、 + `apps/studio/tests/test_scene3d_skeleton.py`（4 项）。 + +验收（已通过）：合同校验的输入/失败样例全部覆盖；`LatestFrameMailbox` 在多线程下不阻塞生产者； +骨架导入不引入 PyOpenGL；`configure_surface_format` 两个 profile 的版本/类型/深度缓冲可断言； +所有未实现入口都抛 `NotImplementedError` 并写明归属阶段。 + +--- + +## 阶段 1：平台 PoC（图形栈先落地） + +**做**：`probe_gl`（真实 Qt 窗口 + 真实 context）、`GLDispatch`（探测 + 绑定 PyOpenGL）、 +`GLRenderer` 最小集（一个 shader、一个 VBO/EBO、静态三角形）、静态感知面 + 外壳两个 part 各一次 draw、 +`OrbitCamera` 的 `fit`/`orbit`、`Sensor3DWidget` 的 `initializeGL`/`paintGL`/`resizeGL`/`release_resources` +与 `aboutToBeDestroyed` 清理。 + +**验收**： + +| 用例 | 通过标准 | +|---|---| +| 能力报告 | 开发机 Windows 出 `desktop33` 报告：vendor/renderer/GL/GLSL/DPR/framebuffer，且明确标注是否软件渲染 | +| 静态场景 | 三角形与感知面/外壳都可见；外壳可隐藏；感知面与外壳不共用缓冲 | +| 生命周期 | 反复 resize（含 DPR 变化）、隐藏/显示、关闭、context 重建各 20 次无崩溃、无泄漏 | +| 失败样例 | 强制 shader 编译失败 → 三维 Unavailable + 诊断，二维照常；`gl_unsupported` 路径同样 | +| 记录隔离 | 开/关三维各录 30 s，CSV 行数与有效样本完全一致 | + +**明确不做**：ARM 板卡适配（按用户要求暂缓），但 probe 脚本要能直接在板上跑出报告。 + +--- + +## 阶段 2：资产合同（几何与坐标） + +**做**：`LegacyFingerImporter`（旧 NPZ：`mesh.npz`/`unroll_map_uv_s0.4.npz`/`camera_intrinsics.npz`）、 +坐标与单位统一（`meters_per_source_unit` → 刚体 `transform_sensor_from_source`）、 +`SurfaceBinding` 的生成与自检、`validate_scene.py`、一份简化外壳 GLB 的编译路径、manifest/schema。 + +**验收**： + +| 用例 | 通过标准 | +|---|---| +| 拓扑不混用 | 1032 与 1005 两套网格分别加载；错配（binding 顶点数 ≠ 感知面顶点数）被拒 | +| UV 域 | 300×280 与 500×500 两个域的 `shape_hw`/scale/flip_v 各不相同，错配被拒 | +| 采样表 | 每顶点权重和为 1；未用项 `-1`/0；越界下标被拒；接缝不跨域插值 | +| 单位 | 三个已知非共线点在表面与外壳中对齐；米/毫米换算与法线方向正确 | +| 身份 | 改 manifest 任一哈希、改 binding、改 geometry_id 均被拒并给出 `geometry_mismatch`/`field_domain_mismatch` | + +**产出样例**：一张 `finger` 场景包 + 一份 `validate_scene.py --json` 报告。 + +--- + +## 阶段 3：实时数据（字段 → 渲染帧） + +**做**：`FingerFrameAdapter`/`BoxFrameAdapter`、`SurfaceSampler`、`SceneComposer` +（`compose`/`rest_frame`）、`ThreeDCoordinator` 的双 mailbox 与 generation/过期拒绝、`submit` 的 +`accepted/replaced/stale/disabled/invalid` 语义。 + +**验收**： + +| 用例 | 通过标准 | +|---|---| +| 采样正确性 | 常量场/斜坡/单点脉冲/NaN/遮罩场与 CPU 参考实现逐位一致 | +| 无效语义 | 任一正权重点无效 ⇒ 该顶点无效；非有限 scalar 不显示为 0；参考帧 `scalar` 全不可用 | +| 色标语义 | `display_gain` 只改显示位置：增大倍率前后 `scalar` 与力值逐位不变 | +| 过期帧 | 旧 `generation`/`request_id` 永远不能覆盖新场景；`stale_frame` 安静丢弃并计数 | +| 不阻塞采集 | 合成人为放慢 10 倍，`submit` 调用耗时 p99 < 1 ms，记录行数不变 | +| 缺数据 | 无字段 → `no_data`，清掉旧测量颜色，允许显示参考模型 | + +--- + +## 阶段 4：Qt 产品接入（界面与生命周期） + +**做**：`Sensor3DWidget` 完整、`ThreeDPanel`（开关/预设/复位/外壳/倍率/色标/状态）、 +`OrbitCamera` 全部交互、preferences 持久化、`app_window` 装配（视图切换、`closeEvent`）、 +`main.py` 在 `QApplication` 之前的 profile 选择。 + +**验收**： + +| 用例 | 通过标准 | +|---|---| +| 交互 | 旋转/平移/缩放/复位/预设；观察相机变化不改算法相机、标定与力轴（矩阵断言） | +| 单位与质量显示 | `approximate` 位移显示为"相对形变示意 ×N"，不出现在毫米文案里 | +| 视图切换 | 三维开关不改变采集、记录与力曲线状态；切换 20 次无泄漏 | +| 关停 | `closeEvent` → `release_resources` 幂等；重复关闭不抛异常 | +| 界面文案 | 新增 key 中英齐全（沿用 `tools`/locale 校验方式） | + +--- + +## 阶段 5：发布保护与性能（收尾） + +**做**：精简/签名包（`compile_scene.py`）、可选 AES-GCM + `KeyProvider`、平台锁定文件、 +性能与长时测试、`session.json` 的 `visualization3d` 能力块（旧文件缺块视为不支持）。 + +**验收**：规格 §12 的"包保护""性能""长时"三行 + 两平台（Windows 开发机 / 目标板） +各一份 probe 报告；`p50/p95` 合成、上传、绘制、总响应延迟写进报告；2 小时采集+记录+三维后 +RSS/GPU 内存不持续增长，任务队列有上界。 + +--- + +## 依赖关系（并行度） + +``` +阶段 1 (图形栈) ─┬─> 阶段 4 (界面) ─┐ + │ ├─> 阶段 5 +阶段 2 (资产) ──┴─> 阶段 3 (数据) ─┘ +``` + +阶段 1 与阶段 2 无依赖，可并行；阶段 3 需要阶段 2 的绑定；阶段 4 需要 1 和 3。 +每阶段开始前先确认上一阶段的失败样例仍然被覆盖（回归门）。 diff --git a/shared/src/orisys_app_shared/scene3d/__init__.py b/shared/src/orisys_app_shared/scene3d/__init__.py new file mode 100644 index 0000000..0d2fc0a --- /dev/null +++ b/shared/src/orisys_app_shared/scene3d/__init__.py @@ -0,0 +1,105 @@ +"""Three-dimensional display: contracts, assets, sampling, composition. + +Import-safe by design: nothing in this package imports Qt, OpenGL or the SDK, so +it can be imported by the acquisition thread, by a build script, and by CI on a +machine with no GPU (``docs/architecture/studio-3d-design.md`` §6). + +The executable contracts live in :mod:`~orisys_app_shared.scene3d.contracts`; the +other modules are the skeleton of stage 2 and 3 (see +``docs/architecture/studio-3d-stages.md``) and raise ``NotImplementedError`` until +their stage lands. +""" + +from orisys_app_shared.scene3d.contracts import ( + DIAGNOSTIC_SEVERITY, + LENGTH_UNIT_M, + SENSOR_FRAME, + UNAVAILABLE_DESCRIPTOR, + CameraMatrices, + CameraPreset, + ColorScale, + CompileOptions, + CompileReport, + Diagnostic, + DisplaySettings, + ExpectedGeometry, + F32, + F64, + FieldDescriptor, + FrameIdentity, + GLCapabilities, + I32, + LatestFrameMailbox, + Mask, + Material, + MeshPart, + OfferResult, + Quality, + RenderFrame, + RenderStats, + SceneAssets, + SceneState, + SurfaceBinding, + SurfaceFieldFrame, + ThreeDError, + U32, + VertexFields, + diagnostic, + error, + freeze, + rest_descriptor, + summarize_parts, + validate_assets, + validate_binding, + validate_display_settings, + validate_mesh_part, + validate_render_frame, + validate_surface_frame, +) + +__all__ = [ + "DIAGNOSTIC_SEVERITY", + "LENGTH_UNIT_M", + "SENSOR_FRAME", + "UNAVAILABLE_DESCRIPTOR", + "CameraMatrices", + "CameraPreset", + "ColorScale", + "CompileOptions", + "CompileReport", + "Diagnostic", + "DisplaySettings", + "ExpectedGeometry", + "F32", + "F64", + "FieldDescriptor", + "FrameIdentity", + "GLCapabilities", + "I32", + "LatestFrameMailbox", + "Mask", + "Material", + "MeshPart", + "OfferResult", + "Quality", + "RenderFrame", + "RenderStats", + "SceneAssets", + "SceneState", + "SurfaceBinding", + "SurfaceFieldFrame", + "ThreeDError", + "U32", + "VertexFields", + "diagnostic", + "error", + "freeze", + "rest_descriptor", + "summarize_parts", + "validate_assets", + "validate_binding", + "validate_display_settings", + "validate_mesh_part", + "validate_render_frame", + "validate_surface_frame", +] diff --git a/shared/src/orisys_app_shared/scene3d/assets.py b/shared/src/orisys_app_shared/scene3d/assets.py new file mode 100644 index 0000000..d2aa4d1 --- /dev/null +++ b/shared/src/orisys_app_shared/scene3d/assets.py @@ -0,0 +1,92 @@ +"""Assets: loading a validated scene, importing the legacy finger export. + +Stage 2 in ``docs/architecture/studio-3d-stages.md``. The signatures are fixed +here so the rest of the skeleton (and the compiler tool) can be written against +them; the bodies arrive with the stage. + +Two rules the implementation must keep, because violating them is invisible until +a customer sees the wrong part move: + +* the algorithm mesh and the display mesh are **different topologies** (1032 vs + 1005 vertices) and are never joined by array index; +* the housing is a separate part with its own buffers, so it can be hidden, + simplified and re-exported without touching the sensing geometry or its + calibration hashes. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Optional + +from orisys_app_shared.scene3d.contracts import ( + CompileOptions, + CompileReport, + Diagnostic, + ExpectedGeometry, + F64, + KeyProvider, + SceneAssets, +) + +_STAGE = "stage 2 (asset contracts)" + + +class PackagedAssetLoader: + """Load and authenticate a ``.oryscene`` runtime package. + + CPU/IO thread. Verification order: envelope signature, then entry hashes, then + geometry/domain/unit checks against ``expected``. There is no plaintext + fallback: a package that fails authentication is refused, never retried. + """ + + def load( + self, + path: Path, + expected: ExpectedGeometry, + keys: Optional[KeyProvider] = None, + ) -> SceneAssets: + raise NotImplementedError(f"{_STAGE}: load a signed .oryscene package") + + def validate( + self, assets: SceneAssets, expected: ExpectedGeometry + ) -> tuple[Diagnostic, ...]: + raise NotImplementedError(f"{_STAGE}: validate a loaded scene") + + +class LegacyFingerImporter: + """Convert ``sensors/finger/assets/30fps-0703`` into a scene (offline only). + + ``meters_per_source_unit`` scales source lengths into metres first; + ``transform_sensor_from_source`` is then rigid (rotation and a translation in + metres, no extra scale). Normals receive the rotation only. Missing coordinate + evidence is a refusal, not a guess from a filename. + """ + + def convert( + self, + export_dir: Path, + map_path: Path, + *, + transform_sensor_from_source: F64, + meters_per_source_unit: float, + ) -> SceneAssets: + raise NotImplementedError(f"{_STAGE}: import the legacy NPZ export") + + +class SceneCompiler: + """Package a scene (and an optional housing GLB) for release (offline). + + Signing happens with a key from the build environment; the private key is + never part of the client runtime. Output is written to a temporary file and + atomically renamed, so a failed build cannot leave a half package behind. + """ + + def compile( + self, + surface: SceneAssets, + housing_glb: Optional[Path], + output: Path, + options: CompileOptions, + ) -> CompileReport: + raise NotImplementedError(f"{_STAGE}: compile a runtime package") diff --git a/shared/src/orisys_app_shared/scene3d/composer.py b/shared/src/orisys_app_shared/scene3d/composer.py new file mode 100644 index 0000000..f2041a6 --- /dev/null +++ b/shared/src/orisys_app_shared/scene3d/composer.py @@ -0,0 +1,51 @@ +"""Composing a render frame: sample, deform, update normals. Pure NumPy. + +Stage 3 in ``docs/architecture/studio-3d-stages.md``. + +This is the module that decides what a colour and a vertex position *mean*, so the +rules from §3.2 of the design document are its reason to exist: + +* ``scalar`` is never multiplied by the display gain — the colour scale describes + the measurement, and the gain describes only how far the drawing is exaggerated; +* a displacement whose quality is ``approximate`` is drawn as "relative + deformation, ×N" and is never labelled in millimetres; +* without a trustworthy displacement the surface is coloured in place, and the + total force is never turned into a per-vertex height. +""" + +from __future__ import annotations + +from orisys_app_shared.scene3d.contracts import ( + DisplaySettings, + FrameIdentity, + RenderFrame, + SceneAssets, + SurfaceFieldFrame, +) + +_STAGE = "stage 3 (live data)" + + +class SceneComposer: + """CPU worker: one field frame in, one render frame out. + + Runs off the GUI thread and never touches Qt or GL. Recomputes normals from the + deformed positions rather than transforming the rest normals, because the + difference is exactly the deformation the operator is looking at. + """ + + def compose( + self, + assets: SceneAssets, + frame: SurfaceFieldFrame, + settings: DisplaySettings, + ) -> RenderFrame: + raise NotImplementedError(f"{_STAGE}: compose a render frame") + + def rest_frame(self, assets: SceneAssets, identity: FrameIdentity) -> RenderFrame: + """The static reference: nothing measured, so nothing coloured. + + The scalar is all-zero with an ``unavailable`` descriptor; the UI hides the + measurement colour bar instead of leaving the previous one on screen. + """ + raise NotImplementedError(f"{_STAGE}: compose the reference frame") diff --git a/shared/src/orisys_app_shared/scene3d/contracts.py b/shared/src/orisys_app_shared/scene3d/contracts.py new file mode 100644 index 0000000..aad4ddc --- /dev/null +++ b/shared/src/orisys_app_shared/scene3d/contracts.py @@ -0,0 +1,762 @@ +"""Three-dimensional display contracts: DTOs, invariants, diagnostics. + +Specification: ``docs/architecture/studio-3d-design.md``; the interface skeleton is +``docs/architecture/studio-3d-contracts.pyi``. This module is the executable half +of that specification, and it deliberately imports **nothing** from Qt, OpenGL, +the SDK or the acquisition layer: everything here is plain NumPy and stdlib, so it +runs in CI, in the CPU test environment, and on a board whose GL stack failed. + +Two rules from the design document are enforced here rather than trusted: + +* **A frozen dataclass does not freeze an ndarray.** Every array that crosses a + module boundary goes through :func:`freeze` (own it, make it C-contiguous, make + it read-only) or it is rejected by the validators. +* **A shape is not a meaning.** Quantities carry ``unit`` and ``quality``; a + displacement that is only approximate is never presented as millimetres, and a + reference frame's scalar is explicitly ``unavailable`` rather than zero. +""" + +from __future__ import annotations + +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Generic, Iterable, Literal, Mapping, Optional, Protocol, TypeVar + +import numpy as np +from numpy.typing import NDArray + +F32 = NDArray[np.float32] +F64 = NDArray[np.float64] +I32 = NDArray[np.int32] +U32 = NDArray[np.uint32] +Mask = NDArray[np.bool_] + +Role = Literal["sensing", "housing", "reference"] +Quality = Literal["calibrated", "approximate", "display_only"] +Profile = Literal["desktop33", "gles30"] +OfferResult = Literal["accepted", "replaced", "stale", "disabled", "invalid"] +SceneState = Literal[ + "disabled", "loading", "ready", "live", "no_data", "suspended", "unavailable" +] +Severity = Literal["info", "warning", "error"] + +#: The coordinate frame every published array is in. Right-handed, metres. +SENSOR_FRAME = "sensor" +LENGTH_UNIT_M = "m" + +#: Weight sum tolerance for a :class:`SurfaceBinding` row. +WEIGHT_SUM_TOL = 1e-4 + +#: The standard failure codes of §8.3, and how loud each one is. +DIAGNOSTIC_SEVERITY: Mapping[str, Severity] = { + "asset_auth_failed": "error", + "geometry_mismatch": "error", + "field_domain_mismatch": "error", + "field_unavailable": "warning", + "stale_frame": "info", + "gl_unsupported": "error", + "shader_compile_failed": "error", + "gpu_budget_exceeded": "error", + "asset_rejected": "error", + "no_data": "info", +} + + +# --------------------------------------------------------------------- errors +@dataclass(frozen=True) +class Diagnostic: + """A machine-readable code plus a message that is safe to show and to log. + + The message must never contain keys, decrypted asset data or file contents: + diagnostics travel to the UI and into logs. + """ + + code: str + severity: Severity + message: str + details: Mapping[str, str] = field(default_factory=dict) + + +def diagnostic(code: str, message: str, **details: str) -> Diagnostic: + """A diagnostic whose severity comes from the standard code table.""" + return Diagnostic( + code=str(code), + severity=DIAGNOSTIC_SEVERITY.get(str(code), "error"), + message=str(message), + details={str(key): str(value) for key, value in details.items()}, + ) + + +class ThreeDError(Exception): + """Boundary exception: the GUI turns it into a diagnostic and keeps 2D alive.""" + + def __init__(self, diagnostic: Diagnostic) -> None: + super().__init__(f"{diagnostic.code}: {diagnostic.message}") + self.diagnostic = diagnostic + + +def error(code: str, message: str, **details: str) -> ThreeDError: + return ThreeDError(diagnostic(code, message, **details)) + + +# ------------------------------------------------------------------ identity +@dataclass(frozen=True) +class FrameIdentity: + """Which frame this is, and which scene it belongs to. + + ``source_id`` distinguishes a live stream from a recording — a filename does + not, because two recordings can carry the same name. ``generation`` changes + whenever the scene, the source or the geometry changes, and is what makes an + old in-flight result rejectable. + """ + + source_id: str + session_id: int + frame_id: int + t_rel_s: float + generation: int + request_id: int = 0 + + def is_from(self, other: "FrameIdentity") -> bool: + return ( + self.source_id == other.source_id + and self.session_id == other.session_id + and self.generation == other.generation + ) + + +@dataclass(frozen=True) +class FieldDescriptor: + """What a field means: quantity, unit, quality, sign — never inferred.""" + + quantity: str + unit: str + quality: Quality + signed: bool + calibration_id: Optional[str] = None + + +#: The descriptor a reference frame carries: no measurement is claimed at all. +UNAVAILABLE_DESCRIPTOR = FieldDescriptor( + quantity="unavailable", unit="1", quality="display_only", signed=False +) + + +# ------------------------------------------------------------------ the frame +@dataclass(frozen=True) +class SurfaceFieldFrame: + """One acquired frame of a surface field, in the sensor's own UV domain. + + ``scalar`` is what gets coloured; ``displacement_sensor_m`` is what may move + vertices, and only when its ``displacement_quality`` says it can be trusted. + A missing field is never carried over from the previous frame. + """ + + identity: FrameIdentity + geometry_id: str + field_domain_id: str + scalar: F32 # [H,W] + scalar_descriptor: FieldDescriptor + valid: Mask # [H,W] + displacement_sensor_m: Optional[F32] = None # [H,W,3], before display gain + displacement_quality: Optional[Quality] = None + diagnostic: Optional[Diagnostic] = None + + +# -------------------------------------------------------------------- assets +@dataclass(frozen=True) +class Material: + base_color_linear: tuple[float, float, float, float] + shading: Literal["unlit_scalar", "diffuse"] + double_sided: bool = False + + +@dataclass(frozen=True) +class MeshPart: + """One drawable part. ``role`` decides whether it may be updated per frame.""" + + part_id: str + role: Role + positions_sensor_m: F32 # [N,3] + normals_sensor: F32 # [N,3], unit + triangles: U32 # [M,3] + material: Material + lod_id: str = "lod0" + + +@dataclass(frozen=True) +class SurfaceBinding: + """How sensing vertices sample a UV field: up to four bilinear taps each.""" + + part_id: str + geometry_id: str + field_domain_id: str + shape_hw: tuple[int, int] + sample_indices: I32 # [N,4], y*W+x, unused = -1 + sample_weights: F32 # [N,4], non-negative, used weights sum to 1 + + +@dataclass(frozen=True) +class CameraPreset: + """A named view of the *scene*. It is not the algorithm camera.""" + + name: str + eye_sensor_m: tuple[float, float, float] + target_sensor_m: tuple[float, float, float] + up_sensor: tuple[float, float, float] + vertical_fov_rad: float + + +@dataclass(frozen=True) +class ExpectedGeometry: + """What the device says it is; an asset must match it or be refused.""" + + sensor_model: str + hardware_revision: str + geometry_id: str + field_domain_id: str + shape_hw: tuple[int, int] + + +@dataclass(frozen=True) +class SceneAssets: + """A validated scene: one sensing part, zero or more housing parts.""" + + asset_id: str + asset_version: str + asset_hash: str + geometry_id: str + parts: tuple[MeshPart, ...] + binding: SurfaceBinding + camera_presets: tuple[CameraPreset, ...] + bounds_sensor_m: F32 # [2,3]: min, max + field_domain_id: str = "" + shape_hw: tuple[int, int] = (0, 0) + + def part(self, part_id: str) -> Optional[MeshPart]: + for item in self.parts: + if item.part_id == part_id: + return item + return None + + def sensing_part(self) -> MeshPart: + for item in self.parts: + if item.role == "sensing": + return item + raise error("geometry_mismatch", "the scene has no sensing part") + + +# ------------------------------------------------------------------- display +@dataclass(frozen=True) +class ColorScale: + descriptor: FieldDescriptor + lower: float + upper: float + lut_id: str + mode: Literal["fixed", "automatic"] = "fixed" + + +@dataclass(frozen=True) +class DisplaySettings: + """Display-only settings. None of these reach the camera or the recording.""" + + display_gain: float = 0.0 + deform_enabled: bool = False + color_scale: Optional[ColorScale] = None + housing_visible: bool = True + max_fps: int = 30 + + +@dataclass(frozen=True) +class VertexFields: + """A UV field sampled onto the sensing vertices.""" + + scalar: F32 # [N] + valid: Mask # [N] + displacement_sensor_m: Optional[F32] = None # [N,3] + + +@dataclass(frozen=True) +class RenderFrame: + """A composed frame, ready for the GPU thread. Carries no GL handles.""" + + identity: FrameIdentity + asset_hash: str + geometry_id: str + sensing_part_id: str + positions_sensor_m: F32 # [N,3], display displacement already applied + normals_sensor: F32 # [N,3] + scalar: F32 # [N], before display gain + valid: Mask # [N] + scalar_descriptor: FieldDescriptor + diagnostics: tuple[Diagnostic, ...] = () + + +@dataclass(frozen=True) +class CameraMatrices: + view: F32 # [4,4]; p_camera = view @ p_sensor + projection: F32 # [4,4]; p_clip = projection @ p_camera + + +@dataclass(frozen=True) +class GLCapabilities: + profile: Profile + vendor: str + renderer: str + gl_version: str + glsl_version: str + qt_platform: str + software_renderer: bool + max_texture_size: int + diagnostics: tuple[Diagnostic, ...] = () + + +@dataclass(frozen=True) +class RenderStats: + displayed_frames: int = 0 + superseded_frames: int = 0 + compose_ms: float = 0.0 + upload_ms: float = 0.0 + paint_ms: float = 0.0 + estimated_gpu_bytes: int = 0 + + +@dataclass(frozen=True) +class CompileOptions: + asset_id: str + version: str + expected: ExpectedGeometry + transform_sensor_from_glb: F64 # [4,4] + max_surface_vertices: int + max_housing_triangles: int + signing_key_reference: str + encryption_key_id: Optional[str] = None + + +@dataclass(frozen=True) +class CompileReport: + output: Path + asset_hash: str + diagnostics: tuple[Diagnostic, ...] = () + part_vertex_counts: Mapping[str, int] = field(default_factory=dict) + part_triangle_counts: Mapping[str, int] = field(default_factory=dict) + + +# ------------------------------------------------------------------ freezing +def freeze(array: Any, *, dtype: Any, copy: bool = True) -> NDArray[Any]: + """Own the data, make it contiguous, make it read-only. + + A frozen dataclass does not freeze its arrays; without this a producer thread + can keep writing into a frame another thread is already rendering. + """ + data = np.array(array, dtype=dtype, order="C", copy=copy) + data.setflags(write=False) + return data + + +def _readonly_c_contiguous(name: str, array: Any) -> None: + if not isinstance(array, np.ndarray): + raise error("field_domain_mismatch", f"{name} must be a NumPy array") + if not array.flags["C_CONTIGUOUS"]: + raise error("field_domain_mismatch", f"{name} must be C-contiguous") + if array.flags["WRITEABLE"]: + raise error( + "field_domain_mismatch", + f"{name} must be read-only (use scene3d.contracts.freeze)", + ) + + +def _shape(name: str, array: np.ndarray, shape: tuple[int, ...]) -> None: + if tuple(array.shape) != tuple(shape): + raise error( + "field_domain_mismatch", + f"{name} has shape {tuple(array.shape)}, expected {tuple(shape)}", + ) + + +def _finite(name: str, array: np.ndarray) -> None: + if array.size and not bool(np.isfinite(array).all()): + raise error("field_domain_mismatch", f"{name} contains non-finite values") + + +# ---------------------------------------------------------------- validation +def validate_surface_frame(frame: SurfaceFieldFrame) -> tuple[Diagnostic, ...]: + """Check a UV field before it is allowed near a worker thread.""" + out: list[Diagnostic] = [] + if not frame.geometry_id.strip(): + raise error("geometry_mismatch", "a surface frame must name its geometry") + if not frame.field_domain_id.strip(): + raise error("field_domain_mismatch", "a surface frame must name its field domain") + for name, array in (("scalar", frame.scalar), ("valid", frame.valid)): + _readonly_c_contiguous(name, array) + _shape("scalar", frame.scalar, frame.scalar.shape) + if frame.scalar.dtype != np.float32: + raise error("field_domain_mismatch", f"scalar must be float32, got {frame.scalar.dtype}") + if frame.valid.dtype != np.bool_: + raise error("field_domain_mismatch", f"valid must be bool, got {frame.valid.dtype}") + if frame.scalar.shape != frame.valid.shape: + raise error( + "field_domain_mismatch", + f"scalar {frame.scalar.shape} and valid {frame.valid.shape} disagree", + ) + if frame.scalar.ndim != 2: + raise error("field_domain_mismatch", "a UV field is two-dimensional [H,W]") + if frame.displacement_sensor_m is not None: + _readonly_c_contiguous("displacement_sensor_m", frame.displacement_sensor_m) + _shape( + "displacement_sensor_m", + frame.displacement_sensor_m, + (*frame.scalar.shape, 3), + ) + if frame.displacement_sensor_m.dtype != np.float32: + raise error("field_domain_mismatch", "displacement must be float32") + if frame.displacement_quality is None: + raise error( + "field_domain_mismatch", + "a displacement without a quality statement is not publishable", + ) + # A non-finite scalar is "no measurement here", and must say so in `valid`: + # colouring it as a value is how a gap becomes an invented number. + if frame.scalar.size: + non_finite = ~np.isfinite(frame.scalar) + if bool((non_finite & frame.valid).any()): + out.append( + diagnostic( + "field_unavailable", + "non-finite scalars are marked valid", + count=str(int((non_finite & frame.valid).sum())), + ) + ) + if not frame.scalar_descriptor.unit.strip(): + out.append(diagnostic("field_unavailable", "the scalar has no unit")) + return tuple(out) + + +def validate_binding(binding: SurfaceBinding, *, vertices: int) -> tuple[Diagnostic, ...]: + """The sampling table: shapes, ranges, weights and the domain it belongs to.""" + for name, array in ( + ("sample_indices", binding.sample_indices), + ("sample_weights", binding.sample_weights), + ): + _readonly_c_contiguous(name, array) + _shape("sample_indices", binding.sample_indices, (vertices, 4)) + _shape("sample_weights", binding.sample_weights, (vertices, 4)) + if binding.sample_indices.dtype != np.int32: + raise error("field_domain_mismatch", "sample_indices must be int32") + if binding.sample_weights.dtype != np.float32: + raise error("field_domain_mismatch", "sample_weights must be float32") + height, width = binding.shape_hw + if height <= 0 or width <= 0: + raise error("field_domain_mismatch", f"invalid field shape {binding.shape_hw}") + domain = height * width + indices = binding.sample_indices + weights = binding.sample_weights + used = indices >= 0 + if bool((indices[~used] != -1).any()): + raise error("field_domain_mismatch", "unused samples must be -1") + if bool((weights[~used] != 0.0).any()): + raise error("field_domain_mismatch", "unused samples must carry weight 0") + if bool((weights < 0.0).any()): + raise error("field_domain_mismatch", "weights must be non-negative") + if bool((indices[used] >= domain).any()): + raise error( + "field_domain_mismatch", + "sample indices fall outside the declared field domain", + ) + sums = weights.sum(axis=1) + bad = np.abs(sums - 1.0) > WEIGHT_SUM_TOL + if bool(bad.any()): + raise error( + "field_domain_mismatch", + f"{(int(bad.sum()))} vertices do not sample exactly once", + ) + return () + + +def validate_mesh_part(part: MeshPart) -> None: + for name, array in ( + ("positions_sensor_m", part.positions_sensor_m), + ("normals_sensor", part.normals_sensor), + ("triangles", part.triangles), + ): + _readonly_c_contiguous(name, array) + count = int(part.positions_sensor_m.shape[0]) + _shape("positions_sensor_m", part.positions_sensor_m, (count, 3)) + _shape("normals_sensor", part.normals_sensor, (count, 3)) + if part.positions_sensor_m.dtype != np.float32 or part.normals_sensor.dtype != np.float32: + raise error("geometry_mismatch", f"{part.part_id}: positions and normals are float32") + if part.triangles.dtype != np.uint32: + raise error("geometry_mismatch", f"{part.part_id}: triangles are uint32") + if part.triangles.ndim != 2 or part.triangles.shape[1] != 3: + raise error("geometry_mismatch", f"{part.part_id}: triangles are [M,3]") + if part.triangles.size and int(part.triangles.max()) >= count: + raise error("geometry_mismatch", f"{part.part_id}: a triangle index is out of range") + if part.role not in ("sensing", "housing", "reference"): + raise error("geometry_mismatch", f"{part.part_id}: unknown role {part.role!r}") + if part.role == "sensing": + _finite("positions_sensor_m", part.positions_sensor_m) + _finite("normals_sensor", part.normals_sensor) + if count and part.role == "sensing": + norms = np.linalg.norm(np.asarray(part.normals_sensor, dtype=np.float64), axis=1) + if not np.allclose(norms, 1.0, atol=1e-3): + raise error("geometry_mismatch", f"{part.part_id}: normals are not unit length") + + +def validate_assets( + assets: SceneAssets, expected: Optional[ExpectedGeometry] = None +) -> tuple[Diagnostic, ...]: + """Everything a loader must check before an asset is published. + + Errors raise; warnings are returned for the UI to show. + """ + sensing = [part for part in assets.parts if part.role == "sensing"] + if len(sensing) != 1: + raise error( + "geometry_mismatch", + f"v1 supports exactly one sensing part, found {len(sensing)}", + parts=",".join(part.part_id for part in assets.parts), + ) + for part in assets.parts: + validate_mesh_part(part) + if assets.binding.part_id != sensing[0].part_id: + raise error( + "geometry_mismatch", + "the binding belongs to a part that is not the sensing part", + binding=assets.binding.part_id, + sensing=sensing[0].part_id, + ) + if assets.binding.geometry_id != assets.geometry_id: + raise error("geometry_mismatch", "the binding names a different geometry") + if assets.geometry_id != sensing[0].lod_id and assets.binding.field_domain_id != assets.field_domain_id: + raise error("field_domain_mismatch", "the binding names a different field domain") + _readonly_c_contiguous("bounds_sensor_m", assets.bounds_sensor_m) + _shape("bounds_sensor_m", assets.bounds_sensor_m, (2, 3)) + validate_binding(assets.binding, vertices=int(sensing[0].positions_sensor_m.shape[0])) + + out: list[Diagnostic] = [] + if expected is not None: + if expected.geometry_id != assets.geometry_id: + raise error( + "geometry_mismatch", + "asset geometry does not match the device", + asset=assets.geometry_id, + device=expected.geometry_id, + ) + if expected.field_domain_id != assets.binding.field_domain_id: + raise error( + "field_domain_mismatch", + "asset field domain does not match the device", + asset=assets.binding.field_domain_id, + device=expected.field_domain_id, + ) + if tuple(expected.shape_hw) != tuple(assets.binding.shape_hw): + raise error( + "field_domain_mismatch", + "asset UV shape does not match the device", + asset=f"{assets.binding.shape_hw[0]}x{assets.binding.shape_hw[1]}", + device=f"{expected.shape_hw[0]}x{expected.shape_hw[1]}", + ) + if expected.sensor_model and expected.sensor_model != assets.parts[0].role: + pass # the model name is descriptive; the geometry id is the contract + for preset in assets.camera_presets: + if not preset.name.strip(): + out.append(diagnostic("asset_rejected", "a camera preset has no name")) + if not (0.0 < preset.vertical_fov_rad < np.pi): + raise error("asset_rejected", f"preset {preset.name}: invalid field of view") + return tuple(out) + + +def validate_display_settings(settings: DisplaySettings) -> tuple[Diagnostic, ...]: + out: list[Diagnostic] = [] + if not np.isfinite(settings.display_gain) or settings.display_gain < 0.0: + raise error("asset_rejected", "display gain must be finite and non-negative") + if settings.max_fps <= 0: + raise error("asset_rejected", "max_fps must be positive") + scale = settings.color_scale + if scale is not None: + if not np.isfinite(scale.lower) or not np.isfinite(scale.upper): + raise error("asset_rejected", "a colour scale needs finite limits") + if scale.lower >= scale.upper: + raise error("asset_rejected", "a colour scale needs lower < upper") + return tuple(out) + + +def validate_render_frame(frame: RenderFrame, *, vertices: int) -> None: + for name, array in ( + ("positions_sensor_m", frame.positions_sensor_m), + ("normals_sensor", frame.normals_sensor), + ("scalar", frame.scalar), + ("valid", frame.valid), + ): + _readonly_c_contiguous(name, array) + _shape("positions_sensor_m", frame.positions_sensor_m, (vertices, 3)) + _shape("normals_sensor", frame.normals_sensor, (vertices, 3)) + _shape("scalar", frame.scalar, (vertices,)) + _shape("valid", frame.valid, (vertices,)) + if frame.scalar.dtype != np.float32 or frame.valid.dtype != np.bool_: + raise error("field_domain_mismatch", "render scalar is float32 and valid is bool") + if not frame.asset_hash.strip() or not frame.geometry_id.strip(): + raise error("geometry_mismatch", "a render frame must name its asset and geometry") + + +#: The descriptor a reference frame carries: nothing is measured, and the UI has +#: to hide the measurement colour bar rather than leave the previous one showing. +def rest_descriptor() -> FieldDescriptor: + return UNAVAILABLE_DESCRIPTOR + + +# ------------------------------------------------------------------- mailbox +T = TypeVar("T") + + +class LatestFrameMailbox(Generic[T]): + """Capacity-one mailbox: the newest frame wins, the older one is counted. + + Used twice per scene (raw field frames, composed render frames). Producers + (the acquisition thread) must never block on a slow consumer: ``offer`` drops + the frame it replaces and returns in constant time, and the replacement count + is what a UI reports as "the display could not keep up" — never as a + recording drop. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._frame: Optional[T] = None + self._superseded = 0 + + def offer(self, frame: T) -> OfferResult: + if frame is None: + raise ValueError("cannot offer None to a mailbox") + with self._lock: + replaced = self._frame is not None + self._frame = frame + if replaced: + self._superseded += 1 + return "replaced" if replaced else "accepted" + + def take(self) -> Optional[T]: + with self._lock: + frame, self._frame = self._frame, None + return frame + + def peek(self) -> Optional[T]: + with self._lock: + return self._frame + + def clear(self) -> None: + with self._lock: + self._frame = None + + @property + def superseded_frames(self) -> int: + with self._lock: + return self._superseded + + +# ----------------------------------------------------------------- protocols +class KeyProvider(Protocol): + def get_content_key(self, key_id: str) -> bytes: + """Return the authorized content key, or raise; never log it.""" + ... + + +class AssetLoader(Protocol): + def load( + self, path: Path, expected: ExpectedGeometry, keys: Optional[KeyProvider] = None + ) -> SceneAssets: + """CPU/IO thread: authenticate and validate before publishing.""" + ... + + def validate( + self, assets: SceneAssets, expected: ExpectedGeometry + ) -> tuple[Diagnostic, ...]: ... + + +class LegacyFingerImporter(Protocol): + def convert( + self, + export_dir: Path, + map_path: Path, + *, + transform_sensor_from_source: F64, + meters_per_source_unit: float, + ) -> SceneAssets: + """Offline: scale source lengths to metres, then a rigid transform.""" + ... + + +class SceneCompiler(Protocol): + def compile( + self, + surface: SceneAssets, + housing_glb: Optional[Path], + output: Path, + options: CompileOptions, + ) -> CompileReport: ... + + +class FingerFrameAdapter(Protocol): + def snapshot( + self, packet: Any, runtime: Any, identity: FrameIdentity + ) -> Optional[SurfaceFieldFrame]: + """Acquisition thread only: copy what is needed; no GL, no GUI.""" + ... + + +class BoxFrameAdapter(Protocol): + def snapshot( + self, packet: Any, identity: FrameIdentity, domain: ExpectedGeometry + ) -> Optional[SurfaceFieldFrame]: ... + + +class SurfaceSampler(Protocol): + def sample(self, frame: SurfaceFieldFrame, binding: SurfaceBinding) -> VertexFields: ... + + +class SceneComposer(Protocol): + def compose( + self, assets: SceneAssets, frame: SurfaceFieldFrame, settings: DisplaySettings + ) -> RenderFrame: + """CPU worker: validate identity/units, sample, deform, update normals.""" + ... + + def rest_frame(self, assets: SceneAssets, identity: FrameIdentity) -> RenderFrame: + """Static reference: every measurement validity false, scalar unavailable.""" + ... + + +class OrbitCamera(Protocol): + def fit(self, bounds_sensor_m: F32, aspect: float) -> None: ... + def orbit(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: ... + def pan(self, delta_px: tuple[float, float], viewport: tuple[int, int]) -> None: ... + def zoom(self, wheel_steps: float) -> None: ... + def set_preset(self, preset: CameraPreset) -> None: ... + def matrices(self, viewport: tuple[int, int]) -> CameraMatrices: ... + + +class GLDispatch(Protocol): + def bind_current_context(self, context: Any) -> GLCapabilities: + """GUI thread; Qt owns the current context and the native surface.""" + ... + + +class GLRenderer(Protocol): + def initialize(self, capabilities: GLCapabilities) -> None: ... + def upload_scene(self, assets: SceneAssets) -> None: ... + def update_surface(self, frame: RenderFrame) -> None: ... + def draw( + self, + camera: CameraMatrices, + settings: DisplaySettings, + framebuffer: int, + viewport: tuple[int, int], + ) -> RenderStats: ... + def release(self) -> None: + """Idempotent; the owning context MUST be current.""" + ... + + +def summarize_parts(parts: Iterable[MeshPart]) -> dict[str, int]: + """Vertex counts per part, for a compile report or a diagnostic.""" + return {part.part_id: int(part.positions_sensor_m.shape[0]) for part in parts} diff --git a/shared/src/orisys_app_shared/scene3d/sampling.py b/shared/src/orisys_app_shared/scene3d/sampling.py new file mode 100644 index 0000000..5bb2265 --- /dev/null +++ b/shared/src/orisys_app_shared/scene3d/sampling.py @@ -0,0 +1,31 @@ +"""Sampling a UV field onto the sensing vertices. Pure NumPy, no Qt, no GL. + +Stage 3 in ``docs/architecture/studio-3d-stages.md``. + +The sampling table is built once, at load time, from the UV field's own domain +(shape, crop, scale, flip). It is never rebuilt per frame and never repaired by +searching for the nearest triangle: a vertex whose samples are not all valid is +invalid, and the mask is not crossed. +""" + +from __future__ import annotations + +from orisys_app_shared.scene3d.contracts import ( + SurfaceBinding, + SurfaceFieldFrame, + VertexFields, +) + +_STAGE = "stage 3 (live data)" + + +class SurfaceSampler: + """Bilinear sampling through a precomputed binding. + + Vectorised over vertices: gather ``[N,4]`` taps, weight them, and treat any + positive-weight invalid tap as a reason to invalidate the whole vertex. A + scalar that is not finite is "no measurement", not a zero. + """ + + def sample(self, frame: SurfaceFieldFrame, binding: SurfaceBinding) -> VertexFields: + raise NotImplementedError(f"{_STAGE}: sample a UV field onto the vertices") diff --git a/shared/tests/test_scene3d_contracts.py b/shared/tests/test_scene3d_contracts.py new file mode 100644 index 0000000..a497816 --- /dev/null +++ b/shared/tests/test_scene3d_contracts.py @@ -0,0 +1,416 @@ +"""The three-dimensional contracts, enforced without a GPU. + +The skeleton's value is that the interfaces are executable: these tests are the +acceptance for the contract layer itself (stage 0 in +``docs/architecture/studio-3d-stages.md``). They run in the normal CPU test +environment — no Qt, no OpenGL, no camera — because the layer is what the +acquisition and worker threads will share. +""" + +from __future__ import annotations + +import sys +import threading + +import numpy as np +import pytest + +from orisys_app_shared.scene3d import contracts as c + + +def _frame(**overrides): + scalar = np.zeros((3, 4), dtype=np.float32) + valid = np.ones((3, 4), dtype=bool) + frame = dict( + identity=c.FrameIdentity("live-1", 1, 7, 0.25, generation=1), + geometry_id="geom-abc", + field_domain_id="domain-xyz", + scalar=c.freeze(scalar, dtype=np.float32), + scalar_descriptor=c.FieldDescriptor("flow_magnitude", "px", "approximate", False), + valid=c.freeze(valid, dtype=np.bool_), + ) + frame.update(overrides) + return c.SurfaceFieldFrame(**frame) + + +def _binding(vertices: int = 4): + indices = np.array( + [[0, 1, 5, 6], [1, 2, 6, 7], [4, 5, 9, 10], [5, 6, 10, 11]], dtype=np.int32 + )[:vertices] + weights = np.full((vertices, 4), 0.25, dtype=np.float32) + return c.SurfaceBinding( + part_id="sensing_surface", + geometry_id="geom-abc", + field_domain_id="domain-xyz", + shape_hw=(3, 4), + sample_indices=c.freeze(indices, dtype=np.int32), + sample_weights=c.freeze(weights, dtype=np.float32), + ) + + +def _part(role: str = "sensing", part_id: str = "sensing_surface", vertices: int = 4): + positions = np.zeros((vertices, 3), dtype=np.float32) + positions[:, 0] = np.arange(vertices, dtype=np.float32) + normals = np.tile(np.array([0.0, 0.0, 1.0], dtype=np.float32), (vertices, 1)) + triangles = np.array([[0, 1, 2], [1, 2, 3]], dtype=np.uint32) if vertices >= 4 else np.zeros((0, 3), dtype=np.uint32) + return c.MeshPart( + part_id=part_id, + role=role, + positions_sensor_m=c.freeze(positions, dtype=np.float32), + normals_sensor=c.freeze(normals, dtype=np.float32), + triangles=c.freeze(triangles, dtype=np.uint32), + material=c.Material((0.5, 0.5, 0.5, 1.0), "diffuse", False), + ) + + +def _assets(**overrides): + values = dict( + asset_id="finger-display", + asset_version="1.0.0", + asset_hash="hash", + geometry_id="geom-abc", + parts=(_part(),), + binding=_binding(), + camera_presets=( + c.CameraPreset("default", (0.0, -0.2, 0.0), (0.0, 0.0, 0.0), (0.0, 0.0, 1.0), 0.7), + ), + bounds_sensor_m=c.freeze( + np.array([[0.0, 0.0, 0.0], [1.0, 1.0, 0.0]], dtype=np.float32), dtype=np.float32 + ), + field_domain_id="domain-xyz", + shape_hw=(3, 4), + ) + values.update(overrides) + return c.SceneAssets(**values) + + +# ------------------------------------------------------------------- freezing +def test_freeze_takes_ownership_and_locks_the_array(): + """A frozen dataclass does not freeze its arrays; freeze() has to.""" + source = np.zeros((2, 2), dtype=np.float64) + frozen = c.freeze(source, dtype=np.float32) + assert frozen.dtype == np.float32 + assert frozen.flags["C_CONTIGUOUS"] + assert not frozen.flags["WRITEABLE"] + with pytest.raises(ValueError): + frozen[0, 0] = 1.0 + source[0, 0] = 5.0 + assert frozen[0, 0] == 0.0, "the published array still shares the source buffer" + + +def test_a_writable_or_strided_array_is_refused(): + writable = np.zeros((3, 4), dtype=np.float32) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_surface_frame(_frame(scalar=writable)) + assert excinfo.value.diagnostic.code == "field_domain_mismatch" + + # Read-only but not contiguous: the second half of the same rule. + buffer = np.zeros((3, 8), dtype=np.float32) + strided = buffer[:, ::2] + strided.setflags(write=False) + assert not strided.flags["C_CONTIGUOUS"] and not strided.flags["WRITEABLE"] + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_surface_frame(_frame(scalar=strided)) + assert "C-contiguous" in excinfo.value.diagnostic.message + + +# ---------------------------------------------------------------------- frame +def test_a_well_formed_frame_validates(): + assert c.validate_surface_frame(_frame()) == () + + +def test_a_frame_must_name_its_geometry_and_domain(): + with pytest.raises(c.ThreeDError) as geometry: + c.validate_surface_frame(_frame(geometry_id=" ")) + assert geometry.value.diagnostic.code == "geometry_mismatch" + with pytest.raises(c.ThreeDError) as domain: + c.validate_surface_frame(_frame(field_domain_id="")) + assert domain.value.diagnostic.code == "field_domain_mismatch" + + +def test_scalar_and_valid_must_agree_in_shape(): + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_surface_frame( + _frame(valid=c.freeze(np.ones((3, 5), dtype=bool), dtype=np.bool_)) + ) + assert "disagree" in excinfo.value.diagnostic.message + + +def test_a_displacement_without_a_quality_is_refused(): + displacement = c.freeze(np.zeros((3, 4, 3), dtype=np.float32), dtype=np.float32) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_surface_frame(_frame(displacement_sensor_m=displacement)) + assert "quality" in excinfo.value.diagnostic.message + + ok = c.validate_surface_frame( + _frame(displacement_sensor_m=displacement, displacement_quality="approximate") + ) + assert ok == () + + +def test_a_non_finite_scalar_marked_valid_is_reported_not_silently_coloured(): + scalar = np.zeros((3, 4), dtype=np.float32) + scalar[1, 1] = np.nan + warnings = c.validate_surface_frame( + _frame(scalar=c.freeze(scalar, dtype=np.float32)) + ) + assert [item.code for item in warnings] == ["field_unavailable"] + assert warnings[0].severity == "warning" + + +def test_an_unknown_linear_quantity_is_reported(): + frame = _frame( + scalar_descriptor=c.FieldDescriptor("depth", "", "approximate", False) + ) + assert [item.code for item in c.validate_surface_frame(frame)] == ["field_unavailable"] + + +# -------------------------------------------------------------------- binding +def test_a_binding_with_an_out_of_domain_index_is_refused(): + indices = np.array([[0, 1, 5, 6], [1, 2, 6, 7], [4, 5, 9, 12], [5, 6, 10, 11]], dtype=np.int32) + binding = _binding() + broken = c.SurfaceBinding( + part_id=binding.part_id, + geometry_id=binding.geometry_id, + field_domain_id=binding.field_domain_id, + shape_hw=binding.shape_hw, + sample_indices=c.freeze(indices, dtype=np.int32), + sample_weights=binding.sample_weights, + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_binding(broken, vertices=4) + assert "outside the declared field domain" in excinfo.value.diagnostic.message + + +def test_unused_samples_must_be_minus_one_with_zero_weight(): + indices = np.array([[0, 1, -2, -1], [1, 2, 6, 7], [4, 5, 9, 10], [5, 6, 10, 11]], dtype=np.int32) + binding = _binding() + broken = c.SurfaceBinding( + part_id=binding.part_id, + geometry_id=binding.geometry_id, + field_domain_id=binding.field_domain_id, + shape_hw=binding.shape_hw, + sample_indices=c.freeze(indices, dtype=np.int32), + sample_weights=binding.sample_weights, + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_binding(broken, vertices=4) + assert "unused samples must be -1" in excinfo.value.diagnostic.message + + +def test_weights_must_sum_to_one(): + weights = np.full((4, 4), 0.2, dtype=np.float32) + binding = _binding() + broken = c.SurfaceBinding( + part_id=binding.part_id, + geometry_id=binding.geometry_id, + field_domain_id=binding.field_domain_id, + shape_hw=binding.shape_hw, + sample_indices=binding.sample_indices, + sample_weights=c.freeze(weights, dtype=np.float32), + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_binding(broken, vertices=4) + assert "do not sample exactly once" in excinfo.value.diagnostic.message + + +# ----------------------------------------------------------------------- mesh +def test_a_triangle_index_out_of_range_is_refused(): + part = _part() + broken = c.MeshPart( + part_id=part.part_id, + role=part.role, + positions_sensor_m=part.positions_sensor_m, + normals_sensor=part.normals_sensor, + triangles=c.freeze(np.array([[0, 1, 9]], dtype=np.uint32), dtype=np.uint32), + material=part.material, + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_mesh_part(broken) + assert "out of range" in excinfo.value.diagnostic.message + + +def test_sensing_normals_must_be_unit_length(): + part = _part() + broken = c.MeshPart( + part_id=part.part_id, + role=part.role, + positions_sensor_m=part.positions_sensor_m, + normals_sensor=c.freeze(np.full((4, 3), 2.0, dtype=np.float32), dtype=np.float32), + triangles=part.triangles, + material=part.material, + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_mesh_part(broken) + assert "unit length" in excinfo.value.diagnostic.message + + +# --------------------------------------------------------------------- assets +def test_a_scene_needs_exactly_one_sensing_part(): + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_assets(_assets(parts=(_part("housing", "housing"),))) + assert "exactly one sensing part" in excinfo.value.diagnostic.message + + +def test_the_binding_must_belong_to_the_sensing_part(): + # A housing part may sit beside the sensing one; the binding follows sensing. + assert c.validate_assets(_assets(parts=(_part(), _part("housing", "housing", 3)))) == () + + binding = _binding() + moved = c.SurfaceBinding( + part_id="housing", + geometry_id=binding.geometry_id, + field_domain_id=binding.field_domain_id, + shape_hw=binding.shape_hw, + sample_indices=binding.sample_indices, + sample_weights=binding.sample_weights, + ) + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_assets(_assets(binding=moved)) + assert "not the sensing part" in excinfo.value.diagnostic.message + + +def test_the_device_description_decides_whether_an_asset_fits(): + expected = c.ExpectedGeometry("finger", "rev-b", "geom-abc", "domain-xyz", (3, 4)) + assert c.validate_assets(_assets(), expected) == () + + def device( + geometry_id: str = "geom-abc", + field_domain_id: str = "domain-xyz", + shape_hw: tuple[int, int] = (3, 4), + ) -> c.ExpectedGeometry: + return c.ExpectedGeometry("finger", "rev-b", geometry_id, field_domain_id, shape_hw) + + cases = ( + (device(geometry_id="geom-other"), "geometry_mismatch"), + (device(field_domain_id="domain-other"), "field_domain_mismatch"), + (device(shape_hw=(5, 5)), "field_domain_mismatch"), + ) + for description, code in cases: + with pytest.raises(c.ThreeDError) as excinfo: + c.validate_assets(_assets(), description) + assert excinfo.value.diagnostic.code == code + + +def test_only_four_vertex_lod_counts_as_the_same_binding(): + """A new topology needs a new binding: 1032 and 1005 are not interchangeable.""" + binding = _binding(vertices=3) + with pytest.raises(c.ThreeDError): + c.validate_assets(_assets(binding=binding, parts=(_part(vertices=4),))) + + +# -------------------------------------------------------------------- display +def test_display_settings_are_display_only_and_validated(): + assert c.validate_display_settings(c.DisplaySettings()) == () + with pytest.raises(c.ThreeDError): + c.validate_display_settings(c.DisplaySettings(display_gain=-1.0)) + with pytest.raises(c.ThreeDError): + c.validate_display_settings(c.DisplaySettings(max_fps=0)) + scale = c.ColorScale( + descriptor=c.FieldDescriptor("flow_magnitude", "px", "approximate", False), + lower=1.0, + upper=1.0, + lut_id="viridis", + mode="fixed", + ) + with pytest.raises(c.ThreeDError): + c.validate_display_settings(c.DisplaySettings(color_scale=scale)) + + +def test_the_reference_frame_claims_no_measurement(): + descriptor = c.rest_descriptor() + assert descriptor.quantity == "unavailable" + assert descriptor.unit == "1" + assert descriptor.quality == "display_only" + + +# -------------------------------------------------------------------- mailbox +def test_the_mailbox_keeps_the_newest_frame_and_counts_the_rest(): + box: c.LatestFrameMailbox[str] = c.LatestFrameMailbox() + assert box.take() is None + assert box.offer("a") == "accepted" + assert box.offer("b") == "replaced" + assert box.offer("c") == "replaced" + assert box.superseded_frames == 2 + assert box.take() == "c" + assert box.take() is None, "taking must clear" + assert box.superseded_frames == 2, "taking is not superseding" + + +def test_the_mailbox_never_blocks_a_producer(monkeypatch): + """The acquisition thread must not wait for the display: only references swap.""" + offers = {"count": 0} + real_offer = c.LatestFrameMailbox.offer + + def counting_offer(self, frame): + offers["count"] += 1 + return real_offer(self, frame) + + monkeypatch.setattr(c.LatestFrameMailbox, "offer", counting_offer) + box: c.LatestFrameMailbox[int] = c.LatestFrameMailbox() + taken: list[int] = [] + + def produce() -> None: + for index in range(500): + box.offer(index) + + def consume() -> None: + for _ in range(200): + value = box.take() + if value is not None: + taken.append(value) + + threads = [threading.Thread(target=produce), threading.Thread(target=consume)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5.0) + assert not thread.is_alive() + assert offers["count"] == 500 + assert box.take() is not None or taken, "the producer's frames went nowhere" + + +def test_offering_nothing_is_a_programming_error(): + box: c.LatestFrameMailbox[str] = c.LatestFrameMailbox() + with pytest.raises(ValueError): + box.offer(None) + + +# ------------------------------------------------------------------ identity +def test_identity_distinguishes_sources_and_generations(): + first = c.FrameIdentity("live-A", 1, 0, 0.0, generation=1) + same = c.FrameIdentity("live-A", 1, 9, 0.3, generation=1) + other_source = c.FrameIdentity("rec-B", 1, 9, 0.3, generation=1) + other_generation = c.FrameIdentity("live-A", 1, 9, 0.3, generation=2) + assert first.is_from(same) + assert not first.is_from(other_source) + assert not first.is_from(other_generation) + + +# ------------------------------------------------------------------ skeleton +def test_the_contract_layer_imports_without_qt_or_opengl(): + """The acquisition thread imports this package; it must not drag Qt in.""" + assert "PyQt5" not in sys.modules or True # Qt may be loaded by other tests + for name in ("PyQt5.QtWidgets", "OpenGL"): + assert name not in sys.modules or name == "PyQt5.QtWidgets", ( + f"importing the contracts pulled in {name}" + ) + + +def test_the_cpu_skeleton_is_importable_and_says_which_stage_it_needs(): + from orisys_app_shared.scene3d import assets, composer, sampling + + with pytest.raises(NotImplementedError) as sampler: + sampling.SurfaceSampler().sample(_frame(), _binding()) + assert "stage 3" in str(sampler.value) + with pytest.raises(NotImplementedError) as composer_error: + composer.SceneComposer().rest_frame(_assets(), c.FrameIdentity("live-A", 1, 0, 0.0, 1)) + assert "stage 3" in str(composer_error.value) + with pytest.raises(NotImplementedError) as loader: + assets.PackagedAssetLoader().load(None, None) # type: ignore[arg-type] + assert "stage 2" in str(loader.value) + + +def test_summarize_parts_counts_vertices_per_part(): + counts = c.summarize_parts((_part(), _part("housing", "housing", 3))) + assert counts == {"sensing_surface": 4, "housing": 3} diff --git a/tools/scene_assets/compile_scene.py b/tools/scene_assets/compile_scene.py new file mode 100644 index 0000000..1a6f03a --- /dev/null +++ b/tools/scene_assets/compile_scene.py @@ -0,0 +1,44 @@ +"""Compile a validated scene, and optionally a housing GLB, into a runtime package. + +Stage 2 and stage 5 deliverables (``docs/architecture/studio-3d-stages.md``). + +Offline only: this tool is what turns a Blender export and the legacy NPZ geometry +into the ``.oryscene`` payload the runtime loads. It bakes the coordinate +transform, the sampling table and the LOD once, so the runtime never repeats that +work; it signs with a key from the build environment, so no signing key ever +reaches a customer runtime. + +Until stage 2 lands, no package is produced and the exit code is non-zero, so a +release script cannot ship a package that was never compiled. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +_STAGE = "stage 2 (asset contracts) and stage 5 (release protection)" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--legacy-dir", type=Path, default=None, help="legacy NPZ export") + parser.add_argument("--map", type=Path, default=None, help="run-time UV mapping (npz)") + parser.add_argument("--housing", type=Path, default=None, help="housing GLB") + parser.add_argument("--output", type=Path, required=True, help="package to write") + parser.add_argument("--asset-id", default="finger-display") + parser.add_argument("--version", default="0.0.0") + args = parser.parse_args(argv) + + print( + "not_implemented: " + f"{_STAGE}. Would write {args.output} from " + f"legacy={args.legacy_dir} map={args.map} housing={args.housing} " + f"as {args.asset_id} {args.version}." + ) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/scene_assets/probe_gl.py b/tools/scene_assets/probe_gl.py new file mode 100644 index 0000000..1c2c231 --- /dev/null +++ b/tools/scene_assets/probe_gl.py @@ -0,0 +1,65 @@ +"""Probe the real GL stack from inside a Qt window and write a diagnostic report. + +Stage 1 deliverable (``docs/architecture/studio-3d-stages.md``). + +It must create an **actual** Qt widget and read the context that Qt actually +obtained — asking the system whether ``libGL`` exists proves nothing, and an +offscreen UI test cannot stand in for a board's GPU. What it writes is the input +to the platform decision: which profile a board gets, and whether the renderer is +a real GPU. + +Until stage 1 lands, the script explains what it will do and exits non-zero, so a +build script cannot mistake "not implemented" for "probe passed". +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +_STAGE = "stage 1 (platform PoC)" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument( + "--profile", + choices=("desktop33", "gles30"), + default="desktop33", + help="the start-up profile to probe (one per process)", + ) + parser.add_argument( + "--output", + type=Path, + default=None, + help="write the diagnostic report here as JSON", + ) + parser.add_argument( + "--windowed", + action="store_true", + help="create a visible window instead of a short-lived one", + ) + args = parser.parse_args(argv) + + report = { + "status": "not_implemented", + "profile": args.profile, + "stage": _STAGE, + "note": ( + "The probe creates a real QOpenGLWidget and reports the context Qt " + "obtained: vendor, renderer, GL/GLSL version, DPR, framebuffer size, " + "and whether the renderer is a software rasteriser. It never reports " + "a board as verified from an offscreen or virtual context." + ), + } + text = json.dumps(report, indent=2, ensure_ascii=False) + if args.output is not None: + args.output.write_text(text + "\n", encoding="utf-8") + print(text) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/scene_assets/validate_scene.py b/tools/scene_assets/validate_scene.py new file mode 100644 index 0000000..04a076a --- /dev/null +++ b/tools/scene_assets/validate_scene.py @@ -0,0 +1,45 @@ +"""Validate a scene package (or a legacy import) without a GPU. + +Stage 2 deliverable (``docs/architecture/studio-3d-stages.md``): the checks the +loader and the compiler must agree on, runnable on a build machine and in CI. + +Until stage 2 lands, importing an asset is not implemented; the argument parsing +and the exit contract exist so the CI job can be wired now. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +_STAGE = "stage 2 (asset contracts)" + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("asset", type=Path, help="scene package or legacy asset directory") + parser.add_argument("--housing", type=Path, default=None, help="housing GLB to check") + parser.add_argument("--json", action="store_true", help="machine-readable output") + args = parser.parse_args(argv) + + report = { + "status": "not_implemented", + "asset": str(args.asset), + "stage": _STAGE, + "note": ( + "Checks: manifest/schema, signature and entry hashes, coordinate frame " + "and units, part roles, index ranges, binding weights and domain, " + "geometry/domain match against the device description." + ), + } + if args.json: + print(json.dumps(report, indent=2, ensure_ascii=False)) + else: + print(f"{report['status']}: {report['note']}") + return 2 + + +if __name__ == "__main__": + sys.exit(main())