Source code for simvx.graphics.gpu.pipeline

"""Graphics pipeline and shader module management."""

import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal

import vulkan as vk

from .pipeline_cache import pipeline_cache_for

log = logging.getLogger(__name__)

__all__ = [
    "PipelineSpec",
    "VertexAttr",
    "VertexBinding",
    "build_pipeline",
    "create_shader_module",
    "POS_COLOUR_VERTEX_STRIDE",
    "POS_COLOUR_VERTEX_ATTRS",
    "UI_VERTEX_STRIDE",
    "UI_VERTEX_ATTRS",
    "UI2D_VERTEX_STRIDE",
    "UI2D_VERTEX_ATTRS",
    "MESH_PUSH_CONSTANT_SIZE",
]

# A vertex attribute: (location, VkFormat, byte offset within the vertex).
VertexAttr = tuple[int, int, int]

# One vertex-buffer binding: (binding number, stride, attributes). Explicit
# binding numbers allow sparse layouts (the skinned mesh pipeline binds
# streams 0, 1, and 3 without a binding-2 extras stream, D5). The 3D mesh
# stream layouts live in ``renderer/vertex_layouts.py`` (single source).
VertexBinding = tuple[int, int, tuple[VertexAttr, ...]]

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

# Mesh push-constant block: mat4 view (64) + mat4 proj (64) + uint hdr_output (4).
# Shared verbatim by cube.vert / skinned.vert (vertex) and cube_textured.frag
# (fragment), so every mesh pipeline layout must reserve the full block.
MESH_PUSH_CONSTANT_SIZE = 132

_COLOUR_WRITE_ALL = (
    vk.VK_COLOR_COMPONENT_R_BIT
    | vk.VK_COLOR_COMPONENT_G_BIT
    | vk.VK_COLOR_COMPONENT_B_BIT
    | vk.VK_COLOR_COMPONENT_A_BIT
)

# ---------------------------------------------------------------------------
# Internal cffi struct builders
# ---------------------------------------------------------------------------

def _make_shader_stages(ffi: Any, vert_module: Any, frag_module: Any) -> tuple[Any, Any]:
    """Build a 2-stage (vert + frag) shader stage array.  Returns (stages, main_name)."""
    stages = ffi.new("VkPipelineShaderStageCreateInfo[2]")
    main_name = ffi.new("char[]", b"main")
    for i, (stage_bit, module) in enumerate([
        (vk.VK_SHADER_STAGE_VERTEX_BIT, vert_module),
        (vk.VK_SHADER_STAGE_FRAGMENT_BIT, frag_module),
    ]):
        stages[i].sType = vk.VK_STRUCTURE_TYPE_PIPELINE_SHADER_STAGE_CREATE_INFO
        stages[i].stage = stage_bit
        stages[i].module = module
        stages[i].pName = main_name
    return stages, main_name

def _make_vertex_input(
    ffi: Any,
    bindings: tuple[VertexBinding, ...],
) -> tuple[Any, Any, Any]:
    """Build vertex input state for one or more vertex-buffer bindings.

    *bindings* is a tuple of ``(binding, stride, attributes)`` triples with
    explicit Vulkan binding numbers (sparse layouts allowed, e.g. 0/1/3 for
    skinned meshes). Each attribute is a (location, format, offset) tuple,
    with the offset relative to its own binding's vertex.
    Returns (vi, binding_descs, attr_descs) -- all three must stay alive until pipeline creation.
    """
    n_bind = len(bindings)
    binding_descs = ffi.new(f"VkVertexInputBindingDescription[{n_bind}]")
    flat_attrs: list[tuple[int, int, int, int]] = []  # (binding, location, format, offset)
    for i, (b, stride, attrs) in enumerate(bindings):
        binding_descs[i].binding = b
        binding_descs[i].stride = stride
        binding_descs[i].inputRate = vk.VK_VERTEX_INPUT_RATE_VERTEX
        flat_attrs.extend((b, loc, fmt, off) for loc, fmt, off in attrs)

    n_attr = len(flat_attrs)
    attr_descs = ffi.new(f"VkVertexInputAttributeDescription[{n_attr}]")
    for i, (b, loc, fmt, off) in enumerate(flat_attrs):
        attr_descs[i].location = loc
        attr_descs[i].binding = b
        attr_descs[i].format = fmt
        attr_descs[i].offset = off

    vi = ffi.new("VkPipelineVertexInputStateCreateInfo*")
    vi.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_VERTEX_INPUT_STATE_CREATE_INFO
    vi.vertexBindingDescriptionCount = n_bind
    vi.pVertexBindingDescriptions = binding_descs
    vi.vertexAttributeDescriptionCount = n_attr
    vi.pVertexAttributeDescriptions = attr_descs
    return vi, binding_descs, attr_descs

