"""Top-level engine entry point."""
import logging
from collections.abc import Callable
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any
import numpy as np
import vulkan as vk
if TYPE_CHECKING:
from .engine_surface import EngineSurface
from .gpu.context import GPUContext
from .materials.texture import TextureManager
from .renderer.forward import Renderer
from .renderer.gpu_batch import GPUBatch
from .renderer.mesh_registry import MeshRegistry
from .renderer.outline_pass import OutlinePass
def _assert_engine_surface(e: Engine) -> EngineSurface:
"""Static conformance: ``Engine`` must satisfy the ``SceneAdapter`` contract."""
return e
from ._engine_init import (
create_depth_resources as _create_depth_resources,
)
from ._engine_init import (
create_device_objects as _create_device_objects,
)
from ._engine_init import (
create_framebuffers as _create_framebuffers,
)
from ._engine_init import (
create_instance_and_surface as _create_instance_and_surface,
)
from ._engine_init import (
destroy_depth_resources as _destroy_depth_resources,
)
from ._engine_init import (
destroy_framebuffers as _destroy_framebuffers,
)
from ._engine_init import (
init_vulkan as _init_vulkan,
)
from ._engine_init import (
recreate_swapchain as _recreate_swapchain,
)
from ._frame_capture import capture_swapchain_frame
from .gpu.capabilities import RenderCapabilities
from .gpu.commands import CommandContext
from .gpu.descriptors import (
allocate_descriptor_set,
create_descriptor_pool,
create_ssbo_layout,
create_texture_descriptor_layout,
create_texture_descriptor_pool,
write_ssbo_descriptor,
write_texture_descriptor,
)
from .gpu.memory import (
create_buffer,
create_sampler,
upload_numpy,
)
from .gpu.pipeline import create_shader_module
from .gpu.swapchain import Swapchain
from .gpu.sync import FrameSync
from .gpu.timestamp_pool import TimestampPool
from .materials.shader_compiler import compile_shader
from .picking.pick_pass import PickPass
from .platform import resolve_backend
from .renderer.render_target import RenderTarget
from .types import (
MAX_TEXTURES,
SHADER_DIR,
MeshHandle,
_VkCommandBuffer,
_VkDebugUtilsMessengerEXT,
_VkDescriptorPool,
_VkDescriptorSet,
_VkDescriptorSetLayout,
_VkDevice,
_VkDeviceMemory,
_VkFramebuffer,
_VkImage,
_VkImageView,
_VkInstance,
_VkPhysicalDevice,
_VkPipeline,
_VkPipelineLayout,
_VkQueue,
_VkRenderPass,
_VkSampler,
_VkShaderModule,
_VkSurfaceKHR,
)
__all__ = ["CubemapHandle", "Engine"]
log = logging.getLogger(__name__)
# VkFormat int -> coarse device-feature key, resolved lazily (vulkan ints only
# touched at call time). Maps a block-compressed format to the optional Vulkan
# feature that must be enabled to sample it: BC -> ASTC-LDR -> ETC2 families.
_COMPRESSION_FAMILY_BY_INT: dict[int, str] = {}
_COMPRESSION_FAMILY_NAMES: dict[str, str] = {
"VK_FORMAT_ASTC_4x4_UNORM_BLOCK": "texture_compression_astc_ldr",
"VK_FORMAT_ASTC_4x4_SRGB_BLOCK": "texture_compression_astc_ldr",
"VK_FORMAT_ETC2_R8G8B8A8_UNORM_BLOCK": "texture_compression_etc2",
"VK_FORMAT_ETC2_R8G8B8A8_SRGB_BLOCK": "texture_compression_etc2",
}
def _compression_family_feature(vk_format: int) -> str | None:
"""Coarse device-feature key for a block-compressed format, or None.
Returns ``texture_compression_astc_ldr`` / ``_etc2`` for the ASTC/ETC2 4x4
formats; defaults to ``texture_compression_bc`` for any other (BC) format,
preserving the historical BC-only gate. None is never returned for the
formats the UASTC target probe can produce.
"""
if not _COMPRESSION_FAMILY_BY_INT:
for name, feature in _COMPRESSION_FAMILY_NAMES.items():
try:
_COMPRESSION_FAMILY_BY_INT[int(getattr(vk, name))] = feature
except AttributeError:
continue
return _COMPRESSION_FAMILY_BY_INT.get(int(vk_format), "texture_compression_bc")
[docs]
@dataclass(frozen=True)
class CubemapHandle:
"""Opaque handle to a GPU cubemap returned by :meth:`Engine.load_cubemap`.
Pass to :meth:`~simvx.graphics.renderer.forward.Renderer.set_skybox`
to install it as the scene skybox / IBL source. The engine takes ownership
of the underlying Vulkan resources and destroys them at shutdown.
"""
view: Any # _VkImageView (cube)
sampler: Any # _VkSampler
image: Any # _VkImage (6 layers)
memory: Any # _VkDeviceMemory
[docs]
class DeviceState(Enum):
"""Lifecycle of the Vulkan logical device, mirroring the web GpuContext.
``BOOTING`` before the first device is created; ``READY`` while rendering;
``LOST`` when ``VK_ERROR_DEVICE_LOST`` is observed; ``RECOVERING`` while a
rebuild is in flight; ``DESTROYED`` after shutdown. The legacy
``_device_lost`` bool remains the per-frame submission gate; this state is
the coarser machine the recovery path drives.
"""
BOOTING = "booting"
READY = "ready"
LOST = "lost"
RECOVERING = "recovering"
DESTROYED = "destroyed"
[docs]
class Engine:
"""Graphics engine: owns the window, GPU context, and render loop."""
def __init__(
self,
width: int = 1280,
height: int = 720,
title: str = "SimVX",
backend: str | None = None,
renderer: str = "deferred",
max_textures: int = MAX_TEXTURES,
visible: bool = True,
vsync: bool = False,
target_fps: int | None = None,
) -> None:
self.width = width
self.height = height
self.title = title
self._backend_name = backend
self._renderer_name = renderer
self._max_textures = max_textures
self._visible = visible
self._vsync = vsync
self._target_fps = target_fps
self._running = False
self._last_image_index: int = 0
# Capture-before-present arming. When armed, ``_draw_frame`` reads the
# swapchain image back WHILE IT IS STILL ACQUIRED (after the render submit,
# before ``vkQueuePresentKHR``) and stashes the pixels here. Presentable
# images may only be used between acquire and present, so reading back after
# present (the pull-based ``capture_frame``) trips a validation error; this
# keeps the headless capture path acquire-before-transition clean. Zero cost
# unless a frame is explicitly armed.
self._capture_armed: bool = False
self._captured_frame: np.ndarray | None = None
# Device-loss state. VK_ERROR_DEVICE_LOST (GPU TDR / driver crash / device
# removed) invalidates every handle from this device. We flip this terminal
# flag, stop the loop, and tear down cleanly rather than crashing mid-frame
# or hanging on a dead fence: the desktop mirror of the web backend's 'lost'
# state + frame gate + deterministic teardown. ``on_device_lost`` lets the
# app react (e.g. surface a dialog). Live device re-creation (rebuild device
# + all GPU resources) is a separate subsystem, intentionally not attempted.
self._device_lost = False
# Where the most recent device loss was observed (Vulkan call site), kept
# for the recovery driver + the terminal ``on_device_lost`` fallback.
self._device_lost_where: str = "unknown"
# Coarse device lifecycle state (mirrors the web GpuContext). BOOTING
# until init_vulkan completes, then READY; LOST on device loss. Drives
# the recovery path; ``_device_lost`` stays the per-frame submit gate.
self._device_state: DeviceState = DeviceState.BOOTING
self.on_device_lost: Callable[[str], None] | None = None
# Thin G-buffer active state. False by default: the HDR target
# has a single colour attachment and every HDR-pass pipeline is built with
# one blend attachment (byte-identical to today). A consumer (SSAO real
# normals, later SSR/SSGI/decals) flips this via the renderer, which then
# rebuilds the HDR target + all HDR-pass pipelines with the 2-attachment
# MRT variant. Read by post_process (target) and every HDR pipeline builder.
self.gbuffer_active: bool = False
# Vulkan handles (populated in _init_vulkan)
self._instance: _VkInstance | None = None
self._debug_messenger: _VkDebugUtilsMessengerEXT | None = None
self._surface: _VkSurfaceKHR | None = None
self._physical_device: _VkPhysicalDevice | None = None
self._capabilities: RenderCapabilities | None = None
self._device: _VkDevice | None = None
self._graphics_queue: _VkQueue | None = None
self._present_queue: _VkQueue | None = None
# Dedicated-queue handles (None on a single-universal-family GPU).
self._compute_queue: _VkQueue | None = None
self._transfer_queue: _VkQueue | None = None
self._async_compute: Any = None # AsyncComputeScheduler (set in init_vulkan)
self._swapchain: Swapchain | None = None
self._render_pass: _VkRenderPass | None = None
self._pipeline: _VkPipeline | None = None
self._pipeline_layout: _VkPipelineLayout | None = None
self._cmd_ctx: CommandContext | None = None
self._cmd_buffers: list[_VkCommandBuffer] = []
self._sync: FrameSync | None = None
self._framebuffers: list[_VkFramebuffer] = []
self._window: Any = None
self._content_scale: tuple[float, float] = (1.0, 1.0) # HiDPI pixel scaling
self._resize_pending: bool = False
self._vert_module: _VkShaderModule | None = None
self._frag_module: _VkShaderModule | None = None
# Depth buffer
self._depth_image: _VkImage | None = None
self._depth_memory: _VkDeviceMemory | None = None
self._depth_view: _VkImageView | None = None
self._use_depth: bool = False
# Render callbacks (set via run())
self._render_callback: Callable[[_VkCommandBuffer, tuple[int, int]], None] | None = None
self._pre_render_callback: Callable[[_VkCommandBuffer], None] | None = None
# Subsystems
self._mesh_registry: MeshRegistry | None = None
self._gpu_batch: GPUBatch | None = None
self._renderer: Renderer | None = None
self._texture_manager: TextureManager | None = None
# Pick pass
self._pick_pass: PickPass | None = None
self._graphics_qf: int = 0
# Selection outline pass
self._outline_pass: OutlinePass | None = None
self._selected_objects: list[tuple[MeshHandle, np.ndarray, int]] = []
# Texture system (lazily initialized)
self._texture_descriptor_pool: _VkDescriptorPool | None = None
self._texture_descriptor_layout: _VkDescriptorSetLayout | None = None
self._texture_descriptor_set: _VkDescriptorSet | None = None
self._default_sampler: _VkSampler | None = None
# Nearest-neighbour sampler: created lazily on first request from a
# Sprite2D with ``filter="nearest"`` so non-pixel-art games pay nothing.
# Same descriptor layout as the default; just trades VK_FILTER_LINEAR
# for VK_FILTER_NEAREST on min/mag.
self._nearest_sampler: _VkSampler | None = None
# Per-(filter, max_lod) sampler cache for mip-chained textures. The two
# base samplers above are the max_lod=0 entries (seeded in
# _init_texture_system) so single-mip uploads stay byte-identical; a
# multi-mip upload binds a distinct maxLod=mip_count-1 sampler so the
# whole chain is actually sampled. Keyed (filter, max_lod_float).
self._sampler_cache: dict[tuple[str, float], _VkSampler] = {}
self._next_texture_index = 0
# Slots released by unregister_texture; reused LIFO by register_texture.
self._free_texture_slots: list[int] = []
self._user_samplers: list[_VkSampler] = []
self._user_render_targets: list[RenderTarget] = []
# User-loaded images (image, memory, view)
self._user_images: list[tuple[_VkImage, _VkDeviceMemory, _VkImageView]] = []
# Texture sizes: index → (width, height)
self._texture_sizes: dict[int, tuple[int, int]] = {}
# User-created resources to clean up
self._user_buffers: list[tuple[Any, _VkDeviceMemory]] = []
self._user_descriptor_pools: list[_VkDescriptorPool] = []
self._user_descriptor_layouts: list[_VkDescriptorSetLayout] = []
self._user_pipelines: list[tuple[_VkPipeline, _VkPipelineLayout]] = []
self._user_shader_modules: list[_VkShaderModule] = []
# KHR extension functions (loaded after instance/device creation)
self._vk_acquire: Callable | None = None
self._vk_present: Callable | None = None
# GPU context (populated in _init_vulkan)
self._ctx: GPUContext | None = None
# Explicit-multi-adapter opt-in + manager. ``_multi_gpu_requested``
# is set by ``App`` before ``run()`` (default OFF). ``_multi_device`` is the
# MultiDeviceManager built in ``_init_vulkan``: on this single-GPU box (or
# when the opt-in is off) it is a single-slot passthrough and the engine
# behaves byte-identically. ``ctx.multi_gpu`` / ``ctx.device_count`` mirror
# its state for the renderer to read.
self._multi_gpu_requested: bool = False
self._multi_device: Any = None
# Per-pass GPU timing: one TimestampPool per frame-in-flight, rotated
# in lockstep with FrameSync. Latest readback is published on
# ``gpu_phase_times`` (label -> ms) for downstream profilers.
self._timestamp_pools: list[TimestampPool] = []
self.gpu_phase_times: dict[str, float] = {}
# Device-loss recovery. Max ``_rebuild_device`` attempts before the
# device-loss path gives up and falls back to terminal clean shutdown +
# ``on_device_lost`` (so a permanently-dead device never infinite-loops).
self.device_recovery_max_retries: int = 3
# Consecutive device losses with no good frame in between. Reset to 0 once a
# frame renders+presents cleanly; when it exceeds ``device_recovery_max_retries``
# the recovery driver gives up and finalises (so a permanently-dead device,
# which keeps re-losing every frame even though each rebuild "succeeds",
# terminates cleanly instead of recover/re-lose looping forever).
self._consecutive_device_losses: int = 0
# Optional render-thread controller, installed by FrameLoop when running
# pipelined (``App(render_thread=True)``). A rebuild must quiesce the
# render thread (close its packet ring + join) BEFORE tearing the device
# down, then spin up a fresh thread + ring afterwards (a Python thread is
# single-use). ``None`` on the synchronous path (no coordination needed).
self._render_thread_controller: Any = None
# Optional callback fired at the end of a device rebuild, after the new
# Renderer exists. Installed by FrameLoop to re-point its SceneAdapter +
# SubViewportManager at the NEW renderer -- without it the adapter keeps
# submitting to the dead old renderer and post-rebuild frames are blank.
self._on_device_rebuilt: Callable[[], None] | None = None
# --- Public API ---
[docs]
@property
def ctx(self) -> GPUContext | None:
"""GPU context holding device, physical_device, queues, and command pool.
``None`` until ``run()`` invokes ``_init_vulkan()``.
"""
return self._ctx
[docs]
@property
def multi_device(self) -> Any:
"""The :class:`~simvx.graphics.gpu.multi_device.MultiDeviceManager`.
``None`` until ``run()`` invokes ``_init_vulkan()``. On the single-GPU /
unopted path it is a single-slot passthrough (``multi_gpu`` is ``False``).
"""
return self._multi_device
[docs]
@property
def render_pass(self) -> _VkRenderPass | None:
"""Main render pass. ``None`` until ``run()`` invokes ``_init_vulkan()``."""
return self._render_pass
[docs]
@property
def extent(self) -> tuple[int, int] | None:
"""Swapchain extent. ``None`` until ``run()`` invokes ``_init_vulkan()``."""
return self._swapchain.extent if self._swapchain else None
[docs]
@property
def content_scale(self) -> tuple[float, float]:
"""HiDPI content scale (e.g. ``(2.0, 2.0)`` on a 200% display)."""
return self._content_scale
[docs]
@property
def vsync(self) -> bool:
"""Whether vertical sync is currently enabled."""
return self._vsync
[docs]
def set_vsync(self, value: bool) -> None:
"""Toggle vsync at runtime by recreating the swapchain with a new present mode.
No-op if the requested state matches the current one. Before the
engine has a swapchain (pre-``run()``) the value is stored and used
at boot.
"""
value = bool(value)
if value == self._vsync:
return
if self._swapchain is None:
self._vsync = value
return
_recreate_swapchain(self, vsync=value)
@property
def pre_render_callback(self) -> Callable[[_VkCommandBuffer], None] | None:
"""Callback invoked after ``vkBeginCommandBuffer`` but before the main render pass."""
return self._pre_render_callback
[docs]
@pre_render_callback.setter
def pre_render_callback(self, value: Callable[[_VkCommandBuffer], None] | None) -> None:
self._pre_render_callback = value
[docs]
@property
def shader_dir(self) -> Path:
return SHADER_DIR
[docs]
@property
def current_frame(self) -> int:
"""Index of the frame-in-flight slot currently being recorded (0..FRAMES_IN_FLIGHT-1).
Stable across a frame's recording; advances at present. Per-frame ring
buffers index their current slot by this so a write never lands on memory
a still-pending previous frame reads.
"""
return self._sync.current_frame if self._sync else 0
[docs]
@property
def current_timestamp_pool(self) -> TimestampPool | None:
"""Return the TimestampPool for the in-flight frame currently recording.
``None`` if the device does not support timestamp queries (the pool
list is left empty by ``_init_vulkan`` in that case). Renderers wrap
each pass with ``pool.begin(cmd, label)`` / ``pool.end(cmd, label)``.
"""
if not self._timestamp_pools or not self._sync:
return None
return self._timestamp_pools[self._sync.current_frame]
[docs]
@property
def mesh_registry(self) -> MeshRegistry:
"""Get mesh registry (lazy init)."""
if not self._mesh_registry:
from .renderer.mesh_registry import MeshRegistry
# Retain CPU geometry only when an opted-in multi-GPU renderer is active
# (>= 2 devices): the D8 offload path mirrors an offloaded SRU's geometry
# onto a secondary device's own registry, which needs the source arrays
# (a VkBuffer cannot cross devices). On the single-GPU / unopted path the
# manager is single-slot, so retention stays OFF and memory is unchanged.
mgr = self._multi_device
retain = bool(mgr is not None and getattr(mgr, "multi_gpu", False))
self._mesh_registry = MeshRegistry(
self._device, self._physical_device, retain_geometry=retain
)
return self._mesh_registry
[docs]
@property
def capabilities(self) -> RenderCapabilities | None:
"""Immutable host + GPU capability snapshot (set during ``init_vulkan``).
``None`` before Vulkan initialisation. Read-only: configuration flows
through ``WorldEnvironment`` / ``App``, never through this object.
"""
return self._capabilities
[docs]
@property
def texture_manager(self) -> TextureManager:
"""Get texture manager (lazy init)."""
if not self._texture_manager:
from .materials.texture import TextureManager
self._texture_manager = TextureManager(self)
return self._texture_manager
[docs]
@property
def batch(self) -> GPUBatch:
"""Get the GPU batch renderer (lazy init)."""
if not self._gpu_batch:
from .renderer.gpu_batch import GPUBatch
self._gpu_batch = GPUBatch(self, self._device, self._physical_device)
return self._gpu_batch
[docs]
@property
def renderer(self) -> Renderer:
"""Get the active renderer (creates the default renderer if none exists)."""
if not self._renderer:
self.create_renderer("forward")
return self._renderer
[docs]
def create_renderer(self, renderer_type: str = "forward") -> Renderer:
"""Create and initialize a renderer."""
if renderer_type == "forward":
from .renderer.forward import Renderer
self._renderer = Renderer(self)
self._renderer.setup()
else:
raise ValueError(f"Unknown renderer type: {renderer_type}")
return self._renderer
[docs]
@property
def texture_descriptor_layout(self) -> _VkDescriptorSetLayout:
"""Get (lazily init) the texture descriptor set layout."""
if not self._texture_descriptor_layout:
self._init_texture_system()
return self._texture_descriptor_layout
[docs]
@property
def texture_descriptor_set(self) -> _VkDescriptorSet:
"""Get the texture descriptor set (set 1)."""
return self._texture_descriptor_set
def _init_texture_system(self) -> None:
"""Lazily initialize the texture descriptor pool, layout, set, and default sampler.
The global bindless texture array is UPDATE_AFTER_BIND: slots are written
(e.g. a SubViewport registering its live render target) while the set is
bound in the recording frame command buffer.
"""
self._texture_descriptor_pool = create_texture_descriptor_pool(
self._device, self._max_textures, update_after_bind=True
)
self._texture_descriptor_layout = create_texture_descriptor_layout(
self._device, self._max_textures, update_after_bind=True
)
self._texture_descriptor_set = allocate_descriptor_set(
self._device,
self._texture_descriptor_pool,
self._texture_descriptor_layout,
)
self._default_sampler = create_sampler(self._device)
self._user_samplers.append(self._default_sampler)
# Seed the cache with the maxLod=0 linear sampler so multi-mip lookups
# for ("linear", 0.0) return the existing default (byte-identical).
self._sampler_cache[("linear", 0.0)] = self._default_sampler
[docs]
def create_render_target(self, width: int, height: int, use_depth: bool = True) -> RenderTarget:
"""Create an offscreen render target for render-to-texture."""
rt = RenderTarget(
self._device, self._physical_device, width, height, use_depth=use_depth,
queue=self._graphics_queue, command_pool=self._cmd_ctx.pool,
)
self._user_render_targets.append(rt)
return rt
[docs]
def load_cubemap(
self,
paths_or_hdr: list[str] | str | None = None,
*,
colour: tuple[float, float, float] | None = None,
face_size: int = 256,
faces: list | None = None,
) -> CubemapHandle:
"""Load a cubemap from 6 face images, an equirectangular HDR, or a solid colour.
Args:
paths_or_hdr: Either a list of 6 face paths
``[+X, -X, +Y, -Y, +Z, -Z]`` for a pre-split cubemap, or a
single ``.hdr`` (Radiance RGBE) equirectangular file projected
onto 6 faces on the CPU. ``None`` falls back to ``colour``.
colour: RGB triple in 0..1 used when no paths are provided.
face_size: Per-face resolution when projecting from an HDR equirect.
Returns:
A :class:`CubemapHandle`; hand this to
:meth:`Renderer.set_skybox` to install it.
"""
from .assets.cubemap_loader import load_cubemap as _load_cubemap
face_paths: list[str] | None = None
hdr_path: str | None = None
if isinstance(paths_or_hdr, str):
ext = paths_or_hdr.lower()
if ext.endswith(".hdr"):
hdr_path = paths_or_hdr
else:
# Authors occasionally pass a single non-HDR path expecting
# something useful: fail loudly with the supported shapes.
raise ValueError(
f"Unsupported single-file cubemap source {paths_or_hdr!r}. "
"Pass an .hdr equirect file or a list of 6 face paths.",
)
else:
face_paths = paths_or_hdr
view, sampler, image, memory = _load_cubemap(
self._device,
self._physical_device,
self._graphics_queue,
self._cmd_ctx.pool,
face_paths=face_paths,
hdr_path=hdr_path,
colour=colour,
face_size=face_size,
faces=faces,
)
return CubemapHandle(view=view, sampler=sampler, image=image, memory=memory)
[docs]
def register_lut(self, tex_id: int, lut_data: np.ndarray) -> int:
"""Register a 3D colour-grading LUT under ``tex_id``.
``lut_data`` is an ``(size, size, size, 4)`` uint8 array (e.g. from
``simvx.core.colour_grading.generate_warm_lut``). Select it
at render time with ``WorldEnvironment.lut_tex_id = tex_id`` and
``lut_enabled = True``. The LUT is a 3D ``rgba8`` image sampled
post-tonemap, matching the web runtime. Returns ``tex_id``.
"""
pp = getattr(self.renderer, "_post_process", None)
if pp is None:
raise RuntimeError("register_lut: post-process pass not initialised")
pp.register_lut(tex_id, lut_data)
return tex_id
[docs]
def register_texture(self, image_view: _VkImageView, *, filter: str = "linear",
mip_count: int = 1) -> int:
"""Register a texture (image view) into the bindless array.
Returns the texture index. Reuses slots freed by
:meth:`unregister_texture`; otherwise allocates a new slot from
the high-water mark.
``filter`` selects the bound sampler: ``"linear"`` (default,
bilinear) or ``"nearest"`` (point sampling, the right choice for
pixel-art sprites). The same image view can be registered twice with
different filter modes to give the same texture two bindless indices,
which is how :class:`SceneAdapter` honours per-Sprite2D ``filter``.
``mip_count`` is the number of mip levels in the view. The default 1
binds the shared maxLod=0 sampler (single-mip behaviour unchanged); a
value > 1 binds a cached maxLod=mip_count-1 sampler so the whole mip
chain is sampled (without it every level beyond 0 is unreachable).
"""
if not self._texture_descriptor_set:
self._init_texture_system()
if mip_count > 1:
sampler = self._sampler_for(filter, float(mip_count - 1))
else:
sampler = self._sampler_for_filter(filter)
if self._free_texture_slots:
idx = self._free_texture_slots.pop()
else:
idx = self._next_texture_index
self._next_texture_index += 1
write_texture_descriptor(
self._device,
self._texture_descriptor_set,
idx,
image_view,
sampler,
)
return idx
def _sampler_for_filter(self, filter: str) -> _VkSampler:
"""Return the bindless sampler for a filter string.
``"nearest"`` lazily creates the dedicated nearest-neighbour sampler so
non-pixel-art games pay nothing for a feature they don't use.
"""
if filter == "nearest":
if self._nearest_sampler is None:
self._nearest_sampler = create_sampler(self._device, filter_mode=vk.VK_FILTER_NEAREST)
self._user_samplers.append(self._nearest_sampler)
return self._nearest_sampler
# Treat unknown values as linear: Sprite2D's enum already validates,
# so reaching here with anything else means a backend internal caller
# passed something exotic. Default beats hard-fail in a hot path.
return self._default_sampler
def _sampler_for(self, filter: str, max_lod: float) -> _VkSampler:
"""Return a cached sampler for ``(filter, max_lod)``, creating it once.
Used for mip-chained textures (max_lod = mip_count-1) so the whole chain
is sampled. max_lod=0 reuses the shared base sampler (the cache is seeded
with it), so single-mip uploads are byte-identical. New samplers are
appended to ``_user_samplers`` for teardown.
"""
key = (filter if filter in ("linear", "nearest") else "linear", max_lod)
cached = self._sampler_cache.get(key)
if cached is not None:
return cached
if max_lod == 0.0:
# Defer to the base path so the two global samplers stay the single
# source for the maxLod=0 case (and get lazily created if needed).
sampler = self._sampler_for_filter(key[0])
else:
filter_mode = vk.VK_FILTER_NEAREST if key[0] == "nearest" else vk.VK_FILTER_LINEAR
sampler = create_sampler(self._device, filter_mode=filter_mode, max_lod=max_lod)
self._user_samplers.append(sampler)
self._sampler_cache[key] = sampler
return sampler
[docs]
def update_texture(self, slot: int, image_view: _VkImageView) -> None:
"""Rewrite the descriptor at an existing bindless slot.
Used when the backing image view changes (e.g. render-target
resize) but the slot id must remain stable so callers that
captured it don't need re-notification.
"""
if not self._texture_descriptor_set:
return
write_texture_descriptor(
self._device,
self._texture_descriptor_set,
slot,
image_view,
self._default_sampler,
)
[docs]
def unregister_texture(self, slot: int) -> None:
"""Release a bindless texture slot for reuse by a later register_texture.
The descriptor at the slot is left pointing at its old view until
something new is bound there; the slot is simply marked free.
"""
if slot < 0:
return
self._texture_sizes.pop(slot, None)
if slot not in self._free_texture_slots:
self._free_texture_slots.append(slot)
[docs]
def upload_texture_pixels(
self, pixels: np.ndarray, width: int, height: int, *,
filter: str = "linear", colour_space: str = "srgb", mipmaps: bool = False,
) -> int:
"""Upload raw RGBA pixel data to GPU. Returns the bindless texture index.
``filter`` is forwarded to :meth:`register_texture`; pass ``"nearest"``
for pixel-art sprites that should keep crisp edges when scaled.
``colour_space`` selects the sampled image view format (linear-workflow
contract, Stage 1): ``"srgb"`` (default) makes a ``VK_FORMAT_R8G8B8A8_SRGB``
view so the GPU decodes the stored sRGB bytes to linear on sample,
round-tripping correctly through the sRGB swapchain (the right choice for
colour textures: 2D sprites, tileset atlases, Draw2D images). ``"linear"``
keeps the ``VK_FORMAT_R8G8B8A8_UNORM`` view: the choice for data textures
that are not perceptual colour (normal/data maps) and for the 3D material
albedo path, which currently treats its uploaded bytes as already-linear.
The pixel bytes uploaded to memory are identical in both cases; only the
view's interpretation differs.
``mipmaps=True`` allocates the full mip chain and generates every level
on the GPU with a linear-filtered ``vkCmdBlitImage`` chain,
then binds the maxLod=chain-1 sampler so minification actually reaches
the chain. Falls back to a single mip when the view format lacks
linear-blit support (so exotic formats degrade gracefully) or when the
image is already 1x1. The default False keeps the historical single-mip
upload byte-identical.
"""
from .gpu.memory import format_blit_supported, mip_chain_length, upload_image_data
view_format = vk.VK_FORMAT_R8G8B8A8_SRGB if colour_space == "srgb" else vk.VK_FORMAT_R8G8B8A8_UNORM
mip_count = 1
if mipmaps:
chain = mip_chain_length(width, height)
if chain > 1 and format_blit_supported(self._physical_device, view_format):
mip_count = chain
image, memory = upload_image_data(
self._device, self._physical_device, self._graphics_queue, self._cmd_ctx.pool,
np.ascontiguousarray(pixels), width, height, view_format, mip_count=mip_count,
)
view_info = vk.VkImageViewCreateInfo(
image=image,
viewType=vk.VK_IMAGE_VIEW_TYPE_2D,
format=view_format,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
baseMipLevel=0, levelCount=mip_count, baseArrayLayer=0, layerCount=1,
),
)
image_view = vk.vkCreateImageView(self._device, view_info, None)
tex_idx = self.register_texture(image_view, filter=filter, mip_count=mip_count)
self._user_images.append((image, memory, image_view))
self._texture_sizes[tex_idx] = (width, height)
return tex_idx
[docs]
def compressed_caps(self) -> dict[str, bool]:
"""Coarse block-compression family support for this device.
Returns the ``texture_compression_{bc,etc2,astc_ldr}`` subset of the
probed capabilities. Consulted by ``TextureManager`` to pick a UASTC
transcode target (BC7 -> ASTC-4x4 -> ETC2). Each coarse bit is necessary
but not sufficient: a per-format SAMPLED check (``format_supported``) is
the second gate. Public accessor so callers never reach into the private
``_capabilities``.
"""
caps = self._capabilities
return {
"texture_compression_bc": bool(caps and caps.texture_compression_bc),
"texture_compression_etc2": bool(caps and caps.texture_compression_etc2),
"texture_compression_astc_ldr": bool(caps and caps.texture_compression_astc_ldr),
}
[docs]
def upload_texture_blocks(
self, blocks: list[bytes], width: int, height: int, fmt: int, block_size: int, *,
filter: str = "linear",
) -> int:
"""Upload block-compressed mip data to GPU. Returns the bindless texture index.
Parallel to :meth:`upload_texture_pixels` for the uncompressed RGBA8
path: builds the image view with the COMPRESSED ``fmt`` and a levelCount
spanning all supplied mips, then registers it into the bindless array
and tracks it for teardown exactly like the uncompressed path.
"""
from .gpu.memory import upload_compressed_image
image, memory = upload_compressed_image(
self._device, self._physical_device, self._graphics_queue, self._cmd_ctx.pool,
blocks, width, height, fmt, block_size,
)
view_info = vk.VkImageViewCreateInfo(
image=image,
viewType=vk.VK_IMAGE_VIEW_TYPE_2D,
format=fmt,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
baseMipLevel=0, levelCount=len(blocks), baseArrayLayer=0, layerCount=1,
),
)
image_view = vk.vkCreateImageView(self._device, view_info, None)
# mip_count = number of supplied levels: a multi-mip chain binds a
# maxLod=mip_count-1 sampler so every level is sampled (a single-mip
# texture keeps the maxLod=0 sampler, unchanged).
tex_idx = self.register_texture(image_view, filter=filter, mip_count=len(blocks))
self._user_images.append((image, memory, image_view))
self._texture_sizes[tex_idx] = (width, height)
return tex_idx
[docs]
def load_mesh(self, file_path: str) -> MeshHandle:
"""Load a glTF file's first mesh primitive from disk and register it.
Returns: MeshHandle for use in rendering.
"""
from .assets.mesh_loader import load_gltf
scene = load_gltf(file_path)
if not scene.meshes:
raise ValueError(f"glTF file has no meshes: {file_path}")
streams, indices = scene.meshes[0]
handle = self.mesh_registry.register(streams, indices)
log.debug(
"Loaded mesh %s: %d verts, %d indices -> handle %d",
file_path,
handle.vertex_count,
handle.index_count,
handle.id,
)
return handle
[docs]
def create_vertex_buffer(self, vertices: np.ndarray) -> tuple[Any, _VkDeviceMemory]:
"""Create a GPU vertex buffer and upload data. Returns (buffer, memory)."""
buf, mem = create_buffer(
self._device,
self._physical_device,
vertices.nbytes,
vk.VK_BUFFER_USAGE_VERTEX_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
upload_numpy(self._device, mem, vertices)
self._user_buffers.append((buf, mem))
return buf, mem
[docs]
def create_index_buffer(self, indices: np.ndarray) -> tuple[Any, _VkDeviceMemory]:
"""Create a GPU index buffer and upload data. Returns (buffer, memory)."""
buf, mem = create_buffer(
self._device,
self._physical_device,
indices.nbytes,
vk.VK_BUFFER_USAGE_INDEX_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
upload_numpy(self._device, mem, indices)
self._user_buffers.append((buf, mem))
return buf, mem
[docs]
def create_ssbo(self, data: np.ndarray) -> tuple[Any, _VkDeviceMemory]:
"""Create an SSBO and upload data. Returns (buffer, memory)."""
buf, mem = create_buffer(
self._device,
self._physical_device,
data.nbytes,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
upload_numpy(self._device, mem, data)
self._user_buffers.append((buf, mem))
return buf, mem
[docs]
def update_ssbo(self, memory: _VkDeviceMemory, data: np.ndarray) -> None:
"""Update SSBO contents in-place via mapped memory."""
upload_numpy(self._device, memory, data)
[docs]
def create_descriptor_pool(self, max_sets: int = 4) -> _VkDescriptorPool:
"""Create a descriptor pool."""
pool = create_descriptor_pool(self._device, max_sets)
self._user_descriptor_pools.append(pool)
return pool
[docs]
def create_descriptor_set_layout(self, binding_count: int = 3) -> _VkDescriptorSetLayout:
"""Create a descriptor set layout with N SSBO bindings."""
layout = create_ssbo_layout(self._device, binding_count)
self._user_descriptor_layouts.append(layout)
return layout
[docs]
def allocate_descriptor_set(self, pool: _VkDescriptorPool, layout: _VkDescriptorSetLayout) -> _VkDescriptorSet:
"""Allocate a descriptor set from pool."""
return allocate_descriptor_set(self._device, pool, layout)
[docs]
def write_descriptor_ssbo(
self,
descriptor_set: _VkDescriptorSet,
binding: int,
buffer: Any,
size: int,
) -> None:
"""Bind an SSBO buffer to a descriptor set binding."""
write_ssbo_descriptor(self._device, descriptor_set, binding, buffer, size)
[docs]
def compile_and_load_shader(self, name: str) -> _VkShaderModule:
"""Compile a shader from SHADER_DIR and return its module."""
spv = compile_shader(SHADER_DIR / name)
module = create_shader_module(self._device, spv)
self._user_shader_modules.append(module)
return module
[docs]
def update_vertex_buffer(self, memory: _VkDeviceMemory, data: np.ndarray) -> None:
"""Update vertex buffer contents in-place via mapped memory."""
upload_numpy(self._device, memory, data)
[docs]
def push_constants(self, cmd: _VkCommandBuffer, pipeline_layout: _VkPipelineLayout, data: bytes | bytearray) -> None:
"""Push constant data (view + proj)."""
ffi = vk.ffi
# cffi needs a writable buffer or char*
cbuf = ffi.new("char[]", data)
vk._vulkan.lib.vkCmdPushConstants(
cmd,
pipeline_layout,
vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
0,
len(data),
cbuf,
)
# --- Picking ---
[docs]
def enable_picking(self, descriptor_layout: _VkDescriptorSetLayout, descriptor_set: _VkDescriptorSet) -> None:
"""Initialize the GPU pick pass for mouse picking."""
self._pick_pass = PickPass(
self._device,
self._physical_device,
self._graphics_queue,
self._graphics_qf,
self._swapchain.extent,
descriptor_layout,
descriptor_set,
)
self._pick_pass.create()
[docs]
def pick_entity(
self,
x: int,
y: int,
view_proj_data: bytes,
vertex_buffer: Any, # VkBuffer
index_buffer: Any, # VkBuffer
index_count: int,
instance_count: int,
) -> int:
"""Read entity ID at screen position (x, y). Returns entity index or -1."""
if not self._pick_pass:
return -1
vk.vkDeviceWaitIdle(self._device)
return self._pick_pass.pick(
x,
y,
view_proj_data,
vertex_buffer,
index_buffer,
index_count,
instance_count,
)
# --- Selection Outline ---
[docs]
def set_selected_objects(
self,
selected: list[tuple[MeshHandle, np.ndarray, int]],
) -> None:
"""Set the list of selected objects to highlight with outlines.
Args:
selected: List of (mesh_handle, transform_4x4, material_id) tuples.
"""
self._selected_objects = selected
[docs]
def clear_selected_objects(self) -> None:
"""Clear all selection outlines."""
self._selected_objects.clear()
[docs]
@property
def outline_pass(self) -> OutlinePass | None:
"""Access outline pass for configuration (colour, width, enabled)."""
return self._outline_pass
def _ensure_outline_pass(self) -> None:
"""Lazily create the outline pass when first needed."""
if self._outline_pass is not None:
return
from .renderer.outline_pass import OutlinePass
self._outline_pass = OutlinePass(self)
self._outline_pass.setup()
# --- Text ---
[docs]
def create_text_texture(
self,
font: str | None = None,
size: int = 32,
width: int = 256,
height: int = 64,
) -> Any:
"""Create a texture with rendered text for use on 3D objects.
Returns a TextTexture with .text, .colour, and .texture_index properties.
Setting .text or .colour re-renders and re-uploads the texture.
Web parity (``simvx.web`` ``EngineStub.create_text_texture``): the ``font``
argument is a no-op on web (text is rasterized from the atlas baked at export
time), and ``texture_index`` is reassigned on every re-render there (the
bindless slot is reused in place on desktop). So set ``.text``/``.colour``
*before* binding ``texture_index`` to a material for identical results on both
backends -- which is how the examples author it.
"""
from .text_utils import create_text_texture
if not self._texture_descriptor_set:
self._init_texture_system()
return create_text_texture(
self._ctx,
self.register_texture,
self._texture_descriptor_set,
self._default_sampler,
font=font,
size=size,
width=width,
height=height,
)
# --- Input ---
[docs]
def set_key_callback(self, callback: Callable[[int, int, int], None]) -> None:
"""Register callback(key, action, mods) for keyboard events."""
if self._window:
self._window.set_key_callback(callback)
[docs]
def set_cursor_pos_callback(self, callback: Callable[[float, float], None]) -> None:
"""Register callback(x, y) for cursor position events."""
if self._window:
self._window.set_cursor_pos_callback(callback)
[docs]
def set_char_callback(self, callback: Callable[[int], None]) -> None:
"""Register callback(codepoint) for character input events."""
if self._window:
self._window.set_char_callback(callback)
[docs]
def set_cursor_shape(self, shape: int) -> None:
"""Set cursor shape. 0=arrow, 1=ibeam, 2=crosshair, 3=hand, 4=hresize, 5=vresize."""
if self._window:
self._window.set_cursor_shape(shape)
[docs]
def set_mouse_capture(self, mode: int) -> None:
"""Apply a MouseCaptureMode int to the OS cursor via the active window backend."""
if self._window is not None and hasattr(self._window, "set_mouse_capture"):
self._window.set_mouse_capture(mode)
[docs]
@property
def cursor_pos(self) -> tuple[float, float]:
"""Current cursor position in screen coordinates."""
return self._window.get_cursor_pos() if self._window else (0.0, 0.0)
# --- Main Loop ---
[docs]
def run(
self,
callback: Callable[[], None] | None = None,
setup: Callable[[], None] | None = None,
render: Callable[[_VkCommandBuffer, tuple[int, int]], None] | None = None,
pre_render: Callable[[_VkCommandBuffer], None] | None = None,
cleanup: Callable[[], None] | None = None,
frame_driver: Callable[[], None] | None = None,
pre_shutdown: Callable[[], None] | None = None,
) -> None:
"""Start the main loop.
Args:
callback: Legacy per-frame callback (called before draw).
setup: Called once after Vulkan init, before the loop.
render: Custom render callback receiving (command_buffer, extent).
If provided, replaces the built-in triangle rendering.
pre_render: Called with command_buffer after vkBeginCommandBuffer
but before the main render pass. Use for offscreen passes.
cleanup: Called once when the loop exits, *after* the final
``vkDeviceWaitIdle`` but *before* ``shutdown()`` destroys
the device. The right place to release caller-owned GPU
resources (offscreen render targets, bindless slots) that
must outlive the loop but die before the device.
pre_shutdown: Called once when the loop exits, BEFORE the final
``vkDeviceWaitIdle``. The pipelined driver stops + joins its
render thread here so no GPU work is issued against the
device while/after it is being torn down. ``None`` (default)
is a no-op, unchanged.
frame_driver: Pipelined-mode hook replacing the per-frame
``self._draw_frame()`` call on the MAIN thread. When given,
the engine loop runs ``poll_events -> callback() ->
frame_driver()`` and never touches the GPU itself: the
driver hands the frame's render packet to a render thread
that owns all GPU work. ``None`` (default) keeps the
synchronous ``_draw_frame`` path, byte-identical to before.
"""
self.begin(setup=setup, render=render, pre_render=pre_render)
import time as _time
frame_budget = (1.0 / self._target_fps) if self._target_fps else 0.0
try:
while True:
frame_start = _time.perf_counter() if frame_budget else 0.0
if not self.step(callback=callback, frame_driver=frame_driver):
break
if frame_budget:
elapsed = _time.perf_counter() - frame_start
remaining = frame_budget - elapsed
if remaining > 0:
_time.sleep(remaining)
finally:
self.end(pre_shutdown=pre_shutdown, cleanup=cleanup)
[docs]
def begin(
self,
*,
setup: Callable[[], None] | None = None,
render: Callable[[_VkCommandBuffer, tuple[int, int]], None] | None = None,
pre_render: Callable[[_VkCommandBuffer], None] | None = None,
) -> None:
"""Create the window + Vulkan device and run ``setup``.
The externally-driven counterpart to :meth:`run`: pair with :meth:`step`
and :meth:`end` to advance frames under a caller-owned clock (the agent
live-session / editor viewport). ``run`` is ``begin`` + a step loop + ``end``.
"""
self._window, self._resolved_backend_name = resolve_backend(self._backend_name)
self._window.create_window(self.width, self.height, self.title, visible=self._visible)
# Query HiDPI content scale (e.g. (2.0, 2.0) on a 200% display)
if hasattr(self._window, "get_content_scale"):
self._content_scale = self._window.get_content_scale()
self._use_depth = render is not None
self._init_vulkan(use_triangle=render is None)
self._render_callback = render
self._pre_render_callback = pre_render
# Subscribe to native window resize events so the swapchain is recreated
# promptly even on drivers that do not emit OutOfDate/Suboptimal.
if hasattr(self._window, "set_resize_callback"):
self._window.set_resize_callback(self._on_window_resize)
if setup:
setup()
self._running = True
[docs]
def step(
self,
*,
callback: Callable[[], None] | None = None,
frame_driver: Callable[[], None] | None = None,
) -> bool:
"""Run one frame iteration (poll -> callback -> draw).
Returns False when the loop should stop (quit requested or window closed),
True otherwise (including a paused/backgrounded frame, which is skipped).
"""
if not self._running or self._window.should_close():
return False
self._window.poll_events()
# Skip rendering while paused (mobile background); keep the loop alive.
if getattr(self._window, "paused", False):
return True
if callback:
callback()
if frame_driver is not None:
frame_driver()
else:
self._draw_frame()
return self._running and not self._window.should_close()
[docs]
def end(
self,
*,
pre_shutdown: Callable[[], None] | None = None,
cleanup: Callable[[], None] | None = None,
) -> None:
"""Tear down: quiesce the render thread, wait-idle, run cleanup, shutdown."""
# Quiesce the render thread (if any) BEFORE touching the device, so it is not
# recording / submitting while we wait-idle + destroy resources.
if pre_shutdown is not None:
pre_shutdown()
# A lost device rejects vkDeviceWaitIdle with VK_ERROR_DEVICE_LOST; skip the
# wait in that case (there is no in-flight work to drain anyway).
if self._device and not self._device_lost:
vk.vkDeviceWaitIdle(self._device)
if cleanup is not None:
cleanup()
self.shutdown()
def _init_vulkan(self, use_triangle: bool = True) -> None:
_init_vulkan(self, use_triangle)
self._device_state = DeviceState.READY
def _create_instance_and_surface(self) -> None:
"""Create instance-level handles that survive a device-only loss."""
_create_instance_and_surface(self)
def _create_device_objects(self, use_triangle: bool = True) -> None:
"""Create everything bound to the logical device (recreated on rebuild)."""
_create_device_objects(self, use_triangle)
def _create_depth_resources(self) -> None:
_create_depth_resources(self)
def _destroy_depth_resources(self) -> None:
_destroy_depth_resources(self)
def _create_framebuffers(self) -> None:
_create_framebuffers(self)
def _destroy_framebuffers(self) -> None:
_destroy_framebuffers(self)
def _recreate_swapchain(self) -> None:
_recreate_swapchain(self)
def _on_window_resize(self, w: int, h: int) -> None:
# A resize to the SAME framebuffer size needs no swapchain rebuild. Some
# windowing backends (SDL3) emit a spurious same-size PIXEL_SIZE_CHANGED
# event for a freshly created window; acting on it recreates the swapchain
# and makes the next _draw_frame skip its render, desyncing frame-indexed
# render state (e.g. the TAA jitter phase, first-frame occlusion cull) from
# the logic clock. Both backends deliver framebuffer-pixel dimensions here,
# so only a genuine dimension change is flagged. (History: BUGS.md
# bug-taa-golden-preexisting-failure, resolved; see git log.)
if self._swapchain is not None and (w, h) == tuple(self._swapchain.extent):
return
self._resize_pending = True
def _draw_triangle(self, cmd: _VkCommandBuffer) -> None:
"""Record built-in triangle draw commands."""
vk.vkCmdBindPipeline(cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline)
viewport = vk.VkViewport(
x=0.0,
y=0.0,
width=float(self._swapchain.extent[0]),
height=float(self._swapchain.extent[1]),
minDepth=0.0,
maxDepth=1.0,
)
vk.vkCmdSetViewport(cmd, 0, 1, [viewport])
scissor = vk.VkRect2D(
offset=vk.VkOffset2D(x=0, y=0),
extent=vk.VkExtent2D(width=self._swapchain.extent[0], height=self._swapchain.extent[1]),
)
vk.vkCmdSetScissor(cmd, 0, 1, [scissor])
vk.vkCmdDraw(cmd, 3, 1, 0, 0)
def _render_selection_outlines(self, cmd: Any) -> None:
"""Render selection outlines for highlighted objects."""
self._ensure_outline_pass()
op = self._outline_pass
if not op or not op.enabled:
return
# Build view+proj push constant data from the first viewport
renderer = self._renderer
if renderer:
viewports = renderer.viewport_manager.viewports
if viewports:
_, viewport = viewports[0]
view_t = np.ascontiguousarray(viewport.camera_view.T)
proj_t = np.ascontiguousarray(viewport.camera_proj.T)
pc_data = view_t.tobytes() + proj_t.tobytes()
else:
return
else:
return
op.render(
cmd,
self._selected_objects,
pc_data,
self.mesh_registry,
self._swapchain.extent,
)
def _handle_device_lost(self, where: str) -> None:
"""React to ``VK_ERROR_DEVICE_LOST``: attempt auto-recovery, else stop cleanly.
Called when a Vulkan call returns ``VK_ERROR_DEVICE_LOST`` (GPU reset /
driver crash / device removed): every handle from this device is now
invalid. The policy is full auto-rebuild (web parity): flip the gate,
then mark the loss as PENDING so the per-frame driver attempts a recovery
rebuild (with a small retry cap). Only when recovery is impossible does the
loop fall back to terminal clean shutdown + ``on_device_lost``.
Recovery is NOT run inline here because under the pipelined render thread
this runs ON the render thread, which cannot join itself: the rebuild must
happen on the quiescent main thread. So this only records the loss; the
frame driver (synchronous: :meth:`_draw_frame`; pipelined: the FrameLoop
on the main thread) calls :meth:`_recover_device` at the frame boundary.
Idempotent.
"""
if self._device_lost:
return
self._device_lost = True
self._device_lost_where = where
self._device_state = DeviceState.LOST
self._consecutive_device_losses += 1
log.error(
"Vulkan device lost during %s (VK_ERROR_DEVICE_LOST): the GPU driver reset, "
"crashed, or the device was removed. Attempting device rebuild...",
where,
)
def _finalise_device_loss(self, where: str) -> None:
"""Terminal fallback when device recovery is exhausted or disabled.
Stops the loop deterministically and fires the optional ``on_device_lost``
hook (the legacy clean-shutdown behaviour, now the recovery fallback).
"""
self._device_state = DeviceState.LOST
self._running = False
log.error(
"Vulkan device rebuild failed after %d attempt(s) (lost during %s): "
"stopping the render loop and shutting down cleanly.",
self.device_recovery_max_retries, where,
)
if self.on_device_lost is not None:
try:
self.on_device_lost(where)
except Exception:
log.exception("on_device_lost hook raised")
def _recover_device(self) -> bool:
"""Drive a device rebuild with a bounded retry cap after a device loss.
Returns ``True`` if the device was rebuilt and rendering can resume,
``False`` if every attempt failed (the caller then finalises the loss).
Runs on the MAIN thread (synchronous loop, or FrameLoop for the pipelined
path), never on the render thread. The next frame re-registers meshes from
the live scene tree, so a successful rebuild resumes from the resident scene.
"""
where = getattr(self, "_device_lost_where", "unknown")
# Give up if the device keeps re-losing every frame with no good frame in
# between (a permanently-dead device): each rebuild may "succeed" yet the
# next submit re-loses, so cap on consecutive losses, not just rebuild tries.
if self._consecutive_device_losses > self.device_recovery_max_retries:
self._finalise_device_loss(where)
return False
for attempt in range(1, self.device_recovery_max_retries + 1):
try:
if self._rebuild_device():
log.warning("Device recovery succeeded on attempt %d.", attempt)
return True
except vk.VkErrorDeviceLost:
# A persistently-faulting device re-loses during the rebuild submit:
# retry up to the cap, then fall back to terminal shutdown.
log.warning("Device rebuild attempt %d re-lost the device.", attempt)
except Exception:
log.exception("Device rebuild attempt %d raised", attempt)
self._finalise_device_loss(where)
return False
def _rebuild_device(self) -> bool:
"""Re-acquire the logical device and recreate all device objects.
Destroys the dead device objects (keeping the surviving instance, surface,
and window), recreates the device + swapchain + passes on the same surface,
recreates the renderer, and resyncs the frame driver's acquire/present/sync
cycle so the next frame renders non-blank. The live scene re-registers its
meshes on the next frame (``Mesh.positions`` etc. live on the nodes), so
the adapter-driven scene path recovers from the resident tree without a
byte journal (unlike the web one-way channel).
``_destroy_device_objects`` frees + clears every device-bound resource
(incl. the default/nearest/mip-cache samplers and the texture descriptor
layout/pool) and nulls the texture-system handles, so the next device's
lazy ``_init_texture_system`` recreates them clean -- no stale handle is
carried across the rebuild. The frame-driver resync below mirrors
:func:`~simvx.graphics._engine_init.recreate_swapchain`: a fresh FrameSync
is created by ``_create_device_objects``; here we reset the cached acquire
index so the next ``vkAcquireNextImageKHR`` starts a clean cycle on the new
swapchain (no "presentable image not acquired" validation error).
Returns ``True`` on success. Out-of-band GPU resources reloaded from
out-of-tree sources (textures, cubemaps, IBL/LUTs, user buffers, render
targets) are not yet journaled and are re-resolved lazily by the live
scene; a real-TDR live check across such assets is owed to the user.
"""
log.warning("Rebuilding Vulkan device after device loss...")
self._device_state = DeviceState.RECOVERING
# Pipelined mode: quiesce + join the render thread (closing its packet ring)
# BEFORE touching the device, so no stale packet referencing dead handles is
# recorded against the device being torn down, and the device rebuild runs on
# a single quiescent thread (the command-pool / single-recorder invariant).
controller = self._render_thread_controller
if controller is not None:
controller.stop()
self._destroy_device_objects()
# The new bindless allocator starts empty: the live scene re-registers its
# textures lazily on the next frame. ``_destroy_device_objects`` already
# freed + nulled the texture-system handles + samplers.
self._next_texture_index = 0
self._free_texture_slots = []
self._texture_sizes = {}
# Recreate the device + swapchain + passes on the surviving surface, then
# the renderer; the live scene re-registers on the next frame.
self._create_device_objects(use_triangle=False)
self.create_renderer("forward")
# Re-point any SceneAdapter / SubViewportManager at the NEW renderer; the
# adapter holds a hard renderer reference (scene_adapter.py:80), so without
# this it would keep submitting to the dead renderer and render blank.
if self._on_device_rebuilt is not None:
self._on_device_rebuilt()
# Frame-driver resync: a fresh FrameSync (current_frame=0, fences signaled,
# images_in_flight all None) was just made by ``_create_device_objects``.
# Reset the cached acquire index so capture / present do not reference a
# stale image index from the dead swapchain, and the next acquire opens a
# clean acquire->submit->present cycle on the new swapchain.
self._last_image_index = 0
self._device_lost = False
self._device_state = DeviceState.READY
# Spin up a fresh render thread + packet ring on the rebuilt device (a Python
# thread is single-use, so it is recreated, not restarted).
if controller is not None:
controller.start()
log.warning("Vulkan device rebuilt; resuming rendering.")
return True
def _draw_frame(self) -> None:
if self._device_lost:
# Device-loss recovery. On the SYNCHRONOUS path (no render-thread
# controller) recover inline here, on this same thread, then continue
# so the next frame resumes rendering. On the PIPELINED path this body
# runs on the render thread, which cannot rebuild the device (it cannot
# join itself): just return so the thread exits and the FrameLoop drives
# ``_recover_device`` on the quiescent main thread.
if self._render_thread_controller is not None:
return
if not self._recover_device():
return
# Fall through: a fresh device is live, but this frame's submission
# lists were built against the dead renderer. Skip drawing this frame;
# the next frame re-submits the live scene against the new renderer.
return
if self._resize_pending:
self._resize_pending = False
self._recreate_swapchain()
return
try:
self._sync.wait_and_reset()
except vk.VkErrorDeviceLost:
self._handle_device_lost("fence wait")
return
frame = self._sync.current_frame
try:
image_index = self._vk_acquire(
self._device,
self._swapchain.handle,
2_000_000_000,
self._sync.image_available[frame],
None,
)
except vk.VkErrorOutOfDateKhr:
self._recreate_swapchain()
return
except vk.VkSuboptimalKhr:
self._recreate_swapchain()
return
except vk.VkErrorSurfaceLostKhr:
log.warning("Vulkan surface lost: skipping frame (awaiting new surface)")
return
except vk.VkErrorDeviceLost:
self._handle_device_lost("swapchain image acquire")
return
self._last_image_index = image_index
try:
self._sync.wait_for_image(image_index)
except vk.VkErrorDeviceLost:
self._handle_device_lost("image fence wait")
return
self._sync.mark_image(image_index)
cmd = self._cmd_buffers[frame]
vk.vkResetCommandBuffer(cmd, 0)
begin_info = vk.VkCommandBufferBeginInfo()
vk.vkBeginCommandBuffer(cmd, begin_info)
# Open the async-compute frame. In passthrough (single-queue) mode this
# binds ``cmd`` as the compute record target so routed passes record
# inline (byte-identical); in async mode it resets/begins the dedicated
# compute command buffer. Compute passes route their dispatches through
# ``engine._async_compute.record_compute`` during ``pre_render``.
if self._async_compute is not None:
self._async_compute.begin_frame(frame, cmd)
# GPU timing: ``wait_and_reset`` above guaranteed the previous frame
# in this slot has finished, so its timestamps are ready. Read them
# before recording new ones into the same pool.
if self._timestamp_pools:
pool = self._timestamp_pools[frame]
results = pool.read_results()
if results:
self.gpu_phase_times = results
pool.reset(cmd)
# Pre-render pass (offscreen rendering etc.)
if self._pre_render_callback:
self._pre_render_callback(cmd)
cc = getattr(self, "clear_colour", [0.0, 0.0, 0.0, 1.0])
clear_values = [vk.VkClearValue(color=vk.VkClearColorValue(float32=cc))]
if self._use_depth:
clear_values.append(vk.VkClearValue(depthStencil=vk.VkClearDepthStencilValue(depth=1.0, stencil=0)))
rp_begin = vk.VkRenderPassBeginInfo(
renderPass=self._render_pass,
framebuffer=self._framebuffers[image_index],
renderArea=vk.VkRect2D(
offset=vk.VkOffset2D(x=0, y=0),
extent=vk.VkExtent2D(width=self._swapchain.extent[0], height=self._swapchain.extent[1]),
),
clearValueCount=len(clear_values),
pClearValues=clear_values,
)
vk.vkCmdBeginRenderPass(cmd, rp_begin, vk.VK_SUBPASS_CONTENTS_INLINE)
if self._render_callback:
self._render_callback(cmd, self._swapchain.extent)
else:
self._draw_triangle(cmd)
# Render selection outlines (inside render pass, after scene geometry)
if self._selected_objects:
self._render_selection_outlines(cmd)
vk.vkCmdEndRenderPass(cmd)
vk.vkEndCommandBuffer(cmd)
# Submit the dedicated compute command buffer (async mode) signalling
# this frame's compute-done semaphore; a no-op in passthrough mode. Done
# before the graphics submit so the compute work is in flight and the
# graphics wait (merged below) is satisfiable.
compute_wait_sems: list[Any] = []
compute_wait_stages: list[int] = []
if self._async_compute is not None:
self._async_compute.submit()
compute_wait_sems, compute_wait_stages = self._async_compute.graphics_wait_semaphores()
# The graphics submit waits on the swapchain image AND (async mode only)
# the compute-done semaphore at the stages that consume compute results
# (DRAW_INDIRECT / VERTEX_SHADER / FRAGMENT_SHADER). In passthrough mode
# ``compute_wait_*`` are empty, so this is the identical single-wait
# submit as before. Use per-image render_finished to avoid semaphore reuse.
wait_sems = [self._sync.image_available[frame], *compute_wait_sems]
wait_stages = [vk.VK_PIPELINE_STAGE_COLOR_ATTACHMENT_OUTPUT_BIT, *compute_wait_stages]
submit_info = vk.VkSubmitInfo(
waitSemaphoreCount=len(wait_sems),
pWaitSemaphores=wait_sems,
pWaitDstStageMask=wait_stages,
commandBufferCount=1,
pCommandBuffers=[cmd],
signalSemaphoreCount=1,
pSignalSemaphores=[self._sync.render_finished[image_index]],
)
try:
vk.vkQueueSubmit(self._graphics_queue, 1, [submit_info], self._sync.fences[frame])
except vk.VkErrorDeviceLost:
self._handle_device_lost("queue submit")
return
# Armed capture reads the image back HERE, before present, while it is still
# acquired (the readback transitions the presentable image, which is only
# legal between acquire and present). ``capture_swapchain_frame`` fully syncs
# the device, so the render above has completed and the pixels are final.
if self._capture_armed:
self._capture_armed = False
self._captured_frame = self.capture_frame()
present_info = vk.VkPresentInfoKHR(
waitSemaphoreCount=1,
pWaitSemaphores=[self._sync.render_finished[image_index]],
swapchainCount=1,
pSwapchains=[self._swapchain.handle],
pImageIndices=[image_index],
)
try:
self._vk_present(self._present_queue, present_info)
except vk.VkErrorOutOfDateKhr:
self._recreate_swapchain()
except vk.VkErrorSurfaceLostKhr:
log.warning("Vulkan surface lost during present")
return
except vk.VkErrorDeviceLost:
self._handle_device_lost("queue present")
return
self._sync.advance()
# A clean frame rendered + presented: the device is healthy again, so the
# consecutive-loss streak (if any) is broken.
self._consecutive_device_losses = 0
[docs]
def arm_capture(self) -> None:
"""Arm a one-shot readback of the NEXT ``_draw_frame`` before it presents.
The captured pixels are retrieved with :meth:`take_captured_frame`. This is
the acquire-before-transition capture path: the readback happens while the
swapchain image is still acquired, so it never trips the "presentable image
must be acquired" validation rule that the post-present :meth:`capture_frame`
does. Used by the synchronous headless capture sink.
"""
self._capture_armed = True
[docs]
def take_captured_frame(self) -> np.ndarray | None:
"""Return (and clear) the frame stashed by an armed capture, or ``None``."""
frame, self._captured_frame = self._captured_frame, None
return frame
[docs]
def capture_frame(self) -> np.ndarray:
"""Capture the last rendered framebuffer as an RGBA numpy array.
Returns (height, width, 4) uint8 array. Must be called after _draw_frame().
"""
return capture_swapchain_frame(
self._device,
self._physical_device,
self._graphics_queue,
self._cmd_ctx.pool,
self._swapchain.images,
self._last_image_index,
self._swapchain.extent,
self._swapchain.image_format,
)
@property
def window_size(self) -> tuple[int, int]:
"""Current window ``(width, height)``. Writable to programmatically resize."""
return (self.width, self.height)
[docs]
@window_size.setter
def window_size(self, value: tuple[int, int]) -> None:
if not isinstance(value, tuple | list) or len(value) != 2:
raise TypeError("window_size must be a (width, height) tuple")
w, h = int(value[0]), int(value[1])
if w < 1 or h < 1:
raise ValueError(f"window_size must be >= 1x1, got {w}x{h}")
self.width = w
self.height = h
if self._window:
self._window.set_window_size(w, h)
[docs]
def shutdown(self) -> None:
"""Clean up all resources."""
if not self._device:
return
self._destroy_device_objects()
self._destroy_instance()
if self._window:
self._window.destroy()
self._device_state = DeviceState.DESTROYED
log.debug("Engine shutdown complete")
def _destroy_device_objects(self) -> None:
"""Destroy everything bound to the logical device, including the device.
The reverse of ``create_device_objects`` plus the renderer / mesh
registry / texture system and user-created GPU resources that live on
the device. Leaves the engine with no device objects (handles nulled) so
a fresh device can be created in their place -- the destroy half of the
device-loss rebuild seam. The instance, surface, debug messenger, and
window are left intact (they survive a device-only loss). Idempotent.
"""
if not self._device:
return
# Wait for every in-flight command buffer to finish before releasing any
# GPU resource. ``Engine.run`` already waits before calling shutdown, but
# other teardown paths may not, so guard here too. A lost device rejects the
# wait (VK_ERROR_DEVICE_LOST) and has no in-flight work to drain, so skip it.
if not self._device_lost:
vk.vkDeviceWaitIdle(self._device)
# The per-frame command buffers remain in the *executable* state and keep
# references to the SSBOs they recorded. Reset the pool first so those
# references drop before the buffers are destroyed below, otherwise the
# validation layer reports VUID-vkDestroyBuffer-buffer-00922 (buffer in
# use by a command buffer) for every per-frame SSBO.
if self._cmd_ctx and self._cmd_ctx.pool:
vk.vkResetCommandPool(self._device, self._cmd_ctx.pool, 0)
if self._renderer:
self._renderer.cleanup()
self._renderer = None
if self._gpu_batch:
self._gpu_batch.destroy()
self._gpu_batch = None
if self._mesh_registry:
self._mesh_registry.destroy()
self._mesh_registry = None
if self._outline_pass:
self._outline_pass.cleanup()
self._outline_pass = None
if self._pick_pass:
self._pick_pass.destroy()
self._pick_pass = None
# Clean up render targets
for rt in self._user_render_targets:
rt.destroy()
self._user_render_targets.clear()
self._destroy_framebuffers()
self._destroy_depth_resources()
for pool in self._timestamp_pools:
pool.destroy()
self._timestamp_pools.clear()
if self._sync:
self._sync.destroy()
if self._async_compute:
self._async_compute.destroy()
if self._cmd_ctx:
self._cmd_ctx.destroy()
# Clean up user-created resources. Each list is CLEARED after its handles
# are destroyed so a device rebuild cannot carry a stale (now-freed) handle
# into the next ``_destroy_device_objects`` and double-free it.
for pipeline, layout in self._user_pipelines:
vk.vkDestroyPipeline(self._device, pipeline, None)
vk.vkDestroyPipelineLayout(self._device, layout, None)
self._user_pipelines.clear()
for module in self._user_shader_modules:
vk.vkDestroyShaderModule(self._device, module, None)
self._user_shader_modules.clear()
for layout in self._user_descriptor_layouts:
vk.vkDestroyDescriptorSetLayout(self._device, layout, None)
self._user_descriptor_layouts.clear()
for pool in self._user_descriptor_pools:
vk.vkDestroyDescriptorPool(self._device, pool, None)
self._user_descriptor_pools.clear()
for buf, mem in self._user_buffers:
vk.vkDestroyBuffer(self._device, buf, None)
vk.vkFreeMemory(self._device, mem, None)
self._user_buffers.clear()
for img, mem, view in self._user_images:
vk.vkDestroyImageView(self._device, view, None)
vk.vkDestroyImage(self._device, img, None)
vk.vkFreeMemory(self._device, mem, None)
self._user_images.clear()
# Clean up texture system. ``_user_samplers`` owns the default / nearest /
# mip-cache samplers (all appended there at creation), so destroying +
# clearing it here frees each exactly once. The handles are nulled so the
# next device's lazy ``_init_texture_system`` recreates them clean; without
# the clear, a rebuild would re-enter this loop with a freed handle and the
# validation layer reports "Couldn't find VkSampler".
for sampler in self._user_samplers:
vk.vkDestroySampler(self._device, sampler, None)
self._user_samplers.clear()
if self._texture_descriptor_layout:
vk.vkDestroyDescriptorSetLayout(self._device, self._texture_descriptor_layout, None)
if self._texture_descriptor_pool:
vk.vkDestroyDescriptorPool(self._device, self._texture_descriptor_pool, None)
self._default_sampler = None
self._nearest_sampler = None
self._sampler_cache = {}
self._texture_descriptor_layout = None
self._texture_descriptor_pool = None
self._texture_descriptor_set = None
self._texture_manager = None
# Clean up built-in triangle resources
if self._pipeline:
vk.vkDestroyPipeline(self._device, self._pipeline, None)
self._pipeline = None
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(self._device, self._pipeline_layout, None)
self._pipeline_layout = None
if self._vert_module:
vk.vkDestroyShaderModule(self._device, self._vert_module, None)
self._vert_module = None
if self._frag_module:
vk.vkDestroyShaderModule(self._device, self._frag_module, None)
self._frag_module = None
if self._render_pass:
vk.vkDestroyRenderPass(self._device, self._render_pass, None)
self._render_pass = None
if self._swapchain:
self._swapchain.destroy()
# Destroy any SECONDARY logical devices (D8 multi-GPU) before the primary
# device. No-op single-slot passthrough on this box.
if self._multi_device is not None:
self._multi_device.destroy()
self._multi_device = None
# Flush the pipeline cache to disk (with everything compiled this run) and
# destroy the handle while the device is still valid.
if self._device and not self._device_lost:
from .gpu.pipeline_cache import destroy_pipeline_cache
destroy_pipeline_cache(self._device)
if self._device:
vk.vkDestroyDevice(self._device, None)
def _destroy_instance(self) -> None:
"""Destroy the instance-level handles (surface, debug messenger, instance).
Run after ``_destroy_device_objects``. Left separate so a device-only
loss can rebuild the device while keeping the window + surface alive.
"""
if self._surface and self._instance:
fn = vk.vkGetInstanceProcAddr(self._instance, "vkDestroySurfaceKHR")
if fn:
fn(self._instance, self._surface, None)
if self._debug_messenger and self._instance:
fn = vk.vkGetInstanceProcAddr(self._instance, "vkDestroyDebugUtilsMessengerEXT")
if fn:
fn(self._instance, self._debug_messenger, None)
if self._instance:
vk.vkDestroyInstance(self._instance, None)