"""Adapter to bridge SceneTree nodes to Renderer submissions."""
import logging
from typing import Any
import numpy as np
from simvx.core import (
Camera2D,
Camera3D,
GPUParticles2D,
GPUParticles3D,
Light2D,
Light3D,
LightOccluder2D,
Material,
MeshInstance3D,
MultiMeshInstance3D,
NinePatchRect,
Node,
ParticleEmitter,
PointLight3D,
SceneTree,
SpotLight3D,
Sprite2D,
Sprite3D,
SubViewport,
Text3D,
)
from simvx.core.light2d import DEFAULT_SHADOW_COLOUR
from simvx.core.tilemap import CHUNK_SIZE, TileMap
from .engine_surface import EngineSurface
from .material_slots import MaterialSlotManager
from .renderer.tile_types import TILE_INSTANCE_DTYPE
from .types import (
LIGHT_DTYPE,
SKIN_DTYPE,
VertexStreams,
)
__all__ = ["SceneAdapter"]
log = logging.getLogger(__name__)
[docs]
def find_all_outside_subviewports(root: Node, node_type: type) -> list:
"""``find_all`` that does NOT descend into ``SubViewport`` subtrees.
A SubViewport's children render into the viewport's own offscreen target
(driven by ``SubViewportManager``), so a main-pass collection must not also
pick them up: a plain ``root.find_all`` would double-submit them (a
``MultiMeshInstance3D`` inside a SubViewport would draw in both the main pass
and the offscreen pass). Mirrors the SubViewport boundary in ``_collect_nodes``.
The root itself is never skipped, so collecting a SubViewport's *own* contents
(root == that SubViewport) still works; nested SubViewports are pruned.
"""
out: list = []
stack = list(root.children)
while stack:
node = stack.pop()
if isinstance(node, node_type):
out.append(node)
if isinstance(node, SubViewport):
continue
stack.extend(node.children)
return out
def _gather_layer_instances(layer, uv_off_lut, uv_size_lut, valid_lut, max_tid, ox, oy, cell_w, cell_h):
"""Build a ``TILE_INSTANCE_DTYPE`` buffer (+ optional tint buffer) for one layer.
Walks the layer's sparse 32x32 chunks and gathers every populated cell with
vectorised numpy: no per-cell Python loop. Positions, atlas-normalised UVs
and tints are sliced straight out of the chunk planes, so the per-frame Python
cost is O(chunks-touched), not O(cells). The tint buffer is returned only when
some chunk actually carries a colour plane, leaving the untinted common path
byte-identical to a colourless submission.
Returns ``(tile_data, colour_data)`` or ``(None, None)`` when the layer is empty.
"""
pos_segs: list = []
uvoff_segs: list = []
uvsize_segs: list = []
colour_segs: list = [] # one entry per accepted chunk: ndarray or None
seg_counts: list = []
any_colour = False
total = 0
for (cx, cy), chunk in layer._chunks.items():
plane = chunk.cells.get(0)
if plane is None:
continue
idx = np.flatnonzero(plane >= 0)
if idx.size == 0:
continue
tids = plane[idx]
# Drop ids the tileset doesn't define (old loop skipped ``get_tile is None``).
in_range = tids <= max_tid
keep = in_range.copy()
keep[in_range] = valid_lut[tids[in_range]]
if not keep.all():
idx = idx[keep]
tids = tids[keep]
if idx.size == 0:
continue
lx = idx % CHUNK_SIZE
ly = idx // CHUNK_SIZE
pos = np.empty((idx.size, 2), dtype=np.float32)
pos[:, 0] = ox + (cx * CHUNK_SIZE + lx) * cell_w
pos[:, 1] = oy + (cy * CHUNK_SIZE + ly) * cell_h
pos_segs.append(pos)
uvoff_segs.append(uv_off_lut[tids])
uvsize_segs.append(uv_size_lut[tids])
colour_plane = chunk.colours.get(0)
if colour_plane is not None:
colour_segs.append(colour_plane[idx])
any_colour = True
else:
colour_segs.append(None)
seg_counts.append(idx.size)
total += idx.size
if total == 0:
return None, None
tile_data = np.zeros(total, dtype=TILE_INSTANCE_DTYPE)
tile_data["position"] = np.concatenate(pos_segs)
tile_data["tile_uv_offset"] = np.concatenate(uvoff_segs)
tile_data["tile_uv_size"] = np.concatenate(uvsize_segs)
# flip_h / flip_v stay zero (per-cell flip is not stored yet).
colour_data = None
if any_colour:
colour_data = np.ones((total, 4), dtype=np.float32)
off = 0
for cnt, seg in zip(seg_counts, colour_segs, strict=True):
if seg is not None:
colour_data[off : off + cnt] = seg
off += cnt
return tile_data, colour_data
[docs]
class SceneAdapter:
"""Bridges SceneTree to Renderer (Vulkan).
Responsibilities:
- Register meshes with the GPU
- Convert materials to Vulkan format
- Traverse scene tree and submit instances
"""
def __init__(self, engine: EngineSurface, renderer: Any):
self._engine = engine
self._renderer = renderer
self._mesh_cache: dict[int, Any] = {} # mesh id -> MeshHandle
# Retained per-layer tilemap instance cache: id(layer) -> (build_key,
# tile_data, colour_data, build_version). A static layer is gathered +
# packed ONCE; subsequent frames reuse the SAME numpy arrays (so the GPU
# passes can skip the re-upload by identity / version) until the layer's
# content, world offset, cell size, or tileset changes the build_key.
# ``build_version`` is a monotonic int the web wire keys its REUSE marker
# on; it bumps only when this cache actually rebuilds the packed buffer.
self._tilemap_cache: dict[int, tuple] = {}
# Bindless material SSBO bookkeeping (slot allocation, content dedup,
# weakref reclamation). Delegates texture loading back to this adapter.
self._materials = MaterialSlotManager(renderer, self._load_texture)
# Cached node collection (invalidated by tree structure changes)
self._cached_nodes: tuple | None = None
self._cached_structure_version: int = -1
# D8 multi-GPU SubViewport-offload coordinator. Built lazily from
# ``engine.multi_device`` ONLY when an opted-in multi-device renderer is
# active (>= 2 devices). Stays ``None`` on the single-GPU / unopted path
# (every box here), where the SRU recording seam takes exactly today's
# primary-device path and the frame is byte-identical.
self._offload_coordinator: Any = None
self._offload_probed: bool = False
[docs]
@property
def offload_coordinator(self) -> Any:
"""The D8 :class:`SRUOffloadCoordinator`, or ``None`` (single-GPU/unopted).
Built once, lazily, from ``engine.multi_device``: a coordinator exists
only when that manager reports an active multi-device renderer. On every
single-GPU box (and any box that did not opt in) this is ``None``, so the
SRU recording seam's ``coordinator is None`` branch keeps the path
byte-identical to today.
"""
if self._offload_probed:
return self._offload_coordinator
self._offload_probed = True
mgr = getattr(self._engine, "multi_device", None)
if mgr is not None and getattr(mgr, "multi_gpu", False):
from .gpu.multi_device import SRUOffloadCoordinator
caps = getattr(self._engine, "capabilities", None)
scale = getattr(self._engine, "content_scale", (1.0, 1.0))
# Bind the primary renderer to slot 0 so the cross-device transfer can
# reach the primary command pool. ``secondary_renderer_factory`` builds a
# ``Renderer(facade)`` on a secondary device; it stays ``None`` on this
# single-GPU box (no secondary exists), so render_offloaded raises the
# clear rig-completion error and the seam never silently drops an SRU.
mgr.attach_renderer(0, self._renderer)
self._offload_coordinator = SRUOffloadCoordinator(
mgr,
caps,
content_scale=scale,
secondary_renderer_factory=self._build_secondary_renderer,
)
return self._offload_coordinator
def _build_secondary_renderer(self, facade: Any) -> Any:
"""Rig-side factory: construct + GPU-``setup`` a secondary-device renderer.
Returns an object exposing ``render_sru_offscreen(sru, target)`` that records
the SRU into ``target`` on the secondary device and submits on its queue. On
the rig this wraps ``Renderer(facade)`` (which builds all pipelines / SSBOs /
descriptors on the secondary device via ``facade.ctx.device``) plus a thin
SRU-record adapter mirroring the primary :meth:`render_sru_from_plan` slice
model. This single-GPU box never calls it (no secondary device is created),
so the heavy GPU build is exercised only on the 4x Arc Pro B70 rig.
"""
from .renderer.forward import Renderer
from .renderer.secondary_sru import SecondarySRURenderer
renderer = Renderer(facade)
renderer.setup()
return SecondarySRURenderer(facade, renderer)
[docs]
def plan_sru_offload(self, srus: list, *, sru_id: Any = None, cost: Any = None) -> None:
"""Compute this frame's per-SRU device routes (no-op without a coordinator).
Called once per frame on the ORDERED SRU list (producer-first) before the
SRUs are recorded, so the per-SRU :meth:`render_to_target` /
:meth:`render_sru_from_plan` can look up their route. Pure decision work;
no Vulkan. On the single-GPU / unopted path there is no coordinator and
this returns immediately, leaving every SRU on the primary (today's path).
``sru_id`` / ``cost`` accessors are forwarded to the coordinator for the
synchronous path, whose ordered items are live SubViewport nodes rather
than :class:`SubViewportSRU` plans.
"""
coord = self.offload_coordinator
if coord is not None:
coord.plan(srus, sru_id=sru_id, cost=cost)
def _maybe_offload_sru(self, sru_id: int, sru: Any = None) -> bool:
"""Route one SRU to a secondary device if this frame's plan assigned it there.
Returns ``True`` when the SRU was handled by the offload path (so the
caller must NOT also record it on the primary), ``False`` when it stays on
the primary and the caller takes today's path. With no coordinator (the
only path on this single-GPU box) this is always ``False``, so the primary
recording below is byte-identical to today.
``sru`` is the OWNED :class:`~.renderer.render_packet.SubViewportSRU` plan
(pipelined ``render_sru_from_plan`` path). Cross-device offload needs the
plan (instances + camera + dims) to render on the secondary, so when no plan
is supplied (the synchronous live-tree ``render_to_target`` path) an
offloaded SRU is NOT routed to a secondary here: it falls through to the
primary path (correct, just unaccelerated). With a plan, the whole render +
cross-device staging transfer runs via
:meth:`SRUOffloadCoordinator.render_sru_offloaded`; on a box without the
rig-side per-device renderer that raises a clear, capability-gated error
rather than silently dropping the SRU. Unreachable on this single-GPU box:
no coordinator is ever built.
"""
coord = self.offload_coordinator
if coord is None:
return False
route = coord.route_for(sru_id)
if route is None or not route.offloaded:
return False
if sru is None:
# Synchronous live-tree path has no plan to render on a secondary;
# render on the primary (unaccelerated) rather than fail.
return False
# Secondary-assigned: render on its device + transfer the colour image into
# the primary bindless image the main pass samples. ``render_sru_offloaded``
# owns the full sequence (residency, secondary record+submit, staging copy).
primary_dst_image = self._sru_primary_image(sru)
return coord.render_sru_offloaded(sru, primary_dst_image)
@staticmethod
def _sru_primary_image(sru: Any) -> Any:
"""The primary-device colour image the main scene samples for this SRU.
The SubViewport's :class:`SubViewportRenderer` registered its offscreen
colour view as a bindless slot the main scene samples; the cross-device
transfer writes the secondary's result INTO that same primary image so the
composite samples the offloaded feed this frame. Resolved off the SRU plan's
renderer (its ``_target.colour_image``).
"""
rend = getattr(sru, "renderer", None)
target = getattr(rend, "_target", None)
return getattr(target, "colour_image", None)
[docs]
def register_mesh(self, mesh: Any) -> Any | None:
"""Register mesh with Vulkan engine, return MeshHandle.
Returns None if the mesh has no vertex data (skipped gracefully).
"""
mesh_id = getattr(mesh, "_uid", id(mesh))
if mesh_id in self._mesh_cache:
return self._mesh_cache[mesh_id]
# Validate mesh has renderable data
if not hasattr(mesh, "positions") or mesh.positions is None or mesh.vertex_count == 0:
log.debug("Skipping mesh registration: no vertex data")
return None
if not hasattr(mesh, "indices") or mesh.indices is None or len(mesh.indices) == 0:
log.debug("Skipping mesh registration: no index data")
return None
# Ensure required attributes
if mesh.normals is None:
mesh.generate_normals()
if not hasattr(mesh, "texcoords") or mesh.texcoords is None:
mesh.texcoords = np.zeros((mesh.vertex_count, 2), dtype=np.float32)
# Assemble the per-stream vertex payload (positions + shading, plus
# the optional extras stream when the mesh carries any of
# tangents/colours/texcoords2).
streams = VertexStreams.build(
mesh.positions,
mesh.normals,
mesh.texcoords,
tangents=getattr(mesh, "tangents", None),
colours=getattr(mesh, "colours", None),
uvs2=getattr(mesh, "texcoords2", None),
)
handle = self._engine.mesh_registry.register(streams, mesh.indices)
self._mesh_cache[mesh_id] = handle
return handle
[docs]
def register_skinned_mesh(self, mesh: Any) -> Any | None:
"""Register skinned mesh with Vulkan engine, return MeshHandle.
Returns None if the mesh has no vertex data.
"""
mesh_id = getattr(mesh, "_uid", id(mesh))
if mesh_id in self._mesh_cache:
return self._mesh_cache[mesh_id]
# Validate mesh has renderable data
if not hasattr(mesh, "indices") or mesh.indices is None or len(mesh.indices) == 0:
log.debug("Skipping skinned mesh registration: no index data")
return None
if not hasattr(mesh, "positions") or mesh.positions is None or mesh.vertex_count == 0:
log.debug("Skipping skinned mesh registration: no vertex data")
return None
if mesh.normals is None:
mesh.generate_normals()
if not hasattr(mesh, "texcoords") or mesh.texcoords is None:
mesh.texcoords = np.zeros((mesh.vertex_count, 2), dtype=np.float32)
# The skinned pipeline requires the skin stream (binding 3): use the
# importer-attached joints/weights (``_skin_stream``), or zeros when a
# mesh is flagged skinned without skin data (renders at bind pose).
skin = getattr(mesh, "_skin_stream", None)
if skin is None:
skin = np.zeros(mesh.vertex_count, dtype=SKIN_DTYPE)
streams = VertexStreams.build(
mesh.positions,
mesh.normals,
mesh.texcoords,
tangents=getattr(mesh, "tangents", None),
colours=getattr(mesh, "colours", None),
uvs2=getattr(mesh, "texcoords2", None),
joints=skin["joints"],
weights=skin["weights"],
)
handle = self._engine.mesh_registry.register(streams, mesh.indices)
self._mesh_cache[mesh_id] = handle
return handle
def _load_texture(self, source, *, colour_space: str = "linear", mipmaps: bool = False) -> int:
"""Load a 3D material texture and return its bindless index.
Accepts any source type understood by ``TextureManager.resolve``
(file path, encoded bytes, numpy ndarray). Returns -1 on failure.
``colour_space`` selects the sampled image view format per map role:
``"srgb"`` (R8G8B8A8_SRGB view, hardware linearises on sample) for COLOUR
maps (albedo / base-colour, emissive); ``"linear"`` (UNORM view) for data
maps (normal, metallic-roughness, ambient occlusion) whose bytes are not
colours and must reach the shader untouched. The 3D PBR shader works in
linear light and the sRGB swapchain does the final linear->sRGB encode.
``mipmaps=True`` requests a runtime-generated mip chain; the
kwarg is only forwarded when set so default loads stay byte-identical
and registrars without the capability fall back to a single mip.
"""
try:
if mipmaps:
return self._engine.texture_manager.resolve(source, colour_space=colour_space, mipmaps=True)
return self._engine.texture_manager.resolve(source, colour_space=colour_space)
except (OSError, ValueError, RuntimeError) as exc:
log.warning("Failed to load texture %s: %s", type(source).__name__, exc)
return -1
def _upload_tileset_atlas(self, ts: Any) -> int:
"""Upload tileset atlas pixels to GPU, return bindless texture index.
The TileSet must have ``_atlas_pixels`` (RGBA uint8 ndarray, shape HxWx4)
and ``_atlas_width`` / ``_atlas_height`` attributes.
"""
pixels = ts._atlas_pixels
w, h = ts._atlas_width, ts._atlas_height
tex_idx = self._engine.upload_texture_pixels(pixels, w, h)
ts._gpu_texture_id = tex_idx
log.debug("TileSet atlas uploaded (%dx%d) -> texture %d", w, h, tex_idx)
return tex_idx
def _collect_nodes(
self,
root: Node,
) -> tuple[list, list, list, list, list, list, list, list, list, list, list, list]:
"""Single-pass collection of all renderable node types.
Walks the tree iteratively (stack-based) to classify every node
into one of: cameras, meshes, sprites, particles,
gpu_particles, nine_patches, lights3d, lights2d, occluders, tilemaps,
billboards3d.
Text2D is intentionally NOT collected: it renders through the one 2D text
builder via the ``Draw2D`` walk (its ``on_draw`` emits a ``TEXT`` op /
native ``GLYPH`` item), not the deleted MSDF overlay pass.
Sprite3D / Text3D (and AnimatedSprite3D, a Sprite3D subclass) ARE
collected: they render as depth-tested camera-facing billboards via the
:class:`Billboard2DPass` inside the 3D scene pass.
"""
cameras: list[Camera3D] = []
meshes: list[MeshInstance3D] = []
sprites: list[Sprite2D] = []
particles: list[ParticleEmitter] = []
gpu_particles: list = []
nine_patches: list[NinePatchRect] = []
lights3d: list[Light3D] = []
lights2d: list[Light2D] = []
occluders: list[LightOccluder2D] = []
tilemaps: list[TileMap] = []
billboards3d: list = []
stack = [root]
while stack:
node = stack.pop()
if isinstance(node, Camera3D):
cameras.append(node)
elif isinstance(node, MeshInstance3D):
meshes.append(node)
elif isinstance(node, Sprite3D | Text3D):
billboards3d.append(node)
elif isinstance(node, NinePatchRect):
nine_patches.append(node)
elif isinstance(node, Sprite2D):
sprites.append(node)
elif isinstance(node, GPUParticles2D | GPUParticles3D):
gpu_particles.append(node)
elif isinstance(node, ParticleEmitter):
particles.append(node)
elif isinstance(node, Light3D):
lights3d.append(node)
elif isinstance(node, Light2D):
lights2d.append(node)
elif isinstance(node, LightOccluder2D):
occluders.append(node)
elif isinstance(node, TileMap):
tilemaps.append(node)
# A SubViewport's subtree renders into its OWN offscreen target
# (driven by SubViewportManager), never the pass that owns the
# outer tree, so don't descend into it here. The exception is when
# the SubViewport *is* the root we were asked to collect: that's the
# SubViewportManager rendering the viewport's own contents.
if isinstance(node, SubViewport) and node is not root:
continue
stack.extend(node.children)
return (
cameras,
meshes,
sprites,
particles,
lights3d,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
billboards3d,
)
def _submit_tilemaps(self, tilemaps: list, tree: SceneTree) -> None:
"""Submit tile layers from pre-collected TileMap nodes to the renderer backend.
Desktop path: calls ``TileMapPass.submit_layer`` on the Vulkan forward renderer.
Web path: calls ``WebRenderer.submit_tilemap_layer`` which queues layers for
serialization into the per-frame binary, to be rendered by the JS tilemap pass.
"""
if not tilemaps:
return
tilemap_pass = getattr(self._renderer, "_tilemap_pass", None)
web_submit = getattr(self._renderer, "submit_tilemap_layer", None)
if tilemap_pass is None and web_submit is None:
return
# Both backends expose the same submit_layer signature; resolve once.
submit = tilemap_pass.submit_layer if tilemap_pass is not None else web_submit
seen_layers: set[int] = set()
for tilemap in tilemaps:
if not tilemap._visible_in_hierarchy:
continue
cell_w, cell_h = tilemap.cell_size
# World-space offset of the tilemap (parent transforms included).
# The tilemap shader works in world coordinates and does not yet
# carry a per-layer model matrix, so we premultiply the translation
# into each tile's position. The highlight overlays add the same
# offset so they share the tiles' world frame. Rotation/scale on the
# tilemap node are NOT yet supported: they would require a per-layer
# model matrix in tilemap.vert.
wp = tilemap.world_position
ox, oy = float(wp[0]), float(wp[1])
# Textured content layers (need a tileset + atlas).
if tilemap.tile_set is not None:
self._submit_tilemap_content(tilemap, submit, ox, oy, float(cell_w), float(cell_h), seen_layers)
# Highlight overlays: independent of the tileset, layered on top.
if tilemap._highlight_groups:
self._submit_tilemap_highlights(tilemap, submit, ox, oy, float(cell_w), float(cell_h))
# Drop cache entries for layers not submitted this frame (freed tilemaps,
# newly-hidden layers): bounds the cache and limits id(layer) recycling.
if len(self._tilemap_cache) != len(seen_layers):
self._tilemap_cache = {k: v for k, v in self._tilemap_cache.items() if k in seen_layers}
def _submit_tilemap_content(
self, tilemap, submit, ox: float, oy: float, cell_w: float, cell_h: float, seen_layers: set
) -> None:
"""Submit the textured content layers of one TileMap (one draw per layer).
``tilemap.tile_set`` is known non-None here. Each visible layer's
populated cells are gathered into a TILE_INSTANCE buffer (+ optional
tint) with vectorised numpy and submitted to the shared tilemap pass.
The gather + pack runs ONLY when the layer's content version, world
offset, cell size, or tileset changed since the last frame (``_tilemap_cache``
keyed by ``id(layer)``); otherwise the cached arrays are reused as-is, so a
static layer's per-frame cost is the cheap key comparison rather than an
O(cells) rebuild and a fresh allocation.
"""
ts = tilemap.tile_set
# Lazy-upload atlas pixels to GPU on first encounter.
tex_id = getattr(ts, "_gpu_texture_id", -1)
if tex_id < 0:
if hasattr(ts, "_atlas_pixels") and ts._atlas_pixels is not None:
tex_id = self._upload_tileset_atlas(ts)
else:
return
atlas_w = getattr(ts, "_atlas_width", 0)
atlas_h = getattr(ts, "_atlas_height", 0)
if atlas_w <= 0 or atlas_h <= 0:
return
# tile_id -> normalised UV region lookup tables, vectorised so the
# per-frame gather below never touches the TileSet dict per cell.
# Cached on the tileset (keyed by tile_count) like the GPU atlas
# upload: a static tileset builds these once, not every frame.
uv_off_lut, uv_size_lut, valid_lut = self._tileset_uv_luts(ts, atlas_w, atlas_h)
max_tid = valid_lut.shape[0] - 1
if max_tid < 0:
return
tile_size = (cell_w, cell_h)
# ``id(uv_off_lut)`` is the tileset's UV-table identity: it changes only
# when ``_tileset_uv_luts`` rebuilds (tileset / atlas swap), so folding it
# into the build key invalidates the cache on a tileset change for free.
uv_sig = id(uv_off_lut)
for layer_idx in range(tilemap.layer_count):
layer = tilemap.get_layer(layer_idx)
if not layer.visible:
continue
lid = id(layer)
seen_layers.add(lid)
build_key = (layer.content_version, ox, oy, cell_w, cell_h, uv_sig)
cached = self._tilemap_cache.get(lid)
if cached is not None and cached[0] == build_key:
_key, tile_data, colour_data, build_version = cached
else:
tile_data, colour_data = _gather_layer_instances(
layer, uv_off_lut, uv_size_lut, valid_lut, max_tid, ox, oy, cell_w, cell_h
)
build_version = (cached[3] + 1) if cached is not None else 0
self._tilemap_cache[lid] = (build_key, tile_data, colour_data, build_version)
if tile_data is not None:
submit(tile_data, tex_id, tile_size, colour_data, layer_id=lid, version=build_version)
@staticmethod
def _submit_tilemap_highlights(tilemap, submit, ox: float, oy: float, cell_w: float, cell_h: float) -> None:
"""Submit each highlight group as a textureless solid-fill tile layer.
Highlights ride the SAME GPU tilemap pass as the content, appended after
it so they layer on top (painter's order, no depth test). ``tex_id`` -1
makes the fragment shader sample white (1,1,1,1), and the per-cell tint
turns each white quad into the translucent overlay fill. One instanced
draw per group; positions are vectorised (O(cells) numpy, no per-cell
Python loop), so a large movement range stays O(1) Python per frame.
"""
tile_size = (cell_w, cell_h)
for cells, colour in tilemap._highlight_groups:
n = cells.shape[0]
if n == 0:
continue
tile_data = np.zeros(n, dtype=TILE_INSTANCE_DTYPE)
pos = tile_data["position"]
pos[:, 0] = ox + cells[:, 0] * cell_w
pos[:, 1] = oy + cells[:, 1] * cell_h
colour_data = np.empty((n, 4), dtype=np.float32)
colour_data[:] = colour
# No layer_id / version: highlights are rebuilt every frame (a fresh
# array each time), so the GPU passes treat them as always-changed and
# never apply the static-layer reuse skip to them.
submit(tile_data, -1, tile_size, colour_data, layer_id=None, version=None)
@staticmethod
def _tileset_uv_luts(ts, atlas_w: int, atlas_h: int):
"""Return ``(uv_offset_lut, uv_size_lut, valid_lut)`` for ``ts``.
Each is indexed by tile id (dense, ``0..max_tile_id``): the UV LUTs hold
the atlas-normalised region offset/size, ``valid_lut`` marks which ids
actually exist. Built by vectorising the TileSet's region tuples and
cached on the tileset, keyed by tile count, so a static tileset pays the
O(tiles) build once rather than per frame. Mirrors the ``_gpu_texture_id``
atlas-upload cache (same immutable-after-load assumption).
"""
tiles = ts._tiles
sig = (len(tiles), atlas_w, atlas_h)
if getattr(ts, "_uv_lut_sig", None) == sig:
return ts._uv_off_lut, ts._uv_size_lut, ts._uv_valid_lut
if tiles:
ids = np.fromiter(tiles.keys(), dtype=np.int64, count=len(tiles))
regions = np.array([td.texture_region for td in tiles.values()], dtype=np.float32)
size = int(ids.max()) + 1
else:
ids = np.empty(0, dtype=np.int64)
regions = np.empty((0, 4), dtype=np.float32)
size = 0
uv_off = np.zeros((size, 2), dtype=np.float32)
uv_size = np.zeros((size, 2), dtype=np.float32)
valid = np.zeros(size, dtype=bool)
if size:
inv = np.array([atlas_w, atlas_h], dtype=np.float32)
uv_off[ids] = regions[:, 0:2] / inv
uv_size[ids] = regions[:, 2:4] / inv
valid[ids] = True
ts._uv_off_lut, ts._uv_size_lut, ts._uv_valid_lut, ts._uv_lut_sig = uv_off, uv_size, valid, sig
return uv_off, uv_size, valid
def _submit_lights(self, lights: list, cull_mask: int = 0xFFFFFFFF) -> None:
"""Upload pre-collected Light3D nodes to GPU SSBO, filtered by light_cull_mask vs camera cull_mask."""
if not lights:
return
# Filter: skip hidden lights and those whose light_cull_mask has no overlap with the camera's cull_mask
visible_lights = [
n for n in lights if n._visible_in_hierarchy and (getattr(n, "light_cull_mask", 0xFFFFFFFF) & cull_mask)
]
if not visible_lights:
return
light_data = np.zeros(len(visible_lights), dtype=LIGHT_DTYPE)
for i, node in enumerate(visible_lights):
pos = node.world_position
fwd = node.forward
colour = node.colour
intensity = node.intensity
if isinstance(node, SpotLight3D):
light_data[i]["position"] = (pos.x, pos.y, pos.z, 2.0)
light_data[i]["direction"] = (fwd.x, fwd.y, fwd.z, 0.0)
light_data[i]["params"] = (
node.range,
node.inner_cone,
node.outer_cone,
0.0,
)
elif isinstance(node, PointLight3D):
light_data[i]["position"] = (pos.x, pos.y, pos.z, 1.0)
light_data[i]["direction"] = (0.0, 0.0, 0.0, 0.0)
light_data[i]["params"] = (node.range, 0.0, 0.0, 0.0)
else: # DirectionalLight3D or base Light3D
light_data[i]["position"] = (fwd.x, fwd.y, fwd.z, 0.0)
light_data[i]["direction"] = (fwd.x, fwd.y, fwd.z, 0.0)
light_data[i]["params"] = (0.0, 0.0, 0.0, 0.0)
light_data[i]["colour"] = (colour[0], colour[1], colour[2], intensity)
# Shadow flag in params[3]: 1.0 = casts shadows
if getattr(node, "shadows", False):
light_data[i]["params"][3] = 1.0
self._renderer.set_lights(light_data)
def _submit_mesh_nodes(self, meshes: list, cull_mask: int = 0xFFFFFFFF) -> None:
"""Submit pre-collected MeshInstance3D nodes to the renderer, filtered by cull_mask."""
register = self._materials.register
for node in meshes:
if not node._visible_in_hierarchy:
continue
if node.mesh is None:
continue
# Filter by render layer vs camera cull mask
if not (getattr(node, "render_layer", 1) & cull_mask):
continue
model_mat = node.model_matrix
if not isinstance(model_mat, np.ndarray):
model_mat = np.ascontiguousarray(np.array(model_mat, dtype=np.float32).reshape(4, 4))
material_id = register(node.material)
skeleton = getattr(node, "skeleton", None)
is_skinned = getattr(node, "_is_skinned", False)
shader_material = getattr(node, "shader_material", None)
if is_skinned and skeleton and hasattr(skeleton, "joint_matrices"):
handle = self.register_skinned_mesh(node.mesh)
if handle is None:
continue
self._renderer.submit_skinned_instance(
mesh_handle=handle,
transform=model_mat,
material_id=material_id,
joint_matrices=skeleton.joint_matrices,
)
elif shader_material is not None and hasattr(self._renderer, "submit_shader_instance"):
# Custom ShaderMaterial path: renderer-specific (Vulkan today)
handle = self.register_mesh(node.mesh)
if handle is None:
continue
self._renderer.submit_shader_instance(
mesh_handle=handle,
transform=model_mat,
material_id=material_id,
shader_material=shader_material,
)
else:
handle = self.register_mesh(node.mesh)
if handle is None:
continue
self._renderer.submit_instance(
mesh_handle=handle,
transform=model_mat,
material_id=material_id,
viewport_id=0,
render_layers=int(getattr(node, "render_layer", 1)),
)
def _submit_multimesh_nodes(self, root: Node, cull_mask: int = 0xFFFFFFFF) -> None:
"""Submit all MultiMeshInstance3D descendants: vectorized batch submission."""
register = self._materials.register
# Per-node compose cache: skip the node_global @ instances matmul for a
# static block (unchanged node transform + MultiMesh version). The renderer
# additionally skips re-uploading it (see Renderer._upload_multimesh_blocks).
compose_cache: dict[int, tuple[int, np.ndarray, np.ndarray | None]] = self.__dict__.setdefault(
"_mm_compose_cache", {}
)
seen: set[int] = set()
for node in find_all_outside_subviewports(root, MultiMeshInstance3D):
if not node._visible_in_hierarchy:
continue
mm = node.multi_mesh
if mm is None or mm.mesh is None or mm.instance_count == 0:
continue
if not (getattr(node, "render_layer", 1) & cull_mask):
continue
handle = self.register_mesh(mm.mesh)
if handle is None:
continue
material_id = register(node.material)
# Node's own global transform (applied to all instances)
node_mat = node.model_matrix
if not isinstance(node_mat, np.ndarray):
node_mat = np.ascontiguousarray(np.array(node_mat, dtype=np.float32).reshape(4, 4))
# Respect visible_instance_count (-1 means all)
vis = node.visible_instance_count
count = mm.instance_count if vis < 0 else min(vis, mm.instance_count)
if count == 0:
continue
# Content generation: changes when the MultiMesh, node transform, material,
# or visible count changes. Drives both the compose cache (here) and the
# renderer's upload-skip. Hash is over tiny data (a 4x4 + a few ints).
cache_key = id(node)
version = hash((mm._version, node_mat.tobytes(), int(material_id), int(count)))
seen.add(cache_key)
cached = compose_cache.get(cache_key)
if cached is not None and cached[0] == version:
# Static block: reuse composed transforms AND per-instance material
# indices, skipping both the matmul and the O(N) register() loop.
final, mat_ids = cached[1], cached[2]
else:
inst_transforms = mm.transforms[:count] # (count, 4, 4)
if np.allclose(node_mat, np.eye(4, dtype=np.float32)):
final = np.ascontiguousarray(inst_transforms)
else:
final = np.ascontiguousarray(node_mat @ inst_transforms) # broadcast (4,4) @ (count,4,4)
# Per-instance colour -> material indices. Rebuilt only when the
# block version changes (a colour edit bumps mm._version via
# set_colour/set_all_colours); the indices are stable across frames
# because the material registry dedups and never resets per frame.
mat_ids = None
if mm.colours is not None:
if mm._colour_materials is None or mm._dirty:
mm._colour_materials = [Material(colour=tuple(mm.colours[i])) for i in range(mm.instance_count)]
mm._dirty = False
mat_ids = np.array([register(mm._colour_materials[i]) for i in range(count)], dtype=np.uint32)
compose_cache[cache_key] = (version, final, mat_ids)
self._renderer.submit_multimesh(
mesh_handle=handle,
transforms=final,
material_id=material_id,
material_ids=mat_ids,
viewport_id=0,
cache_key=cache_key,
version=version,
)
# Drop compose-cache entries for nodes not seen this frame (freed/hidden).
if compose_cache:
for dead in [k for k in compose_cache if k not in seen]:
del compose_cache[dead]
def _submit_billboards3d(self, billboards3d: list, cull_mask: int = 0xFFFFFFFF) -> None:
"""Resolve textures + emit depth-tested billboards for Sprite3D / Text3D.
The depth-tested 2D-in-3D mechanism. This method does
the lazy, render-pass-unsafe GPU work HERE (outside any render pass):
resolves each Sprite3D's bindless ``_texture_id`` and ensures the MSDF
atlas slot for Text3D. It then calls each node's ``on_draw(renderer)`` --
the engine's emit hook -- which forwards to ``Renderer.draw_sprite_3d`` /
``draw_text_3d`` to append ``BILLBOARD_DTYPE`` rows that Billboard2DPass
draws (depth-test/no-write) inside the 3D scene pass.
"""
if not billboards3d:
return
if not hasattr(self._renderer, "draw_sprite_3d"): # backend without billboards (web)
return
tm = self._engine.texture_manager
for node in billboards3d:
if not node._visible_in_hierarchy:
continue
if not (getattr(node, "render_layer", 1) & cull_mask):
continue
if isinstance(node, Sprite3D) and node._texture_id < 0 and node.texture is not None:
tex_id = tm.resolve(node.texture, filter=getattr(node, "filter", "linear"))
if tex_id >= 0:
node._texture_id = tex_id
w, h = tm.get_texture_size(tex_id)
node._texture_width, node._texture_height = w, h
node.on_draw(self._renderer)
[docs]
def submit_scene(self, tree: SceneTree, *, submit_lights: bool = True, submit_2d: bool = True) -> None:
"""Walk scene tree and submit instances to renderer.
Performs a single tree traversal to collect all renderable node types,
then passes the pre-collected lists to individual submission methods.
Gracefully handles scenes with no Camera3D (e.g. editor UI-only scenes)
by skipping 3D submission while still processing 2D overlays.
``submit_lights`` is False for offscreen scene-render units (SubViewport /
reflection-probe faces): the engine keeps a single shared light SSBO, so
in the single-submit multi-scene frame only one light set can survive.
The main scene owns it; offscreen SRUs render with the main scene's lights
(a known limitation). Skipping the upload
also keeps the main scene's light buffer intact without a re-submit.
``submit_2d`` is False for a RenderView SRU: it re-submits the MAIN tree
(already submitted this frame), and the 2D overlay submits are per-frame
APPENDS to shared renderer lists (particles, tilemap layers, 2D lights),
so running them a second time would double that content in the main
pass. A RenderView captures the 3D world only.
"""
if not tree.root:
return
# Cached single-pass collection: only re-walk on tree structure changes
version = tree._structure_version
if self._cached_nodes is not None and self._cached_structure_version == version:
(
cameras,
meshes,
sprites,
particles,
lights3d,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
billboards3d,
) = self._cached_nodes
else:
collected = self._collect_nodes(tree.root)
self._cached_nodes = collected
self._cached_structure_version = version
(
cameras,
meshes,
sprites,
particles,
lights3d,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
billboards3d,
) = collected
# Allow external camera override (e.g. editor orbit camera for textured preview)
camera = getattr(tree, "_render_camera_override", None)
if camera is None:
visible_cameras = [c for c in cameras if c._visible_in_hierarchy]
if not visible_cameras:
# No camera: still process sprite/particle 2D overlays below
if submit_2d:
self._submit_2d_overlays(
tree,
sprites,
particles,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
)
return
camera = visible_cameras[0]
# Setup viewport: use play_viewport_rect if set (editor play mode)
vp_rect = getattr(tree, "play_viewport_rect", None)
if vp_rect is not None:
# Logical coords → physical pixels (HiDPI)
sx, sy = self._engine.content_scale
vp_x, vp_y, w, h = (
vp_rect[0] * sx,
vp_rect[1] * sy,
vp_rect[2] * sx,
vp_rect[3] * sy,
)
else:
w, h = self._engine.extent
vp_x, vp_y = 0, 0
aspect = w / h if h > 0 else 1.0
# Camera matrices (always numpy arrays now)
try:
view_mat = camera.view_matrix
except AttributeError, TypeError:
# Camera parent chain may be incomplete during scene transitions
if submit_2d:
self._submit_2d_overlays(
tree,
sprites,
particles,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
)
return
if not isinstance(view_mat, np.ndarray):
view_mat = np.ascontiguousarray(np.array(view_mat, dtype=np.float32).reshape(4, 4))
proj_mat = camera.projection_matrix(aspect)
if not isinstance(proj_mat, np.ndarray):
proj_mat = np.ascontiguousarray(np.array(proj_mat, dtype=np.float32).reshape(4, 4))
# Ensure the 3x3 rotation submatrix has positive determinant (right-handed, no mirroring).
# A negative determinant causes geometry to render horizontally flipped.
rot_det = np.linalg.det(view_mat[:3, :3])
if rot_det < 0:
log.warning("View matrix 3x3 det=%.4f (mirrored), negating right vector to fix", rot_det)
view_mat[0, :3] = -view_mat[0, :3]
view_mat[0, 3] = -view_mat[0, 3]
# Clear and create viewport
self._renderer.viewport_manager.clear()
self._renderer.viewport_manager.create_viewport(
x=int(vp_x),
y=int(vp_y),
width=int(w),
height=int(h),
camera_view=view_mat,
camera_proj=proj_mat,
)
# Submit all MeshInstance3D nodes (filtered by camera cull_mask)
cull_mask = getattr(camera, "cull_mask", 0xFFFFFFFF)
self._submit_mesh_nodes(meshes, cull_mask)
# Submit all MultiMeshInstance3D nodes
self._submit_multimesh_nodes(tree.root, cull_mask)
# Submit Sprite3D / Text3D as depth-tested billboards. Needs
# the 3D camera viewport (created above), so it runs only on the camera path.
self._submit_billboards3d(billboards3d, cull_mask)
# Submit 2D overlays, particles, lights, materials
if submit_2d:
self._submit_2d_overlays(
tree,
sprites,
particles,
lights2d,
occluders,
tilemaps,
gpu_particles,
nine_patches,
)
# Collect and upload lights (filtered by light_cull_mask vs camera cull_mask).
# Skipped for offscreen SRUs so the main scene's shared light buffer survives.
if submit_lights:
self._submit_lights(lights3d, cull_mask)
# Upload materials (shared global table; same content regardless of SRU)
self._materials.upload()
[docs]
def resolve_2d_textures(self, root: Node) -> None:
"""Resolve Sprite2D / NinePatchRect textures under ``root``.
The item-pipeline RTT-2D collector needs a SubViewport subtree's sprite
``_texture_id`` populated BEFORE it captures ``on_draw`` (a sprite whose
texture is still ``-1`` emits nothing, so the collected item view would be
empty and stay cached empty -- ``_texture_id`` is not a Property, so it
fires no invalidation). The main pass resolves its own sprites in
``submit_scene`` AFTER the item collection runs, and that walk prunes
SubViewports anyway -- so a SubViewport's 2D sprites are resolved here,
from the viewport's own root. Idempotent: an already-resolved sprite
(``_texture_id >= 0``) and a live SubViewport source are both skipped.
"""
from simvx.core import NinePatchRect
tm = self._engine.texture_manager
stack = list(root.children)
while stack:
node = stack.pop()
if isinstance(node, SubViewport):
continue # nested viewport: its own collector resolves it
if isinstance(node, Sprite2D):
if (
not getattr(node.texture, "_is_subviewport", False)
and node._texture_id < 0
and node.texture is not None
):
tex_id = tm.resolve(node.texture, filter=getattr(node, "filter", "linear"))
if tex_id >= 0:
node._texture_id = tex_id
if node.width == 0 or node.height == 0:
tw, th = tm.get_texture_size(tex_id)
fh = getattr(node, "frames_h", 1) or 1
fv = getattr(node, "frames_v", 1) or 1
if node.width == 0:
node.width = max(1, tw // fh)
if node.height == 0:
node.height = max(1, th // fv)
elif isinstance(node, NinePatchRect) and node._texture_id < 0 and node.texture is not None:
tex_id = tm.resolve(node.texture)
if tex_id >= 0:
node._texture_id = tex_id
stack.extend(node.children)
def _submit_2d_overlays(
self,
tree: SceneTree,
sprites: list,
particles: list,
lights2d: list,
occluders: list,
tilemaps: list,
gpu_particles: list | None = None,
nine_patches: list | None = None,
) -> None:
"""Resolve sprite/nine-patch textures and submit particle emitters + 2D lights.
Text2D is no longer handled here: it renders through the one 2D text
builder via the ``Draw2D`` walk, so the MSDF overlay submit
that used to live in this method is deleted.
"""
# Publish the active Camera2D mapping to the renderer so 2D overlays that
# render through the 3D pass (GPU particles, tilemaps) project to screen via
# the SAME canvas_transform sprites are baked with: otherwise
# they fall back to the 3D camera and stay screen-fixed while the 2D world
# pans. With no Camera2D, default to the identity world->screen affine (world
# units ARE screen pixels) so these overlays still render screen-space --
# matching Sprite2D / Text2D / Draw2D, which all draw without an explicit
# camera. Without this default a cameraless 2D scene drops every particle /
# tilemap (render_particles returns early with no projection to use).
ss = tuple(float(s) for s in tree._screen_size)
cam2d = tree._current_camera_2d
affine = cam2d.canvas_transform(ss) if cam2d is not None else (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
self._renderer._camera2d_affine = (affine, ss)
# Load sprite textures lazily (sets _texture_id so Sprite2D.on_draw() works).
# When width/height are still 0 ("use native"), populate them from the
# texture's pixel dimensions so the sprite renders at its source size.
# Honour ``Sprite2D.filter`` so pixel-art ports get nearest-neighbour
# sampling without losing the smooth default for everything else.
tm = self._engine.texture_manager
for node in sprites:
# A SubViewport source resolves its bindless slot live in
# ``Sprite2D.on_draw`` from ``subviewport.texture`` -- it is NOT a path/
# bytes/ndarray the TextureManager loads, so skip the lazy resolve.
if getattr(node.texture, "_is_subviewport", False):
continue
if node._texture_id < 0 and node.texture is not None:
tex_id = tm.resolve(node.texture, filter=getattr(node, "filter", "linear"))
if tex_id >= 0:
node._texture_id = tex_id
if node.width == 0 or node.height == 0:
tw, th = tm.get_texture_size(tex_id)
# Sprite sheets render one frame at a time, so divide
# the native dimensions by the sheet grid.
fh = getattr(node, "frames_h", 1) or 1
fv = getattr(node, "frames_v", 1) or 1
if node.width == 0:
node.width = max(1, tw // fh)
if node.height == 0:
node.height = max(1, th // fv)
# Load NinePatchRect textures lazily (sets _texture_id and texture_size)
for node in nine_patches or []:
if node._texture_id < 0 and node.texture is not None:
tex_id = tm.resolve(node.texture)
if tex_id >= 0:
node._texture_id = tex_id
if node.texture_size is None:
node.texture_size = tm.get_texture_size(tex_id)
# Submit particle emitters
for node in particles:
data = node.particle_data
if data is not None and len(data) > 0:
self._renderer.submit_particles(data)
# Submit GPU particle emitters (compute-shader driven). Both
# backends key persistent SSBOs by ``emitter_id`` so multi-emitter
# scenes render correctly and simulation state survives across
# frames per emitter node: desktop ``ParticleCompute`` and web
# ``GPUParticlePass`` share the same per-emitter ownership model.
if gpu_particles and hasattr(self._renderer, "submit_gpu_particles"):
for node in gpu_particles:
if node.emitting or not getattr(node, "_cycle_complete", False):
self._renderer.submit_gpu_particles(
node.emitter_config,
emitter_id=id(node) & 0xFFFFFFFF,
)
# Submit TileMap layers
self._submit_tilemaps(tilemaps, tree)
# Collect and submit 2D lights and occluders
self._submit_lights_2d(tree, lights2d, occluders)
def _submit_lights_2d(self, tree: SceneTree, lights2d: list, occluders: list) -> None:
"""Submit pre-collected Light2D and LightOccluder2D nodes to renderer.
Uses viewport/camera culling to skip lights and occluders that are
entirely outside the visible area, reducing GPU and CPU work.
"""
if not lights2d:
return
# Compute viewport bounds from 2D camera
cam: Camera2D | None = tree._current_camera_2d
sw, sh = tree._screen_size
if cam is not None:
zoom = cam.zoom if cam.zoom > 0 else 1.0
cx, cy = cam.current.x, cam.current.y
half_w, half_h = (sw / zoom) * 0.5, (sh / zoom) * 0.5
else:
cx, cy = sw * 0.5, sh * 0.5
half_w, half_h = cx, cy
vp_min_x, vp_max_x = cx - half_w, cx + half_w
vp_min_y, vp_max_y = cy - half_h, cy + half_h
# Collect visible lights (viewport-culled, layer-filtered)
cam_cull_mask = getattr(cam, "cull_mask", 0xFFFFFFFF) if cam is not None else 0xFFFFFFFF
max_light_range = 0.0
lights_to_submit: list[dict] = []
for node in lights2d:
if not node.enabled:
continue
# Filter by light_cull_mask vs camera cull_mask
if not (getattr(node, "light_cull_mask", 0xFFFFFFFF) & cam_cull_mask):
continue
data = node._get_light_data()
lx, ly = data["position"]
lr = data["range"]
# Directional lights are global (a fullscreen quad in the disk pass),
# so their node position must not viewport-cull them.
if data.get("type") != "directional":
if lx + lr < vp_min_x or lx - lr > vp_max_x:
continue
if ly + lr < vp_min_y or ly - lr > vp_max_y:
continue
lights_to_submit.append(data)
if lr > max_light_range:
max_light_range = lr
if not lights_to_submit:
return
# Expanded viewport for occluder culling (viewport + max light range)
exp_min_x = vp_min_x - max_light_range
exp_max_x = vp_max_x + max_light_range
exp_min_y = vp_min_y - max_light_range
exp_max_y = vp_max_y + max_light_range
# Camera2D mapping: the light2d pass treats light_pos /
# light_range / occluder verts as SCREEN-space pixels (light2d.vert:
# world_pos/screen_size; the shadow ray-cast shares that space), so world-space
# lights and occluders must be projected through the SAME canvas_transform the
# sprites are baked with, otherwise they stay screen-fixed while the world pans.
# The transform is rotation-free uniform scale + translation, so the ray-edge
# shadow geometry is preserved.
if cam is not None:
a, _b, _c, d, tx, ty = cam.canvas_transform(tree._screen_size)
else:
a = d = 1.0
tx = ty = 0.0
# Submit occluders within expanded viewport (quick position pre-check)
for node in occluders:
if not node.polygon:
continue
# Quick AABB check using node world_position before expensive polygon transform
gp = node.world_position
px, py = float(gp[0]), float(gp[1])
# Estimate occluder extent from polygon local bounds
max_ext = max(abs(v[0]) + abs(v[1]) for v in node.polygon) + 1
if px + max_ext < exp_min_x or px - max_ext > exp_max_x:
continue
if py + max_ext < exp_min_y or py - max_ext > exp_max_y:
continue
poly = [(a * float(vx) + tx, d * float(vy) + ty) for vx, vy in node.global_polygon]
self._renderer.submit_occluder2d(poly)
# Submit visible lights
for data in lights_to_submit:
lx, ly = data["position"]
screen_pos = (a * float(lx) + tx, d * float(ly) + ty)
self._renderer.submit_light2d(
position=screen_pos,
colour=data["colour"],
energy=data["energy"],
light_range=data["range"] * a,
falloff=data.get("falloff", 1.0),
inner_radius=data.get("inner_radius", 0.0) * a,
falloff_gradient=data.get("falloff_gradient"),
blend_mode=data.get("blend_mode", "add"),
shadow_enabled=data.get("shadow_enabled", False),
shadow_colour=data.get("shadow_colour", DEFAULT_SHADOW_COLOUR),
shadow_softness=data.get("shadow_softness", 1.0),
light_type=data.get("type", "point"),
direction=data.get("direction", (0.0, -1.0)),
)
[docs]
def render_to_target(
self,
cmd,
target,
tree: SceneTree,
*,
camera=None,
screen_size: tuple[float, float] | None = None,
draw2d_ops: list | None = None,
item_2d: tuple | None = None,
sru_id: int = 0,
submit_2d: bool = True,
occlusion: Any = None,
) -> None:
"""Record one offscreen scene-render unit (SRU) into the shared command buffer.
Single-submit slice model (no per-scene ``begin_frame`` / submit / wait):
the offscreen scene's instances are submitted into an *isolated* list, a
contiguous slice of the shared transform SSBO is reserved for them, their
transforms are written into that slice, and their draws are recorded with
absolute ``first_instance`` indices into the same primary ``cmd`` that the
main frame uses. The main scene's per-frame submission lists are saved and
restored around this call so its base-0 slice is untouched: the whole frame
still ends in exactly ONE ``vkQueueSubmit``.
The offscreen colour target is written here and sampled by the main pass
later in the same command buffer; ``RenderTarget`` leaves it in
``SHADER_READ_ONLY_OPTIMAL`` (its end-of-pass store + image-layout
transition is an in-cmd barrier), so the within-frame
write-then-sample ordering is valid without any queue wait.
Args:
cmd: The frame's primary Vulkan command buffer (from pre_render).
target: Offscreen render target with begin_pass/end_pass/width/height/ready.
tree: SceneTree (or _SubTreeView) to render.
camera: Optional camera override (e.g. probe face / editor orbit camera).
screen_size: Optional 2D screen size override (for 2D viewport rendering).
draw2d_ops: Pre-extracted Draw2D ops to render as a 2D overlay.
sru_id: Stable identity of this SRU (SubViewport node id / probe-face id),
used to key the frustum visibility cache so SRUs never collide.
submit_2d: Forwarded to :meth:`submit_scene`. False for a RenderView
SRU, whose re-submit of the already-submitted MAIN tree must not
append the tree's 2D overlays into the shared per-frame lists a
second time (they would double in the main pass).
occlusion: Optional per-view :class:`~.renderer.view_occlusion.ViewOcclusion`
bundle (``use_occlusion=True`` on the node). When given,
the SRU's transforms upload their AABBs and the view's own
two-phase Hi-Z cull is recorded before its colour pass, so the
SRU render draws from the culled indirect batches. ``None``
(the default) records exactly the historical un-culled SRU.
"""
if not target.ready:
return
# D8 multi-GPU seam: if this frame's plan assigned this SRU to a secondary
# device, render it there and transfer the result to the primary instead of
# recording it on this (primary) cmd. No coordinator (single-GPU / unopted,
# the only path here) => False => today's primary recording below, unchanged.
if self._maybe_offload_sru(sru_id):
return
# Save tree state
saved_vp = tree.play_viewport_rect
saved_cam = getattr(tree, "_render_camera_override", None)
saved_screen = tree._screen_size if screen_size is not None else None
# Set overrides: viewport rect computed from target dims + HiDPI scale
sx, sy = self._engine.content_scale
tree.play_viewport_rect = (0, 0, target.width / sx, target.height / sy)
if camera is not None:
tree._render_camera_override = camera
if screen_size is not None:
tree._screen_size = screen_size
renderer = self._renderer
vpm = renderer.viewport_manager
# Isolate this SRU's submission lists + viewport so the main scene's
# base-0 slice and its viewport (already submitted this frame) are
# preserved. submit_scene resets viewport_manager to this SRU's single
# viewport; we snapshot and restore the main scene's viewports after.
saved_instances = renderer._instances
saved_skinned = renderer._skinned_instances
saved_base = renderer._first_instance_base
saved_sru = renderer._sru_id
saved_viewports = dict(vpm._viewports)
saved_next_id = vpm._next_id
# MultiMesh blocks are per-pass too: isolate this SRU's so it draws its own
# blocks (a MultiMesh inside a SubViewport) without disturbing the main
# scene's already-resolved blocks.
saved_mm_blocks = renderer._multimesh_blocks
saved_mm_draws = renderer._multimesh_draws
# Billboards (Sprite3D/Text3D) are per-pass: isolate this SRU's so they draw
# into the offscreen target, not leak into the main pass.
saved_billboards = renderer._billboard_submissions
renderer._instances = []
renderer._skinned_instances = []
renderer._multimesh_blocks = []
renderer._multimesh_draws = []
renderer._billboard_submissions = []
try:
self.submit_scene(tree, submit_lights=False, submit_2d=submit_2d)
# Resolve this SRU's MultiMesh blocks BEFORE reserving the instance slice
# so any ineligible blocks expand into _instances and are counted/uploaded
# with the rest (mirrors Renderer.pre_render ordering).
renderer._prepare_multimesh()
# Reserve a slice for this SRU's opaque + skinned instances and write
# its transforms there; draws record first_instance = base + local.
n_slots = len(renderer._instances) + len(renderer._skinned_instances)
base = renderer._buffers.reserve_slots(n_slots)
renderer._first_instance_base = base
renderer._sru_id = sru_id
renderer._buffers.upload_transforms(
renderer._instances,
upload_aabbs=occlusion is not None,
base=base,
)
# Upload the eligible blocks into slots after this SRU's instance slice;
# builds _multimesh_draws that render_scene_content draws for this SRU.
renderer._upload_multimesh_blocks()
# Per-view two-phase Hi-Z occlusion: record the view's own
# cull island HERE, outside any render pass, against the SRU's
# freshly-uploaded slice + its single viewport. Leaves the scene
# renderer's plans pointing at the view's culled batches, which the
# render_scene_content below consumes; the ``finally`` restores the
# main pass's plan state via ``occlusion.finish``.
if occlusion is not None:
occlusion.record(cmd, renderer, target.width, target.height)
# hdr_output=0: tone-map in-shader so the offscreen target holds LDR
# values directly. Without this, HDR values > 1.0 from lights get
# clamped to white when sampled as a texture, making every lit surface
# look pure white. Supplied per-SRU via the mesh push constant.
target.begin_pass(cmd)
renderer._scene_renderer.render_scene_content(cmd, hdr_output=0)
target.end_pass(cmd)
# Render 2D overlay. RTT-2D: submit the published item
# view (Camera2D honoured, Text2D carried) into the offscreen target.
# ``draw2d_ops`` is the alternate input the editor's play-mode game
# viewport supplies (an isolated op stream walked from the running game
# tree). The MSDF atlas must be uploaded before any glyph draws.
text_pass = getattr(renderer, "_text_pass", None)
if item_2d is not None and item_2d[0] is not None and hasattr(target, "render_items"):
if text_pass:
text_pass.upload_atlas_if_dirty()
view, camera = item_2d
target.render_items(cmd, view, camera)
elif draw2d_ops and hasattr(target, "render_draw2d"):
if text_pass:
text_pass.upload_atlas_if_dirty()
target.render_draw2d(cmd, draw2d_ops)
finally:
if occlusion is not None:
occlusion.finish(renderer)
renderer._instances = saved_instances
renderer._skinned_instances = saved_skinned
renderer._multimesh_blocks = saved_mm_blocks
renderer._multimesh_draws = saved_mm_draws
renderer._billboard_submissions = saved_billboards
renderer._first_instance_base = saved_base
renderer._sru_id = saved_sru
vpm._viewports = saved_viewports
vpm._next_id = saved_next_id
tree.play_viewport_rect = saved_vp
if camera is not None:
tree._render_camera_override = saved_cam
if saved_screen is not None:
tree._screen_size = saved_screen
[docs]
def render_sru_from_plan(self, cmd, sru) -> None:
"""Record one SubViewport SRU from an OWNED packet plan (pipelined render thread).
The render-thread counterpart of :meth:`render_to_target`: instead of
walking the live tree (illegal on the render thread, the main thread owns
it) it replays a :class:`~.renderer.render_packet.SubViewportSRU` snapshot
captured on the main thread by :meth:`SubViewportManager.build_srus`.
It isolates the renderer's main-scene per-frame instance lists + viewport
(already installed from the packet for the main pass), binds the SRU's
OWNED instances + single camera viewport, reserves a slice of the shared
transform SSBO, writes those transforms, records ``render_scene_content``
into the SRU's offscreen target with absolute ``first_instance`` indices,
overlays the SRU's isolated Draw2D ops, and restores the main-scene state.
The whole frame still ends in one ``vkQueueSubmit``; the offscreen colour
write is made visible to the main pass's later sample by the render
target's end-of-pass layout transition (same as ``render_to_target``).
"""
renderer = self._renderer
rend = sru.renderer
if rend is None or not rend.ready:
return
# D8 multi-GPU seam (pipelined replay): a secondary-assigned SRU renders on
# its device and transfers back instead of recording on this primary cmd.
# No coordinator (single-GPU / unopted) => today's primary recording below.
if self._maybe_offload_sru(sru.sru_id, sru):
return
vpm = renderer.viewport_manager
saved_instances = renderer._instances
saved_skinned = renderer._skinned_instances
saved_mm_blocks = renderer._multimesh_blocks
saved_mm_draws = renderer._multimesh_draws
saved_base = renderer._first_instance_base
saved_sru = renderer._sru_id
saved_viewports = dict(vpm._viewports)
saved_next_id = vpm._next_id
rend._clear_colour = list(sru.clear_colour)
# Install this SRU's OWN Sprite3D / Text3D billboards (carried in the plan)
# so they render into the offscreen target; the main scene's billboards are
# saved and restored so they do not bleed in. render_scene_content draws
# these against the SRU's single installed camera viewport below.
saved_billboards = renderer._billboard_submissions
renderer._billboard_submissions = list(sru.billboard_submissions)
# Copy the instance list: _prepare_multimesh may append expanded transparent
# MultiMesh instances, and the SRU plan's snapshot must not be mutated.
renderer._instances = list(sru.instances)
renderer._skinned_instances = sru.skinned_instances
renderer._multimesh_blocks = list(sru.multimesh_blocks)
renderer._multimesh_draws = []
try:
# Install the SRU's single camera viewport (3D SRUs). A 2D-only SRU
# has no camera matrices and renders only its Draw2D overlay.
vpm.clear()
if sru.camera_view is not None and sru.camera_proj is not None:
vpm.create_viewport(
0,
0,
sru.width,
sru.height,
sru.camera_view,
sru.camera_proj,
None,
)
# Resolve this SRU's MultiMesh blocks before reserving the instance slice
# so any ineligible (transparent) block expands into _instances and is
# counted/uploaded with the rest (mirrors render_to_target ordering).
renderer._prepare_multimesh()
n_slots = len(renderer._instances) + len(renderer._skinned_instances)
base = renderer._buffers.reserve_slots(n_slots)
renderer._first_instance_base = base
renderer._sru_id = sru.sru_id
renderer._buffers.upload_transforms(
renderer._instances,
upload_aabbs=False,
base=base,
)
# Upload eligible blocks into slots after this SRU's instance slice;
# builds _multimesh_draws that render_scene_content draws for this SRU.
renderer._upload_multimesh_blocks()
# hdr_output=0: tone-map in-shader so the offscreen target holds LDR.
rend.begin_pass(cmd)
renderer._scene_renderer.render_scene_content(cmd, hdr_output=0)
rend.end_pass(cmd)
# RTT-2D: submit the published item view (Camera2D +
# Text2D) into the offscreen target. The MSDF atlas must be uploaded
# before any glyph draws.
text_pass = getattr(renderer, "_text_pass", None)
if getattr(sru, "item_view", None) is not None and hasattr(rend, "render_items"):
if text_pass:
text_pass.upload_atlas_if_dirty()
rend.render_items(cmd, sru.item_view, sru.item_camera)
finally:
renderer._instances = saved_instances
renderer._skinned_instances = saved_skinned
renderer._multimesh_blocks = saved_mm_blocks
renderer._multimesh_draws = saved_mm_draws
renderer._billboard_submissions = saved_billboards
renderer._first_instance_base = saved_base
renderer._sru_id = saved_sru
vpm._viewports = saved_viewports
vpm._next_id = saved_next_id
[docs]
def snapshot_sru(
self,
tree: SceneTree,
*,
camera=None,
screen_size: tuple[float, float] | None = None,
) -> tuple[list, list, Any, Any, list, list]:
"""Submit an offscreen SRU's scene into isolated lists and return OWNED copies.
The CPU-only, no-GPU counterpart of :meth:`render_to_target`: it sets the
same tree overrides (viewport rect / camera / screen size), runs
``submit_scene`` into temporarily-isolated per-frame lists, then copies the
resulting opaque / skinned instances and the SRU camera's view + projection
matrices, and restores the renderer + tree state. No slice reservation, no
transform upload, no draw recording happens here: that is the render
thread's job from the returned plan. Used by
:meth:`SubViewportManager.build_srus` to packetise SubViewports.
Returns ``(instances, skinned_instances, camera_view, camera_proj,
multimesh_blocks, billboard_submissions)`` where the instance lists,
MultiMesh blocks, and billboard rows own copies of their transform / joint
arrays and the matrices are owned copies (or ``None`` for a 2D-only SRU
with no 3D camera viewport).
"""
renderer = self._renderer
saved_vp = tree.play_viewport_rect
saved_cam = getattr(tree, "_render_camera_override", None)
saved_screen = tree._screen_size if screen_size is not None else None
sx, sy = self._engine.content_scale
tw = float(screen_size[0]) if screen_size is not None else tree.screen_size[0]
th = float(screen_size[1]) if screen_size is not None else tree.screen_size[1]
tree.play_viewport_rect = (0, 0, tw / sx, th / sy)
if camera is not None:
tree._render_camera_override = camera
if screen_size is not None:
tree._screen_size = screen_size
vpm = renderer.viewport_manager
saved_instances = renderer._instances
saved_skinned = renderer._skinned_instances
saved_mm_blocks = renderer._multimesh_blocks
saved_viewports = dict(vpm._viewports)
saved_next_id = vpm._next_id
# Isolate billboards: capture this subtree's Sprite3D / Text3D rows the
# submit appends, then restore the main scene's billboard list untouched.
saved_billboards = renderer._billboard_submissions
renderer._instances = []
renderer._skinned_instances = []
renderer._multimesh_blocks = []
renderer._billboard_submissions = []
try:
self.submit_scene(tree, submit_lights=False)
instances = [(mh, t.copy(), mid, vid, rl) for (mh, t, mid, vid, rl) in renderer._instances]
skinned = [(mh, t.copy(), mid, j.copy()) for (mh, t, mid, j) in renderer._skinned_instances]
# MultiMesh blocks for this SRU's subtree (submit_scene populated them).
# Copy the transform / material-id arrays so the render thread owns them.
mm_blocks = [
(mh, t.copy(), mid, (m.copy() if m is not None else None), vid, key, ver)
for (mh, t, mid, m, vid, key, ver) in renderer._multimesh_blocks
]
# Sprite3D / Text3D billboard rows for this SRU's subtree: own copies so
# the render thread installs them into the offscreen target.
billboards = [b.copy() for b in renderer._billboard_submissions]
# submit_scene rebuilt viewport_manager with this SRU's single viewport;
# copy its camera matrices so the render thread reads a stable view/proj.
cam_view = cam_proj = None
vps = vpm.viewports
if vps:
_, vp = vps[0]
cam_view = np.array(vp.camera_view, dtype=np.float32, copy=True)
cam_proj = np.array(vp.camera_proj, dtype=np.float32, copy=True)
finally:
renderer._instances = saved_instances
renderer._skinned_instances = saved_skinned
renderer._multimesh_blocks = saved_mm_blocks
renderer._billboard_submissions = saved_billboards
vpm._viewports = saved_viewports
vpm._next_id = saved_next_id
tree.play_viewport_rect = saved_vp
if camera is not None:
tree._render_camera_override = saved_cam
if saved_screen is not None:
tree._screen_size = saved_screen
return instances, skinned, cam_view, cam_proj, mm_blocks, billboards