def _make_empty_vertex_input(ffi: Any) -> Any:
    """Build an empty vertex input state (no vertex buffers)."""
    vi = ffi.new("VkPipelineVertexInputStateCreateInfo*")
    vi.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_VERTEX_INPUT_STATE_CREATE_INFO
    return vi

def _make_input_assembly(ffi: Any, topology: int = vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST) -> Any:
    """Build input assembly state."""
    ia = ffi.new("VkPipelineInputAssemblyStateCreateInfo*")
    ia.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_INPUT_ASSEMBLY_STATE_CREATE_INFO
    ia.topology = topology
    return ia

def _make_viewport_state(ffi: Any, extent: tuple[int, int]) -> tuple[Any, Any, Any]:
    """Build viewport state.  Returns (vps, viewport, scissor) -- all must stay alive."""
    vps = ffi.new("VkPipelineViewportStateCreateInfo*")
    vps.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_VIEWPORT_STATE_CREATE_INFO
    vps.viewportCount = 1
    viewport = ffi.new("VkViewport*")
    viewport.width = float(extent[0])
    viewport.height = float(extent[1])
    viewport.maxDepth = 1.0
    vps.pViewports = viewport
    scissor = ffi.new("VkRect2D*")
    scissor.extent.width = extent[0]
    scissor.extent.height = extent[1]
    vps.scissorCount = 1
    vps.pScissors = scissor
    return vps, viewport, scissor

def _make_rasterization(
    ffi: Any,
    cull_mode: int = vk.VK_CULL_MODE_BACK_BIT,
    front_face: int = vk.VK_FRONT_FACE_COUNTER_CLOCKWISE,
    depth_bias: tuple[float, float] | None = None,
) -> Any:
    """Build rasterization state.

    *depth_bias*, when given, is ``(constantFactor, slopeFactor)``: enables
    depth bias with those factors (shadow maps use this to fight acne).
    """
    rs = ffi.new("VkPipelineRasterizationStateCreateInfo*")
    rs.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_RASTERIZATION_STATE_CREATE_INFO
    rs.polygonMode = vk.VK_POLYGON_MODE_FILL
    rs.lineWidth = 1.0
    rs.cullMode = cull_mode
    rs.frontFace = front_face
    if depth_bias is not None:
        rs.depthBiasEnable = 1
        rs.depthBiasConstantFactor = depth_bias[0]
        rs.depthBiasSlopeFactor = depth_bias[1]
    return rs

def _make_multisample(ffi: Any) -> Any:
    """Build multisample state (1x, no MSAA)."""
    ms = ffi.new("VkPipelineMultisampleStateCreateInfo*")
    ms.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_MULTISAMPLE_STATE_CREATE_INFO
    ms.rasterizationSamples = vk.VK_SAMPLE_COUNT_1_BIT
    return ms

def _make_depth_stencil(
    ffi: Any,
    *,
    test: bool = True,
    write: bool = True,
    compare_op: int = vk.VK_COMPARE_OP_LESS,
) -> Any:
    """Build depth/stencil state."""
    dss = ffi.new("VkPipelineDepthStencilStateCreateInfo*")
    dss.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_DEPTH_STENCIL_STATE_CREATE_INFO
    dss.depthTestEnable = int(test)
    dss.depthWriteEnable = int(write)
    dss.depthCompareOp = compare_op
    return dss

def _make_colour_blend_state(ffi: Any, attachments: Any, count: int) -> Any:
    """Wrap a pre-filled attachment-state array in a VkPipelineColorBlendStateCreateInfo."""
    cb = ffi.new("VkPipelineColorBlendStateCreateInfo*")
    cb.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_COLOR_BLEND_STATE_CREATE_INFO
    cb.attachmentCount = count
    cb.pAttachments = attachments
    return cb

