orisys-app/docs/architecture/studio-3d-contracts.pyi

"""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 Callable, 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', 'stale']


@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
    computed_frame_id: int | None  # Frame the array was computed for; None = unstated.


@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  # Metre-scaled shape identity; '' when the device cannot state it.
    field_domain_id: str
    shape_hw: tuple[int, int]
    source_geometry_id: str  # Unit-free surface identity a device CAN reproduce.


@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.
    source_geometry_id: str  # Unit-free source geometry this asset was built from.
    compatible_sensor_model: str
    compatible_hardware_revisions: tuple[str, ...]  # '*' means any revision.


@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, ...]
    computed_frame_id: int | None  # Carried through so the display side re-checks it.
    compose_ms: float  # This frame's own compose cost; the shared counter is rewritten.


@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 SceneCapabilities:
    """What the *data source* can provide — not what the pane can draw."""
    quantity: str
    unit: str
    displacement_available: bool  # False for the current adapters: they colour in place.
    displacement_quality: Quality | None
    source_kind: str  # live, recompute, replay.


@dataclass(frozen=True)
class RenderStats:
    displayed_frames: int  # Unique frames presented, by identity (not paint calls).
    superseded_frames: int
    paint_calls: int  # Every paintGL, so a slow camera move is not a slow pipeline.
    compose_ms: float  # The frame's own measured cost (RenderFrame.compose_ms).
    upload_ms: float  # CPU submission time unless a separate GPU timer is used.
    paint_ms: float
    estimated_gpu_bytes: int
    scale_lower: float  # The colour bounds the shader was actually given.
    scale_upper: float
    scale_mode: str  # 'automatic' or 'fixed'.


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, *,
                 clock: Callable[[], float] = ..., stale_grace_s: float = 1.0) -> None: ...
    def activate(self, assets: SceneAssets, source_id: str) -> int: ...
    def submit(self, frame: SurfaceFieldFrame) -> OfferResult: ...
    def note_field_failure(self, item: Diagnostic | None = None) -> None: ...
    def tick(self) -> None: ...
    def set_visible(self, visible: bool) -> None: ...
    def set_settings(self, settings: DisplaySettings) -> None: ...
    def show_rest(self) -> None: ...
    def close(self) -> None: ...
    @property
    def state(self) -> SceneState: ...
    @property
    def generation(self) -> int: ...
    @property
    def epoch(self) -> int: ...
    def last_presented(self) -> tuple[FrameIdentity, float] | None: ...


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: ...