"""Top-level engine entry point."""
from __future__ import annotations
import logging
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any
import numpy as np
import vulkan as vk
if TYPE_CHECKING:
from .gpu.context import GPUContext
from .materials.texture import TextureManager
from .renderer.forward import ForwardRenderer
from .renderer.gpu_batch import GPUBatch
from .renderer.mesh_registry import MeshRegistry
from .renderer.outline_pass import OutlinePass
from ._engine_init import (
create_depth_resources as _create_depth_resources,
create_framebuffers as _create_framebuffers,
destroy_depth_resources as _destroy_depth_resources,
destroy_framebuffers as _destroy_framebuffers,
init_vulkan as _init_vulkan,
recreate_swapchain as _recreate_swapchain,
)
from ._frame_capture import capture_swapchain_frame
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,
)
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_forward_pipeline,
create_line_pipeline,
create_shader_module,
create_textured_quad_pipeline,
create_ui_pipeline,
)
from .gpu.swapchain import Swapchain
from .gpu.sync import FrameSync
from .materials.shader_compiler import compile_shader
from .picking.pick_pass import PickPass
from .platform import resolve_backend
from .renderer.render_target import RenderTarget
__all__ = ["Engine"]
log = logging.getLogger(__name__)
[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,
) -> 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._running = False
self._last_image_index: int = 0
# 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._device: VkDevice | None = None
self._graphics_queue: VkQueue | None = None
self._present_queue: VkQueue | None = None
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)
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: ForwardRenderer | 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
self._next_texture_index = 0
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
# --- Public API ---
@property
def ctx(self) -> GPUContext:
"""GPU context holding device, physical_device, queues, and command pool."""
return self._ctx
@property
def render_pass(self) -> VkRenderPass:
return self._render_pass
@property
def extent(self) -> tuple[int, int]:
return self._swapchain.extent
@property
def shader_dir(self) -> Path:
return SHADER_DIR
@property
def mesh_registry(self) -> MeshRegistry:
"""Get mesh registry (lazy init)."""
if not self._mesh_registry:
from .renderer.mesh_registry import MeshRegistry
self._mesh_registry = MeshRegistry(self._device, self._physical_device)
return self._mesh_registry
@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
@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._device, self._physical_device)
return self._gpu_batch
@property
def renderer(self) -> ForwardRenderer:
"""Get the active renderer (creates forward renderer if none exists)."""
if not self._renderer:
self.create_renderer("forward")
return self._renderer
[docs]
def create_renderer(self, renderer_type: str = "forward") -> ForwardRenderer:
"""Create and initialize a renderer."""
if renderer_type == "forward":
from .renderer.forward import ForwardRenderer
self._renderer = ForwardRenderer(self)
self._renderer.setup()
else:
raise ValueError(f"Unknown renderer type: {renderer_type}")
return self._renderer
@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
@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."""
self._texture_descriptor_pool = create_texture_descriptor_pool(self._device, self._max_textures)
self._texture_descriptor_layout = create_texture_descriptor_layout(self._device, self._max_textures)
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)
[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)
self._user_render_targets.append(rt)
return rt
[docs]
def register_texture(self, image_view: VkImageView) -> int:
"""Register a texture (image view) into the bindless array. Returns the texture index."""
if not self._texture_descriptor_set:
self._init_texture_system()
idx = self._next_texture_index
write_texture_descriptor(
self._device,
self._texture_descriptor_set,
idx,
image_view,
self._default_sampler,
)
self._next_texture_index += 1
return idx
[docs]
def upload_texture_pixels(self, pixels: np.ndarray, width: int, height: int) -> int:
"""Upload raw RGBA pixel data to GPU. Returns the bindless texture index."""
from .gpu.memory import upload_image_data
image, memory = upload_image_data(
self._device, self._physical_device, self._graphics_queue, self._cmd_ctx.pool,
np.ascontiguousarray(pixels), width, height,
)
view_info = vk.VkImageViewCreateInfo(
image=image,
viewType=vk.VK_IMAGE_VIEW_TYPE_2D,
format=vk.VK_FORMAT_R8G8B8A8_UNORM,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
baseMipLevel=0, levelCount=1, baseArrayLayer=0, layerCount=1,
),
)
image_view = vk.vkCreateImageView(self._device, view_info, None)
tex_idx = self.register_texture(image_view)
self._user_images.append((image, memory, image_view))
self._texture_sizes[tex_idx] = (width, height)
return tex_idx
[docs]
def load_texture(self, file_path: str) -> int:
"""Load a PNG/JPG texture from disk. Returns the texture index."""
from .assets.image_loader import load_texture_from_file
image, memory, width, height = load_texture_from_file(
self._device,
self._physical_device,
self._graphics_queue,
self._cmd_ctx.pool,
file_path,
)
view_info = vk.VkImageViewCreateInfo(
image=image,
viewType=vk.VK_IMAGE_VIEW_TYPE_2D,
format=vk.VK_FORMAT_R8G8B8A8_UNORM,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
baseMipLevel=0,
levelCount=1,
baseArrayLayer=0,
layerCount=1,
),
)
image_view = vk.vkCreateImageView(self._device, view_info, None)
tex_idx = self.register_texture(image_view)
self._user_images.append((image, memory, image_view))
self._texture_sizes[tex_idx] = (width, height)
log.debug("Loaded texture %s (%dx%d) → index %d", file_path, width, height, tex_idx)
return tex_idx
[docs]
def load_mesh(self, file_path: str) -> MeshHandle:
"""Load a glTF mesh from disk and register it.
Returns: MeshHandle for use in rendering.
"""
from .assets.mesh_loader import load_gltf
vertices, indices = load_gltf(file_path)
handle = self.mesh_registry.register(vertices, 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_textured_quad_pipeline(
self, vert_module: VkShaderModule, frag_module: VkShaderModule
) -> tuple[VkPipeline, VkPipelineLayout]:
"""Create a textured quad pipeline using the texture descriptor layout. Returns (pipeline, layout)."""
if not self._texture_descriptor_layout:
self._init_texture_system()
pipeline, layout = create_textured_quad_pipeline(
self._device,
vert_module,
frag_module,
self._render_pass,
self._swapchain.extent,
self._texture_descriptor_layout,
)
self._user_pipelines.append((pipeline, layout))
return pipeline, layout
[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 create_forward_pipeline(
self,
vert_module: VkShaderModule,
frag_module: VkShaderModule,
descriptor_layout: VkDescriptorSetLayout,
texture_layout: VkDescriptorSetLayout | None = None,
render_pass: VkRenderPass | None = None,
extent: tuple[int, int] | None = None,
) -> tuple[VkPipeline, VkPipelineLayout]:
"""Create a forward rendering pipeline. Returns (pipeline, pipeline_layout).
If texture_layout is provided, the pipeline uses 2 descriptor set layouts
(set 0 = SSBOs, set 1 = textures).
If render_pass is provided, uses that instead of the engine's main render pass.
"""
rp = render_pass or self._render_pass
ext = extent or self._swapchain.extent
pipeline, layout = create_forward_pipeline(
self._device,
vert_module,
frag_module,
rp,
ext,
descriptor_layout,
texture_layout=texture_layout,
)
self._user_pipelines.append((pipeline, layout))
return pipeline, layout
[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 create_line_pipeline(
self, vert_module: VkShaderModule, frag_module: VkShaderModule
) -> tuple[VkPipeline, VkPipelineLayout]:
"""Create a line rendering pipeline. Returns (pipeline, pipeline_layout)."""
pipeline, layout = create_line_pipeline(
self._device,
vert_module,
frag_module,
self._render_pass,
self._swapchain.extent,
)
self._user_pipelines.append((pipeline, layout))
return pipeline, layout
[docs]
def create_ui_pipeline(
self, vert_module: VkShaderModule, frag_module: VkShaderModule
) -> tuple[VkPipeline, VkPipelineLayout]:
"""Create a solid-colour UI pipeline. Returns (pipeline, pipeline_layout)."""
pipeline, layout = create_ui_pipeline(
self._device,
vert_module,
frag_module,
self._render_pass,
self._swapchain.extent,
)
self._user_pipelines.append((pipeline, layout))
return pipeline, layout
[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()
@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 draw_text(
self,
text: str,
x: float = 10,
y: float = 10,
font: str | None = None,
size: float = 24.0,
colour: tuple[float, ...] = (1.0, 1.0, 1.0, 1.0),
) -> None:
"""Draw 2D overlay text (call each frame).
Args:
text: String to render.
x, y: Top-left position in pixels.
font: Path to .ttf file (auto-detects system font if None).
size: Text size in pixels.
colour: RGBA colour tuple.
"""
if self._renderer:
self._renderer.submit_text(text, x, y, font_path=font, size=size, colour=colour)
[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.
"""
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 get_cursor_pos(self) -> tuple[float, float]:
"""Get 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,
) -> 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.
"""
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
if setup:
setup()
self._running = True
try:
while self._running and not self._window.should_close():
self._window.poll_events()
# Skip rendering while paused (mobile background)
if getattr(self._window, "paused", False):
continue
if callback:
callback()
self._draw_frame()
finally:
if self._device:
vk.vkDeviceWaitIdle(self._device)
self.shutdown()
def _init_vulkan(self, use_triangle: bool = True) -> None:
_init_vulkan(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 _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.get_all()
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 _draw_frame(self) -> None:
self._sync.wait_and_reset()
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
self._last_image_index = image_index
self._sync.wait_for_image(image_index)
self._sync.mark_image(image_index)
cmd = self._cmd_buffers[frame]
vk.vkResetCommandBuffer(cmd, 0)
begin_info = vk.VkCommandBufferBeginInfo()
vk.vkBeginCommandBuffer(cmd, begin_info)
# 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)
# Use per-image render_finished semaphore to avoid semaphore reuse hazard
submit_info = vk.VkSubmitInfo(
waitSemaphoreCount=1,
pWaitSemaphores=[self._sync.image_available[frame]],
pWaitDstStageMask=[vk.VK_PIPELINE_STAGE_COLOR_ATTACHMENT_OUTPUT_BIT],
commandBufferCount=1,
pCommandBuffers=[cmd],
signalSemaphoreCount=1,
pSignalSemaphores=[self._sync.render_finished[image_index]],
)
vk.vkQueueSubmit(self._graphics_queue, 1, [submit_info], self._sync.fences[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
self._sync.advance()
[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,
)
[docs]
def set_window_size(self, width: int, height: int) -> None:
"""Programmatically resize the window."""
if self._window:
self._window.set_window_size(width, height)
[docs]
def shutdown(self) -> None:
"""Clean up all resources."""
if not self._device:
return
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()
if self._sync:
self._sync.destroy()
if self._cmd_ctx:
self._cmd_ctx.destroy()
# Clean up user-created resources
for pipeline, layout in self._user_pipelines:
vk.vkDestroyPipeline(self._device, pipeline, None)
vk.vkDestroyPipelineLayout(self._device, layout, None)
for module in self._user_shader_modules:
vk.vkDestroyShaderModule(self._device, module, None)
for layout in self._user_descriptor_layouts:
vk.vkDestroyDescriptorSetLayout(self._device, layout, None)
for pool in self._user_descriptor_pools:
vk.vkDestroyDescriptorPool(self._device, pool, None)
for buf, mem in self._user_buffers:
vk.vkDestroyBuffer(self._device, buf, None)
vk.vkFreeMemory(self._device, mem, None)
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)
# Clean up texture system
for sampler in self._user_samplers:
vk.vkDestroySampler(self._device, sampler, None)
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)
# Clean up built-in triangle resources
if self._pipeline:
vk.vkDestroyPipeline(self._device, self._pipeline, None)
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(self._device, self._pipeline_layout, None)
if self._vert_module:
vk.vkDestroyShaderModule(self._device, self._vert_module, None)
if self._frag_module:
vk.vkDestroyShaderModule(self._device, self._frag_module, None)
if self._render_pass:
vk.vkDestroyRenderPass(self._device, self._render_pass, None)
if self._swapchain:
self._swapchain.destroy()
if self._surface:
fn = vk.vkGetInstanceProcAddr(self._instance, "vkDestroySurfaceKHR")
if fn:
fn(self._instance, self._surface, None)
if self._debug_messenger:
fn = vk.vkGetInstanceProcAddr(self._instance, "vkDestroyDebugUtilsMessengerEXT")
if fn:
fn(self._instance, self._debug_messenger, None)
if self._device:
vk.vkDestroyDevice(self._device, None)
if self._instance:
vk.vkDestroyInstance(self._instance, None)
if self._window:
self._window.destroy()
log.debug("Engine shutdown complete")