"""PassOrchestrator — owns the forward renderer's pass lifecycle.
Extracted from Renderer so the 13 render-pass subsystems (shadow,
point-shadow, particle, GPU-particle-compute, tilemap, light2d, text,
draw2d, post-process, custom post-process, grid, gizmo, SSAO, fog,
skybox) live in one place with unified setup / resize / cleanup.
Renderer still owns frame-level orchestration (pre_render, render)
and calls into the orchestrator's ``render_overlays`` for the 2D overlay
pass after tonemap.
"""
import logging
from typing import TYPE_CHECKING, Any
import vulkan as vk
from .custom_post_process import CustomPostProcessPass
from .draw2d_pass import Draw2DPass
from .fog_pass import FogPass
from .gizmo_pass import GizmoPass, GizmoRenderData
from .grid_pass import GridPass
from .light2d_pass import Light2DPass
from .particle_pass import ParticlePass
from .post_process import PostProcessPass
from .skybox_pass import SkyboxPass
from .ssao_pass import SSAOPass
from .text_pass import TextPass
from .tilemap_pass import TileMapPass
if TYPE_CHECKING:
from .buffer_manager import BufferManager
from .pipeline_manager import PipelineManager
__all__ = ["PassOrchestrator"]
log = logging.getLogger(__name__)
[docs]
class PassOrchestrator:
"""Owns the lifecycle of every render pass used by the forward renderer."""
def __init__(self, engine: Any) -> None:
self._engine = engine
self.grid_pass: GridPass | None = None
self.shadow_pass: Any = None
self.point_shadow_pass: Any = None
self.particle_pass: ParticlePass | None = None
self.particle_compute: Any = None # lazy-init on first GPU particle submission
self.tilemap_pass: TileMapPass | None = None
self.light2d_pass: Light2DPass | None = None
self.text_pass: TextPass | None = None
self.draw2d_pass: Draw2DPass | None = None
self.post_process: PostProcessPass | None = None
self.custom_pp: CustomPostProcessPass | None = None
self.gizmo_pass: GizmoPass | None = None
self.ssao_pass: SSAOPass | None = None
self.fog_pass: FogPass | None = None
self.skybox_pass: SkyboxPass | None = None
self._skybox_owned: dict | None = None
self.gizmo_render_data: GizmoRenderData | None = None
# ------------------------------------------------------------------ setup
[docs]
def setup(self, ssbo_layout: Any, pipelines: PipelineManager) -> None:
"""Create and initialise every pass.
Order is load-bearing:
1. Grid, shadow, point-shadow, particles, tilemap, light2d come first
and are independent.
2. TextPass → Draw2DPass (Draw2D shares TextPass's atlas/pipeline).
3. PostProcess / CustomPostProcess.
4. Once post-process has built the HDR target, ``pipelines`` is
rebuilt against the HDR render pass so 3D draws inside that
pass match the HDR format.
5. Gizmo, SSAO, Fog — SSAO/Fog need the HDR depth views.
"""
e = self._engine
self.grid_pass = GridPass(e)
self.grid_pass.setup()
from .shadow_pass import ShadowPass
self.shadow_pass = ShadowPass(e)
self.shadow_pass.setup(ssbo_layout)
from .point_shadow_pass import PointShadowPass
self.point_shadow_pass = PointShadowPass(e)
self.point_shadow_pass.setup(ssbo_layout)
self.particle_pass = ParticlePass(e)
self.particle_pass.setup()
self.tilemap_pass = TileMapPass(e)
self.tilemap_pass.setup()
self.light2d_pass = Light2DPass(e)
self.light2d_pass.setup()
self.text_pass = TextPass(e)
self.text_pass.setup()
self.draw2d_pass = Draw2DPass(e, text_pass=self.text_pass, light2d_pass=self.light2d_pass)
self.draw2d_pass.setup()
self.post_process = PostProcessPass(e)
self.post_process.setup()
self.custom_pp = CustomPostProcessPass(e)
self.custom_pp.setup()
# Rebuild 3D pipelines against the HDR render pass when post-processing
# is active — the HDR target (R16G16B16A16_SFLOAT) has a different
# format than the default swapchain pass and pipelines must match.
if self.post_process.enabled and self.post_process.hdr_target:
hdr_rp = self.post_process.hdr_target.render_pass
pipelines.rebuild_for_render_pass(hdr_rp)
if self.particle_pass:
self.particle_pass.rebuild_pipeline(hdr_rp)
self.gizmo_pass = GizmoPass(e)
self.gizmo_pass.setup()
if self.post_process.enabled and self.post_process.hdr_target:
try:
hdr_rt = self.post_process.hdr_target
w, h = e.extent
self.ssao_pass = SSAOPass(e)
self.ssao_pass.setup(w, h, hdr_rt.depth_view, hdr_rt.depth_image)
self.post_process.update_ssao_descriptor(self.ssao_pass.ao_view)
except Exception as exc:
log.warning("SSAO disabled: %s", exc)
self.ssao_pass = None
try:
hdr_rt = self.post_process.hdr_target
w, h = e.extent
self.fog_pass = FogPass(e)
self.fog_pass.setup(w, h, hdr_rt.depth_view, hdr_rt.colour_view, hdr_rt.colour_image)
except Exception as exc:
log.warning("Fog pass disabled: %s", exc)
self.fog_pass = None
# ---------------------------------------------------------------- skybox
[docs]
def set_skybox(self, cubemap_view: Any, cubemap_sampler: Any,
buffers: BufferManager,
cubemap_image: Any = None, cubemap_memory: Any = None) -> None:
"""Set a cubemap as the skybox and enable IBL.
Takes ownership of ``cubemap_image`` / ``cubemap_memory`` when provided
so they can be destroyed at shutdown.
"""
self.skybox_pass = SkyboxPass(self._engine)
# When post-processing is active, the skybox draws inside the HDR
# render pass (R16G16B16A16_SFLOAT). Pipelines compiled for the default
# swapchain pass (B8G8R8A8_SRGB) would trigger render-pass format
# incompatibility warnings at every draw — use the HDR pass.
hdr_rp = None
if self.post_process and self.post_process.hdr_target:
hdr_rp = self.post_process.hdr_target.render_pass
self.skybox_pass.setup(cubemap_view, cubemap_sampler, render_pass=hdr_rp)
self._skybox_owned = {
"view": cubemap_view,
"sampler": cubemap_sampler,
"image": cubemap_image,
"memory": cubemap_memory,
}
buffers.write_cubemap_descriptor(cubemap_view, cubemap_sampler)
def _destroy_skybox_resources(self) -> None:
"""Destroy cubemap resources accepted via set_skybox. Safe to call twice."""
owned = self._skybox_owned
if not owned:
return
device = self._engine.ctx.device
if owned.get("view"):
vk.vkDestroyImageView(device, owned["view"], None)
if owned.get("sampler"):
vk.vkDestroySampler(device, owned["sampler"], None)
if owned.get("image"):
vk.vkDestroyImage(device, owned["image"], None)
if owned.get("memory"):
vk.vkFreeMemory(device, owned["memory"], None)
self._skybox_owned = None
# ---------------------------------------------------------------- resize
[docs]
def resize(self, width: int, height: int) -> None:
"""Resize passes that track the framebuffer size. Returns the render
pass that 3D pipelines should be compiled against after resize."""
if self.post_process and self.post_process.enabled:
self.post_process.resize(width, height)
if self.ssao_pass and self.post_process.hdr_target:
hdr_rt = self.post_process.hdr_target
self.ssao_pass.resize(width, height, hdr_rt.depth_view, hdr_rt.depth_image)
self.post_process.update_ssao_descriptor(self.ssao_pass.ao_view)
if self.fog_pass and self.post_process.hdr_target:
hdr_rt = self.post_process.hdr_target
self.fog_pass.resize(width, height, hdr_rt.depth_view, hdr_rt.colour_view, hdr_rt.colour_image)
if self.custom_pp:
self.custom_pp.resize(width, height)
[docs]
def pipeline_render_pass(self) -> Any:
"""Return the render pass that 3D pipelines should be compiled against."""
pp = self.post_process
if pp and pp.enabled and pp.hdr_target:
return pp.hdr_target.render_pass
return self._engine.render_pass
# ------------------------------------------------------------ per-frame
[docs]
def begin_frame(self) -> None:
"""Reset per-frame state on passes that buffer submissions."""
if self.tilemap_pass:
self.tilemap_pass.begin_frame()
if self.light2d_pass:
self.light2d_pass.begin_frame()
# --------------------------------------------------------------- cleanup
[docs]
def cleanup(self) -> None:
"""Destroy every pass and the skybox cubemap ownership."""
# Destroy skybox cubemap before the device goes down.
self._destroy_skybox_resources()
if self.skybox_pass:
self.skybox_pass.cleanup()
for pass_obj in (
self.gizmo_pass, self.shadow_pass, self.point_shadow_pass,
self.particle_pass, self.particle_compute, self.tilemap_pass,
self.ssao_pass, self.fog_pass, self.custom_pp, self.post_process,
self.grid_pass, self.light2d_pass, self.draw2d_pass, self.text_pass,
):
if pass_obj:
pass_obj.cleanup()