"""WorldEnvironment synchronisation and custom post-process orchestration."""
import logging
from typing import TYPE_CHECKING, Any
import vulkan as vk
from simvx.core.env_sync_spec import apply_spec
if TYPE_CHECKING:
from .forward import Renderer
from .post_process import PostProcessPass
__all__ = ["EnvironmentSync"]
log = logging.getLogger(__name__)
# Map logical ``subsystem`` names from ENV_SYNC_SPEC to renderer attribute
# accessors. Returning ``None`` means "subsystem not initialised: skip the
# write" (matches the previous ``if pp:`` / ``if r._ssao_pass:`` gating).
def _subsystem(renderer: Any, name: str) -> Any:
if name == "post_process":
return renderer._post_process
if name in ("ssao", "ssao_pass"):
# ``ssao`` carries the tuning knobs (radius/bias/intensity); the
# SsaoPass consumes them directly in its compute push-constants.
return renderer._ssao_pass
if name == "dof":
# DoF is parameterised on the PostProcessPass; the CoC→pixel
# conversion happens in the resolver before the write.
return renderer._post_process
if name == "shadow_pass":
return renderer._shadow_pass
if name == "volumetric_fog":
return renderer._volumetric_fog_pass
if name == "light2d":
# 2D light pass holds the ambient floor read by Draw2DPass.
return renderer._light2d_pass
if name == "taa_pass":
return renderer._taa_pass
if name == "renderer":
# Flags that live directly on the Renderer (not a post-process
# subsystem), e.g. the occlusion-culling gate.
return renderer
if name == "frame_globals":
# Wind/wetness inputs to the FrameGlobals UBO. Written onto
# the renderer's FrameGlobalsEnv holder; packed per frame in pre_render.
return renderer._frame_globals_env
return None
# Attribute name aliases for the rare cases where the renderer subsystem
# uses a different field name than ``WorldEnvironment``. Limit additions:
# every entry is a naming inconsistency we should eventually reconcile.
_VULKAN_ATTR_ALIAS = {
# ``WorldEnvironment.film_grain_*`` / web ``_film_grain_*`` /
# PostProcessPass ``grain_*``: the post_process attribute predates the
# ``film_grain_`` naming convention. Reconcile in a follow-up rename.
("post_process", "film_grain_enabled"): "grain_enabled",
("post_process", "film_grain_intensity"): "grain_intensity",
# PostProcessPass exposes the bloom soft-knee field on its inner
# ``_bloom_pass`` rather than as a top-level attribute.
("post_process", "bloom_soft_knee"): "_bloom_pass.soft_knee",
# Renderer-level flags use a private underscore attribute on the Renderer.
("renderer", "occlusion_culling_enabled"): "_occlusion_culling_enabled",
# Pluggable ambient tier: spec transform already mapped the
# string to the AMBIENT_MODE_* integer; the shadow packers read it from here.
("renderer", "ambient_mode"): "_ambient_mode",
# Debug view mode: spec transform already mapped the string
# to the DEBUG_VIEW_* integer; the shadow packers read it from ``_debug_view``.
("renderer", "debug_view"): "_debug_view",
# Reflection-probe quality dials: raw values land
# here; ``apply_probe_quality`` (composite step below) applies transitions.
("renderer", "probe_blend_count"): "_probe_blend_count",
("renderer", "probe_face_size"): "_probe_face_size",
}
[docs]
class EnvironmentSync:
"""Syncs WorldEnvironment node properties to renderer settings and
manages custom post-processing.
"""
def __init__(self, renderer: Renderer) -> None:
self._r = renderer
# Identity of the most recently installed ``WorldEnvironment.environment_map``
# value. Used to avoid re-running the IBL precompute when the env's
# property hasn't changed since the last sync.
self._last_env_map: Any = None
# Gradient-sky IBL: cache the synthesized cubemap handle by
# (top, bottom, size) so a static or repeating sky reuses the precompute.
self._gradient_key: Any = None
self._gradient_handle: Any = None
# Procedural-sky IBL: cache the synthesized Preetham
# cubemap handle by a quantised (sun, turbidity, ground_albedo, size)
# key so a sweeping sun re-bakes the cube + IBL only in discrete steps
# and a static procedural sky is synthesized exactly once.
self._procedural_key: Any = None
self._procedural_handle: Any = None
# Cache for ``tree.root.find(WorldEnvironment)`` / ``find(Camera3D)``.
# Both are full recursive walks that returned ``None`` every frame for
# pure-2D ports (~0.1 ms wasted). We invalidate the cache when the
# SceneTree's structure version bumps (add_child / remove_child /
# reparent all increment it).
self._cached_env: Any = None
self._cached_camera: Any = None
self._cache_version: int = -1
self._cache_tree: Any = None
# Per-CanvasLayer post: cached list of CanvasLayer nodes that
# carry an ``environment``, refreshed by the SAME structure-version watch.
# ``None`` until the first lookup; an empty list = no opted-in layer (the
# common case), so the per-frame cost is one ``if not layers``.
self._cached_post_layers: list[Any] | None = None
# Last HDR view bound into the tonemap descriptor. Used to skip the
# per-frame volumetric-fog descriptor swap: we only rewrite (and drain
# in-flight frames) when the target actually changes.
self._last_tonemap_hdr_views: tuple[Any, Any] | None = None
[docs]
@property
def cached_env(self) -> Any:
"""The ``WorldEnvironment`` node resolved by the last ``sync_world_environment``.
``None`` when the scene has no environment. Read by consumers that need
the env after the per-frame sync (e.g. the ocean pass wants its wind +
quality tier); refreshed on every structure change by the same cache."""
return self._cached_env
def _resolve_env_and_camera(self, tree: Any) -> tuple[Any, Any]:
"""Return (env, camera) for *tree*, using a structure-version cache.
``SceneTree._structure_version`` is bumped by ``add_child`` /
``remove_child`` before the new child's ``_enter_tree`` (or the
removed child's ``_exit_tree``) runs, so a cache hit on the same
version is always consistent with the current tree shape.
"""
from simvx.core.nodes_3d.camera import Camera3D
from simvx.core.world_environment import WorldEnvironment
version = getattr(tree, "_structure_version", None)
if self._cache_tree is tree and version is not None and version == self._cache_version:
return self._cached_env, self._cached_camera
env = tree.root.find(WorldEnvironment) if tree.root else None
cam = tree.root.find(Camera3D) if tree.root else None
self._cached_env = env
self._cached_camera = cam
self._cache_tree = tree
self._cache_version = version if version is not None else -1
# Structure changed: drop the cached opted-in-CanvasLayer list so the next
# layer-post pass re-walks (find_all) once, then reuses it per frame.
self._cached_post_layers = None
return env, cam
def _post_layers(self, tree: Any) -> list[Any]:
"""CanvasLayer nodes with an ``environment`` set, structure-version cached.
The expensive ``find_all`` walk runs only on a tree-structure change; on a
stable tree the cached list is reused. Empty for every scene that uses no
per-layer environment (the zero-cost-when-unused common case)."""
if self._cached_post_layers is not None:
return self._cached_post_layers
from simvx.core.nodes_2d.canvas import CanvasLayer
root = getattr(tree, "root", None)
layers = [n for n in root.find_all(CanvasLayer) if n.environment is not None] if root is not None else []
self._cached_post_layers = layers
return layers
[docs]
def invalidate_cache(self) -> None:
"""Drop any cached ``find()`` results: forces re-lookup on next sync.
Usually unnecessary because the ``_structure_version`` watch in
``_resolve_env_and_camera`` already catches add/remove/reparent. Exposed
for callers that mutate the tree without going through ``add_child`` /
``remove_child`` (e.g. test harnesses that swap root nodes wholesale).
"""
self._cached_env = None
self._cached_camera = None
self._cache_version = -1
self._cache_tree = None
self._cached_post_layers = None
def _sync_layer_post(self, tree: Any) -> None:
"""Populate ``renderer._layer_post_specs`` from opted-in CanvasLayers.
The whole feature is gated here: an empty opted-in-layer list (the common
case) clears the specs dict in O(1) and returns, so the renderer's per-frame
check is one ``if not self._layer_post_specs`` and it takes the exact
existing path. Only when a CanvasLayer actually carries an ``environment``
with an applicable effect does this build a per-band spec.
"""
from .layer_post import LayerPostSpec
r = self._r
layers = self._post_layers(tree)
if not layers:
if r._layer_post_specs:
r._layer_post_specs = {}
return
specs: dict[int, Any] = {}
for layer in layers:
spec = LayerPostSpec(layer.environment)
if spec.applies():
specs[int(layer.layer)] = spec
r._layer_post_specs = specs
[docs]
def sync_world_environment(self) -> None:
"""Sync WorldEnvironment node properties to renderer settings."""
r = self._r
# Whether a WorldEnvironment node exists this frame (pure-2D opt-in): with
# no env the renderer keeps its post setup defaults (pp.bloom_enabled stays
# True), so this flag is what distinguishes "no env" from "env that enabled
# bloom" for the 2D-only HDR-entry gate (forward._wants_2d_post).
r._has_world_env = False
tree = getattr(r._engine, "_scene_tree", None) or getattr(r._engine, "scene_tree", None)
if not tree or not tree.root:
return
env, camera = self._resolve_env_and_camera(tree)
# Per-CanvasLayer post is independent of the global WorldEnvironment node
# (a layer's env need not be in the tree), so collect it BEFORE the no-env
# early-out. Cheap: structure-version-cached + an empty-list fast path.
self._sync_layer_post(tree)
if not env:
return
r._has_world_env = True
# Spec-driven propagation. Composites and structurally divergent
# bridges (camera-exposure × tonemap-exposure, sky_mode → clear,
# custom colour-grading toggle) are handled below.
apply_spec(env, backend="vulkan", resolve=self._resolve_vulkan)
# Composite: tonemap exposure × camera exposure.
camera_exposure = float(camera.exposure) if camera is not None else 1.0
pp = r._post_process
if pp:
pp.exposure = camera_exposure * env.tonemap_exposure
# Screen-space reflections: the tier ``ssr_mode``
# ceiling can force SSR off regardless of the per-env toggle (auto/custom
# keep the engine default "half"). SSR needs the thin G-buffer, so fold it
# into the G-buffer activation below.
from simvx.core.graphics_quality import resolve_tier
_tier = resolve_tier(getattr(env, "quality_tier", "auto"))
_ssr_ceiling = _tier.ssr_mode if _tier is not None else "half"
ssr_active = bool(getattr(env, "ssr_enabled", False)) and _ssr_ceiling != "off"
# Screen-space global illumination: the tier
# ``ssgi_mode`` ceiling ("off" for every tier except ultra) can force it
# off; "auto"/"custom" leave it to the per-env toggle. SSGI shares SSR's
# thin G-buffer requirement, so it also folds into the activation below.
_ssgi_ceiling = _tier.ssgi_mode if _tier is not None else "on"
ssgi_active = bool(getattr(env, "ssgi_enabled", False)) and _ssgi_ceiling != "off"
# Thin G-buffer: activate the second HDR attachment (real
# normals for SSAO, world normal + roughness for SSR/SSGI) when a consumer
# asks for it. Idempotent + only rebuilds on a transition, so this
# per-frame call is free when stable. Requires the post-process HDR path
# (checked inside). SSR/SSGI are activated AFTER, so the G-buffer exists first.
want_gbuffer = (
(bool(getattr(env, "ssao_normals", False)) and bool(env.ssao_enabled)) or ssr_active or ssgi_active
)
if hasattr(r, "apply_gbuffer_state"):
r.apply_gbuffer_state(want_gbuffer)
if hasattr(r, "apply_ssr_state"):
r.apply_ssr_state(
ssr_active,
intensity=float(getattr(env, "ssr_intensity", 1.0)),
max_distance=float(getattr(env, "ssr_max_distance", 40.0)),
roughness_cutoff=float(getattr(env, "ssr_roughness_cutoff", 0.6)),
)
if hasattr(r, "apply_ssgi_state"):
r.apply_ssgi_state(
ssgi_active,
intensity=float(getattr(env, "ssgi_intensity", 1.0)),
max_distance=float(getattr(env, "ssgi_max_distance", 8.0)),
)
# Reflection-probe quality dials: the spec
# rows above wrote the raw values; apply the transitions (shader
# permutation rebuild / capture-target recreation). Idempotent, only
# acts on a change, so this per-frame call is free in steady state.
if hasattr(r, "apply_probe_quality"):
r.apply_probe_quality()
# Ambient: colour + energy reach cube_textured.frag via the shadow SSBO
# (shadow_renderer packs them into ambient_colour rgb/a). Energy scales
# the sky IBL ambient; colour is the flat fallback when no sky is set.
ac = env.ambient_light_colour
r._ambient_colour = (float(ac[0]), float(ac[1]), float(ac[2]))
r._ambient_energy = float(env.ambient_light_energy)
# Volumetric ↔ analytic fog are mutually exclusive: when the ray-march
# pass is active, tell the tonemap pass to suppress its analytic
# distance-fog branch (FLAG_VOLUMETRIC_FOG) so they don't double up.
# The tonemap HDR-input swap is decided here: before the command buffer
# records any descriptor binds, so we never update a bound set mid-frame.
vfog = r._volumetric_fog_pass
vfog_active = bool(env.volumetric_fog_enabled) and vfog is not None and vfog.enabled
taa = r._taa_pass
taa_active = bool(getattr(env, "taa_enabled", False)) and taa is not None and taa.enabled
if pp:
pp.volumetric_fog_active = vfog_active
hdr_target = getattr(pp, "hdr_target", None)
if hdr_target is not None:
# The HDR colour that emerges from the forward+fog stage. TAA (if
# active) consumes this as its current-frame input, and the
# tonemap consumes the TAA output instead. When TAA is off the
# tonemap reads this directly (the existing fog behaviour).
forward_hdr_view = vfog.output_view if vfog_active else hdr_target.colour_view
if taa_active:
# Point TAA's current-frame input (binding 0) at the
# forward+fog HDR. The tonemap's two per-parity sets sample
# the two TAA ping-pong targets (set i = TAA target i); the
# renderer binds the parity written each frame, so following
# the ping-pong needs no per-frame descriptor writes here.
taa.set_inputs(forward_hdr_view, hdr_target.depth_view)
tonemap_views = taa.target_views or (forward_hdr_view, forward_hdr_view)
# TAAU: the resolved output is at the
# OUTPUT extent, not the HDR chain's; tell the tonemap so
# its texel-step effects divide the bound input's size.
resolved_extent = taa.output_extent if getattr(taa, "is_upsampling", False) else None
else:
tonemap_views = (forward_hdr_view, forward_hdr_view)
resolved_extent = None
pp.tonemap_input_extent = resolved_extent
# Custom user post-process: the effect chain consumes the
# fog/TAA-resolved HDR (``tonemap_views``, per parity) and the
# tonemap then samples the chain's final output. Both the reroute
# and the chain's output view are resolved HERE, before recording,
# so no tonemap descriptor is rewritten while a frame is in flight
# (the chain output target is stable across frames).
pp.custom_pp_input_views = tonemap_views
custom_output = self._resolve_custom_post_process(pp, resolved_extent)
if custom_output is not None:
tonemap_views = (custom_output, custom_output)
# Only rewrite when the target views actually change (fog/TAA
# toggle, custom-pp added/removed, or resize): updating a
# descriptor still referenced by an in-flight frame is invalid,
# and a per-frame rewrite would do so.
if tonemap_views != self._last_tonemap_hdr_views:
# The views changed (fog/TAA toggled / resized). Drain in-flight
# frames so we don't update a descriptor still in use, then
# rewrite. Cheap because it happens only on a toggle.
vk.vkDeviceWaitIdle(self._r._engine.ctx.device)
self._update_tonemap_hdr_inputs(tonemap_views)
self._last_tonemap_hdr_views = tonemap_views
# Sky-mode bridge: colour-gradient skies drive the clear colour AND,
# when no explicit environment_map overrides, a synthesized gradient
# cubemap that feeds IBL ambient: parity with the web renderer (which
# always builds a gradient cubemap for colour skies).
if env.sky_mode == "colour":
c = env.sky_colour_top
if len(c) >= 4:
r._engine.clear_colour = [c[0], c[1], c[2], c[3]]
elif len(c) >= 3:
r._engine.clear_colour = [c[0], c[1], c[2], 1.0]
if env.environment_map is None:
self._sync_gradient_sky(env)
elif env.sky_mode == "procedural" and env.environment_map is None:
# Dynamic Preetham sky: synthesize a cube from the sun
# + turbidity + ground albedo and drive the skybox + IBL from it.
self._sync_procedural_sky(env, tree)
# Environment-map bridge: when the property changes, load the
# cubemap (Resource refs go through ``Engine.load_cubemap``) and
# hand it to ``Renderer.set_skybox`` which auto-runs the IBL
# precompute. ``None`` clears any prior install (no-op for the
# first sync since the renderer starts without IBL).
env_map = env.environment_map
if env_map is not self._last_env_map:
self._install_environment_map(env_map)
self._last_env_map = env_map
# Custom colour-grading effects opt in via WorldEnvironment.
cg = getattr(pp, "colour_grading", None) if pp else None
if cg and env.colour_grading_enabled:
cg.enabled = True
def _sync_gradient_sky(self, env: Any) -> None:
"""Synthesize + install a gradient cubemap for a colour sky, driving IBL.
Cached by ``(top, bottom, size)`` so an unchanging sky reuses the
precompute; only re-synthesizes when the gradient colours change.
"""
engine = self._r._engine
if not hasattr(engine, "load_cubemap"):
return
top = tuple(float(x) for x in env.sky_colour_top[:3])
bottom = tuple(float(x) for x in env.sky_colour_bottom[:3])
size = 64
key = (top, bottom, size)
if key == self._gradient_key:
return
from ..assets.cubemap_loader import gradient_cubemap_faces
faces = gradient_cubemap_faces(top, bottom, size)
handle = engine.load_cubemap(faces=faces)
self._gradient_key = key
self._gradient_handle = handle
self._r.set_skybox(handle)
def _find_sun_direction(self, tree: Any) -> tuple[float, float, float]:
"""Direction TOWARD the sun from the first ``DirectionalLight3D``.
A directional light's ``direction`` is its travel direction (pointing
away from the sun), so the sky's sun vector is its negation. Falls back
to a mid-morning sun when no directional light is present.
"""
from simvx.core.nodes_3d.lights import DirectionalLight3D
root = getattr(tree, "root", None)
light = root.find(DirectionalLight3D) if root is not None else None
if light is not None:
d = light.direction
return (-float(d[0]), -float(d[1]), -float(d[2]))
return (0.35, 0.5, 0.8)
def _sync_procedural_sky(self, env: Any, tree: Any) -> None:
"""Synthesize + install a Preetham sky cubemap, driving skybox + IBL.
Mirrors :meth:`_sync_gradient_sky`: CPU-synthesized float32 faces go
through ``Engine.load_cubemap`` + ``set_skybox`` (the existing IBL
precompute path). Cached by a quantised sun/turbidity/albedo key so a
sweeping sun re-bakes only in discrete steps and a static sky is baked
once (zero re-cost). The whole method is inert unless the env selected
``sky_mode="procedural"``, so the common path stays byte-identical.
"""
engine = self._r._engine
if not hasattr(engine, "load_cubemap"):
return
from simvx.core.procedural_sky import preetham_cubemap_faces, sky_cache_key
sun = self._find_sun_direction(tree)
turbidity = float(getattr(env, "sky_turbidity", 2.5))
albedo = tuple(float(x) for x in env.sky_ground_albedo[:3])
size = 64
key = sky_cache_key(sun, turbidity, albedo, size)
if key == self._procedural_key:
return
faces = preetham_cubemap_faces(sun, turbidity=turbidity, ground_albedo=albedo, size=size)
handle = engine.load_cubemap(faces=faces)
self._procedural_key = key
self._procedural_handle = handle
# A colourful horizon-ish clear behind the skybox in case it is skipped.
top = faces[2][0, 0, :3]
self._r._engine.clear_colour = [float(top[0]), float(top[1]), float(top[2]), 1.0]
self._r.set_skybox(handle)
def _install_environment_map(self, env_map: Any) -> None:
"""Resolve ``WorldEnvironment.environment_map`` to a cubemap handle
and hand it to the renderer. Supported value shapes:
* ``CubemapHandle``: installed directly.
* ``dict``: forwarded as kwargs to ``Engine.load_cubemap``
(e.g. ``{"colour": (r, g, b)}`` or ``{"paths_or_hdr": [...]}``).
* ``list[str]``: 6 face paths forwarded as ``paths_or_hdr=``.
Anything else logs a warning. Resource-style asset references will
slot in here once ``Engine.load_cubemap`` grows a Resource overload.
"""
if env_map is None:
return
engine = self._r._engine
if not hasattr(engine, "load_cubemap"):
return
from ..engine import CubemapHandle
if isinstance(env_map, CubemapHandle):
self._r.set_skybox(env_map)
return
if isinstance(env_map, dict):
handle = engine.load_cubemap(**env_map)
self._r.set_skybox(handle)
return
if isinstance(env_map, (list, tuple)) and len(env_map) == 6:
handle = engine.load_cubemap(list(env_map))
self._r.set_skybox(handle)
return
log.warning("environment_map: unsupported value type %r", type(env_map))
def _resolve_vulkan(self, subsystem: str, attr: str, value: Any) -> None:
"""Spec resolver: write ``value`` to the named subsystem field."""
target = _subsystem(self._r, subsystem)
if target is None:
return
# DoF: the canonical user knob is ``max_coc`` (max circle-of-confusion
# in UV units, i.e. fraction of screen width). The desktop tonemap DoF
# parameterises blur in *pixels* (``dof_max_blur``), and the shader
# multiplies the resulting CoC by ``texel_size`` (1/screen) before
# sampling, so a UV-space radius maps to pixels by × screen width.
# This keeps the blur disc resolution-independent and visually matched
# to the web backend (which works directly in UV).
if subsystem == "dof" and attr == "max_coc":
# Internal (HDR-chain) width: the DoF gather samples the HDR target,
# so the pixel radius is relative to it (== output width at scale 1).
w = self._r.internal_extent()[0]
target.dof_max_blur = float(value) * float(w)
return
# Apply naming alias if registered.
alias = _VULKAN_ATTR_ALIAS.get((subsystem, attr))
if alias is not None:
attr = alias
# Walk dotted attr (``_bloom_pass.soft_knee`` style).
parts = attr.split(".")
for seg in parts[:-1]:
target = getattr(target, seg, None)
if target is None:
return
setattr(target, parts[-1], value)
def _resolve_custom_post_process(self, pp: PostProcessPass, resolved_extent: tuple[int, int] | None) -> Any:
"""Sync custom effects and return the view the tonemap should sample, or None.
Called from ``sync_world_environment`` (pre-record). Sizes the effect
chain to the resolved (post-TAA) extent and reports its stable final
output view so the tonemap descriptor can be wired to it without any
mid-frame rewrite. The chain itself is recorded later by
``run_custom_post_process``.
"""
r = self._r
cpp = r._custom_pp
if cpp is None or not pp.hdr_target:
pp.custom_pp_active = False
return None
effects = self._gather_post_process_effects()
cpp.sync_effects(effects)
if not cpp.has_effects:
pp.custom_pp_active = False
return None
w = resolved_extent[0] if resolved_extent else pp.hdr_target.width
h = resolved_extent[1] if resolved_extent else pp.hdr_target.height
cpp.ensure_extent(w, h)
pp.custom_pp_active = True
pp.custom_pp_extent = (w, h)
return cpp.resolved_output_view()
[docs]
def run_custom_post_process(self, cmd: Any, pp: PostProcessPass) -> None:
"""Record the custom effect chain (resolved in ``sync_world_environment``).
Consumes the fog/TAA-resolved HDR for the parity written this frame and
renders into the ping-pong target the tonemap was already wired to, so no
descriptor is rewritten mid-frame.
"""
r = self._r
if not getattr(pp, "custom_pp_active", False) or r._custom_pp is None or not pp.hdr_target:
return
inputs = getattr(pp, "custom_pp_input_views", None)
if not inputs:
return
input_view = inputs[pp.tonemap_set_index]
w, h = getattr(pp, "custom_pp_extent", (pp.hdr_target.width, pp.hdr_target.height))
# Colour runs at the resolved extent; depth is the HDR chain's depth, which
# under TAAU upsampling (render_scale < 1) is the smaller internal extent.
# u_depth_tex samples by normalised UV so this is correct except that a
# depth-neighbourhood effect's 1/u_resolution texel step is in output space.
r._custom_pp.render(cmd, input_view, pp.hdr_target.depth_view, w, h)
def _gather_post_process_effects(self) -> list:
"""Collect PostProcessEffects from all WorldEnvironment nodes in the scene."""
from simvx.core.world_environment import WorldEnvironment
r = self._r
tree = getattr(r._engine, "_scene_tree", None) or getattr(r._engine, "scene_tree", None)
if not tree:
return []
root = getattr(tree, "root", None)
if not root:
return []
effects = []
for node in root.find_all(WorldEnvironment):
effects.extend(node.get_post_processes())
effects.sort(key=lambda e: e.order)
return effects
def _update_tonemap_hdr_inputs(self, new_hdr_views: tuple[Any, Any]) -> None:
"""Rewrite binding 0 of the tonemap's per-parity descriptor sets.
``new_hdr_views[i]`` goes into set ``i``: the two TAA ping-pong views
when TAA is active, or the same view twice otherwise (the sets stay
interchangeable so ``tonemap_set_index`` can be anything).
"""
r = self._r
pp = r._post_process
if not pp or not getattr(pp, "_descriptor_sets", None) or not pp._sampler:
return
writes = [
vk.VkWriteDescriptorSet(
dstSet=ds,
dstBinding=0,
dstArrayElement=0,
descriptorCount=1,
descriptorType=vk.VK_DESCRIPTOR_TYPE_COMBINED_IMAGE_SAMPLER,
pImageInfo=[
vk.VkDescriptorImageInfo(
sampler=pp._sampler,
imageView=view,
imageLayout=vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
)
],
)
for ds, view in zip(pp._descriptor_sets, new_hdr_views, strict=True)
]
vk.vkUpdateDescriptorSets(r._engine.ctx.device, len(writes), writes, 0, None)