def _make_colour_blend_opaque(
    ffi: Any, write_mask: int = _COLOUR_WRITE_ALL, attachment_count: int = 1
) -> tuple[Any, Any]:
    """Build opaque colour blend state (no blending).  Returns (cb, cba).

    The same attachment state is replicated across *attachment_count* colour
    attachments (multi-render-target passes).
    """
    cba = ffi.new(f"VkPipelineColorBlendAttachmentState[{attachment_count}]")
    for i in range(attachment_count):
        cba[i].colorWriteMask = write_mask
    return _make_colour_blend_state(ffi, cba, attachment_count), cba

def _make_colour_blend_alpha(
    ffi: Any,
    dst_alpha_factor: int = vk.VK_BLEND_FACTOR_ZERO,
    attachment_count: int = 1,
) -> tuple[Any, Any]:
    """Build alpha-blended colour blend state.  Returns (cb, cba).

    src colour = srcAlpha, dst colour = 1-srcAlpha.
    *dst_alpha_factor* controls destination alpha (ZERO for overlay, ONE_MINUS_SRC_ALPHA for compositing).
    The same attachment state is replicated across *attachment_count* colour attachments.
    """
    cba = ffi.new(f"VkPipelineColorBlendAttachmentState[{attachment_count}]")
    for i in range(attachment_count):
        cba[i].blendEnable = 1
        cba[i].srcColorBlendFactor = vk.VK_BLEND_FACTOR_SRC_ALPHA
        cba[i].dstColorBlendFactor = vk.VK_BLEND_FACTOR_ONE_MINUS_SRC_ALPHA
        cba[i].colorBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].srcAlphaBlendFactor = vk.VK_BLEND_FACTOR_ONE
        cba[i].dstAlphaBlendFactor = dst_alpha_factor
        cba[i].alphaBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].colorWriteMask = _COLOUR_WRITE_ALL
    return _make_colour_blend_state(ffi, cba, attachment_count), cba

def _make_colour_blend_add(ffi: Any, attachment_count: int = 1) -> tuple[Any, Any]:
    """Additive blend: ``dst += src.rgb * src.a``.  Returns (cb, cba).

    src colour = srcAlpha (so alpha scales the contribution), dst colour = ONE
    (the framebuffer is preserved and the source is added on top). Alpha is
    accumulated with ONE/ONE so the framebuffer alpha saturates rather than
    being overwritten, matching the colour channels' additive intent.
    The same attachment state is replicated across *attachment_count* colour attachments.
    """
    cba = ffi.new(f"VkPipelineColorBlendAttachmentState[{attachment_count}]")
    for i in range(attachment_count):
        cba[i].blendEnable = 1
        cba[i].srcColorBlendFactor = vk.VK_BLEND_FACTOR_SRC_ALPHA
        cba[i].dstColorBlendFactor = vk.VK_BLEND_FACTOR_ONE
        cba[i].colorBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].srcAlphaBlendFactor = vk.VK_BLEND_FACTOR_ONE
        cba[i].dstAlphaBlendFactor = vk.VK_BLEND_FACTOR_ONE
        cba[i].alphaBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].colorWriteMask = _COLOUR_WRITE_ALL
    return _make_colour_blend_state(ffi, cba, attachment_count), cba

def _make_colour_blend_multiply(ffi: Any, attachment_count: int = 1) -> tuple[Any, Any]:
    """Multiply blend: ``dst = dst * src``.  Returns (cb, cba).

    Uses ``srcFactor = DST_COLOR, dstFactor = ZERO`` so the result is the
    product of the source colour and the existing framebuffer colour, the
    standard "multiply" / "darken" overlay. Premultiply caveat: the source
    alpha does NOT scale the multiply here (factor is DST_COLOR, not
    SRC_ALPHA*DST_COLOR), so a multiply overlay should encode its strength in
    the RGB channels (e.g. a 50% grey for half-darkening), not in alpha. Alpha
    is left as the source alpha (ONE/ZERO) so a fully-opaque overlay keeps the
    framebuffer opaque. The same attachment state is replicated across
    *attachment_count* colour attachments.
    """
    cba = ffi.new(f"VkPipelineColorBlendAttachmentState[{attachment_count}]")
    for i in range(attachment_count):
        cba[i].blendEnable = 1
        cba[i].srcColorBlendFactor = vk.VK_BLEND_FACTOR_DST_COLOR
        cba[i].dstColorBlendFactor = vk.VK_BLEND_FACTOR_ZERO
        cba[i].colorBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].srcAlphaBlendFactor = vk.VK_BLEND_FACTOR_ONE
        cba[i].dstAlphaBlendFactor = vk.VK_BLEND_FACTOR_ZERO
        cba[i].alphaBlendOp = vk.VK_BLEND_OP_ADD
        cba[i].colorWriteMask = _COLOUR_WRITE_ALL
    return _make_colour_blend_state(ffi, cba, attachment_count), cba

