"""Overlay rendering: debug lines, text, and particles."""
import logging
from typing import TYPE_CHECKING, Any
import numpy as np
import vulkan as vk
from ..gpu.memory import create_buffer, upload_numpy
from ..materials.shader_compiler import compile_shader
if TYPE_CHECKING:
from .forward import Renderer
__all__ = ["OverlayRenderer"]
log = logging.getLogger(__name__)
[docs]
class OverlayRenderer:
"""Handles rendering of overlays: debug lines, text, and particles."""
def __init__(self, renderer: Renderer) -> None:
self._r = renderer
[docs]
def dispatch_gpu_particles(self, cmd: Any) -> None:
"""Dispatch compute shaders for GPU particle simulation (outside render pass).
Each ``(emitter_id, cfg)`` submission maps to a persistent per-emitter
SSBO owned by :class:`ParticleCompute` (matches the web
``GPUParticlePass`` model). Emitters that no longer appear in the
scene tree are pruned after the dispatch loop so their GPU
resources are released.
"""
r = self._r
if not r._gpu_particle_submissions:
return
# Lazy-init the shared compute pipeline on first use.
if r._particle_compute is None:
from .particle_compute import ParticleCompute
r._particle_compute = ParticleCompute(r._engine)
r._particle_compute.setup()
compute = r._particle_compute
compute.begin_frame()
dt = getattr(r._engine, '_last_dt', 1.0 / 60.0)
active_ids: set[int] = set()
# Route the dispatches through the async-compute scheduler. In
# passthrough mode ``record_compute`` records into the graphics ``cmd``
# inline (byte-identical to recording directly); in async mode it
# records into the dedicated compute command buffer, and the graphics
# submit waits on the compute-done semaphore at VERTEX_SHADER before the
# billboard draw reads the particle SSBO. ``async_compute`` tells the
# pass to drop the trailing intra-queue write->vertex barrier on the
# async path (the semaphore provides that cross-queue ordering, and a
# VERTEX_SHADER dst stage is illegal on a compute-only queue).
sched = getattr(r._engine, "_async_compute", None)
for emitter_id, cfg in r._gpu_particle_submissions:
if sched is not None:
sched.record_compute(lambda c, eid=emitter_id, cf=cfg: compute.dispatch(c, dt, eid, cf))
else:
compute.dispatch(cmd, dt, emitter_id, cfg)
active_ids.add(emitter_id)
compute.prune_inactive(active_ids)
[docs]
def render_particles(self, cmd: Any, extent: tuple[int, int]) -> None:
"""Render all submitted particle systems."""
r = self._r
viewports = r.viewport_manager.viewports
if viewports:
_, viewport = viewports[0]
vp = viewport.camera_proj @ viewport.camera_view
# Extract camera right/up from view matrix (rows 0 and 1 of transpose)
view_inv = np.linalg.inv(viewport.camera_view)
camera_right = view_inv[:3, 0].astype(np.float32)
camera_up = view_inv[:3, 1].astype(np.float32)
elif r._camera2d_affine is not None:
# 2D scene (Camera2D, no 3D camera): billboard the particles screen-aligned
# through the SAME world->screen canvas_transform sprites are baked with, so
# they pan/zoom locked to the 2D world (design §5.7). camera_right/up stay
# axis-aligned; the view_proj carries the world(pixel)->NDC mapping incl.
# the camera's zoom, so particle scale tracks zoom like a sprite.
vp = self._camera2d_view_proj(*r._camera2d_affine)
camera_right = np.array([1.0, 0.0, 0.0], dtype=np.float32)
camera_up = np.array([0.0, 1.0, 0.0], dtype=np.float32)
else:
return
# Concatenate every emitter's particle data into one upload; the
# particle pass reuses a single persistent buffer, so calling
# render() N times would just clobber the buffer and only show the
# last emitter. One vkCmdDraw covers all submissions.
if r._particle_submissions:
all_data = np.concatenate([d for d, _ in r._particle_submissions])
r._particle_pass.render(cmd, all_data, vp, camera_right, camera_up, extent)
# GPU-simulated particles: compute dispatches already ran in
# dispatch_gpu_particles() outside the render pass; here we draw
# them using the shared billboard pipeline but bind the compute-
# owned SSBO via its own descriptor set.
if r._particle_compute is not None and r._gpu_particle_submissions:
r._particle_compute.render(
cmd, r._particle_pass, vp, camera_right, camera_up, extent,
)
@staticmethod
def _camera2d_view_proj(
affine: tuple[float, ...], screen_size: tuple[float, float]
) -> np.ndarray:
"""Row-major mat4 mapping world (pixels) -> Vulkan NDC via the Camera2D affine.
``affine`` is the world->screen ``(a,b,c,d,tx,ty)`` (rotation-free), ``screen_size``
the logical viewport. Screen->NDC is ``2*screen/size - 1`` (Y already points down
in both screen and Vulkan NDC). z collapses to 0 (the 2D plane). The particle pass
transposes this for the column-major shader.
"""
a, _b, _c, d, tx, ty = affine
sw, sh = screen_size
m = np.zeros((4, 4), dtype=np.float32)
m[0, 0] = 2.0 * a / sw
m[0, 3] = 2.0 * tx / sw - 1.0
m[1, 1] = 2.0 * d / sh
m[1, 3] = 2.0 * ty / sh - 1.0
m[3, 3] = 1.0
return m
[docs]
def render_debug_lines(self, cmd: Any, extent: tuple[int, int]) -> None:
"""Render debug wireframe lines if any were submitted."""
r = self._r
from ..debug_draw import DebugDraw
vertex_data = DebugDraw.get_vertex_data()
if vertex_data is None:
return
e = r._engine
device = e.ctx.device
# Lazy-init debug line pipeline
if r._debug_pipeline is None:
from ..gpu.pipeline import (
POS_COLOUR_VERTEX_ATTRS,
POS_COLOUR_VERTEX_STRIDE,
PipelineSpec,
build_pipeline,
create_shader_module,
)
shader_dir = e.shader_dir
vert_spv = compile_shader(shader_dir / "line.vert")
frag_spv = compile_shader(shader_dir / "line.frag")
r._debug_vert_module = create_shader_module(device, vert_spv)
r._debug_frag_module = create_shader_module(device, frag_spv)
# Debug lines: per-vertex position(vec3)+colour(vec4), alpha-blended,
# depth-tested but not depth-written (drawn on top of the scene with
# LESS_OR_EQUAL); push constants are view(mat4)+proj(mat4) = 128 B.
spec = PipelineSpec(
name="debug-line",
topology=vk.VK_PRIMITIVE_TOPOLOGY_LINE_LIST,
vertex_stride=POS_COLOUR_VERTEX_STRIDE,
vertex_attrs=POS_COLOUR_VERTEX_ATTRS,
cull_mode=vk.VK_CULL_MODE_NONE,
depth_write=False,
depth_compare=vk.VK_COMPARE_OP_LESS_OR_EQUAL,
blend="alpha",
push_size=128,
)
# Debug lines draw inside the active scene pass. When post-processing
# is enabled that is the HDR offscreen pass (R16G16B16A16_SFLOAT), not
# the swapchain pass: compile against the matching pass to avoid
# render-pass-format incompatibility (VUID-vkCmdDraw-renderPass-02684).
r._debug_pipeline, r._debug_pipeline_layout = build_pipeline(
device,
spec,
r._passes.pipeline_render_pass(),
extent,
vert_module=r._debug_vert_module,
frag_module=r._debug_frag_module,
)
# Ensure vertex buffer is large enough
needed = vertex_data.nbytes
if needed > r._debug_vb_capacity:
if r._debug_vb:
vk.vkDestroyBuffer(device, r._debug_vb, None)
vk.vkFreeMemory(device, r._debug_vb_mem, None)
new_cap = max(needed, 4096)
r._debug_vb, r._debug_vb_mem = create_buffer(
device,
e.ctx.physical_device,
new_cap,
vk.VK_BUFFER_USAGE_VERTEX_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
r._debug_vb_capacity = new_cap
upload_numpy(device, r._debug_vb_mem, vertex_data)
# Use the same view/proj from the first viewport
viewports = r.viewport_manager.viewports
if not viewports:
DebugDraw._clear()
return
_, viewport = viewports[0]
view_transposed = np.ascontiguousarray(viewport.camera_view.T)
proj_transposed = np.ascontiguousarray(viewport.camera_proj.T)
pc_data = view_transposed.tobytes() + proj_transposed.tobytes()
# Record draw commands
vk.vkCmdBindPipeline(cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, r._debug_pipeline)
vk_viewport = vk.VkViewport(
x=0.0,
y=0.0,
width=float(extent[0]),
height=float(extent[1]),
minDepth=0.0,
maxDepth=1.0,
)
vk.vkCmdSetViewport(cmd, 0, 1, [vk_viewport])
scissor = vk.VkRect2D(
offset=vk.VkOffset2D(x=0, y=0),
extent=vk.VkExtent2D(width=extent[0], height=extent[1]),
)
vk.vkCmdSetScissor(cmd, 0, 1, [scissor])
e.push_constants(cmd, r._debug_pipeline_layout, pc_data)
vk.vkCmdBindVertexBuffers(cmd, 0, 1, [r._debug_vb], [0])
vk.vkCmdDraw(cmd, DebugDraw.vertex_count(), 1, 0, 0)
DebugDraw._clear()