"""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
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",
}
[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
# 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
# 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_view: Any = None
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
return env, cam
[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
[docs]
def sync_world_environment(self) -> None:
"""Sync WorldEnvironment node properties to renderer settings."""
r = self._r
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)
if not env:
return
# 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
# 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
if pp:
pp.volumetric_fog_active = vfog_active
hdr_target = getattr(pp, "hdr_target", None)
if hdr_target is not None:
hdr_view = vfog.output_view if vfog_active else hdr_target.colour_view
# Only rewrite when the target view actually changes (fog toggle
# or resize): updating a descriptor still referenced by an
# in-flight frame is invalid, and a per-frame rewrite would do so.
if hdr_view is not self._last_tonemap_hdr_view:
# The view changed (fog 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_input(hdr_view)
self._last_tonemap_hdr_view = hdr_view
# 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)
# 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 _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":
w = self._r._engine.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)
[docs]
def run_custom_post_process(self, cmd: Any, pp: PostProcessPass) -> None:
"""Execute custom user effects between built-in post-processing and tonemap."""
r = self._r
if not r._custom_pp or not pp.hdr_target:
return
# Collect effects from WorldEnvironment nodes in the scene tree
effects = self._gather_post_process_effects()
if not effects:
return
r._custom_pp.sync_effects(effects)
if not r._custom_pp.has_effects:
return
w, h = r._engine.extent
hdr_view = pp.hdr_target.colour_view
depth_view = pp.hdr_target.depth_view
result_view = r._custom_pp.render(cmd, hdr_view, depth_view, w, h)
# If custom effects produced output, update tonemap's HDR input descriptor
if result_view and result_view != hdr_view:
self._update_tonemap_hdr_input(result_view)
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_input(self, new_hdr_view: Any) -> None:
"""Rewrite the tonemap descriptor set binding 0 to point at the custom output."""
r = self._r
pp = r._post_process
if not pp or not pp._descriptor_set or not pp._sampler:
return
hdr_info = vk.VkDescriptorImageInfo(
sampler=pp._sampler, imageView=new_hdr_view,
imageLayout=vk.VK_IMAGE_LAYOUT_SHADER_READ_ONLY_OPTIMAL,
)
vk.vkUpdateDescriptorSets(r._engine.ctx.device, 1, [vk.VkWriteDescriptorSet(
dstSet=pp._descriptor_set, dstBinding=0, dstArrayElement=0,
descriptorCount=1, descriptorType=vk.VK_DESCRIPTOR_TYPE_COMBINED_IMAGE_SAMPLER,
pImageInfo=[hdr_info],
)], 0, None)