"""Node: Base node class with tree hierarchy, groups, and coroutine support."""
from __future__ import annotations
import ast
import difflib
import functools
import inspect
import logging
import sys
import textwrap
from collections.abc import Callable, Iterator, Mapping
from time import perf_counter
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, overload
from ._draw_context import READS_VIEWPORT_INSETS, READS_VIEWPORT_SIZE, draw_context
from .decorators import TargetedHandlers, collect_hooks, collect_targeted
from .descriptors import Children, Coroutine, CoroutineHandle, NodeNotFound, Notification, Property, UpdateMode
from .input.contexts import InputContext
from .input.controls import DeviceButton, roles_of
from .input.drag import DEFAULT_DRAG_THRESHOLD, drag_threshold_map
from .input.enums import FocusMode, FocusReason, PointerType
from .input.events import GuiInputEvent, InputBinding, InputEvent, PickEvent, PointerEvent, modifiers_held
from .input.players import InputPlayer
from .math.types import Vec2
from .signals import Signal
log = logging.getLogger(__name__)
# How many nodes in the process declare an ``input_player``. Zero for every
# single-player game, and the one test the hook walks and the input dispatch
# make before they spend anything on the player dimension at all.
_PLAYER_SCOPES = 0
# Set to observe what the frame walk visited and what each node cost, which is
# what a profiler needs and what it would otherwise have to walk the tree itself
# to get. ``None`` for every ordinary run, and the one test each node makes
# before it spends anything on timing, the same trade as ``_PLAYER_SCOPES``.
_NODE_OBSERVER: Callable[[Node, str, float], None] | None = None
[docs]
def observe_node_walk(observer: Callable[[Node, str, float], None] | None) -> None:
"""Be told each node the frame walk visits, in which phase, and its cost in ms.
``observer`` receives ``(node, phase, ms)`` with ``phase`` either
``"process"`` or ``"physics"``, once per node per phase, measured around that
node's own hooks and not its children's. Pass ``None`` to stop.
There is one observer for the process, because there is one frame walk. A
profiler is the caller this exists for: it means nothing has to re-implement
the walk to time it, and a walk of one's own silently does less than the
engine's.
"""
global _NODE_OBSERVER
_NODE_OBSERVER = observer
def _default_player() -> InputPlayer:
"""The default player, for a node no tree holds."""
from .input.state import Input
return Input.players[0]
if TYPE_CHECKING:
from .scene_tree import SceneTree
# Lazy-cached reference for circular import
_ui_Control: type | None = None
def _get_control() -> type:
global _ui_Control
if _ui_Control is None:
from .ui import Control
_ui_Control = Control
return _ui_Control
def _where_declared(klass: type) -> str:
"""Where a class was declared, said the way a developer would look for it.
A scene file's module name is derived from its path and reads as noise, so
prefer the file itself; fall back to the module name for a class whose
module is not (yet) importable.
"""
module = sys.modules.get(klass.__module__)
path = getattr(module, "__file__", None)
return f"{path} ({klass.__qualname__})" if path else f"{klass.__module__}.{klass.__qualname__}"
def _init_calls_super(func) -> bool:
"""Check via AST whether *func* contains a super().__init__(...) call."""
try:
source = textwrap.dedent(inspect.getsource(func))
tree = ast.parse(source)
except OSError, TypeError, IndentationError:
return True # Cannot inspect: assume user handles super
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
f = node.func
# super().__init__(...) or super(Cls, self).__init__(...)
if isinstance(f, ast.Attribute) and f.attr == "__init__" and isinstance(f.value, ast.Call):
inner = f.value
if isinstance(inner.func, ast.Name) and inner.func.id == "super":
return True
return False
#: The class-level namespaces a node's managed state is declared in.
#:
#: Managed state is state some machinery other than plain attribute lookup
#: maintains: a ``Property`` (validated, inspector-visible, scene-emitted), a
#: plain ``property`` (a computed accessor over other state), or one of the
#: three ``_ThemeAttr`` registries (a value read from the active theme unless
#: overridden). Each keeps the value somewhere other than the class attribute
#: the name is bound to, which is what makes a class-level rebinding of the
#: name silent rather than loud: see :func:`_managed_kind`.
_THEME_REGISTRIES: tuple[str, ...] = ("__theme_colours__", "__theme_sizes__", "__theme_styleboxes__")
def _managed_kind(owner: type, name: str) -> str | None:
"""How ``owner``'s own class body declares ``name``, or ``None`` if it does not.
Answers for one class, not for its bases: the caller walks the MRO. Only
the class's own ``__dict__`` is consulted, because an inherited declaration
belongs to the class that made it and would otherwise be reported against
every class below it.
"""
declared = owner.__dict__.get(name)
if isinstance(declared, Property):
return "the Property"
if isinstance(declared, property):
return "the property"
for registry in _THEME_REGISTRIES:
if name in owner.__dict__.get(registry, ()):
return "the theme attribute"
return None
def _reject_shadowed_managed_names(cls: type) -> None:
"""Refuse a subclass that rebinds an inherited managed name to a plain value.
``class DemoRunner(Node): update_mode = UpdateMode.ALWAYS`` reads back as
``ALWAYS`` and changes nothing: the class attribute replaces the descriptor,
while the engine keeps reading the storage behind it, which still holds the
default. The mismatch is invisible at the assignment and shows up much later
as behaviour that ignores a value the class plainly states, so it is refused
where it is written.
Only a plain value is refused. Rebinding the name to another descriptor is
how a subclass legitimately redeclares managed state -- a ``Property`` with
a different default, an overriding ``property``, a ``ThemeColour`` naming a
different theme key -- so anything with a ``__get__`` is left alone, methods
included.
"""
for name, value in cls.__dict__.items():
if name.startswith("__") or hasattr(type(value), "__get__"):
continue
for base in cls.__mro__[1:]:
kind = _managed_kind(base, name)
if kind is None:
continue
raise TypeError(
f"{cls.__name__}.{name} = {value!r} shadows {kind} declared on "
f"{base.__name__}: the class attribute replaces the descriptor, so reads "
f"see this value while the engine goes on reading the storage behind it. "
f"Assign it on the instance (self.{name} = ...), or redeclare it as a "
f"descriptor of its own to change the default."
)
T = TypeVar("T", bound="Node")
D = TypeVar("D")
class _NoDefault:
"""Marks "no default given" for :meth:`Node.node_at`, so ``None`` can be one."""
def __repr__(self) -> str:
return "<no default>"
_NO_DEFAULT = _NoDefault()
def _node_matcher(target: type[Node] | str | Callable[[Node], bool]) -> Callable[[Node], bool]:
"""Build a predicate from a :meth:`Node.find` target.
A Node subclass matches by ``isinstance``, a ``str`` matches an exact
``name``, and a callable is used directly as the predicate.
"""
if isinstance(target, type):
return lambda n: isinstance(n, target)
if isinstance(target, str):
return lambda n: n.name == target
if callable(target):
return target
raise TypeError(f"find() target must be a Node subclass, a name str, or a predicate; got {type(target).__name__}")
#: How many declared names to list when a misspelt kwarg has no near miss. A
#: ``Control`` subclass declares dozens, and a wall of them buries the error it
#: is attached to.
_KWARG_NAMES_SHOWN = 12
def _unknown_kwarg_message(cls: type, key: str, props: Any) -> str:
"""The ``TypeError`` text for a kwarg the class does not declare.
Names up to three close matches when there are any, since the name a caller
wants is usually one edit away. Otherwise it lists what the class declares
rather than inventing a plausible-looking substitute.
"""
declared = sorted(props)
near = difflib.get_close_matches(key, declared, n=3)
head = f"{cls.__name__}: unknown kwarg {key!r}"
if near:
options = " or ".join(repr(n) for n in near)
return f"{head}. Did you mean {options}?"
if not declared:
return f"{head}, and it declares no Property at all."
shown = ", ".join(declared[:_KWARG_NAMES_SHOWN])
more = "" if len(declared) <= _KWARG_NAMES_SHOWN else f", and {len(declared) - _KWARG_NAMES_SHOWN} more"
return f"{head}, and nothing it declares is close to it. {cls.__name__} declares: {shown}{more}."
[docs]
class Node:
"""Base node with tree hierarchy, groups, and coroutine support.
Attributes:
name: Unique name within the parent's children. Defaults to the class name.
parent: The parent ``Node``, or ``None`` if this is the root.
children: Ordered collection of child nodes, accessible by name or index.
visible: Whether this node (and its descendants) should be drawn.
update_mode: Controls processing behaviour during pause
(``INHERIT``, ``PAUSABLE``, ``PAUSED_ONLY``, ``ALWAYS``, ``DISABLED``).
unique_name: When ``True``, the node is registered in the tree for
fast lookup via ``SceneTree.get_unique_node()``.
Example::
root = Node(name="Root")
child = Node(name="Child")
root.add_child(child)
assert child.parent is root
assert root.children["Child"] is child
"""
_registry: ClassVar[dict[str, type]] = {}
#: Bumped by every subclass declaration, including one that rebinds a name
#: already in the registry. A cache over the registry keys on this: the
#: registry's own length does not move when a reload replaces a class, and
#: the replacement may declare different properties.
_registry_version: ClassVar[int] = 0
# Raise on script errors; set False for release. Governs EVERY exception class a
# hook can raise, ``AssertionError`` included: an assertion inside a hook is fatal
# in the default dev/test mode and contained per node when the flag is cleared.
# A carve-out for ``assert`` would be a runtime contract the language does not
# offer, since ``python -O`` removes the statement outright.
strict_errors: ClassVar[bool] = True
#: Whether to run the engine's own internal consistency checks: the ones that
#: cost something on every frame and exist to catch an engine defect, not a
#: game's. Distinct from ``strict_errors``, which is a policy about a user
#: script's exceptions.
#:
#: Defaults to ``sys.flags.dev_mode`` (``python -X dev``), which is what
#: asyncio keys its equivalent on and is False in an ordinary run. It is
#: deliberately NOT ``__debug__``, which is True unless ``python -O`` is
#: passed and would therefore leave these checks armed in most shipped games.
dev_checks: ClassVar[bool] = sys.flags.dev_mode
#: Whether to run the checks that catch mistakes in game code: the ones that
#: cost something on every frame or event and exist only to tell a developer
#: what to fix. Each reports a given mistake once, through :mod:`warnings`
#: under :class:`~simvx.core.DevWarning`, so it can be filtered or turned into
#: an error. Distinct from :attr:`dev_checks`, which catches engine and backend
#: defects, and from ``strict_errors``, which is the script-error policy.
#:
#: Covers the 2D freeze detector (a node whose ``on_draw`` changed without
#: ``dynamic`` or ``queue_redraw()``, sampled every
#: :attr:`freeze_check_interval` seconds), a screen-anchored ``Control`` that
#: renders through a panning world camera, and a typed ``Signal`` connected to
#: a callback that cannot take its arguments.
#:
#: Defaults to ``sys.flags.dev_mode``, like :attr:`dev_checks`, so a shipped
#: game never runs them; set it to True while developing.
dev_warnings: ClassVar[bool] = sys.flags.dev_mode
#: Seconds between two freeze-detector samples under :attr:`dev_warnings`, on
#: the tree's clock (``SceneTree.now``, which stops while the tree is paused).
#: Each sample re-draws one node, so a scene of N retained nodes is covered
#: every N intervals. ``0.0`` samples every frame.
freeze_check_interval: ClassVar[float] = 1.0
# -- 2D render-retention contract ----------------------------------------
# ``on_draw`` lives on the base ``Node`` (any node -- plain HUD/menu, Node2D,
# CanvasLayer, even a 3D billboard -- may define it), so the bits that let the
# retained 2D item pipeline (``render2d.RenderItemCache``) know WHEN to re-run a
# node's ``on_draw`` live here too.
#
# ``dynamic`` -- PUBLIC, runtime-toggleable. ``True`` means "my ``on_draw`` reads
# non-Property state (``tree.now`` animation, a frame counter, a live feed), so
# re-collect me every frame." The self-documenting immediate-mode escape hatch.
# ``_render_dirty`` -- appearance/geometry changed; re-capture this node's
# ``on_draw`` next frame. Set by :meth:`queue_redraw`; drained by the cache.
# ``_render_auto_dirty`` -- whether a ``Property`` write auto-dirties this node
# (the blanket ``descriptors.py`` hook). ``False`` here so non-2D nodes (3D,
# audio, camera) never pollute the 2D dirty scan; ``Drawable2D`` sets it ``True``.
dynamic: bool = False
_render_dirty: bool = False
_render_auto_dirty: bool = False
# ``_ambient_reads`` -- the ambient values (``_draw_context.AmbientRead``) this
# node's last capture read; a change to one of them calls ``queue_redraw``.
_ambient_reads: int = 0
# -- Declared state every node has ---------------------------------------
# Both are authoring state -- a user hides a node or takes it out of the
# pause in the editor and expects the scene to save that way -- so both are
# declared Properties.
#
# Both stay storage-backed. Their setters do work beyond storing (a cache
# invalidation, a subtree walk), but the value itself lives exactly where
# ``Property`` would put it, which inside this class body is
# ``self.__update_mode`` and ``self.__visible`` (a Property stores under
# the declaring class's mangled name), so the
# extra work belongs in an ``on_change`` hook and NOT in an overriding
# ``__set__``. A derived Property (``storage_backed = False``, both
# accessors overridden) is for a value with no slot of its own; declaring
# one here would claim there is no storage when there is, and would cost a
# subclass the ability to redeclare the default.
#
# ``on_change`` fires only on a real change, so re-assigning the current
# value does no work.
visible = Property(
True,
coerce=bool,
hint="Whether this node and its subtree are drawn and picked",
on_change="_on_visible_changed",
)
update_mode = Property(
UpdateMode.INHERIT,
hint="Processing behaviour while the tree is paused",
on_change="_invalidate_update_mode_cache",
)
# Whether this node can hold keyboard focus, and how it takes it. Any node
# can: a sprite a pad cursor lands on is as focusable as a button. The tree
# keeps an index of the nodes that can, so Tab never walks the rest.
focus_mode = Property(
FocusMode.NONE,
hint="Whether this node can take keyboard focus, and how",
on_change="_on_focus_mode_changed",
)
# How far a pointer travels from a press this node takes before the
# motion is a drag, per pointer type. ``None``, or a type the map leaves
# out, inherits from the parent and at the top from ``tree.drag_threshold``,
# so a card table can ask more of a finger without touching anything else.
drag_threshold = Property(
None,
coerce=drag_threshold_map,
hint="Travel before a press is a drag, per pointer type; None inherits",
)
#: This node gained keyboard focus.
focus_entered = Signal()
#: This node lost keyboard focus.
focus_exited = Signal()
# Engine kwargs consumed by Node.__init__, not forwarded to user __init__ unless explicitly accepted
_NODE_INIT_KWARGS = frozenset({"name"})
# Stamped on by ``Property.__set_name__``: every Property declared on this
# class, merged with the ones its bases declare. Declared without a value so
# the descriptor's own "is it in this class's __dict__ yet" test still sees
# a fresh subclass as fresh; :meth:`get_properties` is the read for anything
# outside the constructor, which reads it directly to stay off a call.
__properties__: ClassVar[dict[str, Property]]
# Filled by __init_subclass__: hook name -> tuple of method names to invoke per dispatch.
_simvx_hooks: ClassVar[dict[str, tuple[str, ...]]] = {}
# Filled by __init_subclass__: ordered tuple of (method_name, filter_dict) for input handlers.
_simvx_input_handlers: ClassVar[tuple[tuple[str, dict[str, Any]], ...]] = ()
# Filled by __init_subclass__: the class's @on_gui_input and @on_picked handlers,
# indexed by input, or None where it declares none.
_simvx_gui_handlers: ClassVar[TargetedHandlers | None] = None
_simvx_pick_handlers: ClassVar[TargetedHandlers | None] = None
#: Whether the targeted GUI phase offers this class anything: it overrides
#: ``on_gui_input`` or declares an ``@on_gui_input`` handler. The router's
#: bubble skips a node whose class answers False.
_simvx_hears_gui: ClassVar[bool] = False
#: Whether the class declares an ``@on_picked`` handler for a release. A
#: tree casts a ray on a pointer's release only while it holds such a node.
_simvx_picks_release: ClassVar[bool] = False
# Set by SceneTree._register_input_node while this node is registered for input, and
# popped on the way out: the buckets it put entries in, so leaving reaches exactly
# those even if the class changed in between. Declared without a value on purpose --
# its presence in ``__dict__`` is what says the node is registered.
_input_plan: tuple[tuple, ...]
#: Position among siblings, assigned by ``Children`` and renumbered only on
#: a reorder, so an append or a removal moves nobody else's. Input delivery
#: order is built from these and cached against them.
_sibling_rank: int = 0
#: Where this node's ``input_context`` sits in the active stack, or -1 for
#: a node that holds none. ``InputMap`` writes it on every stack change, so
#: a handler under a hiding context finds its own rank by walking up rather
#: than by searching the stack.
_input_context_index: int = -1
#: The context a class body declared, lifted off the class by
#: ``__init_subclass__`` and given to every instance at construction.
_declared_input_context: ClassVar[InputContext | None] = None
# ``on_input`` is here so a plain override is dispatched like every other hook;
# having no filters, it registers as the catch-all. That is why ``Node`` carries no
# documentation-only ``on_input`` stub the way it does for the other hooks: a stub in
# the base would put every node in the tree on the catch-all list.
_PRIMARY_HOOK_METHODS: ClassVar[tuple[str, ...]] = (
"on_ready",
"on_update",
"on_fixed_update",
"on_enter_tree",
"on_exit_tree",
"on_draw",
"on_draw_3d",
"on_input",
)
# Bare names commonly mistaken for SimVX hooks (the engine only dispatches the on_-prefixed forms).
# ``update``/``fixed_update`` are deliberately NOT reserved: the on_ prefix already separates the
# hooks from user code, so a plain ``def update()`` helper on a Node subclass is allowed.
_BARE_HOOK_NAMES: ClassVar[tuple[str, ...]] = (
"ready",
"draw",
"input",
)
# Retired hook names, mapped to the hook that replaces each. The engine dispatches by
# name, so a method under one of these would silently never run; a class defining one
# is refused instead. A renamed or deleted hook adds its name here. Only names the
# engine dispatched belong here: anything else would refuse a user's own method.
_RETIRED_HOOK_NAMES: ClassVar[dict[str, str]] = {
"on_process": "on_update",
"on_physics_process": "on_fixed_update",
"on_unhandled_input": "a bare @on_input handler, which runs last and only if nobody claimed",
"_on_gui_input": "on_gui_input",
}
[docs]
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
previous = Node._registry.get(cls.__name__)
if previous is not None and previous.__module__ != cls.__module__:
# Two different modules, one name: the registry is keyed on the bare
# name and keeps only the last, so a snapshot that names this class
# without a module hint resolves to whichever was declared later.
# Re-running ONE module is not this case and must stay quiet -- that
# is hot reload, and it rebinds the name from the same module.
log.warning(
"Two node classes are named %r, in %s and %s. The later one wins wherever the "
"name alone has to resolve it: the node listing, editor completions, and a "
"snapshot that recorded no module.",
cls.__name__,
_where_declared(previous),
_where_declared(cls),
)
Node._registry[cls.__name__] = cls
Node._registry_version += 1
# ``input_context = {...}`` in a class body is the declarative spelling,
# so the value is lifted off the class here and the property below stays
# reachable for the imperative one. Without this the class attribute
# would replace the property and the assignment form would stop working.
if "input_actions" in cls.__dict__:
raise TypeError(
f"{cls.__name__}.input_actions is gone: write input_context = {{...}} with the same "
"entries. Any node may declare one; it is in force while the node is in the tree."
)
declared_context = cls.__dict__.get("input_context")
if not (declared_context is None or isinstance(declared_context, Mapping | InputContext | property)):
raise TypeError(
f"{cls.__name__}.input_context takes a dict of entries or an InputContext, "
f"got {type(declared_context).__name__}"
)
if isinstance(declared_context, Mapping):
declared_context = InputContext(declared_context, name=cls.__name__)
if isinstance(declared_context, InputContext):
delattr(cls, "input_context")
cls._declared_input_context = declared_context
# Lint: catch a class-level value that silently replaces an inherited descriptor.
_reject_shadowed_managed_names(cls)
# Lint: catch bare hook names (e.g. `def ready(self)`) that the engine silently never invokes.
for name in cls._BARE_HOOK_NAMES:
if name in cls.__dict__ and callable(cls.__dict__[name]):
raise TypeError(f"{cls.__name__}.{name}: not a SimVX hook: did you mean 'on_{name}'?")
# Lint: refuse a retired hook name, which the engine would never call.
for name, replacement in cls._RETIRED_HOOK_NAMES.items():
if name in cls.__dict__ and callable(cls.__dict__[name]):
raise TypeError(
f"{cls.__name__}.{name}: {name} is no longer dispatched; use {replacement}. "
f"The engine dispatches by name, so leaving it there would never call it."
)
# Collect lifecycle and input handlers (decorated + same-named overrides)
# walking the MRO most-derived-last, mirroring Property.__set_name__.
cls._simvx_hooks, cls._simvx_input_handlers = collect_hooks(cls, cls._PRIMARY_HOOK_METHODS)
cls._simvx_gui_handlers, cls._simvx_pick_handlers = collect_targeted(cls)
cls._simvx_hears_gui = cls._simvx_gui_handlers is not None or cls.on_gui_input is not Node.on_gui_input
cls._simvx_picks_release = cls._simvx_pick_handlers is not None and cls._simvx_pick_handlers.picks_release
# Auto-super: wrap user __init__ that doesn't call super().__init__
if "__init__" not in cls.__dict__:
return # No custom __init__: nothing to wrap
if cls.__dict__.get("__auto_init__") is False:
return # Opted out
user_init = cls.__dict__["__init__"]
if _init_calls_super(user_init):
return # User handles super(): don't wrap
user_sig = inspect.signature(user_init)
user_params = user_sig.parameters
# Determine which params (beyond self) the user accepts
user_param_names = [n for n in user_params if n != "self"]
has_var_keyword = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in user_params.values())
has_var_positional = any(p.kind == inspect.Parameter.VAR_POSITIONAL for p in user_params.values())
# Positional param names (POSITIONAL_ONLY or POSITIONAL_OR_KEYWORD), in order
positional_names = [
n
for n, p in user_params.items()
if n != "self" and p.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
]
# Find the parent __init__ to call (the next in MRO that is not the user's)
parent_init = None
for base in cls.__mro__[1:]:
if "__init__" in base.__dict__:
parent_init = base.__dict__["__init__"]
break
if parent_init is None:
parent_init = Node.__init__
# Inspect parent __init__ signature to know what it accepts.
# follow_wrapped=False gets the actual wrapper signature (not the original
# user function's signature that functools.wraps copies).
try:
parent_sig = inspect.signature(parent_init, follow_wrapped=False)
parent_param_names = {n for n in parent_sig.parameters if n != "self"}
parent_has_var_kw = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parent_sig.parameters.values())
except ValueError, TypeError:
parent_param_names = set()
parent_has_var_kw = True # Assume flexible
has_on_change = getattr(cls, "_has_on_change_hooks", False)
@functools.wraps(user_init)
def _wrapped_init(self, *args, **all_kwargs):
# Map positional args to their parameter names so we can split by name
user_args = args
if args and not has_var_positional:
for i, val in enumerate(args):
if i < len(positional_names):
pname = positional_names[i]
if pname in all_kwargs:
raise TypeError(f"__init__() got multiple values for argument '{pname}'")
all_kwargs[pname] = val
user_args = () # All positional mapped to kwargs
# Split kwargs: parent_kw goes to parent __init__, user_fwd goes to user's __init__
props = cls.get_properties()
parent_kw = {}
user_fwd = {}
for k, v in all_kwargs.items():
is_user_param = k in user_param_names
is_parent_named = k in parent_param_names # Explicitly named in parent sig
is_prop = k in props
is_engine = k in Node._NODE_INIT_KWARGS
if is_user_param:
user_fwd[k] = v
if is_parent_named or is_prop or is_engine:
parent_kw[k] = v # Also pass to parent (e.g. 'name', properties)
elif is_parent_named or is_prop or is_engine:
parent_kw[k] = v # Known parent/engine/property kwarg
elif has_var_keyword:
user_fwd[k] = v # Unknown kwarg, user accepts **kwargs
elif parent_has_var_kw:
parent_kw[k] = v # Fallback to parent **kwargs (will warn)
else:
raise TypeError(f"{cls.__name__}.__init__() got unexpected keyword argument {k!r}")
we_set_init_scope = has_on_change and not getattr(self, "_on_change_init", False)
if we_set_init_scope:
self._on_change_init = True
self._on_change_pending = []
try:
# Initialise via parent chain (e.g. Node2D.__init__ -> Node.__init__)
parent_init(self, **parent_kw)
# Forward to user's __init__
if has_var_keyword or has_var_positional:
user_init(self, *user_args, **user_fwd)
elif user_param_names:
forward = {k: v for k, v in user_fwd.items() if k in user_param_names}
user_init(self, *user_args, **forward)
else:
user_init(self)
finally:
if we_set_init_scope:
self._on_change_init = False
self._flush_on_change_pending()
cls.__init__ = _wrapped_init
# Class-level marker that lets ``_draw_recursive`` cheaply detect CanvasLayer
# children without importing the subclass (which lives in a sibling module
# and would otherwise force a per-frame ``from ... import CanvasLayer``).
# ``CanvasLayer`` overrides this to ``True``.
_is_canvas_layer: bool = False
# Class-level marker for a SceneTree's own top node, the parent of every
# singleton and of the current scene. Upward walks (``path``, ``node_at``)
# stop below it, so a scene root reports ``/Root``. ``SceneTree._TreeRoot``
# is the only class that sets it.
_is_tree_root: bool = False
#: What the engine worked out for this node, by property name
#: (:meth:`_record_derived`). Shared and unwritable until a node has
#: something to record, so the ordinary node -- which is nearly all of them
#: -- allocates nothing and reads an empty mapping.
_derived: Mapping[str, Any] = MappingProxyType({})
# Plain ``property`` attributes a constructor keyword may set, alongside the
# declared Properties: views over Properties, such as a control's ``size``
# over ``size_x`` and ``size_y``, which a scene saves through the Properties.
_constructor_views: frozenset[str] = frozenset()
def _record_derived(self, name: str, value: Any) -> None:
"""Remember that the engine, not the author, put ``value`` in ``name``.
A control measures its own size; a container places its children. Both
write into the same properties an author writes into, so without this
record an editor save would replace the author's value with the
engine's arithmetic result.
A recorded value is not emitted while the property still holds it
(:func:`~simvx.core.scene_io.emitter.iter_runtime_kwargs`). Any other
write simply makes the two differ, so no write path is instrumented,
nothing has to be cleared, and there is no per-frame cost.
"""
derived = self.__dict__.get("_derived")
if derived is None:
derived = {}
self._derived = derived
derived[name] = value
def __init__(self, name: str = "", **kwargs):
# The empty default is "no name given", and the node takes its type's
# name. It is the only way to spell that: the setter refuses an empty
# name, because a node that held one could be neither found nor saved.
if name and not isinstance(name, str):
raise TypeError(f"Node name must be a string, got {type(name).__name__}")
self._name = name or type(self).__name__
self.parent: Node | None = None
self.children = Children(self)
# Count of direct children whose ``_is_canvas_layer`` flag is True.
# Maintained by add_child / remove_child so ``_draw_recursive`` can
# skip the sort/partition fast path when zero.
self._canvas_layer_child_count: int = 0
self._tree: SceneTree | None = None
self._coroutines: list[Coroutine] = []
self._groups: set[str] = set()
self._scene_template_path: str | None = None
# Prime the two declared Properties' storage from the DECLARED default
# rather than from a literal, so a subclass that redeclares either one
# (``visible = Property(False)`` on a dialog that opens hidden) starts
# where it says it does. Priming at all is what lets the visibility walk
# below read ``self.__visible`` (the Property's own storage) directly
# rather than through the descriptor, and it is what gives
# ``_visible_in_hierarchy`` its opening value.
declared = type(self).__properties__
self.__visible: bool = declared["visible"].default
self._visible_in_hierarchy: bool = self.__visible
self.__update_mode: UpdateMode = declared["update_mode"].default
self._cached_update_mode: UpdateMode | None = None # cached resolved mode
self._declared_player: InputPlayer | None = None # set by input_player
self._cached_player: int | None = None # cached resolved player id
# Declared in the class body, or assigned later through the property.
# Not activated here: a context is live only while its node is in a
# tree, and this node is not in one yet.
self._input_context: InputContext | None = type(self)._declared_input_context
self.unique_name: bool = False
self._script_error: bool = False
self._destroying: bool = False # queued by destroy(), not yet flushed
self._destroyed: bool = False # destroy() has run its course: this node is finished
self._exiting: bool = False # inside this node's own _exit_tree
self._ready_called: bool = False # on_ready has fired, once and for all
self._outgoing_connections: list = [] # signals connected via this node's bound methods
# Open the on_change deferral scope if this class has any on_change
# hooks AND no enclosing init wrapper has already opened one. The flag
# set here is cleared in the matching `finally` after kwargs are applied.
we_set_init_scope = getattr(type(self), "_has_on_change_hooks", False) and not getattr(
self, "_on_change_init", False
)
if we_set_init_scope:
self._on_change_init = True
self._on_change_pending: list[tuple[str, str]] = []
try:
# Apply Property values passed as kwargs
props = self.get_properties()
views = self._constructor_views
for key, val in kwargs.items():
if key in props or key in views:
setattr(self, key, val)
else:
raise TypeError(_unknown_kwarg_message(type(self), key, [*props, *views]))
finally:
if we_set_init_scope:
self._on_change_init = False
self._flush_on_change_pending()
@property
def name(self) -> str:
"""This node's name, which is never empty.
The empty string is not a name a node can hold: it cannot be found by
:meth:`find`, it cannot appear in a node path, and a scene that saved
one would load back a node named after its type. Assigning it raises;
pass no ``name`` to the constructor to get the type name instead.
"""
return self._name
[docs]
@name.setter
def name(self, value: str):
if not isinstance(value, str):
raise TypeError(f"Node name must be a string, got {type(value).__name__}")
if not value:
raise ValueError(
f"{type(self).__name__}: a node name cannot be empty. "
f"Construct without a name to be named after the type."
)
old = self._name
self._name = value
parent = getattr(self, "parent", None)
if parent is not None:
parent.children._rename(self, old, value)
def _on_visible_changed(self) -> None:
"""Carry a visibility flip down the subtree. Fired by the ``visible`` Property."""
parent_effective = True if self.parent is None else self.parent._visible_in_hierarchy
self._propagate_visibility(parent_effective)
self._notification(Notification.VISIBILITY_CHANGED)
[docs]
@property
def visible_in_tree(self) -> bool:
"""Whether this node is effectively visible: its own flag AND every ancestor's.
``visible`` answers only for this node, so a visible node under a hidden
parent still reports ``True`` there. This is the answer the draw walk uses,
and it is the one to test before skipping work for a hidden subtree.
Maintained on every visibility change and every reparent, so reading it is
a cached lookup rather than a walk up the parents. A node with no parent
reports its own ``visible``.
"""
return self._visible_in_hierarchy
def _propagate_visibility(self, parent_effective: bool) -> None:
"""Update _visible_in_hierarchy for self and descendants.
Propagates the effective visibility down the subtree; prunes branches
whose effective state didn't change so toggles cost O(changed-subtree)
rather than O(whole-subtree).
"""
new_effective = parent_effective and self.__visible
if self._visible_in_hierarchy == new_effective:
return
self._visible_in_hierarchy = new_effective
# A flip in effective visibility changes what this node's ``on_draw``
# contributes to the retained 2D item set (it is drawn iff effectively
# visible). A descendant's flip comes from an ancestor, not from a write
# to its own Property, so mark render-dirty here and the
# ``RenderItemCache`` drops (or re-adds) this node's slice instead of
# leaving its last frame painted.
self.queue_redraw()
for child in self.children:
child._propagate_visibility(new_effective)
def _invalidate_update_mode_cache(self):
"""Clear cached process mode for this node and descendants that inherit."""
self._clear_update_mode_cache()
# The same switch decides whether this subtree hears input, so a
# context held anywhere in it may have just gone deaf or woken up.
self._refresh_input_contexts()
def _clear_update_mode_cache(self):
"""Drop the resolved mode here and on the descendants that inherit it."""
self._cached_update_mode = None
for child in self.children:
if child.update_mode == UpdateMode.INHERIT:
child._clear_update_mode_cache()
def _refresh_input_contexts(self) -> None:
"""Re-flatten if this node going deaf or waking changed the active contexts.
Free for a tree whose only context is the root's, which is every game
that declares its bindings once: the map answers from the stack length
without looking at a node.
"""
tree = self._tree
if tree is not None:
tree.input_map._refresh_active()
# ------------------------------------------------------------------ input player
@property
def input_player(self) -> InputPlayer:
"""Whose input this subtree answers to.
Reading it gives the player this node belongs to, inherited from the
nearest ancestor that declared one and player 0 where nobody did.
Assigning one declares it here and for everything below::
two = Input.add_player()
two.pair(Input.controllers[1])
player_two_ship.input_player = two
Inside that subtree ``Input.pressed("thrust")`` answers for player two
with no argument, and an ``@on_input("jump")`` handler on one of
its nodes hears only player two's edge. That is what keeps local co-op
from being a rewrite: a game gains the dimension by declaring it once
rather than threading an id through every call site, and a
single-player game never sees it because one implicit player owns
every device.
The scoping covers this subtree's per-frame code (``on_update``,
``on_fixed_update``) and the input handlers on its nodes. Elsewhere,
and everywhere in a game that declared nothing, ``Input`` answers for
the default player.
Assign ``None`` to stop declaring one here and inherit again.
"""
return self.tree.input.players[self._effective_player_id()] if self._tree else _default_player()
# ------------------------------------------------------------------ input context
@property
def input_context(self) -> InputContext | None:
"""The bindings this node puts on top of the stack while it can hear input.
A context is how a game says "while this is on screen, these are the
controls": a vehicle whose ``move`` steers rather than walks, a pause
menu that takes every name below it, a minigame with its own verbs::
class Game(Node2D):
input_context = {"move": [Axis2D.WASD, Controller.LEFT_STICK], "jump": [Key.SPACE]}
class PauseMenu(Control):
input_context = InputContext({"resume": [Key.ESCAPE]}, hides=InputContext.ALL)
A plain dict is a context of its own entries; ``InputContext`` is the
spelling with a name or ``hides``. Any node may declare one: the
root's covers the whole tree, a branch's is in force while the branch
is, a reparented node carries its own with it, and a scene that
replaces the root takes the old root's away.
Declared on the class, as above, or assigned at any point; either way
it is **active exactly while this node can receive input**, so no
exception, early return or scene swap can leave one behind, and the
switch that stops the node processing takes its context down too: a
paused vehicle stops steering and gives ``move`` back to whatever
declared it below. Contexts stack in activation order, later on top,
and shadow by action name: the vehicle's ``move`` steers while the
walking controller's ``jump`` still jumps. ``hides=InputContext.ALL``
is the stronger form and takes every name below, which is what a pause
menu wants. A menu whose ``update_mode`` is ``PAUSED_ONLY`` therefore
hides nothing until the game is actually paused.
Assigning ``None`` takes this node's context off the stack.
Changing the stack releases the actions whose bindings changed, with a
real release edge, and presses nothing: a key held as the context
changes reads unpressed until the player presses it again.
"""
return self._input_context
[docs]
@input_context.setter
def input_context(self, context: InputContext | Mapping[Any, Any] | None) -> None:
if isinstance(context, Mapping):
context = InputContext(context, name=type(self).__name__)
previous = self._input_context
if context is previous:
return
tree = self._tree
if previous is not None and tree is not None:
tree.input_map._deactivate(previous, self)
self._input_context = context
if context is None:
self._input_context_index = -1
elif tree is not None:
tree.input_map._activate(context, self)
def _input_context_rank(self) -> int:
"""Which context this node's handlers sit under, as a stack index.
The nearest ancestor-or-self that holds one, and 0 (the root context,
which is the floor) for a node with none above it. What ``hides`` is
measured against: a handler is hidden when the context it belongs to
sits below the hiding one.
"""
node: Node | None = self
while node is not None:
index = node._input_context_index
if index >= 0:
return index
node = node.parent
return 0
def _invalidate_player_cache(self) -> None:
"""Clear the resolved player here and on descendants that inherit it."""
self._cached_player = None
for child in self.children:
if child._declared_player is None:
child._invalidate_player_cache()
def _effective_player_id(self) -> int:
"""Resolve the inherited player by walking up, cached per node."""
cached = self._cached_player
if cached is not None:
return cached
declared = self._declared_player
if declared is not None:
resolved = declared.id
elif self.parent is not None:
resolved = self.parent._effective_player_id()
else:
resolved = 0
self._cached_player = resolved
return resolved
def _enter_player(self) -> tuple[Any, int] | None:
"""Make ``Input`` answer for this node's player, until :meth:`_leave_player`.
Returns what restores it, or None when there is nothing to restore:
a node with no tree, or one whose player is already the one being
answered for, which is every node of a game that declared none.
"""
tree = self._tree
if tree is None:
return None
state = tree.input
resolved = self._effective_player_id()
previous = state._current_player
if resolved == previous:
return None
state._current_player = resolved
return state, previous
@staticmethod
def _leave_player(scoped: tuple[Any, int]) -> None:
"""Put back whichever player was being answered for before."""
state, previous = scoped
state._current_player = previous
def _hears_player(self, player_id: int) -> bool:
"""Whether an action belonging to *player_id* is this node's to hear.
True for everything while no node in the process declares a player,
which is every single-player game: there is one player, every action
belongs to it, and the question never has to be asked.
"""
if not _PLAYER_SCOPES:
return True
return self._effective_player_id() == player_id
def _notification(self, what: Notification) -> None:
"""Called when a notification is dispatched. Override to handle."""
[docs]
def reset_error(self) -> None:
"""Clear script error flag to re-enable processing."""
self._script_error = False
def _flush_on_change_pending(self) -> None:
"""Dispatch on_change hooks queued during ``__init__``.
Hooks are deduplicated by ``(property_attr, method_name)`` so multiple
sets of the same property during construction fire the hook once. Order
of first occurrence is preserved.
"""
pending = getattr(self, "_on_change_pending", None)
if not pending:
return
self._on_change_pending = []
seen: set[tuple[str, str]] = set()
for prop_attr, method_name in pending:
key = (prop_attr, method_name)
if key in seen:
continue
seen.add(key)
method = getattr(self, method_name, None)
if method is not None:
method()
def _handle_script_error(self, method_name: str) -> None:
"""Disable this node after a script exception and surface the traceback.
Only reached in non-strict mode: callers re-raise first when
``Node.strict_errors`` is set. Must be called from inside the active
``except`` block so ``traceback.format_exc()`` sees the exception.
"""
self._script_error = True
import sys
import traceback
tb = traceback.format_exc()
# The node's PATH, not its name: "which node?" is the whole diagnostic value of
# a contained error, and a scene routinely holds several nodes of one name.
path = self.path
# Always print to stderr so errors are never invisible
print(f"Script error in {path}.{method_name}: node disabled:\n{tb}", file=sys.stderr)
log.error("Script error in %s.%s: node disabled", path, method_name)
tree = self._tree
if tree is not None:
try:
tree.script_error.emit(self, method_name, tb)
except Exception:
# justified: signal-handler errors must not derail error recovery itself
pass
# Last, because the flatten it may trigger runs its own try blocks and
# ``format_exc`` above reads the exception still being handled. A
# disabled node hears nothing further, its context included: it must
# not go on hiding names from the rest of the tree.
self._refresh_input_contexts()
def _safe_call(self, method, *args: Any) -> None:
"""Call a lifecycle method with error recovery."""
if self._script_error:
return
try:
method(*args)
except Exception:
if Node.strict_errors:
raise
self._handle_script_error(method.__name__)
def _safe_decide(self, method_name: str, event: Any) -> bool | None:
"""Run an input handler and return what it decided about the claim.
``True`` claims the event, ``False`` declines it for this call, and
``None`` leaves the claim to the handler's own default: a decorated
handler's ``consume=``, and not claiming for ``on_gui_input``.
Anything else is a mistake in the handler rather than in the input, so
it raises naming the handler; the raise happens inside the containment
below, which is the repo's rule for an error in a lifecycle hook (it
propagates under ``Node.strict_errors``, and in editor and release mode
it is logged and the node disabled).
**A handler that raises claims nothing.** A claim is the handler saying
it dealt with the event, and one that blew up did not. Letting the
claim stand would make a crashed pause menu swallow every key, which is
invisible; declining lets the rest of the tree go on responding, which
is a visible, ordinary degradation. It is also the only answer that
agrees with the node then being disabled: a node that will never run
again must not go on owning input.
Every route a handler can be reached by shares this, so the decorated
broadcast handlers and the UI's ``on_gui_input`` answer to one rule.
"""
if self._script_error:
return False
try:
decided = getattr(self, method_name)(event)
if decided is None or decided is True or decided is False:
return decided
raise TypeError(
f"{type(self).__name__}.{method_name}() returned {decided!r}: an input handler "
f"returns True to claim the event, False to decline it, or None to leave the "
f"claim to its default"
)
except Exception:
if Node.strict_errors:
raise
self._handle_script_error(method_name)
return False
def _rebind_class(self, new_class: type) -> None:
"""Change this node's class in place, keeping its registrations true.
Swapping ``__class__`` on a live node is how hot reload runs edited
code without re-parenting anything, and how the editor retypes a node
it has just written a class for. Both do it *because* the behaviour
should change, and input handlers are behaviour, so the registration
follows the class. The node is retired from the
buckets it registered in, takes the new class, and registers again
under whatever handlers that class declares. Nothing else about the
node moves: it keeps its parent, its children, its groups, its unique
name and its place in the tree, which is the point of swapping the
class rather than rebuilding the node.
Assigning ``__class__`` directly is safe -- the dispatch tables
remember what each node registered with -- but the node goes on
delivering the old class's handlers until it next enters a tree. Use
this instead.
"""
tree = self._tree
if tree is not None and "_input_plan" in self.__dict__:
tree._unregister_input_node(self)
self.__class__ = new_class
if tree is not None:
if type(self)._simvx_input_handlers:
tree._register_input_node(self)
if type(self)._simvx_picks_release:
tree._release_pickers.add(self)
else:
tree._release_pickers.discard(self)
[docs]
def add_child(self, node: T) -> T:
"""Add a node as a child, reparenting it if already in a tree.
Returns ``node`` itself, at the type it was passed, so the
``self.hero = self.add_child(Sprite2D(...))`` the tutorials teach
keeps the child's own type rather than widening it to ``Node``.
Args:
node: The node to add. Removed from its current parent first.
Raises:
ValueError: ``node`` is ``self`` or one of ``self``'s ancestors --
either would create a cycle in the scene tree -- or
:meth:`destroy` has been called on ``node``, which is final.
"""
self._attach_child(node)
if self._tree:
tree = self._tree
# ``_attach_child`` has already bumped ``_structure_version`` through
# ``Children``, which is the one place the shape is allowed to move.
if tree._input_span_depth:
# Already inside the tree's input span: the common case, a node spawned
# from on_update, on_ready or an input handler. Entering another span
# would only bump a depth counter, so skip it entirely and pay nothing.
node._enter_tree(tree)
node._ready_recursive()
else:
# Spawned from outside the tree's own work, so open a span: this node's
# on_enter_tree and on_ready then see the tree they are joining rather
# than whichever Input and InputMap happen to be active for the caller.
with tree.activate_input():
node._enter_tree(tree)
node._ready_recursive()
return node
def _attach_child(self, node: Node) -> None:
"""Link ``node`` into this node's children WITHOUT running the entry path.
The half of :meth:`add_child` that is pure structure, and the exact
counterpart of :meth:`_detach_child`. Split out for
``SceneTree.root``'s setter, whose published contract is to re-link the
current scene and run no lifecycle, so the two cannot drift apart.
"""
if node is self:
raise ValueError(f"Cannot add node {node.name!r} as child of itself")
if node._destroying or node._destroyed:
raise ValueError(
f"Cannot add destroyed node {node.name!r} as a child of {self.name!r}: "
"destroy() is final, so build a fresh node instead of reviving this one"
)
ancestor = self.parent
while ancestor is not None:
if ancestor is node:
raise ValueError(
f"Cannot reparent {node.name!r} under its descendant {self.name!r}: would create a cycle"
)
ancestor = ancestor.parent
if node.parent:
node.parent.remove_child(node)
node.parent = self
self.children._add(node)
if node._is_canvas_layer:
self._canvas_layer_child_count += 1
node._notification(Notification.PARENTED)
node._invalidate_update_mode_cache()
node._invalidate_player_cache()
node._propagate_visibility(self._visible_in_hierarchy)
if hasattr(node, "_invalidate_transform"):
node._invalidate_transform()
[docs]
def remove_child(self, node: Node) -> None:
"""Remove a child node from this node's children.
Immediate: ``node`` has left the tree by the time this returns, so its
``on_exit_tree`` has already run. Prefer :meth:`destroy` from inside a
signal handler or a lifecycle hook, which runs the same teardown at the
end of the frame instead of in the middle of a dispatch.
"""
if node not in self.children:
return
tree = self._tree
if tree is None:
node._exit_tree()
self._detach_child(node)
return
if tree._input_span_depth:
# Already inside the tree's input span: the common case, a removal from
# a signal handler or a lifecycle hook. Entering another span would only
# bump a depth counter, so skip it entirely and pay nothing.
node._exit_tree()
else:
# Removed from outside the tree's own work, so open a span: this node's
# on_exit_tree then sees the tree it is leaving rather than whichever
# Input and InputMap happen to be active for the caller.
with tree.activate_input():
node._exit_tree()
self._detach_child(node)
def _detach_child(self, node: Node) -> None:
"""Unlink ``node`` from this node's children WITHOUT running the exit path.
The half of :meth:`remove_child` that is pure bookkeeping. Split out for
the delete-queue drain, which must unlink a node that a destroyed
ancestor already carried out of the tree: re-running ``_exit_tree`` there
would fire ``on_exit_tree`` a second time on the same node.
"""
if node.parent is not self:
# Already unlinked, which is what an ``on_exit_tree`` that removes
# its own node leaves behind for the removal that is unwinding.
return
self.children._remove(node)
if node._is_canvas_layer and self._canvas_layer_child_count > 0:
self._canvas_layer_child_count -= 1
node._notification(Notification.UNPARENTED)
node.parent = None
node._invalidate_update_mode_cache()
node._invalidate_player_cache()
node._propagate_visibility(True)
[docs]
def reparent(self, new_parent: Node):
"""Remove from current parent and add to new_parent."""
if self.parent:
self.parent.remove_child(self)
new_parent.add_child(self)
@overload
def node_at(self, path: str) -> Node: ...
@overload
def node_at(self, path: str, default: D) -> Node | D: ...
[docs]
def node_at(self, path, default=_NO_DEFAULT):
"""Navigate the tree by path: ``'Child/GrandChild'`` or ``'../Sibling'``.
A leading ``/`` starts from the tree root, which the path may name as its
first segment (``'/Root/Player'``).
``default`` works as it does on :func:`getattr`: without one, a path that
names no such child raises :class:`NodeNotFound`; with one, that miss
returns the default instead, so ``node_at("HUD/Minimap", None)`` is how
an optional node is read. A malformed path -- ``'..'`` past the root --
is a bug in the caller rather than an absent node, and raises either way.
"""
current = self
parts = [p for p in path.split("/") if p]
if path.startswith("/"):
# Stop below the tree's own top node: "the root" is the scene root,
# never the internal parent that also owns the singletons.
while current.parent is not None and not current.parent._is_tree_root:
current = current.parent
# An absolute path may optionally name the root as its first segment
# (e.g. '/Root/Player'). Consume it once so it resolves to root.
if parts and parts[0] == current.name:
parts.pop(0)
for part in parts:
if part == "..":
parent = current.parent
if parent is None or parent._is_tree_root:
raise ValueError("Already at root")
current = parent
else:
try:
current = current.children[part]
except NodeNotFound:
if default is _NO_DEFAULT:
raise
return default
return current
@overload
def find(self, target: type[T], *, direct: bool = False) -> T | None: ...
@overload
def find(self, target: str, *, direct: bool = False) -> Node | None: ...
@overload
def find(self, target: Callable[[Node], bool], *, direct: bool = False) -> Node | None: ...
[docs]
def find(self, target, *, direct: bool = False):
"""First descendant matching ``target``, or ``None``.
``target`` may be:
- a :class:`Node` subclass: matches the first ``isinstance`` descendant.
The result is typed as that subclass (``find(Player) -> Player | None``),
so no cast is needed.
- a ``str``: matches the first descendant whose ``name`` equals it.
- a predicate ``(Node) -> bool``: matches the first descendant it accepts.
Search is depth-first, pre-order, and recursive by default. Pass
``direct=True`` to consider only this node's direct children.
"""
return self._find_first(_node_matcher(target), direct)
def _find_first(self, match: Callable[[Node], bool], direct: bool) -> Node | None:
for child in self.children:
if match(child):
return child
if not direct:
found = child._find_first(match, False)
if found is not None:
return found
return None
@overload
def find_all(self, target: type[T], *, direct: bool = False) -> list[T]: ...
@overload
def find_all(self, target: str, *, direct: bool = False) -> list[Node]: ...
@overload
def find_all(self, target: Callable[[Node], bool], *, direct: bool = False) -> list[Node]: ...
[docs]
def find_all(self, target, *, direct: bool = False):
"""All descendants matching ``target`` (same matcher rules as :meth:`find`),
in depth-first pre-order. Recursive by default; ``direct=True`` limits the
search to direct children. Returns ``[]`` when nothing matches."""
out: list[Node] = []
self._find_all_into(_node_matcher(target), direct, out)
return out
def _find_all_into(self, match: Callable[[Node], bool], direct: bool, out: list[Node]) -> None:
for child in self.children:
if match(child):
out.append(child)
if not direct:
child._find_all_into(match, False, out)
@overload
def expect(self, target: type[T], *, direct: bool = False) -> T: ...
@overload
def expect(self, target: str, *, direct: bool = False) -> Node: ...
@overload
def expect(self, target: Callable[[Node], bool], *, direct: bool = False) -> Node: ...
[docs]
def expect(self, target, *, direct: bool = False):
""":meth:`find`, but a miss is an error rather than a ``None``.
Same matcher rules and the same typed result. Use it wherever the scene
is expected to contain the node and a missing one is a bug in the scene:
it fails at the lookup, naming what was asked for, instead of handing
back a ``None`` that raises somewhere later with no clue why.
Raises:
NodeNotFound: nothing under this node matches ``target``.
"""
found = self._find_first(_node_matcher(target), direct)
if found is None:
what = target.__name__ if isinstance(target, type) else target
raise NodeNotFound(f"no descendant of {self.name!r} matches {what!r}")
return found
@overload
def ancestor(self, target: type[T]) -> T | None: ...
@overload
def ancestor(self, target: str) -> Node | None: ...
@overload
def ancestor(self, target: Callable[[Node], bool]) -> Node | None: ...
[docs]
def ancestor(self, target):
"""Nearest ANCESTOR matching ``target``, or ``None``.
The upward counterpart of :meth:`find`, with the same matcher rules and
the same typed result, walking parents from this node outwards. ``self``
is never a candidate. This is how a node reaches the container it lives
under (``self.ancestor(Inventory)``) without hard-coding how deep it sits.
"""
match = _node_matcher(target)
node = self.parent
# Stops below the tree's own top node for the same reason ``path`` and
# ``node_at`` do: the scene root's ancestry ends at the scene root.
while node is not None and not node._is_tree_root:
if match(node):
return node
node = node.parent
return None
[docs]
def walk(self, *, include_self: bool = True) -> Iterator[Node]:
"""Iterate this node and all descendants in DFS pre-order."""
if include_self:
yield self
for child in self.children:
yield from child.walk(include_self=True)
[docs]
@property
def path(self) -> str:
parent = self.parent
if parent is None or parent._is_tree_root:
return f"/{self.name}"
return f"{parent.path}/{self.name}"
[docs]
@property
def is_scene_root(self) -> bool:
"""Whether this node is the top of its scene.
True for a node with no parent, and for the current scene root of a
:class:`~simvx.core.scene_tree.SceneTree`, whose parent is the tree's
own top node rather than another scene node. This is the test to make
before offering to delete, duplicate or reparent a node: a scene root
can do none of the three.
"""
parent = self.parent
return parent is None or parent._is_tree_root
# --- Groups ---
[docs]
def add_to_group(self, group: str):
"""Add this node to a named group.
The tree lists a group's members in join order: a node in the tree joins
last now, and one outside it joins when it enters. Adding a group the
node already has keeps its place. A node that has called :meth:`destroy`
records the group but joins no member list.
"""
self._groups.add(group)
if self._tree and not self._destroying:
self._tree._group_add(group, self)
[docs]
def remove_from_group(self, group: str):
"""Remove this node from a named group."""
self._groups.discard(group)
if self._tree:
self._tree._group_remove(group, self)
[docs]
def is_in_group(self, group: str) -> bool:
"""Check if this node belongs to a named group."""
return group in self._groups
# --- Lifecycle (override in subclasses) ---
[docs]
def on_ready(self) -> None:
"""Called once, ever, after the node and all its children enter the tree.
Override to perform initialisation that requires the scene tree --
finding sibling nodes, connecting signals, spawning children. The
``tree`` property is available. Called after ``on_enter_tree()`` and
after all children's ``on_ready()``.
Decorate other methods with ``@on_ready`` to register additional
ready handlers; they fire after the override in declaration order.
Once per node instance: a node that leaves the tree and comes back
does **not** ready again, so the children spawned here are built once
rather than once per entry, and the signals connected here are
connected once rather than stacking up a duplicate per entry.
Which of the two entry hooks::
on_enter_tree() every entry, for as long as the node keeps
coming back -- per-stay setup
on_ready() once per node instance, ever -- one-time setup
The mirror rule follows from that: whatever ``on_exit_tree()`` undoes
has to be redone in ``on_enter_tree()``. Putting it here leaves the
node dead the second time it joins a tree.
Example::
def on_ready(self):
self.sprite = self.node_at("Sprite")
self.health_changed.connect(self._update_hud)
"""
[docs]
def on_enter_tree(self) -> None:
"""Called on every entry into the scene tree, before ``on_ready()``.
Override for the setup that belongs to each stay in the tree: claiming
a slot on the tree, subscribing to a tree signal, restarting whatever
``on_exit_tree()`` stopped. Unlike ``on_ready()``, which fires once per
node instance and never again, this fires as often as the node is
added.
The whole incoming subtree is bound to the tree before any of its entry
hooks run, so a hook may look up nodes the walk has not reached yet --
by group, by unique name or by path -- and every one of them already
answers ``node.tree``. Their own ``on_enter_tree`` may not have run yet,
though, so read their state rather than depending on their setup;
``on_ready()`` is where the subtree is fully built.
Example::
def on_enter_tree(self):
self.add_to_group("enemies")
self.tree.screen_resized.connect(self._relayout)
"""
[docs]
def on_exit_tree(self) -> None:
"""Called when the node is about to leave the scene tree.
Override to clean up resources, disconnect external signals, or
persist state. Children have already exited by the time this fires
on the parent.
Example::
def on_exit_tree(self):
self.save_progress()
self.remove_from_group("enemies")
"""
[docs]
def on_update(self, dt: float) -> None:
"""Called every frame for game logic.
Args:
dt: Seconds elapsed since the previous frame (variable timestep).
Override for movement, AI, animation triggers, or any per-frame
update. Obeys ``update_mode`` -- disabled or paused nodes are
skipped automatically.
Decorate other methods with ``@on_update`` to register additional
per-frame handlers; they fire after the override in declaration
order. For state held while a button is pressed, poll
``Input.pressed("name")`` from inside ``on_update``.
Example::
def on_update(self, dt):
self.position += self.velocity * dt
"""
[docs]
def on_fixed_update(self, dt: float) -> None:
"""Called at a fixed timestep (default 60 Hz) for physics logic.
Args:
dt: Fixed time step in seconds (e.g. 1/60).
Override for deterministic physics updates -- forces, collision
responses, rigid-body integration. Runs independently of the
render frame rate.
Example::
def on_fixed_update(self, dt):
self.velocity += self.gravity * dt
self.move_and_slide(dt)
"""
[docs]
def on_draw(self, canvas) -> None:
"""Called each frame for custom 2D drawing.
Args:
canvas: The 2D drawing surface this frame is recorded into.
Override to issue immediate-mode draw calls: ``rect``, ``circle``,
``line``, ``polygon``, ``text``, ``texture`` and ``nine_patch``, each
painted with a ``fill`` colour, a ``stroke`` colour, or both. Called only
when ``visible`` is ``True``.
The 2D renderer is retained ("build once"): output is re-collected only
when a ``Property`` changes. If ``on_draw`` reads non-Property state (a
plain attribute updated by a signal or timer, ``tree.now`` animation),
call :meth:`queue_redraw` when that state changes so the new frame is
collected. This is identical on live and headless: a body that mutates
without ``queue_redraw`` freezes on both.
Drawing happens in the node's own space: ``(0, 0)`` is the node's origin,
and its position, rotation and scale, and every ancestor's, apply to what
it draws.
Example::
def on_draw(self, canvas):
canvas.circle((0, 0), 10, fill=(1, 0, 0, 1))
"""
[docs]
def on_draw_3d(self, gizmos) -> None:
"""Called each frame for custom immediate-mode 3D drawing.
Args:
gizmos: The active 3D draw surface, which issues primitives that
render inside the 3D scene pass.
Override to issue immediate 3D primitives positioned in world space, as
distinct from the retained mesh pipeline that a ``MeshInstance3D`` feeds.
Called once per frame while the node is visible, and never for a scene
with no 3D camera.
This is not :meth:`on_draw`, which draws in 2D and receives an entirely
different surface. A node may implement both.
Example::
def on_draw_3d(self, gizmos):
gizmos.draw_text_3d(text=self.label, position=self.world_position)
"""
[docs]
def on_picked(self, event: PickEvent) -> bool | None:
"""Called when a 3D mouse-pick event hits this node's collision shape.
Args:
event: The pick, carrying the camera ray, the button, and where
along the ray the hit landed.
Returns:
True to claim the press, which stops it there: no action state, no
edge and no broadcast behind it. False or None lets it through, so
clicking a barrel can mean both touching it and firing at it,
which is usually what a game wants. Anything else raises naming
the handler, the rule every input handler answers to.
Override to react to direct interaction with this 3D object --
selection, dragging, context menus. The override hears a press's pick
alone; a release's pick reaches only an ``@on_picked`` handler that
asks for it with ``edge="release"`` or ``"both"``.
Example::
def on_picked(self, event):
if event.button in Pointer.PRIMARY:
self.selected = True
The decorator form, :func:`~simvx.core.decorators.on_picked`, routes
by button, so one method per button replaces the test::
@on_picked(Pointer.PRIMARY)
def select(self, event):
self.selected = True
@on_picked(Mouse.RIGHT)
def open_menu(self, event):
...
Decorated handlers run first, in declaration order, and this override
runs after them only if none claimed.
"""
return None
def _internal_gui_input(self, event: GuiInputEvent) -> bool | None:
"""Run the ``@on_gui_input`` handlers *event* reaches, then :meth:`on_gui_input`, until one claims.
The targeted dispatch, which only the input router calls. A widget never
hands an event to a child through it: a child is a real child, so the
router delivers to it directly, with its disabled rule and signals.
"""
handlers = type(self)._simvx_gui_handlers
if handlers is not None and self._run_targeted(handlers, event):
return True
return self._safe_decide("on_gui_input", event)
def _internal_picked(self, event: PickEvent) -> bool | None:
"""Run the ``@on_picked`` handlers *event* reaches, then :meth:`on_picked`, until one claims.
A release pick reaches only the handlers whose edge asks for it, never the override.
"""
handlers = type(self)._simvx_pick_handlers
if handlers is not None and self._run_targeted(handlers, event):
return True
if not event.pressed:
return None
return self._safe_decide("on_picked", event)
def _run_targeted(self, handlers: TargetedHandlers, event: InputEvent) -> bool:
"""Offer *event* to the decorated targeted handlers it reaches, in declaration order; True on a claim.
Each answers by the rule :meth:`_safe_decide` holds every handler to,
and ``None`` defers to its ``consume=``. A target holding controls,
``Key.CTRL + Key.C``, hears the event only while they are down, read
from the modifiers the event itself reports. A handler reached under
several targets runs once, under the first whose chord is held.
"""
ran: set[str] = set()
for (_rank, method_name, chord, consume), delivered in handlers.matching(event, self._targeted_button_forms):
if method_name in ran or (chord is not None and not self._chord_held(chord, event)):
continue
ran.add(method_name)
decided = self._safe_decide(method_name, delivered)
if decided is True or (decided is None and consume):
return True
return False
def _targeted_button_forms(self, event: InputEvent) -> tuple[DeviceButton, ...]:
"""Every control a filter can name *event*'s button by: the control, its index form and its roles.
A pick is named by the pointer press it was cast for, where it carries one.
"""
if isinstance(event, PickEvent) and isinstance(event.source, PointerEvent):
event = event.source
button = getattr(event, "button", None)
if button is None:
return ()
tree = self._tree
if tree is not None and isinstance(event, PointerEvent):
forms: tuple[DeviceButton, ...] = tree.input._button_forms(event)
return forms
return (button, *roles_of(button))
def _chord_held(self, chord: InputBinding, event: InputEvent) -> bool:
"""Whether *chord*'s held controls are down, by the modifiers the event or its pick's press reports."""
tree = self._tree
if tree is None:
return False
source = event.source if isinstance(event, PickEvent) and event.source is not None else event
inp = tree.input
outer = inp._event_modifiers
inp._event_modifiers = frozenset(modifiers_held(source))
try:
held: bool = inp._binding_held(chord)
return held
finally:
inp._event_modifiers = outer
# --- Coroutine support ---
[docs]
def start_coroutine(self, gen: Coroutine) -> CoroutineHandle:
"""Register a generator coroutine to run each frame. Returns a cancellable handle."""
handle = CoroutineHandle(gen)
self._coroutines.append(handle)
return handle
[docs]
def stop_coroutine(self, gen_or_handle):
"""Stop and remove a running coroutine (accepts generator or CoroutineHandle)."""
if isinstance(gen_or_handle, CoroutineHandle):
gen_or_handle.cancel()
if gen_or_handle in self._coroutines:
self._coroutines.remove(gen_or_handle)
return
for h in self._coroutines:
if h._gen is gen_or_handle:
h.cancel()
self._coroutines.remove(h)
return
def _tick_coroutines(self, dt: float):
if not self._coroutines:
return
finished = []
# Snapshot: a coroutine that raises reports through the tree's script_error, and a
# listener is free to cancel coroutines on this node while that signal is dispatching.
for handle in list(self._coroutines):
if handle.is_cancelled:
finished.append(handle)
continue
gen = handle._gen
try:
if handle._primed:
gen.send(dt)
else:
next(gen)
handle._primed = True
except StopIteration:
finished.append(handle)
except Exception:
# A coroutine that raises is dropped on its own, so its siblings on
# this node still tick this frame. Beyond that it follows the same
# policy as a failed lifecycle hook: re-raise in strict mode, else
# disable the node and surface the traceback.
finished.append(handle)
if Node.strict_errors:
raise
self._handle_script_error(getattr(gen, "__qualname__", repr(gen)))
for handle in finished:
# stop_coroutine() invoked from inside the coroutine may have already removed it.
if handle in self._coroutines:
self._coroutines.remove(handle)
# -- Focus -------------------------------------------------------------
[docs]
@property
def focused(self) -> bool:
"""Whether this node holds keyboard focus. Read-only: :meth:`grab_focus` and :meth:`release_focus` move it."""
tree = self._tree
return tree is not None and tree._ui._focused_control is self
[docs]
@property
def focus_rect(self) -> tuple[float, float, float, float] | None:
"""Where this node is on screen, ``(x, y, width, height)``, as directional focus reads it.
None for a node with no extent on screen, which directional focus
passes over. A control answers its rect; a node that should be a
directional-focus destination and is not a control overrides this.
"""
return None
[docs]
def drag_threshold_for(self, pointer_type: PointerType | str) -> float:
"""The drag threshold in force here for *pointer_type*: this node's, an ancestor's, or the tree's."""
kind = PointerType(pointer_type)
node: Node | None = self
while node is not None:
own: dict[PointerType, float] | None = node.drag_threshold
if own is not None and kind in own:
return own[kind]
node = node.parent
tree = self._tree
if tree is not None:
return tree.drag_threshold.get(kind, DEFAULT_DRAG_THRESHOLD[kind])
return DEFAULT_DRAG_THRESHOLD[kind]
[docs]
def grab_focus(self) -> None:
"""Take keyboard focus, which ``tree.focus_owner`` then names. Outside a tree this does nothing.
Raises:
ValueError: ``focus_mode`` is ``NONE``. A widget that focuses itself
declares the mode it takes focus in, ``FocusMode.CLICK`` when
it is not a Tab stop.
"""
if not self.focus_mode.takes_focus:
raise ValueError(
f"{type(self).__name__} {self.name!r} cannot take focus: its focus_mode is NONE. "
"Set focus_mode to FocusMode.CLICK (focusable, not a Tab stop) or FocusMode.ALL first"
)
if self._tree is not None:
self._tree._set_focused_control(self, FocusReason.CODE)
[docs]
def release_focus(self) -> None:
"""Give keyboard focus up, if this node holds it."""
tree = self._tree
if tree is not None and tree._ui._focused_control is self:
tree._set_focused_control(None, FocusReason.CODE)
def _on_focus_mode_changed(self, _old: object = None, _new: object = None) -> None:
"""Keep the tree's index of focusable nodes in step, and give focus up when it can no longer be held."""
tree = self._tree
if tree is None:
return
if self.focus_mode.takes_focus:
tree._focusables.add(self)
else:
tree._focusables.discard(self)
self.release_focus()
if self.focus_mode.follows_hover:
tree._hover_focusables.add(self)
else:
tree._hover_focusables.discard(self)
[docs]
def capture_pointer(self, pointer_id: int | None = None) -> None:
"""Route every event from *pointer_id* to this node until it is released.
A drag needs the pointer past the edge of whatever started it: a slider
thumb the cursor has run off, a splitter, a canvas being panned. While
the capture is held the hit test is skipped for that pointer and the
events go straight here.
Args:
pointer_id: Which pointer. 0 is the mouse; every other id is
the opaque one its events carry. Left unset, the
pointer of the event being handled, so a slider dragged by a
finger holds that finger; outside a handler, the mouse.
Capture belongs to the node rather than to the control: a 3D gizmo and
a world-space handle drag exactly as a widget does. It lasts until
:meth:`release_pointer`, until the node leaves the tree, or until the
window loses focus, because a node the tree no longer holds receives
nothing and an interrupted press was never completed.
"""
if self._tree is not None:
self._tree._ui.capture(self, pointer_id)
[docs]
def release_pointer(self, pointer_id: int | None = None) -> None:
"""Give *pointer_id* back. A no-op when this node is not what holds it.
Left unset, the pointer of the event being handled, as
:meth:`capture_pointer`.
"""
if self._tree is not None:
self._tree._ui.release(self, pointer_id)
# --- Tree internals ---
def _enter_tree(self, tree: SceneTree):
"""Attach this subtree to ``tree``, in two passes over it.
The first pass binds every node's tree reference and registers its
groups, unique name and input handlers. Only once the whole subtree is
registered does the second pass run the entry hooks, walking the flat
list the first pass collected. An ``on_enter_tree`` therefore sees a
fully attached subtree: a parent can look its own descendants up by
group or by unique name from its entry hook, and every node it reaches
already answers ``node.tree``.
The recursive step goes through ``_enter_tree`` again, so a subclass
override fires once per node. The walk in progress is tracked
on the tree: the nested calls register only, and the outermost one owns
the second pass.
"""
attaching = tree._attaching
outermost = attaching is None
if attaching is None:
attaching = []
tree._attaching = attaching
try:
# First pass for this node: bind and register, running no user code.
self._tree = tree
if self.unique_name:
tree._unique_nodes[self.name] = self
if not self._destroying:
for group in self._groups:
tree._group_add(group, self)
# Register @on_input handlers with the tree's dispatch tables.
if type(self)._simvx_input_handlers:
tree._register_input_node(self)
if type(self)._simvx_picks_release:
tree._release_pickers.add(self)
# A context is on the stack exactly while its node is in the tree,
# which is what makes it impossible to leak one. Whether it counts
# while it is there is ``_can_receive_input``, read by the flatten.
if self._input_context is not None:
tree.input_map._activate(self._input_context, self)
if self.focus_mode.takes_focus:
tree._focusables.add(self)
if self.focus_mode.follows_hover:
tree._hover_focusables.add(self)
attaching.append(self)
for child in self.children:
child._enter_tree(tree)
finally:
if outermost:
tree._attaching = None
if not outermost:
return
for node in attaching:
# An earlier hook may have removed or destroyed part of the subtree;
# a node it carried out has no entry to announce.
if node._tree is not tree:
continue
node._notification(Notification.ENTER_TREE)
for method_name in type(node)._simvx_hooks.get("enter_tree", ()):
node._safe_call(getattr(node, method_name))
# The contexts this subtree brought with it are on the stack now, and
# whatever they took away has a release waiting.
tree._flush_action_edges()
def _exit_tree(self):
if self._exiting:
# Re-entered from inside this node's own exit: an on_exit_tree that
# removes or destroys the node it belongs to. The teardown already
# running finishes the job, so the nested call has nothing to do.
return
self._exiting = True
# A release edge a departing context produces waits for the teardown
# to finish: a handler must not run against a tree halfway through
# losing its nodes.
tree = self._tree
if tree is not None:
tree._input_edge_hold += 1
try:
self._exit_tree_inner()
finally:
self._exiting = False
if tree is not None:
tree._input_edge_hold -= 1
tree._flush_action_edges()
def _exit_tree_inner(self):
if self._tree is not None:
# Before the children detach, while the parent links this reads are
# all still intact. Focus and pointer capture are delivery paths of
# their own, held in tables rather than derived from the tree, so a
# subtree leaving has to take its entries with it: otherwise the
# drag goes on moving something that was deleted.
self._tree._notify_ui_subtree_removed(self)
for child in self.children:
child._exit_tree()
self._notification(Notification.EXIT_TREE)
for method_name in type(self)._simvx_hooks.get("exit_tree", ()):
self._safe_call(getattr(self, method_name))
# Close any in-flight coroutines so their ``finally:`` blocks run
# (releases signal-handler subscriptions, restores transforms, etc.).
# Without this, ``wait_signal`` lambdas stay attached to the signal
# for the lifetime of the emitter: a slow leak on scene churn.
# A ``finally:`` that raises is a script error: every coroutine is still
# closed, then under ``Node.strict_errors`` the first error propagates.
if self._coroutines:
close_error: Exception | None = None
for handle in self._coroutines:
try:
handle._gen.close()
except Exception as exc:
if Node.strict_errors:
close_error = close_error or exc
else:
log.exception("Coroutine close raised on node exit for %r", self)
self._coroutines.clear()
if close_error is not None:
raise close_error
if self._tree:
# Keyed on having registered rather than on the current class: a
# node that entered carrying handlers and had its class swapped for
# one carrying none still has entries to retire.
if "_input_plan" in self.__dict__:
self._tree._unregister_input_node(self)
self._tree._release_pickers.discard(self)
self._tree._focusables.discard(self)
self._tree._hover_focusables.discard(self)
if self._input_context is not None:
self._tree.input_map._deactivate(self._input_context, self)
self._input_context_index = -1
if self.unique_name:
self._tree._unique_nodes.pop(self.name, None)
for group in self._groups:
self._tree._group_remove(group, self)
self._tree = None
def _ready_recursive(self):
# Snapshot: a child's on_ready may add a sibling (via parent.add_child),
# which already readies it inline. Iterating the live list would re-visit
# that node and fire its on_ready twice. The copy makes ready exactly-once.
for child in list(self.children):
child._ready_recursive()
if self._ready_called:
# Ready is once per node instance, ever: a subtree that leaves the
# tree and comes back re-enters, but does not ready again.
return
self._ready_called = True
self._notification(Notification.READY)
for method_name in type(self)._simvx_hooks.get("ready", ()):
self._safe_call(getattr(self, method_name))
def _effective_update_mode(self) -> UpdateMode:
"""Resolve INHERIT by walking up the tree (cached)."""
cached = self._cached_update_mode
if cached is not None:
return cached
mode: UpdateMode = self.update_mode
if mode == UpdateMode.INHERIT:
mode = self.parent._effective_update_mode() if self.parent else UpdateMode.PAUSABLE
self._cached_update_mode = mode
return mode
def _can_update(self, paused: bool) -> bool:
"""Check if this node should process given the tree's pause state."""
mode = self._effective_update_mode()
if mode == UpdateMode.DISABLED:
return False
if mode == UpdateMode.ALWAYS:
return True
if mode == UpdateMode.PAUSED_ONLY:
return paused
# PAUSABLE (or resolved INHERIT → PAUSABLE)
return not paused
def _can_receive_input(self) -> bool:
"""Whether input reaches this node at this instant.
The switch that stops a node processing is the switch that deafens it,
so a paused player neither polls nor hears while a menu that runs
through the pause does. One predicate answers for all three delivery
routes and for the context stack: the broadcast skips a node that
fails it, the targeted phase does not offer it the pointer, the focus
owner or the pick, and its ``input_context`` is left out of the
flatten, so pausing a subtree stops it hiding names from everything
below it.
Three things say no: the node is in no tree, so nothing can reach it;
a script error has disabled it, and a node that will never run again
must not go on owning input; or its effective ``update_mode`` is off
against the tree's own ``paused``.
An inert overlay is deliberately not part of the pause this reads,
while it is part of the one processing reads. An inert overlay freezes
the world *behind* it and is itself an ordinary node in that world, so
folding it in here would deafen the dialog that opened it. What keeps
input off the world behind an overlay is the overlay registry's own
capture scope.
"""
tree = self._tree
if tree is None or self._script_error:
return False
return self._can_update(tree.paused)
def _process_recursive(self, dt: float, paused: bool = False):
if self._script_error:
return
# Inlined _can_update: resolve mode from cache and check pause state
mode = self._cached_update_mode
if mode is None:
mode = self._effective_update_mode()
if mode != UpdateMode.DISABLED and (
mode == UpdateMode.ALWAYS or (not paused if mode == UpdateMode.PAUSABLE else paused)
):
self._notification(Notification.PROCESS)
handlers = type(self)._simvx_hooks.get("update", ())
# A node inside a subtree that declares a player answers ``Input``
# for that player while its own code runs. Gated on any node in
# the process declaring one, so a single-player game pays one
# global read per node per frame and nothing else.
scoped = self._enter_player() if _PLAYER_SCOPES and handlers else None
observer = _NODE_OBSERVER
started = perf_counter() if observer is not None else 0.0
try:
for method_name in handlers:
try:
getattr(self, method_name)(dt)
except Exception:
if Node.strict_errors:
raise
self._handle_script_error(method_name)
return
finally:
if scoped is not None:
self._leave_player(scoped)
if observer is not None:
observer(self, "process", (perf_counter() - started) * 1000.0)
if self._coroutines:
self._tick_coroutines(dt)
for child in self.children.safe_iter():
child._process_recursive(dt, paused)
def _physics_process_recursive(self, dt: float, paused: bool = False):
if self._script_error:
return
# Inlined _can_update: resolve mode from cache and check pause state
mode = self._cached_update_mode
if mode is None:
mode = self._effective_update_mode()
if mode != UpdateMode.DISABLED and (
mode == UpdateMode.ALWAYS or (not paused if mode == UpdateMode.PAUSABLE else paused)
):
self._notification(Notification.PHYSICS_PROCESS)
handlers = type(self)._simvx_hooks.get("fixed_update", ())
scoped = self._enter_player() if _PLAYER_SCOPES and handlers else None
observer = _NODE_OBSERVER
started = perf_counter() if observer is not None else 0.0
try:
for method_name in handlers:
try:
getattr(self, method_name)(dt)
except Exception:
if Node.strict_errors:
raise
self._handle_script_error(method_name)
return
finally:
if scoped is not None:
self._leave_player(scoped)
if observer is not None:
observer(self, "physics", (perf_counter() - started) * 1000.0)
for child in self.children.safe_iter():
child._physics_process_recursive(dt, paused)
# ------------------------------------------------------------------ draw walk
#
# ONE unified, layer-banded, pluggable-key walker.
#
# ``_draw_recursive`` is the single skeleton for every node type. It owns the
# ``visible`` guard, the ``_script_error`` short-circuit, the self-dispatch,
# and the below/self/above interleave. Per-type behaviour is supplied through
# three narrow hooks, NOT by re-implementing the walk:
#
# * ``_draw_self(renderer)`` -- how this node draws itself
# (default: ``_draw_dispatch``).
# * ``_ordered_children()`` -- the pluggable ORDERING KEY: returns
# ``(below, above)`` child bands (self
# draws between them). The CanvasLayer
# band is folded into ``above`` by the
# types that partition it. Returning
# ``(None, None)`` selects the FAST
# PATH (no sort, tree order).
# * ``_draw_children(renderer)`` -- full override of child traversal,
# for types that wrap each child
# (Control's per-child clip+offset).
# * ``_draw_script_error(renderer)`` -- error presentation (default: walk
# children only).
def _draw_recursive(self, renderer):
if not self.visible:
return
if self._script_error:
self._draw_script_error(renderer)
return
below, above = self._ordered_children()
if below is None:
# FAST PATH: self first, then children in their traversal policy.
self._draw_self(renderer)
self._draw_children(renderer)
return
for child in below:
child._draw_recursive(renderer)
self._draw_self(renderer)
for child in above:
child._draw_recursive(renderer)
@staticmethod
def _banded_children(children, canvas_count, z_key=None):
"""Partition ``children`` into ``(below, above)`` with CanvasLayers LAST.
Shared helper for the default layer-banded policy (plain ``Node`` and
``Node2D``): CanvasLayer children are sorted by ``layer`` and split into
a negative-layer prefix (folded into ``below``) and a zero/positive
suffix (folded at the end of ``above``); world children optionally sort
by ``z_key``. CanvasLayers always draw last in their band, after world
``above`` content.
"""
below = []
above = []
canvas_layers = []
for c in children:
if c._is_canvas_layer:
canvas_layers.append(c)
elif z_key is not None and z_key(c) < 0:
below.append(c)
else:
above.append(c)
if z_key is not None:
below.sort(key=z_key)
above.sort(key=z_key)
if canvas_count:
canvas_layers.sort(key=lambda c: c.layer)
below = [c for c in canvas_layers if c.layer < 0] + below
above = above + [c for c in canvas_layers if c.layer >= 0]
return below, above
def _ordered_children(self):
"""Return ``(below, above)`` child bands for the walk (ORDERING KEY hook).
Plain ``Node`` has no z/Y ordering of its own. Its only banding concern
is CanvasLayer children (HUD layers drawn last); with none present it
takes the FAST PATH (tree order). Negative-layer CanvasLayers draw before
non-layer siblings, zero/positive after.
"""
if self._canvas_layer_child_count == 0:
return None, None
return self._banded_children(self.children.safe_iter(), self._canvas_layer_child_count)
def _draw_self(self, renderer):
"""Draw this node's own content in its own space.
The node's world affine is pushed around its own draw, so what it emits
lands where the node is, turned and scaled with it; its children push
their own. A node with no transform of its own (a plain ``Node``) pushes
nothing and pays nothing: it draws in the canvas scope it is walked in.
CanvasLayer overrides this to open its layer scope for the whole subtree.
"""
if type(self)._draw_affine is not Node._draw_affine and hasattr(renderer, "push_transform"):
push = getattr(renderer, "push_node_transform", renderer.push_transform)
push(*self._draw_affine())
try:
self._draw_own(renderer)
finally:
renderer.pop_transform()
else:
self._draw_own(renderer)
def _draw_own(self, renderer):
"""Fire this node's ``on_draw`` handlers. Control overrides it with its draw cache."""
self._draw_dispatch(renderer)
def _draw_affine(self) -> tuple[float, float, float, float, float, float]:
"""The camera-free affine ``(a, b, c, d, tx, ty)`` its local drawing maps through.
A plain Node has no 2D transform, so it draws in its parent's space.
"""
return (1.0, 0.0, 0.0, 1.0, 0.0, 0.0)
def _draw_children(self, renderer):
"""Walk children in tree order (FAST-PATH traversal).
Control overrides this to synthesise a per-child clip + transform offset.
Only invoked on the fast path; the banded path walks ``below``/``above``
directly.
"""
for child in self.children.safe_iter():
child._draw_recursive(renderer)
def _draw_script_error(self, renderer):
"""Present a script-errored node. Default: skip own draw, walk children.
Control overrides to paint an error box over its rect.
"""
for child in self.children.safe_iter():
child._draw_recursive(renderer)
def _draw_dispatch(self, renderer):
"""Invoke all ``on_draw`` handlers (override + decorated) for this node.
Used by ``_draw_recursive`` and by Control/Node2D subclasses that
wrap drawing with caching, transforms, or clipping but still want
to fire the same ordered set of handlers. While they run this node is
the capturing node, so an ambient value they read (``viewport_size``
and the rest) is recorded against it and a change to that value
re-captures it.
"""
context = draw_context
previous = context.node
context.node = self
if self._ambient_reads:
self._ambient_reads = 0
try:
for method_name in type(self)._simvx_hooks.get("draw", ()):
self._safe_call(getattr(self, method_name), renderer)
finally:
context.node = previous
# -- 2D render-retention ---------------------------------------------------
[docs]
def queue_redraw(self) -> None:
"""Mark this node's ``on_draw`` output stale (re-capture it next frame).
The manual escape hatch for an ``on_draw`` body that reads non-Property
state and changes ONCE (a signal/timer poke). For per-frame animation set
:attr:`dynamic` instead. Idempotent and cheap (a no-op once already dirty).
``Drawable2D`` (every ``Node2D`` / ``Control`` / ``CanvasLayer``) also gets
this called automatically by the blanket ``Property.__set__`` hook on any
changed Property, so drawing from Property state never needs it. A plain
``Node`` HUD/menu (``_render_auto_dirty`` is ``False``) calls it by hand.
"""
self._render_dirty = True
def _clear_render_dirty(self) -> None:
"""Drain the render-dirty bit. Called ONLY by the cache's upload step."""
self._render_dirty = False
[docs]
@property
def render_dirty(self) -> bool:
"""Whether ``on_draw`` output changed since the last upload (introspection)."""
return self._render_dirty
[docs]
def clear_children(self):
"""Destroy all children of this node."""
for child in list(self.children):
child.destroy()
[docs]
def destroy(self):
"""Schedule this node for removal at the end of the current frame.
**This is the engine's deferred removal**, and the only spelling of it:
a node marked here stays alive, in the tree, for the rest of the frame,
and its subtree is carried out in one piece at the end-of-frame sync
point. That is what makes it safe to call from a signal handler, a
collision callback or a lifecycle hook, where tearing a node down on the
spot would mutate a structure the caller is still walking.
:meth:`remove_child` is the immediate counterpart, for the cases that
genuinely need the node gone before the call returns (reparenting).
Removal does not cost a node the events it was still owed, and that is
true of ``remove_child`` too: the physics seam reports the contacts a
destroyed body was in on the step after it goes, and ``separated`` /
``body_exited`` carry that node. The seam holds it for that step itself,
so the naming does not depend on which removal path was taken.
Always final. A node that is not in a tree has no frame to defer to, so
it is finished on the spot rather than ignoring the call; one that is in
a tree is queued for the end of the frame. Either way the node is spent
afterwards, and :meth:`add_child` refuses to put it back in a tree.
Idempotent: a second call does nothing, before or after the removal has
run, so the same enemy can be destroyed by two handlers in one frame.
:attr:`destroying` reports the pending state.
The whole subtree is finished with the node, as it is removed with it:
every descendant still attached when the removal runs is spent too. A
child meant to outlive its parent is moved out before then.
The node leaves its groups on the call, so ``tree.group`` stops listing
it at once; its descendants leave theirs when the removal runs.
Signal connections made through this node's bound methods, ``EventBus``
subscriptions among them, are disconnected on the call, so emitters stop
dispatching to it on the next emit; its descendants' connections end
when the removal runs. Lazy weak-ref cleanup in ``Signal.emit``
covers nodes that are collected without ``destroy()``.
"""
if self._destroying or self._destroyed:
return
self._disconnect_outgoing()
tree = self._tree
if tree is not None:
self._destroying = True
for group in self._groups:
tree._group_remove(group, self)
tree._queue_delete(self)
else:
self._spend_subtree()
def _disconnect_outgoing(self) -> None:
"""End every signal connection made through this node's bound methods."""
for conn in list(self._outgoing_connections):
conn.disconnect()
self._outgoing_connections.clear()
def _spend_subtree(self) -> None:
"""Spend this node and every descendant: connections end, none can enter a tree again.
The last step of :meth:`destroy`, run once the subtree is out of the
tree, so every ``on_exit_tree`` in it has already run.
"""
stack: list[Node] = [self]
while stack:
node = stack.pop()
node._disconnect_outgoing()
node._destroying = False
node._destroyed = True
stack.extend(node.children)
[docs]
@property
def destroying(self) -> bool:
"""Whether :meth:`destroy` has been called and the removal has not run yet.
True from the ``destroy()`` call until the end-of-frame drain carries the
node out of the tree, and False again afterwards (the node object itself
stays valid; Python frees it when the last reference goes). The call
drops the node from its groups, so group queries already leave it out;
use this for a reference held some other way::
if self.target is not None and self.target.destroying:
self.target = None
Per node, not per subtree: a child of a destroying node reports False
until it is itself queued. Only nodes queued while in a tree are ever
marked: a detached node has no frame to defer to, so ``destroy()`` on
one finishes it immediately and this never turns True for it.
"""
return self._destroying
[docs]
@property
def destroyed(self) -> bool:
"""Whether this node has been spent: :meth:`destroy` has run its course.
True once the removal has carried the node and its subtree out of the
tree (at once for a node destroyed while detached), and never False
again: a spent node cannot be added back to a tree. Something holding a
reference to the node, such as a signal observer, uses this to know the
node's lifetime is over.
"""
return self._destroyed
[docs]
def call_deferred(self, method: Callable[..., Any], *args: Any) -> None:
"""Escape hatch: run ``method(*args)`` at the end of this frame, outside
tree traversal, instead of now.
**Discouraged: prefer a safe-by-default path when one exists.** SimVX
already makes the common cases safe without deferring: the process loop
and signal dispatch iterate snapshots (so adding/removing nodes mid-loop
does not corrupt iteration), :meth:`destroy` is already a deferred
delete, ``Property(coalesce=True)`` collapses repeated writes, and
``tree.events.publish_deferred(...)`` decouples event delivery. Reach for
``call_deferred`` only when you must mutate from a context none of those
cover, and document why at the call site.
``method`` is a bound method or any callable (type-safe; never a string
method name). Calls run once, in queue order, at the end-of-frame sync
point; anything queued *during* that drain runs on the next frame. A
call bound to this node is dropped if the node has left the tree by the
time the queue drains, and runs through :meth:`_safe_call` so a failure
obeys the same strict/release policy as any other lifecycle hook.
"""
if self._tree is None:
raise RuntimeError(
f"{self.name!r}.call_deferred() needs the node to be in a SceneTree "
"(there is no frame to defer to otherwise)."
)
self._tree.call_deferred(method, *args)
[docs]
@property
def app(self):
"""The App running this node's scene tree. Available after enter_tree()."""
return self._tree.app if self._tree else None
[docs]
@property
def tree(self) -> SceneTree:
"""The SceneTree this node belongs to."""
return self._tree
[docs]
@property
def viewport_size(self) -> Vec2:
"""The size of the render target this node draws into, right now.
What a node needs to lay anything out. Read it where it is used and
compute from it; do not copy it into a field, because a window resize,
a fullscreen toggle and a phone rotating all move it. An ``on_draw``
that read it is re-captured when it changes, with no ``queue_redraw``
and no ``dynamic``; code outside ``on_draw`` that lays out from it
listens to ``tree.screen_resized``::
def on_draw(self, renderer):
width, height = self.viewport_size
renderer.rect((0, height * 0.7), (width, height * 0.3), fill=UI_BG)
**In the pixels of the space you draw in.** Engine units are logical
pixels, and the answer is measured in the units this node draws in. A
world node gets screen pixels. A ``Control`` under the tree's
``ui_scale`` gets UI pixels: at 1.5 a 1200-pixel-wide window is 800 of
them, because a 200-pixel button is still 200 pixels to the control and
fewer fit, which is browser zoom's model. :attr:`viewport_insets`
follows the same rule.
**Which target.** A :class:`~simvx.core.viewport.SubViewport` renders
its children into its own offscreen texture, so a node under one gets
that texture's size rather than the window's. Everything else gets the
tree's ``screen_size``, a ``RenderView``'s children included: a
RenderView captures the main world from a second camera and its
children are ordinary main-scene content, so the target they draw into
is the screen. Asked of a SubViewport itself the answer is its own
target, the same number as its ``texture_size``, which is what lets a
SubViewport lay its own children out without naming its size twice.
**No default.** Off-tree and outside any SubViewport this raises rather
than answering with a window size nobody asked for: a fallback would
be a literal that is right on one machine, wrong on the next, and
invisible until someone runs the game on a phone. A node is in a tree from ``on_enter_tree`` onward, which is
before every hook that draws or lays out.
Raises:
RuntimeError: when the node is in no tree and under no SubViewport.
"""
reader = draw_context.node
if reader is not None:
reader._ambient_reads |= READS_VIEWPORT_SIZE
# ``Any`` because the walk is structural: it asks each ancestor whether
# it is a viewport by the same marker the graphics item pipeline prunes
# the main walk at, rather than importing the viewport module here to
# answer a question about sizes.
node: Any = self
while node is not None:
if getattr(node, "_is_subviewport", False):
width, height = node.size
return Vec2(float(width), float(height))
node = node.parent
if self._tree is None:
raise RuntimeError(
f"{self.name!r}.viewport_size needs the node to be in a SceneTree, or under a "
"SubViewport, to have a target to report. It is valid from on_enter_tree() onward; "
"in __init__ there is no viewport yet."
)
width, height = self._tree.screen_size
return Vec2(float(width), float(height))
[docs]
@property
def viewport_insets(self) -> tuple[float, float, float, float]:
"""How much of each edge of this node's render target is covered, as ``(left, top, right, bottom)``.
A soft keyboard sliding up is a bottom inset, a notch a top or side
one, a home indicator a small bottom one. Measured inward from the
edges of :attr:`viewport_size` and in the same units, the pixels of the
space this node draws in, so a control can use them as margins
directly and a world node can compare them with its own coordinates::
def on_update(self, dt):
left, top, right, bottom = self._panel.viewport_insets
self._panel.margin_left = left + PAD
self._panel.margin_bottom = -(bottom + PAD)
Read where it is used rather than remembered: a keyboard opens and a
phone rotates. An ``on_draw`` that read it is re-captured when it
changes; code outside ``on_draw`` listens to
``tree.screen_insets_changed``. The screen's own insets, in
screen pixels, are ``tree.screen_insets``, which is what a backend
writes. A node under a ``SubViewport`` draws into an offscreen texture
that nothing on the glass covers, so its insets are zero.
Raises:
RuntimeError: when the node is in no tree and under no SubViewport.
"""
reader = draw_context.node
if reader is not None:
reader._ambient_reads |= READS_VIEWPORT_INSETS
node: Any = self
while node is not None:
if getattr(node, "_is_subviewport", False):
return (0.0, 0.0, 0.0, 0.0)
node = node.parent
if self._tree is None:
raise RuntimeError(
f"{self.name!r}.viewport_insets needs the node to be in a SceneTree, or under a "
"SubViewport, to have a target to report. It is valid from on_enter_tree() onward."
)
return self._tree.screen_insets
[docs]
@property
def physics(self):
"""Spatial-query accessor bound to this node's physics world, or ``None``.
Mirrors :attr:`app` / :attr:`tree`: available once in-tree. Returns a
``PhysicsQuery`` scoped to the same world the node's body lives in
(resolved via the nearest ``PhysicsRoot`` ancestor, else the tree
default), exposing ``raycast`` / ``raycast_all`` / ``shapecast`` /
``overlap`` with typed results and ``mask`` / ``exclude`` filters. Built
fresh per access (not cached): the resolved world can change across
re-parent / change_scene, and the wrapper is a thin two-reference object
on the cold query path. Bind it locally if a hot loop wants to reuse it.
"""
if self._tree is None:
return None
from .physics.query import PhysicsQuery
from .physics.root import resolve_world
world = resolve_world(self)
node_map = self._tree._physics_nodes.get(world)
return PhysicsQuery(world, node_map)
[docs]
@property
def physics_2d(self):
"""2D spatial-query accessor bound to this node's 2D physics world, or ``None``.
The 2D sibling of :attr:`physics`: available once in-tree, returns a
``PhysicsQuery2D`` scoped to the same 2D world the node's body lives in
(resolved via the nearest ``PhysicsRoot2D`` ancestor, else the tree's 2D
default), exposing ``raycast`` / ``raycast_all`` / ``shapecast`` /
``overlap`` with typed 2D results and ``mask`` / ``exclude`` filters.
Built fresh per access (cold query path).
"""
if self._tree is None:
return None
from .physics.query2d import PhysicsQuery2D
from .physics.root import resolve_world_2d
world = resolve_world_2d(self)
node_map = self._tree._physics_nodes.get(world)
return PhysicsQuery2D(world, node_map)
[docs]
def __getitem__(self, key: str):
"""Shorthand for node_at: ``self["Child/Path"]``."""
return self.node_at(key)
[docs]
@classmethod
def get_properties(cls) -> dict[str, Property]:
"""Return all Property descriptors declared on this node class and its bases."""
return getattr(cls, "__properties__", {})
[docs]
def __repr__(self):
return f"<{type(self).__name__} '{self.name}'>"
# Register Node itself (not covered by __init_subclass__ which only fires for subclasses)
Node._registry["Node"] = Node
# Collect primary lifecycle hooks defined directly on Node (no decorators on the
# base class) so bare Node instances dispatch through the same code path as
# subclasses.
Node._simvx_hooks, Node._simvx_input_handlers = collect_hooks(Node, Node._PRIMARY_HOOK_METHODS)
# ============================================================================
# Timer
# ============================================================================
[docs]
class Timer(Node):
"""Node that emits :attr:`timeout` after ``duration`` seconds.
A Timer counts down in ``on_update``, so it runs on frame time, not on the
fixed-step physics clock, and it obeys ``update_mode``: a ``PAUSABLE``
Timer stops counting while the tree is paused. Nothing happens until
:meth:`start` is called or ``autostart`` is set at construction.
With ``one_shot`` (the default) the timer fires once and stops. With
``one_shot=False`` it repeats, and the leftover time from the frame that
crossed zero is carried into the next cycle, so a repeating timer does not
drift at low frame rates.
Connect to it like any other signal::
timer = Timer(duration=0.5, one_shot=False, autostart=True)
timer.timeout.connect(self.spawn_enemy)
self.add_child(timer)
:attr:`time_left` reports the seconds remaining and :attr:`stopped` whether
the countdown is idle; :meth:`stop` resets both without emitting.
"""
duration = Property(1.0, range=(0.001, 3600), hint="Countdown length in seconds")
one_shot = Property(True, hint="Fire once and stop; False repeats until stopped")
autostart = Property(False, hint="Begin counting down on construction, without waiting for start()")
timeout = Signal()
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._running = bool(self.autostart)
self._time_left = self.duration if self._running else 0.0
[docs]
def start(self, duration: float = 0):
"""Start or restart the timer, optionally overriding duration."""
if duration > 0:
self.duration = duration
self._time_left = self.duration
self._running = True
[docs]
def stop(self):
"""Stop the timer and reset time_left to zero."""
self._running = False
self._time_left = 0.0
[docs]
@property
def stopped(self) -> bool:
return not self._running
[docs]
@property
def time_left(self) -> float:
return self._time_left
[docs]
def on_update(self, dt: float):
if not self._running:
return
self._time_left -= dt
if self._time_left <= 0:
self.timeout.emit()
if self.one_shot:
self._running = False
else:
self._time_left += self.duration