"""SceneTree: Central manager for the node tree, groups, input routing, and UI focus."""
from __future__ import annotations
import logging
import math
import weakref
from collections import deque
from collections.abc import Callable, Iterable, Mapping
from contextvars import Token
from dataclasses import dataclass, field
from enum import StrEnum
from typing import TYPE_CHECKING, Any, overload
import numpy as np
from .event_bus import EventBus
if TYPE_CHECKING:
from .audio_listener import AudioListener2D, AudioListener3D
from .audio_protocol import (
AudioBackend,
AudioBusBackend,
AudioPlaybackBackend,
AudioStreamingBackend,
)
from .input.contexts import InputContext
from .input.map import _InputMap
from .input.state import _Input
from .physics.world import BodyHandle, PhysicsWorld
from .physics.world2d import Physics2DWorld
from .ui.scene_host import SceneHost
# Either dimensionality of world. The tree's per-world registries and the
# per-world step helpers are dimension-agnostic: they key on the world object
# and read widths off it, so both kinds travel the same path.
#
# A plain assignment rather than a PEP 695 ``type`` statement, which would
# make the alias lazy and force every reader of it through ``__value__``.
AnyPhysicsWorld = PhysicsWorld | Physics2DWorld
from . import node as _node_module
from ._draw_context import READS_UI_SCALE, READS_VIEWPORT_INSETS, READS_VIEWPORT_SIZE, draw_context, redraw_readers
from .decorators import _POINTER_TYPES, _naming
from .descriptors import UpdateMode
from .input.actions import Action, ActionKind
from .input.controls import (
MODIFIER_OF,
DeviceAxis,
DeviceButton,
DeviceControl,
DeviceStick,
DeviceType,
Key,
Pointer,
key_code,
)
from .input.drag import DEFAULT_DRAG_THRESHOLD, drag_threshold_map
from .input.enums import FocusReason, PointerPhase, PointerType
from .input.events import (
ActionEvent,
AxisEvent,
ButtonEvent,
InputBinding,
InputEvent,
KeyEvent,
PickEvent,
PointerEvent,
ScrollEvent,
TextEvent,
modifiers_held,
)
from .input.gestures import recognised_in
from .input.map import _active_input_map
from .input.state import _active_input
from .input.tick import TickInput
from .math.raycast import Ray, ray_intersect_sphere
from .math.types import Vec3
from .node import Node
from .nodes_2d.camera import Camera2D
from .nodes_3d.camera import Camera3D
from .physics.nodes import CollisionShape3D, PhysicsObject3D
from .physics.query import PhysicsQuery
from .signals import Signal
# The "events" singleton name is reserved for the engine-provided EventBus.
# Project TOMLs and runtime code may not declare or replace it.
RESERVED_SINGLETON_EVENTS = "events"
log = logging.getLogger(__name__)
[docs]
class ContentScaleMode(StrEnum):
"""How a scene's space relates to the render target it is presented in.
Set through :attr:`SceneTree.content_scale_mode`, which documents what each
one does to a layout and to the three sizes a game can read.
"""
#: An engine unit is a logical pixel of the render target. A wider window
#: is a wider scene and shows more of the world.
EXPAND = "expand"
#: The scene is always ``content_scale_size`` units across, scaled up
#: uniformly to the largest that fits the window and centred, with bars
#: over the remainder.
FIT = "fit"
class _InputSpan:
"""Makes one tree's ``Input`` and ``InputMap`` the active pair for a block.
Returned by :meth:`SceneTree.activate_input`, one fresh instance per ``with``
so that nested spans, and spans opened on the same tree from two threads,
each keep their own reset tokens.
Entering is on the hot path: it runs once per tick, once per physics step,
once per mount and once per dispatched input event. So the contextvars are
written only when they do not already hold the exact objects the span would
install, which is the whole cost for a tree that shares the process-wide
default pair and for any span nested inside another on the same tree. The
depth counter is bumped unconditionally either way: it is what tells node
code the engine is running it, and skipping it would silently disarm the
late-registration warning and the mount/unmount fast paths.
"""
__slots__ = ("_tree", "_input_token", "_map_token")
def __init__(self, tree: SceneTree):
self._tree = tree
self._input_token: Token[_Input | None] | None = None
self._map_token: Token[_InputMap | None] | None = None
def __enter__(self) -> _InputSpan:
tree = self._tree
own_input = tree._own_input
if _active_input.get() is not own_input:
self._input_token = _active_input.set(own_input)
own_map = tree._own_input_map
if _active_input_map.get() is not own_map:
self._map_token = _active_input_map.set(own_map)
tree._input_span_depth += 1
return self
def __exit__(self, exc_type, exc, tb) -> None:
self._tree._input_span_depth -= 1
if self._map_token is not None:
_active_input_map.reset(self._map_token)
self._map_token = None
if self._input_token is not None:
_active_input.reset(self._input_token)
self._input_token = None
class _SceneEnded(BaseException):
"""Unwinds the code a scene was running, because that scene has ended.
Raised by :meth:`SceneTree.change_scene` when the scene's own code asks for
the change, and caught where the tree began running that code. A
``BaseException`` so that neither a game's ``except Exception`` nor the
engine's script-error containment mistakes the end of a scene for an error
in it.
"""
class _SceneSpan:
"""The stretch of a frame in which the tree is running its scene's code.
Entered around the scene's process walk, its physics walk and step, and
each input event's delivery. Inside one, :meth:`SceneTree.change_scene`
ends the scene where it stands: it records the new root and raises
:class:`_SceneEnded`, and the outermost span of that tree swallows it, so
the walk stops once, wherever it had got to, rather than being asked at
every node whether it should go on. The phase that opened the span then
mounts the new root from its own stack.
One instance per tree, reused: a span is two integer writes, which is its
whole cost on a frame that changes no scene.
"""
__slots__ = ("_tree",)
def __init__(self, tree: SceneTree):
self._tree = tree
def __enter__(self) -> None:
self._tree._scene_depth += 1
def __exit__(self, exc_type, exc, tb) -> bool:
tree = self._tree
tree._scene_depth -= 1
# Only this tree's own end is swallowed, and only once the unwind is
# back where the tree started running the scene. A guest tree's span
# lets a host's scene change through, and an inner span of this tree
# lets it on to the outer one.
return exc_type is _SceneEnded and tree._scene_depth == 0 and tree._next_scene is not None
@dataclass(slots=True)
class _BodySync:
"""Per-world handle->node sync state connecting physics bodies to their nodes.
Drives the bulk transform read-back: ``nodes`` is a weak handle->body map
(a dead node auto-drops, never leaks or pins the tree); ``order`` + ``buf``
are rebuilt only when membership changes (``dirty``), so a world with stable
membership pays one bulk ``read_transforms`` per frame with zero allocation.
``prev_buf``/``cur_buf`` hold the last two STEP states (never render states);
render-time :meth:`SceneTree.interpolate_physics` blends between them by the
fixed-step accumulator fraction. ``seeded`` is reset on a membership rebuild
so a freshly added body re-seeds ``prev == cur`` and never lerps from origin.
``width``/``dims`` make the bulk plumbing dimension-aware: 3D worlds use a
``(N, 7)`` transform row (pos xyz + quat xyzw) with ``dims == 3``; 2D worlds
use ``(N, 4)`` (pos xy + cos/sin) with ``dims == 2``. Both are set from the
world's type at the first registration (see :meth:`SceneTree._world_dims`) so
the scatter / capture / interpolate leaves branch on ``dims`` only.
"""
nodes: weakref.WeakValueDictionary[BodyHandle, Node] = field(default_factory=weakref.WeakValueDictionary)
order: list[BodyHandle] = field(default_factory=list)
buf: np.ndarray | None = None # scratch target for read_transforms
prev_buf: np.ndarray | None = None # (N,width) f32: step state at t-1 (render interpolation)
cur_buf: np.ndarray | None = None # (N,width) f32: step state at t
dirty: bool = False
seeded: bool = False # False until prev/cur first populated (re-seed on membership change)
width: int = 7 # bulk transform row width: 7 for 3D (pos+quat), 4 for 2D (pos+cos/sin)
dims: int = 3 # spatial dimensions: 3 (3D) or 2 (2D); selects the scatter/interp branch
def _input_order_key(node: Node, top: Node) -> tuple[int, ...] | None:
"""A node's position in pre-order, as its sibling ranks from *top* down.
Lexicographic order over these tuples *is* pre-order: an ancestor's key is a
prefix of its descendants' and therefore sorts before them, and siblings
compare by position. Sorting a dispatch bucket by it in reverse therefore
gives the reverse pre-order a broadcast delivers in, a node before its
parent and later siblings before earlier ones, with *top* last.
``None`` when the walk does not reach *top*, which means the node is not in
this tree's shape however registered it still is. That is not a theoretical
case: assigning ``SceneTree.root`` re-links the scene and runs no
lifecycle, so the outgoing scene keeps its handlers while its parent link
is gone. Such a node would otherwise measure its position from its own
detached top and land on a key a live node already holds (a detached root
computes ``()`` and its first child ``(0,)``, which is also the first
singleton's), so an order that is supposed to be total would have
duplicates in it. Dropping it is also the honest answer to what the caller
asked for: a node the tree no longer holds does not receive its input.
The rank is a sibling's position in its parent's list, but assigned rather
than counted: ``Children`` hands out an increasing rank per child and
renumbers only on a reorder. A counted index would change for every
sibling after a departed one, which would quietly falsify every key already
cached in a bucket that nothing invalidated, and those keys are compared
across buckets when an event matches more than one. An assigned rank is
unmoved by an append or a removal, so a cached key stays true until
something actually reorders.
"""
path: list[int] = []
while node is not top:
parent = node.parent
if parent is None:
return None
path.append(node._sibling_rank)
node = parent
path.reverse()
return tuple(path)
#: What a pointer handover takes back: every device with a position.
_POINTER_DEVICES: frozenset[DeviceType] = frozenset(_POINTER_TYPES)
#: What a keys handover takes back: the keyboard, and the devices whose buttons have no position.
_KEY_DEVICES: frozenset[DeviceType] = frozenset({DeviceType.KEYBOARD, DeviceType.CONTROLLER, DeviceType.TABLET})
class _HandlerTables:
"""One phase's device handler tables, each keyed by the input it routes."""
__slots__ = ("key", "key_repeat", "pointer", "device_button", "device_axis", "event_class", "catch_all")
def __init__(self) -> None:
self.key: dict[tuple[DeviceButton, bool], _HandlerBucket] = {}
#: The key targets that asked for a held key's auto-repeats, ``repeat=True``.
self.key_repeat: dict[DeviceButton, _HandlerBucket] = {}
self.pointer: dict[tuple[DeviceButton | None, PointerPhase], _HandlerBucket] = {}
self.device_button: dict[tuple[DeviceButton, bool], _HandlerBucket] = {}
self.device_axis: dict[DeviceAxis, _HandlerBucket] = {}
self.event_class: dict[type, _HandlerBucket] = {}
self.catch_all = _HandlerBucket()
class _HandlerBucket(list):
"""The handlers registered for one routing key, and their delivery order.
Registration is append-only and therefore cheap, which is what a scene
change of a few thousand nodes needs. The order a broadcast wants is
reverse pre-order, and nothing about attach order expresses it, so the
bucket records which arrangement its order was computed for and re-sorts
itself on the first event that reaches it after the arrangement changed. A
scene change pays one sort per bucket an event actually touches, not one
walk per structural edit and not a walk per event.
The version it records is ``SceneTree._input_order_version``, which moves
only when the *relative* order of nodes changes, and not on every
structural edit. Delivery is by relative position, so appending a child or
removing one shifts indices and reorders nothing; keying off the general
structure counter made one spawned and freed bullet, binding no input at
all, re-sort every bucket in the tree on the next event.
``snapshot`` is the ordered tuple dispatch iterates. Holding it here rather
than copying the list per event is what keeps the loop allocation-free, and
it doubles as the safety copy a handler that adds or removes a node in the
middle of a dispatch needs.
It is dropped the moment a node leaves, not merely marked out of date. A
snapshot holds its nodes strongly, and the rebuild is lazy: it waits for an
event to arrive at this bucket, and ``_matching_buckets`` skips an empty
bucket without reaching ``_ordered`` at all, so an emptied bucket would
never rebuild and would hold the outgoing scene for the life of the tree.
Measured at five hundred nodes across a ``change_scene``, plus every plain
sibling they hold through ``parent``. Dropping it costs nothing, because
the next dispatch was going to re-sort anyway: the same ``drop`` has
already invalidated the order.
**Leaving is a tombstone, not a removal.** Rebuilding the list per departing
node makes a scene teardown quadratic: three thousand handler-carrying nodes
leaving one bucket is four and a half million comparisons, which measured at
seventy-eight milliseconds against one for the same nodes carrying no
handler. ``drop`` therefore records the node and sweeps only once the dead
entries outnumber the living, so a teardown pays a geometric series rather
than a square, and a bucket never holds more than twice the entries it
needs.
The count is of dead **entries**, not of dead nodes, and the difference is
not academic: a class and its subclass can both bind the same key, which is
ordinary UI code, and then every node puts two entries in this one bucket.
Comparing dead nodes against total entries makes the threshold
``D(2 - k) >= kL`` for k entries per node, which for k of two or more can
never be met while anything is live. A bucket like that swept only when an
event happened to reach it, so a routing key nothing ever pressed grew for
the life of the tree and held every departed node: 1008 entries where 8
were needed, measured. ``drop`` is called once per entry, so counting calls
counts entries.
"""
__slots__ = ("order_version", "snapshot", "pending", "dead")
def __init__(self) -> None:
super().__init__()
self.order_version: int = -1
self.snapshot: tuple[tuple, ...] = ()
#: Nodes that have left and whose entries are still in the list.
self.pending: set[Node] = set()
#: How many entries those nodes account for.
self.dead: int = 0
def add(self, entry: tuple) -> None:
if entry[0] in self.pending:
# Re-registered before its tombstone was swept: compact now, or the
# sweep would take the fresh entry along with the stale ones.
self.compact()
self.append(entry)
self.order_version = -1
def drop(self, node: Node) -> None:
"""Tombstone one of *node*'s entries; the caller knows it is in here."""
self.pending.add(node)
self.dead += 1
self.order_version = -1
self.snapshot = ()
if self.dead * 2 >= len(self):
self.compact()
def compact(self) -> None:
"""Drop every tombstoned entry. One pass, and it releases the nodes."""
if not self.pending:
return
dead = self.pending
self[:] = [entry for entry in self if entry[0] not in dead]
dead.clear()
self.dead = 0
self.order_version = -1
self.snapshot = ()
def _entry_order(entry: tuple) -> tuple[int, ...]:
"""The pre-order key of a snapshot entry, for merging two ordered buckets."""
return entry[0]
def _check_group_class(of: object) -> None:
"""Raise ``TypeError`` unless ``of`` is a class or a non-empty tuple of classes."""
if isinstance(of, type) or (isinstance(of, tuple) and of and all(isinstance(c, type) for c in of)):
return
raise TypeError(
f"a group query's class filter must be a class or a tuple of classes, not {of!r}: "
"tree.group(name, Enemy) or tree.group(name, (Enemy, Mine))"
)
[docs]
class SceneTreeTimer:
"""A transient one-shot timer owned by the :class:`SceneTree`.
Created via :meth:`SceneTree.create_timer`. Emits ``timeout`` once after its
delay elapses, then the tree drops it. Lighter than a :class:`~simvx.core.Timer`
node: it lives only on the tree's timer list, not in the scene graph.
"""
__slots__ = ("timeout", "_time_left")
def __init__(self, seconds: float):
self.timeout = Signal()
self._time_left = float(seconds)
[docs]
@property
def time_left(self) -> float:
return self._time_left
def _tick(self, dt: float) -> bool:
"""Advance by ``dt``; return ``True`` once the timer has fired."""
self._time_left -= dt
if self._time_left <= 0.0:
self.timeout.emit()
return True
return False
class _TreeRoot(Node):
"""A tree's own top node: parent of every singleton and of the current scene.
Not a scene node. It is created with the tree, never enters or leaves it, is
never drawn, never serialised and never reaches the editor. Its only job is
to give the tree exactly one walk source, so that a node registered as a
singleton and a node parented into the scene are the same kind of thing and
neither can be processed twice.
It must stay a plain :class:`Node`. A ``Node2D``, ``Node3D`` or ``Control``
here would enter every world transform and every global rect in the engine,
so subclassing it further, or giving it a ``position``, moves every pixel the
engine draws.
Its ``update_mode`` must stay ``INHERIT``. An unparented node resolves
``INHERIT`` to ``PAUSABLE``, which is what a singleton and a scene root
resolved to before this node existed; ``DISABLED`` would poison every
``INHERIT`` descendant and ``ALWAYS`` would break pausing.
It holds its tree strongly (``_tree``) and the tree holds it, so the pair is
a refcount cycle from construction. A teardown that promises refcount-only
reclamation owes the two lines that break it: clear this node's children and
set its ``_tree`` to ``None`` (see ``UITestHarness.teardown``).
"""
_is_tree_root = True
def _detach_child(self, node: Node) -> None:
# Unlinking the current scene is the only way ``tree.root`` can stop
# being a child of this node, so clearing the tree's reference here is
# what keeps "the scene is one of my children" true rather than hoped
# for. Every unlink funnels through here: ``remove_child`` calls it, the
# delete-queue drain calls it, and ``add_child`` reaches it through the
# reparent-away.
was_child = node.parent is self
super()._detach_child(node)
tree = self._tree
if was_child and tree is not None and tree._scene_root is node:
tree._scene_root = None
[docs]
class SceneTree:
"""Central manager for the node tree, groups, input routing, and UI focus.
Owns the root node and drives per-frame ``process`` / ``physics_process`` /
``draw`` traversals. Also manages pause state, scene changes, and the UI
input pipeline (mouse, keyboard, popups).
"""
# Most-recently-active tree, set on every ``set_root``/``change_scene``.
# InputMap reads it (best-effort) so ``add_action`` can warn when called
# after the first tick: a common mistake that silently drops bindings in
# the web exporter (which never invokes ``main()``). Held as a weakref so
# the last tree of a session does not pin its whole node graph in memory:
# the tree owns its nodes, nothing owns the tree, so a strong class-level
# ref is the one thing that survives ``gc.collect()`` once every external
# reference is dropped. ``current()`` then correctly reports "no active
# tree" once the tree is gone, which all consumers already handle.
_active_tree: weakref.ref[SceneTree] | None = None
#: Recheck every cached input dispatch order against the live tree, and
#: raise when they disagree. Off by default because it costs exactly what
#: the cache exists to save: turn it on for a run that is hunting a wrong
#: delivery order, which is the one input failure that produces no error
#: and no log of its own. Process-wide, like ``Node.strict_errors``.
verify_input_order: bool = False
#: Every input event, once delivery is over: ``(event, claimed_by)``, the
#: node that claimed it or None. The seam a recorder, a replay or a
#: debugger watches; an event does not carry who claimed it.
input_delivered = Signal()
#: A lifecycle hook, input handler or coroutine on a node in this tree
#: raised, and the node was disabled: ``(node, method_name, traceback)``.
#: Fires only outside ``Node.strict_errors``, where the error propagates
#: instead. Error panels and test harnesses connect to the tree they watch.
script_error = Signal(Node, str, str)
[docs]
@classmethod
def current(cls) -> SceneTree | None:
"""Return the most recently activated SceneTree, or ``None``.
Activation happens automatically on ``set_root`` / ``change_scene``.
Used by ``InputSimulator`` to deliver scene-tree + UI-tree events
without requiring callers to thread a tree reference through.
Held weakly: returns ``None`` once the tree has no remaining strong
references (e.g. a test dropped it), which is the correct "no active
tree" semantics.
"""
ref = cls._active_tree
return ref() if ref is not None else None
[docs]
@property
def app(self):
"""The App instance running this tree (set by graphics backend)."""
return getattr(self, "_app", None)
[docs]
@property
def events(self) -> EventBus:
"""Engine-provided typed event bus.
Use ``tree.events.subscribe(EventCls, handler)`` to register a
handler and ``tree.events.publish(event)`` (or
``publish_deferred``) to dispatch. The bus survives
``change_scene()``: subscriptions held by singletons or other
long-lived objects keep firing across scene swaps, and those the
replaced scene's nodes held end when it is destroyed. Deferred
events queued during a frame are dispatched at the start of the
next ``process()`` tick, before any node ``_process`` runs.
"""
return self._events
[docs]
@property
def audio_backend(self) -> AudioBackend | None:
"""The active audio backend, or ``None`` if none was initialised.
Returns the union :class:`AudioBackend` type for backwards
compatibility: callers that only need one facet should prefer the
narrowed :attr:`audio_playback` / :attr:`audio_streaming` /
:attr:`audio_buses` properties so the type checker enforces the
boundary. Reaching for the underscore-prefixed attribute is
engine-private and may change.
The engine sets this during ``App.run`` via ``make_backend``. Tests
and headless harnesses that don't initialise audio see ``None``.
"""
return getattr(self, "_audio_backend", None)
[docs]
def install_audio_backend(self, backend: AudioPlaybackBackend) -> None:
"""Install the audio backend for this tree: the one canonical path.
This is the only supported way to attach a backend; assigning the
private ``_audio_backend`` slot is no longer part of the contract.
Used by ``App.run`` / ``WebApp`` at startup and by tests that inject a
:class:`NullAudioBackend` or a mock.
Semantics mirror startup driver selection in other engines (Godot's
``--audio-driver``, pyglet's ``audio`` option): a backend is installed
once, before the tree starts producing sound. Re-installing shuts the
previous backend down first so device handles don't leak.
Facet conformance is enforced where it is consumed, not here: the
:attr:`audio_playback` / :attr:`audio_streaming` / :attr:`audio_buses`
accessors narrow via ``isinstance`` and return ``None`` for a backend
that doesn't implement that facet, and callers raise
:class:`AudioCapabilityError` on ``None``. Gating here as well would
duplicate that boundary and reject legitimate partial backends.
A backend that paces itself by the engine rather than by a device
clock declares :class:`~simvx.core.audio_protocol.TreeBoundAudioBackend`,
and installing hands it this tree: several trees can be alive at once,
so it has to be told which one's ticks are its own.
"""
if backend is None:
raise TypeError("install_audio_backend requires a backend, got None")
previous = getattr(self, "_audio_backend", None)
if previous is not None and previous is not backend:
shutdown = getattr(previous, "shutdown", None)
if callable(shutdown):
shutdown()
self._audio_backend = backend
from .audio_protocol import TreeBoundAudioBackend as _TBA
if isinstance(backend, _TBA):
backend.bind_tree(self)
[docs]
@property
def audio_playback(self) -> AudioPlaybackBackend | None:
"""The active backend narrowed to :class:`AudioPlaybackBackend`, or ``None``.
Always equals :attr:`audio_backend` cast to the playback facet when
a backend is present: every shipped backend implements playback,
including the silent :class:`NullAudioBackend`.
"""
from .audio_protocol import AudioPlaybackBackend as _APB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _APB) else None
[docs]
@property
def audio_streaming(self) -> AudioStreamingBackend | None:
"""The active backend narrowed to :class:`AudioStreamingBackend`, or ``None``.
Returns ``None`` when the active backend doesn't implement streaming
(the Null backend, any future no-device test stub). Callers that
need streaming (:class:`AudioSynth` driver, AudioWorklet feeds)
should raise :class:`AudioCapabilityError` on ``None``.
"""
from .audio_protocol import AudioStreamingBackend as _ASB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _ASB) else None
[docs]
@property
def audio_buses(self) -> AudioBusBackend | None:
"""The active backend narrowed to :class:`AudioBusBackend`, or ``None``.
Always equals :attr:`audio_backend` cast to the bus facet when a
backend is present: every shipped backend implements bus + capability
advertisement.
"""
from .audio_protocol import AudioBusBackend as _ABB
backend = self.audio_backend
if backend is None:
return None
return backend if isinstance(backend, _ABB) else None
[docs]
def audio_listener_3d(self) -> AudioListener3D | None:
"""The active 3D audio listener, lazy-creating one if none exists.
Returns the most recently entered :class:`AudioListener3D` in the
scene. If none has been added, auto-creates a fallback parented
to the active ``Camera3D`` with a one-time warning. Returns
``None`` only if there's no camera either.
Audio players call this every frame, so the auto-creation is
cheap once it's happened (the cached listener is returned).
"""
listener = self._current_audio_listener_3d
if listener is not None:
return listener
from .audio_listener import _autocreate_listener_3d
return _autocreate_listener_3d(self)
[docs]
def audio_listener_2d(self) -> AudioListener2D | None:
"""The active 2D audio listener, lazy-creating one if none exists.
Same contract as :meth:`audio_listener_3d` but for 2D.
"""
listener = self._current_audio_listener_2d
if listener is not None:
return listener
from .audio_listener import _autocreate_listener_2d
return _autocreate_listener_2d(self)
def __init__(self, screen_size=None, *, input_base: SceneTree | None = None):
"""Build a tree.
Args:
screen_size: The render target's size, ``(800, 600)`` by default.
input_base: Another tree whose input entries (its actions, device
aliases and redefinitions of engine entries) this one is
built on, read live, as a branch inherits the entries above
it. None, the default, starts from the engine's entries and
the process-wide map alone, which is what an editor hosting a
game wants: the choice is made here and never inferred.
**A tree owns its input.** It has its own
:class:`~simvx.core.input.map.InputMap`, its own contexts, its own
focus, and its own **delivered record** of the device elements that
actually reached it, which is the only thing its actions resolve
from. So an editor and the game it is running never share action
state, and neither has to ask for that.
Two things are underneath it and shared. The process-wide **device
record** says what is physically down: ``Input.is_key_pressed``,
``Input.capabilities``, ``Input.controllers`` and
``Input.input_method`` describe the hardware under the player and
say the same thing in every tree. And the process-wide ``InputMap``
is the bottom layer of this tree's, read live, so an action
registered before any tree existed, which is what a game's
``main()`` does, is one this tree resolves.
A test that needs the machine clean calls
:func:`simvx.core.input.devices.reset` and
:func:`simvx.core.input.capabilities.reset`, or ``Input._reset()``,
which covers the first.
"""
from .input.map import _InputMap
from .input.state import _Input
self._own_input_map = _InputMap(base=input_base.input_map if input_base is not None else None)
#: A new scene root's input context, carried across the swap that
#: mounts it so an entry both scenes declare is never gone in between.
self._input_ahead: InputContext | None = None
self._own_input = _Input(input_map=self._own_input_map)
self._own_input_map._state = self._own_input
# The tree's own top node, and the current scene as one of its children.
# Constructed with its tree reference already bound, so that mounting a
# singleton or a scene through ``_tree_root.add_child`` runs the ordinary
# enter/ready path and bumps ``_structure_version`` exactly as any other
# mount does.
self._scene_root: Node | None = None
self._tree_root = _TreeRoot(name="_tree_root")
self._tree_root._tree = self
# Depth of open activate_input() spans (re-entrant): > 0 while this tree
# is running its own node code, which is a tick, a physics step, a mount,
# a teardown, a layout flush, drawing, a group broadcast, or an input,
# UI, touch, pick or signal dispatch. Read by the InputMap late-
# registration warning to exempt the registrations node code makes while
# the engine is running it, and by the mount/unmount fast paths that skip
# opening a redundant nested span.
self._input_span_depth: int = 0
# The subtree an attach walk is currently binding, or None outside one.
# Node._enter_tree registers into this list on the way down and runs the
# entry hooks off it afterwards, so a hook sees a fully attached subtree.
self._attaching: list[Node] | None = None
# The render target this tree is presented in, in engine units: what a
# backend writes through ``screen_size``, and what the renderer reads.
# Under EXPAND it is also what ``screen_size`` reads back; under FIT
# the scene is laid out at ``content_scale_size`` instead and this is
# only the glass it is letterboxed onto. See :attr:`content_scale_mode`.
self._screen_size: tuple[float, float] = SceneTree._normalize_size(screen_size or (800, 600))
self._content_scale_mode: ContentScaleMode = ContentScaleMode.EXPAND
self._content_scale_size: tuple[float, float] = self._screen_size
# ``_letterbox()``'s answer, dropped whenever any of its three inputs
# moves. Every pointer event that arrives asks for it.
self._content_scale: tuple[float, float, float] | None = None
# Edges of the scene's space something is covering: a soft keyboard, a
# notch, a home indicator. Zero until a backend that can ask says so.
self._screen_insets: tuple[float, float, float, float] = (0.0, 0.0, 0.0, 0.0)
# The player's interface size: see ``ui_scale``.
self._ui_scale: float = 1.0
self._paused: bool = False
# How deep the tree is inside code its scene is running, and the root a
# scene change made from that code is waiting to mount: see
# ``change_scene``.
self._scene_depth: int = 0
self._next_scene: Node | None = None
# True while ``_swap_scene`` is carrying the old scene out: a scene
# change asked for from its ``on_exit_tree`` is not a second swap.
self._scene_swapping: bool = False
self._scene_span = _SceneSpan(self)
self._delete_queue: list[Node] = []
# Each group's members in join order: a dict keyed by node is an
# insertion-ordered set with O(1) add and remove.
self._groups: dict[str, dict[Node, None]] = {}
self._singletons: dict[str, Node] = {}
# Transient fire-and-forget timers created via ``create_timer``. Ticked
# right after singletons each frame and dropped once they fire, so a
# one-shot delay needs no Node in the scene and no manual cleanup.
self._scene_timers: list[SceneTreeTimer] = []
# Escape-hatch deferred calls queued via ``Node.call_deferred``. Drained
# once per frame at the end-of-frame sync point (after traversal, before
# the delete flush). Calls queued during the drain run next frame.
self._deferred_calls: list[tuple[Callable[..., Any], tuple]] = []
# Engine-provided typed event bus, accessible as ``tree.events``. Held
# outside ``_singletons`` because EventBus is not a Node: entries there
# are children of the tree's own top node, and the bus does not
# participate in the process/physics traversal -- the per-frame tick
# explicitly calls ``self._events.flush_deferred()``. The name
# ``events`` is reserved (see add_singleton()).
self._events: EventBus = EventBus()
self._unique_nodes: dict[str, Node] = {}
# Input dispatch tables: populated as @on_input handlers register on tree-enter.
# Keyed for O(1) lookup so input dispatch is O(handlers per event), not O(nodes).
# A filter naming several targets registers in one bucket per target, so a
# bucket is always reached by a single lookup and never scanned for a match.
self._action_handlers: dict[tuple[str, bool], _HandlerBucket] = {}
#: Every node in this tree whose focus_mode lets it hold focus: what Tab
#: and directional focus search, so neither walks the other nodes.
self._focusables: set[Node] = set()
#: The subset whose focus_mode follows hover. Empty, hover focus costs nothing.
self._hover_focusables: set[Node] = set()
self._drag_threshold: dict[PointerType, float] = dict(DEFAULT_DRAG_THRESHOLD)
#: How long a finger rests, unmoved, before it is a ``Touch.LONG_PRESS``, seconds.
self.long_press_time: float = 0.5
#: How quickly two fingers must touch and lift to be a ``Touch.TWO_FINGER_TAP``, seconds.
self.tap_time: float = 0.3
#: The recognised controls (a gesture, the wheel) some handler here names,
#: so the router recognises them; never shrinks, like the capture flag.
self._recognised_by_handlers: frozenset[DeviceControl] = frozenset()
#: The SceneHost this tree is the guest of, or None for a window's tree.
self._host: SceneHost | None = None
#: The SceneHosts inside this tree, each audited once a frame; empty, nothing is.
self._hosts: set[SceneHost] = set()
#: The controls naming a ``cursor_shape``. Empty, the cursor is never looked up.
self._cursor_shaped: set[Node] = set()
# The device tables of the ordinary, deepest-first phase, and of the
# capture phase, which runs root-first ahead of every control.
self._bubble_tables = _HandlerTables()
self._capture_tables = _HandlerTables()
#: Whether any capture handler has ever registered here: a tree with
#: none never looks for one.
self._has_capture_handlers = False
#: The ``samples=True`` handlers, which hear each motion sample as it
#: arrives rather than the frame's merged move, and whether any ever
#: registered: a tree with none has no sample built for it.
self._sample_handlers = _HandlerBucket()
self._has_sample_handlers = False
#: Whether any ordinary-phase ``repeat=True`` key handler ever registered:
#: a tree with none never offers an auto-repeat to the broadcast.
self._has_repeat_handlers = False
#: The node whose capture handler claimed each held key's press, which
#: owns that key's auto-repeats until its release.
self._capture_keys: dict[DeviceButton, Node] = {}
#: The node that claimed each held key's press after the capture phase,
#: kept only in a tree with ordinary-phase ``repeat=True`` handlers: it
#: owns that key's auto-repeats in the broadcast until its release.
self._repeat_keys: dict[DeviceButton, Node] = {}
#: The nodes whose class hears a release's pick. Empty, a release casts no ray.
self._release_pickers: set[Node] = set()
self._key_handlers = self._bubble_tables.key
self._pointer_handlers = self._bubble_tables.pointer
self._device_button_handlers = self._bubble_tables.device_button
self._device_axis_handlers = self._bubble_tables.device_axis
self._event_class_handlers = self._bubble_tables.event_class
self._catch_all_handlers = self._bubble_tables.catch_all
# Every bucket ever created, for the diagnostics that want to see all of
# them at once. Dispatch reaches a bucket by routing key and tree-exit
# by the index below; neither walks this.
self._input_buckets: list[_HandlerBucket] = [
self._catch_all_handlers,
self._capture_tables.catch_all,
self._sample_handlers,
]
# Where each node class registers, resolved on its first mount in this
# tree. Registration and tree-exit both read it, so neither walks the
# filters per node nor asks every bucket whether it has heard of one.
self._registration_plans: dict[type, tuple[tuple, ...]] = {}
# Publishing from inside a dispatch queues rather than nests: the frame's
# events stay in arrival order and the current one finishes first. Held on
# the tree rather than the router because several routers can feed one tree
# and they must share one queue.
# Bumped only when the *relative* order of nodes changes, which is a
# reorder and an unlink-without-exit and nothing else. Appending or
# removing a child shifts sibling indices and leaves every pair in the
# order it was, so a spawned and freed bullet that binds no input costs
# the dispatch tables nothing.
self._input_order_version: int = 0
self._input_dispatching: bool = False
self._input_queue: deque[tuple[Any, InputEvent]] = deque()
#: The InputListener nodes in this tree, in activation order: the last hears first.
self._input_listeners: list[Any] = []
#: Action releases a context change produced while a dispatch, an
#: attach or a teardown was in flight, drained once that finishes.
self._deferred_action_edges: deque[ActionEvent] = deque()
#: Depth of the subtree teardowns running on this tree. A release edge
#: waits for them: a handler must not run against a tree halfway
#: through losing its nodes.
self._input_edge_hold: int = 0
# The routers feeding this tree, so a scene change can tell them the
# pointers they were tracking are gone. Weak, because a router holds its
# tree: a strong set would make every tree and router pair a cycle that
# only the collector could break.
self._routers: weakref.WeakSet[Any] = weakref.WeakSet()
from .ui.ui_input import UIInputManager # local import avoids cycle via ui/testing.py
self._ui = UIInputManager()
self._ui._tree = self # back-ref so the manager can read the overlay registry
self._current_camera_2d: Camera2D | None = None
# Active audio listeners: set by AudioListener2D/3D's on_enter_tree.
# Audio players read these via the audio_listener_2d() / _3d()
# accessors below, which lazy-create a fallback at the active
# camera if no explicit listener exists in the scene.
self._current_audio_listener_2d: AudioListener2D | None = None
self._current_audio_listener_3d: AudioListener3D | None = None
self.auto_physics: bool = True # Automatically step the physics world(s) each physics tick
self.physics_interpolation: bool = True # render-time lerp of new-world bodies between fixed steps
# Default physics worlds. Both slots stay empty/lazy so non-physics scenes
# allocate nothing.
self._physics_world: PhysicsWorld | None = None # lazily-created default 3D world
self._physics_world_2d: Physics2DWorld | None = None # lazily-created default 2D world
# Registry of active PhysicsRoot / PhysicsRoot2D worlds. Holds BOTH 3D and
# 2D worlds (each registers via register_physics_world); the step/sync/
# dispatch loop handles them polymorphically once the leaves are width-aware.
self._physics_worlds: list[Any] = []
# Handle->node sync registry: per-world _BodySync used to scatter
# simulated transforms back onto body nodes after step.
# Empty for non-physics / all-static scenes (zero overhead).
self._physics_bodies: dict[AnyPhysicsWorld, _BodySync] = {}
# Handle->node map for collision-event dispatch.
# Holds EVERY PhysicsBody3D (static included, unlike _physics_bodies),
# weakly, so _dispatch_contact_events can resolve both sides of a pair.
# Empty for non-physics scenes (zero overhead).
self._physics_nodes: dict[AnyPhysicsWorld, weakref.WeakValueDictionary[BodyHandle, Node]] = {}
# Bodies destroyed but still owed their closing contact/overlap events:
# handle -> (node, step sequence at which it was retired). STRONG refs, so
# a peer the game has already dropped survives long enough to be named.
# Released one step after retirement; empty for non-physics scenes.
self._physics_retired: dict[AnyPhysicsWorld, dict[BodyHandle, tuple[Node, int]]] = {}
# Per-world count of steps this tree has driven, the clock the retirement
# grace period is measured on.
self._physics_step_seq: dict[AnyPhysicsWorld, int] = {}
# The tree-owned overlay layer: ordered registry of on-top UI overlays
# (dropdowns, dialogs, tooltips). Empty registry is an O(1) zero-cost gate.
from .ui.overlay import OverlayLayer
self.overlays = OverlayLayer(self)
# Layout: dirty containers awaiting reflow, drained by ``flush_layout``
# (end-of-tick and synchronously on overlay/popup open). Empty set is the
# O(1) zero-cost gate. ``_layout_frame`` is this tree's private rect-cache
# epoch (per-tree so a second ticking tree does not invalidate this one's
# cached global rects); ``flush_layout`` and ``tick`` advance it.
self._dirty_layout: set = set()
self._layout_frame: int = 0
self.play_viewport_rect: tuple[float, float, float, float] | None = (
None # (x, y, w, h) to constrain 3D rendering
)
self._structure_version: int = 0 # Incremented on add_child/remove_child for cache invalidation
self._camera_3d_cache: tuple[tuple[int, Node], tuple[Camera3D, ...]] | None = None
self._running: bool = True
# Frame counter: bumped at the start of every ``tick`` so callers
# (notably InputMap.add_action's late-call warning) can tell whether
# the scene has begun ticking. Stays at 0 inside the root node's
# initial ``ready`` callback so registrations there are silent.
self._tick_count: int = 0
# Monotonically-increasing scene time, in seconds. Accumulates ``dt``
# at the start of every ``tick()`` (after any ``App.time_scale``
# scaling has been applied to ``dt``) so node code can read a single
# shared clock instead of maintaining its own ``_time += dt``
# accumulator. Frozen while ``self.paused`` is True.
self._now: float = 0.0
# ``Property(on_change=..., coalesce=True)`` enqueues ``(obj, method_name)``
# pairs here instead of firing synchronously. Drained once at the end of
# ``process()``: multiple writes within one tick collapse to a single call.
# Set semantics dedupe automatically.
self._pending_coalesced_hooks: set[tuple[Any, str]] = set()
self.quit_requested: Signal = Signal()
# Overlay lifecycle: fires when any Control's show_overlay()/close_overlay()
# opens/closes a registry entry. Receivers get the overlay Control so
# external listeners (e.g. game-side pause bridges) can react without being
# involved in the overlay itself.
self.overlay_opened: Signal = Signal()
self.overlay_closed: Signal = Signal()
# Fires whenever ``screen_size`` changes (window resize, fullscreen
# toggle, viewport swap). Receivers get the new ``(width, height)``
# tuple and can recompute layout once instead of every frame.
self.screen_resized: Signal = Signal()
# Fires when ``ui_scale`` changes, with the new value. The UI space
# (``screen_size / ui_scale``) moved while the screen did not.
self.ui_scale_changed: Signal = Signal()
# Fires when ``screen_insets`` changes, with the new
# ``(left, top, right, bottom)``: a soft keyboard opened or closed, a
# phone rotated its notch to another edge.
self.screen_insets_changed: Signal = Signal()
# -- UI state forwarding (preserves external interface) --
@property
def _last_mouse_pos(self):
return self._ui._last_mouse_pos
@_last_mouse_pos.setter
def _last_mouse_pos(self, v):
self._ui._last_mouse_pos = v
@staticmethod
def _normalize_size(sz) -> tuple[float, float]:
"""Coerce any screen_size representation to a plain tuple once."""
if isinstance(sz, tuple) and len(sz) == 2:
return sz
if hasattr(sz, "x"):
return (float(sz.x), float(sz.y))
return (float(sz[0]), float(sz[1]))
@property
def paused(self) -> bool:
"""Whether the scene is suspended: a node runs only if it says so.
A node's ``update_mode`` decides what the pause means for it:
``PAUSABLE`` stops, ``ALWAYS`` carries on, ``PAUSED_ONLY`` runs only
while this is True, which is what a pause menu is. The switch reaches
input as well as processing, so a paused player neither polls nor
hears, and an ``input_context`` on a paused node stops declaring and
stops hiding until the node wakes.
"""
return self._paused
[docs]
@paused.setter
def paused(self, value: bool) -> None:
value = bool(value)
if value == self._paused:
return
self._paused = value
# Which contexts are live is a function of this, so the flatten has to
# run again. It is also what releases the paused subtree's held
# actions, through the ordinary release-on-change path.
self._own_input_map._refresh_active()
[docs]
@property
def is_running(self) -> bool:
"""Whether the tree is actively being ticked by its driving app.
Set to False by :meth:`quit` (or the graphics backend's app ``quit()``),
signalling the main loop to exit at the end of the current frame.
"""
return self._running
[docs]
@property
def now(self) -> float:
"""Monotonically-increasing scene time, in seconds.
Accumulates ``dt`` at the start of every :meth:`tick` call (after
any :attr:`simvx.graphics.App.time_scale` scaling has been applied),
so slow-motion and hitstop also slow this clock. Frozen while
:attr:`paused` is True or any inert overlay is open. Resets to
``0.0`` only by constructing a fresh ``SceneTree``.
Use for time-based animation, slow-mo gating, and any "how long has
the scene been running" query: preferable to per-node
``self._time += dt`` accumulators because every consumer reads the
same monotonic value.
"""
return self._now
[docs]
def quit(self) -> None:
"""Request a clean shutdown of the running tree.
Emits :attr:`quit_requested` and flips :attr:`is_running` to False. The
driving app polls this state and exits its loop at the end of the
current frame. Safe to call from node callbacks or signal handlers.
"""
if not self._running:
return
self._running = False
with self.activate_input():
self.quit_requested.emit()
# Release the physics worlds on shutdown (default world + PhysicsRoot
# registry) so a torn-down tree leaks nothing. change_scene deliberately
# does NOT drop these: the default world is tree-scoped (survives swaps).
self._drop_physics_worlds()
@property
def screen_size(self) -> tuple[float, float]:
"""The size of the space this scene is laid out in, in engine units.
Under :attr:`content_scale_mode` ``EXPAND``, the default, that is the
render target: a wider window is a wider scene and shows more.
Under ``FIT`` it is :attr:`content_scale_size`, the design resolution,
whatever the window is doing, because under FIT the scene's space *is*
design units all the way through.
Assigning says the render target is now this big, which is what a
backend does on a resize. Under EXPAND that is the same statement as
the one above; under FIT it changes the letterbox and leaves what this
reads back alone. The three sizes and which question each answers are
set out in :attr:`content_scale_mode`.
An ``on_draw`` that read it is re-captured when it changes.
"""
reader = draw_context.node
if reader is not None:
reader._ambient_reads |= READS_VIEWPORT_SIZE
if self._content_scale_mode is ContentScaleMode.FIT:
return self._content_scale_size
return self._screen_size
[docs]
@screen_size.setter
def screen_size(self, value):
self._set_window_size(SceneTree._normalize_size(value))
def _set_window_size(self, size: tuple[float, float]) -> None:
"""Record the render target's size and announce it if the scene's space moved.
Under FIT the scene's space is the design resolution, so a window
resize moves the letterbox without moving anything the scene is laid
out against: nothing is invalidated and ``screen_resized`` does not
fire, because for the scene nothing resized.
"""
old_window, old_space = self._screen_size, self.screen_size
self._screen_size = size
if self._screen_size != old_window:
self._content_scale = None
new_space = self.screen_size
if new_space != old_space:
with self.activate_input():
self._invalidate_for_target_change()
self.screen_resized.emit(new_space)
[docs]
@property
def window_size(self) -> tuple[float, float]:
"""The render target this scene is presented in, in engine units.
The glass, as opposed to :attr:`screen_size`, which is the space the
scene is laid out in. They are the same under ``EXPAND`` and differ
under ``FIT``.
"""
return self._screen_size
@property
def content_scale_mode(self) -> ContentScaleMode:
"""How the scene's space relates to the render target it is shown in.
``EXPAND`` (the default) means an engine unit is a logical pixel of the
render target: the scene's space is the window, a wider window shows
more world, and a HUD anchored to a screen edge is on the glass. This
is what anything with a HUD wants and it is what every backend does.
``FIT`` preserves a design resolution for a fixed playfield: the scene
is always :attr:`content_scale_size` units across, that rect is scaled
up uniformly to the largest that fits the window and centred, and the
leftover bars belong to no viewport. Under FIT the root viewport's
space **is** design units: the letterbox is applied once, where input
enters, so ``PointerEvent.position``, ``Input.mouse_position``,
:attr:`screen_size` and every ``Control`` rect are already design units
and no consumer transforms anything. A pointer over a bar has a design
coordinate outside the design rect and therefore hits nothing, by
ordinary geometry rather than by a special case.
**What each size answers, which is the thing to be careful about.**
These three agree under EXPAND and deliberately differ under FIT:
========================== ==========================================
``app.width`` the window. The glass, always, because a
game placing an OS dialog or remembering a
window size needs the real one
``tree.screen_size`` the space the scene is laid out in: design
units under FIT
``node.viewport_size`` the target that node draws into, which is
``screen_size`` unless a ``SubViewport``
ancestor renders it offscreen
========================== ==========================================
A game that wants a screen-edge HUD wants EXPAND. Under FIT a HUD
anchored to an edge lands on the edge of the design rect, not of the
glass, which is right for a playfield and wrong for a HUD that wants
the whole screen: FIT buys one coherent space at the price of the bars.
"""
return self._content_scale_mode
[docs]
@content_scale_mode.setter
def content_scale_mode(self, value: ContentScaleMode | str) -> None:
mode = ContentScaleMode(value)
if mode is self._content_scale_mode:
return
old_space = self.screen_size
self._content_scale_mode = mode
self._content_scale = None
new_space = self.screen_size
if new_space != old_space:
with self.activate_input():
self._invalidate_for_target_change()
self.screen_resized.emit(new_space)
@property
def content_scale_size(self) -> tuple[float, float]:
"""The design resolution ``FIT`` preserves, in engine units.
Defaults to the size the tree was built at, which is the size the game
asked its window for, so a game that switches to FIT and says nothing
else keeps the layout it authored. Ignored under ``EXPAND``.
"""
return self._content_scale_size
[docs]
@content_scale_size.setter
def content_scale_size(self, value) -> None:
size = SceneTree._normalize_size(value)
if size[0] <= 0 or size[1] <= 0:
raise ValueError(f"content_scale_size must be positive on both axes, got {size}")
if size == self._content_scale_size:
return
old_space = self.screen_size
self._content_scale_size = size
self._content_scale = None
if self.screen_size != old_space:
with self.activate_input():
self._invalidate_for_target_change()
self.screen_resized.emit(size)
[docs]
@property
def content_scale(self) -> float:
"""Render-target units per engine unit: 1.0 under ``EXPAND``.
Under ``FIT`` this is how much the design rect is magnified to fill the
window, which is what a delta measured on the glass has to be divided
by to become a delta in the scene's space. :meth:`window_to_viewport`
does that for a position; this is for everything else.
"""
return self._letterbox()[0]
[docs]
@property
def presentation_rect(self) -> tuple[float, float, float, float]:
"""``(x, y, w, h)`` of the glass the scene's space is presented into.
In engine units, measured on :attr:`window_size`. The whole window
under ``EXPAND``; under ``FIT`` the design rect scaled uniformly and
centred, with the letterbox bars in the margin.
The exact inverse of :meth:`window_to_viewport`, and computed from the
same placement, so what a renderer draws and what a pointer hits cannot
drift: a point at the rect's top-left corner converts to ``(0, 0)`` in
the scene's space, a point at its bottom-right to
:attr:`screen_size`, and a point in a bar to a coordinate outside both.
"""
scale, offset_x, offset_y = self._letterbox()
width, height = self.screen_size
return (offset_x, offset_y, width * scale, height * scale)
def _letterbox(self) -> tuple[float, float, float]:
"""``(scale, offset_x, offset_y)`` placing the scene's space on the glass.
Remembered until the window, the mode or the design size moves, because
every pointer event that arrives asks for it.
"""
cached = self._content_scale
if cached is not None:
return cached
if self._content_scale_mode is ContentScaleMode.FIT:
window_w, window_h = self._screen_size
design_w, design_h = self._content_scale_size
scale = min(window_w / design_w, window_h / design_h)
result = (scale, (window_w - design_w * scale) * 0.5, (window_h - design_h * scale) * 0.5)
else:
result = (1.0, 0.0, 0.0)
self._content_scale = result
return result
@property
def screen_insets(self) -> tuple[float, float, float, float]:
"""Edges of the scene's space something is covering, as ``(left, top, right, bottom)``.
In screen pixels, measured inward from the edges of :attr:`screen_size`,
and all zero unless a backend says otherwise. A soft keyboard sliding
up is a bottom inset, a notch is a top or side one, and a home
indicator is a small bottom one.
A node keeping something out of them reads ``node.viewport_insets``
instead, which answers in the pixels of the space that node draws in:
a ``Control`` under :attr:`ui_scale` gets UI pixels it can use as
margins directly, where these would be too large by the factor.
Only a backend that can actually ask the platform writes this, and one
that can declares ``Input.capabilities.screen_insets`` when it does.
A backend that cannot declares nothing and this stays zero, which is
the honest answer rather than a gap: zero insets is exactly what a
desktop window has.
Assigning takes insets measured on the glass, which is what a platform
reports, and is the backend's door in the way :attr:`screen_size` is.
Under ``FIT`` they are converted: the letterbox bars already sit
between the window edge and the scene's, so an inset that only reaches
into a bar occludes nothing and arrives here as zero.
A change, and only a change, re-captures every ``on_draw`` that read
the insets and then fires ``screen_insets_changed`` with the new
value, which is where code outside ``on_draw`` that keeps something
clear of them listens. A backend may therefore write them every frame.
"""
reader = draw_context.node
if reader is not None:
reader._ambient_reads |= READS_VIEWPORT_INSETS
return self._screen_insets
[docs]
@screen_insets.setter
def screen_insets(self, value) -> None:
left, top, right, bottom = (float(v) for v in value)
if min(left, top, right, bottom) < 0.0:
raise ValueError(f"screen_insets are distances inward and cannot be negative, got {value!r}")
scale, offset_x, offset_y = self._letterbox()
if scale != 1.0 or offset_x != 0.0 or offset_y != 0.0:
left = max(0.0, left - offset_x) / scale
right = max(0.0, right - offset_x) / scale
top = max(0.0, top - offset_y) / scale
bottom = max(0.0, bottom - offset_y) / scale
insets = (left, top, right, bottom)
if insets == self._screen_insets:
return
self._screen_insets = insets
redraw_readers(self._tree_root, READS_VIEWPORT_INSETS)
with self.activate_input():
self.screen_insets_changed.emit(insets)
@property
def ui_scale(self) -> float:
"""How big the interface is drawn, as the player chose it. 1.0 is as authored.
Every ``Control`` in this tree is laid out and drawn at this size; the
game world is not. A settings slider assigns it and the interface
resizes on the next frame, the slider included::
def on_ready(self):
slider = self.add_child(Slider(1.0, 2.0, value=self.tree.ui_scale))
slider.value_changed.connect(self._set_ui_scale)
def _set_ui_scale(self, value):
self.tree.ui_scale = value
The scale lives in the space rather than in the widgets: the outermost
control of each interface lays out against ``screen_size / ui_scale``
and carries the factor in its transform, so fonts, paddings, minimum
sizes and a game's own hard-coded ``40.0`` all grow together, and a
click finds the control it lands on because hit testing goes through
the same transform. Text is laid out at the size it is drawn at rather
than laid out small and magnified, so glyphs stay on the pixel grid at
fractional factors. ``Control.viewport_size`` answers in the divided
space, which is the space a control's ``on_draw`` works in.
It is a preference, not a density: an engine unit is already one
density-independent pixel, so a 44-unit button is the same physical
size on every screen at 1.0 and nothing here should be multiplied by a
display's pixel ratio. It belongs to this tree, so an editor and the
game running inside it are scaled independently, and it survives
:meth:`change_scene` because the tree does. Keeping it across runs is
the game's job: load it and assign it in the root's ``on_ready``.
A world node's ``on_draw`` that read it is re-captured when it
changes, as every control's is.
Raises:
ValueError: the value is not a finite number above zero.
"""
reader = draw_context.node
if reader is not None:
reader._ambient_reads |= READS_UI_SCALE
return self._ui_scale
[docs]
@ui_scale.setter
def ui_scale(self, value: float) -> None:
scale = float(value)
if not (math.isfinite(scale) and scale > 0.0):
raise ValueError(f"ui_scale must be a finite number above zero, got {value!r}")
if scale == self._ui_scale:
return
self._ui_scale = scale
# Every control's rect, transform and text layout depend on the factor.
# The transform caches hold for one layout epoch, so a new epoch retires
# them; the retained draws are marked here, the way a resize marks them.
self._layout_frame += 1
self._invalidate_ui_draws()
with self.activate_input():
self.ui_scale_changed.emit(scale)
def _invalidate_ui_draws(self) -> None:
"""Mark every control, and everything drawn inside one, for a fresh draw.
A change of :attr:`ui_scale` moves and resizes every interface and
changes the size its text is laid out at, so both the transform and the
captured geometry of each node under a ``Control`` are stale, and so is
every clip a control opened over its children, which only a fresh
collection re-opens: ``_layout_rect_dirty`` asks the retained pipeline
for one. Nodes outside every control, the game world, are left alone
unless their last draw read the scale, ``Control.viewport_size``
included, in which case they are re-captured too.
"""
from .nodes_2d.node2d import Node2D
from .ui import Control
stack: list[tuple[Node, bool]] = [(self._tree_root, False)]
while stack:
node, in_ui = stack.pop()
if isinstance(node, Control):
in_ui = True
node._draw_dirty = True
node._draw_cache = None
node._rect_frame = -1
node._layout_rect_dirty = True
if in_ui and isinstance(node, Node2D):
node._transform_dirty = True
node._render_dirty = True
node._transform_render_dirty = True
elif node._ambient_reads & READS_UI_SCALE:
node.queue_redraw()
stack.extend((child, in_ui) for child in node.children)
[docs]
def window_to_viewport(self, x: float, y: float) -> tuple[float, float]:
"""A point on the glass, as a point in the space the scene is laid out in.
The identity under ``EXPAND``. Under ``FIT`` it undoes the letterbox,
and a point in a bar comes back outside the design rect rather than
clamped to its edge: the bars belong to no viewport, so a pointer over
one has to hit nothing, and a coordinate that is simply out of range
does that without anything having to check for it.
"""
scale, offset_x, offset_y = self._letterbox()
if scale == 1.0 and offset_x == 0.0 and offset_y == 0.0:
return (x, y)
return ((x - offset_x) / scale, (y - offset_y) / scale)
@property
def root(self) -> Node | None:
"""The current scene's root node.
A sibling of this tree's singletons under the tree's own top node, and
the node every scene-facing walk starts from: rendering, 3D submission,
saving, hot reload and the editor's scene panel all begin here, so a
singleton is ticked but is not part of the scene.
Assigning re-links the child and runs no lifecycle, which is what
assigning to it has always done. Use :meth:`set_root` to mount a scene
(enter + ready) and :meth:`change_scene` to swap one.
"""
return self._scene_root
[docs]
@root.setter
def root(self, node: Node | None) -> None:
old = self._scene_root
if old is not None:
# Clears ``_scene_root`` through the _TreeRoot override.
self._tree_root._detach_child(old)
self._scene_root = node
if node is not None:
self._tree_root._attach_child(node)
# The scene is always the tree root's last child, so it processes
# after every singleton (see ``tick``).
self._tree_root.children.move_last(node)
[docs]
def set_root(self, root: Node):
"""Mount ``root`` as the current scene.
The root's ``input_context``, like any node's, comes into force as it
enters, and goes with it when another scene replaces it.
A scene already mounted is carried out first, so its ``on_exit_tree``
runs, but it is not destroyed. ``change_scene`` remains the call for a
swap: it destroys the old scene and also clears popup state and the
active 2D camera.
The mount itself runs inside this tree's input span, so
``on_enter_tree`` / ``on_ready`` code sees this tree's ``Input`` and
``InputMap``, and any action it registers lands in this tree's own map.
"""
log.debug("SceneTree.set_root(%s)", root)
# Publish "this is the active tree" so input map / asset paths that
# need cross-cutting context can find us without a parameter chain.
# Weak so we do not pin the previous/last tree's node graph.
SceneTree._active_tree = weakref.ref(self)
self._bring_input_ahead(root)
self._detach_scene_root()
# Published before the mount: an ``on_enter_tree`` reads ``tree.root``
# (``_FullscreenOverlay`` decides whether to register on the overlay
# layer by it), so the reference has to be live by the time the entry
# hooks run.
self._scene_root = root
with self.activate_input():
# attach + enter + ready, and the scene lands last among the tree
# root's children.
self._tree_root.add_child(root)
ahead, self._input_ahead = self._input_ahead, None
if ahead is not None:
# The root holds its own entry now; the stand-in goes, and the
# entries it carried across the swap never left.
self._own_input_map._deactivate(ahead, None)
def _detach_scene_root(self) -> None:
"""Carry the current scene out of the tree, if there is one."""
old = self._scene_root
if old is None:
return
if old.parent is not self._tree_root:
# Somebody reparented the scene root without going through
# ``_detach_child``, so "the scene is a child of the tree root" no
# longer holds and there is nothing here to carry out.
self._scene_root = None
if Node.strict_errors:
raise RuntimeError(
f"Scene root {old.name!r} is no longer a child of the tree. "
"Reparent a scene root with change_scene(), not add_child()."
)
return
# Opens the input span, runs _exit_tree, and clears _scene_root.
self._tree_root.remove_child(old)
self._prune_singletons()
self._prune_dead_index_entries()
[docs]
def change_scene(self, new_root: Node):
"""End the current scene and mount ``new_root`` in its place.
The old scene is destroyed: every node in it runs ``on_exit_tree``,
then is finished exactly as :meth:`Node.destroy` finishes a node, so
its signal connections and ``EventBus`` subscriptions end there and
then, and none of its nodes can be added to a tree again. A lambda or
other strongly held callable it subscribed still fires until it is
unsubscribed, as it would after ``destroy()``. To carry state into the
next scene, pass values to ``new_root``, not nodes of the old one.
``new_root`` then runs the full ``_enter_tree`` / ``_ready_recursive``
path, identical to the initial root. Singletons the tree owns are left
in place, keeping their groups and unique names; a singleton
registered on a node the old scene owned leaves with that scene, and
its name binding is dropped. UI popup state and the active 2D camera
are cleared.
**Called from the scene's own code, the call does not return.** A hook,
an input handler, a signal a control emitted or a contact handler that
the tree is running for the scene ends that scene where it stands: the
code that asked unwinds (its ``finally`` blocks run, with the old scene
still in place), nothing further in the old scene runs that frame, and
the tree then carries the old scene out and mounts ``new_root`` from its
own frame, never from inside the old scene's walk. So the first scene
change a scene makes is the one that happens; a second node that would
have decided otherwise later in the same walk is never reached. A scene
changed from ``on_update`` starts on the next frame; one changed from an
input handler or ``on_fixed_update`` runs the rest of the frame.
Called from anywhere else (a singleton, a deferred call, a timer, code
outside the frame), it swaps at once and returns with ``tree.root``
already the new scene.
The new scene is given no edge for anything already held when it
arrived, because that press was delivered to the scene that ended: a
held key reads at rest there until it is let go and pressed again.
A change asked for while the old scene is being carried out (from one
of its ``on_exit_tree`` hooks) is ignored: the change already under
way is the one that happens.
Use this for title → gameplay → game-over navigation. See
:doc:`../patterns` for a full example.
Raises:
ValueError: ``new_root`` is the current scene or a node inside it,
or has been destroyed or is about to be (a scene an earlier
``change_scene`` replaced, kept to come back to). Nothing has
changed when it raises.
"""
log.debug("SceneTree.change_scene(%s), old root=%s", new_root, self._scene_root)
self._refuse_unmountable(new_root)
if self._scene_swapping:
log.debug("change_scene(%s) ignored: the old scene is already being replaced", new_root)
return
if self._scene_depth:
# A second request while the scene is already ending comes from its
# own unwind (a ``finally``): the first decision stands.
if self._next_scene is None:
self._next_scene = new_root
raise _SceneEnded
self._swap_scene(new_root)
def _refuse_unmountable(self, new_root: Node) -> None:
"""Raise, before anything is torn down, when *new_root* cannot become the scene."""
if new_root._destroyed or new_root._destroying:
raise ValueError(
f"change_scene({new_root.name!r}) names a destroyed node. A scene that change_scene "
"replaced is destroyed with it, so it cannot be mounted again: build a fresh root instead"
)
old = self._scene_root
if old is not None:
node: Node | None = new_root
while node is not None:
if node is old:
raise ValueError(
f"change_scene({new_root.name!r}) names the scene it replaces, or a node inside it, "
"and the old scene is destroyed: pass a freshly built root instead"
)
node = node.parent
def _mount_next_scene(self) -> None:
"""Mount the root a scene change asked for, once the scene's code has unwound."""
new_root = self._next_scene
if new_root is not None:
self._next_scene = None
self._swap_scene(new_root)
def _bring_input_ahead(self, root: Node) -> None:
"""Put *root*'s input context on the stack before the scene it replaces leaves.
So an entry both declare (a "start" on the title and in the game) is
never gone in between, and the press that started the game is not
released under it. Refused here, with the old scene intact, where the
new root's entries cannot resolve. The mount then finds it active.
"""
context = root._input_context
if context is not None and self._scene_root is not None and self._input_ahead is None:
# Held by no node while the swap runs, so it counts from now: the
# root is not in the tree yet and could not hear anything.
self._own_input_map._activate(context, None)
self._input_ahead = context
def _swap_scene(self, new_root: Node) -> None:
"""End the current scene, destroying it, and mount ``new_root``, from outside the scene's code."""
# Checked again here: a root recorded from inside the scene may have
# been destroyed while the scene's code unwound.
self._refuse_unmountable(new_root)
self._bring_input_ahead(new_root)
old = self._scene_root
self._scene_swapping = True
try:
self._detach_scene_root()
if old is not None and old._tree is None:
# Out of the tree with every on_exit_tree run, so it is finished
# at once: its connections and bus subscriptions end here rather
# than whenever the collector reaches it. A delete it had pending
# went with it; one a singleton queued still runs at the end of
# the frame, even when this swap runs inside that flush.
old._spend_subtree()
self._delete_queue[:] = [node for node in self._delete_queue if not node._destroyed]
finally:
self._scene_swapping = False
# The old scene left in one piece, so nothing is owed a closing contact
# event; release the retired bodies rather than hold a whole dead scene
# alive for one more step.
self._physics_retired.clear()
self._ui.reset()
# The pointers in flight belonged to the scene that just left: the
# finger holding the emulated mouse is not on the new scene's glass.
for router in tuple(self._routers):
router.reset()
self._ui._hover_dirty = False
self._current_camera_2d = None
# Events the old scene published while one of its own was being
# delivered went with it.
self._input_queue.clear()
self._retire_delivered_input()
self.set_root(new_root)
def _retire_delivered_input(self) -> None:
"""Leave what the outgoing scene was given with it.
The edges not yet read by a fixed tick were delivered to the scene that
ended, and whatever is held now was pressed before the new scene
existed, so it reads at rest there until the player lets go and presses
again: the rule a context change and an input handover already follow.
The new root's own actions are marked the same way as they arrive.
"""
state = self._own_input
state._pending_edges.clear()
if not state._anything_held():
return
with self.activate_input():
for entry in self._own_input_map._actions.values():
state._mark_stale(entry.bindings)
#: Backstop for pathological layout feedback (a container that re-dirties
#: itself every pass). Layout converges in ~2 passes by idempotence (size
#: ``on_change`` fires only on an actual change), so this only ever caps a bug.
_LAYOUT_FLUSH_CAP = 8
def _register_dirty_layout(self, container) -> None:
"""Record ``container`` as needing reflow (called by ``Container.mark_layout_dirty``)."""
self._dirty_layout.add(container)
[docs]
def flush_layout(self) -> None:
"""Reflow every dirty container now, root-first, and expire the rect cache.
Synchronous layout settle: after this returns, container children have
their final positions/sizes for this frame, so ``get_global_rect`` is
valid immediately (no one-tick lag). Called at end-of-tick and, crucially,
right when an overlay/popup opens so a freshly-shown dialog can be
hit-tested and focus-elected on the same tick. Idempotent and O(1) when no
container is dirty.
The reflow itself runs inside this tree's input span: ``_do_update_layout``
is an overridable hook, so a custom container laying itself out sees this
tree's ``Input`` and ``InputMap``.
"""
dirty = self._dirty_layout
if not dirty:
return
def _depth(node) -> int:
d, cur = 0, node.parent
while cur is not None:
d += 1
cur = cur.parent
return d
with self.activate_input():
for _ in range(self._LAYOUT_FLUSH_CAP):
if not dirty:
break
# Root-first: a parent's size feeds its children's layout, so lay out
# shallower containers before deeper ones (topological, same shape as
# SubViewport's producer-before-consumer ordering).
batch = sorted(dirty, key=_depth)
dirty.clear()
for c in batch:
if c._tree is self and c._layout_dirty:
c._do_update_layout()
dirty.clear()
# Containers moved; descendants cache global rects by walking ancestors,
# so bump this tree's epoch to expire those caches (mirrors the direct
# ``_rect_frame`` reset a node does on its own move).
self._layout_frame += 1
[docs]
def tick(self, dt: float):
"""Run process callbacks and coroutines on all nodes for one frame.
Order: singletons' ``_process`` first, in registration order (global
nodes, before the scene), then ``self.events.flush_deferred()`` so
deferred events queued during the previous frame (process, physics, or
input) reach all subscribers before scene logic runs, then the scene
root's ``_process``. Anything ``emit_deferred``'d during this frame's
singleton pass is also drained in the same flush, keeping the scene's
view of the world consistent.
Singletons and the current scene are siblings under the tree's own top
node, and the scene is always its last child, so every node the tree
ticks is reached through exactly one containment path and processes
exactly once. A node registered as a singleton *and* parented into the
scene is a scene node that happens to have a name: it processes in its
scene position, not ahead of the scene.
"""
with self.activate_input():
from .assets import AssetServer
from .ui.core import Control
# A per-second binding read this frame integrates over this frame.
self._own_input._set_step_dt(dt)
# Per-tick rect-cache auto-expire (catches ancestor moves; direct
# moves self-invalidate via _invalidate_transform). Per-tree so a
# sibling tree's tick doesn't churn this tree's caches. The global
# Control._current_frame is kept as the detached-control fallback.
self._layout_frame += 1
Control._current_frame += 1
self._tick_count += 1
# Hover, once, from wherever the frame's motion left the pointer,
# and before any node code reads ``mouse_over``. Rects settled at
# the end of the previous tick, so this reads laid-out geometry.
self._ui.flush_hover(self.root)
# Each host checks whether input still reaches its guest, and hands
# over what stopped; a tree hosting nothing looks at nothing.
if self._hosts:
for host in tuple(self._hosts):
host._audit()
# A held pad direction has no auto-repeat of its own, so the UI
# generates one here, off the monotonic clock: a pause menu is
# exactly where a direction is held and the scene clock has stopped.
self._ui.flush_nav_repeat(self.root)
# A press whose owner has since hidden, been disabled or paused is
# cancelled here, once a frame; an empty table costs one test.
if self._ui._owners:
self._ui.audit_owners()
# An inert overlay pauses the tree (derived, so closing it auto-resumes).
# ``any_inert`` short-circuits False on the empty registry (zero-cost).
paused = self.paused or self.overlays.any_inert()
if not paused:
self._now += dt
# Asset-loader completions arrive on worker threads; drain them
# onto the main thread before any node code runs so handlers see
# a stable world.
if AssetServer._instance is not None:
AssetServer._instance.flush()
scene = self._scene_root
for child in self._tree_root.children.safe_iter():
# ``safe_iter`` snapshots before the loop, so a hook that
# removes a later sibling leaves it here with its tree already
# cleared; a node that has left is not processed.
if child is not scene and child._tree is self:
child._process_recursive(dt, paused)
if self._scene_timers and not paused:
self._scene_timers = [t for t in self._scene_timers if not t._tick(dt)]
self._events.flush_deferred()
# Re-read: a singleton's on_update or a deferred handler may have
# swapped the scene, and it is the new one that processes this frame.
scene = self._scene_root
if scene is not None and scene._tree is self:
with self._scene_span:
scene._process_recursive(dt, paused)
# A scene that ended during its walk is carried out here, and
# its successor starts on the next frame.
if self._next_scene is not None:
self._mount_next_scene()
self._flush_deferred_calls()
self._flush_deletes()
self._flush_coalesced_hooks()
# Reflow any container dirtied during this tick (mark_layout_dirty,
# size on_change, add/remove child) so draw + next-frame hit-testing
# see settled positions. O(1) when nothing is dirty.
self.flush_layout()
# Audio backend reconciliation. Native MiniaudioBackend uses this
# to detect AudioBus volume / effect-chain changes (Property
# mutations on AudioBus aren't on_change-instrumented: diffing
# snapshot is the canonical sync path). Legacy + Null backends
# are no-ops; web reads bus state during its drain. One-line
# call so live UI changes (sliders, toggles) take effect next
# frame on every backend.
backend = getattr(self, "_audio_backend", None)
if backend is not None:
from .audio_bus import AudioBusLayout
from .audio_errors import AudioError, raise_or_warn
try:
backend.sync_bus_layout(AudioBusLayout.get_default())
except AudioError as exc:
# Per-frame call: let the audio system's strict/warn-once
# policy decide whether to raise or log. Other exceptions
# propagate (genuine engine bugs shouldn't be hidden by
# the per-frame catch-all).
raise_or_warn(
exc,
key="audio.scene_tree.sync_bus_layout_failed",
message="audio backend sync_bus_layout failed",
)
def _flush_coalesced_hooks(self) -> None:
"""Drain pending ``on_change`` callbacks queued by ``coalesce=True`` Properties.
Iterates a snapshot so hooks that themselves write to ``coalesce`` properties
enqueue onto the next frame (the natural debounce). Missing methods are
ignored: the owning node may have been removed mid-frame.
"""
if not self._pending_coalesced_hooks:
return
pending = self._pending_coalesced_hooks
self._pending_coalesced_hooks = set()
for obj, method_name in pending:
method = getattr(obj, method_name, None)
if method is not None:
method()
# -- Physics world ownership ------------------------------------------
[docs]
@property
def physics_world(self) -> PhysicsWorld:
"""The tree's default PhysicsWorld, created lazily on first access.
Bodies with no PhysicsRoot ancestor resolve here. Lazy so non-physics
scenes never allocate a world. Backend follows the selection precedence
(no node override -> project ``physics_backend`` setting > auto-discovered
native > Builtin), gravity -Y. Tree-scoped: it persists across
``change_scene`` (like ``events``) and is released only by ``quit()`` / GC.
"""
if self._physics_world is None:
from .math import Vec3
from .physics.backends import resolve_world_factory
self._physics_world = resolve_world_factory()(Vec3(0.0, -9.81, 0.0))
return self._physics_world
[docs]
@property
def has_physics_world(self) -> bool:
"""True iff a default world has been lazily created (no allocation)."""
return self._physics_world is not None
[docs]
@property
def physics_world_2d(self) -> Physics2DWorld:
"""The tree's default 2D PhysicsWorld, created lazily on first access.
The 2D sibling of :attr:`physics_world`. 2D bodies with no ``PhysicsRoot2D``
ancestor resolve here. Lazy so non-2D-physics scenes never allocate it.
Backend follows the selection precedence (no node override -> project
``physics_backend`` setting > auto-discovered native > Builtin), gravity
``Vec2(0, -9.81)`` (Y-up). Tree-scoped: persists across ``change_scene``,
released by ``quit()`` / GC.
"""
if self._physics_world_2d is None:
from .math import Vec2
from .physics.backends import resolve_world_factory_2d
self._physics_world_2d = resolve_world_factory_2d()(Vec2(0.0, -9.81))
return self._physics_world_2d
@staticmethod
def _world_dims(world: Any) -> tuple[int, int]:
"""Return ``(width, dims)`` for a physics world: ``(4, 2)`` 2D, ``(7, 3)`` 3D.
Selects the bulk-array width + scatter/interpolate branch from the world's
type. 2D worlds subclass ``Physics2DWorld`` (``(N,4)`` transform rows);
everything else is a 3D world (``(N,7)``). Imported lazily to keep
``simvx.core.scene_tree`` import-cheap for non-physics scenes.
"""
from .physics.world2d import Physics2DWorld
if isinstance(world, Physics2DWorld):
return (4, 2)
return (7, 3)
[docs]
def register_physics_world(self, world: PhysicsWorld) -> None:
"""Register a PhysicsRoot's isolated world so physics_tick steps it.
Idempotent: a world already present is not added twice (defends against
re-entrant enter_tree). Called from PhysicsRoot.on_enter_tree.
"""
if world not in self._physics_worlds:
self._physics_worlds.append(world)
[docs]
def unregister_physics_world(self, world: PhysicsWorld) -> None:
"""Unregister a PhysicsRoot's world (PhysicsRoot.on_exit_tree). No-op if absent.
The world will not be stepped by this tree again, so any body still in
its retirement grace period is released here rather than left holding a
node alive (see :meth:`retire_physics_node`).
"""
try:
self._physics_worlds.remove(world)
except ValueError:
pass
self._physics_retired.pop(world, None)
self._physics_step_seq.pop(world, None)
# -- Physics body sync (the world layer below is node-agnostic) -------
[docs]
def register_physics_body(self, world: PhysicsWorld, handle: BodyHandle, node: Node) -> None:
"""Register a body node for post-step transform read-back.
Called by ``PhysicsBody3D.on_enter_tree`` for non-static bodies. Marks
the world's membership ``dirty`` so the next sync rebuilds the bulk
order + buffer. Weak reference: a GC'd node auto-drops.
"""
sync = self._physics_bodies.get(world)
if sync is None:
width, dims = self._world_dims(world)
sync = self._physics_bodies[world] = _BodySync(width=width, dims=dims)
sync.nodes[handle] = node
sync.dirty = True
[docs]
def unregister_physics_body(self, world: PhysicsWorld, handle: BodyHandle) -> None:
"""Unregister a body node (``PhysicsBody3D.on_exit_tree``). No-op if absent.
Drops the whole per-world ``_BodySync`` once empty so idle worlds cost
nothing.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
sync.nodes.pop(handle, None)
sync.dirty = True
if not sync.nodes:
del self._physics_bodies[world]
[docs]
def register_physics_node(self, world: PhysicsWorld, handle: BodyHandle, node: Node) -> None:
"""Register a body node in the collision-event handle->node map.
Called by ``PhysicsBody3D.on_enter_tree`` for ALL modes (static
included), so :meth:`_dispatch_contact_events` can resolve both sides of
a contact pair. Mode-independent: unlike ``register_physics_body`` this
map never changes on a mode flip. Weak reference: a GC'd node auto-drops.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
nodes = self._physics_nodes[world] = weakref.WeakValueDictionary()
nodes[handle] = node
[docs]
def retire_physics_node(self, world: AnyPhysicsWorld, handle: BodyHandle) -> None:
"""Move a body node out of the live map and into the retirement grace period.
Called by ``PhysicsObject3D.on_exit_tree`` immediately before the backend
body is destroyed. Destroying a body ENDS every contact and sensor
overlap it was in, and the seam reports those closing ``EXIT``s on the
first drain after the NEXT step (see
:meth:`~simvx.core.physics.world.PhysicsWorld.destroy_body`). Dropping
the handle->node entry here would leave that dispatch unable to name the
peer that went away, so the entry is retired instead: held, with a STRONG
reference to the node (the live map is weak, and the game has usually let
go of a node it destroyed), until the events it is owed have been
dispatched.
The grace period is one step, measured on this tree's per-world step
count, and :meth:`_release_retired_physics_nodes` ends it. Retirement is
never consulted before the live map, so a backend that recycles handles
resolves a reused handle to its NEW body.
One STEP, not one frame: a tree that stops stepping keeps the entry until
it steps again, so pausing (or clearing ``auto_physics``) immediately
after a kill holds that one node for the duration of the pause. It is
released on the next step regardless, and by ``change_scene`` and by
unregistering the world, so it cannot outlive the scene either way.
No-op if the handle is not registered.
"""
nodes = self._physics_nodes.get(world)
if nodes is None:
return
node = nodes.pop(handle, None)
if not nodes:
del self._physics_nodes[world]
if node is None:
return
retired = self._physics_retired.get(world)
if retired is None:
retired = self._physics_retired[world] = {}
retired[handle] = (node, self._physics_step_seq.get(world, 0))
def _resolve_physics_node(self, world: AnyPhysicsWorld, handle: BodyHandle) -> Node | None:
"""The node behind ``handle``: live first, then retired, else ``None``.
Live-first is what keeps a recycled handle honest (see
:meth:`retire_physics_node`).
"""
nodes = self._physics_nodes.get(world)
node = nodes.get(handle) if nodes is not None else None
if node is not None:
return node
retired = self._physics_retired.get(world)
if retired is None:
return None
entry = retired.get(handle)
return entry[0] if entry is not None else None
def _release_retired_physics_nodes(self, world: AnyPhysicsWorld) -> None:
"""End the grace period for bodies retired before this world's last step.
A body retired DURING the dispatch that follows a step (a handler calling
``destroy`` on the peer it just touched) carries the current sequence
number and so survives this pass; its own closing events arrive after the
next step. Anything older has had its dispatch and is dropped.
"""
retired = self._physics_retired.get(world)
if not retired:
return
seq = self._physics_step_seq.get(world, 0)
for handle in [h for h, (_node, at) in retired.items() if at < seq]:
del retired[handle]
if not retired:
del self._physics_retired[world]
def _dispatch_contact_events(self, world: AnyPhysicsWorld) -> None:
"""Drain ``world``'s buffered contact events and fire node Signals.
Called once per stepped world, AFTER :meth:`PhysicsWorld.step` and
:meth:`_sync_physics_world`, so node transforms are already current and
a Signal handler may safely mutate the tree (spawn/destroy): any change
takes effect on the next step, and a handler-triggered destroy this turn
is safe because dispatch runs outside the broad/narrow loops.
The world layer knows nothing about nodes: its events are keyed by body
handles. This resolves both handles via :meth:`_resolve_physics_node` and
builds one node-level :class:`Contact` per side, with ``other`` set to the
peer and ``normal`` / ``velocity`` reoriented for that side (the world
reports the normal as ``a -> b``; the ``b`` side is negated).
A peer DESTROYED mid-touch is still named: it is resolved out of the
retirement grace period (:meth:`retire_physics_node`), so ``separated``
fires with a ``Contact`` whose ``other`` is the real node, detached
(``other.tree`` and ``other.handle`` are ``None``). An event is skipped
only when a side has no node at all, which now means garbage-collected
without ever having been in this tree.
"""
if world not in self._physics_nodes and world not in self._physics_retired:
return
events = world.drain_contact_events()
if not events:
return
# Width-aware: build the 2D node-level Contact for a 2D world, else the 3D
# one. ContactPhase is shared (dimension-agnostic). The world-level
# ContactEvent field layout is identical across dims, so only the payload
# type differs.
from .physics.world import ContactPhase
if self._world_dims(world)[1] == 2:
from .physics.nodes2d import Contact2D as Contact
else:
from .physics.nodes import Contact
for ev in events:
a = self._resolve_physics_node(world, ev.a)
b = self._resolve_physics_node(world, ev.b)
if a is None or b is None:
continue # never registered, or GC'd: nothing to notify
# Seam normal is a->b and rel_velocity is b w.r.t. a. Each node sees a
# normal pointing TOWARD itself (its separating direction) and the
# peer's velocity w.r.t. itself: that is -normal/+rel_vel for the ``a``
# side and +normal/-rel_vel for the ``b`` side.
if ev.phase is ContactPhase.ENTER:
a.collided.emit(
Contact(
other=b,
point=ev.point,
normal=-ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
b.collided.emit(
Contact(
other=a,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=-ev.rel_velocity,
)
)
else:
# EXIT: degenerate payload (no live manifold), only ``other`` is
# meaningful. ``impulse`` is forwarded rather than zeroed: a
# backend that cannot measure reports None, and hardcoding 0.0
# here would relabel that as "measured, and it was zero". The
# estimate needs no such care: it is a number on every backend,
# and the seam already reports 0.0 for an exit.
a.separated.emit(
Contact(
other=b,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
b.separated.emit(
Contact(
other=a,
point=ev.point,
normal=ev.normal,
impulse=ev.impulse,
impulse_estimate=ev.impulse_estimate,
velocity=ev.rel_velocity,
)
)
def _dispatch_overlap_events(self, world: AnyPhysicsWorld) -> None:
"""Drain ``world``'s buffered sensor-overlap events and fire Area Signals.
The SEPARATE sensor stream, parallel to :meth:`_dispatch_contact_events`
but DIRECTED ``sensor -> other``: dispatch fires ONLY on the detecting
sensor's :class:`Area3D` node, never the reverse, so sensor-vs-sensor
produces up to two independent directed events (one per observer) with no
double-firing logic here. Reuses the same mode-independent
:attr:`_physics_nodes` handle->node map (Area sensor bodies register via
``register_physics_node`` too).
Called once per stepped world AFTER :meth:`_dispatch_contact_events`, so a
Signal handler may safely mutate the tree (deferred: effect next step).
Node-agnostic: routes ``body_entered`` / ``body_exited`` vs
``area_entered`` / ``area_exited`` by the OTHER node's type. The Area's
live overlap sets are maintained HERE (ENTER add / EXIT discard) so its
polling accessors stay consistent with the last dispatched step.
A peer destroyed mid-overlap is named out of the retirement grace period
exactly as on the contact stream, so ``body_exited`` / ``area_exited``
fires with the detached node AND the Area's overlap set loses it.
"""
if world not in self._physics_nodes and world not in self._physics_retired:
return
events = world.drain_overlap_events()
if not events:
return
# Width-aware: resolve the Area / physics-object node types for this
# world's dim so the routing (body_* vs area_* by the OTHER node's type) is
# dimension-correct. The non-area branch tests the shared physics-object
# base, so a character routes to body_entered / body_exited like any other
# body without a second isinstance.
from .physics.world import ContactPhase
if self._world_dims(world)[1] == 2:
from .physics.nodes2d import Area2D as Area
from .physics.nodes2d import PhysicsObject2D as Body
else:
from .physics.nodes import Area3D as Area
from .physics.nodes import PhysicsObject3D as Body
for ev in events:
# The detecting sensor MUST be an Area node.
area = self._resolve_physics_node(world, ev.sensor)
other = self._resolve_physics_node(world, ev.other)
if area is None or other is None:
continue # never registered, or GC'd
if not isinstance(area, Area):
continue # defensive: only Area nodes own sensors
if ev.phase is ContactPhase.ENTER:
if isinstance(other, Area):
area._overlapping_areas.add(other)
area.area_entered.emit(other)
elif isinstance(other, Body):
area._overlapping_bodies.add(other)
area.body_entered.emit(other)
else: # EXIT
if isinstance(other, Area):
area._overlapping_areas.discard(other)
area.area_exited.emit(other)
elif isinstance(other, Body):
area._overlapping_bodies.discard(other)
area.body_exited.emit(other)
def _sync_physics_world(self, world: AnyPhysicsWorld) -> None:
"""Scatter ``world``'s simulated transforms back onto its body nodes.
Called once per stepped world, after :meth:`PhysicsWorld.step`. Uses the
bulk contract only: ``register_bodies`` (re-fixed lazily on membership
change) + ``read_transforms`` into a reused preallocated buffer. One bulk
transfer per world per frame, no hot-path allocation.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
if sync.dirty:
sync.order = list(sync.nodes.keys())
world.register_bodies(sync.order)
sync.buf = np.empty((len(sync.order), sync.width), np.float32)
sync.dirty = False
if not sync.order:
return
world.read_transforms(sync.buf)
buf = sync.buf
if sync.dims == 2:
# 2D row is [px, py, cos, sin]; rotation is a scalar = atan2(sin, cos).
from .math import Vec2
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None: # GC'd between unregister and sync
continue
row = buf[i]
# _write_simulated_pose guards the assignment so the pose-reconcile
# hook does not bounce this simulated pose back into the handle.
node._write_simulated_pose(Vec2(row[0], row[1]), math.atan2(float(row[3]), float(row[2])))
return
from .math import Quat, Vec3
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None: # GC'd between unregister and sync
continue
row = buf[i]
# read_transforms emits xyzw (scalar-last); engine Quat is (w,x,y,z).
node._write_simulated_pose(Vec3(row[0], row[1], row[2]), Quat(row[6], row[3], row[4], row[5]))
def _capture_physics_world(self, world: AnyPhysicsWorld) -> None:
"""Capture this step's transforms into the per-world prev/cur ring.
Rolls cur -> prev, then reads the post-step pose into cur. Writes nothing
to nodes: render-time interpolation (interpolate_physics) does the scatter.
A membership change (dirty) rebuilds order/buffers and re-seeds prev == cur
so a newly added body never lerps from origin on its first rendered frame.
"""
sync = self._physics_bodies.get(world)
if sync is None:
return
if sync.dirty:
sync.order = list(sync.nodes.keys())
world.register_bodies(sync.order)
n = len(sync.order)
sync.buf = np.empty((n, sync.width), np.float32)
sync.prev_buf = np.empty((n, sync.width), np.float32)
sync.cur_buf = np.empty((n, sync.width), np.float32)
sync.dirty = False
sync.seeded = False
if not sync.order:
return
world.read_transforms(sync.buf)
if not sync.seeded:
sync.prev_buf[:] = sync.buf # seed BOTH: first frame holds steady, no origin lerp
sync.cur_buf[:] = sync.buf
sync.seeded = True
else:
sync.prev_buf, sync.cur_buf = sync.cur_buf, sync.prev_buf # roll cur -> prev (swap, no alloc)
sync.cur_buf[:] = sync.buf # new state into cur
[docs]
def interpolate_physics(self, alpha: float) -> None:
"""Write the render pose of every captured new-world body: lerp(prev, cur, alpha).
``alpha`` is the fixed-step fraction ``physics_accum / physics_dt`` in [0, 1).
Position lerps linearly; rotation slerps (shortest-arc, Quat.slerp). A
sleeping body has prev == cur, so it lerps to a constant (correct, cheap).
No-op when ``physics_interpolation`` is False (driver still calls it; the raw
write already happened in _sync_physics_world during physics_tick).
"""
if not self.physics_interpolation:
return
from .math import Quat, Vec2, Vec3
a = 0.0 if alpha < 0.0 else 1.0 if alpha > 1.0 else alpha
for sync in self._physics_bodies.values():
if not sync.seeded or not sync.order:
continue
prev, cur = sync.prev_buf, sync.cur_buf
if sync.dims == 2:
# 2D: lerp position xy + lerp the cos/sin unit vector, then atan2
# (no +-pi wraparound: the load-bearing reason transforms carry
# cos/sin rather than the bare angle).
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None:
continue
p, c = prev[i], cur[i]
cos = float(p[2]) + (float(c[2]) - float(p[2])) * a
sin = float(p[3]) + (float(c[3]) - float(p[3])) * a
# Guarded write: this is a render-only interpolated pose; it
# must NOT be pushed back into the handle (would corrupt the
# next step's start pose).
node._write_simulated_pose(Vec2(p[0], p[1]).lerp(Vec2(c[0], c[1]), a), math.atan2(sin, cos))
continue
for i, handle in enumerate(sync.order):
node = sync.nodes.get(handle)
if node is None:
continue
p, c = prev[i], cur[i]
# read_transforms emits xyzw (scalar-last); engine Quat is (w,x,y,z).
qp = Quat(p[6], p[3], p[4], p[5])
qc = Quat(c[6], c[3], c[4], c[5])
node._write_simulated_pose(Vec3(p[0], p[1], p[2]).lerp(Vec3(c[0], c[1], c[2]), a), qp.slerp(qc, a))
def _drop_physics_worlds(self) -> None:
"""Release the default world and clear the PhysicsRoot + body registries.
Registered worlds are normally removed by each PhysicsRoot.on_exit_tree;
this is the belt-and-braces sweep so a torn-down tree leaks nothing even
if a root was force-detached. Safe to call repeatedly.
"""
self._physics_worlds.clear()
self._physics_world = None
self._physics_world_2d = None
self._physics_bodies.clear()
self._physics_nodes.clear()
self._physics_retired.clear()
self._physics_step_seq.clear()
[docs]
def physics_tick(self, dt: float, *, tick_input: TickInput | None = None):
"""Run physics_process callbacks on all nodes, then auto-step physics.
The node walk runs against one :class:`~.input.tick.TickInput`, which
is what ``Input`` answers from inside ``on_fixed_update``: held state
as it was when the tick started, and the frame's edges on the frame's
first tick and no other. Pass *tick_input* to drive a tick from a
recorded or hand-built snapshot instead of from the live record; the
physics step below is outside the span, since the solver asks nothing
of input.
"""
# An inert overlay freezes the whole world, physics included (parity with
# the process pass in ``tick``); ``any_inert`` short-circuits on the empty
# registry, so this is zero-cost when no inert overlay is open.
paused = self.paused or self.overlays.any_inert()
with self.activate_input():
inp = self._own_input
outer_dt = inp._step_dt
inp._set_step_dt(dt)
try:
with inp._fixed_tick(tick_input):
scene = self._scene_root
for child in self._tree_root.children.safe_iter():
if child is not scene and child._tree is self:
child._physics_process_recursive(dt, paused)
if scene is not None and scene._tree is self:
with self._scene_span:
scene._physics_process_recursive(dt, paused)
finally:
# A per-second binding integrates over the step it is read in.
inp._set_step_dt(outer_dt)
if self._next_scene is not None:
# The scene ended during its fixed walk, so this step was its
# and is not taken: the new scene's bodies start on its own.
self._mount_next_scene()
return
if self.auto_physics:
# Step the default world (only if it was actually created) plus
# every registered PhysicsRoot world, once each, with this tick's
# dt. Skip empty worlds for zero overhead.
#
# Frozen while paused: a paused tree must not advance the sim or
# dispatch contact/overlap events. Render interpolation still runs
# (driver-side) so a paused world holds its last pose steady. The
# capture-vs-immediate-write split honours physics_interpolation:
# ON captures prev/cur for interpolate_physics; OFF writes nodes now.
if not paused:
# Contact and overlap handlers are the scene's code too.
with self._scene_span:
if self._physics_world is not None:
self._step_physics_world(self._physics_world, dt)
# Default 2D world: same pause guard, same sync/dispatch
# path (the leaves are width-aware), parallel to the 3D default.
if self._physics_world_2d is not None:
self._step_physics_world(self._physics_world_2d, dt)
for world in self._physics_worlds:
self._step_physics_world(world, dt)
if self._next_scene is not None:
self._mount_next_scene()
def _step_physics_world(self, world: AnyPhysicsWorld, dt: float) -> None:
"""Advance one world by ``dt``, scatter its poses and dispatch its events.
Also ends the retirement grace period of bodies destroyed before this
step, which is why it runs even for an empty world: a scene that destroys
its LAST body would otherwise hold that node alive forever, and with no
body left there is no open edge for the seam still to close.
"""
if world.body_count == 0:
self._physics_retired.pop(world, None)
return
self._physics_step_seq[world] = self._physics_step_seq.get(world, 0) + 1
world.step(dt)
if self.physics_interpolation:
self._capture_physics_world(world)
else:
self._sync_physics_world(world)
self._dispatch_contact_events(world)
self._dispatch_overlap_events(world)
self._release_retired_physics_nodes(world)
def _broadcast_repeat(self, event: KeyEvent) -> Node | None:
"""Offer a held key's auto-repeat to the ordinary-phase ``repeat=True`` key handlers; return who claimed it.
The router calls this for a repeat no control claimed, and only in a
tree where such a handler registered. A repeat resolves to no action
and reaches no event family or catch-all here: those hear one press
per physical press. Deepest first, like the ordinary phase, and a
claim stops the rest.
"""
bucket = self._bubble_tables.key_repeat.get(event.key)
if not bucket:
return None
# A repeat of a press some handler claimed belongs to that node: only
# its own ``repeat=True`` handlers hear it. A press nobody claimed
# leaves its repeats to every handler that asked.
owner = self._repeat_keys.get(event.key)
if owner is not None and owner._tree is not self:
owner = None
sequence = [(entry, event) for entry in self._ordered(bucket) if owner is None or entry[1] is owner]
if not sequence:
return None
claimant: Node | None = None
with self.activate_input():
inp = self.input
outer = inp._event_modifiers
inp._event_modifiers = frozenset(modifiers_held(event))
try:
with self._scene_span:
claimant = self._deliver_sequence(event, sequence)
finally:
inp._event_modifiers = outer
if self._next_scene is not None:
self._mount_next_scene()
return claimant
def _deliver_samples(self, event: PointerEvent) -> Node | None:
"""Offer one motion sample to the ``samples=True`` handlers, and return who claimed it.
The router calls this for each sample as it arrives, ahead of the
frame's merged move, which every other handler hears. A sample reaches
only these handlers, deepest first like the ordinary phase, and a claim
stops the rest of them; it writes no state and reaches no control.
"""
if not self._has_sample_handlers:
return None
sequence = [(entry, event) for entry in self._ordered(self._sample_handlers)]
if not sequence:
return None
claimant: Node | None = None
with self.activate_input():
inp = self.input
outer = inp._event_modifiers
inp._event_modifiers = frozenset(modifiers_held(event))
try:
with self._scene_span:
claimant = self._deliver_sequence(event, sequence)
finally:
inp._event_modifiers = outer
return claimant
def _broadcast_input(self, event: InputEvent, *, handlers: bool = True) -> Node | None:
"""Resolve *event* to its actions and deliver it to every matching handler.
Phases three and four of :meth:`InputRouter.publish
<simvx.core.input.router.InputRouter.publish>`, which is the only door
input comes through: nothing else calls this. Resolution reads the
active bindings, and delivery visits every handler the event matches in
one reverse pre-order sequence, the action handlers and the device
handlers for the same press among them.
Lookup is indexed by the input, so a node with no handler for this
event costs nothing and the traversal is O(handlers per event) rather
than O(nodes). The first handler to claim the event ends the sequence;
the claim changes no device state, so a poll still answers honestly.
Dispatch runs inside this tree's input span, so a handler querying
``Input`` or registering on ``InputMap`` sees the tree the event was
delivered to and not whichever pair is active for the caller draining
the event queue.
With *handlers* False the event resolves and records its action edges
and reaches no handler: a claim taken before this phase that must still
let the action layer close what its press opened.
"""
if SceneTree.verify_input_order:
for bucket in self._input_buckets:
self._check_bucket(bucket)
claimant: Node | None = None
with self.activate_input():
with self._scene_span:
claimant = self._dispatch_input(event, handlers)
if self._next_scene is not None:
# A handler ended the scene: the event went with it.
self._mount_next_scene()
return claimant
def _matching_buckets(
self, event: InputEvent, action_edges: list[tuple[str, bool, int]]
) -> list[tuple[tuple[tuple, ...], InputEvent]]:
"""The ordered handler snapshots *event* reaches, each with the event it delivers.
*action_edges* is what the event resolved to, passed in rather than
resolved here because the same resolution is what a physics tick is
given: one event opens and closes actions once.
**An action handler receives an** :class:`~.input.events.ActionEvent`,
not the device event: it asked about an action, and the action is what
carries the value, the player it belongs to and the device event as its
``source``. A device handler receives the device event itself, and the
two travel in one sequence for one press, so a claim on either stops
the rest.
"""
found: list[tuple[tuple[tuple, ...], InputEvent]] = []
for action_name, action_pressed, player in action_edges:
bucket = self._action_handlers.get((action_name, not action_pressed))
if bucket:
found.append((self._ordered(bucket), self._action_event(event, action_name, action_pressed, player)))
found.extend(self._device_buckets(event, self._bubble_tables))
return found
def _device_buckets(self, event: InputEvent, tables: _HandlerTables) -> list[tuple[tuple[tuple, ...], InputEvent]]:
"""The ordered handler snapshots a device event reaches in one set of *tables*: bubble or capture."""
found: list[tuple[tuple[tuple, ...], InputEvent]] = []
# The event's class selects the table, so a device family is one branch
# and the tables stay indexed by input.
if isinstance(event, KeyEvent):
# An auto-repeat reaches the key targets that asked for it, and the
# families below, which hear every event their phase delivers.
bucket = tables.key_repeat.get(event.key) if event.echo else tables.key.get((event.key, not event.pressed))
if bucket:
found.append((self._ordered(bucket), event))
elif isinstance(event, PointerEvent):
phase = event.phase
# A filter naming a button and one naming only a phase are separate
# routing keys, so a press reaches both a ``Pointer.PRIMARY``
# handler and the any-button one. A press is looked up under the
# control, its index form and its roles, so a filter on any of them
# hears it alike; a move names no button and has only the one key.
handlers = tables.pointer
for name in self.input._button_forms(event):
bucket = handlers.get((name, phase))
if bucket:
found.append((self._ordered(bucket), _naming(event, name)))
bucket = handlers.get((None, phase))
if bucket:
found.append((self._ordered(bucket), event))
elif isinstance(event, ButtonEvent):
# A pad's or a tablet's button, under the control, its index form
# and its unit's own spelling of each.
released = not event.pressed
for name in self.input._button_forms(event):
bucket = tables.device_button.get((name, released))
if bucket:
found.append((self._ordered(bucket), _naming(event, name)))
elif isinstance(event, AxisEvent):
for name in self.input._axis_forms(event):
bucket = tables.device_axis.get(name)
if bucket:
found.append((self._ordered(bucket), _naming(event, name)))
# An `@on_input(KeyEvent)`-style filter: the whole family, no target.
bucket = tables.event_class.get(type(event))
if bucket:
found.append((self._ordered(bucket), event))
if tables.catch_all:
found.append((self._ordered(tables.catch_all), event))
return found
def _deliver_action_release(self, action: str, player: int) -> None:
"""Close *action* for *player* with a real edge, because a context change took it.
An action that goes dark has to be released rather than simply
forgotten: a game that stops sprinting on the release would otherwise
sprint for ever after a pause menu took the action away. The edge
reaches the ``@on_input("jump", edge="release")`` handlers and
nothing else; there is no device event under it, because no device
moved.
Deferred while a dispatch or a subtree attach is in flight, so a
context arriving with a scene neither nests inside the event being
delivered nor runs a handler against a half-attached tree.
"""
held = self.input._keys_delivered
event = ActionEvent(
action=action,
pressed=False,
value=False,
player=player,
# No device moved, so the modifiers are read off what is held: a
# handler filtered on a chord is still asking about now.
ctrl=key_code(Key.LEFT_CONTROL) in held or key_code(Key.RIGHT_CONTROL) in held,
shift=key_code(Key.LEFT_SHIFT) in held or key_code(Key.RIGHT_SHIFT) in held,
alt=key_code(Key.LEFT_ALT) in held or key_code(Key.RIGHT_ALT) in held,
meta=key_code(Key.LEFT_SUPER) in held or key_code(Key.RIGHT_SUPER) in held,
)
if self._input_dispatching or self._attaching is not None or self._input_edge_hold:
self._deferred_action_edges.append(event)
return
self._dispatch_action_edge(event)
def _flush_action_edges(self) -> None:
"""Deliver the release edges a dispatch, an attach or a teardown held back."""
if self._input_dispatching or self._attaching is not None or self._input_edge_hold:
return
while self._deferred_action_edges:
self._dispatch_action_edge(self._deferred_action_edges.popleft())
def _dispatch_action_edge(self, event: ActionEvent) -> None:
"""Deliver one action edge that no device event produced."""
bucket = self._action_handlers.get((event.action, True))
if not bucket:
return
# Neither the hide floor nor the processing gate applies here. This
# edge closes something that was live before the change, and the
# handler that has to hear it is by definition the one the change just
# hid or paused: a sprint that stops on the release would otherwise
# run for ever behind a pause menu.
with self.activate_input():
for _order, node, method_name, _chord, consume in self._ordered(bucket):
if not node._hears_player(event.player):
continue
scoped = node._enter_player() if _node_module._PLAYER_SCOPES else None
try:
decided = node._safe_decide(method_name, event)
finally:
if scoped is not None:
node._leave_player(scoped)
if decided is True or (decided is None and consume):
return
def _action_event(self, source: InputEvent, action: str, pressed: bool, player: int) -> ActionEvent:
"""The action edge an ``action=`` handler receives, over the device event.
It carries what the action reads at this edge, resolved for the player
it belongs to, so a handler on a movement action is handed the vector
rather than being told a key went down. The modifiers and the time come
from the device event, because they are facts about the same moment.
"""
with self.input._as_player(player):
value = self.input.value(action)
return ActionEvent(
action=action,
pressed=pressed,
value=value,
player=player,
context=self.input_map._origin.get(action),
source=source,
time=source.time,
ctrl=source.ctrl,
shift=source.shift,
alt=source.alt,
meta=source.meta,
)
def _dispatch_input(self, event: InputEvent, handlers: bool) -> Node | None:
"""Handler half of :meth:`_broadcast_input`, with the event's modifiers as the ones held.
The platform reports modifier state on every event, and that report
is the truth for the event's own resolution: a Ctrl held since before
the window had focus was never delivered, and Ctrl+S must still be
Ctrl+S. Everything read while the event resolves, a binding's held
modifiers and an ``Input.pressed`` inside a handler alike, reads it.
"""
inp = self.input
outer = inp._event_modifiers
inp._event_modifiers = frozenset(modifiers_held(event))
try:
return self._dispatch_input_under(event, handlers)
finally:
inp._event_modifiers = outer
def _dispatch_input_under(self, event: InputEvent, handlers: bool) -> Node | None:
"""Handler half of :meth:`_broadcast_input`, run inside the span.
Every handler goes through :meth:`Node._safe_decide`, so a decorated
handler answers to ``Node.strict_errors`` exactly as every other hook
does, a node already disabled by a script error receives nothing
further, and the UI's ``on_gui_input`` claims by the same rule.
The action edges the event resolved to are recorded on the way past,
for the next physics tick to read. They are recorded here rather than
on the state write, because an event a control claimed never reaches
this phase and opens no action: what a tick is told is exactly what
the handlers were told.
"""
action_edges = self._actions_for_event(event)
if action_edges:
state = self.input
for action_name, action_pressed, player in action_edges:
state._record_edge(action_name, action_pressed, player, event.time)
if not handlers:
return None
found = self._matching_buckets(event, action_edges)
if not found:
return None
if len(found) == 1:
bucket, delivered = found[0]
sequence: Iterable[tuple[tuple, InputEvent]] = ((entry, delivered) for entry in bucket)
else:
# Each bucket is already in reverse pre-order, so the merge is a
# sort of the concatenation on the key each entry carries. The
# event travels with its entry, because the action handlers and the
# device handlers for one press share the sequence and receive
# different objects.
sequence = sorted(
((entry, delivered) for bucket, delivered in found for entry in bucket),
key=lambda pair: _entry_order(pair[0]),
reverse=True,
)
return self._deliver_sequence(event, sequence)
def _deliver_sequence(self, event: InputEvent, sequence: Iterable[tuple[tuple, InputEvent]]) -> Node | None:
"""Offer *event* to each handler of *sequence* in turn until one claims it; return the claimant."""
floor = self.input_map._hide_floor
paused = self._paused
held_now = self.input._binding_held
for (_order, node, method_name, chord, consume), delivered in sequence:
# A target that holds controls, Ctrl+S or LB+A, hears the press only
# while they are down, by the rule an action's binding is read by.
if chord is not None and not held_now(chord):
continue
# ``Node._can_receive_input`` inlined, the way the process walk
# inlines ``_can_update``: the switch that stops a node processing
# is the switch that deafens it, and this runs once per handler
# the event reaches. A script error is left to ``_safe_decide``,
# which already answers False for a disabled node.
mode = node._cached_update_mode
if mode is None:
mode = node._effective_update_mode()
if mode == UpdateMode.DISABLED or (
paused if mode == UpdateMode.PAUSABLE else (not paused and mode == UpdateMode.PAUSED_ONLY)
):
# DISABLED ever, PAUSABLE while paused, PAUSED_ONLY while the
# scene is running: a paused player does not hear its own
# keys, and a menu running through the pause does.
continue
if floor and node._input_context_rank() < floor:
# A context hiding everything below it hides the handlers
# there too, not only the action names: the game under a pause
# menu stops hearing its own keys as well as its own actions.
continue
if isinstance(delivered, ActionEvent) and not node._hears_player(delivered.player):
# An action belongs to one player, and a handler inside another
# player's subtree is not listening for it.
continue
# A handler that polls Input reads its own player, the same as its
# node's per-frame code does.
scoped = node._enter_player() if _node_module._PLAYER_SCOPES else None
try:
decided = node._safe_decide(method_name, delivered)
finally:
if scoped is not None:
node._leave_player(scoped)
if decided is True or (decided is None and consume):
# A claim takes the event and the action together: they are one
# sequence for one press, so claiming either stops the rest.
claimant: Node = node
return claimant
return None
def _ordered(self, bucket: _HandlerBucket) -> tuple[tuple, ...]:
"""*bucket*'s handlers in reverse pre-order, re-sorted only when stale.
A broadcast delivers deepest-and-last-added first and the tree's top
node last, and registration order is attach order, which is not tree
order after any reorder or reparent. Sorting here rather than inserting
at registration is what keeps a scene change of a few thousand nodes an
append per handler: the cost lands once per bucket, on the first event
that reaches it after something actually reordered, and a bucket holds
only the handlers registered for one input.
The sort is also where tombstones are swept, since it is already a pass
over every entry: a bucket that has to be re-sorted is compacted for
free, and one that does not was not holding dead entries worth sweeping.
"""
if bucket.order_version != self._input_order_version:
if bucket.pending:
bucket.compact()
bucket.snapshot = self._sorted_handlers(bucket)
bucket.order_version = self._input_order_version
return bucket.snapshot
def _sorted_handlers(self, bucket: _HandlerBucket) -> tuple[tuple, ...]:
"""*bucket*'s entries keyed by tree position and sorted into delivery order."""
top = self._tree_root
keyed = []
for entry in bucket:
order = _input_order_key(entry[0], top)
if order is not None:
keyed.append((order, *entry))
keyed.sort(key=_entry_order, reverse=True)
return tuple(keyed)
def _check_bucket(self, bucket: _HandlerBucket) -> None:
"""Assert everything a bucket is supposed to be true about itself.
Dispatch is correct only while seven pieces of mutable state agree, and
none of them is checkable from what a handler observes. Every defect
this machinery has had was invisible to behaviour: a bucket held five
hundred nodes while ``len(bucket)`` read zero, a sweep threshold was
unsatisfiable while delivery stayed perfectly correct, an order was
silently wrong after a reorder with nothing to compare against. None of
those was a missing test; each needed a weakref, a counter or an
oracle. This is the oracle.
Checked over *every* bucket on every event, not only the ones the event
reaches. That is the expensive half and it is also the necessary half:
an emptied bucket is skipped by ``_matching_buckets`` and re-sorted by
nothing, so the one defect that held five hundred nodes across a scene
change lived exactly where a matched-buckets-only check could not look.
It costs what the caches exist to save, so it is off by default. Turn
it on for a run that is hunting anything in this area, and any ordinary
test then catches the whole class.
"""
dead_entries = sum(1 for entry in bucket if entry[0] in bucket.pending)
living = [entry[0] for entry in bucket if entry[0] not in bucket.pending]
held = {id(node) for node in living}
strays = [entry[1].name for entry in bucket.snapshot if id(entry[1]) not in held]
if strays:
raise AssertionError(
f"the cached snapshot names {len(strays)} node(s) this bucket no longer holds "
f"({strays[:5]}), and a snapshot holds them strongly. An emptied bucket is "
"skipped by _matching_buckets and re-sorted by nothing, so they are kept for the "
"life of the tree: drop() must clear the contents and not only the order."
)
if living and len(bucket) > 2 * len(living):
raise AssertionError(
f"{len(bucket)} entries where {len(living)} are live. The sweep threshold is not "
"firing, so departed nodes pile up until an event happens to reach this bucket, "
"and never if nothing presses this input. It has to compare dead entries against "
"the total, not dead nodes."
)
if bucket.dead != dead_entries:
raise AssertionError(
f"the sweep threshold is counting the wrong thing: dead={bucket.dead} against "
f"{dead_entries} entries actually tombstoned. It has to count entries, because a "
"class and its subclass can both bind one key and put two entries in one bucket."
)
orphans = [node.name for node in bucket.pending if all(entry[0] is not node for entry in bucket)]
if orphans:
raise AssertionError(f"tombstones for nodes with no entry in this bucket: {orphans}")
top = self._tree_root
for node, method_name, _chord, _consume in bucket:
if node in bucket.pending:
continue
if not hasattr(node, method_name):
raise AssertionError(
f"{node.name!r} has a live entry naming {method_name!r}, which its class "
f"{type(node).__name__} does not have: the next event through this bucket "
"raises. Its class changed while it was mounted, without the registration "
"following."
)
if _input_order_key(node, top) is None:
raise AssertionError(
f"{node.name!r} has a live entry here and is not in the tree: it was unlinked "
"without its handlers being retired, so it is held for ever and would receive "
"input again the moment it is re-attached."
)
plan = node.__dict__.get("_input_plan")
if plan is None:
raise AssertionError(
f"{node.name!r} has a live entry here and no registration plan, so nothing will ever retire it."
)
if not any(planned is bucket for planned, *_rest in plan):
raise AssertionError(
f"{node.name!r} has a live entry in a bucket its plan does not name, so "
"leaving the tree will retire it from somewhere else and strand this one. "
"Its class changed while it was mounted."
)
# Ranks decide delivery and the list decides everything else, so the two
# have to agree. Nothing that goes through Children can part them; this
# is here because a reach past it cannot be caught by comparing the
# cache against a recomputation, since both read the ranks.
checked: set[int] = set()
for entry in bucket:
node = entry[0]
while node is not None and node is not top:
parent = node.parent
if parent is None or id(parent) in checked:
break
checked.add(id(parent))
ranks = [child._sibling_rank for child in parent.children]
if ranks != sorted(ranks):
raise AssertionError(
f"{parent.name!r} lists its children in one order and ranks them in "
f"another ({ranks}): delivery follows the ranks, so it no longer matches "
"the tree. Something permuted the list without going through Children."
)
node = parent
if bucket.order_version != self._input_order_version:
# The cache does not claim to be current, so there is no order to
# disagree with: the next event through this bucket rebuilds it.
return
# Last, because a stale order is usually the symptom of one of the above
# and naming the cause is more use than naming the consequence.
fresh = self._sorted_handlers(bucket)
if fresh != bucket.snapshot:
raise AssertionError(
"input dispatch order is stale: the tree's shape changed without "
f"the order counter moving. Cached {[(e[0], e[1].name, e[2]) for e in bucket.snapshot]}, "
f"recomputed {[(e[0], e[1].name, e[2]) for e in fresh]}. Something permuted a child "
"list without going through Children."
)
# -- Input dispatch table maintenance ---------------------------------
def _input_bucket(self, table: dict, key: Any) -> _HandlerBucket:
"""The bucket for one routing key, created on first use."""
bucket = table.get(key)
if bucket is None:
bucket = table[key] = _HandlerBucket()
self._input_buckets.append(bucket)
return bucket
def _registration_plan(self, node_type: type[Node]) -> tuple[tuple, ...]:
"""Where a node of *node_type* registers, resolved once per class.
Which buckets a node lands in and what it puts in them is a property of
its class, not of the instance: the filters are collected at class
definition. Resolving that per node made attach pay a filter walk and a
per-node index entry for every mount; resolving it per class makes both
registration and tree-exit a dict lookup and one append or tombstone per
handler. Buckets are created on first use and never removed, so a plan
stays valid for the life of the tree.
"""
plan = self._registration_plans.get(node_type)
if plan is not None:
return plan
entries: list[tuple] = []
for method_name, filt in node_type._simvx_input_handlers:
kind = filt["kind"]
if kind in ("button", "axis"):
named = {found for target in filt["targets"] for found in recognised_in(target)}
if named:
self._recognised_by_handlers |= named
buckets: list[tuple[_HandlerBucket, InputBinding | None]] = []
tables = self._capture_tables if filt["capture"] else self._bubble_tables
if filt["capture"]:
self._has_capture_handlers = True
if kind == "action":
for name in filt["targets"]:
for released in filt["released"]:
buckets.append((self._input_bucket(self._action_handlers, (name, released)), None))
elif kind == "button":
# One spelling for every device's button; the device type
# says which event carries it, so which table it routes in.
for button, chord in zip(filt["targets"], filt["chords"], strict=True):
device = button.device
if device is DeviceType.KEYBOARD:
for released in filt["released"]:
buckets.append((self._input_bucket(tables.key, (button, released)), chord))
if filt["repeat"]:
buckets.append((self._input_bucket(tables.key_repeat, button), chord))
if not filt["capture"]:
self._has_repeat_handlers = True
elif device in _POINTER_TYPES:
for phase in filt["phases"]:
buckets.append((self._input_bucket(tables.pointer, (button, phase)), chord))
else:
for released in filt["released"]:
key = (button, released)
buckets.append((self._input_bucket(tables.device_button, key), chord))
elif kind == "pointer":
for phase in filt["phases"]:
if phase is PointerPhase.MOVE and filt.get("samples"):
# Every sample rather than the merged move, so never both.
self._has_sample_handlers = True
buckets.append((self._sample_handlers, None))
else:
buckets.append((self._input_bucket(tables.pointer, (None, phase)), None))
elif kind == "axis":
for target in filt["targets"]:
# A stick is its two axes: a handler on it hears either move.
axes = (target.x, target.y) if isinstance(target, DeviceStick) else (target,)
for axis in axes:
buckets.append((self._input_bucket(tables.device_axis, axis), None))
elif kind == "event_class":
for event_class in filt["event_classes"]:
buckets.append((self._input_bucket(tables.event_class, event_class), None))
else:
buckets.append((tables.catch_all, None))
entries.extend((bucket, method_name, chord, filt["consume"]) for bucket, chord in buckets)
plan = tuple(entries)
self._registration_plans[node_type] = plan
return plan
def _register_input_node(self, node: Node) -> None:
"""Insert *node*'s ``@on_input`` handlers into the dispatch tables.
One append per routing key the filter names. Appending rather than
inserting in tree order is deliberate: attach is hot during a scene
change, and :meth:`_ordered` pays for the order later, once, and only
for the buckets an event actually reaches.
The plan is kept on the node, not looked up again on the way out. A
node's class can change while it is mounted -- a hot reload swaps it in
place, precisely so the new code runs -- and asking the class at exit
would then retire the node from the buckets the *new* class registers
in and strand its real entries where nothing can reach them.
"""
plan = self._registration_plan(type(node))
for bucket, method_name, chord, consume in plan:
bucket.add((node, method_name, chord, consume))
node._input_plan = plan
def _unregister_input_node(self, node: Node) -> None:
"""Retire *node*'s entries from the dispatch tables on tree-exit.
The plan the node registered with says which buckets hold it, so this
reaches exactly those, and each one tombstones rather than rebuilding.
Between them those are what make a scene teardown linear: asking every
bucket, and rebuilding each one per departing node, are two separate
factors of the node count, and together they measured ninety
milliseconds for three thousand nodes against one and a half for the
same nodes carrying no handler.
"""
plan = node.__dict__.pop("_input_plan", ())
for bucket, _method_name, _chord, _consume in plan:
bucket.drop(node)
def _actions_for_event(self, event: InputEvent) -> list[tuple[str, bool, int]]:
"""Return ``(action name, pressed, player)`` triples whose bindings match *event*.
A discrete device carries its own edge, so its pairs simply repeat
``event.pressed``. A stick axis is continuous: it becomes an action edge
only where it crosses a binding's deadzone in the direction that binding
names, so one axis event can be a release of one action and a press of
another (a left/right pair driven by the same stick, flipped in a single
move) and each pair states its own edge.
Axis resolution is per event, comparing the event's value against the
value the axis held at the start of the frame. On the desktop, where a
pad is polled once per frame, that is exact. A test writing the same axis
twice within one frame measures both events against the same frame-start
value, so the second event repeats rather than continues the first one's
edge; end the frame between writes to measure them separately.
**A typed action is resolved whole.** Buttons, an ``Axis1D`` and a
stick all move one value, so a continuous action has no per-binding
edge to read: its edge is that value crossing the deadzone, and it is
resolved once per event against what the action read after the previous
one, whichever device moved.
**Every triple names a player.** One is the player who hears the
device the input came from, and there is normally exactly one: the
default player owns everything until a game pairs a device to somebody
else. Two players sharing one keyboard produce a triple each, which is
how the same key opens an action for both of them.
Empty list if nothing matches.
"""
# The device state this event carries is already written, so the
# continuous actions are resolved first and whatever their value did
# travels with the same event.
inp = self.input
matched: list[tuple[str, bool, int]] = inp._continuous_edges()
target_key: DeviceButton | None = None
target_names: frozenset[DeviceButton] = frozenset()
target_device = 0
pad_id: int | None = None
target_code: int | None = None
if isinstance(event, KeyEvent):
target_key = event.key
target_code = key_code(event.key)
target_device = event.device_id
# A side of a modifier is also the modifier: Key.CTRL hears either Control.
group = MODIFIER_OF.get(event.key)
target_names = frozenset((event.key,) if group is None else (event.key, group))
pressed = event.pressed
elif isinstance(event, PointerEvent):
phase = event.phase
if event.button is None or (phase is not PointerPhase.DOWN and phase is not PointerPhase.UP):
return matched
# Every name the press put in front of the bindings: the device's
# own, and the roles when it is the pointer standing in for the
# single pointer an action models.
target_names = inp._binding_names(event)
target_device = event.device_id
pressed = phase is PointerPhase.DOWN
elif isinstance(event, ButtonEvent):
target_names = inp._binding_names(event)
target_device = event.device_id
pressed = event.pressed
if event.button.device is DeviceType.CONTROLLER:
pad_id = event.device_id
elif isinstance(event, AxisEvent):
matched.extend(self._actions_for_device_axis(event))
return matched
else:
return matched
owners = inp._pad_players(pad_id) if pad_id is not None else inp._desk_players()
if not owners:
return matched
def hit(entry: Action) -> bool:
for b in entry.bindings:
if inp._stale and inp._binding_stale(b):
# The element does not count yet: a control took the press,
# or a context change arrived while it was already held.
# Either way the action was never opened, so there is no
# edge to close when the player finally lets go.
continue
control = b.control
if not isinstance(control, DeviceButton):
continue
if control.on(None) not in target_names and (target_key is None or key_code(control) != target_code):
continue
# A binding pinned to one device hears that device only.
if control.unit is not None and control.unit != target_device:
continue
# Ctrl+S is S while Ctrl is held, on the press and the release alike.
if not inp._binding_held(b):
continue
return True
return False
# A player who rebound an action reads their own copy of it, so the
# owner is inside the loop rather than outside. Empty is the usual
# case and costs one dict test.
per_player = self.input_map._player_actions
for action_name, shared in self.input_map._actions.items():
if shared.kind is not ActionKind.BUTTON:
# Resolved above, from the action's value rather than from one
# binding: a continuous action carries no device edge of its
# own, and every binding behind it moves the same value.
continue
for who in owners:
own = per_player.get(who) if per_player else None
entry = own.get(action_name, shared) if own else shared
if hit(entry):
matched.append((action_name, pressed, who))
return matched
def _actions_for_device_axis(self, event: AxisEvent) -> list[tuple[str, bool, int]]:
"""Resolve a pad's or a tablet's axis motion to the button-action edges it crosses.
The same crossing test the polling path uses, so a handler and a poll
agree on when a half of the axis opens and closes.
"""
inp = self.input
owners = (
inp._pad_players(event.device_id) if event.axis.device is DeviceType.CONTROLLER else inp._desk_players()
)
if not owners:
return []
names = inp._axis_forms(event)
matched: list[tuple[str, bool, int]] = []
per_player = self.input_map._player_actions
for action_name, shared in self.input_map._actions.items():
if shared.kind is not ActionKind.BUTTON:
continue
for who in owners:
own = per_player.get(who) if per_player else None
entry = own.get(action_name, shared) if own else shared
for b in entry.bindings:
control = b.control
if not isinstance(control, DeviceAxis) or control.whole not in names:
continue
if control.unit is not None and control.unit != event.device_id:
continue
now = inp._axis_beyond_deadzone(b, inp._device_axis(b))
before = inp._axis_beyond_deadzone(b, inp._device_axis(b, before=True))
if now != before:
matched.append((action_name, now, who))
break
return matched
[docs]
def render(self, renderer):
"""Draw the scene, then the overlay layer, into ``renderer``.
Runs inside this tree's input span: ``on_draw`` is game code like any
other hook, so a handler that reads ``Input`` or registers an action
sees this tree rather than the process-wide default. That matters for
an isolated tree, which is how the editor renders a game in play mode.
"""
with self.activate_input():
cam = self._current_camera_2d
_has = hasattr(renderer, "push_transform")
if _has and cam is not None:
# One Camera2D mapping: the renderer bakes exactly the
# matrix Camera2D.canvas_transform returns, which world_to_screen and the
# Family-B passes also use, so hit-testing and rendering agree.
renderer.push_transform(*cam.canvas_transform(self.screen_size))
if self.root:
self.root._draw_recursive(renderer)
if _has and cam is not None:
renderer.pop_transform()
# Overlay pass: the registry-driven on-top layer (dropdowns, dialogs,
# tooltips). Drawn after the main tree so submission order = GPU order =
# on top. ``if self.overlays`` is an O(1) empty check, so this is
# zero-cost when no overlay is open. The SAME iter_overlay_draws feeds the
# item pipeline (graphics), so desktop == web and headless == live.
if self.overlays:
from .ui.overlay import iter_overlay_draws
if hasattr(renderer, "reset_clip"):
renderer.reset_clip()
for kind, ctrl, entry in iter_overlay_draws(self, viewport=None):
if kind == "dim":
self._draw_overlay_scrim(renderer, self.screen_size, entry.dim_colour)
elif ctrl.visible:
ctrl._draw_recursive(renderer)
@staticmethod
def _draw_overlay_scrim(renderer, screen_size, colour):
"""Draw one full-screen scrim rect in screen space (overlay dim backdrop).
``reset_clip`` resets the scissor; the root walk already popped the
Camera2D transform before this pass, so the renderer is at the root screen
transform. Push identity defensively so a panned/zoomed Camera2D never
offsets the scrim, then draw the full-viewport rect.
"""
w, h = float(screen_size[0]), float(screen_size[1])
has_xf = hasattr(renderer, "push_transform")
if has_xf:
renderer.push_transform(1, 0, 0, 1, 0, 0)
renderer.rect((0.0, 0.0), (w, h), fill=tuple(colour))
if has_xf:
renderer.pop_transform()
def _active_camera_3d(self) -> Camera3D | None:
"""The camera this tree's 3D scene renders through, or ``None``.
The renderers' rule: an explicit render-camera override wins, else the
first ``Camera3D`` visible in the hierarchy, visited in the order the
scene adapter collects nodes (a stack walk that does not enter a
``SubViewport``, which renders its own subtree through its own camera).
The camera list is cached per tree structure; visibility is read on
every call.
"""
override = getattr(self, "_render_camera_override", None)
if isinstance(override, Camera3D):
return override
if self.root is None:
return None
key = (self._structure_version, self.root)
cached = self._camera_3d_cache
if cached is None or cached[0] != key:
from .viewport import SubViewport
found: list[Camera3D] = []
stack = [self.root]
while stack:
node = stack.pop()
if isinstance(node, Camera3D):
found.append(node)
if isinstance(node, SubViewport) and node is not self.root:
continue
stack.extend(node.children)
cached = self._camera_3d_cache = (key, tuple(found))
for camera in cached[1]:
if camera._visible_in_hierarchy:
return camera
return None
def _cast(
self,
screen_pos: tuple[float, float] | np.ndarray,
button: DeviceButton = Pointer.PRIMARY,
*,
source: InputEvent | None = None,
pressed: bool = True,
) -> tuple[Node | None, bool]:
""":meth:`input_cast`, also saying whether the hit node's ``on_picked`` claimed the press."""
hit = self._ray_pick(screen_pos)
if hit is None:
return None, False
best_node, origin, direction, best_t, point, normal = hit
event = PickEvent(
source=source,
button=button,
pressed=pressed,
origin=origin,
direction=direction,
distance=best_t,
node=best_node,
point=point,
normal=normal,
)
with self.activate_input():
claimed = best_node._internal_picked(event) is True
# ``on_picked`` claims the way every other handler does, by saying so,
# and the claim reaches the pointer event it came from: the press
# stops there, with no action edge and no broadcast behind it. A pick
# that happens and says nothing takes nothing away.
return best_node, claimed
def _ray_pick(self, screen_pos) -> tuple[Node, Any, Any, float, Any, Any] | None:
"""The nearest pickable owner the camera's ray through *screen_pos* meets, and where; delivers nothing."""
if not self.root:
return None
camera = self._active_camera_3d()
if camera is None:
return None
if camera._tree is self:
ray = camera.ray_at(screen_pos)
else:
# A render override from outside the tree (an editor's preview
# camera) has no viewport of its own; it draws into this tree's.
sw, sh = self.screen_size
proj = camera.projection_matrix(sw / sh if sh > 0 else 1.0)
ray = Ray.from_screen(screen_pos, self.screen_size, camera.view_matrix, proj)
origin, direction = ray.origin, ray.direction
best_t = float("inf")
best_node: Node | None = None
best_point = best_normal = None
# Owners hit through their physics world, grouped by world so each world
# is cast once whatever the number of pickable bodies in it.
by_world: dict[Any, set[Node]] = {}
for shape in self.root.find_all(CollisionShape3D):
if not shape.pickable:
continue
owner = shape.parent if shape.parent else shape
if not owner._can_receive_input():
# Paused or disabled: the ray goes through it to whatever is
# behind, the same as a control that cannot hear a pointer is
# not in the hit chain.
continue
if isinstance(owner, PhysicsObject3D) and owner._handle is not None and owner.collision_layer:
by_world.setdefault(owner._world, set()).add(owner)
continue
t = ray_intersect_sphere(origin, direction, shape.world_position, shape.pick_radius)
if t is not None and t < best_t:
best_t, best_node = t, owner
best_point = ray.at(t)
# The pick volume is the bounding sphere, so the normal is the
# radial direction out of its centre.
offset = best_point - shape.world_position
length = float(np.linalg.norm(offset))
best_normal = Vec3(offset / length) if length > 0.0 else Vec3(-direction)
for world, owners in by_world.items():
query = PhysicsQuery(world, self._physics_nodes.get(world))
for hit in query.raycast_all(origin, direction, distance=best_t):
if hit.node in owners:
if hit.distance < best_t:
best_t, best_node = hit.distance, hit.node
best_point, best_normal = Vec3(hit.point), Vec3(hit.normal)
break
if best_node is None:
return None
return best_node, origin, direction, best_t, best_point, best_normal
@overload
def group(self, name: str, /) -> list[Node]: ...
@overload
def group[T](self, name: str, of: type[T], /) -> list[T]: ...
@overload
def group[A, B](self, name: str, of: tuple[type[A], type[B]], /) -> list[A | B]: ...
@overload
def group[A, B, C](self, name: str, of: tuple[type[A], type[B], type[C]], /) -> list[A | B | C]: ...
@overload
def group[T](self, name: str, of: tuple[type[T], ...], /) -> list[T]: ...
[docs]
def group(self, name, of=None, /):
"""The members of group ``name``, in the order they joined it, as a new list.
A node joins a group when ``add_to_group`` runs while it is in this
tree, or when it enters the tree already carrying the group; leaving
the group or the tree drops it, and joining again puts it last.
:meth:`Node.destroy` drops the node from its groups at once, although
it stays in the tree until the end of the frame.
``of``, a class or a tuple of classes, keeps only the members that are
instances of it, and the list is typed as that class, so a group may
hold several types and a loop calls the right one's methods::
for enemy in tree.group("enemies", Enemy):
enemy.stun()
A ``runtime_checkable`` Protocol class as ``of`` matches by methods, not by inheritance;
mypy refuses one where ``type[T]`` is expected, so such a call carries
``# type: ignore[type-abstract]`` under mypy (pyright accepts it).
The list is a snapshot: a member an earlier iteration destroys is
still visited, as with any copy.
"""
members = self._groups.get(name)
if members is None:
if of is not None:
_check_group_class(of)
return []
if of is None:
return list(members)
_check_group_class(of)
return [node for node in members if isinstance(node, of)]
@overload
def first(self, name: str, /) -> Node | None: ...
@overload
def first[T](self, name: str, of: type[T], /) -> T | None: ...
@overload
def first[A, B](self, name: str, of: tuple[type[A], type[B]], /) -> A | B | None: ...
@overload
def first[A, B, C](self, name: str, of: tuple[type[A], type[B], type[C]], /) -> A | B | C | None: ...
@overload
def first[T](self, name: str, of: tuple[type[T], ...], /) -> T | None: ...
[docs]
def first(self, name, of=None, /):
"""The earliest-joined member of ``name`` that is an instance of ``of`` (any type if omitted), or ``None``."""
if of is not None:
_check_group_class(of)
members = self._groups.get(name, ())
if of is None:
return next(iter(members), None)
return next((node for node in members if isinstance(node, of)), None)
[docs]
def create_timer(self, seconds: float) -> SceneTreeTimer:
"""Create a one-shot timer that emits ``timeout`` after ``seconds``.
Returns immediately with a :class:`SceneTreeTimer`; connect its
``timeout`` signal to run code after the delay. The timer is owned and
ticked by the tree (right after singletons) and discarded once it fires,
so it needs no Node in the scene and no manual cleanup. Honours
``tree.paused``. For delays inside a coroutine, prefer ``wait(seconds)``.
"""
timer = SceneTreeTimer(seconds)
self._scene_timers.append(timer)
return timer
[docs]
def call_deferred(self, method: Callable[..., Any], *args: Any) -> None:
"""Queue ``method(*args)`` to run at the end of this frame (escape hatch).
The tree-level counterpart to :meth:`Node.call_deferred`, for code that
has no node to defer from (plugins, singleton helpers, non-node systems).
Same discouraged-escape-hatch guidance applies: prefer a safe-by-default
path (``destroy``, ``Property(coalesce=True)``, ``events.publish_deferred``)
when one exists. ``method`` is a callable, never a string.
"""
self._deferred_calls.append((method, args))
# --- Singletons (nodes that persist across scene changes) ---
[docs]
@property
def singletons(self) -> dict[str, Node]:
"""Read-only view of registered singletons."""
return self._singletons
[docs]
def add_singleton(self, name: str, node: Node):
"""Register ``node`` as a persistent singleton attached to the tree.
An unparented node is mounted as a child of the tree's own top node: it
enters the tree, runs ``on_ready()`` immediately, processes ahead of the
scene in registration order, and survives ``change_scene()``. Reach for it
when a node must outlive the scene that started it: music that keeps
playing across a scene change, or a save file held open. Retrieve one via
``tree.singletons[name]``; it is not reachable from the scene root.
It is not a home for the game's own state. A score belongs to whatever
owns the run, and reaching it through a name from anywhere makes every
reader of that name a writer of it.
A node the scene already owns may also be given a name. It stays where it
is, so it processes in its scene position rather than ahead of the scene,
and it leaves with its scene: after a ``change_scene`` the name is gone.
A singleton is ticked but not drawn -- drawing walks the scene. A global
HUD is either a child of the scene or a full-screen overlay.
The name ``"events"`` is reserved for the engine-provided
:class:`~simvx.core.event_bus.EventBus`; use ``tree.events`` instead.
Raises:
ValueError: the name is ``"events"``; the name is already bound to a
different node; ``node`` already holds a different name; or
``node`` is a child of some other tree.
"""
if name == RESERVED_SINGLETON_EVENTS:
raise ValueError(
"Singleton name 'events' is reserved for the engine-provided "
"EventBus. Access it via tree.events; pick a different name "
"for your singleton."
)
bound = self._singletons.get(name)
if bound is not None and bound is not node:
raise ValueError(
f"Singleton name {name!r} is already registered to {bound.name!r}. "
f"Call remove_singleton({name!r}) first, or pick a different name."
)
for other_name, other in self._singletons.items():
if other is node and other_name != name:
raise ValueError(
f"{node.name!r} is already registered as the {other_name!r} singleton. "
"A node holds at most one singleton name."
)
parent = node.parent
if parent is not None and node._tree is not self:
raise ValueError(
f"Cannot register {node.name!r} as a singleton: it is a child of "
f"{parent.name!r}, which is not in this tree. Detach it first, or "
"pass a node that has no parent."
)
self._singletons[name] = node
if parent is None:
self._tree_root.add_child(node)
if self._scene_root is not None:
# The scene stays the last child, so singletons keep processing
# ahead of it however late they register.
self._tree_root.children.move_last(self._scene_root)
[docs]
def remove_singleton(self, name: str):
"""Drop a singleton's name binding, and unmount it if the tree owns it.
A singleton the tree mounted (registered while unparented) is carried out
of the tree, so its ``on_exit_tree`` runs and anything it registered on
the way in is released. A singleton the scene owns keeps its place in the
scene: dropping the name is the whole of the operation, because the node
is a scene node that happens to have a name.
"""
node = self._singletons.pop(name, None)
if node is not None and node.parent is self._tree_root:
# Opens the input span itself and runs _exit_tree.
self._tree_root.remove_child(node)
def _prune_singletons(self) -> None:
"""Drop registry entries whose node has left this tree.
A singleton registered on a node the scene owns leaves with that scene,
exactly as a ``unique_name`` registration does. One pass over a dict that
holds single digits of entries, at scene-swap and delete-flush time only;
never on the per-frame path.
"""
dead = [name for name, node in self._singletons.items() if node._tree is not self]
for name in dead:
del self._singletons[name]
def _prune_dead_index_entries(self) -> None:
"""Drop group and unique-name entries naming a node that has left.
Every exit deregisters its own node (``Node._exit_tree_inner``), so this
finds nothing in a correct engine. It replaces the clear-and-rebuild
``change_scene`` used to do, and it is a detector rather than a cover-up:
under ``Node.strict_errors`` a survivor raises, so a deregistration bug
fails a test instead of being silently mopped up, and in a release build
it is mopped up so no game gets a dead node out of ``tree.group()``.
"""
stale = [(group, node) for group, members in self._groups.items() for node in members if node._tree is not self]
stale_unique = [n for n, node in self._unique_nodes.items() if node._tree is not self]
if (stale or stale_unique) and Node.strict_errors:
raise RuntimeError(f"Nodes left the tree without deregistering: groups={stale}, unique={stale_unique}")
for group, node in stale:
self._groups[group].pop(node, None)
for name in stale_unique:
del self._unique_nodes[name]
# --- Unique nodes ---
[docs]
def unique(self, name: str) -> Node | None:
"""Get a unique node by name. Returns None if not found."""
return self._unique_nodes.get(name)
# --- Internal helpers ---
def _group_add(self, group: str, node: Node):
members = self._groups.get(group)
if members is None:
members = self._groups[group] = {}
members.setdefault(node, None)
def _group_remove(self, group: str, node: Node):
members = self._groups.get(group)
if members is not None:
members.pop(node, None)
def _invalidate_for_target_change(self):
"""Re-capture whatever drew against the screen's old size, after the scene's space resized.
Two dependencies on the size, found in one walk.
**Draws that read it.** A node whose last ``on_draw`` read
``viewport_size`` or ``screen_size`` is marked with ``queue_redraw``,
whatever its type, so a board that fills the window follows the window
with no code of its own. The walk does not enter a ``SubViewport``:
nodes there draw into its texture, whose size has not moved. The read
is recorded against the node whose draw ran, so a node that read
another target's size in its own draw is marked when its own target
changes instead; that is the one case still needing ``queue_redraw``.
**Anchored controls.** A screen-size change moves or resizes every
anchored control (any of anchor_{left,top,right,bottom} != 0) and
carries every Control in its subtree along: descendants are laid out
relative to the moved rect, so their absolute coordinates change even
with zero anchors of their own. That rect is computed by layout, not
read in a draw, so it is marked here by shape. Controls outside every
anchored subtree keep rects computed purely from their own
position/size, so their caches stay valid here: own-size changes are
handled by Property setters and own-position changes by
_invalidate_transform.
Both draw paths bake absolute screen coordinates, so both are
invalidated across the whole moved set: the legacy recorder cache
(_draw_dirty/_draw_cache) and the retained item pipeline's bits
(_render_dirty on each anchored control, whose rect may have changed
size as well as position; _transform_render_dirty on everything that
moved) that RenderItemCache scans to re-capture stale geometry.
The walk starts at the tree's own top node, not at the scene root, so a
node that a singleton owns is reached as well. An overlay parented to
a singleton (the boot splash, the attribution watermark) is anchored to
the screen by definition, so a resize is exactly the event that moves it,
and starting at the scene root would leave its retained geometry pinned to
the corner the old screen had.
Uses an iterative stack to avoid Python recursion overhead on deep trees.
"""
from .ui import Control
# (node, moved with an anchored ancestor, outside every SubViewport)
stack = [(self._tree_root, False, True)]
while stack:
node, moved, on_screen = stack.pop()
if on_screen:
if getattr(node, "_is_subviewport", False):
on_screen = False
elif node._ambient_reads & READS_VIEWPORT_SIZE:
node.queue_redraw()
if isinstance(node, Control):
anchored = bool(node.anchor_left or node.anchor_top or node.anchor_right or node.anchor_bottom)
if anchored or moved:
moved = True
if anchored:
node._render_dirty = True
node._transform_render_dirty = True
if node._draw_cache is not None:
node._draw_dirty = True
node._draw_cache = None
children = node.children
if children:
stack.extend((child, moved, on_screen) for child in children)
def _queue_delete(self, node: Node):
self._delete_queue.append(node)
[docs]
def clear_delete_queue(self) -> None:
"""Abandon every pending :meth:`Node.destroy` without running it.
For a teardown that is carrying the whole tree out anyway (a harness
dropping its tree): the queued nodes go with it, so the removal
they were waiting for is moot. Clears their pending state as well, since a
node left marked as destroying would report a removal that is never coming,
and puts each one still in this tree back in its groups.
"""
for node in self._delete_queue:
node._destroying = False
if node._tree is self:
for group in node._groups:
self._group_add(group, node)
self._delete_queue.clear()
def _flush_deletes(self):
"""Carry every node queued by :meth:`Node.destroy` out of the tree.
Runs at the end-of-frame sync point. A node queued *during* the drain
(an ``on_exit_tree`` that destroys a sibling, say) is picked up by the
same pass, so a cascade completes in the frame it was started in; the
queue is finite because :meth:`Node.destroy` is idempotent, and a node
this pass has finished is marked as such and can never queue again.
Each removed subtree is finished whole (:meth:`Node._spend_subtree`): every
node in it is spent and its signal connections end here.
Exactly one exit per node. A queued node whose ancestor was destroyed
first has already been carried out by that ancestor's ``_exit_tree`` and
so holds no tree, leaving only the parent link to unlink
(``_detach_child``); running ``remove_child`` on it would fire its
``on_exit_tree`` a second time. A node that is still in a tree goes
through the full ``remove_child`` instead, so it never leaves a
half-registered entry behind in a tree that is still live.
"""
if not self._delete_queue:
return
# Drained a batch at a time: what the batch's exit hooks queue lands in
# a fresh list and runs next, and an exit hook that changes scene prunes
# that live list, never the batch being walked. An old-scene entry left
# in the batch is already spent, so it is only unlinked again.
while self._delete_queue:
batch, self._delete_queue = self._delete_queue, []
for node in batch:
parent = node.parent
if parent is not None:
if node._tree is None:
parent._detach_child(node)
else:
parent.remove_child(node)
node._spend_subtree()
# A destroyed singleton has left the tree through the ordinary removal
# above, so its name binding goes with it.
self._prune_singletons()
def _flush_deferred_calls(self):
"""Drain calls queued via :meth:`Node.call_deferred` / :meth:`call_deferred`.
Runs once per frame at the end-of-frame sync point, outside tree
traversal. A snapshot swap means calls queued *during* the drain are
held for the next frame, so a self-requeuing deferred call cannot spin
the current frame.
A call bound to a node that has since left this tree is dropped (its
context is gone, matching Godot's freed-object behaviour). Node-bound
calls run through the owner's :meth:`Node._safe_call`, so a failing
deferred call obeys the same strict-raise / release-isolate policy as
every other hook dispatch instead of silently eating the rest of the
queue.
"""
if not self._deferred_calls:
return
pending = self._deferred_calls
self._deferred_calls = []
for fn, args in pending:
owner = getattr(fn, "__self__", None)
if isinstance(owner, Node):
if owner._tree is not self:
continue # node left the tree after queueing: drop the call
owner._safe_call(fn, *args)
elif Node.strict_errors:
fn(*args)
else:
try:
fn(*args)
except Exception:
log.exception("deferred call %r failed", fn)
# ========================================================================
# UI Input System (delegated to UIInputManager)
# ========================================================================
def _deliver_to_ui(self, route: Callable[[Node | None, Any], Node | None], event: Any) -> Node | None:
"""Run one UI routing call as the scene's code, and say who took the event.
A control's handler, or a signal it emits, may end the scene. The event
then went with the scene that ended, so the tree's own top node is
reported as having taken it: nothing after the targeted phase delivers
it to the scene that replaced it, and the release of the press it began
is claimed the same way.
"""
claimant = None
with self.activate_input():
with self._scene_span:
claimant = route(self.root, event)
if self._next_scene is not None:
self._mount_next_scene()
return self._tree_root
return claimant
[docs]
def ui_key(self, event: KeyEvent) -> Node | None:
"""Route a key press or release to the UI.
The UI is handed the object the tree received, so a widget reads the
physical key and the modifier flags off the same event a broadcast
handler sees. Typed characters take :meth:`ui_text`, pointers
:meth:`ui_pointer` and the wheel :meth:`ui_scroll`.
Returns the node that took the key, or ``None`` when the UI let it
through. A focused text field swallowing ``W`` is what stops the
character behind it walking.
Routing runs inside this tree's input span: ``on_gui_input``, focus
signals and the ``pressed`` / ``released`` handlers reached from here
are arbitrary game code, and they see this tree's ``Input`` and
``InputMap``.
"""
return self._deliver_to_ui(self._ui.ui_key, event)
[docs]
def ui_text(self, event: TextEvent) -> Node | None:
"""Route a typed character to the UI.
Returns the node that took it, or ``None``. Runs inside this tree's
input span for the same reason :meth:`ui_key` does.
"""
return self._deliver_to_ui(self._ui.ui_text, event)
[docs]
def ui_pointer(self, event: PointerEvent) -> Node | None:
"""Route a pointer to the controls under it.
The UI is handed the object the tree received rather than a flattened
copy of it, so a widget reads the phase, the button, the pressure and
the pointer id off the same event a broadcast handler sees.
Returns the node that took the pointer, or ``None`` when nothing in the
UI stops it.
"""
return self._deliver_to_ui(self._ui.ui_pointer, event)
[docs]
def ui_pad(self, event) -> Node | None:
"""Route a controller button or axis to UI navigation.
The pad's half of ``ui_key``: the UI resolves it through the same
navigation actions a key goes through, so a controller can move focus,
activate what it lands on and dismiss a dialog. Runs inside this
tree's input span for the same reason :meth:`ui_key` does.
Returns the node the pad's press moved, activated or dismissed, or
``None`` when navigation did nothing with it.
"""
return self._deliver_to_ui(self._ui.ui_pad, event)
[docs]
def flush_pointer_motion(self) -> None:
"""Deliver the frame's accumulated pointer motion, from every router.
Motion is coalesced into one event per frame carrying the frame's
total (:meth:`InputRouter.flush_motion
<simvx.core.input.router.InputRouter.flush_motion>`), and this is
where the frame spends it: at the top of the input phase, before
physics, so a fixed tick reads a pointer position that is this
frame's. Several routers can feed one tree, and each holds its own
device's motion.
"""
for router in tuple(self._routers):
router.flush_motion()
[docs]
def cancel_capture(self, pointer_id: int, event: PointerEvent):
"""Take a captured pointer back, telling whoever held it it was taken."""
with self.activate_input():
self._ui.cancel_capture(pointer_id, event)
[docs]
@property
def focus_active(self) -> bool:
"""Whether the keyboard reaches this tree now.
A window's tree has it while the window has the operating system's
focus; a hosted tree while its :class:`SceneHost` holds focus in a
tree that has it and is forwarding. A tree whose keyboard went
elsewhere keeps its focused widget, drawn as it was, with this False.
"""
host = self._host
if host is not None:
outer = host.tree
return bool(host.forwarding and host.focused and outer is not None and outer.focus_active)
return not self._ui._unfocused_sources
[docs]
def hit_test(
self, point: tuple[float, float], *, pointer_type: PointerType | str = PointerType.MOUSE, source_id: int = 0
) -> Node | None:
"""What *point* would hit: the nearest control under it, else the pickable body its ray meets.
Writes no state: no hover, no focus, no pick delivered. The same
answer the pointer pipeline acts on, so code that crosses a host
boundary (a drag session, a tooltip) asks the tree on the far side the
question it would ask its own. *pointer_type* matters because a
finger hits a touch target anywhere in its grown area.
"""
root = self.root
if root is None:
return None
probe = PointerEvent(
phase=PointerPhase.MOVE,
pointer_type=PointerType(pointer_type),
position=(float(point[0]), float(point[1])),
source_id=source_id,
)
chain = self._ui._hit_chain(self._ui._scope_root(root), probe.position, probe)
if chain:
return chain[0]
hit = self._ray_pick(probe.position)
return hit[0] if hit is not None else None
def _set_focused_control(self, control, reason: FocusReason = FocusReason.CODE):
"""Move keyboard focus to *control*, or nowhere, recording *reason*.
The engine's one door into the router's focus: ``grab_focus``,
``release_focus``, Tab traversal from code and the overlay registry
all come through it. Spanned because the change emits
``focus_entered`` and ``focus_exited``, and these callers are not a
dispatch already inside a span.
"""
with self.activate_input():
self._ui._set_focused_control(control, reason)
@property
def drag_threshold(self) -> dict[PointerType, float]:
"""How far a pointer travels from its press before the motion is a drag, per pointer type.
The default every node inherits; ``node.drag_threshold`` overrides it
for a subtree. Assigning a map replaces the types it names and keeps
the rest.
"""
return self._drag_threshold
[docs]
@drag_threshold.setter
def drag_threshold(self, value: Mapping[PointerType | str, float]) -> None:
given = drag_threshold_map(value)
if given is None:
raise TypeError(
"tree.drag_threshold takes a mapping from pointer type to distance; it has nothing to inherit"
)
self._drag_threshold.update(given)
[docs]
@property
def focus_owner(self) -> Node | None:
"""The node holding keyboard focus, or None. Read-only: ``grab_focus()`` and ``release_focus()`` move it."""
return self._ui._focused_control
[docs]
@property
def focus_reason(self) -> FocusReason:
"""Why keyboard focus last moved: a click, Tab, a direction, hover, an overlay, code, or the owner leaving."""
return self._ui._focus_reason
def _notify_ui_subtree_removed(self, node):
"""Release focus and pointer capture for a subtree leaving the tree.
Every node exit runs this, so the gate comes first: with nothing
focused and no pointer captured there is no table to search, and the
input span is not worth opening to find that out.
"""
ui = self._ui
if ui._focused_control is None and not ui._owners:
return
with self.activate_input():
ui.notify_subtree_removed(node)
def _update_mouse_over_states(self, mouse_pos):
"""Update mouse_over state for all controls (forwarded to UIInputManager).
Spanned because the update emits ``mouse_entered`` / ``mouse_exited``.
"""
with self.activate_input():
self._ui._update_mouse_over_states(self.root, mouse_pos)
def _find_control_at_point(self, point):
"""Find topmost control at screen position (forwarded to UIInputManager)."""
return self._ui._find_control_at_point(self.root, point)