def _make_colour_blend_none(ffi: Any) -> tuple[Any]:
    """Build a colour blend state with zero attachments (depth-only passes).

    Returns a 1-tuple ``(cb,)`` so callers can splat it into a ``keep`` list
    the same way the other ``_make_colour_blend_*`` helpers return their
    sub-allocations.
    """
    cb = ffi.new("VkPipelineColorBlendStateCreateInfo*")
    cb.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_COLOR_BLEND_STATE_CREATE_INFO
    cb.attachmentCount = 0
    return (cb,)

def _make_dynamic_state(ffi: Any) -> tuple[Any, Any]:
    """Build dynamic state for viewport + scissor.  Returns (ds, dyn_states)."""
    dyn_states = ffi.new("VkDynamicState[2]", [vk.VK_DYNAMIC_STATE_VIEWPORT, vk.VK_DYNAMIC_STATE_SCISSOR])
    ds = ffi.new("VkPipelineDynamicStateCreateInfo*")
    ds.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_DYNAMIC_STATE_CREATE_INFO
    ds.dynamicStateCount = 2
    ds.pDynamicStates = dyn_states
    return ds, dyn_states

def _create_pipeline_layout(
    ffi: Any,
    device: Any,
    descriptor_layouts: list[Any] | None = None,
    push_constant_size: int = 0,
    push_stage_flags: int = vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT,
) -> Any:
    """Create a VkPipelineLayout via raw cffi.

    Returns the pipeline layout handle.  Raises RuntimeError on failure.
    """
    layout_ci = ffi.new("VkPipelineLayoutCreateInfo*")
    layout_ci.sType = vk.VK_STRUCTURE_TYPE_PIPELINE_LAYOUT_CREATE_INFO

    # Keep cffi allocations alive until the call completes
    _set_layouts = None
    _push_range = None

    if descriptor_layouts:
        n = len(descriptor_layouts)
        _set_layouts = ffi.new(f"VkDescriptorSetLayout[{n}]", descriptor_layouts)
        layout_ci.setLayoutCount = n
        layout_ci.pSetLayouts = _set_layouts

    if push_constant_size > 0:
        _push_range = ffi.new("VkPushConstantRange*")
        _push_range.stageFlags = push_stage_flags
        _push_range.offset = 0
        _push_range.size = push_constant_size
        layout_ci.pushConstantRangeCount = 1
        layout_ci.pPushConstantRanges = _push_range

    layout_out = ffi.new("VkPipelineLayout*")
    result = vk._vulkan._callApi(
        vk._vulkan.lib.vkCreatePipelineLayout,
        device,
        layout_ci,
        ffi.NULL,
        layout_out,
    )
    if result != vk.VK_SUCCESS:
        raise RuntimeError(f"vkCreatePipelineLayout failed: {result}")
    return layout_out[0]

def _build_pipeline(ffi: Any, device: Any, pi: Any, name: str) -> Any:
    """Call vkCreateGraphicsPipelines and return the pipeline handle."""
    pipeline_out = ffi.new("VkPipeline*")
    result = vk._vulkan._callApi(
        vk._vulkan.lib.vkCreateGraphicsPipelines,
        device,
        pipeline_cache_for(device),
        1,
        pi,
        ffi.NULL,
        pipeline_out,
    )
    if result != vk.VK_SUCCESS:
        raise RuntimeError(f"vkCreateGraphicsPipelines failed: {result}")
    log.debug("%s pipeline created", name)
    return pipeline_out[0]

# ---------------------------------------------------------------------------
# Vertex format presets: (location, format, offset) tuples
# ---------------------------------------------------------------------------

# 3D mesh vertex layouts (positions / normal+uv / extras / skin streams, D5)
# live in ``simvx.graphics.renderer.vertex_layouts`` -- the single source for
# every mesh pipeline's vertex bindings. Only the non-mesh, single-buffer
# presets below stay here.

