"""GPU-batched TileMap renderer: single draw call per layer via instanced SSBO quads."""
import logging
from typing import Any
import numpy as np
import vulkan as vk
from ..gpu.memory import create_buffer, upload_numpy
from ..gpu.pipeline import PipelineSpec, build_pipeline, create_shader_module
from ..materials.shader_compiler import compile_shader
from .tile_types import TILE_COLOUR_DTYPE, TILE_INSTANCE_DTYPE
__all__ = ["TILE_INSTANCE_DTYPE", "TileMapPass"]
log = logging.getLogger(__name__)
MAX_TILES = 65_536 # Max tiles per frame across all layers
_TILE_STRIDE = TILE_INSTANCE_DTYPE.itemsize # 32 bytes
_COLOUR_STRIDE = TILE_COLOUR_DTYPE.itemsize # 16 bytes (vec4 RGBA)
[docs]
class TileMapPass:
"""Renders tilemap layers as instanced quads via SSBO.
Each layer is submitted as a contiguous block of tile instances.
All layers share a single SSBO upload; each layer draws with an offset.
"""
def __init__(self, engine: Any):
self._engine = engine
self._pipeline: Any = None
self._pipeline_layout: Any = None
self._vert_module: Any = None
self._frag_module: Any = None
self._ssbo_layout: Any = None
self._ssbo_pool: Any = None
self._ssbo_set: Any = None
self._tile_buf: Any = None
self._tile_mem: Any = None
self._colour_buf: Any = None
self._colour_mem: Any = None
self._ready = False
# Identity signature of the last SSBO upload. When this frame's submissions
# reference the byte-identical arrays as last frame (the scene adapter hands
# back its cached buffer for an unchanged layer), the SSBO already holds the
# right data, so the concatenate + upload is skipped entirely. ``None`` ==
# nothing uploaded yet (first frame always uploads).
self._last_upload_sig: tuple | None = None
# Per-frame submissions: (tile_data, tileset_texture_id, tile_size, colours|None)
self._submissions: list[tuple[np.ndarray, int, tuple[float, float], np.ndarray | None]] = []
[docs]
def setup(self, render_pass: Any = None) -> None:
"""Create GPU resources: SSBO, pipeline, descriptors.
``render_pass`` defaults to ``engine.render_pass``; pass an HDR pass
when the pipeline will be bound inside the post-process offscreen pass.
"""
e = self._engine
device = e.ctx.device
phys = e.ctx.physical_device
# Tile instance SSBO
buf_size = MAX_TILES * _TILE_STRIDE
self._tile_buf, self._tile_mem = create_buffer(
device,
phys,
buf_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
# Optional per-tile tint SSBO (set 0, binding 1), indexed parallel to the
# tile buffer. Always bound so the pipeline has a valid descriptor; only
# uploaded (and read by the shader) when a layer carries a tint.
colour_size = MAX_TILES * _COLOUR_STRIDE
self._colour_buf, self._colour_mem = create_buffer(
device,
phys,
colour_size,
vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
# Descriptor set for tile + colour SSBOs (set 0, bindings 0 and 1)
from ..gpu.descriptors import (
allocate_descriptor_set,
create_descriptor_pool,
create_ssbo_layout,
write_ssbo_descriptor,
)
self._ssbo_layout = create_ssbo_layout(device, binding_count=2)
self._ssbo_pool = create_descriptor_pool(device, max_sets=1)
self._ssbo_set = allocate_descriptor_set(device, self._ssbo_pool, self._ssbo_layout)
write_ssbo_descriptor(device, self._ssbo_set, 0, self._tile_buf, buf_size)
write_ssbo_descriptor(device, self._ssbo_set, 1, self._colour_buf, colour_size)
# Compile shaders
shader_dir = e.shader_dir
vert_spv = compile_shader(shader_dir / "tilemap.vert")
frag_spv = compile_shader(shader_dir / "tilemap.frag")
self._vert_module = create_shader_module(device, vert_spv)
self._frag_module = create_shader_module(device, frag_spv)
# Pipeline
self._create_pipeline(device, render_pass or e.render_pass, e.extent)
self._ready = True
log.debug("TileMap pass initialized (max %d tiles)", MAX_TILES)
[docs]
def rebuild_pipeline(self, render_pass: Any) -> None:
"""Recreate the tilemap pipeline against a different render pass (e.g. HDR)."""
if not self._ready:
return
device = self._engine.ctx.device
if self._pipeline:
vk.vkDestroyPipeline(device, self._pipeline, None)
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._pipeline_layout, None)
self._create_pipeline(device, render_pass, self._engine.extent)
def _create_pipeline(self, device: Any, render_pass: Any, extent: tuple[int, int]) -> None:
"""Create the tilemap pipeline: alpha blend, no depth test/write, no vertex input.
Declares its fixed-function state via :class:`PipelineSpec` and defers all
cffi sub-struct plumbing (and lifetime management) to :func:`build_pipeline`.
Set 0: tile + colour SSBOs, Set 1: bindless texture array. Push constants:
mat4 view(64) + vec2 tile_size(8) + int tex_id(4) + int has_colour(4) = 80
bytes (vertex + fragment).
The shaders are compiled at runtime, so the pre-created modules are passed
directly rather than via SPIR-V paths in the spec.
"""
spec = PipelineSpec(
name="tilemap",
topology=vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
vertex_stride=0, # 6 vertices per tile generated in the shader from the SSBO
cull_mode=vk.VK_CULL_MODE_NONE, # 2D tiles: no culling (front_face inert -> spec default CCW)
depth_test=False,
depth_write=False,
blend="alpha",
dst_alpha_factor=vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA,
set_layouts=(self._ssbo_layout, self._engine.texture_descriptor_layout),
push_size=80,
push_stages=vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
# Thin G-buffer: share the HDR pass's second attachment, mask writes.
attachment_count=(2 if self._engine.gbuffer_active else 1),
)
self._pipeline, self._pipeline_layout = build_pipeline(
device,
spec,
render_pass,
extent,
vert_module=self._vert_module,
frag_module=self._frag_module,
)
[docs]
def begin_frame(self) -> None:
"""Clear per-frame submissions."""
self._submissions.clear()
[docs]
def submit_layer(
self,
tile_data: np.ndarray,
tileset_texture_id: int,
tile_size: tuple[float, float],
colours: np.ndarray | None = None,
*,
layer_id: int | None = None,
version: int | None = None,
) -> None:
"""Queue a tile layer for rendering.
Args:
tile_data: Structured array with TILE_INSTANCE_DTYPE.
tileset_texture_id: Bindless texture index for the tileset atlas.
tile_size: (width, height) of each tile in world units.
colours: Optional ``(N, 4)`` float32 RGBA tints parallel to
``tile_data``; ``None`` (the common case) leaves the layer
untinted (white) and uploads no colour bytes.
layer_id, version: Retention hints from the scene adapter (the
source layer's identity + packed-buffer version). The desktop
pass skips the SSBO re-upload by array identity (the adapter
hands back the SAME cached array for an unchanged layer), so it
ignores these; they exist so this signature matches the web
backend's, whose REUSE wire keys on them.
"""
if len(tile_data) == 0:
return
self._submissions.append((tile_data, tileset_texture_id, tile_size, colours))
def _upload_if_changed(self, subs: list, total: int, grand: int) -> bool:
"""Upload the concatenated tile (and tint) SSBOs, skipping unchanged frames.
The submission signature folds each layer's array identity, texture id and
tile size: the scene adapter hands back the SAME cached array for a layer
whose content / transform / tileset did not change, so an all-static frame's
signature matches the last upload's and the concatenate + ``upload_numpy``
are skipped entirely (the SSBO still holds the right bytes). Returns whether
an upload happened (``True`` on the first frame / any change)."""
upload_sig = tuple((id(s[0]), s[1], tuple(s[2]), id(s[3])) for s in subs)
if upload_sig == self._last_upload_sig:
return False
device = self._engine.ctx.device
all_data = np.concatenate([s[0] for s in subs])
upload_numpy(device, self._tile_mem, all_data[:total])
# Tints: only build + upload a parallel colour buffer when some layer is
# tinted. Untinted layers keep has_colour=0 and never read it, so the
# all-untinted common case uploads zero colour bytes.
if any(s[3] is not None for s in subs):
colour_all = np.ones((grand, 4), dtype=np.float32)
off2 = 0
for tile_data, _tex_id, _tile_size, colours in subs:
n = len(tile_data)
if colours is not None:
colour_all[off2 : off2 + n] = colours
off2 += n
upload_numpy(device, self._colour_mem, colour_all[:total])
self._last_upload_sig = upload_sig
return True
[docs]
def render(
self,
cmd: Any,
view_matrix: np.ndarray,
extent: tuple[int, int],
submissions: list[tuple[np.ndarray, int, tuple[float, float], np.ndarray | None]] | None = None,
) -> None:
"""Record draw commands for all queued tile layers.
``submissions`` defaults to the live ``self._submissions`` list (the
synchronous path). In pipelined mode the render thread passes the
packet's OWNED tilemap snapshot so it never reads the live list the main
thread is concurrently rebuilding.
"""
subs = self._submissions if submissions is None else submissions
if not self._ready or not subs:
return
e = self._engine
# Per-frame upload guard: when every submission references the byte-identical
# array as last frame (the adapter reuses its cached buffer for an unchanged
# static layer), the SSBO already holds the right contents, so skip the
# concatenate + GPU re-upload. The draw loop below still re-pushes the view
# matrix per layer, so a static tilemap renders correctly under a moving
# camera while paying ~0 buffer cost on unchanged frames.
grand = sum(len(s[0]) for s in subs)
total = min(grand, MAX_TILES)
if total < grand:
log.warning("TileMap overflow: %d tiles (max %d)", grand, MAX_TILES)
self._upload_if_changed(subs, total, grand)
# Viewport + scissor
vk_vp = 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_vp])
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])
# Bind pipeline
vk.vkCmdBindPipeline(cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline)
# Bind SSBO descriptor (set 0)
vk.vkCmdBindDescriptorSets(
cmd,
vk.VK_PIPELINE_BIND_POINT_GRAPHICS,
self._pipeline_layout,
0,
1,
[self._ssbo_set],
0,
None,
)
# Bind texture array descriptor (set 1) from engine
tex_ds = e.texture_descriptor_set
if tex_ds:
vk.vkCmdBindDescriptorSets(
cmd,
vk.VK_PIPELINE_BIND_POINT_GRAPHICS,
self._pipeline_layout,
1,
1,
[tex_ds],
0,
None,
)
# Transpose view matrix for column-major GLSL
view_transposed = np.ascontiguousarray(view_matrix.T, dtype=np.float32)
# Draw each layer with its own push constants
offset = 0
ffi = vk.ffi
for tile_data, tex_id, tile_size, colours in subs:
count = min(len(tile_data), total - offset)
if count <= 0:
break
# Push constants: mat4(64) + vec2 tile_size(8) + int tex_id(4) + int has_colour(4) = 80 bytes
pc = np.zeros(20, dtype=np.float32)
pc[:16] = view_transposed.ravel()
pc[16] = tile_size[0]
pc[17] = tile_size[1]
# Pack texture ID + has_colour as int32 at indices 18, 19
pc_i = pc.view(np.int32)
pc_i[18] = np.int32(tex_id)
pc_i[19] = np.int32(1 if colours is not None else 0)
pc_bytes = pc.tobytes()
cbuf = ffi.new("char[]", pc_bytes)
vk._vulkan.lib.vkCmdPushConstants(
cmd,
self._pipeline_layout,
vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
0,
len(pc_bytes),
cbuf,
)
# 6 vertices per tile (quad), draw as instanced with base vertex offset
vk.vkCmdDraw(cmd, count * 6, 1, offset * 6, 0)
offset += count
[docs]
def cleanup(self) -> None:
"""Destroy all GPU resources."""
if not self._ready:
return
device = self._engine.ctx.device
for obj, fn in [
(self._pipeline, vk.vkDestroyPipeline),
(self._pipeline_layout, vk.vkDestroyPipelineLayout),
(self._vert_module, vk.vkDestroyShaderModule),
(self._frag_module, vk.vkDestroyShaderModule),
(self._ssbo_layout, vk.vkDestroyDescriptorSetLayout),
(self._ssbo_pool, vk.vkDestroyDescriptorPool),
]:
if obj:
fn(device, obj, None)
if self._tile_buf:
vk.vkDestroyBuffer(device, self._tile_buf, None)
if self._tile_mem:
vk.vkFreeMemory(device, self._tile_mem, None)
if self._colour_buf:
vk.vkDestroyBuffer(device, self._colour_buf, None)
if self._colour_mem:
vk.vkFreeMemory(device, self._colour_mem, None)
self._ready = False