"""Descriptors, signals, enums, and type aliases used throughout the engine."""
import inspect
import logging
import weakref
from collections.abc import Callable, Generator
from enum import IntEnum, auto
from typing import TYPE_CHECKING, Any, NamedTuple
from .math.types import Vec2, Vec3
def _has_shape(v: Any) -> bool:
"""Return True for array-like values (numpy ndarray, Vec2/Vec3) whose
equality comparison would return another array rather than a scalar bool.
"""
return hasattr(v, "shape")
log = logging.getLogger(__name__)
if TYPE_CHECKING:
pass
Coroutine = Generator[None]
# Sentinel used by Property to distinguish "no default supplied" from a default of None.
_UNSET: Any = object()
[docs]
class CoroutineHandle:
"""Cancellable handle returned by Node.start_coroutine()."""
__slots__ = ('_gen', '_cancelled')
def __init__(self, gen: Coroutine):
self._gen = gen
self._cancelled = False
[docs]
def cancel(self):
"""Cancel this coroutine. It will be removed on the next tick."""
self._cancelled = True
[docs]
@property
def is_cancelled(self) -> bool:
return self._cancelled
# ============================================================================
# ProcessMode — Controls node processing when tree is paused
# ============================================================================
[docs]
class ProcessMode(IntEnum):
"""Controls whether a node processes when the SceneTree is paused.
Set via ``node.process_mode = ProcessMode.ALWAYS`` (and similar). The
effective mode is resolved by walking up the tree until a non-INHERIT
ancestor is found; the resolved value is cached and invalidated on reparent.
Pause the tree with ``tree.paused = True``. When paused, only ``ALWAYS``
and ``PAUSED_ONLY`` nodes run their ``process`` / ``physics_process``.
Because INHERIT walks ancestors, an ``ALWAYS`` mode set high in the tree
propagates to every INHERIT descendant — defeating the pause for the whole
subtree. Set ALWAYS only on leaf nodes or CanvasLayers, never on a game
root with gameplay children.
"""
INHERIT = 0 # Use parent's mode (default)
PAUSABLE = 1 # Stops when paused (normal game nodes)
PAUSED_ONLY = 2 # Only runs when paused (pause menus)
ALWAYS = 3 # Always runs regardless of pause state
DISABLED = 4 # Never runs
[docs]
class Notification(IntEnum):
"""Notifications dispatched to nodes during lifecycle and property changes."""
TRANSFORM_CHANGED = auto()
VISIBILITY_CHANGED = auto()
ENTER_TREE = auto()
EXIT_TREE = auto()
READY = auto()
PARENTED = auto()
UNPARENTED = auto()
PROCESS = auto()
PHYSICS_PROCESS = auto()
# ============================================================================
# Collision — Result of move_and_slide collision
# ============================================================================
[docs]
class Collision(NamedTuple):
"""Collision info returned by move_and_slide.
Attributes:
normal: Collision normal pointing away from the other body.
collider: The other CharacterBody that was hit.
position: Contact point on collision surface.
depth: Penetration depth.
"""
normal: Vec2 | Vec3
collider: Any
position: Vec2 | Vec3
depth: float
# ============================================================================
# Property — Editor-visible property descriptor
# ============================================================================
[docs]
class Property:
"""Descriptor for editor-visible, serializable node properties.
Declares a typed, validated property that the editor inspector can display
and that the scene serializer persists automatically.
Args:
default: Default value. Type is inferred from this (float, str, bool, Vec2, ...).
range: ``(lo, hi)`` clamp bounds for numeric values.
enum: Allowed values list — the editor renders a dropdown.
hint: Tooltip / description shown in the inspector.
link: When ``True``, the resolved value is the *sum* of this node's
stored value and the parent's value (numeric / vector types), or
the parent's value for other types. Useful for cumulative offsets
that propagate down the tree.
propagate: When True, bool/enum Properties inherit disabling values from parents.
persist: When True, the value is included in ``SaveManager`` snapshots.
save_version: Optional integer schema version recorded alongside the persisted value.
Serialization:
``simvx.core.scene_io`` walks ``node.get_properties()`` and emits any
value that differs from *default* into the ``.py`` scene file.
Loading passes those stored values as kwargs to the node constructor,
which feeds them through ``__set__`` for validation.
Usage::
class Player(Node2D):
speed = Property(5.0, range=(0, 20), hint="Movement speed")
mode = Property("walk", enum=["walk", "run", "fly"])
"""
__slots__ = (
'default', 'default_factory', 'range', 'enum', 'hint', 'name', 'attr', 'link', '_propagate',
'group', 'on_change', 'persist', 'save_version',
)
def __init__(
self,
default: Any = _UNSET,
*,
default_factory: Callable[[], Any] | None = None,
range=None,
enum=None,
hint="",
link=False,
propagate=False,
group="",
on_change: str | None = None,
persist: bool = False,
save_version: int | None = None,
):
"""Create an editor-visible property descriptor.
Args:
default: Default value for the property. Mutually exclusive with
``default_factory``. Mutable containers (``list``, ``dict``,
``set``, ``bytearray``) are rejected here because a single
shared instance would alias across every owning object — pass
``default_factory`` instead.
default_factory: Zero-arg callable invoked the first time each
instance reads the property. The result is cached on the
instance, matching ``functools.cached_property`` and
``dataclasses.field(default_factory=...)`` semantics.
range: Optional (min, max) tuple for numeric clamping.
enum: Optional list of allowed values.
hint: Description shown in the editor inspector.
link: When True, child values are offset from the parent's value.
propagate: When True, bool/enum Settings inherit disabling values from parents.
group: Inspector section name for grouping. Empty string = default "Properties" section.
on_change: Name of a bound method to invoke on the owning instance after a
successful value change (i.e. when the new value differs from the old).
Hooks may fire during ``__init__`` if the Property is passed as a kwarg,
so they must be robust to partially-initialised state. Hooks must be O(1)
and must not write to other Properties (no recursion guard is enforced).
persist: When True, the value is included in ``SaveManager`` snapshots.
save_version: Optional integer schema version recorded alongside the persisted value.
"""
if default is not _UNSET and default_factory is not None:
raise TypeError(
"Property: pass either `default` or `default_factory`, not both"
)
if default is _UNSET and default_factory is None:
raise TypeError(
"Property: must supply `default` or `default_factory`"
)
if default_factory is None and isinstance(default, list | dict | set | bytearray):
raise ValueError(
f"Property has mutable default {default!r}. "
f"Mutable defaults alias across instances. "
f"Use Property(default_factory={type(default).__name__}) instead."
)
self.default = default # Stays as _UNSET when default_factory is supplied.
self.default_factory = default_factory
self.range = range
self.enum = enum
self.hint = hint
self.link = link or propagate # Enable parent-child linking
self._propagate = propagate # Enhanced propagation for bool/enum
self.group = group
self.on_change = on_change
self.persist = persist
self.save_version = save_version
[docs]
def __set_name__(self, owner, name):
self.name = name
self.attr = f"_{name}"
if '__properties__' not in owner.__dict__:
inherited = {}
for base in owner.__mro__[1:]:
if hasattr(base, '__properties__'):
inherited.update(base.__properties__)
break
owner.__properties__ = dict(inherited)
owner.__properties__[name] = self
[docs]
def __get__(self, obj, objtype=None):
if obj is None:
return self
value = getattr(obj, self.attr, _UNSET)
if value is _UNSET:
if self.default_factory is not None:
# Lazily materialise the per-instance default. Bypass __set__ so
# validation/clamping/on_change don't fire — semantics match
# ``functools.cached_property``.
value = self.default_factory()
setattr(obj, self.attr, value)
else:
value = self.default
# Apply parent linking if enabled
if self.link and obj.parent and hasattr(obj.parent, self.name):
parent_value = getattr(obj.parent, self.name)
return self._apply_link(parent_value, value)
return value
[docs]
def __set__(self, obj, value):
if self.range is not None and isinstance(value, int | float):
lo, hi = self.range
clamped = max(lo, min(hi, value))
value = clamped
if self.enum is not None and value not in self.enum:
log.warning("Property %r rejected invalid value %r (allowed: %s)", self.name, value, self.enum)
raise ValueError(f"{self.name} must be one of {self.enum}, got {value!r}")
old = getattr(obj, self.attr, self.default)
setattr(obj, self.attr, value)
if old is value:
changed = False
elif _has_shape(old) or _has_shape(value):
# Arrays (numpy, Vec2/Vec3, …) don't have a scalar != so a distinct
# instance is treated as changed without an elementwise compare.
changed = True
else:
try:
changed = bool(old != value)
except (ValueError, TypeError):
changed = True # numpy arrays, etc.
# Auto-redraw UI controls when property changes
if changed and hasattr(obj, "queue_redraw"):
obj.queue_redraw()
# Fire on_change hook. May fire during __init__ if the Property is passed
# as a kwarg, so hooks must tolerate partially-initialised state.
if changed and self.on_change is not None:
method = getattr(obj, self.on_change, None)
if method is not None:
method()
def _apply_link(self, parent_value, child_value):
"""Apply parent-child linking / propagation based on value type."""
# For bool/enum with propagate: disabling parent overrides child, otherwise child value stands
if self._propagate:
if isinstance(parent_value, bool):
if not parent_value:
return parent_value
return child_value
if isinstance(parent_value, IntEnum):
try:
disabled = type(parent_value)['DISABLED']
if parent_value == disabled:
return parent_value
except (KeyError, TypeError):
pass
return child_value
# For numeric types: child is offset from parent
if isinstance(child_value, int | float) and isinstance(parent_value, int | float):
return parent_value + child_value
# For vectors: child is offset from parent
if isinstance(child_value, Vec2 | Vec3) and isinstance(parent_value, Vec2 | Vec3):
return parent_value + child_value
# For other types: inherit parent value
return parent_value
[docs]
def __repr__(self):
parts = [f"default={self.default!r}"]
if self.range:
parts.append(f"range={self.range}")
if self.enum:
parts.append(f"enum={self.enum}")
return f"Property({', '.join(parts)})"
# ============================================================================
# Child — Declarative child node descriptor
# ============================================================================
[docs]
class Child:
"""Declarative child node. Auto-creates and adds during enter_tree.
Usage:
class Player(Node3D):
camera = Child(Camera3D, fov=90)
health_bar = Child(Node2D, name="HealthBar")
"""
def __init__(self, node_type: type, *args, **kwargs):
self._type = node_type
self._args = args
self._kwargs = kwargs
self._attr_name: str | None = None
[docs]
def __set_name__(self, owner, name):
self._attr_name = name
if '_declared_children' not in owner.__dict__:
inherited = {}
for base in owner.__mro__[1:]:
if hasattr(base, '_declared_children'):
inherited.update(base._declared_children)
break
owner._declared_children = dict(inherited)
owner._declared_children[name] = self
[docs]
def __get__(self, obj, objtype=None):
if obj is None:
return self
return obj.__dict__.get(self._attr_name)
[docs]
def __set__(self, obj, value):
obj.__dict__[self._attr_name] = value
# ============================================================================
# OnReady — Lazy child lookup resolved after ready()
# ============================================================================
[docs]
class OnReady:
"""Lazy child lookup, resolved after ready().
Usage:
class Game(Node):
player = OnReady("Player") # lookup by name
camera = OnReady["MainCamera"] # __class_getitem__ syntax
score = OnReady(lambda n: n.find(ScoreDisplay)) # callable
"""
def __init__(self, path_or_callable):
if isinstance(path_or_callable, str):
self._lookup = lambda node, p=path_or_callable: node.get_node(p)
else:
self._lookup = path_or_callable
self._attr_name: str | None = None
[docs]
def __class_getitem__(cls, key):
"""OnReady["ChildName"] syntax."""
return cls(key)
[docs]
def __set_name__(self, owner, name):
self._attr_name = name
_UNSET = object() # sentinel to distinguish "not cached" from None result
[docs]
def __get__(self, obj, objtype=None):
if obj is None:
return self
cache_key = f'_onready_{self._attr_name}'
cache = obj.__dict__.get(cache_key, self._UNSET)
if cache is not self._UNSET:
return cache
result = self._lookup(obj)
obj.__dict__[cache_key] = result
return result
# ============================================================================
# Connection — returned by Signal.connect()
# ============================================================================
[docs]
class Connection:
"""Handle returned by ``Signal.connect()``. Can disconnect and acts as a callable proxy.
For bound methods on Nodes (objects exposing ``_outgoing_connections``), the
callback is held weakly and the connection auto-cleans when the node is
destroyed — matching Godot 4's signal lifecycle. Other callables (lambdas,
free functions, methods on non-Node objects) keep a strong reference.
"""
__slots__ = ('_signal', '_fn', '_weak', '_connected')
def __init__(self, signal: Signal, fn: Callable):
self._signal = signal
owner = getattr(fn, '__self__', None) if inspect.ismethod(fn) else None
if owner is not None and hasattr(owner, '_outgoing_connections'):
try:
self._weak = weakref.WeakMethod(fn)
self._fn = None
owner._outgoing_connections.append(self)
except TypeError:
self._weak = None
self._fn = fn
else:
self._weak = None
self._fn = fn
self._connected = True
[docs]
def disconnect(self):
"""Disconnect this callback from the signal."""
if self._connected:
self._signal._callbacks[:] = [c for c in self._signal._callbacks if c is not self]
self._connected = False
[docs]
@property
def connected(self) -> bool:
return self._connected
[docs]
def __call__(self, *args, **kwargs):
if self._weak is not None:
method = self._weak()
if method is None:
self._connected = False
return None
return method(*args, **kwargs)
return self._fn(*args, **kwargs)
def _is_alive(self) -> bool:
"""True if the callback can still be invoked (used by Signal pruning)."""
if not self._connected:
return False
if self._weak is not None and self._weak() is None:
return False
return True
[docs]
def __bool__(self):
return True
[docs]
def __repr__(self):
target = self._fn if self._weak is None else self._weak()
return f"Connection({target!r}, connected={self._connected})"
# ============================================================================
# Signal
# ============================================================================
[docs]
class Signal:
"""Observable event dispatcher with optional type metadata.
Used as a class attribute, ``Signal`` is a non-data descriptor: each
instance accessing it lazily gets its own Signal copy stored in
``obj.__dict__``. Class-level access (``Cls.signal_name``) returns the
shared signal — useful for global event hubs like
``Node.script_error_raised``.
Example::
class Player(Node):
health_changed = Signal(int) # typed: emits one int
p1, p2 = Player(), Player()
p1.health_changed.connect(on_p1) # instance-scoped
p1.health_changed(50) # only p1 listeners fire
"""
__slots__ = ('_callbacks', '_types', '_name')
def __init__(self, *types: type):
self._callbacks: list[Connection] = []
self._types: tuple[type, ...] = types
self._name: str | None = None
[docs]
def __class_getitem__(cls, params) -> Signal:
"""Bracket syntax: ``Signal[int]``, ``Signal[int, str]``."""
if not isinstance(params, tuple):
params = (params,)
sig = cls.__new__(cls)
sig._callbacks = []
sig._types = params
sig._name = None
return sig
[docs]
def __set_name__(self, owner, name):
self._name = name
[docs]
def __get__(self, obj, objtype=None):
if obj is None or self._name is None:
return self
sig = obj.__dict__.get(self._name)
if sig is None:
sig = Signal.__new__(Signal)
sig._callbacks = []
sig._types = self._types
sig._name = self._name
obj.__dict__[self._name] = sig
return sig
def _validate_arity(self, fn: Callable) -> None:
"""Warn if *fn* cannot accept the number of args this signal emits."""
try:
import inspect
sig = inspect.signature(fn)
params = sig.parameters.values()
has_var_positional = any(p.kind == p.VAR_POSITIONAL for p in params)
if has_var_positional:
return
max_params = sum(
1 for p in params if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
)
n_types = len(self._types)
if max_params < n_types:
type_names = ", ".join(t.__name__ if isinstance(t, type) else repr(t) for t in self._types)
logging.getLogger(__name__).warning(
"Signal(%s) connected to %r which accepts at most %d args (signal emits %d)",
type_names, fn, max_params, n_types,
)
except (ValueError, TypeError):
pass # Can't inspect (builtin, etc.) — skip validation
[docs]
def connect(self, fn: Callable, *, once: bool = False) -> Connection:
"""Subscribe a callback. Returns a Connection handle.
Args:
fn: Callback to invoke on emit.
once: If True, auto-disconnect after first emit.
"""
if self._types:
self._validate_arity(fn)
if once:
original_fn = fn
conn_ref: list[Connection] = []
def _once_wrapper(*args, **kwargs):
original_fn(*args, **kwargs)
if conn_ref:
conn_ref[0].disconnect()
conn = Connection(self, _once_wrapper)
conn_ref.append(conn)
else:
conn = Connection(self, fn)
self._callbacks.append(conn)
return conn
[docs]
def disconnect(self, fn_or_conn):
"""Remove a previously connected callback or Connection."""
if isinstance(fn_or_conn, Connection):
fn_or_conn.disconnect()
return
# Match by stored fn for strong-ref connections, or by deref'd bound
# method for weak ones. Use == so bound methods compare correctly
# (bound methods create a new object on each attribute access).
def _target(c: Connection):
return c._weak() if c._weak is not None else c._fn
self._callbacks[:] = [c for c in self._callbacks if _target(c) != fn_or_conn]
[docs]
def __call__(self, *args, **kwargs):
"""Emit the signal, calling all connected callbacks with the given arguments.
Connections whose weak target was garbage-collected are skipped and
pruned from ``_callbacks`` after the dispatch loop.
"""
had_dead = False
for cb in self._callbacks[:]:
if cb._weak is not None and cb._weak() is None:
cb._connected = False
had_dead = True
continue
cb(*args, **kwargs)
if had_dead:
self._callbacks[:] = [c for c in self._callbacks if c._connected]
[docs]
def clear(self):
"""Remove all connected callbacks."""
self._callbacks.clear()
emit = __call__
[docs]
def __repr__(self):
if self._types:
type_names = ", ".join(t.__name__ if isinstance(t, type) else repr(t) for t in self._types)
return f"Signal({type_names}, connections={len(self._callbacks)})"
return f"Signal(connections={len(self._callbacks)})"
# ============================================================================
# Children — Smart child container
# ============================================================================
[docs]
class Children:
"""List-like container with named child access.
node.children[0] # by index
node.children['Camera'] # by name string
for c in node.children: # iteration
len(node.children) # count
"""
__slots__ = ('_list', '_names', '_snapshot', '_dirty')
def __init__(self):
self._list: list = []
self._names: dict[str, Any] = {}
self._snapshot: list = [] # cached copy for safe iteration
self._dirty: bool = False
def _add(self, node):
self._list.append(node)
if node.name:
self._names[node.name] = node
self._dirty = True
def _remove(self, node):
self._list.remove(node)
if node.name and self._names.get(node.name) is node:
del self._names[node.name]
self._dirty = True
[docs]
def safe_iter(self) -> list:
"""Return a snapshot safe for iteration during mutation. Avoids per-frame copy when children are unchanged."""
if self._dirty:
self._snapshot = list(self._list)
self._dirty = False
return self._snapshot
[docs]
def __getitem__(self, key):
if isinstance(key, int):
return self._list[key]
if isinstance(key, str):
if key in self._names:
return self._names[key]
raise KeyError(f"No child named '{key}'")
raise TypeError(f"Invalid key type: {type(key)}")
[docs]
def __iter__(self):
return iter(self._list)
[docs]
def __len__(self):
return len(self._list)
[docs]
def __contains__(self, item):
return item in self._list
[docs]
def __bool__(self):
return bool(self._list)
[docs]
def __repr__(self):
return f"Children({[c.name for c in self._list]})"