"""Buffer and image allocation helpers."""
import logging
from typing import Any
import numpy as np
import vulkan as vk
log = logging.getLogger(__name__)
__all__ = [
"create_buffer",
"create_image",
"create_compressed_image",
"create_sampler",
"format_blit_supported",
"format_sampled_supported",
"mip_chain_length",
"transition_image_layout",
"upload_numpy",
"upload_image_data",
"upload_compressed_image",
"update_image_data",
"create_indirect_buffer",
"begin_single_time_commands",
"end_single_time_commands",
]
def _find_memory_type(physical_device: Any, type_filter: int, properties: int) -> int:
mem_props = vk.vkGetPhysicalDeviceMemoryProperties(physical_device)
for i in range(mem_props.memoryTypeCount):
if (type_filter & (1 << i)) and (mem_props.memoryTypes[i].propertyFlags & properties) == properties:
return i
raise RuntimeError("Failed to find suitable memory type")
[docs]
def create_buffer(
device: Any,
physical_device: Any,
size: int,
usage: int,
memory_flags: int,
*,
concurrent_families: list[int] | None = None,
) -> tuple[Any, Any]:
"""Create a VkBuffer with bound memory. Returns (buffer, memory).
``concurrent_families`` opts a buffer into ``VK_SHARING_MODE_CONCURRENT``:
pass the distinct queue-family indices that will touch it (e.g.
``[graphics_qf, compute_qf]``) so the async-compute scheduler can read/write
it from a separate compute queue without an explicit ownership transfer. The
list must have >= 2 *distinct* families to be meaningful: a single family (or
duplicates collapsing to one) falls back to ``EXCLUSIVE`` because CONCURRENT
with one family is invalid (VUID-VkBufferCreateInfo-sharingMode-00914). The
default (``None``) keeps the historical ``EXCLUSIVE`` path byte-identical, so
callers that do not opt in are unaffected.
"""
families = sorted(set(concurrent_families)) if concurrent_families else []
if len(families) >= 2:
buf_info = vk.VkBufferCreateInfo(
size=size,
usage=usage,
sharingMode=vk.VK_SHARING_MODE_CONCURRENT,
queueFamilyIndexCount=len(families),
pQueueFamilyIndices=families,
)
else:
buf_info = vk.VkBufferCreateInfo(
size=size,
usage=usage,
sharingMode=vk.VK_SHARING_MODE_EXCLUSIVE,
)
buffer = vk.vkCreateBuffer(device, buf_info, None)
mem_reqs = vk.vkGetBufferMemoryRequirements(device, buffer)
alloc_info = vk.VkMemoryAllocateInfo(
allocationSize=mem_reqs.size,
memoryTypeIndex=_find_memory_type(physical_device, mem_reqs.memoryTypeBits, memory_flags),
)
memory = vk.vkAllocateMemory(device, alloc_info, None)
vk.vkBindBufferMemory(device, buffer, memory, 0)
return buffer, memory
[docs]
def create_image(
device: Any,
physical_device: Any,
width: int,
height: int,
fmt: int,
usage: int,
*,
mip_levels: int = 1,
) -> tuple[Any, Any]:
"""Create a VkImage with bound memory. Returns (image, memory).
``mip_levels`` sizes the image's mip chain; the default of 1 keeps every
existing single-mip caller byte-identical.
"""
img_info = vk.VkImageCreateInfo(
imageType=vk.VK_IMAGE_TYPE_2D,
format=fmt,
extent=vk.VkExtent3D(width=width, height=height, depth=1),
mipLevels=mip_levels,
arrayLayers=1,
samples=vk.VK_SAMPLE_COUNT_1_BIT,
tiling=vk.VK_IMAGE_TILING_OPTIMAL,
usage=usage,
sharingMode=vk.VK_SHARING_MODE_EXCLUSIVE,
initialLayout=vk.VK_IMAGE_LAYOUT_UNDEFINED,
)
image = vk.vkCreateImage(device, img_info, None)
mem_reqs = vk.vkGetImageMemoryRequirements(device, image)
alloc_info = vk.VkMemoryAllocateInfo(
allocationSize=mem_reqs.size,
memoryTypeIndex=_find_memory_type(
physical_device,
mem_reqs.memoryTypeBits,
vk.VK_MEMORY_PROPERTY_DEVICE_LOCAL_BIT,
),
)
memory = vk.vkAllocateMemory(device, alloc_info, None)
vk.vkBindImageMemory(device, image, memory, 0)
return image, memory
[docs]
def create_compressed_image(
device: Any,
physical_device: Any,
width: int,
height: int,
fmt: int,
mip_count: int,
) -> tuple[Any, Any]:
"""Create a multi-mip block-compressed VkImage with bound memory.
Mirrors :func:`create_image` but takes a compressed ``VkFormat`` and a mip
count. Usage is TRANSFER_DST | SAMPLED. Returns (image, memory).
"""
img_info = vk.VkImageCreateInfo(
imageType=vk.VK_IMAGE_TYPE_2D,
format=fmt,
extent=vk.VkExtent3D(width=width, height=height, depth=1),
mipLevels=mip_count,
arrayLayers=1,
samples=vk.VK_SAMPLE_COUNT_1_BIT,
tiling=vk.VK_IMAGE_TILING_OPTIMAL,
usage=vk.VK_IMAGE_USAGE_TRANSFER_DST_BIT | vk.VK_IMAGE_USAGE_SAMPLED_BIT,
sharingMode=vk.VK_SHARING_MODE_EXCLUSIVE,
initialLayout=vk.VK_IMAGE_LAYOUT_UNDEFINED,
)
image = vk.vkCreateImage(device, img_info, None)
mem_reqs = vk.vkGetImageMemoryRequirements(device, image)
alloc_info = vk.VkMemoryAllocateInfo(
allocationSize=mem_reqs.size,
memoryTypeIndex=_find_memory_type(
physical_device,
mem_reqs.memoryTypeBits,
vk.VK_MEMORY_PROPERTY_DEVICE_LOCAL_BIT,
),
)
memory = vk.vkAllocateMemory(device, alloc_info, None)
vk.vkBindImageMemory(device, image, memory, 0)
return image, memory
[docs]
def mip_chain_length(width: int, height: int) -> int:
"""Number of levels in a full mip chain for a ``width`` x ``height`` image.
Level 0 included: a 1x1 image has a chain of 1, and each level halves the
larger dimension (``floor(log2(max)) + 1``, i.e. ``max.bit_length()``).
"""
return max(int(width), int(height), 1).bit_length()
[docs]
def create_sampler(
device: Any,
filter_mode: int = vk.VK_FILTER_LINEAR,
max_lod: float = 0.0,
address_mode: int = vk.VK_SAMPLER_ADDRESS_MODE_REPEAT,
) -> Any:
"""Create a VkSampler with configurable filtering and addressing.
``address_mode`` applies to all three axes. REPEAT is the correct default for
tiled mesh textures; screen-space post-process passes must pass CLAMP_TO_EDGE so
neighbour-texel taps at the frame edge do not wrap to the opposite edge.
"""
sampler_info = vk.VkSamplerCreateInfo(
magFilter=filter_mode,
minFilter=filter_mode,
mipmapMode=vk.VK_SAMPLER_MIPMAP_MODE_LINEAR,
addressModeU=address_mode,
addressModeV=address_mode,
addressModeW=address_mode,
mipLodBias=0.0,
anisotropyEnable=False,
maxAnisotropy=1.0,
minLod=0.0,
maxLod=max_lod,
)
return vk.vkCreateSampler(device, sampler_info, None)
[docs]
def transition_image_layout(
device: Any,
queue: Any,
cmd_pool: Any,
image: Any,
old_layout: int,
new_layout: int,
aspect_mask: int = vk.VK_IMAGE_ASPECT_COLOR_BIT,
level_count: int = 1,
) -> None:
"""Transition image layout via one-time command buffer.
``level_count`` transitions that many mip levels from ``baseMipLevel=0``;
the default of 1 keeps the single-mip uncompressed path byte-identical.
"""
alloc_info = vk.VkCommandBufferAllocateInfo(
commandPool=cmd_pool,
level=vk.VK_COMMAND_BUFFER_LEVEL_PRIMARY,
commandBufferCount=1,
)
cmd = vk.vkAllocateCommandBuffers(device, alloc_info)[0]
vk.vkBeginCommandBuffer(
cmd,
vk.VkCommandBufferBeginInfo(
flags=vk.VK_COMMAND_BUFFER_USAGE_ONE_TIME_SUBMIT_BIT,
),
)
# Determine access masks and stages from layouts
src_access = 0
src_stage = vk.VK_PIPELINE_STAGE_TOP_OF_PIPE_BIT
dst_access = 0
dst_stage = vk.VK_PIPELINE_STAGE_BOTTOM_OF_PIPE_BIT
if old_layout == vk.VK_IMAGE_LAYOUT_COLOR_ATTACHMENT_OPTIMAL:
src_access = vk.VK_ACCESS_COLOR_ATTACHMENT_WRITE_BIT
src_stage = vk.VK_PIPELINE_STAGE_COLOR_ATTACHMENT_OUTPUT_BIT
elif old_layout == vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL:
src_access = vk.VK_ACCESS_TRANSFER_WRITE_BIT
src_stage = vk.VK_PIPELINE_STAGE_TRANSFER_BIT
elif old_layout == vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL:
src_access = vk.VK_ACCESS_SHADER_READ_BIT
src_stage = vk.VK_PIPELINE_STAGE_FRAGMENT_SHADER_BIT
if new_layout == vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL:
dst_access = vk.VK_ACCESS_SHADER_READ_BIT
dst_stage = vk.VK_PIPELINE_STAGE_FRAGMENT_SHADER_BIT
elif new_layout == vk.VK_IMAGE_LAYOUT_COLOR_ATTACHMENT_OPTIMAL:
dst_access = vk.VK_ACCESS_COLOR_ATTACHMENT_WRITE_BIT
dst_stage = vk.VK_PIPELINE_STAGE_COLOR_ATTACHMENT_OUTPUT_BIT
elif new_layout == vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL:
dst_access = vk.VK_ACCESS_TRANSFER_WRITE_BIT
dst_stage = vk.VK_PIPELINE_STAGE_TRANSFER_BIT
barrier = vk.VkImageMemoryBarrier(
srcAccessMask=src_access,
dstAccessMask=dst_access,
oldLayout=old_layout,
newLayout=new_layout,
srcQueueFamilyIndex=vk.VK_QUEUE_FAMILY_IGNORED,
dstQueueFamilyIndex=vk.VK_QUEUE_FAMILY_IGNORED,
image=image,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=aspect_mask,
baseMipLevel=0,
levelCount=level_count,
baseArrayLayer=0,
layerCount=1,
),
)
vk.vkCmdPipelineBarrier(cmd, src_stage, dst_stage, 0, 0, None, 0, None, 1, [barrier])
vk.vkEndCommandBuffer(cmd)
submit = vk.VkSubmitInfo(commandBufferCount=1, pCommandBuffers=[cmd])
vk.vkQueueSubmit(queue, 1, [submit], None)
vk.vkQueueWaitIdle(queue)
vk.vkFreeCommandBuffers(device, cmd_pool, 1, [cmd])
[docs]
def upload_numpy(device: Any, memory: Any, data: np.ndarray, byte_offset: int = 0) -> None:
"""Map device memory and copy a numpy array into it at ``byte_offset``."""
size = data.nbytes
if size == 0:
return
src = vk.ffi.cast("void*", data.ctypes.data)
dst = vk.vkMapMemory(device, memory, byte_offset, size, 0)
vk.ffi.memmove(dst, src, size)
vk.vkUnmapMemory(device, memory)
[docs]
def begin_single_time_commands(device: Any, cmd_pool: Any) -> Any:
"""Allocate and begin a one-time command buffer."""
alloc_info = vk.VkCommandBufferAllocateInfo(
commandPool=cmd_pool,
level=vk.VK_COMMAND_BUFFER_LEVEL_PRIMARY,
commandBufferCount=1,
)
cmd = vk.vkAllocateCommandBuffers(device, alloc_info)[0]
vk.vkBeginCommandBuffer(
cmd,
vk.VkCommandBufferBeginInfo(
flags=vk.VK_COMMAND_BUFFER_USAGE_ONE_TIME_SUBMIT_BIT,
),
)
return cmd
[docs]
def end_single_time_commands(device: Any, queue: Any, cmd_pool: Any, cmd: Any) -> None:
"""End, submit, wait, and free a one-time command buffer."""
vk.vkEndCommandBuffer(cmd)
submit = vk.VkSubmitInfo(commandBufferCount=1, pCommandBuffers=[cmd])
vk.vkQueueSubmit(queue, 1, [submit], None)
vk.vkQueueWaitIdle(queue)
vk.vkFreeCommandBuffers(device, cmd_pool, 1, [cmd])
def _mip_level_barrier(
cmd: Any,
image: Any,
level: int,
old_layout: int,
new_layout: int,
src_access: int,
dst_access: int,
src_stage: int,
dst_stage: int,
) -> None:
"""Record a single-mip-level layout barrier into an open command buffer."""
barrier = vk.VkImageMemoryBarrier(
srcAccessMask=src_access,
dstAccessMask=dst_access,
oldLayout=old_layout,
newLayout=new_layout,
srcQueueFamilyIndex=vk.VK_QUEUE_FAMILY_IGNORED,
dstQueueFamilyIndex=vk.VK_QUEUE_FAMILY_IGNORED,
image=image,
subresourceRange=vk.VkImageSubresourceRange(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
baseMipLevel=level,
levelCount=1,
baseArrayLayer=0,
layerCount=1,
),
)
vk.vkCmdPipelineBarrier(cmd, src_stage, dst_stage, 0, 0, None, 0, None, 1, [barrier])
def _record_mip_chain(cmd: Any, image: Any, width: int, height: int, mip_count: int) -> None:
"""Record a ``vkCmdBlitImage`` downsample chain for levels 1..mip_count-1.
On entry every level is TRANSFER_DST with level 0 already filled; on exit
every level is SHADER_READ_ONLY. Each source level flips to TRANSFER_SRC
for its blit (linear filter), then to SHADER_READ_ONLY once consumed.
"""
transfer = vk.VK_PIPELINE_STAGE_TRANSFER_BIT
fragment = vk.VK_PIPELINE_STAGE_FRAGMENT_SHADER_BIT
src_w, src_h = width, height
for level in range(1, mip_count):
_mip_level_barrier(
cmd, image, level - 1,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL, vk.VK_IMAGE_LAYOUT_TRANSFER_SRC_OPTIMAL,
vk.VK_ACCESS_TRANSFER_WRITE_BIT, vk.VK_ACCESS_TRANSFER_READ_BIT,
transfer, transfer,
)
dst_w, dst_h = max(1, src_w // 2), max(1, src_h // 2)
blit = vk.VkImageBlit(
srcSubresource=vk.VkImageSubresourceLayers(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT, mipLevel=level - 1, baseArrayLayer=0, layerCount=1,
),
srcOffsets=[vk.VkOffset3D(x=0, y=0, z=0), vk.VkOffset3D(x=src_w, y=src_h, z=1)],
dstSubresource=vk.VkImageSubresourceLayers(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT, mipLevel=level, baseArrayLayer=0, layerCount=1,
),
dstOffsets=[vk.VkOffset3D(x=0, y=0, z=0), vk.VkOffset3D(x=dst_w, y=dst_h, z=1)],
)
vk.vkCmdBlitImage(
cmd,
image, vk.VK_IMAGE_LAYOUT_TRANSFER_SRC_OPTIMAL,
image, vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
1, [blit], vk.VK_FILTER_LINEAR,
)
_mip_level_barrier(
cmd, image, level - 1,
vk.VK_IMAGE_LAYOUT_TRANSFER_SRC_OPTIMAL, vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
vk.VK_ACCESS_TRANSFER_READ_BIT, vk.VK_ACCESS_SHADER_READ_BIT,
transfer, fragment,
)
src_w, src_h = dst_w, dst_h
_mip_level_barrier(
cmd, image, mip_count - 1,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL, vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
vk.VK_ACCESS_TRANSFER_WRITE_BIT, vk.VK_ACCESS_SHADER_READ_BIT,
transfer, fragment,
)
[docs]
def upload_image_data(
device: Any,
physical_device: Any,
queue: Any,
cmd_pool: Any,
pixels: np.ndarray,
width: int,
height: int,
fmt: int = vk.VK_FORMAT_R8G8B8A8_UNORM,
*,
mip_count: int = 1,
) -> tuple[Any, Any]:
"""Upload pixel data to a device-local image via staging buffer.
``mip_count`` > 1 allocates the full chain (plus TRANSFER_SRC usage),
uploads level 0, and generates the remaining levels on the GPU with a
linear-filtered ``vkCmdBlitImage`` chain. The caller is responsible for
checking :func:`format_blit_supported` first. The default of 1 keeps the
historical single-mip path byte-identical.
Args:
pixels: Contiguous RGBA uint8 array, shape (height, width, 4).
Returns: (image, memory)
"""
ffi = vk.ffi
pixel_size = pixels.nbytes
# Staging buffer
staging_buf, staging_mem = create_buffer(
device,
physical_device,
pixel_size,
vk.VK_BUFFER_USAGE_TRANSFER_SRC_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
dst = vk.vkMapMemory(device, staging_mem, 0, pixel_size, 0)
ffi.memmove(dst, ffi.cast("void*", pixels.ctypes.data), pixel_size)
vk.vkUnmapMemory(device, staging_mem)
# Device-local image (mipgen blits read back from the image, hence TRANSFER_SRC)
usage = vk.VK_IMAGE_USAGE_TRANSFER_DST_BIT | vk.VK_IMAGE_USAGE_SAMPLED_BIT
if mip_count > 1:
usage |= vk.VK_IMAGE_USAGE_TRANSFER_SRC_BIT
image, image_mem = create_image(
device,
physical_device,
width,
height,
fmt,
usage,
mip_levels=mip_count,
)
# Transition UNDEFINED → TRANSFER_DST (all levels)
transition_image_layout(
device, queue, cmd_pool, image, vk.VK_IMAGE_LAYOUT_UNDEFINED, vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
level_count=mip_count,
)
# Copy staging buffer → image level 0, then blit the chain down
cmd = begin_single_time_commands(device, cmd_pool)
region = vk.VkBufferImageCopy(
bufferOffset=0,
bufferRowLength=0,
bufferImageHeight=0,
imageSubresource=vk.VkImageSubresourceLayers(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
mipLevel=0,
baseArrayLayer=0,
layerCount=1,
),
imageOffset=vk.VkOffset3D(x=0, y=0, z=0),
imageExtent=vk.VkExtent3D(width=width, height=height, depth=1),
)
vk.vkCmdCopyBufferToImage(
cmd,
staging_buf,
image,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
1,
[region],
)
if mip_count > 1:
# Per-level barriers + blits; leaves every level SHADER_READ_ONLY.
_record_mip_chain(cmd, image, width, height, mip_count)
end_single_time_commands(device, queue, cmd_pool, cmd)
if mip_count == 1:
# Transition TRANSFER_DST → SHADER_READ_ONLY
transition_image_layout(
device,
queue,
cmd_pool,
image,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
)
# Cleanup staging
vk.vkDestroyBuffer(device, staging_buf, None)
vk.vkFreeMemory(device, staging_mem, None)
return image, image_mem
[docs]
def upload_compressed_image(
device: Any,
physical_device: Any,
queue: Any,
cmd_pool: Any,
mip_block_bytes: list[bytes],
width: int,
height: int,
fmt: int,
block_size: int,
) -> tuple[Any, Any]:
"""Upload block-compressed mip data to a device-local image.
Concatenates all mip levels into ONE staging buffer at block-aligned
offsets (tight concatenation keeps every offset a multiple of
``block_size``), then issues one VkBufferImageCopy per mip. Block-row math
follows the Vulkan spec: ``bufferRowLength``/``bufferImageHeight`` are in
TEXELS rounded up to whole 4x4 blocks; ``imageExtent`` is the true texel
size of the mip (Vulkan derives the block count internally).
Args:
mip_block_bytes: tightly-packed block bytes per mip, level 0 first.
block_size: 8 (BC1/BC4) or 16 (BC2/BC3/BC5/BC6H/BC7).
Returns: (image, memory).
"""
ffi = vk.ffi
mip_count = len(mip_block_bytes)
staging = b"".join(mip_block_bytes)
total = len(staging)
staging_buf, staging_mem = create_buffer(
device,
physical_device,
total,
vk.VK_BUFFER_USAGE_TRANSFER_SRC_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
dst = vk.vkMapMemory(device, staging_mem, 0, total, 0)
ffi.memmove(dst, ffi.from_buffer(staging), total)
vk.vkUnmapMemory(device, staging_mem)
image, image_mem = create_compressed_image(device, physical_device, width, height, fmt, mip_count)
transition_image_layout(
device, queue, cmd_pool, image,
vk.VK_IMAGE_LAYOUT_UNDEFINED, vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
level_count=mip_count,
)
cmd = begin_single_time_commands(device, cmd_pool)
regions = []
running_offset = 0
for i in range(mip_count):
mw = max(1, width >> i)
mh = max(1, height >> i)
blocks_w = max(1, (mw + 3) // 4)
blocks_h = max(1, (mh + 3) // 4)
regions.append(vk.VkBufferImageCopy(
bufferOffset=running_offset,
bufferRowLength=blocks_w * 4, # texels, multiple of the 4px block width
bufferImageHeight=blocks_h * 4, # texels, multiple of the 4px block height
imageSubresource=vk.VkImageSubresourceLayers(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
mipLevel=i, baseArrayLayer=0, layerCount=1,
),
imageOffset=vk.VkOffset3D(x=0, y=0, z=0),
imageExtent=vk.VkExtent3D(width=mw, height=mh, depth=1),
))
running_offset += blocks_w * blocks_h * block_size
vk.vkCmdCopyBufferToImage(
cmd, staging_buf, image, vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL, mip_count, regions,
)
end_single_time_commands(device, queue, cmd_pool, cmd)
transition_image_layout(
device, queue, cmd_pool, image,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL, vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
level_count=mip_count,
)
vk.vkDestroyBuffer(device, staging_buf, None)
vk.vkFreeMemory(device, staging_mem, None)
return image, image_mem
[docs]
def update_image_data(
device: Any,
physical_device: Any,
queue: Any,
cmd_pool: Any,
image: Any,
pixels: np.ndarray,
width: int,
height: int,
) -> None:
"""Re-upload pixel data to an existing VkImage (same dimensions).
Transitions the image from SHADER_READ_ONLY → TRANSFER_DST, copies
the new pixel data via a staging buffer, then transitions back to
SHADER_READ_ONLY.
"""
ffi = vk.ffi
pixel_size = pixels.nbytes
# Staging buffer
staging_buf, staging_mem = create_buffer(
device, physical_device, pixel_size,
vk.VK_BUFFER_USAGE_TRANSFER_SRC_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
)
dst = vk.vkMapMemory(device, staging_mem, 0, pixel_size, 0)
ffi.memmove(dst, ffi.cast("void*", pixels.ctypes.data), pixel_size)
vk.vkUnmapMemory(device, staging_mem)
# SHADER_READ_ONLY → TRANSFER_DST
transition_image_layout(
device, queue, cmd_pool, image,
vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
)
# Copy staging → image
cmd = begin_single_time_commands(device, cmd_pool)
region = vk.VkBufferImageCopy(
bufferOffset=0, bufferRowLength=0, bufferImageHeight=0,
imageSubresource=vk.VkImageSubresourceLayers(
aspectMask=vk.VK_IMAGE_ASPECT_COLOR_BIT,
mipLevel=0, baseArrayLayer=0, layerCount=1,
),
imageOffset=vk.VkOffset3D(x=0, y=0, z=0),
imageExtent=vk.VkExtent3D(width=width, height=height, depth=1),
)
vk.vkCmdCopyBufferToImage(cmd, staging_buf, image, vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL, 1, [region])
end_single_time_commands(device, queue, cmd_pool, cmd)
# TRANSFER_DST → SHADER_READ_ONLY
transition_image_layout(
device, queue, cmd_pool, image,
vk.VK_IMAGE_LAYOUT_TRANSFER_DST_OPTIMAL,
vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
)
# Cleanup staging
vk.vkDestroyBuffer(device, staging_buf, None)
vk.vkFreeMemory(device, staging_mem, None)
[docs]
def create_indirect_buffer(
device: Any,
physical_device: Any,
draw_count: int,
*,
concurrent_families: list[int] | None = None,
) -> tuple[Any, Any]:
"""Create a host-visible buffer for VkDrawIndexedIndirectCommand array.
STORAGE_BUFFER usage is included so the GPU occlusion-cull compute (phase O3)
can patch ``instance_count`` in place. It stays HOST_VISIBLE|HOST_COHERENT so
the batch upload (host write) and the telemetry readback both map it directly.
``concurrent_families`` forwards to :func:`create_buffer`: when the
occlusion compute that writes ``instance_count`` runs on a *dedicated*
compute queue (async-compute path), the indirect buffer must be visible to
both that queue and the graphics queue that consumes it via
``vkCmdDrawIndexedIndirect``. ``None`` (single-queue path) keeps EXCLUSIVE.
"""
buffer_size = draw_count * 20 # 5 x uint32 = 20 bytes per command
return create_buffer(
device,
physical_device,
buffer_size,
vk.VK_BUFFER_USAGE_INDIRECT_BUFFER_BIT | vk.VK_BUFFER_USAGE_STORAGE_BUFFER_BIT,
vk.VK_MEMORY_PROPERTY_HOST_VISIBLE_BIT | vk.VK_MEMORY_PROPERTY_HOST_COHERENT_BIT,
concurrent_families=concurrent_families,
)