"""Default renderer: Vulkan forward path; implements the RendererBackend ABC."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
import numpy as np
import vulkan as vk
if TYPE_CHECKING:
from ..engine import CubemapHandle
from ..gpu.memory import upload_numpy
from ..scene.frustum import Frustum
from ..types import (
LIGHT_DTYPE,
MATERIAL_DTYPE,
MeshHandle,
)
from ._base import RendererBackend
from .buffer_manager import SHADOW_DATA_SIZE, BufferManager
from .custom_post_process import CustomPostProcessPass
from .draw2d_pass import Draw2DPass
from .environment_sync import EnvironmentSync
from .gizmo_pass import GizmoRenderData
from .gpu_batch import GPUBatch
from .light2d_pass import Light2DPass
from .overlay_renderer import OverlayRenderer
from .pass_orchestrator import PassOrchestrator
from .pipeline_manager import PipelineManager
from .post_process import PostProcessPass
from .scene_renderer import SceneContentRenderer
from .shadow_renderer import ShadowRenderer
from .ssao_pass import SSAOPass
from .text_pass import TextPass
from .viewport_manager import ViewportManager
__all__ = ["Renderer"]
log = logging.getLogger(__name__)
# Default ambient colour: used by the IBL-only fallback path that bypasses
# the regular shadow-pass upload.
_DEFAULT_AMBIENT = np.array([0.15, 0.15, 0.2, 1.0], dtype=np.float32)
[docs]
class Renderer(RendererBackend):
"""Default Vulkan forward renderer: 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]] = []
# GPU resources (created in setup)
self._pipelines = PipelineManager(engine)
self._buffers = BufferManager(engine, max_objects)
self._passes = PassOrchestrator(engine)
self._batch: GPUBatch | None = None
# Material/light data (set externally via set_materials/set_lights)
self._materials: np.ndarray = np.zeros(1, dtype=MATERIAL_DTYPE)
self._lights: np.ndarray = np.zeros(1, dtype=LIGHT_DTYPE)
# Per-frame particle submissions (stored here; passes consume them)
self._particle_submissions: list[tuple[np.ndarray, int]] = [] # (data, count)
# (emitter_id, emitter_config) tuples: ``emitter_id`` is the stable
# ``id(node) & 0xFFFFFFFF`` minted by ``scene_adapter`` so that
# ``ParticleCompute`` can key per-emitter SSBOs (matches web).
self._gpu_particle_submissions: list[tuple[int, dict]] = []
# Per-frame ShaderMaterial-backed submissions. Each entry is
# (mesh_handle, transform, material_id, shader_material). SceneAdapter
# populates this when a MeshInstance3D carries a custom shader;
# ContentRenderer.render picks up the bucket and issues draws with
# the per-material pipeline (lazy-compiled + cached via the manager).
self._shader_material_submissions: list = []
self._shader_material_manager: Any = None
# Reflection probes: captured local IBL bound at forward bindings 10/11/12.
# Created lazily in setup() so the shared cubemap arrays exist before the
# first probe captures.
self._reflection_probe_pass: Any = None
self._probe_descriptors_bound: bool = False
# IBL enabled once a skybox cubemap is set
self._ibl_enabled: bool = False
# The IBLPass that precomputed the current sky's irradiance / prefiltered
# specular / BRDF LUT. Retained (not cleaned up after process) so the
# forward shader can sample the maps; replaced on each new skybox.
self._ibl_pass: Any = None
self._ibl_handle_id: int | None = None
# Ambient lighting, driven by WorldEnvironment via EnvironmentSync.
# Defaults match the legacy static _AMBIENT so a scene with no
# WorldEnvironment renders identically to before. ``_ambient_energy``
# scales the sky IBL ambient (ambient_colour.a in the shadow SSBO).
self._ambient_colour: tuple[float, float, float] = (0.15, 0.15, 0.2)
self._ambient_energy: float = 1.0
# Skinned mesh instances (mesh_handle, transform, material_id, joint_matrices)
self._skinned_instances: list[tuple[MeshHandle, np.ndarray, int, np.ndarray]] = []
# 2D overlay text renderer (shared across frames, set up lazily)
self._text_renderer: Any = None
# Debug line rendering (OverlayRenderer lazily populates these)
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
self._ready = False
# -- Pipeline accessors (delegate to PipelineManager; keep attribute names
# stable so SceneContentRenderer and other consumers keep working).
@property
def _pipeline(self) -> Any:
return self._pipelines.opaque
@property
def _pipeline_layout(self) -> Any:
return self._pipelines.opaque_layout
@property
def _nocull_pipeline(self) -> Any:
return self._pipelines.nocull
@property
def _nocull_pipeline_layout(self) -> Any:
return self._pipelines.nocull_layout
@property
def _transparent_pipeline(self) -> Any:
return self._pipelines.transparent
@property
def _transparent_pipeline_layout(self) -> Any:
return self._pipelines.transparent_layout
@property
def _skinned_pipeline(self) -> Any:
return self._pipelines.skinned
@property
def _skinned_pipeline_layout(self) -> Any:
return self._pipelines.skinned_layout
@property
def _vert_module(self) -> Any:
return self._pipelines.vert_module
@property
def _frag_module(self) -> Any:
return self._pipelines.frag_module
@property
def _skinned_vert_module(self) -> Any:
return self._pipelines.skinned_vert_module
# -- Buffer accessors (delegate to BufferManager; keep attribute names
# stable so ShadowRenderer, SceneContentRenderer and OverlayRenderer
# continue to access `r._shadow_mem`, `r._transform_mem` etc. unchanged).
@property
def _ssbo_layout(self) -> Any:
return self._buffers.ssbo_layout
@property
def _ssbo_set(self) -> Any:
return self._buffers.ssbo_set
@property
def _transform_mem(self) -> Any:
return self._buffers.transform_mem
@property
def _material_mem(self) -> Any:
return self._buffers.material_mem
@property
def _light_mem(self) -> Any:
return self._buffers.light_mem
@property
def _shadow_buf(self) -> Any:
return self._buffers.shadow_buf
@property
def _shadow_mem(self) -> Any:
return self._buffers.shadow_mem
@property
def _joint_layout(self) -> Any:
return self._buffers.joint_layout
@property
def _joint_set(self) -> Any:
return self._buffers.joint_set
@property
def _joint_mem(self) -> Any:
return self._buffers.joint_mem
@property
def _max_materials(self) -> int:
return self._buffers.max_materials
# -- Pass accessors (delegate to PassOrchestrator; keep old names stable).
@property
def _post_process(self) -> PostProcessPass | None:
return self._passes.post_process
@property
def _custom_pp(self) -> CustomPostProcessPass | None:
return self._passes.custom_pp
@property
def _ssao_pass(self) -> SSAOPass | None:
return self._passes.ssao_pass
@property
def _volumetric_fog_pass(self) -> Any:
return self._passes.volumetric_fog_pass
@property
def _skybox_pass(self) -> Any:
return self._passes.skybox_pass
@property
def _shadow_pass(self) -> Any:
return self._passes.shadow_pass
@property
def _point_shadow_pass(self) -> Any:
return self._passes.point_shadow_pass
@property
def _particle_pass(self) -> Any:
return self._passes.particle_pass
@property
def _particle_compute(self) -> Any:
return self._passes.particle_compute
@_particle_compute.setter
def _particle_compute(self, value: Any) -> None:
"""OverlayRenderer lazily creates the compute pass on first GPU submission."""
self._passes.particle_compute = value
@property
def _tilemap_pass(self) -> Any:
return self._passes.tilemap_pass
@property
def _light2d_pass(self) -> Light2DPass | None:
return self._passes.light2d_pass
@property
def _text_pass(self) -> TextPass | None:
return self._passes.text_pass
@property
def _draw2d_pass(self) -> Draw2DPass | None:
return self._passes.draw2d_pass
@property
def _grid_pass(self) -> Any:
return self._passes.grid_pass
@property
def _gizmo_pass(self) -> Any:
return self._passes.gizmo_pass
@property
def _gizmo_render_data(self) -> GizmoRenderData | None:
return self._passes.gizmo_render_data
@_gizmo_render_data.setter
def _gizmo_render_data(self, value: GizmoRenderData | None) -> None:
self._passes.gizmo_render_data = value
[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 + descriptor sets + IBL cubemap placeholder
self._buffers.setup()
upload_numpy(device, self._buffers.material_mem, self._materials)
upload_numpy(device, self._buffers.light_mem, self._lights)
# 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)
# 3D graphics pipelines (opaque, double-sided, transparent, skinned)
self._pipelines.setup(self._buffers.ssbo_layout, self._buffers.joint_layout)
# All render passes (shadow/particle/tilemap/2D/post-process/SSAO/fog/...)
self._passes.setup(self._buffers.ssbo_layout, self._pipelines)
# Text overlay renderer (shared across frames, caches atlases)
from ..text_renderer import get_shared_text_renderer
self._text_renderer = get_shared_text_renderer()
# Reflection-probe pass: owns the shared cubemap arrays + box SSBO.
from .reflection_probe_pass import ReflectionProbePass
self._reflection_probe_pass = ReflectionProbePass(e)
self._reflection_probe_pass.setup()
# Bind the probe arrays once (capture writes into them in-place).
self._buffers.write_probe_descriptors(
self._reflection_probe_pass.get_irradiance_array_view(),
self._reflection_probe_pass.get_prefilter_array_view(),
self._reflection_probe_pass.get_sampler(),
)
self._ready = True
[docs]
def set_skybox(self, cubemap: CubemapHandle) -> None:
"""Set a cubemap as the skybox and run the IBL precompute.
Accepts a :class:`~simvx.graphics.engine.CubemapHandle` returned by
:meth:`Engine.load_cubemap`. The renderer takes ownership of the
underlying Vulkan resources and destroys them at shutdown. IBL
precompute (irradiance + prefiltered specular + BRDF LUT) is run
once per unique handle and cached, mirroring the web renderer's
``IBLPass._byCubemapId`` lazy-precompute behaviour. Re-setting the
same handle is a no-op past the cache check.
"""
from ..engine import CubemapHandle
if not isinstance(cubemap, CubemapHandle):
raise TypeError(
f"set_skybox expects a CubemapHandle (from Engine.load_cubemap), "
f"got {type(cubemap).__name__}",
)
self._passes.set_skybox(
cubemap.view, cubemap.sampler, self._buffers, cubemap.image, cubemap.memory,
)
self._ibl_enabled = True
# IBL precompute: cached by ``CubemapHandle`` identity. The
# ``IBLPass`` is a one-shot operation per cubemap (irradiance +
# prefiltered specular sampled into renderer-side textures) so we
# avoid redoing it when the same handle is re-installed.
if self._ibl_handle_id != id(cubemap):
try:
from .ibl_pass import IBLPass
except ImportError:
# IBL module not bundled (e.g. minimal builds): skybox is
# still set, just without environment lighting.
self._ibl_handle_id = id(cubemap)
return
# Destroy the previous sky's maps before precomputing the new ones.
if self._ibl_pass is not None:
self._ibl_pass.cleanup()
self._ibl_pass = None
ibl = IBLPass(self._engine)
ibl.setup()
ibl.process_cubemap(cubemap.view, cubemap.sampler)
# Retain the pass: its irradiance / prefiltered / BRDF views are
# bound to the forward descriptor set (bindings 7/8/9) so the shader
# can sample them in the split-sum IBL path.
self._ibl_pass = ibl
self._ibl_handle_id = id(cubemap)
self._buffers.write_ibl_descriptors(
ibl.get_irradiance_view(), ibl.get_prefiltered_view(),
ibl.get_brdf_lut_view(), ibl.get_sampler(),
)
[docs]
@property
def post_processing(self) -> PostProcessPass | None:
"""Access post-processing pass for configuration."""
return self._post_process
[docs]
@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
[docs]
@property
def ssao(self) -> SSAOPass | None:
"""Access SSAO pass for configuration."""
return self._ssao_pass
[docs]
def set_materials(self, materials: np.ndarray) -> None:
"""Set material array and upload to GPU (skips if unchanged)."""
self._materials = self._buffers.set_materials(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
self._buffers.set_lights(lights)
def _directional_sun(self) -> tuple[Any, Any, float]:
"""Return (direction, colour, intensity) of the first directional light.
Direction is the light's travel direction (w<0.5 marks directional in
``LIGHT_DTYPE``). Falls back to a default sun when none is present.
"""
lights = self._lights
if lights is not None and len(lights) > 0:
for i in range(len(lights)):
if lights[i]["position"][3] < 0.5: # directional marker
direction = lights[i]["position"][:3].copy()
colour = lights[i]["colour"][:3].copy()
intensity = float(lights[i]["colour"][3])
if float(np.linalg.norm(direction)) > 1e-6:
return direction, colour, intensity
return (np.array([-1.0, -1.0, -1.0], dtype=np.float32),
np.array([1.0, 1.0, 1.0], dtype=np.float32), 1.0)
def _collect_fog_volumes(self) -> list[Any]:
"""Collect all visible ``FogVolume3D`` nodes from the scene tree."""
from simvx.core.fog_volume import FogVolume3D
tree = getattr(self._engine, "_scene_tree", None) or getattr(self._engine, "scene_tree", None)
root = getattr(tree, "root", None) if tree else None
if root is None:
return []
return [n for n in root.find_all(FogVolume3D) if getattr(n, "visible", True)]
def _collect_reflection_probes(self) -> list[Any]:
"""Collect all visible ``ReflectionProbe3D`` nodes from the scene tree."""
from simvx.core.reflection_probe import ReflectionProbe3D
tree = getattr(self._engine, "_scene_tree", None) or getattr(self._engine, "scene_tree", None)
root = getattr(tree, "root", None) if tree else None
if root is None:
return []
return [n for n in root.find_all(ReflectionProbe3D) if getattr(n, "visible", True)]
[docs]
def capture_reflection_probes(self, adapter: Any) -> bool:
"""Capture any new / requested reflection probes. Returns True if any captured.
Driven from the app's pre_render hook (before the main render pass) so a
capture this frame is visible the same frame. Probe face rendering reuses
the offscreen scene path and clobbers per-frame submission lists, so the
caller must re-submit the main scene when this returns True: exactly the
SubViewport contract.
"""
if not self._ready or self._reflection_probe_pass is None:
return False
tree = getattr(self._engine, "_scene_tree", None) or getattr(self._engine, "scene_tree", None)
if tree is None:
return False
probes = self._collect_reflection_probes()
if not probes:
return False
return self._reflection_probe_pass.update_probes(adapter, tree, probes)
[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._particle_submissions.clear()
self._gpu_particle_submissions.clear()
self._skinned_instances.clear()
self._shader_material_submissions.clear()
self._passes.begin_frame()
if self._text_renderer:
self._text_renderer.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_shader_instance(
self,
mesh_handle: MeshHandle,
transform: np.ndarray,
material_id: int,
shader_material: Any,
) -> None:
"""Submit a MeshInstance3D that carries a ShaderMaterial.
The per-material pipeline is lazy-compiled + cached on first use via
ShaderMaterialManager. During the forward draw, the renderer splits
these submissions into their own bucket, binds the custom pipeline,
updates the uniform buffer, and draws one mesh at a time, per-
material pipeline switching is the cost of the custom-shader path.
"""
self._shader_material_submissions.append(
(mesh_handle, transform, material_id, shader_material),
)
[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, *, emitter_id: int = 0) -> None:
"""Submit a GPU particle emitter config for compute-shader simulation this frame.
``emitter_id`` is a stable per-node identifier (``id(node) & 0xFFFFFFFF``
from :class:`SceneAdapter`). :class:`ParticleCompute` keys its
persistent SSBOs by this id so multi-emitter scenes render
correctly: matches the web ``GPUParticlePass`` ownership model.
"""
self._gpu_particle_submissions.append((int(emitter_id) & 0xFFFFFFFF, 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))
def _upload_transforms(self) -> None:
"""Upload ALL instance transforms to the SSBO once per frame."""
self._buffers.upload_transforms(self._instances)
def _set_hdr_flag(self, enabled: bool) -> None:
"""Set hdr_output (byte offset 216) flag in shadow SSBO."""
self._buffers.set_hdr_flag(enabled)
[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()
# GPU timing pool (None if device lacks timestamp queries).
pool = self._engine.current_timestamp_pool
# Dispatch GPU particle compute shaders (must happen outside render pass)
if self._gpu_particle_submissions:
if pool:
pool.begin(cmd, "particle_compute")
self._overlay_renderer.dispatch_gpu_particles(cmd)
if pool:
pool.end(cmd, "particle_compute")
# 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(SHADOW_DATA_SIZE, 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)
shadow_data[336:352] = _DEFAULT_AMBIENT.view(np.uint8)
self._buffers.write_shadow_data(shadow_data)
# Render 2D lights to accumulation texture
if self._light2d_pass and self._light2d_pass.has_lights:
if pool:
pool.begin(cmd, "light2d")
self._light2d_pass.render(cmd, self._engine.extent)
if pool:
pool.end(cmd, "light2d")
if self._shadow_pass and self._instances:
if pool:
pool.begin(cmd, "shadow")
self._shadow_renderer.render_shadows(cmd, self._engine.mesh_registry)
if pool:
pool.end(cmd, "shadow")
if self._point_shadow_pass and self._instances:
if pool:
pool.begin(cmd, "point_shadow")
self._shadow_renderer.render_point_spot_shadows(cmd, self._engine.mesh_registry)
if pool:
pool.end(cmd, "point_shadow")
# When post-processing is enabled, render 3D scene to HDR target here.
# Also enter the HDR pass for scenes with no 3D mesh instances but that
# still have content that renders through render_scene_content, e.g.
# tilemap-only scenes or skybox + particles, otherwise nothing is
# ever drawn and the tonemap samples an undefined HDR target.
pp = self._post_process
has_scene_content = bool(
self._instances
or (self._tilemap_pass and self._tilemap_pass._submissions)
or (self._particle_pass and getattr(self._particle_pass, "_submissions", None))
or self._gpu_particle_submissions
or (self._skybox_pass and getattr(self._skybox_pass, "enabled", False))
)
if pp and pp.enabled and has_scene_content:
# Update camera matrices in UBO (needed by motion blur AND fog depth reconstruction)
viewports = self.viewport_manager.viewports
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)
if pool:
pool.begin(cmd, "forward")
pp.begin_hdr_pass(cmd)
self._scene_renderer.render_scene_content(cmd)
pp.end_hdr_pass(cmd)
if pool:
pool.end(cmd, "forward")
self._hdr_rendered = True
# Run bloom pass (extract + blur) between HDR and tonemap
if pp.bloom_enabled:
if pool:
pool.begin(cmd, "bloom")
pp.render_bloom(cmd)
if pool:
pool.end(cmd, "bloom")
# 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.viewports
if viewports:
_, vp = viewports[0]
if pool:
pool.begin(cmd, "ssao")
self._ssao_pass.render(cmd, vp.camera_proj)
if pool:
pool.end(cmd, "ssao")
# Distance/height fog is applied in tonemap.frag. Volumetric fog is
# a separate ray-march pass (below) that composites in HDR space and
# mutually excludes the analytic tonemap fog branch.
vfog = self._volumetric_fog_pass
self._volumetric_fog_active = False
if vfog and vfog.enabled:
viewports = self.viewport_manager.viewports
if viewports:
_, vp = viewports[0]
sun_dir, sun_col, sun_int = self._directional_sun()
volumes = self._collect_fog_volumes()
vfog.set_frame_data(vp.camera_view, vp.camera_proj,
sun_dir, sun_col, sun_int, volumes)
if pool:
pool.begin(cmd, "volumetric_fog")
vfog.render(cmd)
if pool:
pool.end(cmd, "volumetric_fog")
# Tonemap samples the fog-composited HDR copy. The descriptor
# swap was done in sync_world_environment (before recording)
# to avoid updating a bound descriptor set mid-frame.
self._volumetric_fog_active = True
# Run custom user post-process effects (after bloom/SSAO, before tonemap)
if pool and self._custom_pp and getattr(self._custom_pp, "has_effects", False):
pool.begin(cmd, "custom_pp")
self._env_sync.run_custom_post_process(cmd, pp)
pool.end(cmd, "custom_pp")
else:
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
pool = e.current_timestamp_pool
# 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:
if pool:
pool.begin(cmd, "tonemap")
pp.render_tonemap(cmd, e.extent[0], e.extent[1])
if pool:
pool.end(cmd, "tonemap")
# Restore tonemap's HDR input to the original HDR target for next
# frame (custom_pp swaps it mid-frame; the volumetric-fog swap is
# re-decided each frame in sync_world_environment).
if self._custom_pp and self._custom_pp.has_effects and pp.hdr_target:
self._env_sync.update_tonemap_hdr_input(pp.hdr_target.colour_view)
elif not (pp and pp.enabled):
# Direct rendering to swapchain (no post-processing)
if pool:
pool.begin(cmd, "forward")
self._scene_renderer.render_scene_content(cmd)
if pool:
pool.end(cmd, "forward")
# 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 pool:
pool.begin(cmd, "draw2d")
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])
if pool:
pool.end(cmd, "draw2d")
# Render text overlay (2D)
if self._text_renderer and self._text_renderer.has_text and self._text_pass:
if pool:
pool.begin(cmd, "text")
self._overlay_renderer.render_text(cmd, e.extent)
if pool:
pool.end(cmd, "text")
[docs]
def resize(self, width: int, height: int) -> None:
"""Handle framebuffer resize: recreate post-process targets + 3D pipelines."""
if not self._ready:
return
# Resize post-process first so HDR render-pass exists when pipelines rebuild.
self._passes.resize(width, height)
self._pipelines.rebuild_for_resize(width, height, self._passes.pipeline_render_pass())
[docs]
def cleanup(self) -> None:
"""Release all GPU resources."""
if not self._ready:
return
device = self._engine.ctx.device
# IBL precompute maps (retained from the last set_skybox).
if self._ibl_pass is not None:
self._ibl_pass.cleanup()
self._ibl_pass = None
# Reflection-probe pass (shared cubemap arrays + capture scratch).
if self._reflection_probe_pass is not None:
self._reflection_probe_pass.cleanup()
self._reflection_probe_pass = None
# All render passes + skybox cubemap (orchestrator owns lifecycle).
self._passes.cleanup()
# Batch objects
for batch in (self._batch, self._transparent_batch):
if batch:
batch.destroy()
# Debug line pipeline (lazily created by OverlayRenderer, not owned by PipelineManager)
if self._debug_pipeline:
vk.vkDestroyPipeline(device, self._debug_pipeline, None)
if self._debug_pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._debug_pipeline_layout, None)
# 3D pipelines + shader modules
self._pipelines.cleanup()
# Debug shader modules (lazily created)
for mod in (self._debug_vert_module, self._debug_frag_module):
if mod:
vk.vkDestroyShaderModule(device, mod, None)
# Debug vertex buffer (lazily created by OverlayRenderer)
if self._debug_vb:
vk.vkDestroyBuffer(device, self._debug_vb, None)
if self._debug_vb_mem:
vk.vkFreeMemory(device, self._debug_vb_mem, None)
# SSBOs + descriptor sets + placeholder cubemap
self._buffers.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 upload_texture_pixels(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()