Source code for simvx.graphics.renderer.render_graph

"""RenderGraph: declarative pass scheduler.

Topologically sorts passes by declared input/output dependencies at compile() time.
Frame-time iteration is a tight loop over pre-computed tuples: no per-frame graph
traversal, no per-frame validation.
"""

import logging
from typing import Any

from simvx.core._topo_sort import CycleError, topo_sort

from .render_pass import FrameContext, RenderPass

__all__ = ["RenderGraph"]

log = logging.getLogger(__name__)




class RenderGraph:
    """Schedules render passes by declared dependencies.

    Usage::

        graph = RenderGraph()
        graph.add(shadow_pass)
        graph.add(ssao_pass)
        graph.compile()          # once at setup
        graph.record_pre_render(cmd, frame)  # every frame
        graph.record_render(cmd, frame)      # every frame
    """

    __slots__ = ("_passes", "_pre_render_order", "_render_order")

    def __init__(self) -> None:
        self._passes: list[RenderPass] = []
        self._pre_render_order: tuple[RenderPass, ...] = ()
        self._render_order: tuple[RenderPass, ...] = ()



    def add(self, pass_: RenderPass) -> None:
        """Register a pass. Call before compile()."""
        self._passes.append(pass_)




    def compile(self) -> None:
        """Topologically sort passes by input/output dependencies. Called once.

        Validates no cycles, no dangling inputs, no duplicate outputs.
        Splits result into pre_render and render tuples.
        """
        # Build output → producer mapping
        producers: dict[str, RenderPass] = {}
        for p in self._passes:
            for out in p.outputs:
                if out in producers:
                    raise ValueError(
                        f"Duplicate output '{out}': both '{producers[out].name}' " f"and '{p.name}' produce it"
                    )
                producers[out] = p

        # Build adjacency: pass → set of passes that must run before it
        deps: dict[str, set[str]] = {p.name: set() for p in self._passes}
        pass_by_name: dict[str, RenderPass] = {p.name: p for p in self._passes}

        for p in self._passes:
            for inp in p.inputs:
                if inp in producers:
                    deps[p.name].add(producers[inp].name)
                # Dangling inputs are allowed (resource may come from outside the graph,
                # e.g. previous frame's depth buffer).

        # Shared Kahn topological sort (simvx.core._topo_sort, also driving the
        # SceneTargetGraph). Passing the name-sorted node list keeps the
        # historical deterministic tiebreak (alphabetical by pass name). An
        # intra-scene pass cycle is always a bug, so CycleError is fatal here.
        try:
            ordered = topo_sort(sorted(deps), deps)
        except CycleError as exc:
            raise ValueError(f"Cycle detected in render graph involving: {set(exc.remaining)}") from exc

        # Split into pre_render and render tuples
        pre = []
        ren = []
        for name in ordered:
            p = pass_by_name[name]
            if p.stage == "pre_render":
                pre.append(p)
            else:
                ren.append(p)

        self._pre_render_order = tuple(pre)
        self._render_order = tuple(ren)

        log.debug(
            "Render graph compiled: pre_render=[%s], render=[%s]",
            ", ".join(p.name for p in self._pre_render_order),
            ", ".join(p.name for p in self._render_order),
        )




    def record_pre_render(self, cmd: Any, frame: FrameContext) -> None:
        """Record all pre-render passes in dependency order."""
        for p in self._pre_render_order:
            if p.enabled:
                p.record(cmd, frame)




    def record_render(self, cmd: Any, frame: FrameContext) -> None:
        """Record all render-stage passes in dependency order."""
        for p in self._render_order:
            if p.enabled:
                p.record(cmd, frame)




    def resize(self, width: int, height: int) -> None:
        """Propagate resize to all passes."""
        for p in self._passes:
            p.resize(width, height)




    def destroy(self) -> None:
        """Destroy all passes."""
        for p in self._passes:
            p.destroy()




    @property
    def passes(self) -> list[RenderPass]:
        """All registered passes (unordered)."""
        return self._passes




    @property
    def pre_render_order(self) -> tuple[RenderPass, ...]:
        """Compiled pre-render execution order."""
        return self._pre_render_order




    @property
    def render_order(self) -> tuple[RenderPass, ...]:
        """Compiled render execution order."""
        return self._render_order