"""Forward renderer — implements the Renderer ABC for the engine backend."""
from __future__ import annotations
import logging
from typing import Any
import numpy as np
import vulkan as vk
from .._types import (
LIGHT_DTYPE,
MATERIAL_DTYPE,
TRANSFORM_DTYPE,
MeshHandle,
)
from ..gpu.descriptors import (
allocate_descriptor_set,
create_descriptor_pool,
create_ssbo_layout,
write_image_descriptor,
write_ssbo_descriptor,
)
from ..gpu.memory import create_buffer, upload_numpy
from ..gpu.pipeline import (
create_forward_pipeline,
create_shader_module,
create_skinned_pipeline,
create_transparent_pipeline,
)
from ..materials.shader_compiler import compile_shader
from ..scene.frustum import Frustum
from ._base import Renderer
from .custom_post_process import CustomPostProcessPass
from .draw2d_pass import Draw2DPass
from .environment_sync import EnvironmentSync
from .fog_pass import FogPass
from .gizmo_pass import GizmoPass, GizmoRenderData
from .gpu_batch import GPUBatch
from .grid_pass import GridPass
from .light2d_pass import Light2DPass
from .mesh_registry import MeshRegistry
from .overlay_renderer import OverlayRenderer
from .particle_compute import ParticleCompute
from .particle_pass import ParticlePass
from .post_process import PostProcessPass
from .scene_renderer import SceneContentRenderer
from .shadow_renderer import ShadowRenderer
from .skybox_pass import SkyboxPass
from .ssao_pass import SSAOPass
from .text_pass import TextPass
from .tilemap_pass import TileMapPass
from .viewport_manager import ViewportManager
__all__ = ["ForwardRenderer"]
log = logging.getLogger(__name__)
[docs]
class ForwardRenderer(Renderer):
"""Forward renderer with multi-draw indirect, per-viewport frustum culling."""
def __init__(self, engine: Any, max_objects: int = 10_000):
self._engine = engine
self._max_objects = max_objects
# Subsystems
self.viewport_manager = ViewportManager()
self._frustum = Frustum()
# Delegated renderers (created after __init__ state is set up)
self._scene_renderer = SceneContentRenderer(self)
self._shadow_renderer = ShadowRenderer(self)
self._overlay_renderer = OverlayRenderer(self)
self._env_sync = EnvironmentSync(self)
# Per-frame submission lists
self._instances: list[tuple[MeshHandle, np.ndarray, int, int]] = []
self._dynamic_draws: list[tuple[np.ndarray, np.ndarray, np.ndarray, int, int]] = []
# GPU resources (created in setup)
self._pipeline: Any = None
self._pipeline_layout: Any = None
self._ssbo_layout: Any = None
self._ssbo_pool: Any = None
self._ssbo_set: Any = None
self._transform_buf: Any = None
self._transform_mem: Any = None
self._material_buf: Any = None
self._material_mem: Any = None
self._light_buf: Any = None
self._light_mem: Any = None
self._batch: GPUBatch | None = None
self._vert_module: Any = None
self._frag_module: Any = None
# Material data (set externally via set_materials)
self._materials: np.ndarray = np.zeros(1, dtype=MATERIAL_DTYPE)
self._lights: np.ndarray = np.zeros(1, dtype=LIGHT_DTYPE)
# Dirty-tracking: skip redundant GPU uploads when data hasn't changed
self._materials_hash: int = 0
self._lights_hash: int = 0
# Text rendering
self._text_pass: TextPass | None = None
self._text_renderer: Any = None
# 2D drawing pass
self._draw2d_pass: Draw2DPass | None = None
# 2D lighting pass
self._light2d_pass: Light2DPass | None = None
# Particle rendering
self._particle_pass: ParticlePass | None = None
self._particle_submissions: list[tuple[np.ndarray, int]] = [] # (data, count)
self._gpu_particle_submissions: list[dict] = [] # emitter_config dicts
self._particle_compute: ParticleCompute | None = None
# TileMap rendering
self._tilemap_pass: TileMapPass | None = None
# Post-processing
self._post_process: PostProcessPass | None = None
self._custom_pp: CustomPostProcessPass | None = None
# Grid overlay (editor)
self._grid_pass: GridPass | None = None
# SSAO
self._ssao_pass: SSAOPass | None = None
# Fog
self._fog_pass: FogPass | None = None
# Skybox + IBL
self._skybox_pass: SkyboxPass | None = None
self._ibl_enabled: bool = False
self._placeholder_cubemap_view: Any = None
self._placeholder_cubemap_sampler: Any = None
self._placeholder_cubemap_img: Any = None
self._placeholder_cubemap_mem: Any = None
# Shadow mapping
self._shadow_pass: Any = None
self._point_shadow_pass: Any = None
self._shadow_buf: Any = None
self._shadow_mem: Any = None
# Skinned mesh rendering
self._skinned_pipeline: Any = None
self._skinned_pipeline_layout: Any = None
self._skinned_vert_module: Any = None
self._joint_layout: Any = None
self._joint_pool: Any = None
self._joint_set: Any = None
self._joint_buf: Any = None
self._joint_mem: Any = None
self._skinned_instances: list[tuple[MeshHandle, np.ndarray, int, np.ndarray]] = []
# (mesh_handle, transform, material_id, joint_matrices)
# Gizmo overlay
self._gizmo_pass: GizmoPass | None = None
self._gizmo_render_data: GizmoRenderData | None = None
# Debug line rendering
self._debug_pipeline: Any = None
self._debug_pipeline_layout: Any = None
self._debug_vb: Any = None
self._debug_vb_mem: Any = None
self._debug_vb_capacity: int = 0
self._debug_vert_module: Any = None
self._debug_frag_module: Any = None
# Track whether HDR was rendered this frame (guards tonemap)
self._hdr_rendered = False
# Double-sided opaque rendering (no backface culling, depth-write on)
self._nocull_pipeline: Any = None
self._nocull_pipeline_layout: Any = None
# Transparent object rendering (alpha blending, depth-write off)
self._transparent_pipeline: Any = None
self._transparent_pipeline_layout: Any = None
self._ready = False
[docs]
def setup(self) -> None:
"""Initialize GPU resources — called once after engine Vulkan init."""
e = self._engine
device = e.ctx.device
phys = e.ctx.physical_device
# SSBOs
transform_size = self._max_objects * TRANSFORM_DTYPE.itemsize
self._transform_buf, self._transform_mem = create_buffer(
device,
phys,
transform_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
self._max_materials = 1024
material_size = self._max_materials * MATERIAL_DTYPE.itemsize
self._material_buf, self._material_mem = create_buffer(
device,
phys,
material_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
light_size = 256 * LIGHT_DTYPE.itemsize # Pre-allocate for 256 lights
self._light_buf, self._light_mem = create_buffer(
device,
phys,
light_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
upload_numpy(device, self._material_mem, self._materials)
upload_numpy(device, self._light_mem, self._lights)
# Shadow data SSBO: CSM (224) + point/spot shadow data (112) = 336 bytes
shadow_data_size = 336
self._shadow_buf, self._shadow_mem = create_buffer(
device,
phys,
shadow_data_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
# Initialize with no-shadow sentinels
init_shadow = np.zeros(shadow_data_size, dtype=np.uint8)
sentinel = np.array([0xFF, 0xFF, 0xFF, 0xFF], dtype=np.uint8)
init_shadow[208:212] = sentinel # shadow_tex_index
init_shadow[220:224] = sentinel # point_shadow_tex
init_shadow[224:228] = sentinel # spot_shadow_tex
upload_numpy(device, self._shadow_mem, init_shadow)
# Descriptors: 4 SSBOs (transforms, materials, lights, shadow), 1 cubemap sampler (IBL),
# 2 trailing SSBOs (tile light indices + tile info for Forward+ culling)
self._ssbo_layout = create_ssbo_layout(device, binding_count=4, extra_samplers=1, trailing_ssbos=2)
self._ssbo_pool = create_descriptor_pool(device, max_sets=1, extra_samplers=1, ssbo_count=6)
self._ssbo_set = allocate_descriptor_set(device, self._ssbo_pool, self._ssbo_layout)
write_ssbo_descriptor(device, self._ssbo_set, 0, self._transform_buf, transform_size)
write_ssbo_descriptor(device, self._ssbo_set, 1, self._material_buf, material_size)
write_ssbo_descriptor(device, self._ssbo_set, 2, self._light_buf, light_size)
write_ssbo_descriptor(device, self._ssbo_set, 3, self._shadow_buf, 336)
# Placeholder buffers for Forward+ tile culling (bindings 5-6); real buffers set by LightCullPass
self._tile_light_idx_buf, self._tile_light_idx_mem = create_buffer(
device, phys, 16, vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
self._tile_info_buf, self._tile_info_mem = create_buffer(
device, phys, 16, vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
write_ssbo_descriptor(device, self._ssbo_set, 5, self._tile_light_idx_buf, 16)
write_ssbo_descriptor(device, self._ssbo_set, 6, self._tile_info_buf, 16)
# IBL cubemap descriptor (binding 4) — placeholder until set_skybox() is called
self._ibl_enabled = False
from ..assets.cubemap_loader import load_cubemap
(
self._placeholder_cubemap_view, self._placeholder_cubemap_sampler,
self._placeholder_cubemap_img, self._placeholder_cubemap_mem,
) = load_cubemap(
device, phys, e.ctx.graphics_queue, e.ctx.command_pool, colour=(0.0, 0.0, 0.0)
)
write_image_descriptor(
device, self._ssbo_set, 4, self._placeholder_cubemap_view, self._placeholder_cubemap_sampler
)
# Pipeline
shader_dir = e.shader_dir
vert_spv = compile_shader(shader_dir / "cube.vert")
frag_spv = compile_shader(shader_dir / "cube_textured.frag")
self._vert_module = create_shader_module(device, vert_spv)
self._frag_module = create_shader_module(device, frag_spv)
tex_layout = e.texture_descriptor_layout
self._pipeline, self._pipeline_layout = create_forward_pipeline(
device,
self._vert_module,
self._frag_module,
e.render_pass,
e.extent,
self._ssbo_layout,
texture_layout=tex_layout,
)
# Double-sided opaque pipeline (no backface culling, same depth/blend as forward)
self._nocull_pipeline, self._nocull_pipeline_layout = create_forward_pipeline(
device,
self._vert_module,
self._frag_module,
e.render_pass,
e.extent,
self._ssbo_layout,
texture_layout=tex_layout,
double_sided=True,
)
# Transparent pipeline (alpha blending, no depth write)
self._transparent_pipeline, self._transparent_pipeline_layout = create_transparent_pipeline(
device,
self._vert_module,
self._frag_module,
e.render_pass,
e.extent,
self._ssbo_layout,
texture_layout=tex_layout,
)
# Batch renderers — separate indirect buffers so opaque/transparent
# don't overwrite each other before the GPU executes draw commands.
use_mdi = getattr(e, "_has_mdi", True)
self._batch = GPUBatch(device, phys, max_draws=self._max_objects, use_mdi=use_mdi)
self._transparent_batch = GPUBatch(device, phys, max_draws=1000, use_mdi=use_mdi)
# Skinned pipeline (joint matrix SSBO at set 2, binding 0)
max_joints = 256 # Max bones across all skinned meshes
joint_buf_size = max_joints * 64 # 256 * mat4(64 bytes) = 16KB
self._joint_buf, self._joint_mem = create_buffer(
device,
phys,
joint_buf_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
self._joint_layout = create_ssbo_layout(device, binding_count=1)
self._joint_pool = create_descriptor_pool(device, max_sets=2)
self._joint_set = allocate_descriptor_set(device, self._joint_pool, self._joint_layout)
write_ssbo_descriptor(device, self._joint_set, 0, self._joint_buf, joint_buf_size)
skinned_vert_spv = compile_shader(shader_dir / "skinned.vert")
self._skinned_vert_module = create_shader_module(device, skinned_vert_spv)
self._skinned_pipeline, self._skinned_pipeline_layout = create_skinned_pipeline(
device,
self._skinned_vert_module,
self._frag_module,
e.render_pass,
e.extent,
self._ssbo_layout,
texture_layout=tex_layout,
joint_layout=self._joint_layout,
)
# Grid overlay (editor viewport)
self._grid_pass = GridPass(self._engine)
self._grid_pass.setup()
# Shadow mapping pass (directional CSM)
from .shadow_pass import ShadowPass
self._shadow_pass = ShadowPass(self._engine)
self._shadow_pass.setup(self._ssbo_layout)
# Point/spot shadow pass
from .point_shadow_pass import PointShadowPass
self._point_shadow_pass = PointShadowPass(self._engine)
self._point_shadow_pass.setup(self._ssbo_layout)
# Particle pass
self._particle_pass = ParticlePass(self._engine)
self._particle_pass.setup()
# GPU particle compute (lazy-initialized on first submission)
self._particle_compute = None
# TileMap pass
self._tilemap_pass = TileMapPass(self._engine)
self._tilemap_pass.setup()
# 2D lighting pass
self._light2d_pass = Light2DPass(self._engine)
self._light2d_pass.setup()
# Text rendering pass (shared atlas owner)
self._text_pass = TextPass(self._engine)
self._text_pass.setup()
# 2D drawing pass — shares TextPass for text pipeline/atlas
self._draw2d_pass = Draw2DPass(self._engine, text_pass=self._text_pass)
self._draw2d_pass.setup()
# Text renderer (shared across frames, caches atlases)
from ..text_renderer import get_shared_text_renderer
self._text_renderer = get_shared_text_renderer()
# Post-processing (HDR → tone mapping + FXAA)
self._post_process = PostProcessPass(self._engine)
self._post_process.setup()
# Custom user post-process effects (runs after built-in post-processing)
self._custom_pp = CustomPostProcessPass(self._engine)
self._custom_pp.setup()
# When post-processing is enabled, 3D content renders into the HDR target
# which uses R16G16B16A16_SFLOAT. Recreate all pipelines that draw inside
# the HDR render pass so the formats match.
if self._post_process.enabled and self._post_process.hdr_target:
hdr_rp = self._post_process.hdr_target.render_pass
self._rebuild_3d_pipelines(hdr_rp)
if self._particle_pass:
self._particle_pass.rebuild_pipeline(hdr_rp)
# Gizmo overlay pass
self._gizmo_pass = GizmoPass(self._engine)
self._gizmo_pass.setup()
# SSAO (needs HDR target depth view + image, compute shader may not be available)
if self._post_process.enabled and self._post_process.hdr_target:
try:
hdr_rt = self._post_process.hdr_target
w, h = e.extent
self._ssao_pass = SSAOPass(self._engine)
self._ssao_pass.setup(w, h, hdr_rt.depth_view, hdr_rt.depth_image)
self._post_process.update_ssao_descriptor(self._ssao_pass.ao_view)
except Exception as exc:
log.warning("SSAO disabled: %s", exc)
self._ssao_pass = None
# Fog (compute pass on HDR colour image, needs depth + colour views)
if self._post_process.enabled and self._post_process.hdr_target:
try:
hdr_rt = self._post_process.hdr_target
w, h = e.extent
self._fog_pass = FogPass(self._engine)
self._fog_pass.setup(w, h, hdr_rt.depth_view, hdr_rt.color_view, hdr_rt.color_image)
except Exception as exc:
log.warning("Fog pass disabled: %s", exc)
self._fog_pass = None
self._ready = True
[docs]
def set_skybox(self, cubemap_view: Any, cubemap_sampler: Any) -> None:
"""Set a cubemap as the skybox and enable IBL."""
self._skybox_pass = SkyboxPass(self._engine)
self._skybox_pass.setup(cubemap_view, cubemap_sampler)
# Write cubemap to SSBO set binding 4 for IBL
write_image_descriptor(
self._engine.ctx.device,
self._ssbo_set,
4,
cubemap_view,
cubemap_sampler,
)
self._ibl_enabled = True
@property
def post_processing(self) -> PostProcessPass | None:
"""Access post-processing pass for configuration."""
return self._post_process
@property
def custom_post_processing(self) -> CustomPostProcessPass | None:
"""Access custom user post-process pass for configuration."""
return self._custom_pp
[docs]
def set_gizmo_data(self, data: GizmoRenderData | None) -> None:
"""Set gizmo render data for the current frame (or None to hide)."""
self._gizmo_render_data = data
@property
def ssao(self) -> SSAOPass | None:
"""Access SSAO pass for configuration."""
return self._ssao_pass
@property
def fog(self) -> FogPass | None:
"""Access fog pass for configuration."""
return self._fog_pass
[docs]
def set_materials(self, materials: np.ndarray) -> None:
"""Set material array and upload to GPU (skips if unchanged)."""
if len(materials) > self._max_materials:
log.warning("Material count (%d) exceeds max (%d), clamping", len(materials), self._max_materials)
materials = materials[: self._max_materials]
self._materials = materials
if self._material_mem:
h = hash(materials.tobytes())
if h == self._materials_hash:
return
self._materials_hash = h
upload_numpy(self._engine.ctx.device, self._material_mem, materials)
[docs]
def set_lights(self, lights: np.ndarray) -> None:
"""Set light array and upload to GPU (skips if unchanged).
Prepends uint32 light_count to match GLSL LightBuffer layout:
[uint32 count][Light[0]][Light[1]]...
"""
self._lights = lights
if self._light_mem:
h = hash(lights.tobytes())
if h == self._lights_hash:
return
self._lights_hash = h
count = np.array([len(lights)], dtype=np.uint32)
# Pad count to 16-byte alignment (vec4 boundary)
padding = np.zeros(3, dtype=np.uint32)
header = np.concatenate([count, padding])
buf = np.concatenate([header.view(np.uint8), lights.view(np.uint8)])
upload_numpy(self._engine.ctx.device, self._light_mem, buf)
[docs]
def submit_text(
self,
text: str,
x: float,
y: float,
font_path: str | None = None,
size: float = 24.0,
colour: tuple = (1.0, 1.0, 1.0, 1.0),
) -> None:
"""Submit text for 2D overlay rendering."""
if self._text_renderer:
self._text_renderer.draw_text(text, x, y, font_path=font_path, size=size, colour=colour)
# --- Renderer ABC ---
[docs]
def init(self, device: Any, swapchain: Any) -> None:
"""Initialize (called by ABC contract — use setup() instead)."""
self.setup()
[docs]
def begin_frame(self) -> Any:
"""Begin frame — clear submission lists."""
self._instances.clear()
self._dynamic_draws.clear()
self._particle_submissions.clear()
self._gpu_particle_submissions.clear()
self._skinned_instances.clear()
if self._tilemap_pass:
self._tilemap_pass.begin_frame()
if self._text_renderer:
self._text_renderer.begin_frame()
if self._light2d_pass:
self._light2d_pass.begin_frame()
return None # Command buffer managed by engine
[docs]
def submit_instance(
self,
mesh_handle: MeshHandle,
transform: np.ndarray,
material_id: int = 0,
viewport_id: int = 0,
) -> None:
"""Submit a mesh instance for rendering this frame."""
self._instances.append((mesh_handle, transform, material_id, viewport_id))
[docs]
def submit_multimesh(
self,
mesh_handle: MeshHandle,
transforms: np.ndarray,
material_id: int = 0,
material_ids: np.ndarray | None = None,
viewport_id: int = 0,
) -> None:
"""Bulk-submit many instances of the same mesh — avoids per-instance Python loops.
Args:
mesh_handle: Shared mesh for all instances.
transforms: (N, 4, 4) float32 array of model matrices.
material_id: Material index for all instances (ignored if *material_ids* given).
material_ids: Optional (N,) uint32 array of per-instance material indices.
viewport_id: Viewport index.
"""
n = transforms.shape[0]
if material_ids is not None:
for i in range(n):
self._instances.append((mesh_handle, transforms[i], int(material_ids[i]), viewport_id))
else:
for i in range(n):
self._instances.append((mesh_handle, transforms[i], material_id, viewport_id))
[docs]
def submit_particles(self, particle_data: np.ndarray) -> None:
"""Submit particle data for rendering this frame."""
self._particle_submissions.append((particle_data, len(particle_data)))
[docs]
def submit_gpu_particles(self, emitter_config: dict) -> None:
"""Submit a GPU particle emitter config for compute-shader simulation this frame."""
self._gpu_particle_submissions.append(emitter_config)
[docs]
def submit_light2d(self, **kwargs) -> None:
"""Submit a 2D light for this frame (forwarded to Light2DPass)."""
if self._light2d_pass:
self._light2d_pass.submit_light(**kwargs)
[docs]
def submit_occluder2d(self, polygon_vertices: list[tuple[float, float]]) -> None:
"""Submit a 2D occluder polygon for shadow casting this frame."""
if self._light2d_pass:
self._light2d_pass.submit_occluder(polygon_vertices)
[docs]
def submit_skinned_instance(
self,
mesh_handle: MeshHandle,
transform: np.ndarray,
material_id: int,
joint_matrices: np.ndarray,
) -> None:
"""Submit a skinned mesh instance with joint matrices for this frame."""
self._skinned_instances.append((mesh_handle, transform, material_id, joint_matrices))
[docs]
def submit_dynamic(
self,
vertices: np.ndarray,
indices: np.ndarray,
transform: np.ndarray,
material_id: int = 0,
viewport_id: int = 0,
) -> None:
"""Submit dynamic geometry (uploaded and drawn this frame only)."""
self._dynamic_draws.append((vertices, indices, transform, material_id, viewport_id))
def _upload_transforms(self) -> None:
"""Upload ALL instance transforms to the SSBO once per frame.
Both shadow and main passes reference the same indices, avoiding
CPU-overwrite-before-GPU-execute race conditions.
"""
if not self._instances:
return
count = min(len(self._instances), self._max_objects)
if len(self._instances) > self._max_objects:
log.warning(
"Instance count (%d) exceeds max_objects (%d), clamping", len(self._instances), self._max_objects
)
instances = self._instances[:count]
# Gather all transforms into a contiguous (N,4,4) array
model_mats = np.empty((count, 4, 4), dtype=np.float32)
mat_ids = np.empty(count, dtype=np.uint32)
for i, (_mh, xform, mid, _vp) in enumerate(instances):
model_mats[i] = xform if xform.shape == (4, 4) else xform.T
mat_ids[i] = mid
# Transpose all at once: (N,4,4) -> column-major for GPU
model_mats_T = np.ascontiguousarray(model_mats.transpose(0, 2, 1))
# Batch normal matrix: inv(upper-left 3x3).T, packed into 4x4
m3x3 = model_mats[:, :3, :3]
try:
inv3x3 = np.linalg.inv(m3x3) # (N,3,3)
except np.linalg.LinAlgError:
# Fallback: per-instance with singular check
inv3x3 = np.empty_like(m3x3)
for j in range(count):
try:
inv3x3[j] = np.linalg.inv(m3x3[j])
except np.linalg.LinAlgError:
inv3x3[j] = m3x3[j].T
# normal_mat = transpose(inverse(M3x3)), stored column-major in 4x4
# inv3x3 is already inv(M), we need inv(M).T then column-major transpose
# GPU wants column-major of (inv(M3x3).T) = column-major(inv(M3x3).T)
# column-major of A = A.T in row-major, so: (inv3x3.T).T = inv3x3
normal4x4 = np.zeros((count, 4, 4), dtype=np.float32)
normal4x4[:, 3, 3] = 1.0
# We need: GPU column-major normal_mat = transpose(inv(model_3x3).T)
# In row-major storage that's: inv(model_3x3)
normal4x4[:, :3, :3] = inv3x3
transforms = np.zeros(count, dtype=TRANSFORM_DTYPE)
transforms["model"] = model_mats_T
transforms["normal_mat"] = normal4x4
transforms["material_index"] = mat_ids
upload_numpy(self._engine.ctx.device, self._transform_mem, transforms)
def _set_hdr_flag(self, enabled: bool) -> None:
"""Set hdr_output (byte offset 216) flag in shadow SSBO."""
flag = np.array([1 if enabled else 0], dtype=np.uint32)
shadow_data = np.zeros(336, dtype=np.uint8)
ffi = vk.ffi
device = self._engine.ctx.device
src = vk.vkMapMemory(device, self._shadow_mem, 0, 336, 0)
ffi.memmove(ffi.cast("void*", shadow_data.ctypes.data), src, 336)
vk.vkUnmapMemory(device, self._shadow_mem)
shadow_data[216:220] = flag.view(np.uint8)
upload_numpy(device, self._shadow_mem, shadow_data)
# --- Delegation methods (preserve backward compatibility for tests/examples) ---
def _sync_world_environment(self) -> None:
"""Delegate to EnvironmentSync."""
self._env_sync.sync_world_environment()
def _render_scene_content(self, cmd: Any) -> None:
"""Delegate to SceneContentRenderer."""
self._scene_renderer.render_scene_content(cmd)
def _render_shadows(self, cmd: Any, registry: Any) -> None:
"""Delegate to ShadowRenderer."""
self._shadow_renderer.render_shadows(cmd, registry)
def _render_point_spot_shadows(self, cmd: Any, registry: Any) -> None:
"""Delegate to ShadowRenderer."""
self._shadow_renderer.render_point_spot_shadows(cmd, registry)
def _render_debug_lines(self, cmd: Any, extent: tuple[int, int]) -> None:
"""Delegate to OverlayRenderer."""
self._overlay_renderer.render_debug_lines(cmd, extent)
def _render_text(self, cmd: Any, extent: tuple[int, int]) -> None:
"""Delegate to OverlayRenderer."""
self._overlay_renderer.render_text(cmd, extent)
def _render_particles(self, cmd: Any, extent: tuple[int, int]) -> None:
"""Delegate to OverlayRenderer."""
self._overlay_renderer.render_particles(cmd, extent)
def _dispatch_gpu_particles(self, cmd: Any) -> None:
"""Delegate to OverlayRenderer."""
self._overlay_renderer.dispatch_gpu_particles(cmd)
def _run_custom_post_process(self, cmd: Any, pp: PostProcessPass) -> None:
"""Delegate to EnvironmentSync."""
self._env_sync.run_custom_post_process(cmd, pp)
def _gather_post_process_effects(self) -> list:
"""Delegate to EnvironmentSync."""
return self._env_sync._gather_post_process_effects()
def _update_tonemap_hdr_input(self, new_hdr_view: Any) -> None:
"""Delegate to EnvironmentSync."""
self._env_sync.update_tonemap_hdr_input(new_hdr_view)
[docs]
def pre_render(self, cmd: Any) -> None:
"""Record offscreen passes (shadow maps, HDR) before main render pass begins."""
if not self._ready:
return
# Sync WorldEnvironment properties to renderer
self._env_sync.sync_world_environment()
# Dispatch GPU particle compute shaders (must happen outside render pass)
if self._gpu_particle_submissions:
self._overlay_renderer.dispatch_gpu_particles(cmd)
# Upload all transforms ONCE — shared by shadow and main passes
self._upload_transforms()
# Upload MSDF atlas outside render pass (staging transfers not allowed inside)
# Single upload serves both TextPass (3D overlay) and Draw2DPass (2D UI text)
if self._text_pass:
self._text_pass.upload_atlas_if_dirty()
# Track whether HDR content was rendered this frame
self._hdr_rendered = False
# Update IBL flag in shadow buffer (even without shadow pass)
if self._ibl_enabled and not self._shadow_pass:
shadow_data = np.zeros(336, dtype=np.uint8)
sentinel = np.array([0xFF, 0xFF, 0xFF, 0xFF], dtype=np.uint8)
shadow_data[208:212] = sentinel
shadow_data[220:224] = sentinel
shadow_data[224:228] = sentinel
shadow_data[212:216] = np.array([1], dtype=np.uint32).view(np.uint8)
pp = self._post_process
hdr_flag = 1 if (pp and pp.enabled) else 0
shadow_data[216:220] = np.array([hdr_flag], dtype=np.uint32).view(np.uint8)
upload_numpy(self._engine.ctx.device, self._shadow_mem, shadow_data)
# Render 2D lights to accumulation texture
if self._light2d_pass and self._light2d_pass.has_lights:
self._light2d_pass.render(cmd, self._engine.extent)
if self._shadow_pass and self._instances:
self._shadow_renderer.render_shadows(cmd, self._engine.mesh_registry)
if self._point_shadow_pass and self._instances:
self._shadow_renderer.render_point_spot_shadows(cmd, self._engine.mesh_registry)
# When post-processing is enabled, render 3D scene to HDR target here
pp = self._post_process
if pp and pp.enabled and self._instances:
# Update camera matrices in UBO (needed by motion blur AND fog depth reconstruction)
viewports = self.viewport_manager.get_all()
if viewports:
_, vp = viewports[0]
pp.update_motion_blur_matrices(vp.camera_view, vp.camera_proj)
# Set hdr_output flag so fragment shader skips tone mapping
self._set_hdr_flag(True)
pp.begin_hdr_pass(cmd)
self._scene_renderer.render_scene_content(cmd)
pp.end_hdr_pass(cmd)
self._hdr_rendered = True
# Run bloom pass (extract + blur) between HDR and tonemap
if pp.bloom_enabled:
pp.render_bloom(cmd)
# Run SSAO after HDR pass (needs depth from HDR target)
if self._ssao_pass:
pp.ssao_enabled = self._ssao_pass.enabled
if self._ssao_pass.enabled:
viewports = self.viewport_manager.get_all()
if viewports:
_, vp = viewports[0]
self._ssao_pass.render(cmd, vp.camera_proj)
# Fog is now applied post-tonemap in the fullscreen pass (see tonemap.frag).
# The fog compute pass is kept for compatibility but not dispatched.
# Run custom user post-process effects (after bloom/SSAO, before tonemap)
self._env_sync.run_custom_post_process(cmd, pp)
[docs]
def render(self, cmd: Any) -> None:
"""Record draw commands for all viewports."""
if not self._ready:
return
e = self._engine
pp = self._post_process
# If post-processing rendered HDR content in pre_render, tonemap it now.
# Skip tonemap when no 3D content was rendered (e.g. editor with only UI nodes)
# to avoid sampling an undefined/stale HDR render target.
if pp and pp.enabled and self._hdr_rendered:
pp.render_tonemap(cmd, e.extent[0], e.extent[1])
# Restore tonemap's HDR input to the original HDR target for next frame
if self._custom_pp and self._custom_pp.has_effects and pp.hdr_target:
self._env_sync.update_tonemap_hdr_input(pp.hdr_target.color_view)
elif not (pp and pp.enabled):
# Direct rendering to swapchain (no post-processing)
self._scene_renderer.render_scene_content(cmd)
# 2D overlays always go to swapchain
# Render 2D drawing overlay — pass window size for UI coordinate conversion
if self._draw2d_pass:
win = self._engine._window
ws = win.get_window_size() if win and hasattr(win, "get_window_size") else None
if ws:
self._draw2d_pass.render(cmd, e.extent[0], e.extent[1], ws[0], ws[1])
else:
self._draw2d_pass.render(cmd, e.extent[0], e.extent[1])
# Render text overlay (2D)
if self._text_renderer and self._text_renderer.has_text and self._text_pass:
self._overlay_renderer.render_text(cmd, e.extent)
def _rebuild_3d_pipelines(self, render_pass: Any) -> None:
"""Recreate forward/transparent/skinned pipelines against a different render pass.
Called when post-processing is enabled and the HDR target's render pass
(R16G16B16A16_SFLOAT) differs from the swapchain render pass (B8G8R8A8_SRGB).
"""
e = self._engine
device = e.ctx.device
tex_layout = e.texture_descriptor_layout
extent = e.extent
if self._pipeline:
vk.vkDestroyPipeline(device, self._pipeline, None)
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._pipeline_layout, None)
self._pipeline, self._pipeline_layout = create_forward_pipeline(
device, self._vert_module, self._frag_module, render_pass,
extent, self._ssbo_layout, texture_layout=tex_layout,
)
if self._nocull_pipeline:
vk.vkDestroyPipeline(device, self._nocull_pipeline, None)
if self._nocull_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._nocull_pipeline_layout, None)
self._nocull_pipeline, self._nocull_pipeline_layout = create_forward_pipeline(
device, self._vert_module, self._frag_module, render_pass,
extent, self._ssbo_layout, texture_layout=tex_layout, double_sided=True,
)
if self._transparent_pipeline:
vk.vkDestroyPipeline(device, self._transparent_pipeline, None)
if self._transparent_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._transparent_pipeline_layout, None)
self._transparent_pipeline, self._transparent_pipeline_layout = create_transparent_pipeline(
device, self._vert_module, self._frag_module, render_pass,
extent, self._ssbo_layout, texture_layout=tex_layout,
)
if self._skinned_pipeline:
vk.vkDestroyPipeline(device, self._skinned_pipeline, None)
if self._skinned_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._skinned_pipeline_layout, None)
self._skinned_pipeline, self._skinned_pipeline_layout = create_skinned_pipeline(
device, self._skinned_vert_module, self._frag_module, render_pass,
extent, self._ssbo_layout, texture_layout=tex_layout,
joint_layout=self._joint_layout,
)
[docs]
def resize(self, width: int, height: int) -> None:
"""Handle framebuffer resize — recreate pipeline."""
if not self._ready:
return
# Recreate post-processing FIRST so HDR target render pass exists
if self._post_process and self._post_process.enabled:
self._post_process.resize(width, height)
if self._ssao_pass and self._post_process.hdr_target:
hdr_rt = self._post_process.hdr_target
self._ssao_pass.resize(width, height, hdr_rt.depth_view, hdr_rt.depth_image)
self._post_process.update_ssao_descriptor(self._ssao_pass.ao_view)
if self._fog_pass and self._post_process.hdr_target:
hdr_rt = self._post_process.hdr_target
self._fog_pass.resize(width, height, hdr_rt.depth_view, hdr_rt.color_view, hdr_rt.color_image)
if self._custom_pp:
self._custom_pp.resize(width, height)
# Choose the correct render pass for 3D pipelines
pp = self._post_process
if pp and pp.enabled and pp.hdr_target:
rp = pp.hdr_target.render_pass
else:
rp = self._engine.render_pass
e = self._engine
device = e.ctx.device
tex_layout = e.texture_descriptor_layout
if self._pipeline:
vk.vkDestroyPipeline(device, self._pipeline, None)
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._pipeline_layout, None)
self._pipeline, self._pipeline_layout = create_forward_pipeline(
device, self._vert_module, self._frag_module, rp,
(width, height), self._ssbo_layout, texture_layout=tex_layout,
)
if self._nocull_pipeline:
vk.vkDestroyPipeline(device, self._nocull_pipeline, None)
if self._nocull_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._nocull_pipeline_layout, None)
self._nocull_pipeline, self._nocull_pipeline_layout = create_forward_pipeline(
device, self._vert_module, self._frag_module, rp,
(width, height), self._ssbo_layout, texture_layout=tex_layout, double_sided=True,
)
if self._transparent_pipeline:
vk.vkDestroyPipeline(device, self._transparent_pipeline, None)
if self._transparent_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._transparent_pipeline_layout, None)
self._transparent_pipeline, self._transparent_pipeline_layout = create_transparent_pipeline(
device, self._vert_module, self._frag_module, rp,
(width, height), self._ssbo_layout, texture_layout=tex_layout,
)
if self._skinned_pipeline:
vk.vkDestroyPipeline(device, self._skinned_pipeline, None)
if self._skinned_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._skinned_pipeline_layout, None)
self._skinned_pipeline, self._skinned_pipeline_layout = create_skinned_pipeline(
device, self._skinned_vert_module, self._frag_module, rp,
(width, height), self._ssbo_layout, texture_layout=tex_layout,
joint_layout=self._joint_layout,
)
[docs]
def cleanup(self) -> None:
"""Release all GPU resources."""
if not self._ready:
return
device = self._engine.ctx.device
# Batch objects
for batch in (self._batch, self._transparent_batch):
if batch:
batch.destroy()
# Pipeline + layout pairs
for pipeline, layout in [
(self._debug_pipeline, self._debug_pipeline_layout),
(self._nocull_pipeline, self._nocull_pipeline_layout),
(self._skinned_pipeline, self._skinned_pipeline_layout),
(self._transparent_pipeline, self._transparent_pipeline_layout),
(self._pipeline, self._pipeline_layout),
]:
if pipeline:
vk.vkDestroyPipeline(device, pipeline, None)
if layout:
vk.vkDestroyPipelineLayout(device, layout, None)
# Shader modules
for mod in (self._debug_vert_module, self._debug_frag_module, self._skinned_vert_module,
self._vert_module, self._frag_module):
if mod:
vk.vkDestroyShaderModule(device, mod, None)
# Descriptor layouts + pools
for layout in (self._joint_layout, self._ssbo_layout):
if layout:
vk.vkDestroyDescriptorSetLayout(device, layout, None)
for pool in (self._joint_pool, self._ssbo_pool):
if pool:
vk.vkDestroyDescriptorPool(device, pool, None)
# Buffer + memory pairs
for buf, mem in [
(self._debug_vb, self._debug_vb_mem),
(self._joint_buf, self._joint_mem),
(self._transform_buf, self._transform_mem),
(self._material_buf, self._material_mem),
(self._light_buf, self._light_mem),
(self._shadow_buf, self._shadow_mem),
(self._tile_light_idx_buf, self._tile_light_idx_mem),
(self._tile_info_buf, self._tile_info_mem),
]:
if buf:
vk.vkDestroyBuffer(device, buf, None)
if mem:
vk.vkFreeMemory(device, mem, None)
# Placeholder cubemap resources
if self._placeholder_cubemap_sampler:
vk.vkDestroySampler(device, self._placeholder_cubemap_sampler, None)
if self._placeholder_cubemap_view:
vk.vkDestroyImageView(device, self._placeholder_cubemap_view, None)
if self._placeholder_cubemap_img:
vk.vkDestroyImage(device, self._placeholder_cubemap_img, None)
if self._placeholder_cubemap_mem:
vk.vkFreeMemory(device, self._placeholder_cubemap_mem, None)
# Render passes
for pass_obj in (self._gizmo_pass, self._shadow_pass, self._point_shadow_pass,
self._particle_pass, self._particle_compute, self._tilemap_pass,
self._ssao_pass, self._fog_pass, self._custom_pp, self._post_process,
self._grid_pass, self._skybox_pass, self._light2d_pass, self._draw2d_pass,
self._text_pass):
if pass_obj:
pass_obj.cleanup()
self._ready = False
[docs]
def destroy(self) -> None:
"""ABC destroy — delegates to cleanup."""
self.cleanup()
# -- Resource management (delegate to engine) --
[docs]
def register_mesh(self, vertices: np.ndarray, indices: np.ndarray) -> MeshHandle:
"""Register mesh data on GPU via engine's mesh registry."""
return self._engine.mesh_registry.register(vertices, indices)
[docs]
def register_texture(self, pixels: np.ndarray, width: int, height: int) -> int:
"""Upload RGBA pixel data to GPU, return bindless texture index."""
return self._engine.upload_texture_pixels(pixels, width, height)
# -- Frame capture --
[docs]
def capture_frame(self) -> np.ndarray:
"""Capture the last rendered frame as (H, W, 4) uint8 RGBA numpy array."""
return self._engine.capture_frame()