"""2D text overlay pass using MSDF atlas: renders after 3D geometry."""
import logging
from typing import Any
import numpy as np
import vulkan as vk
from ..types import SHADER_DIR
from ..gpu.descriptors import (
create_texture_descriptor_layout,
write_texture_descriptor,
)
from ..gpu.pipeline import PipelineSpec, build_pipeline
from ..gpu.memory import (
create_buffer,
upload_image_data,
upload_numpy,
)
from .pass_helpers import create_linear_sampler, create_sampler_descriptor_pool, load_shader_modules
__all__ = ["TextPass"]
log = logging.getLogger(__name__)
# Vertex format: pos(vec2) + uv(vec2) + colour(vec4) = 32 bytes
VERTEX_STRIDE = 32
MAX_CHARS = 4096
VERTEX_BUF_SIZE = MAX_CHARS * 4 * VERTEX_STRIDE # 4 verts per char
INDEX_BUF_SIZE = MAX_CHARS * 6 * 4 # 6 uint32 indices per char
[docs]
class TextPass:
"""GPU text rendering with MSDF atlas, proper CFFI pipeline."""
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._vertex_buffer: Any = None
self._vertex_memory: Any = None
self._index_buffer: Any = None
self._index_memory: Any = None
self._index_count = 0
self._sampler: Any = None
self._descriptor_layout: Any = None
self._descriptor_pool: Any = None
self._descriptor_set: Any = None
self._atlas_image: Any = None
self._atlas_memory: Any = None
self._atlas_view: Any = None
self._atlas_version = 0 # Tracks which atlas version is on GPU
self._px_range: float = 4.0 # SDF range in atlas pixels
self._ready = False
# Per-render-pass MSDF pipeline cache (design §5.3 RTT-2D): the default
# pipeline targets the swapchain (sRGB); a SubViewport offscreen target is
# R16F, a different (incompatible) colour format, so text rendered into one
# needs a pipeline compiled against THAT pass. Keyed by ``id(render_pass)``;
# built lazily, freed at cleanup. The main path uses ``pipeline`` unchanged.
self._extra_pipelines: dict[int, tuple[Any, Any]] = {}
[docs]
def setup(self) -> None:
"""Initialize GPU resources."""
e = self._engine
device = e.ctx.device
phys = e.ctx.physical_device
# Compile shaders
self._vert_module, self._frag_module = load_shader_modules(
device, SHADER_DIR, "text.vert", "text.frag",
)
# Sampler (linear filtering for MSDF)
self._sampler = create_linear_sampler(device)
# Descriptor layout + pool + set for atlas texture
self._descriptor_layout = create_texture_descriptor_layout(device, max_textures=1)
self._descriptor_pool, desc_sets = create_sampler_descriptor_pool(
device, self._descriptor_layout,
)
self._descriptor_set = desc_sets[0]
# Pipeline
self._create_pipeline(device, e.render_pass, e.extent)
# Vertex/index buffers (host-visible for per-frame updates)
self._vertex_buffer, self._vertex_memory = create_buffer(
device, phys, VERTEX_BUF_SIZE,
vk.VK_BUFFER_USAGE_VERTEX_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
self._index_buffer, self._index_memory = create_buffer(
device, phys, INDEX_BUF_SIZE,
vk.VK_BUFFER_USAGE_INDEX_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
self._ready = True
def _create_pipeline(self, device: Any, render_pass: Any, extent: tuple[int, int]) -> None:
"""Create the 2D MSDF text pipeline via :class:`PipelineSpec`.
Vertex format: pos(vec2)@0 + uv(vec2)@8 + colour(vec4)@16 = 32 bytes.
Push constant: screen_size(vec2) + px_range(float) = 12 bytes (VS+FS).
Descriptor set 0: MSDF atlas (combined image sampler).
No depth test, cull none, alpha blending (dst alpha = 1 - src alpha).
cffi sub-struct lifetime is owned by :func:`build_pipeline`.
"""
spec = PipelineSpec(
name="Text",
topology=vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
vertex_stride=VERTEX_STRIDE,
vertex_attrs=(
(0, vk.VK_FORMAT_R32G32_SFLOAT, 0), # position
(1, vk.VK_FORMAT_R32G32_SFLOAT, 8), # uv
(2, vk.VK_FORMAT_R32G32B32A32_SFLOAT, 16), # colour
),
cull_mode=vk.VK_CULL_MODE_NONE, # 2D text, no culling
depth_test=False,
depth_write=False,
blend="alpha",
dst_alpha_factor=vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA,
set_layouts=(self._descriptor_layout,),
push_size=12, # vec2 screen_size + float px_range
push_stages=vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
)
self._pipeline, self._pipeline_layout = build_pipeline(
device, spec, render_pass, extent,
vert_module=self._vert_module, frag_module=self._frag_module,
)
[docs]
@property
def pipeline(self) -> Any:
return self._pipeline
[docs]
@property
def pipeline_layout(self) -> Any:
return self._pipeline_layout
[docs]
def pipeline_for(self, render_pass: Any, extent: tuple[int, int]) -> Any:
"""Return the MSDF pipeline compiled against ``render_pass`` (design §5.3).
The default :attr:`pipeline` targets the swapchain (sRGB). Text drawn into
an offscreen RTT target (a SubViewport's R16F colour) needs a pipeline
compiled against that pass -- a different colour format is render-pass-
INCOMPATIBLE, so the swapchain pipeline silently fails to draw there. This
lazily builds + caches one pipeline per distinct ``render_pass`` (keyed by
id); the layout is shared (same descriptor set / push constants). Returns
the default pipeline when ``render_pass`` is the engine's main pass.
"""
if render_pass is None or render_pass == self._engine.render_pass:
return self._pipeline
key = id(render_pass)
cached = self._extra_pipelines.get(key)
if cached is None:
from ..gpu.pipeline import PipelineSpec, build_pipeline
spec = PipelineSpec(
name="TextRTT",
topology=vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
vertex_stride=VERTEX_STRIDE,
vertex_attrs=(
(0, vk.VK_FORMAT_R32G32_SFLOAT, 0),
(1, vk.VK_FORMAT_R32G32_SFLOAT, 8),
(2, vk.VK_FORMAT_R32G32B32A32_SFLOAT, 16),
),
cull_mode=vk.VK_CULL_MODE_NONE,
depth_test=False,
depth_write=False,
blend="alpha",
dst_alpha_factor=vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA,
set_layouts=(self._descriptor_layout,),
push_size=12,
push_stages=vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
)
cached = build_pipeline(
self._engine.ctx.device, spec, render_pass, extent,
vert_module=self._vert_module, frag_module=self._frag_module,
)
self._extra_pipelines[key] = cached
return cached[0]
[docs]
@property
def descriptor_set(self) -> Any:
return self._descriptor_set
[docs]
@property
def px_range(self) -> float:
return self._px_range
[docs]
@property
def atlas_version(self) -> int:
return self._atlas_version
[docs]
def upload_atlas_if_dirty(self) -> None:
"""Upload the shared TextRenderer's atlas if its version changed.
Also checks Draw2D's atlas (same shared atlas after unification).
Must be called outside the render pass (staging transfers).
"""
if not self._ready:
return
from ..draw2d import Draw2D
atlas = Draw2D._font
if atlas is None:
return
if atlas.version <= self._atlas_version:
return
self.upload_atlas(atlas.atlas, version=atlas.version, px_range=atlas.sdf_range)
[docs]
def upload_atlas(self, atlas_data: np.ndarray, version: int = 1, px_range: float = 4.0) -> None:
"""Upload MSDF atlas (RGBA uint8) to GPU via staging buffer.
Skips upload if the GPU already has this version. On version change,
destroys old image/view/memory before creating new ones.
"""
self._px_range = px_range
if not self._ready or version <= self._atlas_version:
return
e = self._engine
device = e.ctx.device
# Destroy old atlas resources if re-uploading.
# Must wait for GPU to finish: the old view may still be referenced
# by in-flight command buffers / descriptor sets.
if self._atlas_view:
vk.vkDeviceWaitIdle(device)
vk.vkDestroyImageView(device, self._atlas_view, None)
self._atlas_view = None
if self._atlas_image:
vk.vkDestroyImage(device, self._atlas_image, None)
self._atlas_image = None
if self._atlas_memory:
vk.vkFreeMemory(device, self._atlas_memory, None)
self._atlas_memory = None
h, w = atlas_data.shape[:2]
# Ensure RGBA and contiguous
if atlas_data.ndim == 2:
rgba = np.stack([atlas_data] * 4, axis=-1)
elif atlas_data.shape[2] == 3:
rgba = np.zeros((h, w, 4), dtype=np.uint8)
rgba[:, :, :3] = atlas_data
rgba[:, :, 3] = 255
else:
rgba = atlas_data
rgba = np.ascontiguousarray(rgba)
# Upload via staging buffer
self._atlas_image, self._atlas_memory = upload_image_data(
device, e.ctx.physical_device, e.ctx.graphics_queue, e.ctx.command_pool,
rgba, w, h, vk.VK_FORMAT_R8G8B8A8_UNORM,
)
# Create image view
view_ci = vk.VkImageViewCreateInfo(
image=self._atlas_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,
),
)
self._atlas_view = vk.vkCreateImageView(device, view_ci, None)
# Bind to descriptor set
write_texture_descriptor(
device, self._descriptor_set,
0, self._atlas_view, self._sampler,
)
self._atlas_version = version
[docs]
def upload_geometry(self, vertices: np.ndarray, indices: np.ndarray) -> None:
"""Upload per-frame text vertex/index data."""
if vertices is None or len(indices) == 0:
self._index_count = 0
return
max_verts = MAX_CHARS * 4
max_idx = MAX_CHARS * 6
if len(vertices) > max_verts:
log.warning("TextPass overflow: %d verts (max %d), truncating", len(vertices), max_verts)
vertices = vertices[:max_verts]
indices = indices[:max_idx]
if len(indices) > max_idx:
indices = indices[:max_idx]
upload_numpy(self._engine.ctx.device, self._vertex_memory, vertices)
upload_numpy(self._engine.ctx.device, self._index_memory, indices)
self._index_count = len(indices)
[docs]
def render(self, cmd: Any, width: int, height: int) -> None:
"""Record text draw commands into the active render pass."""
if not self._ready or self._index_count == 0 or self._atlas_version == 0:
return
vk.vkCmdBindPipeline(cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline)
# Bind atlas descriptor
vk.vkCmdBindDescriptorSets(
cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline_layout,
0, 1, [self._descriptor_set], 0, None,
)
# Push screen size + px_range
pc_data = np.array([width, height, self._px_range], dtype=np.float32)
self._engine.push_constants(cmd, self._pipeline_layout, pc_data.tobytes())
# Viewport/scissor
vk_viewport = vk.VkViewport(
x=0.0, y=0.0,
width=float(width), height=float(height),
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=width, height=height),
)
vk.vkCmdSetScissor(cmd, 0, 1, [scissor])
# Bind buffers and draw
vk.vkCmdBindVertexBuffers(cmd, 0, 1, [self._vertex_buffer], [0])
vk.vkCmdBindIndexBuffer(cmd, self._index_buffer, 0, vk.VK_INDEX_TYPE_UINT32)
vk.vkCmdDrawIndexed(cmd, self._index_count, 1, 0, 0, 0)
[docs]
def cleanup(self) -> None:
"""Release all GPU resources."""
if not self._ready:
return
device = self._engine.ctx.device
for pipe, _layout in self._extra_pipelines.values():
if pipe:
vk.vkDestroyPipeline(device, pipe, None)
self._extra_pipelines.clear()
for obj, fn in [
(self._vertex_buffer, vk.vkDestroyBuffer),
(self._index_buffer, vk.vkDestroyBuffer),
(self._atlas_image, vk.vkDestroyImage),
(self._pipeline, vk.vkDestroyPipeline),
(self._pipeline_layout, vk.vkDestroyPipelineLayout),
(self._vert_module, vk.vkDestroyShaderModule),
(self._frag_module, vk.vkDestroyShaderModule),
(self._sampler, vk.vkDestroySampler),
(self._descriptor_layout, vk.vkDestroyDescriptorSetLayout),
(self._descriptor_pool, vk.vkDestroyDescriptorPool),
]:
if obj:
fn(device, obj, None)
for mem in [self._vertex_memory, self._index_memory, self._atlas_memory]:
if mem:
vk.vkFreeMemory(device, mem, None)
if self._atlas_view:
vk.vkDestroyImageView(device, self._atlas_view, None)
self._ready = False