# position(vec3) + colour(vec4) = 28 bytes. Shared verbatim by the debug-line
# pipeline (overlay_renderer) and the gizmo overlay pipelines (gizmo_pass).
POS_COLOUR_VERTEX_STRIDE = 28
POS_COLOUR_VERTEX_ATTRS: tuple[VertexAttr, ...] = (
    (0, vk.VK_FORMAT_R32G32B32_SFLOAT, 0),      # position
    (1, vk.VK_FORMAT_R32G32B32A32_SFLOAT, 12),  # colour
)

# 2D UI vertex (ui.vert): position(vec2) + uv(vec2) + colour(vec4) = 32 bytes.
# Shared verbatim by all three Draw2DPass pipelines (fill, line, textured quad);
# matches UI_VERTEX_DTYPE in draw2d_vertex.py.
UI_VERTEX_STRIDE = 32
UI_VERTEX_ATTRS: tuple[VertexAttr, ...] = (
    (0, vk.VK_FORMAT_R32G32_SFLOAT, 0),         # position
    (1, vk.VK_FORMAT_R32G32_SFLOAT, 8),         # uv
    (2, vk.VK_FORMAT_R32G32B32A32_SFLOAT, 16),  # colour
)

# Extended 2D UI vertex for the bindless co-batched item path (ui2d.vert):
# the 32-byte UI vertex plus a per-vertex bindless texture slot (int) + flags
# (uint, bit0 = is_msdf). Matches UI2D_VERTEX_DTYPE in draw2d_vertex.py. 40 bytes.
UI2D_VERTEX_STRIDE = 40
UI2D_VERTEX_ATTRS: tuple[VertexAttr, ...] = (
    (0, vk.VK_FORMAT_R32G32_SFLOAT, 0),         # position
    (1, vk.VK_FORMAT_R32G32_SFLOAT, 8),         # uv
    (2, vk.VK_FORMAT_R32G32B32A32_SFLOAT, 16),  # colour
    (3, vk.VK_FORMAT_R32_SINT, 32),             # tex_id (bindless slot, -1 = none)
    (4, vk.VK_FORMAT_R32_UINT, 36),             # flags (bit0 = is_msdf)
)

# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

