"""The collection seam: walk a Node tree and emit render Items (design §2.2).
The collection pass of the build-once 2D pipeline (design P1.2): it walks a scene
tree and *produces* an :class:`~simvx.graphics.render2d.item_list.ItemList`. The
:class:`RenderItemCache` drives it each frame on the live render path.
What the collection pass is (design §2.2, §2.3, §2.5)
-----------------------------------------------------
The pass walks the tree once **in the same order the live ``_draw_recursive``
visits it** (depth-first, parent-before-children, siblings partitioned into
``below / self / above`` z-bands, CanvasLayers last by ``layer``, the YSort
y-order policy) and, per node, emits one or more Items into an ``ItemList``:
- ``seq`` -- a monotonic counter assigned **in emission order**, so a stable
``(layer, seq)`` sort of the produced list reproduces walk order exactly
(design §2.2: the walk folds per-sibling-group z into ``seq``; the global sort
is *not* a z lexsort -- P0 gate 9). ``z`` is recorded as data only.
- ``layer`` -- the CanvasLayer band (world content defaults to 0; a
``CanvasLayer(layer=N)`` shifts its whole subtree to band ``N``, §5.5).
- ``z`` -- ``absolute_z_index`` of the emitting node (data, never a sort input).
- ``clip_scope`` -- the active :class:`ClipScopeTable` scope, opened/closed around
clipped subtrees and the synthesised Control per-child wrap (§2.5).
- ``transform`` -- a per-item **LOCAL** transform row = the node's
``world_transform``, *not* composed with any camera (the view lives in a UBO,
§2.3). Camera-free verts + a local transform row is the keystone that lets a
later camera pan touch one UBO row instead of N item rows (Decision B).
- ``pipeline`` / ``blend`` / ``texture`` / ``flags`` / ``geometry`` -- taken from
the captured draw op.
The transitional op-adapter bridge (replaced in P3a)
----------------------------------------------------
Real nodes (Sprite2D, the shape primitives, Text2D's family) still draw through
the immediate-mode ``Draw2D`` op API in their ``on_draw`` bodies. Rewriting every
node to emit Items natively is P3a work. To stay isolated **and** make real nodes
produce Items today, this builder runs each node's ``on_draw`` against a
lightweight recording renderer (:class:`_OpRecorder`, the ``_DrawRecorder`` /
``DrawLog`` pattern) that captures the ops a node *would* draw -- **with an
identity transform so the captured geometry is camera-free and parent-free** --
then converts each captured op into one Item. The op's ``kind`` maps to a
:class:`PipelineKind`, its ``blend`` to a :class:`BlendMode`, its ``tex_id`` to a
texture slot, and its verts/indices are stowed in the :class:`GeometryStore`
behind a handle the Item references. This adapter is the seam P3a replaces with
native per-node item emission; the surrounding walk/sort/clip/transform machinery
stays.
Deferred (explicitly out of this increment):
- **P1.3** collapses the six live ``_draw_recursive`` variants into THIS one walk
(this builder reproduces their order so the collapse is a drop-in); today the
live walkers are untouched.
- **P1.4** adds dirty/retention on top of :class:`GeometryStore` (per-item reuse,
frame-level skip). Here the store is a plain handle list rebuilt each call.
"""
from __future__ import annotations
import math
from typing import Any, NamedTuple
from .clip_scope import ClipScopeTable
from .item_list import BlendMode, ItemFlags, ItemList, PipelineKind
__all__ = [
"CollectResult",
"Geometry",
"GeometryStore",
"ItemBuilder",
"NodeEntry",
"affine_row",
"build_item_list",
]
# A compact affine row matching draw2d's ``_xf`` layout: x' = a*x + b*y + tx.
_AffineRow = tuple[float, float, float, float, float, float]
_IDENTITY: _AffineRow = (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
# Op kind -> PipelineKind. The integer codes already agree (FILL=0, LINE=1,
# TEXT/GLYPH=2, TEX/TEXTURED=3, see item_list.PipelineKind), so this is a rename,
# not a remap.
_KIND_TO_PIPELINE = {
0: PipelineKind.FILL,
1: PipelineKind.LINE,
2: PipelineKind.GLYPH,
3: PipelineKind.TEXTURED,
}
_BLEND_TO_MODE = {
"alpha": BlendMode.ALPHA,
"add": BlendMode.ADD,
"multiply": BlendMode.MULTIPLY,
}
class _CapturedOp(NamedTuple):
"""One draw op captured from a node's ``on_draw`` (the bridge unit).
Mirrors ``draw2d_ops.Op`` but is local to this module so the builder does
not depend on the live op type's identity. ``clip`` is the recorder-relative
clip rect active at capture (folded into the ClipScopeTable, never baked into
geometry); ``verts``/``indices`` are camera-free, parent-free geometry.
"""
kind: int
clip: tuple[int, int, int, int] | None
verts: list[tuple]
indices: list[int] | None
tex_id: int
blend: str
is_msdf: bool
screen_space: bool = False
# Draw methods the recorder turns into ops. Mirrors ui.core._DRAW_METHODS but the
# recorder maps each to a concrete op rather than forwarding to a live renderer.
_TEXT_METHODS = frozenset({"draw_text", "draw_text_aligned"})
class _OpRecorder:
"""A camera-free recording renderer for the op-adapter bridge.
Implements the ``Draw2D`` drawing surface a node's ``on_draw`` calls, but
instead of baking the active camera into verts and appending to a global op
list, it records ops with **identity geometry** (no camera, no parent
transform) into :attr:`ops`. The captured geometry is therefore the node's
own local draw shape; the node's ``world_transform`` is recorded separately
by the builder as a per-item transform row (design §2.3).
It honours ``push_clip`` / ``pop_clip`` so a node that clips inside its own
``on_draw`` is reproduced, and exposes ``push_transform`` / ``push_identity``
/ ``pop_transform`` as no-op-on-geometry stack ops (the builder, not the
recorder, owns transform composition -- §2.5). Geometry stays local: a
``push_transform`` from the Control per-child wrap must NOT bake into verts
here, or the per-item transform row would double-apply it.
"""
__slots__ = ("ops", "_clip", "_clip_stack")
def __init__(self) -> None:
self.ops: list[_CapturedOp] = []
self._clip: tuple[int, int, int, int] | None = None
self._clip_stack: list[tuple[int, int, int, int] | None] = []
# -- clip stack (recorded, intersected like the live Draw2D clip) ---------
def push_clip(self, x: int, y: int, w: int, h: int) -> None:
new = (int(x), int(y), int(w), int(h))
if self._clip:
cx, cy, cw, ch = self._clip
nx, ny = max(cx, new[0]), max(cy, new[1])
nx2, ny2 = min(cx + cw, new[0] + new[2]), min(cy + ch, new[1] + new[3])
new = (nx, ny, max(0, nx2 - nx), max(0, ny2 - ny))
self._clip_stack.append(self._clip)
self._clip = new
def pop_clip(self) -> None:
self._clip = self._clip_stack.pop() if self._clip_stack else None
def reset_clip(self) -> None:
self._clip_stack.clear()
self._clip = None
# -- transform stack: accepted but geometry stays local (see class doc) ---
def push_transform(self, *_args) -> None: # noqa: D401 - no-op on geometry
pass
def push_identity(self) -> None:
pass
def pop_transform(self) -> None:
pass
# -- query passthrough ---------------------------------------------------
def text_width(self, text, scale=1.0, *_args, **_kwargs) -> float:
"""Real measured text width (the one builder), so widgets centre correctly.
A widget's ``on_draw`` queries ``renderer.text_width`` to centre/right-
align its own text (Button, LineEdit, ...). Returning 0 -- the P1.6 stub --
broke that under the item path (the Control coverage gap: button text
drifted off-centre). It must answer with the SAME measurement the glyph
layout uses, so flag-ON Control text lands pixel-identically to flag-OFF.
"""
from ..draw2d import Draw2D
return Draw2D.text_width(text, scale)
def text_height(self, text, scale=1.0, *_args, **_kwargs) -> float:
from ..draw2d import Draw2D
return Draw2D.text_height(text, scale)
def text_size(self, text, scale=1.0, *_args, **_kwargs):
from ..draw2d import Draw2D
return Draw2D.text_size(text, scale)
def fit_scale(self, *args, **kwargs) -> float:
from ..draw2d import Draw2D
return Draw2D.fit_scale(*args, **kwargs)
# -- primitive recording -------------------------------------------------
def _add(self, kind, verts, indices, tex_id=-1, blend="alpha", is_msdf=False, screen_space=False) -> None:
self.ops.append(
_CapturedOp(int(kind), self._clip, verts, indices, int(tex_id), blend, is_msdf, screen_space)
)
def draw_rect(self, pos, size, *, colour=None, filled=False, thickness=1.0, screen_space=False, blend="alpha"):
x, y = float(pos[0]), float(pos[1])
w, h = float(size[0]), float(size[1])
c = _norm_colour(colour)
if filled:
verts = [(x, y, 0, 0, *c), (x + w, y, 0, 0, *c), (x + w, y + h, 0, 0, *c), (x, y + h, 0, 0, *c)]
self._add(0, verts, [0, 1, 2, 0, 2, 3], blend=blend, screen_space=screen_space)
else:
self._add(1, _rect_outline_verts(x, y, w, h, c), None, screen_space=screen_space)
def draw_line(self, a, b, *, colour=None, thickness=1.0, screen_space=False):
c = _norm_colour(colour)
self._add(
1,
[(float(a[0]), float(a[1]), 0, 0, *c), (float(b[0]), float(b[1]), 0, 0, *c)],
None,
screen_space=screen_space,
)
def draw_lines(self, points, closed=True, colour=None):
c = _norm_colour(colour)
verts: list[tuple] = []
pts = [(float(p[0]), float(p[1])) for p in points]
for a, b in zip(pts, pts[1:], strict=False):
verts += [(a[0], a[1], 0, 0, *c), (b[0], b[1], 0, 0, *c)]
if closed and len(pts) > 2:
verts += [(pts[-1][0], pts[-1][1], 0, 0, *c), (pts[0][0], pts[0][1], 0, 0, *c)]
if verts:
self._add(1, verts, None)
def draw_circle(self, center, radius, *, colour=None, filled=False, segments=32, screen_space=False):
cx, cy, r = float(center[0]), float(center[1]), float(radius)
c = _norm_colour(colour)
step = math.tau / segments
pts = [(cx + math.cos(i * step) * r, cy + math.sin(i * step) * r) for i in range(segments)]
if filled:
verts = [(cx, cy, 0, 0, *c)] + [(px, py, 0, 0, *c) for px, py in pts]
idx = [j for i in range(segments) for j in (0, 1 + i, 1 + (i + 1) % segments)]
self._add(0, verts, idx, blend="alpha", screen_space=screen_space)
else:
verts = []
for i in range(segments):
a, b = pts[i], pts[(i + 1) % segments]
verts += [(a[0], a[1], 0, 0, *c), (b[0], b[1], 0, 0, *c)]
self._add(1, verts, None, screen_space=screen_space)
def draw_polygon(self, vertices, *, colour=None, filled=True, blend="alpha"):
c = _norm_colour(colour)
pts = [(float(v[0]), float(v[1])) for v in vertices]
if len(pts) < 3:
return
if filled:
# Same triangulation policy as Draw2D.draw_polygon: convex polygons hit
# the cheap fan from vertex 0, concave ones are ear-clipped so star/gear/
# multi-reflex shapes don't web across their notches. (Previously this
# recorder always fanned from vertex 0, which webbed every concave shape
# whose vertex 0 could not see the whole interior.)
from ..draw2d import _ear_clip_triangulate, _polygon_is_convex
verts = [(px, py, 0, 0, *c) for px, py in pts]
if _polygon_is_convex(pts):
idx = [j for i in range(1, len(pts) - 1) for j in (0, i, i + 1)]
else:
idx = _ear_clip_triangulate(pts)
self._add(0, verts, idx, blend=blend)
else:
verts = []
for a, b in zip(pts, pts[1:] + pts[:1], strict=True):
verts += [(a[0], a[1], 0, 0, *c), (b[0], b[1], 0, 0, *c)]
self._add(1, verts, None)
def fill_triangle(self, x1, y1, x2, y2, x3, y3, *, colour=None):
c = _norm_colour(colour)
self._add(0, [(x1, y1, 0, 0, *c), (x2, y2, 0, 0, *c), (x3, y3, 0, 0, *c)], [0, 1, 2])
def fill_quad(self, x1, y1, x2, y2, x3, y3, x4, y4, *, colour=None):
c = _norm_colour(colour)
verts = [(x1, y1, 0, 0, *c), (x2, y2, 0, 0, *c), (x3, y3, 0, 0, *c), (x4, y4, 0, 0, *c)]
self._add(0, verts, [0, 1, 2, 0, 2, 3])
def draw_thick_line(self, x1, y1, x2, y2, width=2.0, *, colour=None):
dx, dy = x2 - x1, y2 - y1
ln = math.hypot(dx, dy)
if ln < 1e-6:
return
hw = width * 0.5
px, py = -dy / ln * hw, dx / ln * hw
self.fill_quad(x1 + px, y1 + py, x2 + px, y2 + py, x2 - px, y2 - py, x1 - px, y1 - py, colour=colour)
def fill_rect_gradient(self, x, y, w, h, colour_top, colour_bottom):
ct, cb = _norm_colour(colour_top), _norm_colour(colour_bottom)
verts = [(x, y, 0, 0, *ct), (x + w, y, 0, 0, *ct), (x + w, y + h, 0, 0, *cb), (x, y + h, 0, 0, *cb)]
self._add(0, verts, [0, 1, 2, 0, 2, 3])
draw_gradient_rect = fill_rect_gradient
def draw_texture(self, texture_id, x, y, w, h, colour=None, rotation=0.0, *, blend="alpha", screen_space=False):
self.draw_texture_region(
texture_id, x, y, w, h, 0.0, 0.0, 1.0, 1.0, colour, rotation, blend=blend, screen_space=screen_space
)
def draw_image(self, path, x, y, w, h, *, colour=None, rotation=0.0, filter="linear", blend="alpha"):
# Mirror Draw2D.draw_image: resolve the path to a bindless tex_id (cached
# via the running App's TextureManager) and record a textured quad. Returns
# silently with no engine (CPU-only mode), exactly like the live method.
from ..draw2d import Draw2D
tex_id = Draw2D._resolve_path_texture(str(path), filter)
if tex_id < 0:
return
self.draw_texture_region(tex_id, x, y, w, h, 0.0, 0.0, 1.0, 1.0, colour, rotation, blend=blend)
def draw_texture_region(
self,
texture_id,
x,
y,
w,
h,
u0=0.0,
v0=0.0,
u1=1.0,
v1=1.0,
colour=None,
rotation=0.0,
*,
blend="alpha",
screen_space=False,
):
# Mirror Draw2D.draw_texture_region exactly: build the (rotated) quad
# corners in the node's own coordinate space (the node bakes its world
# position into x/y, as Sprite2D does), camera-free. The four UV corners
# follow the (u0, v0)-(u1, v1) region; rotation pivots around the quad
# centre. Geometry is captured local (no camera); the camera is applied
# uniformly at submit (design §2.3).
c = _norm_colour(colour)
x, y, w, h = float(x), float(y), float(w), float(h)
if abs(rotation) < 1e-6:
x0, y0, x1, y1 = x, y, x + w, y
x2, y2, x3, y3 = x + w, y + h, x, y + h
else:
cx, cy = x + w * 0.5, y + h * 0.5
cos_r, sin_r = math.cos(rotation), math.sin(rotation)
hw, hh = w * 0.5, h * 0.5
corners = [(-hw, -hh), (hw, -hh), (hw, hh), (-hw, hh)]
(x0, y0), (x1, y1), (x2, y2), (x3, y3) = [
(cx + px * cos_r - py * sin_r, cy + px * sin_r + py * cos_r) for px, py in corners
]
verts = [
(x0, y0, u0, v0, *c),
(x1, y1, u1, v0, *c),
(x2, y2, u1, v1, *c),
(x3, y3, u0, v1, *c),
]
self._add(3, verts, [0, 1, 2, 0, 2, 3], tex_id=texture_id, blend=blend, screen_space=screen_space)
def draw_text(
self,
text,
pos=None,
*,
colour=None,
scale=1.0,
rect=None,
alignment="left",
vertical_alignment="top",
fit_to_width=False,
min_scale=None,
outline=0.0,
outline_colour=None,
screen_space=False,
):
# Native GLYPH emission (P3a): run the ONE 2D text layout
# (:func:`layout_glyph_run`) to produce real kerned MSDF glyph quads in
# local space (the node bakes its world origin into ``pos`` like every
# other primitive, so the geometry is camera-/parent-free; design §2.3).
# The resulting indexed quad mesh becomes a single GLYPH item the item
# submitter draws through the existing (non-bindless) TEXT pipeline -- the
# bridge placeholder is gone. ``screen_space`` rides ``flags`` so the
# submit step can skip the camera for screen-pinned text (Text2D).
from ..draw2d_text import layout_glyph_run
verts, indices = layout_glyph_run(
text, pos, colour=colour, scale=scale, rect=rect,
alignment=alignment, vertical_alignment=vertical_alignment,
fit_to_width=fit_to_width, min_scale=min_scale,
outline=outline, outline_colour=outline_colour,
)
if verts:
self._add(2, verts, indices, is_msdf=True, screen_space=screen_space)
[docs]
def affine_row(node: Any) -> _AffineRow:
"""Return ``node``'s ``world_transform`` as a camera-free LOCAL affine row.
The compact ``(a, b, c, d, tx, ty)`` row draw2d's ``_xf`` uses
(``x' = a*x + b*y + tx``). Camera-free by construction: ``world_transform``
composes only the node's own + ancestor 2D transforms, never a Camera2D (the
view lives in a per-frame UBO, design §2.3), so a camera pan rewrites one UBO
row, not N item rows (Decision B). Nodes without a 2D transform (plain
``Node``, ``Control``, ``CanvasLayer``) record identity. Shared by the
collection walk and the P2 transform-only patch so both produce identical rows.
"""
wt = getattr(node, "world_transform", None)
if wt is None:
return _IDENTITY
pos, scale, rot = wt
c, s = math.cos(rot), math.sin(rot)
sx, sy = float(scale[0]), float(scale[1])
return (sx * c, -sy * s, sx * s, sy * c, float(pos[0]), float(pos[1]))
def _bake_affine(verts: list[tuple], affine: _AffineRow) -> list[tuple]:
"""Apply ``affine`` to the positions of captured 8-float verts (pos only).
Each vertex is ``(x, y, u, v, r, g, b, a)``; only ``(x, y)`` transform, exactly
as ``Draw2D._xf_pt`` bakes during the legacy walk. Used to compose a
CanvasLayer's ``offset/rotation/scale_val`` into its subtree's local geometry
(design §5.5, P4). The identity affine returns positions unchanged.
"""
a, b, c, d, tx, ty = affine
return [(a * v[0] + b * v[1] + tx, c * v[0] + d * v[1] + ty, *v[2:]) for v in verts]
def _norm_colour(c) -> tuple[float, float, float, float]:
if c is None:
return (1.0, 1.0, 1.0, 1.0)
if len(c) == 3:
return (float(c[0]), float(c[1]), float(c[2]), 1.0)
return (float(c[0]), float(c[1]), float(c[2]), float(c[3]))
def _rect_outline_verts(x, y, w, h, c):
p = [(x, y), (x + w, y), (x + w, y + h), (x, y + h)]
out: list[tuple] = []
for a, b in zip(p, p[1:] + p[:1], strict=True):
out += [(a[0], a[1], 0, 0, *c), (b[0], b[1], 0, 0, *c)]
return out
# ---------------------------------------------------------------------------
# Retained geometry store (seed of P1.4 retention)
# ---------------------------------------------------------------------------
[docs]
class Geometry(NamedTuple):
"""Captured vert/index arrays an Item's ``geometry`` handle resolves to.
The unit of the retained geometry store. ``verts`` are 8-float tuples
``(x, y, u, v, r, g, b, a)`` in the node's LOCAL space (camera-free,
parent-free); ``indices`` is ``None`` for line lists. P1.4 layers
dirty/reuse on top, keying handles by ``item_id`` so a clean item keeps its
slice; here every handle is freshly appended.
"""
verts: list[tuple]
indices: list[int] | None
[docs]
class GeometryStore:
"""A handle-indexed store of captured :class:`Geometry` (design §2.1).
The seed of the retained geometry store. Today it is a plain growable list:
:meth:`add` appends and returns the handle an Item records in its
``geometry`` column; :meth:`get` resolves it. P1.4 adds dirty tracking,
item_id keying and arena slices on top of this surface.
"""
__slots__ = ("_geoms",)
def __init__(self) -> None:
self._geoms: list[Geometry] = []
[docs]
def __len__(self) -> int:
return len(self._geoms)
[docs]
def add(self, verts: list[tuple], indices: list[int] | None) -> int:
handle = len(self._geoms)
self._geoms.append(Geometry(verts, indices))
return handle
[docs]
def get(self, handle: int) -> Geometry:
return self._geoms[handle]
[docs]
def set(self, handle: int, verts: list[tuple], indices: list[int] | None) -> None:
"""Overwrite the geometry behind an existing handle (P2 item-level patch).
Lets a render-dirty node re-capture its ``on_draw`` and replace its
geometry slice in place, keeping the handle stable so the Item column and
every consumer keying by handle stay valid (design §2.7 item granularity).
"""
if not 0 <= handle < len(self._geoms):
raise IndexError(f"geometry handle {handle} out of range")
self._geoms[handle] = Geometry(verts, indices)
[docs]
class NodeEntry(NamedTuple):
"""The retained per-node record an incremental patch (P2) locates items by.
Built during collection so a later per-item / transform-only update finds the
exact rows + transform slot + geometry handles a single node produced, without
a full re-walk (design §2.7 item / transform-only granularity).
Attributes
----------
transform_id
The node's row index in :attr:`CollectResult.transforms` (its LOCAL
affine). A transform-only change rewrites just this row.
rows
The physical :class:`ItemList` row indices the node's ops produced.
geometry
The geometry handles (parallel to ``rows``) the node's ops produced. A
render-dirty change re-captures the node's ``on_draw`` and overwrites
these in place.
canvas_affine
The enclosing CanvasLayer affine baked into this node's captured geometry
(``None`` for world content). A P2 in-place re-capture re-bakes with this
so the canvas offset/rotation/scale_val survives an item-level patch
(design §5.5, P4).
"""
transform_id: int
rows: list[int]
geometry: list[int]
canvas_affine: _AffineRow | None = None
[docs]
class CollectResult(NamedTuple):
"""The product of one :meth:`ItemBuilder.collect` (design §2.2).
Bundles the three artefacts the later sort/batch/submit consume: the SoA
:class:`ItemList`, the :class:`ClipScopeTable` its ``clip_scope`` indices
point into, the :class:`GeometryStore` its ``geometry`` handles resolve
against, and the per-item LOCAL transform rows its ``transform`` indices
select. ``node_index`` maps each emitting node (by ``id``) to its
:class:`NodeEntry`, the substrate for P2's in-place per-item / transform-only
patching (design §2.7).
"""
items: ItemList
clips: ClipScopeTable
geometry: GeometryStore
transforms: list[_AffineRow]
node_index: dict[int, NodeEntry]
# ---------------------------------------------------------------------------
# The collection walk
# ---------------------------------------------------------------------------
[docs]
class ItemBuilder:
"""Walks a Node tree and emits render Items into an :class:`ItemList`.
Reproduces the live ``_draw_recursive`` visit order (so a stable
``(layer, seq)`` sort of the result equals walk order, design §2.2) while
keeping geometry camera-free and recording per-item LOCAL transforms +
nested clip scopes. See the module docstring for the op-adapter bridge.
One builder is single-use per :meth:`collect`; construct a fresh one (or
just call :func:`build_item_list`) per collection.
"""
__slots__ = (
"_items", "_clips", "_geom", "_transforms", "_seq", "_node_index",
"_canvas_affine", "_screen_space", "_root",
)
def __init__(self) -> None:
self._items = ItemList()
self._clips = ClipScopeTable()
self._geom = GeometryStore()
self._transforms: list[_AffineRow] = []
self._seq = 0
self._node_index: dict[int, NodeEntry] = {}
# The node ``collect`` was asked to walk. The SubViewport prune (below)
# never fires for the root, so collecting a SubViewport's OWN 2D subtree
# (root == that SubViewport) descends; a nested SubViewport is pruned.
self._root: Any = None
# The active CanvasLayer bake state (design §5.5, P4): the affine the
# current layer composites its whole subtree through, and whether that
# subtree is screen-pinned (follow_viewport=False -> SCREEN_SPACE so the
# submit skips the camera). World content (no enclosing CanvasLayer) uses
# identity + camera-relative -- byte-identical to today.
self._canvas_affine: _AffineRow | None = None
self._screen_space = False
[docs]
def collect(self, root: Any, *, layer: int = 0) -> CollectResult:
"""Walk ``root`` and return the produced :class:`CollectResult`.
``layer`` is the starting CanvasLayer band (0 = world content). A
``CanvasLayer`` child shifts its subtree's band; nested CanvasLayers are
absolute (the layer does not add), matching §5.5.
"""
self._root = root
self._walk(root, layer)
return CollectResult(self._items, self._clips, self._geom, self._transforms, self._node_index)
# -- transform column ----------------------------------------------------
def _transform_row(self, node: Any) -> int:
"""Append the node's world_transform as a LOCAL affine row; return its id."""
tid = len(self._transforms)
self._transforms.append(affine_row(node))
return tid
# -- per-node emission (the op-adapter bridge) ---------------------------
def _emit_node(self, node: Any, layer: int) -> None:
"""Run ``node``'s ``on_draw`` through a recorder and convert ops to Items.
Each captured op becomes one Item carrying its kind->pipeline,
blend->mode, tex_id, the node's LOCAL transform row, the active clip
scope (intersected with any clip the node opened mid-``on_draw``), and a
geometry handle into the store. ``seq`` is the monotonic emission
counter so the stable ``(layer, seq)`` sort reproduces walk order.
"""
rec = _OpRecorder()
# Reuse the engine's ordered hook dispatch so override + decorated
# @on_draw handlers fire in the same order the live walk uses.
node._draw_dispatch(rec)
if not rec.ops:
return
z = int(getattr(node, "absolute_z_index", 0))
transform_id = self._transform_row(node)
base_scope = self._clips.current
rows: list[int] = []
handles: list[int] = []
for op in rec.ops:
# A clip opened inside on_draw nests under the active scope.
scope = base_scope if op.clip is None else self._clips.open(base_scope, op.clip)
# Under an enclosing CanvasLayer, bake the layer's affine into the
# captured (local) geometry so offset/rotation/scale_val apply -- the
# legacy path bakes it via Draw2D._xf during the walk; here the op-
# recorder is identity, so the builder bakes it (design §5.5, P4).
verts = op.verts if self._canvas_affine is None else _bake_affine(op.verts, self._canvas_affine)
geom = self._geom.add(verts, op.indices)
flags = ItemFlags.NONE
if op.is_msdf:
flags |= ItemFlags.IS_MSDF
# A screen-pinned CanvasLayer (follow_viewport=False) or a per-call
# screen_space=True both mean "skip the camera at submit".
if op.screen_space or self._screen_space:
flags |= ItemFlags.SCREEN_SPACE
# N1: the per-node ``hdr`` override forces the HDR lane in/out (None =
# by role). Resolved per node (all its ops share it).
flags |= hdr_flag(node)
row = self._items.add(
layer=layer,
z=z,
seq=self._seq,
pipeline=_KIND_TO_PIPELINE[op.kind],
clip_scope=scope,
blend=_BLEND_TO_MODE.get(op.blend, BlendMode.ALPHA),
transform=transform_id,
geometry=geom,
texture=op.tex_id,
flags=flags,
item_id=id(node) ^ (self._seq << 1),
)
rows.append(row)
handles.append(geom)
self._seq += 1
# Record the node's retained slot so a P2 in-place patch (transform-only
# or item-level) finds its rows/transform/geometry without a re-walk. The
# active canvas affine rides along so a re-capture re-bakes it (§5.5, P4).
self._node_index[id(node)] = NodeEntry(transform_id, rows, handles, self._canvas_affine)
# -- the walk (mirrors the live _draw_recursive variants, §7.1) ----------
def _walk(self, node: Any, layer: int) -> None:
"""Visit ``node`` in live ``_draw_recursive`` order.
Dispatches on node kind to reproduce each live walker:
``CanvasLayer`` (screen-space band shift), ``YSortContainer`` (y-order),
``Node2D`` (below/self/above z-bands + CanvasLayers last), ``Control``
(per-child clip+offset synthesis), and plain ``Node`` (self then
children, CanvasLayers last). P1.3 collapses these into this one method.
"""
if not getattr(node, "visible", True):
return
# Prune SubViewport subtrees from the MAIN walk (design §2.6 / §5.3):
# a SubViewport's children render into the viewport's OWN offscreen target
# (collected separately, per-target, with that viewport's Camera2D), so the
# main item collection must stop here -- mirrors the 3D-path
# ``find_all_outside_subviewports`` prune. Without it the SubViewport's 2D
# content double-draws into the main framebuffer (the gate-4 main-FB leak).
# The SubViewport node itself has no ``on_draw`` (zero items); we only skip
# descending into its children -- UNLESS the SubViewport IS the root we were
# asked to collect (the SubViewportManager collecting the viewport's OWN 2D
# subtree), exactly mirroring ``find_all_outside_subviewports``'s root rule.
if node is not self._root and getattr(node, "_is_subviewport", False):
return
# Script-error nodes skip their own dispatch but still walk children,
# matching every live _draw_recursive's _script_error branch.
if getattr(node, "_script_error", False):
for child in _children(node):
self._walk(child, layer)
return
# Order matters: Control exposes absolute_z_index/world_transform too, so
# it must be matched before the Node2D probe (it has its own walker --
# the per-child clip/offset synthesis, §2.5).
if getattr(node, "_is_canvas_layer", False):
self._walk_canvas_layer(node)
return
if _is_control(node):
self._walk_control(node, layer)
return
if _is_ysort(node):
self._walk_ysort(node, layer)
return
if _is_node2d(node):
self._walk_node2d(node, layer)
return
self._walk_plain(node, layer)
def _walk_plain(self, node: Any, layer: int) -> None:
"""``Node._draw_recursive``: self, then children, CanvasLayers last."""
self._emit_node(node, layer)
canvas_layers, others = _split_canvas_layers(_children(node))
below = [c for c in canvas_layers if c.layer < 0]
above = [c for c in canvas_layers if c.layer >= 0]
below.sort(key=lambda c: c.layer)
above.sort(key=lambda c: c.layer)
for c in below:
self._walk(c, c.layer)
for c in others:
self._walk(c, layer)
for c in above:
self._walk(c, c.layer)
def _walk_node2d(self, node: Any, layer: int) -> None:
"""``Node2D._draw_recursive``: below / self / above z-bands + layers last."""
children = _children(node)
canvas_layers, world = _split_canvas_layers(children)
below = [c for c in world if _is_node2d(c) and c.absolute_z_index < 0]
above = [c for c in world if c not in below]
below.sort(key=_z_key)
above.sort(key=_z_key)
canvas_layers.sort(key=lambda c: c.layer)
for c in below:
self._walk(c, layer)
self._emit_node(node, layer)
for c in above:
self._walk(c, layer)
for c in canvas_layers:
self._walk(c, c.layer)
def _walk_ysort(self, node: Any, layer: int) -> None:
"""``YSortContainer._draw_recursive``: self, then children by world/local Y."""
self._emit_node(node, layer)
children = sorted(_children(node), key=_y_key)
for c in children:
self._walk(c, layer)
def _walk_canvas_layer(self, node: Any) -> None:
"""``CanvasLayer._draw_recursive``: self + children in this layer's band (P4).
The band is the CanvasLayer's own ``layer`` (absolute, not additive --
§5.5; a nested CanvasLayer is its own world, replacing the parent's band
AND canvas affine, not composing). For the subtree's duration the builder:
- bakes the layer's ``offset/rotation/scale_val`` affine into every
captured (local) op so the canvas transform applies (the legacy walk
bakes it via ``Draw2D._xf``; the op-recorder is identity, so the builder
does it here -- design §5.5);
- marks the subtree ``SCREEN_SPACE`` when ``follow_viewport`` is False
(the screen-pinned HUD default) so the submit skips the camera.
``follow_viewport=True`` keeps the camera-relative path (no screen flag);
the canvas affine still bakes (the layer scrolls *with* the world plus its
own offset). The bake/screen state is saved + restored so siblings outside
the layer are unaffected (an unconfigured layer's identity affine + the
save/restore make activation byte-identical for existing scenes).
"""
band = int(node.layer)
affine = node._canvas_affine() if hasattr(node, "_canvas_affine") else None
follow = bool(getattr(node, "follow_viewport", False))
saved_affine, saved_screen = self._canvas_affine, self._screen_space
self._canvas_affine = affine
self._screen_space = not follow
try:
self._emit_node(node, band)
for c in _children(node):
self._walk(c, band)
finally:
self._canvas_affine, self._screen_space = saved_affine, saved_screen
def _walk_control(self, node: Any, layer: int) -> None:
"""``Control._draw_recursive`` + ``_draw_children_default`` (§2.5).
Reproduces the Control per-child wrap: a non-Control, non-overlay child
is drawn under a synthesised ClipScope + transform offset = the parent's
``get_global_rect()`` (a clip the walk synthesises, not declared by
nodes). Here we open a ClipScope for that rect so the child's items
record it; the offset rides the child's own world transform row.
"""
self._emit_node(node, layer)
rect = _global_rect(node)
for child in _children(node):
if _is_control(child) and getattr(child, "top_level", False):
continue
if _is_control(child) or getattr(child, "draw_overlay", False):
self._walk(child, layer)
elif rect is not None:
x, y, w, h = rect
self._clips.push((round(x), round(y), round(w), round(h)))
self._walk(child, layer)
self._clips.pop()
else:
self._walk(child, layer)
# ---------------------------------------------------------------------------
# Node-kind probes + child helpers (duck-typed: no core import at module load)
# ---------------------------------------------------------------------------
[docs]
def hdr_flag(node: Any) -> ItemFlags:
"""Resolve a node's ``hdr`` override to its HDR-lane flag (N1, 2D-in-HDR).
``hdr=True`` -> :attr:`ItemFlags.HDR_OPT_IN` (force the HDR lane), ``hdr=False``
-> :attr:`ItemFlags.HDR_OPT_OUT` (force the LDR lane), ``None``/absent ->
:attr:`ItemFlags.NONE` (by role: world->HDR, screen->LDR). Read per node and
applied to all its items; the submit lane logic (``submit.item_in_hdr_lane``)
combines this with the screen-space role. Resolved fresh in both the build walk
and the in-place re-capture patch, so toggling ``hdr`` takes effect next frame
without a full re-collect.
"""
h = getattr(node, "hdr", None)
if h is True:
return ItemFlags.HDR_OPT_IN
if h is False:
return ItemFlags.HDR_OPT_OUT
return ItemFlags.NONE
def _children(node: Any) -> list[Any]:
"""Children in tree order, using ``safe_iter`` when present (live-walk parity)."""
ch = getattr(node, "children", None)
if ch is None:
return []
safe = getattr(ch, "safe_iter", None)
return list(safe()) if safe is not None else list(ch)
def _split_canvas_layers(children: list[Any]) -> tuple[list[Any], list[Any]]:
"""Partition into (canvas_layers, others) preserving order."""
layers = [c for c in children if getattr(c, "_is_canvas_layer", False)]
others = [c for c in children if not getattr(c, "_is_canvas_layer", False)]
return layers, others
def _is_control(node: Any) -> bool:
return hasattr(node, "get_global_rect")
def _is_node2d(node: Any) -> bool:
# Control also carries absolute_z_index/world_transform, but the live z
# partition keys on isinstance(c, Node2D), which excludes Control; mirror that.
return hasattr(node, "absolute_z_index") and hasattr(node, "world_transform") and not _is_control(node)
def _is_ysort(node: Any) -> bool:
# YSortContainer is the Node2D whose draw order is its children sorted by Y;
# identify it structurally by its sorted-cache marker attribute.
return _is_node2d(node) and hasattr(node, "_sorted_cache")
def _z_key(c: Any) -> int:
return c.absolute_z_index if _is_node2d(c) else 0
def _y_key(c: Any) -> float:
pos = getattr(c, "position", None)
if pos is not None and hasattr(pos, "y"):
return float(pos.y)
return 0.0
def _global_rect(node: Any):
fn = getattr(node, "get_global_rect", None)
return fn() if fn is not None else None
[docs]
def build_item_list(root: Any, *, layer: int = 0) -> CollectResult:
"""Collect ``root`` into a :class:`CollectResult` (one-shot convenience).
Equivalent to ``ItemBuilder().collect(root, layer=layer)``. This is the
on-demand entry point for tests and the (default-OFF) flagged path; it does
not touch the live render path.
"""
return ItemBuilder().collect(root, layer=layer)