"""SceneTree: Central manager for the node tree, groups, input routing, and UI focus."""
import logging
import math
import weakref
from collections.abc import Callable
from contextvars import Token
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
import numpy as np
from .event_bus import EventBus
from .events import InputEvent, TreeInputEvent
if TYPE_CHECKING:
from .audio_listener import AudioListener2D, AudioListener3D
from .audio_protocol import (
AudioBackend,
AudioBusBackend,
AudioPlaybackBackend,
AudioStreamingBackend,
)
from .input.map import _InputMap
from .input.state import _Input
from .physics.world import BodyHandle, PhysicsWorld
from .physics.world2d import Physics2DWorld
# Either dimensionality of world. The tree's per-world registries and the
# per-world step helpers are dimension-agnostic: they key on the world object
# and read widths off it, so both kinds travel the same path.
#
# A plain assignment rather than a PEP 695 ``type`` statement, which parso
# cannot parse and every file here is gated on parsing. See AGENTS.md.
AnyPhysicsWorld = PhysicsWorld | Physics2DWorld
from .input.enums import JoyAxis, JoyButton, Key, MouseButton
from .input.map import _active_input_map
from .input.state import _active_input
from .math.raycast import ray_intersect_sphere, screen_to_ray
from .node import Node
from .nodes_2d.camera import Camera2D
from .nodes_3d.camera import Camera3D
from .physics.nodes import CollisionShape3D
from .signals import Signal
# The "events" singleton name is reserved for the engine-provided EventBus.
# Project TOMLs and runtime code may not declare or replace it.
RESERVED_SINGLETON_EVENTS = "events"
log = logging.getLogger(__name__)
class _InputSpan:
"""Makes one tree's ``Input`` and ``InputMap`` the active pair for a block.
Returned by :meth:`SceneTree.activate_input`, one fresh instance per ``with``
so that nested spans, and spans opened on the same tree from two threads,
each keep their own reset tokens.
Entering is on the hot path: it runs once per tick, once per physics step,
once per mount and once per dispatched input event. So the contextvars are
written only when they do not already hold the exact objects the span would
install, which is the whole cost for a tree that shares the process-wide
default pair and for any span nested inside another on the same tree. The
depth counter is bumped unconditionally either way: it is what tells node
code the engine is running it, and skipping it would silently disarm the
late-registration warning and the mount/unmount fast paths.
"""
__slots__ = ("_tree", "_input_token", "_map_token")
def __init__(self, tree: SceneTree):
self._tree = tree
self._input_token: Token[_Input] | None = None
self._map_token: Token[_InputMap] | None = None
def __enter__(self) -> _InputSpan:
tree = self._tree
own_input = tree._own_input
if _active_input.get() is not own_input:
self._input_token = _active_input.set(own_input)
own_map = tree._own_input_map
if _active_input_map.get() is not own_map:
self._map_token = _active_input_map.set(own_map)
tree._input_span_depth += 1
return self
def __exit__(self, exc_type, exc, tb) -> None:
self._tree._input_span_depth -= 1
if self._map_token is not None:
_active_input_map.reset(self._map_token)
self._map_token = None
if self._input_token is not None:
_active_input.reset(self._input_token)
self._input_token = None
@dataclass(slots=True)
class _BodySync:
"""Per-world handle->node sync state connecting physics bodies to their nodes.
Drives the bulk transform read-back: ``nodes`` is a weak handle->body map
(a dead node auto-drops, never leaks or pins the tree); ``order`` + ``buf``
are rebuilt only when membership changes (``dirty``), so a world with stable
membership pays one bulk ``read_transforms`` per frame with zero allocation.
``prev_buf``/``cur_buf`` hold the last two STEP states (never render states);
render-time :meth:`SceneTree.interpolate_physics` blends between them by the
fixed-step accumulator fraction. ``seeded`` is reset on a membership rebuild
so a freshly added body re-seeds ``prev == cur`` and never lerps from origin.
``width``/``dims`` make the bulk plumbing dimension-aware: 3D worlds use a
``(N, 7)`` transform row (pos xyz + quat xyzw) with ``dims == 3``; 2D worlds
use ``(N, 4)`` (pos xy + cos/sin) with ``dims == 2``. Both are set from the
world's type at the first registration (see :meth:`SceneTree._world_dims`) so
the scatter / capture / interpolate leaves branch on ``dims`` only.
"""
nodes: weakref.WeakValueDictionary[BodyHandle, Node] = field(default_factory=weakref.WeakValueDictionary)
order: list[BodyHandle] = field(default_factory=list)
buf: np.ndarray | None = None # scratch target for read_transforms
prev_buf: np.ndarray | None = None # (N,width) f32: step state at t-1 (render interpolation)
cur_buf: np.ndarray | None = None # (N,width) f32: step state at t
dirty: bool = False
seeded: bool = False # False until prev/cur first populated (re-seed on membership change)
width: int = 7 # bulk transform row width: 7 for 3D (pos+quat), 4 for 2D (pos+cos/sin)
dims: int = 3 # spatial dimensions: 3 (3D) or 2 (2D); selects the scatter/interp branch
def _match_mods(mods: tuple, event: TreeInputEvent) -> bool:
"""Match a filter's (ctrl, shift, alt, meta) spec against an event's modifier state.
Each spec entry is True (must be pressed), False (must not be pressed),
or None (don't care).
"""
cs, ss, als, ms = mods
if cs is not None and event.ctrl != cs:
return False
if ss is not None and event.shift != ss:
return False
if als is not None and event.alt != als:
return False
if ms is not None and event.meta != ms:
return False
return True
[docs]
class SceneTreeTimer:
"""A transient one-shot timer owned by the :class:`SceneTree`.
Created via :meth:`SceneTree.create_timer`. Emits ``timeout`` once after its
delay elapses, then the tree drops it. Lighter than a :class:`~simvx.core.Timer`
node: it lives only on the tree's timer list, not in the scene graph.
"""
__slots__ = ("timeout", "_time_left")
def __init__(self, seconds: float):
self.timeout = Signal()
self._time_left = float(seconds)
[docs]
@property
def time_left(self) -> float:
return self._time_left
def _tick(self, dt: float) -> bool:
"""Advance by ``dt``; return ``True`` once the timer has fired."""
self._time_left -= dt
if self._time_left <= 0.0:
self.timeout()
return True
return False
class _TreeRoot(Node):
"""A tree's own top node: parent of every singleton and of the current scene.
Not a scene node. It is created with the tree, never enters or leaves it, is
never drawn, never serialised and never reaches the editor. Its only job is
to give the tree exactly one walk source, so that a node registered as a
singleton and a node parented into the scene are the same kind of thing and
neither can be processed twice.
It must stay a plain :class:`Node`. A ``Node2D``, ``Node3D`` or ``Control``
here would enter every world transform and every global rect in the engine,
so subclassing it further, or giving it a ``position``, moves every pixel the
engine draws.
Its ``update_mode`` must stay ``INHERIT``. An unparented node resolves
``INHERIT`` to ``PAUSABLE``, which is what a singleton and a scene root
resolved to before this node existed; ``DISABLED`` would poison every
``INHERIT`` descendant and ``ALWAYS`` would break pausing.
It holds its tree strongly (``_tree``) and the tree holds it, so the pair is
a refcount cycle from construction. A teardown that promises refcount-only
reclamation owes the two lines that break it: clear this node's children and
set its ``_tree`` to ``None`` (see ``UITestHarness.teardown``).
"""
_is_tree_root = True
def _detach_child(self, node: Node) -> None:
# Unlinking the current scene is the only way ``tree.root`` can stop
# being a child of this node, so clearing the tree's reference here is
# what keeps "the scene is one of my children" true rather than hoped
# for. Every unlink funnels through here: ``remove_child`` calls it, the
# delete-queue drain calls it, and ``add_child`` reaches it through the
# reparent-away.
was_child = node.parent is self
super()._detach_child(node)
tree = self._tree
if was_child and tree is not None and tree._scene_root is node:
tree._scene_root = None
[docs]
class SceneTree:
"""Central manager for the node tree, groups, input routing, and UI focus.
Owns the root node and drives per-frame ``process`` / ``physics_process`` /
``draw`` traversals. Also manages pause state, scene changes, and the UI
input pipeline (mouse, keyboard, popups).
"""
# Most-recently-active tree, set on every ``set_root``/``change_scene``.
# InputMap reads it (best-effort) so ``add_action`` can warn when called
# after the first tick: a common mistake that silently drops bindings in
# the web exporter (which never invokes ``main()``). Held as a weakref so
# the last tree of a session does not pin its whole node graph in memory:
# the tree owns its nodes, nothing owns the tree, so a strong class-level
# ref is the one thing that survives ``gc.collect()`` once every external
# reference is dropped. ``current()`` then correctly reports "no active
# tree" once the tree is gone, which all consumers already handle.
_active_tree: weakref.ref[SceneTree] | None = None
[docs]
@classmethod
def current(cls) -> SceneTree | None:
"""Return the most recently activated SceneTree, or ``None``.
Activation happens automatically on ``set_root`` / ``change_scene``.
Used by ``InputSimulator`` to deliver scene-tree + UI-tree events
without requiring callers to thread a tree reference through.
Held weakly: returns ``None`` once the tree has no remaining strong
references (e.g. a test dropped it), which is the correct "no active
tree" semantics.
"""
ref = cls._active_tree
return ref() if ref is not None else None
[docs]
@property
def app(self):
"""The App instance running this tree (set by graphics backend)."""
return getattr(self, "_app", None)
[docs]
@property
def events(self) -> EventBus:
"""Engine-provided typed event bus.
Use ``tree.events.subscribe(EventCls, handler)`` to register a
handler and ``tree.events.publish(event)`` (or
``publish_deferred``) to dispatch. The bus survives
``change_scene()`` -- subscriptions held by singletons or other
long-lived objects keep firing across scene swaps. Deferred
events queued during a frame are dispatched at the start of the
next ``process()`` tick, before any node ``_process`` runs.
"""
return self._events
[docs]
@property
def audio_backend(self) -> AudioBackend | None:
"""The active audio backend, or ``None`` if none was initialised.
Returns the union :class:`AudioBackend` type for backwards
compatibility: callers that only need one facet should prefer the
narrowed :attr:`audio_playback` / :attr:`audio_streaming` /
:attr:`audio_buses` properties so the type checker enforces the
boundary. Reaching for the underscore-prefixed attribute is
engine-private and may change.
The engine sets this during ``App.run`` via ``make_backend``. Tests
and headless harnesses that don't initialise audio see ``None``.
"""
return getattr(self, "_audio_backend", None)
[docs]
def install_audio_backend(self, backend: AudioPlaybackBackend) -> None:
"""Install the audio backend for this tree: the one canonical path.
This is the only supported way to attach a backend; assigning the
private ``_audio_backend`` slot is no longer part of the contract.
Used by ``App.run`` / ``WebApp`` at startup and by tests that inject a
:class:`NullAudioBackend` or a mock.
Semantics mirror startup driver selection in other engines (Godot's
``--audio-driver``, pyglet's ``audio`` option): a backend is installed
once, before the tree starts producing sound. Re-installing shuts the
previous backend down first so device handles don't leak.
Facet conformance is enforced where it is consumed, not here: the
:attr:`audio_playback` / :attr:`audio_streaming` / :attr:`audio_buses`
accessors narrow via ``isinstance`` and return ``None`` for a backend
that doesn't implement that facet, and callers raise
:class:`AudioCapabilityError` on ``None``. Gating here as well would
duplicate that boundary and reject legitimate partial backends.
A backend that paces itself by the engine rather than by a device
clock declares :class:`~simvx.core.audio_protocol.TreeBoundAudioBackend`,
and installing hands it this tree: several trees can be alive at once,
so it has to be told which one's ticks are its own.
"""
if backend is None:
raise TypeError("install_audio_backend requires a backend, got None")
previous = getattr(self, "_audio_backend", None)
if previous is not None and previous is not backend:
shutdown = getattr(previous, "shutdown", None)
if callable(shutdown):
shutdown()
self._audio_backend = backend
from .audio_protocol import TreeBoundAudioBackend as _TBA
if isinstance(backend, _TBA):
backend.bind_tree(self)
[docs]
@property
def audio_playback(self) -> AudioPlaybackBackend | None:
"""The active backend narrowed to :class:`AudioPlaybackBackend`, or ``None``.
Always equals :attr:`audio_backend` cast to the playback facet when
a backend is present: every shipped backend implements playback,
including the silent :class:`NullAudioBackend`.
"""
from .audio_protocol import AudioPlaybackBackend as _APB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _APB) else None
[docs]
@property
def audio_streaming(self) -> AudioStreamingBackend | None:
"""The active backend narrowed to :class:`AudioStreamingBackend`, or ``None``.
Returns ``None`` when the active backend doesn't implement streaming
(the Null backend, any future no-device test stub). Callers that
need streaming (:class:`AudioSynth` driver, AudioWorklet feeds)
should raise :class:`AudioCapabilityError` on ``None``.
"""
from .audio_protocol import AudioStreamingBackend as _ASB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _ASB) else None
[docs]
@property
def audio_buses(self) -> AudioBusBackend | None:
"""The active backend narrowed to :class:`AudioBusBackend`, or ``None``.
Always equals :attr:`audio_backend` cast to the bus facet when a
backend is present: every shipped backend implements bus + capability
advertisement.
"""
from .audio_protocol import AudioBusBackend as _ABB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _ABB) else None
[docs]
def audio_listener_3d(self) -> AudioListener3D | None:
"""The active 3D audio listener, lazy-creating one if none exists.
Returns the most recently entered :class:`AudioListener3D` in the
scene. If none has been added, auto-creates a fallback parented
to the active ``Camera3D`` with a one-time warning. Returns
``None`` only if there's no camera either.
Audio players call this every frame, so the auto-creation is
cheap once it's happened (the cached listener is returned).
"""
listener = self._current_audio_listener_3d
if listener is not None:
return listener
from .audio_listener import _autocreate_listener_3d
return _autocreate_listener_3d(self)
[docs]
def audio_listener_2d(self) -> AudioListener2D | None:
"""The active 2D audio listener, lazy-creating one if none exists.
Same contract as :meth:`audio_listener_3d` but for 2D.
"""
listener = self._current_audio_listener_2d
if listener is not None:
return listener
from .audio_listener import _autocreate_listener_2d
return _autocreate_listener_2d(self)
def __init__(self, screen_size=None, *, isolated_input: bool = False):
if isolated_input:
from .input.map import _InputMap
from .input.state import _Input
self._own_input_map = _InputMap()
self._own_input = _Input(input_map=self._own_input_map)
else:
from .input.map import _default_input_map
from .input.state import _default_input
self._own_input_map = _default_input_map
self._own_input = _default_input
# The tree's own top node, and the current scene as one of its children.
# Constructed with its tree reference already bound, so that mounting a
# singleton or a scene through ``_tree_root.add_child`` runs the ordinary
# enter/ready path and bumps ``_structure_version`` exactly as any other
# mount does.
self._scene_root: Node | None = None
self._tree_root = _TreeRoot(name="_tree_root")
self._tree_root._tree = self
# Depth of open activate_input() spans (re-entrant): > 0 while this tree
# is running its own node code, which is a tick, a physics step, a mount,
# a teardown, a layout flush, drawing, a group broadcast, or an input,
# UI, touch, pick or signal dispatch. Read by the InputMap late-
# registration warning to exempt the registrations node code makes while
# the engine is running it, and by the mount/unmount fast paths that skip
# opening a redundant nested span.
self._input_span_depth: int = 0
# The subtree an attach walk is currently binding, or None outside one.
# Node._enter_tree registers into this list on the way down and runs the
# entry hooks off it afterwards, so a hook sees a fully attached subtree.
self._attaching: list[Node] | None = None
self._screen_size: tuple[float, float] = SceneTree._normalize_size(screen_size or (800, 600))
self.paused: bool = False
self._delete_queue: list[Node] = []
self._groups: dict[str, set[Node]] = {}
self._singletons: dict[str, Node] = {}
# Transient fire-and-forget timers created via ``create_timer``. Ticked
# right after singletons each frame and dropped once they fire, so a
# one-shot delay needs no Node in the scene and no manual cleanup.
self._scene_timers: list[SceneTreeTimer] = []
# Escape-hatch deferred calls queued via ``Node.call_deferred``. Drained
# once per frame at the end-of-frame sync point (after traversal, before
# the delete flush). Calls queued during the drain run next frame.
self._deferred_calls: list[tuple[Callable[..., Any], tuple]] = []
# Engine-provided typed event bus, accessible as ``tree.events``. Held
# outside ``_singletons`` because EventBus is not a Node: entries there
# are children of the tree's own top node, and the bus does not
# participate in the process/physics traversal -- the per-frame tick
# explicitly calls ``self._events.flush_deferred()``. The name
# ``events`` is reserved (see add_singleton()).
self._events: EventBus = EventBus()
self._unique_nodes: dict[str, Node] = {}
# Input dispatch tables: populated as @on_input handlers register on tree-enter.
# Keyed for O(1) lookup so input dispatch is O(handlers per event), not O(nodes).
self._action_handlers: dict[tuple[str, bool], list[tuple[Node, str, tuple]]] = {}
self._key_handlers: dict[tuple[Key, bool], list[tuple[Node, str, tuple]]] = {}
self._key_handlers_any: dict[bool, list[tuple[Node, str, tuple[Key, ...], tuple]]] = {True: [], False: []}
self._button_handlers: dict[tuple[MouseButton, bool], list[tuple[Node, str, tuple]]] = {}
self._motion_handlers: list[tuple[Node, str]] = []
self._scroll_handlers: list[tuple[Node, str]] = []
self._joy_button_handlers: dict[tuple[JoyButton, bool], list[tuple[Node, str]]] = {}
self._joy_axis_handlers: dict[JoyAxis, list[tuple[Node, str]]] = {}
self._catch_all_handlers: list[tuple[Node, str]] = []
self._unhandled_input_dirty: bool = True # rebuilt from tree walk on next propagate_input
from .ui.ui_input import UIInputManager # local import avoids cycle via ui/testing.py
self._ui = UIInputManager()
self._ui._tree = self # back-ref so the manager can read the overlay registry
self._current_camera_2d: Camera2D | None = None
# Active audio listeners: set by AudioListener2D/3D's on_enter_tree.
# Audio players read these via the audio_listener_2d() / _3d()
# accessors below, which lazy-create a fallback at the active
# camera if no explicit listener exists in the scene.
self._current_audio_listener_2d: AudioListener2D | None = None
self._current_audio_listener_3d: AudioListener3D | None = None
self.auto_physics: bool = True # Automatically step the physics world(s) each physics tick
self.physics_interpolation: bool = True # render-time lerp of new-world bodies between fixed steps
# Default physics worlds. Both slots stay empty/lazy so non-physics scenes
# allocate nothing.
self._physics_world: PhysicsWorld | None = None # lazily-created default 3D world
self._physics_world_2d: Physics2DWorld | None = None # lazily-created default 2D world
# Registry of active PhysicsRoot / PhysicsRoot2D worlds. Holds BOTH 3D and
# 2D worlds (each registers via register_physics_world); the step/sync/
# dispatch loop handles them polymorphically once the leaves are width-aware.
self._physics_worlds: list[Any] = []
# Handle->node sync registry: per-world _BodySync used to scatter
# simulated transforms back onto body nodes after step.
# Empty for non-physics / all-static scenes (zero overhead).
self._physics_bodies: dict[AnyPhysicsWorld, _BodySync] = {}
# Handle->node map for collision-event dispatch.
# Holds EVERY PhysicsBody3D (static included, unlike _physics_bodies),
# weakly, so _dispatch_contact_events can resolve both sides of a pair.
# Empty for non-physics scenes (zero overhead).
self._physics_nodes: dict[AnyPhysicsWorld, weakref.WeakValueDictionary[BodyHandle, Node]] = {}
# Bodies destroyed but still owed their closing contact/overlap events:
# handle -> (node, step sequence at which it was retired). STRONG refs, so
# a peer the game has already dropped survives long enough to be named.
# Released one step after retirement; empty for non-physics scenes.
self._physics_retired: dict[AnyPhysicsWorld, dict[BodyHandle, tuple[Node, int]]] = {}
# Per-world count of steps this tree has driven, the clock the retirement
# grace period is measured on.
self._physics_step_seq: dict[AnyPhysicsWorld, int] = {}
self.overlay_offset: tuple[float, float] = (0.0, 0.0) # 2D offset for Text2D/particle overlays
# The tree-owned overlay layer: ordered registry of on-top UI overlays
# (dropdowns, dialogs, tooltips). Empty registry is an O(1) zero-cost gate.
from .ui.overlay import OverlayLayer
self.overlays = OverlayLayer(self)
# Layout: dirty containers awaiting reflow, drained by ``flush_layout``
# (end-of-tick and synchronously on overlay/popup open). Empty set is the
# O(1) zero-cost gate. ``_layout_frame`` is this tree's private rect-cache
# epoch (per-tree so a second ticking tree does not invalidate this one's
# cached global rects); ``flush_layout`` and ``tick`` advance it.
self._dirty_layout: set = set()
self._layout_frame: int = 0
self.play_viewport_rect: tuple[float, float, float, float] | None = (
None # (x, y, w, h) to constrain 3D rendering
)
self._structure_version: int = 0 # Incremented on add_child/remove_child for cache invalidation
self._running: bool = True
# Frame counter: bumped at the start of every ``tick`` so callers
# (notably InputMap.add_action's late-call warning) can tell whether
# the scene has begun ticking. Stays at 0 inside the root node's
# initial ``ready`` callback so registrations there are silent.
self._tick_count: int = 0
# Monotonically-increasing scene time, in seconds. Accumulates ``dt``
# at the start of every ``tick()`` (after any ``App.time_scale``
# scaling has been applied to ``dt``) so node code can read a single
# shared clock instead of maintaining its own ``_time += dt``
# accumulator. Frozen while ``self.paused`` is True.
self._now: float = 0.0
# ``Property(on_change=..., coalesce=True)`` enqueues ``(obj, method_name)``
# pairs here instead of firing synchronously. Drained once at the end of
# ``process()``: multiple writes within one tick collapse to a single call.
# Set semantics dedupe automatically.
self._pending_coalesced_hooks: set[tuple[Any, str]] = set()
self.quit_requested: Signal = Signal()
# Overlay lifecycle: fires when any Control's show_overlay()/close_overlay()
# opens/closes a registry entry. Receivers get the overlay Control so
# external listeners (e.g. game-side pause bridges) can react without being
# involved in the overlay itself.
self.overlay_opened: Signal = Signal()
self.overlay_closed: Signal = Signal()
# Fires whenever ``screen_size`` changes (window resize, fullscreen
# toggle, viewport swap). Receivers get the new ``(width, height)``
# tuple and can recompute layout once instead of every frame.
self.screen_resized: Signal = Signal()
# -- UI state forwarding (preserves external interface) --
@property
def _focused_control(self):
return self._ui._focused_control
@_focused_control.setter
def _focused_control(self, v):
self._ui._focused_control = v
@property
def _mouse_grab(self) -> Any:
return self._ui._mouse_grab
@_mouse_grab.setter
def _mouse_grab(self, v):
self._ui._mouse_grab = v
@property
def _last_mouse_pos(self):
return self._ui._last_mouse_pos
@_last_mouse_pos.setter
def _last_mouse_pos(self, v):
self._ui._last_mouse_pos = v
@property
def _shortcut_handler(self):
return self._ui._shortcut_handler
@_shortcut_handler.setter
def _shortcut_handler(self, v):
self._ui._shortcut_handler = v
@staticmethod
def _normalize_size(sz) -> tuple[float, float]:
"""Coerce any screen_size representation to a plain tuple once."""
if isinstance(sz, tuple) and len(sz) == 2:
return sz
if hasattr(sz, "x"):
return (float(sz.x), float(sz.y))
return (float(sz[0]), float(sz[1]))
[docs]
@property
def is_running(self) -> bool:
"""Whether the tree is actively being ticked by its driving app.
Set to False by :meth:`quit` (or the graphics backend's app ``quit()``),
signalling the main loop to exit at the end of the current frame.
"""
return self._running
[docs]
@property
def now(self) -> float:
"""Monotonically-increasing scene time, in seconds.
Accumulates ``dt`` at the start of every :meth:`tick` call (after
any :attr:`simvx.graphics.App.time_scale` scaling has been applied),
so slow-motion and hitstop also slow this clock. Frozen while
:attr:`paused` is True or any inert overlay is open. Resets to
``0.0`` only by constructing a fresh ``SceneTree``.
Use for time-based animation, slow-mo gating, and any "how long has
the scene been running" query: preferable to per-node
``self._time += dt`` accumulators because every consumer reads the
same monotonic value.
"""
return self._now
[docs]
def quit(self) -> None:
"""Request a clean shutdown of the running tree.
Emits :attr:`quit_requested` and flips :attr:`is_running` to False. The
driving app polls this state and exits its loop at the end of the
current frame. Safe to call from node callbacks or signal handlers.
"""
if not self._running:
return
self._running = False
with self.activate_input():
self.quit_requested()
# Release the physics worlds on shutdown (default world + PhysicsRoot
# registry) so a torn-down tree leaks nothing. change_scene deliberately
# does NOT drop these: the default world is tree-scoped (survives swaps).
self._drop_physics_worlds()
@property
def screen_size(self) -> tuple[float, float]:
return self._screen_size
[docs]
@screen_size.setter
def screen_size(self, value):
old = self._screen_size
self._screen_size = SceneTree._normalize_size(value)
if self._screen_size != old:
with self.activate_input():
self._invalidate_draw_caches()
self.screen_resized(self._screen_size)
@property
def root(self) -> Node | None:
"""The current scene's root node.
A sibling of this tree's singletons under the tree's own top node, and
the node every scene-facing walk starts from: rendering, 3D submission,
saving, hot reload and the editor's scene panel all begin here, so a
singleton is ticked but is not part of the scene.
Assigning re-links the child and runs no lifecycle, which is what
assigning to it has always done. Use :meth:`set_root` to mount a scene
(enter + ready) and :meth:`change_scene` to swap one.
"""
return self._scene_root
[docs]
@root.setter
def root(self, node: Node | None) -> None:
old = self._scene_root
if old is not None:
# Clears ``_scene_root`` through the _TreeRoot override.
self._tree_root._detach_child(old)
self._scene_root = node
if node is not None:
self._tree_root._attach_child(node)
# The scene is always the tree root's last child, so it processes
# after every singleton (see ``tick``).
self._tree_root.children.move_last(node)
[docs]
def set_root(self, root: Node):
"""Mount ``root`` as the current scene.
Before the root enters the tree, any ``input_actions`` declared on
the root (class- or instance-level ``dict[str, list]``) is bulk-
registered with this tree's ``InputMap``. This is the canonical
replacement for the wrapper-class + ``on_ready`` boilerplate and
survives ``change_scene`` swaps -- every new root's actions are
re-registered automatically.
A scene already mounted is carried out first, so its ``on_exit_tree``
runs, exactly as :meth:`change_scene` does. ``change_scene`` remains the
call for a swap: it also clears pending deletes, popup state and the
active 2D camera.
The mount itself runs inside this tree's input span, so
``on_enter_tree`` / ``on_ready`` code sees this tree's ``Input`` and
``InputMap``, and any action it registers lands in this tree's own map.
"""
log.debug("SceneTree.set_root(%s)", root)
# Publish "this is the active tree" so input map / asset paths that
# need cross-cutting context can find us without a parameter chain.
# Weak so we do not pin the previous/last tree's node graph.
SceneTree._active_tree = weakref.ref(self)
self._detach_scene_root()
self._register_declared_input_actions(root)
# Published before the mount: an ``on_enter_tree`` reads ``tree.root``
# (``_FullscreenOverlay`` decides whether to register on the overlay
# layer by it), so the reference has to be live by the time the entry
# hooks run.
self._scene_root = root
with self.activate_input():
# attach + enter + ready, and the scene lands last among the tree
# root's children.
self._tree_root.add_child(root)
def _detach_scene_root(self) -> None:
"""Carry the current scene out of the tree, if there is one."""
old = self._scene_root
if old is None:
return
if old.parent is not self._tree_root:
# Somebody reparented the scene root without going through
# ``_detach_child``, so "the scene is a child of the tree root" no
# longer holds and there is nothing here to carry out.
self._scene_root = None
if Node.strict_errors:
raise RuntimeError(
f"Scene root {old.name!r} is no longer a child of the tree. "
"Reparent a scene root with change_scene(), not add_child()."
)
return
# Opens the input span, runs _exit_tree, and clears _scene_root.
self._tree_root.remove_child(old)
self._prune_singletons()
self._prune_dead_index_entries()
def _register_declared_input_actions(self, root: Node) -> None:
actions = getattr(root, "input_actions", None)
if not actions:
return
if not isinstance(actions, dict):
log.warning(
"Root %s.input_actions must be a dict[str, list]; got %s -- skipping.",
type(root).__name__,
type(actions).__name__,
)
return
for name, bindings in actions.items():
self._own_input_map.add_action(name, list(bindings) if bindings else None, _quiet=True)
[docs]
def change_scene(self, new_root: Node):
"""Swap the active root with ``new_root``.
The old root receives ``_exit_tree``; ``new_root`` then runs the full
``_enter_tree`` / ``_ready_recursive`` path, identical to the initial
root. Singletons the tree owns are left in place, keeping their groups
and unique names; a singleton registered on a node the old scene owned
leaves with that scene, and its name binding is dropped. Pending
deletes, UI popup state, and the active 2D camera are cleared.
Use this for title → gameplay → game-over navigation. See
:doc:`../patterns` for a full example.
"""
log.debug("SceneTree.change_scene(%s), old root=%s", new_root, self._scene_root)
self._detach_scene_root()
self.clear_delete_queue()
# The old scene left in one piece, so nothing is owed a closing contact
# event; release the retired bodies rather than hold a whole dead scene
# alive for one more step.
self._physics_retired.clear()
self._ui.reset()
self._current_camera_2d = None
self.set_root(new_root)
#: Backstop for pathological layout feedback (a container that re-dirties
#: itself every pass). Layout converges in ~2 passes by idempotence (size
#: ``on_change`` fires only on an actual change), so this only ever caps a bug.
_LAYOUT_FLUSH_CAP = 8
def _register_dirty_layout(self, container) -> None:
"""Record ``container`` as needing reflow (called by ``Container.mark_layout_dirty``)."""
self._dirty_layout.add(container)
[docs]
def flush_layout(self) -> None:
"""Reflow every dirty container now, root-first, and expire the rect cache.
Synchronous layout settle: after this returns, container children have
their final positions/sizes for this frame, so ``get_global_rect`` is
valid immediately (no one-tick lag). Called at end-of-tick and, crucially,
right when an overlay/popup opens so a freshly-shown dialog can be
hit-tested and focus-elected on the same tick. Idempotent and O(1) when no
container is dirty.
The reflow itself runs inside this tree's input span: ``_do_update_layout``
is an overridable hook, so a custom container laying itself out sees this
tree's ``Input`` and ``InputMap``.
"""
dirty = self._dirty_layout
if not dirty:
return
def _depth(node) -> int:
d, cur = 0, node.parent
while cur is not None:
d += 1
cur = cur.parent
return d
with self.activate_input():
for _ in range(self._LAYOUT_FLUSH_CAP):
if not dirty:
break
# Root-first: a parent's size feeds its children's layout, so lay out
# shallower containers before deeper ones (topological, same shape as
# SubViewport's producer-before-consumer ordering).
batch = sorted(dirty, key=_depth)
dirty.clear()
for c in batch:
if c._tree is self and c._layout_dirty:
c._do_update_layout()
dirty.clear()
# Containers moved; descendants cache global rects by walking ancestors,
# so bump this tree's epoch to expire those caches (mirrors the direct
# ``_rect_frame`` reset a node does on its own move).
self._layout_frame += 1
[docs]
def tick(self, dt: float):
"""Run process callbacks and coroutines on all nodes for one frame.
Order: singletons' ``_process`` first, in registration order (global
nodes, before the scene), then ``self.events.flush_deferred()`` so
deferred events queued during the previous frame (process, physics, or
input) reach all subscribers before scene logic runs, then the scene
root's ``_process``. Anything ``emit_deferred``'d during this frame's
singleton pass is also drained in the same flush, keeping the scene's
view of the world consistent.
Singletons and the current scene are siblings under the tree's own top
node, and the scene is always its last child, so every node the tree
ticks is reached through exactly one containment path and processes
exactly once. A node registered as a singleton *and* parented into the
scene is a scene node that happens to have a name: it processes in its
scene position, not ahead of the scene.
"""
with self.activate_input():
from .assets import AssetServer
from .ui.core import Control
# Per-tick rect-cache auto-expire (catches ancestor moves; direct
# moves self-invalidate via _invalidate_transform). Per-tree so a
# sibling tree's tick doesn't churn this tree's caches. The global
# Control._current_frame is kept as the detached-control fallback.
self._layout_frame += 1
Control._current_frame += 1
self._tick_count += 1
# An inert overlay pauses the tree (derived, so closing it auto-resumes).
# ``any_inert`` short-circuits False on the empty registry (zero-cost).
paused = self.paused or self.overlays.any_inert()
if not paused:
self._now += dt
# Asset-loader completions arrive on worker threads; drain them
# onto the main thread before any node code runs so handlers see
# a stable world.
if AssetServer._instance is not None:
AssetServer._instance.flush()
scene = self._scene_root
for child in self._tree_root.children.safe_iter():
# ``safe_iter`` snapshots before the loop, so a hook that
# removes a later sibling leaves it here with its tree already
# cleared; a node that has left is not processed.
if child is not scene and child._tree is self:
child._process_recursive(dt, paused)
if self._scene_timers and not paused:
self._scene_timers = [t for t in self._scene_timers if not t._tick(dt)]
self._events.flush_deferred()
# Re-read: a singleton's on_update or a deferred handler may have
# swapped the scene, and it is the new one that processes this frame.
scene = self._scene_root
if scene is not None and scene._tree is self:
scene._process_recursive(dt, paused)
self._flush_deferred_calls()
self._flush_deletes()
self._flush_coalesced_hooks()
# Reflow any container dirtied during this tick (mark_layout_dirty,
# size on_change, add/remove child) so draw + next-frame hit-testing
# see settled positions. O(1) when nothing is dirty.
self.flush_layout()
# Audio backend reconciliation. Native MiniaudioBackend uses this
# to detect AudioBus volume / effect-chain changes (Property
# mutations on AudioBus aren't on_change-instrumented: diffing
# snapshot is the canonical sync path). Legacy + Null backends
# are no-ops; web reads bus state during its drain. One-line
# call so live UI changes (sliders, toggles) take effect next
# frame on every backend.
backend = getattr(self, "_audio_backend", None)
if backend is not None:
from .audio_bus import AudioBusLayout
from .audio_errors import AudioError, raise_or_warn
try:
backend.sync_bus_layout(AudioBusLayout.get_default())
except AudioError as exc:
# Per-frame call: let the audio system's strict/warn-once
# policy decide whether to raise or log. Other exceptions
# propagate (genuine engine bugs shouldn't be hidden by
# the per-frame catch-all).
raise_or_warn(
exc,
key="audio.scene_tree.sync_bus_layout_failed",
message="audio backend sync_bus_layout failed",
)
def _flush_coalesced_hooks(self) -> None:
"""Drain pending ``on_change`` callbacks queued by ``coalesce=True`` Properties.
Iterates a snapshot so hooks that themselves write to ``coalesce`` properties
enqueue onto the next frame (the natural debounce). Missing methods are
ignored: the owning node may have been removed mid-frame.
"""
if not self._pending_coalesced_hooks:
return
pending = self._pending_coalesced_hooks
self._pending_coalesced_hooks = set()
for obj, method_name in pending:
method = getattr(obj, method_name, None)
if method is not None:
method()
# -- Physics world ownership ------------------------------------------
[docs]
@property
def physics_world(self) -> PhysicsWorld:
"""The tree's default PhysicsWorld, created lazily on first access.
Bodies with no PhysicsRoot ancestor resolve here. Lazy so non-physics
scenes never allocate a world. Backend follows the selection precedence
(no node override -> project ``physics_backend`` setting > auto-discovered
native > Builtin), gravity -Y. Tree-scoped: it persists across
``change_scene`` (like ``events``) and is released only by ``quit()`` / GC.
"""
if self._physics_world is None:
from .math import Vec3
from .physics.backends import resolve_world_factory
self._physics_world = resolve_world_factory()(Vec3(0.0, -9.81, 0.0))
return self._physics_world
[docs]
@property
def has_physics_world(self) -> bool:
"""True iff a default world has been lazily created (no allocation)."""
return self._physics_world is not None
[docs]
@property
def physics_world_2d(self) -> Physics2DWorld:
"""The tree's default 2D PhysicsWorld, created lazily on first access.
The 2D sibling of :attr:`physics_world`. 2D bodies with no ``PhysicsRoot2D``
ancestor resolve here. Lazy so non-2D-physics scenes never allocate it.
Backend follows the selection precedence (no node override -> project
``physics_backend`` setting > auto-discovered native > Builtin), gravity
``Vec2(0, -9.81)`` (Y-up). Tree-scoped: persists across ``change_scene``,
released by ``quit()`` / GC.
"""
if self._physics_world_2d is None:
from .math import Vec2
from .physics.backends import resolve_world_factory_2d
self._physics_world_2d = resolve_world_factory_2d()(Vec2(0.0, -9.81))
return self._physics_world_2d
@staticmethod
def _world_dims(world: Any) -> tuple[int, int]:
"""Return ``(width, dims)`` for a physics world: ``(4, 2)`` 2D, ``(7, 3)`` 3D.
Selects the bulk-array width + scatter/interpolate branch from the world's
type. 2D worlds subclass ``Physics2DWorld`` (``(N,4)`` transform rows);
everything else is a 3D world (``(N,7)``). Imported lazily to keep
``simvx.core.scene_tree`` import-cheap for non-physics scenes.
"""
from .physics.world2d import Physics2DWorld
if isinstance(world, Physics2DWorld):
return (4, 2)
return (7, 3)
[docs]
def register_physics_world(self, world: PhysicsWorld) -> None:
"""Register a PhysicsRoot's isolated world so physics_tick steps it.
Idempotent: a world already present is not added twice (defends against
re-entrant enter_tree). Called from PhysicsRoot.on_enter_tree.
"""
if world not in self._physics_worlds:
self._physics_worlds.append(world)
[docs]
def unregister_physics_world(self, world: PhysicsWorld) -> None:
"""Unregister a PhysicsRoot's world (PhysicsRoot.on_exit_tree). No-op if absent.
The world will not be stepped by this tree again, so any body still in
its retirement grace period is released here rather than left holding a
node alive (see :meth:`retire_physics_node`).
"""
try:
self._physics_worlds.remove(world)
except ValueError:
pass
self._physics_retired.pop(world, None)
self._physics_step_seq.pop(world, None)
# -- Physics body sync (the world layer below is node-agnostic) -------
[docs]
def register_physics_body(self, world: PhysicsWorld, handle: BodyHandle, node: Node) -> None:
"""Register a body node for post-step transform read-back.
Called by ``PhysicsBody3D.on_enter_tree`` for non-static bodies. Marks
the world's membership ``dirty`` so the next sync rebuilds the bulk
order + buffer. Weak reference: a GC'd node auto-drops.
"""
sync = self._physics_bodies.get(world)
if sync is None:
width, dims = self._world_dims(world)
sync = self._physics_bodies[world] = _BodySync(width=width, dims=dims)
sync.nodes[handle] = node
sync.dirty = True
[docs]
def unregister_physics_body(self, world: PhysicsWorld, handle: BodyHandle) -> None:
"""Unregister a body node (``PhysicsBody3D.on_exit_tree``). No-op if absent.
Drops the whole per-world ``_BodySync`` once empty so idle worlds cost
nothing.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
sync.nodes.pop(handle, None)
sync.dirty = True
if not sync.nodes:
del self._physics_bodies[world]
[docs]
def register_physics_node(self, world: PhysicsWorld, handle: BodyHandle, node: Node) -> None:
"""Register a body node in the collision-event handle->node map.
Called by ``PhysicsBody3D.on_enter_tree`` for ALL modes (static
included), so :meth:`_dispatch_contact_events` can resolve both sides of
a contact pair. Mode-independent: unlike ``register_physics_body`` this
map never changes on a mode flip. Weak reference: a GC'd node auto-drops.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
nodes = self._physics_nodes[world] = weakref.WeakValueDictionary()
nodes[handle] = node
[docs]
def retire_physics_node(self, world: AnyPhysicsWorld, handle: BodyHandle) -> None:
"""Move a body node out of the live map and into the retirement grace period.
Called by ``PhysicsObject3D.on_exit_tree`` immediately before the backend
body is destroyed. Destroying a body ENDS every contact and sensor
overlap it was in, and the seam reports those closing ``EXIT``s on the
first drain after the NEXT step (see
:meth:`~simvx.core.physics.world.PhysicsWorld.destroy_body`). Dropping
the handle->node entry here would leave that dispatch unable to name the
peer that went away, so the entry is retired instead: held, with a STRONG
reference to the node (the live map is weak, and the game has usually let
go of a node it destroyed), until the events it is owed have been
dispatched.
The grace period is one step, measured on this tree's per-world step
count, and :meth:`_release_retired_physics_nodes` ends it. Retirement is
never consulted before the live map, so a backend that recycles handles
resolves a reused handle to its NEW body.
One STEP, not one frame: a tree that stops stepping keeps the entry until
it steps again, so pausing (or clearing ``auto_physics``) immediately
after a kill holds that one node for the duration of the pause. It is
released on the next step regardless, and by ``change_scene`` and by
unregistering the world, so it cannot outlive the scene either way.
No-op if the handle is not registered.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
return
node = nodes.pop(handle, None)
if not nodes:
del self._physics_nodes[world]
if node is None:
return
retired = self._physics_retired.get(world)
if retired is None:
retired = self._physics_retired[world] = {}
retired[handle] = (node, self._physics_step_seq.get(world, 0))
def _resolve_physics_node(self, world: AnyPhysicsWorld, handle: BodyHandle) -> Node | None:
"""The node behind ``handle``: live first, then retired, else ``None``.
Live-first is what keeps a recycled handle honest (see
:meth:`retire_physics_node`).
"""
nodes = self._physics_nodes.get(world)
node = nodes.get(handle) if nodes is not None else None
if node is not None:
return node
retired = self._physics_retired.get(world)
if retired is None:
return None
entry = retired.get(handle)
return entry[0] if entry is not None else None
def _release_retired_physics_nodes(self, world: AnyPhysicsWorld) -> None:
"""End the grace period for bodies retired before this world's last step.
A body retired DURING the dispatch that follows a step (a handler calling
``destroy`` on the peer it just touched) carries the current sequence
number and so survives this pass; its own closing events arrive after the
next step. Anything older has had its dispatch and is dropped.
"""
retired = self._physics_retired.get(world)
if not retired:
return
seq = self._physics_step_seq.get(world, 0)
for handle in [h for h, (_node, at) in retired.items() if at < seq]:
del retired[handle]
if not retired:
del self._physics_retired[world]
def _dispatch_contact_events(self, world: AnyPhysicsWorld) -> None:
"""Drain ``world``'s buffered contact events and fire node Signals.
Called once per stepped world, AFTER :meth:`PhysicsWorld.step` and
:meth:`_sync_physics_world`, so node transforms are already current and
a Signal handler may safely mutate the tree (spawn/destroy): any change
takes effect on the next step, and a handler-triggered destroy this turn
is safe because dispatch runs outside the broad/narrow loops.
The world layer knows nothing about nodes: its events are keyed by body
handles. This resolves both handles via :meth:`_resolve_physics_node` and
builds one node-level :class:`Contact` per side, with ``other`` set to the
peer and ``normal`` / ``velocity`` reoriented for that side (the world
reports the normal as ``a -> b``; the ``b`` side is negated).
A peer DESTROYED mid-touch is still named: it is resolved out of the
retirement grace period (:meth:`retire_physics_node`), so ``separated``
fires with a ``Contact`` whose ``other`` is the real node, detached
(``other.tree`` and ``other.handle`` are ``None``). An event is skipped
only when a side has no node at all, which now means garbage-collected
without ever having been in this tree.
"""
if world not in self._physics_nodes and world not in self._physics_retired:
return
events = world.drain_contact_events()
if not events:
return
# Width-aware: build the 2D node-level Contact for a 2D world, else the 3D
# one. ContactPhase is shared (dimension-agnostic). The world-level
# ContactEvent field layout is identical across dims, so only the payload
# type differs.
from .physics.world import ContactPhase
if self._world_dims(world)[1] == 2:
from .physics.nodes2d import Contact2D as Contact
else:
from .physics.nodes import Contact
for ev in events:
a = self._resolve_physics_node(world, ev.a)
b = self._resolve_physics_node(world, ev.b)
if a is None or b is None:
continue # never registered, or GC'd: nothing to notify
# Seam normal is a->b and rel_velocity is b w.r.t. a. Each node sees a
# normal pointing TOWARD itself (its separating direction) and the
# peer's velocity w.r.t. itself: that is -normal/+rel_vel for the ``a``
# side and +normal/-rel_vel for the ``b`` side.
if ev.phase is ContactPhase.ENTER:
a.collided(
Contact(
other=b,
point=ev.point,
normal=-ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
b.collided(
Contact(
other=a,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=-ev.rel_velocity,
)
)
else:
# EXIT: degenerate payload (no live manifold), only ``other`` is
# meaningful. ``impulse`` is forwarded rather than zeroed: a
# backend that cannot measure reports None, and hardcoding 0.0
# here would relabel that as "measured, and it was zero". The
# estimate needs no such care: it is a number on every backend,
# and the seam already reports 0.0 for an exit.
a.separated(
Contact(
other=b,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
b.separated(
Contact(
other=a,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
def _dispatch_overlap_events(self, world: AnyPhysicsWorld) -> None:
"""Drain ``world``'s buffered sensor-overlap events and fire Area Signals.
The SEPARATE sensor stream, parallel to :meth:`_dispatch_contact_events`
but DIRECTED ``sensor -> other``: dispatch fires ONLY on the detecting
sensor's :class:`Area3D` node, never the reverse, so sensor-vs-sensor
produces up to two independent directed events (one per observer) with no
double-firing logic here. Reuses the same mode-independent
:attr:`_physics_nodes` handle->node map (Area sensor bodies register via
``register_physics_node`` too).
Called once per stepped world AFTER :meth:`_dispatch_contact_events`, so a
Signal handler may safely mutate the tree (deferred: effect next step).
Node-agnostic: routes ``body_entered`` / ``body_exited`` vs
``area_entered`` / ``area_exited`` by the OTHER node's type. The Area's
live overlap sets are maintained HERE (ENTER add / EXIT discard) so its
polling accessors stay consistent with the last dispatched step.
A peer destroyed mid-overlap is named out of the retirement grace period
exactly as on the contact stream, so ``body_exited`` / ``area_exited``
fires with the detached node AND the Area's overlap set loses it.
"""
if world not in self._physics_nodes and world not in self._physics_retired:
return
events = world.drain_overlap_events()
if not events:
return
# Width-aware: resolve the Area / physics-object node types for this
# world's dim so the routing (body_* vs area_* by the OTHER node's type) is
# dimension-correct. The non-area branch tests the shared physics-object
# base, so a character routes to body_entered / body_exited like any other
# body without a second isinstance.
from .physics.world import ContactPhase
if self._world_dims(world)[1] == 2:
from .physics.nodes2d import Area2D as Area
from .physics.nodes2d import PhysicsObject2D as Body
else:
from .physics.nodes import Area3D as Area
from .physics.nodes import PhysicsObject3D as Body
for ev in events:
# The detecting sensor MUST be an Area node.
area = self._resolve_physics_node(world, ev.sensor)
other = self._resolve_physics_node(world, ev.other)
if area is None or other is None:
continue # never registered, or GC'd
if not isinstance(area, Area):
continue # defensive: only Area nodes own sensors
if ev.phase is ContactPhase.ENTER:
if isinstance(other, Area):
area._overlapping_areas.add(other)
area.area_entered(other)
elif isinstance(other, Body):
area._overlapping_bodies.add(other)
area.body_entered(other)
else: # EXIT
if isinstance(other, Area):
area._overlapping_areas.discard(other)
area.area_exited(other)
elif isinstance(other, Body):
area._overlapping_bodies.discard(other)
area.body_exited(other)
def _sync_physics_world(self, world: AnyPhysicsWorld) -> None:
"""Scatter ``world``'s simulated transforms back onto its body nodes.
Called once per stepped world, after :meth:`PhysicsWorld.step`. Uses the
bulk contract only: ``register_bodies`` (re-fixed lazily on membership
change) + ``read_transforms`` into a reused preallocated buffer. One bulk
transfer per world per frame, no hot-path allocation.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
if sync.dirty:
sync.order = list(sync.nodes.keys())
world.register_bodies(sync.order)
sync.buf = np.empty((len(sync.order), sync.width), np.float32)
sync.dirty = False
if not sync.order:
return
world.read_transforms(sync.buf)
buf = sync.buf
if sync.dims == 2:
# 2D row is [px, py, cos, sin]; rotation is a scalar = atan2(sin, cos).
from .math import Vec2
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None: # GC'd between unregister and sync
continue
row = buf[i]
# _write_simulated_pose guards the assignment so the pose-reconcile
# hook does not bounce this simulated pose back into the handle.
node._write_simulated_pose(Vec2(row[0], row[1]), math.atan2(float(row[3]), float(row[2])))
return
from .math import Quat, Vec3
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None: # GC'd between unregister and sync
continue
row = buf[i]
# read_transforms emits xyzw (scalar-last); engine Quat is (w,x,y,z).
node._write_simulated_pose(Vec3(row[0], row[1], row[2]), Quat(row[6], row[3], row[4], row[5]))
def _capture_physics_world(self, world: AnyPhysicsWorld) -> None:
"""Capture this step's transforms into the per-world prev/cur ring.
Rolls cur -> prev, then reads the post-step pose into cur. Writes nothing
to nodes: render-time interpolation (interpolate_physics) does the scatter.
A membership change (dirty) rebuilds order/buffers and re-seeds prev == cur
so a newly added body never lerps from origin on its first rendered frame.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
if sync.dirty:
sync.order = list(sync.nodes.keys())
world.register_bodies(sync.order)
n = len(sync.order)
sync.buf = np.empty((n, sync.width), np.float32)
sync.prev_buf = np.empty((n, sync.width), np.float32)
sync.cur_buf = np.empty((n, sync.width), np.float32)
sync.dirty = False
sync.seeded = False
if not sync.order:
return
world.read_transforms(sync.buf)
if not sync.seeded:
sync.prev_buf[:] = sync.buf # seed BOTH: first frame holds steady, no origin lerp
sync.cur_buf[:] = sync.buf
sync.seeded = True
else:
sync.prev_buf, sync.cur_buf = sync.cur_buf, sync.prev_buf # roll cur -> prev (swap, no alloc)
sync.cur_buf[:] = sync.buf # new state into cur
[docs]
def interpolate_physics(self, alpha: float) -> None:
"""Write the render pose of every captured new-world body: lerp(prev, cur, alpha).
``alpha`` is the fixed-step fraction ``physics_accum / physics_dt`` in [0, 1).
Position lerps linearly; rotation slerps (shortest-arc, Quat.slerp). A
sleeping body has prev == cur, so it lerps to a constant (correct, cheap).
No-op when ``physics_interpolation`` is False (driver still calls it; the raw
write already happened in _sync_physics_world during physics_tick).
"""
if not self.physics_interpolation:
return
from .math import Quat, Vec2, Vec3
a = 0.0 if alpha < 0.0 else 1.0 if alpha > 1.0 else alpha
for sync in self._physics_bodies.values():
if not sync.seeded or not sync.order:
continue
prev, cur = sync.prev_buf, sync.cur_buf
if sync.dims == 2:
# 2D: lerp position xy + lerp the cos/sin unit vector, then atan2
# (no +-pi wraparound: the load-bearing reason transforms carry
# cos/sin rather than the bare angle).
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None:
continue
p, c = prev[i], cur[i]
cos = float(p[2]) + (float(c[2]) - float(p[2])) * a
sin = float(p[3]) + (float(c[3]) - float(p[3])) * a
# Guarded write: this is a render-only interpolated pose; it
# must NOT be pushed back into the handle (would corrupt the
# next step's start pose).
node._write_simulated_pose(Vec2(p[0], p[1]).lerp(Vec2(c[0], c[1]), a), math.atan2(sin, cos))
continue
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None:
continue
p, c = prev[i], cur[i]
# read_transforms emits xyzw (scalar-last); engine Quat is (w,x,y,z).
qp = Quat(p[6], p[3], p[4], p[5])
qc = Quat(c[6], c[3], c[4], c[5])
node._write_simulated_pose(Vec3(p[0], p[1], p[2]).lerp(Vec3(c[0], c[1], c[2]), a), qp.slerp(qc, a))
def _drop_physics_worlds(self) -> None:
"""Release the default world and clear the PhysicsRoot + body registries.
Registered worlds are normally removed by each PhysicsRoot.on_exit_tree;
this is the belt-and-braces sweep so a torn-down tree leaks nothing even
if a root was force-detached. Safe to call repeatedly.
"""
self._physics_worlds.clear()
self._physics_world = None
self._physics_world_2d = None
self._physics_bodies.clear()
self._physics_nodes.clear()
self._physics_retired.clear()
self._physics_step_seq.clear()
[docs]
def physics_tick(self, dt: float):
"""Run physics_process callbacks on all nodes, then auto-step physics."""
# An inert overlay freezes the whole world, physics included (parity with
# the process pass in ``tick``); ``any_inert`` short-circuits on the empty
# registry, so this is zero-cost when no inert overlay is open.
paused = self.paused or self.overlays.any_inert()
with self.activate_input():
scene = self._scene_root
for child in self._tree_root.children.safe_iter():
if child is not scene and child._tree is self:
child._physics_process_recursive(dt, paused)
if scene is not None and scene._tree is self:
scene._physics_process_recursive(dt, paused)
if self.auto_physics:
# Step the default world (only if it was actually created) plus
# every registered PhysicsRoot world, once each, with this tick's
# dt. Skip empty worlds for zero overhead.
#
# Frozen while paused: a paused tree must not advance the sim or
# dispatch contact/overlap events. Render interpolation still runs
# (driver-side) so a paused world holds its last pose steady. The
# capture-vs-immediate-write split honours physics_interpolation:
# ON captures prev/cur for interpolate_physics; OFF writes nodes now.
if not paused:
if self._physics_world is not None:
self._step_physics_world(self._physics_world, dt)
# Default 2D world: same pause guard, same sync/dispatch
# path (the leaves are width-aware), parallel to the 3D default.
if self._physics_world_2d is not None:
self._step_physics_world(self._physics_world_2d, dt)
for world in self._physics_worlds:
self._step_physics_world(world, dt)
def _step_physics_world(self, world: AnyPhysicsWorld, dt: float) -> None:
"""Advance one world by ``dt``, scatter its poses and dispatch its events.
Also ends the retirement grace period of bodies destroyed before this
step, which is why it runs even for an empty world: a scene that destroys
its LAST body would otherwise hold that node alive forever, and with no
body left there is no open edge for the seam still to close.
"""
if world.body_count == 0:
self._physics_retired.pop(world, None)
return
self._physics_step_seq[world] = self._physics_step_seq.get(world, 0) + 1
world.step(dt)
if self.physics_interpolation:
self._capture_physics_world(world)
else:
self._sync_physics_world(world)
self._dispatch_contact_events(world)
self._dispatch_overlap_events(world)
self._release_retired_physics_nodes(world)
def _dispatch_input(self, event: TreeInputEvent) -> None:
"""Handler-table half of :meth:`propagate_input`, run inside the span.
Every handler goes through :meth:`_safe_invoke`, so a decorated handler
answers to ``Node.strict_errors`` exactly as ``on_unhandled_input`` and
every other hook does, and a node already disabled by a script error
receives nothing further.
"""
handled = False
# Action filters: route via the active InputMap's bindings.
for action_name in self._actions_for_event(event):
released_key = (action_name, not event.pressed)
for node, method_name, mods in tuple(self._action_handlers.get(released_key, ())):
if not _match_mods(mods, event):
continue
if self._safe_invoke(node, method_name, event):
handled = True
# Direct typed dispatch.
if event.type == "key" and event.key is not None:
released = not event.pressed
for node, method_name, mods in tuple(self._key_handlers.get((event.key, released), ())):
if not _match_mods(mods, event):
continue
if self._safe_invoke(node, method_name, event):
handled = True
for node, method_name, keys, mods in tuple(self._key_handlers_any.get(released, ())):
if event.key not in keys:
continue
if not _match_mods(mods, event):
continue
if self._safe_invoke(node, method_name, event):
handled = True
elif event.type == "mouse_button" and event.mouse_button is not None:
released = not event.pressed
for node, method_name, mods in tuple(self._button_handlers.get((event.mouse_button, released), ())):
if not _match_mods(mods, event):
continue
if self._safe_invoke(node, method_name, event):
handled = True
elif event.type == "mouse_motion":
for node, method_name in tuple(self._motion_handlers):
if self._safe_invoke(node, method_name, event):
handled = True
elif event.type == "scroll":
for node, method_name in tuple(self._scroll_handlers):
if self._safe_invoke(node, method_name, event):
handled = True
elif event.type == "joy_button" and event.joy_button is not None:
released = not event.pressed
for node, method_name in tuple(self._joy_button_handlers.get((event.joy_button, released), ())):
if self._safe_invoke(node, method_name, event):
handled = True
elif event.type == "joy_axis" and event.joy_axis is not None:
for node, method_name in tuple(self._joy_axis_handlers.get(event.joy_axis, ())):
if self._safe_invoke(node, method_name, event):
handled = True
# Catch-all fires for every event type.
for node, method_name in tuple(self._catch_all_handlers):
if self._safe_invoke(node, method_name, event):
handled = True
event.handled = event.handled or handled
# Unhandled chain: only fires if nothing consumed.
if not handled:
for node, method_name in self._collect_unhandled_handlers():
self._safe_invoke(node, method_name, event)
# -- Input dispatch table maintenance ---------------------------------
def _register_input_node(self, node: Node) -> None:
"""Insert *node*'s ``@on_input`` handlers into the dispatch tables."""
for method_name, filt in type(node)._simvx_input_handlers:
kind = filt["kind"]
target = filt["target"]
released = filt["released"]
mods = filt["mods"]
if kind == "action":
self._action_handlers.setdefault((target, released), []).append((node, method_name, mods))
elif kind == "key":
if len(target) == 1:
self._key_handlers.setdefault((target[0], released), []).append((node, method_name, mods))
else:
self._key_handlers_any[released].append((node, method_name, target, mods))
elif kind == "button":
self._button_handlers.setdefault((target, released), []).append((node, method_name, mods))
elif kind == "motion":
self._motion_handlers.append((node, method_name))
elif kind == "scroll":
self._scroll_handlers.append((node, method_name))
elif kind == "joy_button":
self._joy_button_handlers.setdefault((target, released), []).append((node, method_name))
elif kind == "joy_axis":
self._joy_axis_handlers.setdefault(target, []).append((node, method_name))
elif kind == "catch_all":
self._catch_all_handlers.append((node, method_name))
def _unregister_input_node(self, node: Node) -> None:
"""Remove *node*'s entries from the dispatch tables on tree-exit."""
def _drop(seq):
seq[:] = [t for t in seq if t[0] is not node]
for bucket in self._action_handlers.values():
_drop(bucket)
for bucket in self._key_handlers.values():
_drop(bucket)
for bucket in self._key_handlers_any.values():
_drop(bucket)
for bucket in self._button_handlers.values():
_drop(bucket)
_drop(self._motion_handlers)
_drop(self._scroll_handlers)
for bucket in self._joy_button_handlers.values():
_drop(bucket)
for bucket in self._joy_axis_handlers.values():
_drop(bucket)
_drop(self._catch_all_handlers)
def _actions_for_event(self, event: TreeInputEvent) -> list[str]:
"""Return action names whose bindings match *event*. Empty list if none."""
if event.type == "key":
target_key = event.key
target_button = None
target_joy_button = None
elif event.type == "mouse_button":
target_key = None
target_button = event.mouse_button
target_joy_button = None
elif event.type == "joy_button":
target_key = None
target_button = None
target_joy_button = event.joy_button
else:
return []
if target_key is None and target_button is None and target_joy_button is None:
return []
matched: list[str] = []
for action_name, bindings in self.input_map._actions.items():
for b in bindings:
if target_key is not None and b.key == target_key:
# Modifiers on a binding are required, never exclusive: a
# binding that names none matches whatever else is held.
if (b.ctrl and not event.ctrl) or (b.shift and not event.shift) or (b.alt and not event.alt):
continue
matched.append(action_name)
break
if target_button is not None and b.mouse_button == target_button:
matched.append(action_name)
break
if target_joy_button is not None and b.joy_button == target_joy_button:
matched.append(action_name)
break
return matched
def _collect_unhandled_handlers(self) -> list[tuple[Node, str]]:
"""Walk the tree to gather ``on_unhandled_input`` overrides + decorated handlers.
Computed lazily; tree mutations bump ``_structure_version`` so we
rebuild only when the topology changes.
"""
cached = getattr(self, "_unhandled_cache", None)
cached_version = getattr(self, "_unhandled_cache_version", -1)
if cached is not None and cached_version == self._structure_version:
return cached
result: list[tuple[Node, str]] = []
# One walk from the tree's own top node covers the singletons and the
# current scene: a node that is both is listed once. ``include_self`` is
# False because the top node declares no hooks of its own.
for node in self._tree_root.walk(include_self=False):
methods = type(node)._simvx_hooks.get("unhandled_input", ())
for m in methods:
result.append((node, m))
self._unhandled_cache = result
self._unhandled_cache_version = self._structure_version
return result
def _safe_invoke(self, node: Node, method_name: str, *args: Any) -> Any:
"""Call ``node.method_name(*args)`` with the same error containment Node uses.
Every input handler runs through here -- the eight decorated dispatch
tables, the catch-all, and the ``on_unhandled_input`` chain -- so
``Node.strict_errors`` decides for every exception type on all of them.
A node the containment has already disabled is skipped rather than
called again, which is what stops one broken script from filling the log
with the same error once per event.
"""
if node._script_error:
return None
try:
return getattr(node, method_name)(*args)
except Exception:
if Node.strict_errors:
raise
node._handle_script_error(method_name)
return None
[docs]
def render(self, renderer):
"""Draw the scene, then the overlay layer, into ``renderer``.
Runs inside this tree's input span: ``on_draw`` is game code like any
other hook, so a handler that reads ``Input`` or registers an action
sees this tree rather than the process-wide default. That matters for
an isolated tree, which is how the editor renders a game in play mode.
"""
with self.activate_input():
cam = self._current_camera_2d
_has = hasattr(renderer, "push_transform")
if _has and cam is not None:
# One Camera2D mapping: the renderer bakes exactly the
# matrix Camera2D.canvas_transform returns, which world_to_screen and the
# Family-B passes also use, so hit-testing and rendering agree.
renderer.push_transform(*cam.canvas_transform(self._screen_size))
if self.root:
self.root._draw_recursive(renderer)
if _has and cam is not None:
renderer.pop_transform()
# Overlay pass: the registry-driven on-top layer (dropdowns, dialogs,
# tooltips). Drawn after the main tree so submission order = GPU order =
# on top. ``if self.overlays`` is an O(1) empty check, so this is
# zero-cost when no overlay is open. The SAME iter_overlay_draws feeds the
# item pipeline (graphics), so desktop == web and headless == live.
if self.overlays:
from .ui.overlay import iter_overlay_draws
if hasattr(renderer, "reset_clip"):
renderer.reset_clip()
for kind, ctrl, entry in iter_overlay_draws(self, viewport=None):
if kind == "dim":
self._draw_overlay_scrim(renderer, self._screen_size, entry.dim_colour)
elif ctrl.visible:
ctrl._draw_recursive(renderer)
@staticmethod
def _draw_overlay_scrim(renderer, screen_size, colour):
"""Draw one full-screen scrim rect in screen space (overlay dim backdrop).
``reset_clip`` resets the scissor; the root walk already popped the
Camera2D transform before this pass, so the renderer is at the root screen
transform. Push identity defensively so a panned/zoomed Camera2D never
offsets the scrim, then draw the full-viewport rect.
"""
w, h = float(screen_size[0]), float(screen_size[1])
has_xf = hasattr(renderer, "push_transform")
if has_xf:
renderer.push_transform(1, 0, 0, 1, 0, 0)
renderer.draw_rect((0.0, 0.0), (w, h), colour=tuple(colour), filled=True)
if has_xf:
renderer.pop_transform()
[docs]
def group(self, name: str) -> list[Node]:
"""Get all nodes in a group."""
return list(self._groups.get(name, ()))
[docs]
def get_first_in_group(self, name: str) -> Node | None:
"""Return one node from ``name`` (any member), or ``None`` if empty."""
return next(iter(self._groups.get(name, ())), None)
[docs]
def call_group(self, name: str, method: str, *args) -> None:
"""Call ``method(*args)`` on every node in ``name`` that defines it.
Convenience for the common broadcast (``for n in tree.group(...):
n.method(...)``). Nodes lacking ``method`` are skipped, so mixed-type
groups are safe. The broadcast runs inside this tree's input span, so
the methods it calls see this tree's ``Input`` and ``InputMap`` whether
the caller was node code or a tool driving the tree from outside.
"""
with self.activate_input():
for node in self.group(name):
handler = getattr(node, method, None)
if callable(handler):
handler(*args)
[docs]
def create_timer(self, seconds: float) -> SceneTreeTimer:
"""Create a one-shot timer that emits ``timeout`` after ``seconds``.
Returns immediately with a :class:`SceneTreeTimer`; connect its
``timeout`` signal to run code after the delay. The timer is owned and
ticked by the tree (right after singletons) and discarded once it fires,
so it needs no Node in the scene and no manual cleanup. Honours
``tree.paused``. For delays inside a coroutine, prefer ``wait(seconds)``.
"""
timer = SceneTreeTimer(seconds)
self._scene_timers.append(timer)
return timer
[docs]
def call_deferred(self, method: Callable[..., Any], *args: Any) -> None:
"""Queue ``method(*args)`` to run at the end of this frame (escape hatch).
The tree-level counterpart to :meth:`Node.call_deferred`, for code that
has no node to defer from (plugins, singleton helpers, non-node systems).
Same discouraged-escape-hatch guidance applies: prefer a safe-by-default
path (``destroy``, ``Property(coalesce=True)``, ``events.publish_deferred``)
when one exists. ``method`` is a callable, never a string.
"""
self._deferred_calls.append((method, args))
# --- Singletons (nodes that persist across scene changes) ---
[docs]
@property
def singletons(self) -> dict[str, Node]:
"""Read-only view of registered singletons."""
return self._singletons
[docs]
def add_singleton(self, name: str, node: Node):
"""Register ``node`` as a persistent singleton attached to the tree.
An unparented node is mounted as a child of the tree's own top node: it
enters the tree, runs ``on_ready()`` immediately, processes ahead of the
scene in registration order, and survives ``change_scene()``. That makes
singletons the canonical home for global state (score, settings, audio
manager). Retrieve one via ``tree.singletons[name]``; it is not reachable
from the scene root. See :doc:`../patterns`.
A node the scene already owns may also be given a name. It stays where it
is, so it processes in its scene position rather than ahead of the scene,
and it leaves with its scene: after a ``change_scene`` the name is gone.
A singleton is ticked but not drawn -- drawing walks the scene. A global
HUD is either a child of the scene or a full-screen overlay.
The name ``"events"`` is reserved for the engine-provided
:class:`~simvx.core.event_bus.EventBus`; use ``tree.events`` instead.
Raises:
ValueError: the name is ``"events"``; the name is already bound to a
different node; ``node`` already holds a different name; or
``node`` is a child of some other tree.
"""
if name == RESERVED_SINGLETON_EVENTS:
raise ValueError(
"Singleton name 'events' is reserved for the engine-provided "
"EventBus. Access it via tree.events; pick a different name "
"for your singleton."
)
bound = self._singletons.get(name)
if bound is not None and bound is not node:
raise ValueError(
f"Singleton name {name!r} is already registered to {bound.name!r}. "
f"Call remove_singleton({name!r}) first, or pick a different name."
)
for other_name, other in self._singletons.items():
if other is node and other_name != name:
raise ValueError(
f"{node.name!r} is already registered as the {other_name!r} singleton. "
"A node holds at most one singleton name."
)
parent = node.parent
if parent is not None and node._tree is not self:
raise ValueError(
f"Cannot register {node.name!r} as a singleton: it is a child of "
f"{parent.name!r}, which is not in this tree. Detach it first, or "
"pass a node that has no parent."
)
self._singletons[name] = node
if parent is None:
self._tree_root.add_child(node)
if self._scene_root is not None:
# The scene stays the last child, so singletons keep processing
# ahead of it however late they register.
self._tree_root.children.move_last(self._scene_root)
[docs]
def remove_singleton(self, name: str):
"""Drop a singleton's name binding, and unmount it if the tree owns it.
A singleton the tree mounted (registered while unparented) is carried out
of the tree, so its ``on_exit_tree`` runs and anything it registered on
the way in is released. A singleton the scene owns keeps its place in the
scene: dropping the name is the whole of the operation, because the node
is a scene node that happens to have a name.
"""
node = self._singletons.pop(name, None)
if node is not None and node.parent is self._tree_root:
# Opens the input span itself and runs _exit_tree.
self._tree_root.remove_child(node)
def _prune_singletons(self) -> None:
"""Drop registry entries whose node has left this tree.
A singleton registered on a node the scene owns leaves with that scene,
exactly as a ``unique_name`` registration does. One pass over a dict that
holds single digits of entries, at scene-swap and delete-flush time only;
never on the per-frame path.
"""
dead = [name for name, node in self._singletons.items() if node._tree is not self]
for name in dead:
del self._singletons[name]
def _prune_dead_index_entries(self) -> None:
"""Drop group and unique-name entries naming a node that has left.
Every exit deregisters its own node (``Node._exit_tree_inner``), so this
finds nothing in a correct engine. It replaces the clear-and-rebuild
``change_scene`` used to do, and it is a detector rather than a cover-up:
under ``Node.strict_errors`` a survivor raises, so a deregistration bug
fails a test instead of being silently mopped up, and in a release build
it is mopped up so no game gets a dead node out of ``tree.group()``.
"""
stale = [(group, node) for group, members in self._groups.items() for node in members if node._tree is not self]
stale_unique = [n for n, node in self._unique_nodes.items() if node._tree is not self]
if (stale or stale_unique) and Node.strict_errors:
raise RuntimeError(f"Nodes left the tree without deregistering: groups={stale}, unique={stale_unique}")
for group, node in stale:
self._groups[group].discard(node)
for name in stale_unique:
del self._unique_nodes[name]
# --- Unique nodes ---
[docs]
def unique(self, name: str) -> Node | None:
"""Get a unique node by name. Returns None if not found."""
return self._unique_nodes.get(name)
# --- Internal helpers ---
def _group_add(self, group: str, node: Node):
if group not in self._groups:
self._groups[group] = set()
self._groups[group].add(node)
def _group_remove(self, group: str, node: Node):
if group in self._groups:
self._groups[group].discard(node)
def _invalidate_draw_caches(self):
"""Invalidate draw caches of Controls whose rect depends on the screen size.
A screen-size change moves or resizes every anchored control (any of
anchor_{left,top,right,bottom} != 0) and carries every Control in its
subtree along: descendants are laid out relative to the moved rect, so
their absolute coordinates change even with zero anchors of their own.
Controls outside every anchored subtree keep rects computed purely from
their own position/size, so their caches stay valid here: own-size
changes are handled by Property setters and own-position changes by
_invalidate_transform.
Both draw paths bake absolute screen coordinates, so both are
invalidated across the whole moved set: the legacy recorder cache
(_draw_dirty/_draw_cache) and the retained item pipeline's bits
(_render_dirty on each anchored control, whose rect may have changed
size as well as position; _transform_render_dirty on everything that
moved) that RenderItemCache scans to re-capture stale geometry.
The walk starts at the tree's own top node, not at the scene root, so a
Control that a singleton owns is reached as well. An overlay parented to
a singleton (the boot splash, the attribution watermark) is anchored to
the screen by definition, so a resize is exactly the event that moves it,
and starting at the scene root would leave its retained geometry pinned to
the corner the old screen had.
Uses an iterative stack to avoid Python recursion overhead on deep trees.
"""
from .ui import Control
stack = [(self._tree_root, False)]
while stack:
node, moved = stack.pop()
if isinstance(node, Control):
anchored = bool(node.anchor_left or node.anchor_top or node.anchor_right or node.anchor_bottom)
if anchored or moved:
moved = True
if anchored:
node._render_dirty = True
node._transform_render_dirty = True
if node._draw_cache is not None:
node._draw_dirty = True
node._draw_cache = None
children = node.children
if children:
stack.extend((child, moved) for child in children)
def _queue_delete(self, node: Node):
self._delete_queue.append(node)
[docs]
def clear_delete_queue(self) -> None:
"""Abandon every pending :meth:`Node.destroy` without running it.
For a teardown that is carrying the whole tree out anyway (a scene swap,
a harness dropping its tree): the queued nodes go with it, so the removal
they were waiting for is moot. Clears their pending state as well, since a
node left marked as destroying would report a removal that is never coming.
"""
for node in self._delete_queue:
node._destroying = False
self._delete_queue.clear()
def _flush_deletes(self):
"""Carry every node queued by :meth:`Node.destroy` out of the tree.
Runs at the end-of-frame sync point. A node queued *during* the drain
(an ``on_exit_tree`` that destroys a sibling, say) is picked up by the
same pass, so a cascade completes in the frame it was started in; the
queue is finite because :meth:`Node.destroy` is idempotent, and a node
this pass has finished is marked as such and can never queue again.
Exactly one exit per node. A queued node whose ancestor was destroyed
first has already been carried out by that ancestor's ``_exit_tree`` and
so holds no tree, leaving only the parent link to unlink
(``_detach_child``); running ``remove_child`` on it would fire its
``on_exit_tree`` a second time. A node that is still in a tree goes
through the full ``remove_child`` instead, so it never leaves a
half-registered entry behind in a tree that is still live.
"""
queue = self._delete_queue
if not queue:
return
i = 0
while i < len(queue):
node = queue[i]
i += 1
parent = node.parent
if parent is not None:
if node._tree is None:
parent._detach_child(node)
else:
parent.remove_child(node)
node._destroying = False
node._destroyed = True
queue.clear()
# A destroyed singleton has left the tree through the ordinary removal
# above, so its name binding goes with it.
self._prune_singletons()
def _flush_deferred_calls(self):
"""Drain calls queued via :meth:`Node.call_deferred` / :meth:`call_deferred`.
Runs once per frame at the end-of-frame sync point, outside tree
traversal. A snapshot swap means calls queued *during* the drain are
held for the next frame, so a self-requeuing deferred call cannot spin
the current frame.
A call bound to a node that has since left this tree is dropped (its
context is gone, matching Godot's freed-object behaviour). Node-bound
calls run through the owner's :meth:`Node._safe_call`, so a failing
deferred call obeys the same strict-raise / release-isolate policy as
every other hook dispatch instead of silently eating the rest of the
queue.
"""
if not self._deferred_calls:
return
pending = self._deferred_calls
self._deferred_calls = []
for fn, args in pending:
owner = getattr(fn, "__self__", None)
if isinstance(owner, Node):
if owner._tree is not self:
continue # node left the tree after queueing: drop the call
owner._safe_call(fn, *args)
elif Node.strict_errors:
fn(*args)
else:
try:
fn(*args)
except Exception:
log.exception("deferred call %r failed", fn)
# ========================================================================
# UI Input System (delegated to UIInputManager)
# ========================================================================
def _set_focused_control(self, control):
"""Set the focused control (forwarded to UIInputManager).
Spanned because the focus change emits ``focus_entered`` /
``focus_exited``, and this is the path ``Control.grab_focus`` and the
overlay registry take, neither of which is a dispatch already inside a
span.
"""
with self.activate_input():
self._ui._set_focused_control(control)
def _notify_ui_subtree_removed(self, control):
"""Release focus / mouse grab for a control leaving the tree (forwarded)."""
with self.activate_input():
self._ui.notify_subtree_removed(control)
def _update_mouse_over_states(self, mouse_pos):
"""Update mouse_over state for all controls (forwarded to UIInputManager).
Spanned because the update emits ``mouse_entered`` / ``mouse_exited``.
"""
with self.activate_input():
self._ui._update_mouse_over_states(self.root, mouse_pos)
def _find_control_at_point(self, point):
"""Find topmost control at screen position (forwarded to UIInputManager)."""
return self._ui._find_control_at_point(self.root, point)