"""Studio 3D v1 design contracts, 2026-09-14. Specification only: split into the proposed modules when implementing. No rendering, asset loading, decryption or SDK changes are implemented here. All published NumPy arrays must be C-contiguous and read-only (frozen dataclasses alone do not freeze ndarray contents). Shapes are contractual, not type-checked by NDArray. See studio-3d-design.md for threading and validation requirements. """ from dataclasses import dataclass from pathlib import Path from typing import Generic, Literal, Mapping, Protocol, TypeAlias, TypeVar from concurrent.futures import Executor import numpy as np from numpy.typing import NDArray from PyQt5.QtCore import QEvent, pyqtBoundSignal from PyQt5.QtGui import QMouseEvent, QOpenGLContext, QWheelEvent from PyQt5.QtWidgets import QOpenGLWidget, QWidget from orisys_app_shared.acquisition.models import SdkPacket from orisys.finger.runtime import FingerRuntime F32: TypeAlias = NDArray[np.float32] F64: TypeAlias = NDArray[np.float64] I32: TypeAlias = NDArray[np.int32] U32: TypeAlias = NDArray[np.uint32] Mask: TypeAlias = NDArray[np.bool_] Role: TypeAlias = Literal['sensing', 'housing', 'reference'] Quality: TypeAlias = Literal['calibrated', 'approximate', 'display_only'] Profile: TypeAlias = Literal['desktop33', 'gles30'] OfferResult: TypeAlias = Literal['accepted', 'replaced', 'stale', 'disabled', 'invalid'] SceneState: TypeAlias = Literal['disabled', 'loading', 'ready', 'live', 'no_data', 'suspended', 'unavailable'] @dataclass(frozen=True) class Diagnostic: code: str # Stable machine-readable code, e.g. geometry_mismatch. severity: Literal['info', 'warning', 'error'] message: str # Human-readable, never contains keys or decrypted asset data. details: Mapping[str, str] class ThreeDError(Exception): """Boundary exception; GUI converts to diagnostic and keeps 2D operational.""" diagnostic: Diagnostic def __init__(self, diagnostic: Diagnostic) -> None: ... @dataclass(frozen=True) class FrameIdentity: source_id: str # Live stream UUID or recording UUID; not just a filename. session_id: int frame_id: int t_rel_s: float # Acquisition-relative monotonic time, never wall-clock for ordering. generation: int # Increment on activation/source/geometry change. request_id: int # Replay cursor request; 0 for ordinary live frames. @dataclass(frozen=True) class FieldDescriptor: quantity: str # Examples: flow_magnitude, signed_normal_displacement, depth_response. unit: str # px, m, or explicit algorithm-response unit; never inferred from shape. quality: Quality signed: bool calibration_id: str | None @dataclass(frozen=True) class SurfaceFieldFrame: identity: FrameIdentity geometry_id: str field_domain_id: str # Includes geometry, UV shape, crop/scale/flip convention. scalar: F32 # [H,W]; measured/displayed quantity, not RGB. scalar_descriptor: FieldDescriptor valid: Mask # [H,W]; false or non-finite scalar means unavailable, not zero. displacement_sensor_m: F32 | None # [H,W,3]; base displacement BEFORE display gain. displacement_quality: Quality | None diagnostic: Diagnostic | None @dataclass(frozen=True) class Material: base_color_linear: tuple[float, float, float, float] shading: Literal['unlit_scalar', 'diffuse'] double_sided: bool @dataclass(frozen=True) class MeshPart: part_id: str role: Role positions_sensor_m: F32 # [N,3], baked coordinates. normals_sensor: F32 # [N,3], finite unit normals. triangles: U32 # [M,3], in [0,N). material: Material lod_id: str @dataclass(frozen=True) class SurfaceBinding: part_id: str # The sensing part only. geometry_id: str field_domain_id: str shape_hw: tuple[int, int] sample_indices: I32 # [N,4], row-major y*W+x; unused index=-1 with weight=0. sample_weights: F32 # [N,4], nonnegative, used weights sum to one. @dataclass(frozen=True) class CameraPreset: 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: sensor_model: str hardware_revision: str geometry_id: str field_domain_id: str shape_hw: tuple[int, int] @dataclass(frozen=True) class SceneAssets: asset_id: str asset_version: str asset_hash: str # Digest of canonical manifest/content, not encryption nonce. geometry_id: str parts: tuple[MeshPart, ...] binding: SurfaceBinding # v1 supports exactly one sensing part. camera_presets: tuple[CameraPreset, ...] bounds_sensor_m: F32 # [2,3]: min, max. @dataclass(frozen=True) class ColorScale: descriptor: FieldDescriptor lower: float # Finite; lower < upper, in descriptor.unit. upper: float lut_id: str # Approved sequential/diverging LUT. mode: Literal['fixed', 'automatic'] @dataclass(frozen=True) class DisplaySettings: display_gain: float # Finite positive; affects only dynamic positions. deform_enabled: bool color_scale: ColorScale housing_visible: bool max_fps: int # Display limit; never forwarded as camera/recording FPS. @dataclass(frozen=True) class VertexFields: scalar: F32 # [N], unscaled scientific value. valid: Mask # [N]; any positive-weight invalid sample invalidates the vertex. displacement_sensor_m: F32 | None # [N,3]. @dataclass(frozen=True) class RenderFrame: 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], valid even at vertices excluded by mask. 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 superseded_frames: int compose_ms: float upload_ms: float # CPU submission time unless a separate GPU timer is used. paint_ms: float estimated_gpu_bytes: int class KeyProvider(Protocol): def get_content_key(self, key_id: str) -> bytes: """Return authorized content key, or raise ThreeDError; never log it.""" ... class AssetLoader(Protocol): def load(self, path: Path, expected: ExpectedGeometry, keys: KeyProvider | None = None) -> SceneAssets: """CPU/IO thread. Authenticate and validate before publishing assets.""" ... 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 meters, then apply a rigid transform. transform translation is meters; its rotation contains no extra scale. """ ... @dataclass(frozen=True) class CompileOptions: asset_id: str version: str expected: ExpectedGeometry transform_sensor_from_glb: F64 # [4,4]; GLB lengths are already meters. max_surface_vertices: int max_housing_triangles: int signing_key_reference: str # Build environment reference, not private key bytes. encryption_key_id: str | None @dataclass(frozen=True) class CompileReport: output: Path asset_hash: str diagnostics: tuple[Diagnostic, ...] part_vertex_counts: Mapping[str, int] part_triangle_counts: Mapping[str, int] class SceneCompiler(Protocol): def compile(self, surface: SceneAssets, housing_glb: Path | None, output: Path, options: CompileOptions) -> CompileReport: ... class FingerFrameAdapter(Protocol): def snapshot(self, packet: SdkPacket, runtime: FingerRuntime, identity: FrameIdentity) -> SurfaceFieldFrame | None: """Acquisition thread only. Copy/own one frame; no GL and no GUI calls.""" ... class BoxFrameAdapter(Protocol): def snapshot(self, packet: SdkPacket, identity: FrameIdentity, domain: ExpectedGeometry) -> SurfaceFieldFrame | None: ... 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; all measurement validity false, scalar unavailable.""" ... T = TypeVar('T') class LatestFrameMailbox(Generic[T]): def __init__(self) -> None: ... def offer(self, frame: T) -> OfferResult: """Thread-safe capacity-one replacement, constant-time under the lock.""" ... def take(self) -> T | None: ... def clear(self) -> None: ... @property def superseded_frames(self) -> int: ... 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: ... def configure_surface_format(profile: Profile) -> None: """Before QApplication and before importing a platform-specific GL binding.""" ... class GLDispatch(Protocol): def bind_current_context(self, context: QOpenGLContext) -> GLCapabilities: """GUI thread; Qt owns the current context and 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; correct owning context MUST be current.""" ... class Sensor3DWidget(QOpenGLWidget): capabilitiesChanged: pyqtBoundSignal # Payload: GLCapabilities. diagnosticRaised: pyqtBoundSignal # Payload: Diagnostic. renderStatsChanged: pyqtBoundSignal # Payload: RenderStats; rate-limited. def __init__(self, renderer: GLRenderer, camera: OrbitCamera, parent: QWidget | None = None) -> None: ... def set_scene(self, assets: SceneAssets) -> None: """GUI: retain CPU data, schedule upload when context is current.""" ... def present(self, frame: RenderFrame) -> None: ... def set_settings(self, settings: DisplaySettings) -> None: ... def initializeGL(self) -> None: ... def paintGL(self) -> None: ... def resizeGL(self, w: int, h: int) -> None: ... def release_resources(self) -> None: ... def mousePressEvent(self, event: QMouseEvent) -> None: ... def mouseMoveEvent(self, event: QMouseEvent) -> None: ... def wheelEvent(self, event: QWheelEvent) -> None: ... def hideEvent(self, event: QEvent) -> None: ... def showEvent(self, event: QEvent) -> None: ... class ThreeDCoordinator: def __init__(self, widget: Sensor3DWidget, composer: SceneComposer, executor: Executor, max_fps: int = 30) -> None: ... def activate(self, assets: SceneAssets, source_id: str) -> int: ... def submit(self, frame: SurfaceFieldFrame) -> OfferResult: ... def tick(self) -> None: ... def set_visible(self, visible: bool) -> None: ... def set_settings(self, settings: DisplaySettings) -> None: ... def close(self) -> None: ... @property def state(self) -> SceneState: ... @property def generation(self) -> int: ... class ThreeDPanel(QWidget): """UI-only controls; no asset IO, SDK reads, or direct GL calls.""" def __init__(self, coordinator: ThreeDCoordinator, widget: Sensor3DWidget, parent: QWidget | None = None) -> None: ... def set_diagnostic(self, diagnostic: Diagnostic | None) -> None: ... def set_state(self, state: SceneState) -> None: ... def set_render_stats(self, stats: RenderStats) -> None: ...