[docs] def create_shader_module(device: Any, spirv_path: Path) -> Any: """Load a SPIR-V file and create a VkShaderModule.""" code = spirv_path.read_bytes() create_info = vk.VkShaderModuleCreateInfo( codeSize=len(code), pCode=code, ) module = vk.vkCreateShaderModule(device, create_info, None) log.debug("Shader module loaded: %s", spirv_path.name) return module
# --------------------------------------------------------------------------- # Declarative pipeline builder # ---------------------------------------------------------------------------
[docs] @dataclass(frozen=True) class PipelineSpec: """Declarative description of a graphics pipeline's varying state. Captures *only* the fixed-function and layout state that genuinely differs between SimVX's render passes; everything constant across every surveyed pass (polygon mode FILL, line width 1.0, 1x MSAA, entry point ``"main"``, dynamic viewport + scissor) is fixed inside :func:`build_pipeline`. The render pass, framebuffer extent, and (optionally) pre-created shader modules are *late-bound* arguments to :func:`build_pipeline` rather than spec fields, so one immutable spec can be rebuilt against a different render pass (e.g. the HDR ``R16G16B16A16_SFLOAT`` offscreen pass vs the swapchain ``B8G8R8A8_SRGB`` pass) without copying. Vertex input: either the legacy single-binding pair (*vertex_stride* + *vertex_attrs*, binding 0) or the multi-binding *vertex_bindings* tuple of ``(binding, stride, attrs)`` triples with explicit (possibly sparse) Vulkan binding numbers (vertex stream split, D5). The two forms are mutually exclusive; the :attr:`vertex_input_bindings` property presents both uniformly. ``vertex_stride == 0`` with no *vertex_bindings* selects an empty vertex input state (geometry generated in the shader). Colour blend: ``"opaque"`` (no blending), ``"alpha"`` (src-alpha / one-minus-src-alpha), ``"add"`` (additive, src.a-scaled over ONE), ``"multiply"`` (dst * src), or ``"none"`` (zero colour attachments, for depth-only passes). *attachment_count* (default 1) replicates the blend state across that many colour attachments for multi-render-target passes (thin G-buffer, D3); it must stay 1 when blend is ``"none"``. """ # --- identity / shaders --- name: str vert_spirv: Path | None = None # .spv path; ignored if modules passed to build_pipeline frag_spirv: Path | None = None # --- input assembly + vertex input --- topology: int = vk.VK_PRIMITIVE_TOPOLOGY_TRIANGLE_LIST vertex_stride: int = 0 # 0 => empty vertex input (shader-generated geometry) vertex_attrs: tuple[VertexAttr, ...] = () vertex_bindings: tuple[VertexBinding, ...] = () # multi-binding form; excludes the two above # --- rasterization --- cull_mode: int = vk.VK_CULL_MODE_BACK_BIT front_face: int = vk.VK_FRONT_FACE_COUNTER_CLOCKWISE depth_bias: tuple[float, float] | None = None # (constantFactor, slopeFactor) # --- depth / stencil --- depth_test: bool = True depth_write: bool = True depth_compare: int = vk.VK_COMPARE_OP_LESS # --- colour blend --- blend: Literal["opaque", "alpha", "add", "multiply", "none"] = "opaque" colour_write_mask: int = _COLOUR_WRITE_ALL dst_alpha_factor: int = vk.VK_BLEND_FACTOR_ZERO # only used when blend == "alpha" attachment_count: int = 1 # colour attachments sharing the blend state (MRT) # Thin G-buffer: when the pass has a second colour attachment # (attachment_count >= 2) this flag decides whether THIS pipeline writes it. # The lit uber sets it True (emits octahedral normal + roughness into # attachment 1); every non-lit / transparent pipeline leaves it False, so # build_pipeline masks their extra attachments to zero writes (they share # the render pass but never touch the normal target). Ignored when # attachment_count == 1, so single-attachment specs are byte-identical. writes_gbuffer: bool = False # --- layout --- set_layouts: tuple[Any, ...] = () # VkDescriptorSetLayout handles push_size: int = 0 # bytes; 0 => no push-constant range push_stages: int = field( default=vk.VK_SHADER_STAGE_VERTEX_BIT | vk.VK_SHADER_STAGE_FRAGMENT_BIT )
[docs] def __post_init__(self) -> None: if self.vertex_bindings and (self.vertex_stride or self.vertex_attrs): raise ValueError( f"PipelineSpec({self.name}): vertex_bindings excludes vertex_stride/vertex_attrs" ) if self.attachment_count < 1: raise ValueError(f"PipelineSpec({self.name}): attachment_count must be >= 1") if self.blend == "none" and self.attachment_count != 1: raise ValueError( f"PipelineSpec({self.name}): blend 'none' has zero attachments; leave attachment_count at 1" )
[docs] @property def vertex_input_bindings(self) -> tuple[VertexBinding, ...]: """The vertex input as a binding tuple, whichever form the spec used. Empty means no vertex buffers (shader-generated geometry). The legacy single-binding fields wrap into a one-entry tuple so both forms build identical pipelines through one path. """ if self.vertex_bindings: return self.vertex_bindings if self.vertex_stride == 0: return () return ((0, self.vertex_stride, self.vertex_attrs),)
[docs] def build_pipeline( device: Any, spec: PipelineSpec, render_pass: Any, extent: tuple[int, int], *, vert_module: Any = None, frag_module: Any = None, ) -> tuple[Any, Any]: """Create a ``(VkPipeline, VkPipelineLayout)`` from a declarative *spec*. This is the single high-level pipeline-creation entry point: a render pass declares *what* pipeline it wants via :class:`PipelineSpec` and never touches ``ffi.new`` itself. Internally it composes the private ``_make_*`` sub-struct builders so there is one shared creation path. Args: device: The ``VkDevice``. spec: The immutable pipeline description. render_pass: The ``VkRenderPass`` the pipeline targets (late-bound so one spec can build against the HDR or swapchain pass). extent: ``(width, height)`` for the (dynamic) viewport/scissor placeholders required at create time. vert_module / frag_module: Optional pre-created ``VkShaderModule`` handles. When supplied, they are used directly and *not* destroyed by this function (the caller owns them); this serves passes that compile GLSL -> SPIR-V at runtime. When omitted, modules are loaded from ``spec.vert_spirv`` / ``spec.frag_spirv`` and likewise returned via the pipeline (the caller still owns lifetime). cffi lifetime: every ``ffi.new`` sub-struct is appended to a local *keep* list that stays reachable for the whole function body, so it is alive across the ``vkCreatePipelineLayout`` and ``vkCreateGraphicsPipelines`` calls (Vulkan reads ``pCreateInfos`` only during those calls). No caller ever has to root a sub-struct by hand. *keep* only becomes collectable after this function returns, strictly after the create calls have returned. """ ffi = vk.ffi keep: list[Any] = [] def hold(*objs: Any) -> Any: """Root every allocation in *keep*, return the first (the wired parent).""" keep.extend(objs) return objs[0] # Shader modules: use the supplied handles, else load from the spec paths. vmod = vert_module fmod = frag_module if vmod is None: if spec.vert_spirv is None: raise ValueError(f"build_pipeline({spec.name}): no vert_module and no spec.vert_spirv") vmod = create_shader_module(device, spec.vert_spirv) if fmod is None: if spec.frag_spirv is None: raise ValueError(f"build_pipeline({spec.name}): no frag_module and no spec.frag_spirv") fmod = create_shader_module(device, spec.frag_spirv) # Pipeline layout (its sub-allocations live in keep until the create call # inside _create_pipeline_layout returns, which is correct: that function # scopes them to its own frame). layout = _create_pipeline_layout( ffi, device, list(spec.set_layouts) or None, push_constant_size=spec.push_size, push_stage_flags=spec.push_stages, ) pi = ffi.new("VkGraphicsPipelineCreateInfo*") keep.append(pi) pi.sType = vk.VK_STRUCTURE_TYPE_GRAPHICS_PIPELINE_CREATE_INFO stages = hold(*_make_shader_stages(ffi, vmod, fmod)) pi.stageCount = 2 pi.pStages = stages bindings = spec.vertex_input_bindings if not bindings: vi = hold(_make_empty_vertex_input(ffi)) else: vi = hold(*_make_vertex_input(ffi, bindings)) pi.pVertexInputState = vi pi.pInputAssemblyState = hold(_make_input_assembly(ffi, topology=spec.topology)) pi.pViewportState = hold(*_make_viewport_state(ffi, extent)) pi.pRasterizationState = hold( _make_rasterization( ffi, cull_mode=spec.cull_mode, front_face=spec.front_face, depth_bias=spec.depth_bias, ) ) pi.pMultisampleState = hold(_make_multisample(ffi)) pi.pDepthStencilState = hold( _make_depth_stencil( ffi, test=spec.depth_test, write=spec.depth_write, compare_op=spec.depth_compare, ) ) n_att = spec.attachment_count cba = None if spec.blend == "opaque": cb_pair = _make_colour_blend_opaque(ffi, write_mask=spec.colour_write_mask, attachment_count=n_att) elif spec.blend == "alpha": cb_pair = _make_colour_blend_alpha(ffi, dst_alpha_factor=spec.dst_alpha_factor, attachment_count=n_att) elif spec.blend == "add": cb_pair = _make_colour_blend_add(ffi, attachment_count=n_att) elif spec.blend == "multiply": cb_pair = _make_colour_blend_multiply(ffi, attachment_count=n_att) elif spec.blend == "none": cb_pair = _make_colour_blend_none(ffi) else: # pragma: no cover - dataclass typing guards this raise ValueError(f"build_pipeline({spec.name}): unknown blend {spec.blend!r}") cb = hold(*cb_pair) if len(cb_pair) > 1: cba = cb_pair[1] # Thin G-buffer: in a multi-attachment pass every pipeline must declare # as many blend attachments as the render pass has, but only the lit uber # writes the normal target. Pipelines that do not (skybox/grid/particle/ # billboard/tilemap/transparent/2D) mask attachment 1+ to zero writes so # they leave the G-buffer untouched. Attachment 0 keeps its blend as built. if n_att >= 2 and not spec.writes_gbuffer and cba is not None: for i in range(1, n_att): cba[i].colorWriteMask = 0 pi.pColorBlendState = cb pi.pDynamicState = hold(*_make_dynamic_state(ffi)) pi.layout = layout pi.renderPass = render_pass pipeline = _build_pipeline(ffi, device, pi, spec.name) # `keep` stays reachable for the whole body and only becomes collectable # when this function returns -- strictly after vkCreateGraphicsPipelines # has returned, so every sub-struct was live across the create call. return pipeline, layout