"""Descriptors, signals, enums, and type aliases used throughout the engine."""
from __future__ import annotations
import logging
from collections.abc import Callable, Generator
from enum import IntEnum, auto
from typing import TYPE_CHECKING, Any, NamedTuple
from .math.types import Vec2, Vec3
log = logging.getLogger(__name__)
if TYPE_CHECKING:
pass
Coroutine = Generator[None]
[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
@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."""
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.
Serialization:
``save_scene`` calls ``node.get_properties()`` and stores any value
that differs from *default*. ``load_scene`` passes 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', 'range', 'enum', 'hint', 'name', 'attr', 'link', '_propagate', 'group', 'on_change')
def __init__(
self,
default: Any,
*,
range=None,
enum=None,
hint="",
link=False,
propagate=False,
group="",
on_change: str | None = None,
):
"""Create an editor-visible property descriptor.
Args:
default: Default value for the property.
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).
"""
self.default = default
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
[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, 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)
try:
changed = old is not value and 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."""
__slots__ = ('_signal', '_fn', '_connected')
def __init__(self, signal: Signal, fn: Callable):
self._signal = signal
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
@property
def connected(self) -> bool:
return self._connected
[docs]
def __call__(self, *args, **kwargs):
return self._fn(*args, **kwargs)
[docs]
def __bool__(self):
return True
[docs]
def __repr__(self):
return f"Connection({self._fn!r}, connected={self._connected})"
# ============================================================================
# Signal
# ============================================================================
[docs]
class Signal:
"""Observable event dispatcher with optional type metadata.
Example::
health_changed = Signal(int) # typed: emits one int
damage = Signal(int, str) # typed: emits int and str
generic = Signal() # untyped: no arity checking
conn = health_changed.connect(lambda hp: print(f"HP: {hp}"))
health_changed(50) # prints "HP: 50"
conn.disconnect()
"""
__slots__ = ('_callbacks', '_types')
def __init__(self, *types: type):
self._callbacks: list[Connection] = []
self._types: tuple[type, ...] = types
[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
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()
else:
# Use != instead of `is not` so bound methods compare correctly
# (bound methods create a new object on each attribute access)
self._callbacks[:] = [c for c in self._callbacks if c._fn != fn_or_conn]
[docs]
def __call__(self, *args, **kwargs):
"""Emit the signal, calling all connected callbacks with the given arguments."""
for cb in self._callbacks[:]:
cb(*args, **kwargs)
[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]})"