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.
"""

from __future__ import annotations

import logging
from typing import Any

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).

        # Kahn's algorithm (topological sort)
        in_degree = {name: len(d) for name, d in deps.items()}
        # Seed with zero-degree nodes, sorted by name for deterministic order
        queue = sorted(name for name, deg in in_degree.items() if deg == 0)
        ordered: list[str] = []

        # Reverse adjacency for Kahn's
        dependents: dict[str, list[str]] = {name: [] for name in deps}
        for name, dep_set in deps.items():
            for dep_name in dep_set:
                dependents[dep_name].append(name)

        while queue:
            node = queue.pop(0)
            ordered.append(node)
            # Sort dependents by name for deterministic tiebreaking
            for dep in sorted(dependents[node]):
                in_degree[dep] -= 1
                if in_degree[dep] == 0:
                    queue.append(dep)
            queue.sort()  # maintain sorted order

        if len(ordered) != len(self._passes):
            missing = set(deps) - set(ordered)
            raise ValueError(f"Cycle detected in render graph involving: {missing}")

        # 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