"""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 contextlib import contextmanager
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 .physics.world import BodyHandle, PhysicsWorld
from .physics.world2d import Physics2DWorld
from .input.enums import JoyAxis, JoyButton, Key, MouseButton
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" autoload name is reserved for the engine-provided EventBus.
# Project TOMLs and runtime code may not declare or replace it.
RESERVED_AUTOLOAD_EVENTS = "events"
log = logging.getLogger(__name__)
@dataclass(slots=True)
class _BodySync:
"""Per-world handle->node sync state for the new physics seam.
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
[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 autoloads 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.
"""
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
[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
self.root: Node | None = None
# Depth counter for activate_input() spans (re-entrant): > 0 while this
# tree is processing (tick / input dispatch), used by the InputMap
# late-registration warning to exempt node-callback registrations.
self._in_processing: int = 0
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._autoloads: dict[str, Node] = {}
# Transient fire-and-forget timers created via ``create_timer``. Ticked
# right after autoloads 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 ``_autoloads`` because EventBus is not a Node and 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_autoload() and project.py).
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
# New transport seam (additive; runs alongside the old PhysicsServer until Stage 4).
# Both slots stay empty/lazy so non-physics scenes allocate nothing.
self._physics_world: PhysicsWorld | None = None # lazily-created default 3D world (new seam)
self._physics_world_2d: Physics2DWorld | None = None # lazily-created default 2D world (T2f)
# 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 (new seam, Stage 3a): 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[PhysicsWorld, _BodySync] = {}
# Handle->node map for collision-event dispatch (new seam, Stage R2a).
# 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[PhysicsWorld, weakref.WeakValueDictionary[BodyHandle, Node]] = {}
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. 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
self.quit_requested()
# Release the new-seam 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:
self._invalidate_draw_caches()
self.screen_resized(self._screen_size)
[docs]
def set_root(self, root: Node):
"""Set the root node of the scene tree.
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.
"""
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.root = root
self._register_declared_input_actions(root)
root._enter_tree(self)
root._ready_recursive()
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. Autoloads are left in place and their groups and unique-name
entries are re-registered on the rebuilt tree. 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.root)
if self.root:
self.root._exit_tree()
# Rebuild groups from autoloads only (scene groups were cleared by _exit_tree)
self._groups.clear()
for node in self._autoloads.values():
self._reregister_groups(node)
self._unique_nodes.clear()
for node in self._autoloads.values():
self._reregister_unique(node)
self._delete_queue.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.
"""
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
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: autoloads' ``_process`` first (Godot-style global singletons),
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 autoload pass is also drained
in the same flush, keeping the scene's view of the world consistent.
"""
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()
for node in self._autoloads.values():
node._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()
if self.root:
self.root._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 (new transport seam) ---------------------
[docs]
@property
def physics_world(self) -> PhysicsWorld:
"""The tree's default PhysicsWorld (new seam), 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 (T2f), 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 the 3D seam (``(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."""
try:
self._physics_worlds.remove(world)
except ValueError:
pass
# -- Physics body sync (new transport seam, node-agnostic above the seam) --
[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 unregister_physics_node(self, world: PhysicsWorld, handle: BodyHandle) -> None:
"""Unregister a body node from the collision-event map. No-op if absent.
Drops the whole per-world map once empty so idle worlds cost nothing.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
return
nodes.pop(handle, None)
if not nodes:
del self._physics_nodes[world]
def _dispatch_contact_events(self, world: PhysicsWorld) -> 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.
Node-agnostic seam: events are keyed by body handles; this resolves both
handles via :attr:`_physics_nodes` and builds one node-level
:class:`Contact` per side, with ``other`` set to the peer and
``normal`` / ``velocity`` reoriented for that side (the seam normal is
``a -> b``; the ``b`` side is negated). Skips an event if either node is
no longer tracked (GC'd / destroyed).
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
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 seam 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 = nodes.get(ev.a)
b = nodes.get(ev.b)
if a is None or b is None:
continue # GC'd or destroyed: 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, velocity=ev.rel_velocity)
)
b.collided(
Contact(other=a, point=ev.point, normal=ev.normal, impulse=ev.impulse, velocity=-ev.rel_velocity)
)
else: # EXIT: degenerate payload (no live manifold), only ``other`` is meaningful.
a.separated(Contact(other=b, point=ev.point, normal=ev.normal, impulse=0.0, velocity=ev.rel_velocity))
b.separated(Contact(other=a, point=ev.point, normal=ev.normal, impulse=0.0, velocity=ev.rel_velocity))
def _dispatch_overlap_events(self, world: PhysicsWorld) -> 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.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
return
events = world.drain_overlap_events()
if not events:
return
# Width-aware: resolve the Area / Body node types for this world's dim so
# the routing (body_* vs area_* by the OTHER node's type) is dimension-correct.
from .physics.world import ContactPhase
if self._world_dims(world)[1] == 2:
from .physics.nodes2d import Area2D as Area
from .physics.nodes2d import PhysicsBody2D as Body
else:
from .physics.nodes import Area3D as Area
from .physics.nodes import PhysicsBody3D as Body
for ev in events:
area = nodes.get(ev.sensor) # the detecting sensor MUST be an Area node
other = nodes.get(ev.other)
if area is None or other is None:
continue # GC'd or destroyed
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: PhysicsWorld) -> 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_pose_from_seam guards the assignment so the pose-reconcile
# hook does not bounce this simulated pose back into the handle.
node._write_pose_from_seam(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_pose_from_seam(Vec3(row[0], row[1], row[2]), Quat(row[6], row[3], row[4], row[5]))
def _capture_physics_world(self, world: PhysicsWorld) -> 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_pose_from_seam(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_pose_from_seam(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()
[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():
for node in self._autoloads.values():
node._physics_process_recursive(dt, paused)
if self.root:
self.root._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 and self._physics_world.body_count > 0:
self._physics_world.step(dt)
if self.physics_interpolation:
self._capture_physics_world(self._physics_world)
else:
self._sync_physics_world(self._physics_world)
self._dispatch_contact_events(self._physics_world)
self._dispatch_overlap_events(self._physics_world)
# Default 2D world (T2f): 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 and self._physics_world_2d.body_count > 0:
self._physics_world_2d.step(dt)
if self.physics_interpolation:
self._capture_physics_world(self._physics_world_2d)
else:
self._sync_physics_world(self._physics_world_2d)
self._dispatch_contact_events(self._physics_world_2d)
self._dispatch_overlap_events(self._physics_world_2d)
for world in self._physics_worlds:
if world.body_count > 0:
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)
# -- 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:
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]] = []
if self.root is not None:
for node in self.root.walk(include_self=True):
methods = type(node)._simvx_hooks.get("unhandled_input", ())
for m in methods:
result.append((node, m))
for autoload in self._autoloads.values():
for node in autoload.walk(include_self=True):
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."""
if node._script_error:
return None
try:
return getattr(node, method_name)(*args)
except AssertionError:
raise
except Exception:
if Node.strict_errors:
raise
node._script_error = True
import sys
import traceback
tb = traceback.format_exc()
print(f"Script error in {node.name}.{method_name}: node disabled:\n{tb}", file=sys.stderr)
log.error("Script error in %s.%s: node disabled", node.name, method_name)
try:
Node.script_error_raised.emit(node, method_name, tb)
except Exception:
# A handler of the global script-error signal itself raised.
# Log and continue: error-recovery code must not derail.
log.debug("script_error_raised handler failed", exc_info=True)
return None
[docs]
def render(self, renderer):
cam = self._current_camera_2d
_has = hasattr(renderer, "push_transform")
if _has and cam is not None:
# One Camera2D mapping (design SS7.3): 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 get_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.get_group(...):
n.method(...)``). Nodes lacking ``method`` are skipped, so mixed-type
groups are safe.
"""
for node in self.get_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 autoloads) 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, autoload 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))
# --- Autoloads (singletons that persist across scene changes) ---
[docs]
@property
def autoloads(self) -> dict[str, Node]:
"""Read-only view of registered autoloads."""
return self._autoloads
[docs]
def add_autoload(self, name: str, node: Node):
"""Register ``node`` as a persistent singleton attached to the tree.
The node enters the tree and runs ``on_ready()`` immediately. Unlike a
regular child, it is not reachable via the scene root: retrieve it via
``tree.autoloads[name]``. Autoloads survive ``change_scene()``, making
them the canonical home for global state (score, settings, audio
manager). See :doc:`../patterns`.
The name ``"events"`` is reserved for the engine-provided
:class:`~simvx.core.event_bus.EventBus`; use ``tree.events`` instead.
"""
if name == RESERVED_AUTOLOAD_EVENTS:
raise ValueError(
"Autoload name 'events' is reserved for the engine-provided "
"EventBus. Access it via tree.events; pick a different name "
"for your autoload."
)
self._autoloads[name] = node
node._enter_tree(self)
node._ready_recursive()
[docs]
def remove_autoload(self, name: str):
"""Unregister and tear down an autoload. Calls ``_exit_tree`` on the node."""
node = self._autoloads.pop(name, None)
if node:
node._exit_tree()
# --- Unique nodes ---
[docs]
def get_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 _reregister_groups(self, node: Node):
"""Re-add a node (and descendants) to the group index."""
for group in node._groups:
self._group_add(group, node)
for child in node.children:
self._reregister_groups(child)
def _reregister_unique(self, node: Node):
"""Re-add a node (and descendants) to the unique-node index."""
if node.unique_name:
self._unique_nodes[node.name] = node
for child in node.children:
self._reregister_unique(child)
def _invalidate_draw_caches(self):
"""Invalidate draw caches of Controls whose rect depends on screen/parent size.
Only anchored controls (any of anchor_{left,top,right,bottom} != 0) have
their absolute rect affected by a screen-size change. Controls with the
default zero anchors have 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.
Uses an iterative stack to avoid Python recursion overhead on deep trees.
"""
if not self.root:
return
from .ui import Control
stack = [self.root]
while stack:
node = stack.pop()
if isinstance(node, Control) and node._draw_cache is not None:
if node.anchor_left or node.anchor_top or node.anchor_right or node.anchor_bottom:
node._draw_dirty = True
node._draw_cache = None
children = node.children
if children:
stack.extend(children)
def _queue_delete(self, node: Node):
self._delete_queue.append(node)
def _flush_deletes(self):
for node in self._delete_queue:
if node.parent:
node.parent.remove_child(node)
self._delete_queue.clear()
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)."""
self._ui._set_focused_control(control)
def _notify_ui_subtree_removed(self, control):
"""Release focus / mouse grab for a control leaving the tree (forwarded)."""
self._ui.notify_subtree_removed(control)
def _update_mouse_over_states(self, mouse_pos):
"""Update mouse_over state for all controls (forwarded to UIInputManager)."""
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)