"""Per-CanvasLayer post-processing chain (design §5.6).
A :class:`LayerPostChain` post-processes ONE CanvasLayer band whose
``environment`` Property is set. It is created lazily, the first frame that band
is both opted-in AND present, so a scene with no per-layer environment allocates
nothing here (the zero-cost-when-unused contract: the renderer never instantiates
this class unless ``Renderer._layer_post_specs`` is non-empty).
Pipeline (mirrors the global 2D-in-HDR + bloom path, scoped to one band):
1. Draw the band's 2D items into an offscreen RGBA16F HDR target (cleared
transparent, alpha 0) via a band-filtered bindless submit. Only items whose
``layer`` column equals this band are drawn (``submit.build_bindless_geometry``
``only_band=``).
2. Run :class:`~simvx.graphics.renderer.bloom_pass.BloomPass` on that HDR colour
to produce the glow (when the band env enabled bloom).
3. Composite onto the swapchain with ``layer_post_composite.frag``: bloom add +
tonemap + vignette / grain / chromatic, with the band's COVERAGE ALPHA
preserved, alpha-blended over whatever is already on the swapchain. The host
composites the bands in ascending-band order so painter order with the plain
bands (drawn through the global path) is correct.
Everything is reused (RenderTarget, BloomPass, the bindless 2D submitter, the
WorldEnvironment effect surface); the only new artefact is the composite shader,
which is plumbing (a tonemap-with-preserved-alpha blit), not a new effect class.
"""
from __future__ import annotations
import time
from typing import Any
import numpy as np
import vulkan as vk
from ..gpu.memory import create_sampler
from ..gpu.pipeline import PipelineSpec, build_pipeline, create_shader_module
from ..materials.shader_compiler import compile_shader
from ..render2d.submit import BindlessItemSubmitter
from ..types import SHADER_DIR
from .bindless_draw2d_pass import BindlessDraw2DPass
from .bloom_pass import BloomPass
from .render_target import RenderTarget
__all__ = ["LayerPostChain", "LayerPostSpec"]
# Composite push constants (48 bytes); must match layer_post_composite.frag PC.
_PC_SIZE = 48
_FLAG_BLOOM = 1 << 1
_FLAG_GRAIN = 1 << 4
_FLAG_VIGNETTE = 1 << 5
_FLAG_CHROMATIC = 1 << 6
# Tonemap mode names -> the int the composite shader switches on (matches the web
# WGSL order; 4 = linear, used for flat 2D art so it is not ACES-crushed).
_TONEMAP_MODES = {"aces": 0, "neutral": 1, "reinhard": 2, "uchimura": 3, "linear": 4}
[docs]
class LayerPostSpec:
"""Resolved, render-thread-readable snapshot of a band env's post settings.
Built by ``environment_sync`` from a CanvasLayer's ``environment``
WorldEnvironment. Flat scalars only (no node references) so the render thread
never touches the live scene tree.
"""
__slots__ = (
"bloom_enabled", "bloom_threshold", "bloom_intensity",
"vignette_enabled", "vignette_intensity", "vignette_smoothness",
"grain_enabled", "grain_intensity",
"chromatic_enabled", "chromatic_intensity",
"tonemap_mode", "tonemap_exposure",
)
def __init__(self, env: Any) -> None:
self.bloom_enabled = bool(env.bloom_enabled)
self.bloom_threshold = float(env.bloom_threshold)
self.bloom_intensity = float(env.bloom_intensity)
self.vignette_enabled = bool(env.vignette_enabled)
self.vignette_intensity = float(env.vignette_intensity)
self.vignette_smoothness = float(env.vignette_smoothness)
self.grain_enabled = bool(env.film_grain_enabled)
self.grain_intensity = float(env.film_grain_intensity)
self.chromatic_enabled = bool(env.chromatic_aberration_enabled)
self.chromatic_intensity = float(env.chromatic_aberration_intensity)
# Per-layer flat 2D art uses linear tonemap by default (no ACES crush), the
# same choice the pure-2D global path makes; an author can still pick a
# real operator via the band env's tonemap_mode.
mode = getattr(env, "tonemap_mode", "linear")
self.tonemap_mode = _TONEMAP_MODES.get(str(mode), 4)
self.tonemap_exposure = float(getattr(env, "tonemap_exposure", 1.0))
[docs]
def __eq__(self, other: Any) -> bool:
if not isinstance(other, LayerPostSpec):
return NotImplemented
return all(getattr(self, s) == getattr(other, s) for s in self.__slots__)
[docs]
def applies(self) -> bool:
"""Whether this band actually needs a post chain (any effect enabled)."""
return (
self.bloom_enabled or self.vignette_enabled
or self.grain_enabled or self.chromatic_enabled
)
[docs]
class LayerPostChain:
"""Lazily-created per-band HDR target + bloom + composite (design §5.6).
One instance per opted-in, present band. Built against the current swapchain
extent; recreated on resize. The renderer reaps it when the band's env is
cleared (mirrors the velocity/occlusion lazy-pass cleanup).
"""
def __init__(self, engine: Any, band: int, text_pass: Any) -> None:
self._engine = engine
self.band = band
w, h = engine.extent
# Offscreen HDR target for this band's 2D (cleared transparent each frame).
self._hdr = RenderTarget(
engine.ctx.device, engine.ctx.physical_device, w, h,
colour_format=vk.VK_FORMAT_R16G16B16A16_SFLOAT, use_depth=False,
queue=engine.ctx.graphics_queue, command_pool=engine.ctx.command_pool,
)
# Band 2D pass bound to the band HDR target's (single-colour) render pass.
self._pass = BindlessDraw2DPass(engine, text_pass=text_pass)
self._pass.setup(render_pass=self._hdr.render_pass, extent=(w, h))
self._submitter = BindlessItemSubmitter(self._pass)
# Bloom on the band HDR colour (always built; only run when the band wants
# bloom, exactly like the global PostProcessPass).
self._bloom = BloomPass(engine)
self._bloom.setup(self._hdr.colour_view)
self._sampler = create_sampler(engine.ctx.device)
self._start_time = time.perf_counter()
self._last_spec: LayerPostSpec | None = None
self._pipeline = self._pipeline_layout = self._vert = self._frag = None
self._build_composite_pipeline()
# -- composite pipeline (HDR colour + bloom -> swapchain, alpha-blended) --
def _build_composite_pipeline(self) -> None:
e = self._engine
device = e.ctx.device
bindings = [
vk.VkDescriptorSetLayoutBinding(
binding=b, descriptorType=vk.VK_DESCRIPTOR_TYPE_COMBINED_IMAGE_SAMPLER,
descriptorCount=1, stageFlags=vk.VK_SHADER_STAGE_FRAGMENT_BIT,
)
for b in (0, 1)
]
self._desc_layout = vk.vkCreateDescriptorSetLayout(
device, vk.VkDescriptorSetLayoutCreateInfo(bindingCount=2, pBindings=bindings), None,
)
self._desc_pool = vk.vkCreateDescriptorPool(
device, vk.VkDescriptorPoolCreateInfo(
maxSets=1, poolSizeCount=1,
pPoolSizes=[vk.VkDescriptorPoolSize(
type=vk.VK_DESCRIPTOR_TYPE_COMBINED_IMAGE_SAMPLER, descriptorCount=2,
)],
), None,
)
self._desc_set = vk.vkAllocateDescriptorSets(
device, vk.VkDescriptorSetAllocateInfo(
descriptorPool=self._desc_pool, descriptorSetCount=1, pSetLayouts=[self._desc_layout],
),
)[0]
self._write_composite_descriptors()
self._vert = create_shader_module(device, compile_shader(SHADER_DIR / "tonemap.vert"))
self._frag = create_shader_module(device, compile_shader(SHADER_DIR / "layer_post_composite.frag"))
spec = PipelineSpec(
name=f"layer_post_composite_{self.band}",
topology=vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST,
vertex_stride=0, 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._desc_layout,),
push_size=_PC_SIZE, push_stages=vk.VK_SHADER_STAGE_FRAGMENT_BIT,
)
self._pipeline, self._pipeline_layout = build_pipeline(
device, spec, e.render_pass, e.extent, vert_module=self._vert, frag_module=self._frag,
)
def _write_composite_descriptors(self) -> None:
device = self._engine.ctx.device
infos = [
vk.VkDescriptorImageInfo(
sampler=self._sampler, imageView=self._hdr.colour_view,
imageLayout=vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
),
vk.VkDescriptorImageInfo(
sampler=self._sampler, imageView=self._bloom.bloom_image_view or self._hdr.colour_view,
imageLayout=vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
),
]
writes = [
vk.VkWriteDescriptorSet(
dstSet=self._desc_set, dstBinding=i, dstArrayElement=0, descriptorCount=1,
descriptorType=vk.VK_DESCRIPTOR_TYPE_COMBINED_IMAGE_SAMPLER, pImageInfo=[info],
)
for i, info in enumerate(infos)
]
vk.vkUpdateDescriptorSets(device, len(writes), writes, 0, None)
[docs]
def sync_atlas_slot(self, slot: int) -> None:
"""Mirror the swapchain pass's MSDF atlas slot (shared bindless array)."""
self._pass.set_atlas_slot(slot)
[docs]
def render_band_hdr(self, cmd: Any, view: Any, camera: tuple, ui_size: tuple[int, int]) -> None:
"""Draw this band's items into the HDR target, then run bloom (outside it).
Records the offscreen HDR render pass (cleared transparent) with the band's
items, ends it, then runs bloom on the resulting colour. Bloom records its
own passes (illegal to nest), so it must run after the HDR pass ends.
"""
rt = self._hdr
rt.begin_frame_barrier(cmd)
clear = vk.VkClearValue(color=vk.VkClearColorValue(float32=[0.0, 0.0, 0.0, 0.0]))
vk.vkCmdBeginRenderPass(cmd, vk.VkRenderPassBeginInfo(
renderPass=rt.render_pass, framebuffer=rt.framebuffer,
renderArea=vk.VkRect2D(offset=vk.VkOffset2D(x=0, y=0),
extent=vk.VkExtent2D(width=rt.width, height=rt.height)),
clearValueCount=1, pClearValues=[clear],
), vk.VK_SUBPASS_CONTENTS_INLINE)
self._submitter.render(
cmd, view, rt.width, rt.height, ui_width=ui_size[0], ui_height=ui_size[1],
camera=camera, lane="all", only_band=self.band,
)
vk.vkCmdEndRenderPass(cmd)
if self._spec_bloom():
self._bloom.render(cmd)
[docs]
def configure(self, spec: LayerPostSpec) -> None:
"""Apply the band env's settings (bloom threshold/intensity etc.)."""
self._last_spec = spec
self._bloom.threshold = spec.bloom_threshold
def _spec_bloom(self) -> bool:
return bool(self._last_spec and self._last_spec.bloom_enabled)
[docs]
def composite(self, cmd: Any, width: int, height: int) -> None:
"""Alpha-blend this band's post result onto the current (swapchain) pass."""
spec = self._last_spec
if spec is None:
return
vk.vkCmdSetViewport(cmd, 0, 1, [vk.VkViewport(
x=0.0, y=0.0, width=float(width), height=float(height), minDepth=0.0, maxDepth=1.0)])
vk.vkCmdSetScissor(cmd, 0, 1, [vk.VkRect2D(
offset=vk.VkOffset2D(x=0, y=0), extent=vk.VkExtent2D(width=width, height=height))])
vk.vkCmdBindPipeline(cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline)
vk.vkCmdBindDescriptorSets(
cmd, vk.VK_PIPELINE_BIND_POINT_GRAPHICS, self._pipeline_layout,
0, 1, [self._desc_set], 0, None,
)
flags = 0
if spec.bloom_enabled:
flags |= _FLAG_BLOOM
if spec.vignette_enabled:
flags |= _FLAG_VIGNETTE
if spec.grain_enabled:
flags |= _FLAG_GRAIN
if spec.chromatic_enabled:
flags |= _FLAG_CHROMATIC
elapsed = time.perf_counter() - self._start_time
pc = bytearray(_PC_SIZE)
pc[0:8] = np.array([float(width), float(height)], dtype=np.float32).tobytes()
pc[8:12] = np.array([spec.tonemap_exposure], dtype=np.float32).tobytes()
pc[12:16] = np.array([flags], dtype=np.uint32).tobytes()
pc[16:20] = np.array([spec.bloom_intensity], dtype=np.float32).tobytes()
pc[20:24] = np.array([spec.grain_intensity], dtype=np.float32).tobytes()
pc[24:28] = np.array([spec.vignette_intensity], dtype=np.float32).tobytes()
pc[28:32] = np.array([spec.vignette_smoothness], dtype=np.float32).tobytes()
pc[32:36] = np.array([spec.chromatic_intensity], dtype=np.float32).tobytes()
pc[36:40] = np.array([elapsed], dtype=np.float32).tobytes()
pc[40:44] = np.array([float(spec.tonemap_mode)], dtype=np.float32).tobytes()
pc[44:48] = np.array([1.0], dtype=np.float32).tobytes()
ffi = vk.ffi
vk._vulkan.lib.vkCmdPushConstants(
cmd, self._pipeline_layout, vk.VK_SHADER_STAGE_FRAGMENT_BIT, 0, _PC_SIZE,
ffi.new("char[]", bytes(pc)),
)
vk.vkCmdDraw(cmd, 3, 1, 0, 0)
[docs]
def cleanup(self) -> None:
device = self._engine.ctx.device
vk.vkDeviceWaitIdle(device)
if self._pipeline:
vk.vkDestroyPipeline(device, self._pipeline, None)
if self._pipeline_layout:
vk.vkDestroyPipelineLayout(device, self._pipeline_layout, None)
if self._vert:
vk.vkDestroyShaderModule(device, self._vert, None)
if self._frag:
vk.vkDestroyShaderModule(device, self._frag, None)
if self._desc_pool:
vk.vkDestroyDescriptorPool(device, self._desc_pool, None)
if self._desc_layout:
vk.vkDestroyDescriptorSetLayout(device, self._desc_layout, None)
if self._sampler:
vk.vkDestroySampler(device, self._sampler, None)
# The band 2D pass shares the engine bindless atlas slot (owned by the
# swapchain pass); clear the borrowed slot first so cleanup() does not
# unregister it (mirrors the N1 HDR-lane pass cleanup in forward.py).
self._pass.set_atlas_slot(-1)
self._pass.cleanup()
self._bloom.cleanup()
self._hdr.destroy()