"""GPU material SSBO slot allocation, deduplication, and reclamation.
Extracted from :class:`~simvx.graphics.scene_adapter.SceneAdapter` so the
material subsystem (the registry/dedup dicts, the SSBO numpy array, the
free-list, and the weakref finalizers) lives on its own cohesive object
rather than tangled into the scene-tree traversal adapter.
"""
import logging
import weakref
from collections.abc import Callable
from typing import Any
import numpy as np
from simvx.core import Material
from simvx.core.colour import srgb_to_linear_rgb
from .types import ALPHA_BLEND, ALPHA_OPAQUE, MATERIAL_DTYPE, Feature
__all__ = ["MaterialSlotManager"]
log = logging.getLogger(__name__)
def _release_material_slot(manager_ref: Callable[[], Any], mat_id: int, slot_idx: int) -> None:
"""Weakref-finalize callback that delegates to the manager's release.
Top-level (not a method) so it doesn't itself keep a strong reference to
the manager: the finalize machinery holds the manager by weakref, and this
function's closure keeps nothing alive.
"""
manager = manager_ref()
if manager is None:
return
try:
manager.release(mat_id, slot_idx)
except Exception:
log.exception("release_material failed during finalize")
[docs]
class MaterialSlotManager:
"""Allocates, deduplicates, and reclaims rows of the bindless material SSBO.
Converts ``simvx.core`` :class:`Material` objects into integer indices into
a flat numpy SSBO array, deduplicating by content so two materials with
identical rendering properties share one slot. Slots are reclaimed via a
``weakref.finalize`` attached to each owning Material, keeping long-running
sessions (one Material per frame) from walking the SSBO off the end.
Texture loading is delegated through the ``load_texture`` callable supplied
at construction (the adapter owns the bindless texture manager); this class
owns only material-slot bookkeeping.
"""
def __init__(self, renderer: Any, load_texture: Callable[..., int]):
self._renderer = renderer
self._load_texture = load_texture
self._registry: dict[int, int] = {} # material id -> material index
self._dedup: dict[tuple, int] = {} # content_key -> material index
self._array = np.zeros(256, dtype=MATERIAL_DTYPE)
self._next_id = 0
# Free-list of reclaimed slots. Populated by release() (typically from a
# weakref.finalize on the owning Material). Allocation always prefers
# reclaimed slots so long-running demos that create one Material per
# frame don't walk the SSBO off the end.
self._free_slots: list[int] = []
# Reverse map (slot_idx -> content_key) so a finalizer can evict the
# dedup entry without needing the original Material object.
self._slot_to_content_key: dict[int, tuple] = {}
# Weakref finalizers keyed by mat_id so repeated register() calls for
# the same id don't leak finalizers.
self._finalizers: dict[int, Any] = {}
self._default_mat: Material | None = None
def _default_material(self) -> Material:
"""Return the shared default material (opaque mid-grey PBR).
Materials submitted with no explicit material still need a real entry in
the material SSBO, otherwise the shader reads uninitialised memory and
the mesh renders incorrectly (or is skipped entirely when upload() sees
``_next_id == 0``).
"""
if self._default_mat is None:
self._default_mat = Material(colour=(0.8, 0.8, 0.8, 1.0))
return self._default_mat
[docs]
def register(self, material: Material | None) -> int:
"""Convert a core Material to a Vulkan material index.
Deduplicates materials by content: two Material objects with identical
rendering properties share the same SSBO slot.
"""
if material is None:
return self.register(self._default_material())
# First-class ``Material(albedo=subviewport)`` (design §5.4): resolve the
# SubViewport's live bindless slot into ``albedo_tex_index`` so the existing
# direct-index path (initial register + the per-frame refresh below) samples
# the viewport's current feed. The slot is -1 until the SubViewportManager
# has rendered the viewport once; the per-frame refresh branch lands it then.
svp = getattr(material, "_subviewport_albedo", None)
if svp is not None:
material.albedo_tex_index = int(svp.texture)
mat_id = getattr(material, "_uid", id(material))
if mat_id in self._registry:
# Update in case colour/properties changed at runtime
idx = self._registry[mat_id]
colour = material.colour if hasattr(material, "colour") else (1, 1, 1, 1)
self._array[idx]["albedo"] = srgb_to_linear_rgb(colour)
self._array[idx]["metallic"] = getattr(material, "metallic", 0.0)
self._array[idx]["roughness"] = getattr(material, "roughness", 0.5)
# Refresh a runtime-assigned *direct* GPU texture index (e.g. a
# SubViewport feed whose bindless slot is published after the
# material was first registered). Only the direct-index path is
# cheap enough for the per-frame update branch: URI loads stay in
# the full-register path. Without this, a material that gains an
# albedo_tex_index after first registration never samples it.
direct_albedo = getattr(material, "albedo_tex_index", -1)
if direct_albedo >= 0 and self._array[idx]["albedo_tex"] != direct_albedo:
self._array[idx]["albedo_tex"] = direct_albedo
self._array[idx]["features"] = int(Feature(int(self._array[idx]["features"])) | Feature.HAS_ALBEDO)
# Update alpha mode on existing materials
blend = getattr(material, "blend", "opaque")
if blend in ("alpha", "additive"):
self._array[idx]["alpha_mode"] = ALPHA_BLEND
elif colour[3] < 1.0:
self._array[idx]["alpha_mode"] = ALPHA_BLEND
else:
self._array[idx]["alpha_mode"] = ALPHA_OPAQUE
return idx
# Content-based deduplication: reuse slot if identical material already registered
content_key = material.content_key if hasattr(material, "content_key") else None
if content_key is not None and content_key in self._dedup:
idx = self._dedup[content_key]
self._registry[mat_id] = idx
return idx
# Prefer a reclaimed slot over bumping the monotonic counter so
# long-running sessions that create Materials ad-hoc don't drain the
# SSBO. New allocations grow the array on demand when the free list
# is empty.
if self._free_slots:
idx = self._free_slots.pop()
else:
idx = self._next_id
if idx >= len(self._array):
new_array = np.zeros(idx + 128, dtype=MATERIAL_DTYPE)
new_array[: len(self._array)] = self._array
self._array = new_array
self._next_id += 1
# Convert Material.colour (RGBA tuple) to Vulkan format. The albedo
# constant is a COLOUR: decode sRGB->linear so it matches sRGB-decoded
# albedo textures and the shader's linear-light maths. Alpha stays linear.
colour = material.colour if hasattr(material, "colour") else (1, 1, 1, 1)
self._array[idx]["albedo"] = srgb_to_linear_rgb(colour)
self._array[idx]["metallic"] = getattr(material, "metallic", 0.0)
self._array[idx]["roughness"] = getattr(material, "roughness", 0.5)
# Load textures if URIs or direct indices are provided
features = Feature.NONE
albedo_tex = -1
normal_tex = -1
mr_tex = -1
emissive_tex = -1
ao_tex = -1
# Direct GPU texture index (bypasses URI loading). ``albedo_uri`` may
# hold any ``TextureManager.resolve`` source (str / bytes / ndarray);
# use ``is not None`` rather than truthiness so ndarrays don't trip
# "truth value is ambiguous".
if getattr(material, "albedo_tex_index", -1) >= 0:
albedo_tex = material.albedo_tex_index
features |= Feature.HAS_ALBEDO
elif getattr(material, "albedo_uri", None) is not None:
albedo_tex = self._load_texture(material.albedo_uri, colour_space="srgb")
if albedo_tex >= 0:
features |= Feature.HAS_ALBEDO
if getattr(material, "normal_uri", None) is not None:
normal_tex = self._load_texture(material.normal_uri, colour_space="linear")
if normal_tex >= 0:
features |= Feature.HAS_NORMAL
if getattr(material, "metallic_roughness_uri", None) is not None:
mr_tex = self._load_texture(material.metallic_roughness_uri, colour_space="linear")
if mr_tex >= 0:
features |= Feature.HAS_METALLIC_ROUGHNESS
if getattr(material, "emissive_uri", None) is not None:
emissive_tex = self._load_texture(material.emissive_uri, colour_space="srgb")
if emissive_tex >= 0:
features |= Feature.HAS_EMISSIVE
if getattr(material, "ao_uri", None) is not None:
ao_tex = self._load_texture(material.ao_uri, colour_space="linear")
if ao_tex >= 0:
features |= Feature.HAS_AO
# Emissive colour. The rgb is a COLOUR (decode sRGB->linear); the 4th
# component is emissive *intensity* (a scalar multiplier, not a colour)
# and must pass through untouched.
ec = getattr(material, "emissive_colour", None)
if ec is not None:
ec4 = ec[:4] if len(ec) >= 4 else (*ec[:3], 1.0)
self._array[idx]["emissive_colour"] = srgb_to_linear_rgb(ec4)
features |= Feature.HAS_EMISSIVE_COLOR
self._array[idx]["albedo_tex"] = albedo_tex
self._array[idx]["normal_tex"] = normal_tex
self._array[idx]["metallic_roughness_tex"] = mr_tex
self._array[idx]["emissive_tex"] = emissive_tex
self._array[idx]["ao_tex"] = ao_tex
self._array[idx]["features"] = int(features)
# Alpha mode from Material.blend
blend = getattr(material, "blend", "opaque")
if blend == "alpha":
self._array[idx]["alpha_mode"] = ALPHA_BLEND
self._array[idx]["alpha_cutoff"] = 0.0
elif blend == "additive":
# Additive blending also uses the transparent pipeline
self._array[idx]["alpha_mode"] = ALPHA_BLEND
self._array[idx]["alpha_cutoff"] = 0.0
else:
# Check if the albedo alpha implies transparency
if colour[3] < 1.0:
self._array[idx]["alpha_mode"] = ALPHA_BLEND
else:
self._array[idx]["alpha_mode"] = ALPHA_OPAQUE
self._array[idx]["alpha_cutoff"] = 0.5
# Double-sided flag (disables backface culling)
self._array[idx]["double_sided"] = 1 if getattr(material, "double_sided", False) else 0
self._registry[mat_id] = idx
if content_key is not None:
self._dedup[content_key] = idx
self._slot_to_content_key[idx] = content_key
# Attach a weakref.finalize so the slot is reclaimed when the owning
# Material goes out of scope. Some Material subclasses (dataclass
# without __weakref__ slot) can't be weakref'd, in that case we fall
# back to the monotonic-only path silently. Keyed by mat_id so repeated
# registrations for the same object don't stack finalizers.
if mat_id not in self._finalizers:
try:
manager_ref = weakref.ref(self)
self._finalizers[mat_id] = weakref.finalize(
material,
_release_material_slot,
manager_ref,
mat_id,
idx,
)
except TypeError:
pass # Material class doesn't support weakrefs: slot leaks.
return idx
[docs]
def release(self, mat_id: int, slot_idx: int) -> None:
"""Reclaim a material SSBO slot.
Called by the weakref.finalize attached to a Material when it goes out
of scope. Pushes the slot onto the free list, clears the dedup/registry
entries, and zeroes the SSBO row so stale colour data doesn't leak
through if the slot is re-used before upload() runs this frame.
"""
if mat_id in self._registry:
self._registry.pop(mat_id, None)
if slot_idx in self._slot_to_content_key:
ck = self._slot_to_content_key.pop(slot_idx)
if self._dedup.get(ck) == slot_idx:
self._dedup.pop(ck, None)
if slot_idx < len(self._array):
self._array[slot_idx] = 0
if slot_idx not in self._free_slots:
self._free_slots.append(slot_idx)
self._finalizers.pop(mat_id, None)
[docs]
def upload(self) -> None:
"""Upload the material array to the renderer."""
if self._next_id == 0:
# No materials registered, skip upload to avoid size-0 buffer error
return
used = self._array[: self._next_id]
self._renderer.set_materials(used)