Source code for simvx.core.descriptors

"""Descriptors, enums, and type aliases used throughout the engine.

``Signal`` and ``Connection`` live in ``simvx.core.signals``.
"""

import inspect
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


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(). The runtime drives coroutines with ``gen.send(dt)`` after the initial prime, so coroutines may receive the per-tick delta time via ``dt = yield``. Coroutines that yield without binding the value still work: the sent dt is simply discarded. """ __slots__ = ('_gen', '_cancelled', '_primed') def __init__(self, gen: Coroutine): self._gen = gen self._cancelled = False self._primed = 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', 'coalesce', 'persist', 'save_version', 'scalar', ) 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, coalesce: bool = False, persist: bool = False, save_version: int | None = None, scalar: bool = False, ): """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 fired during ``__init__`` are deferred until ``__init__`` returns, then dispatched once each (deduplicated by ``(property, hook)`` pair) so the hook always sees a fully constructed object. After construction, hooks fire synchronously inside ``__set__`` (unless ``coalesce=True``: see below). coalesce: When ``True`` *and* ``on_change`` is set, post-init hook calls fire at most once per scene-tree frame. Multiple writes within one tick collapse to a single deferred call drained at the end of :meth:`SceneTree.tick`: useful for expensive handlers like HUD rasterisation that don't care about intermediate values (Tower Defence: ``coins -= 1`` repeated five times in one frame rasterised the HUD text five times). The owning object must be attached to a SceneTree (``obj._tree`` set); for tree-less owners the flag is ignored and hooks fire synchronously. persist: When True, the value is included in ``SaveManager`` snapshots. save_version: Optional integer schema version recorded alongside the persisted value. scalar: When True, reject array-like values (numpy ndarray with ndim >= 1, Vec2/Vec3, list, tuple, dict, set) at assignment with a clear ``TypeError``. Plain Python numbers, numpy scalar types (``np.float32``, ``np.int64`` etc.), and 0-d numpy arrays are accepted. Use for properties that semantically represent a single number (e.g. ``Camera2D.zoom``) so a stray ``Vec2`` doesn't leak into downstream math and produce visual glitches. """ 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.coalesce = coalesce self.persist = persist self.save_version = save_version self.scalar = scalar
[docs] def __set_name__(self, owner, name): self.name = name self.attr = f"_{name}" if '__properties__' not in owner.__dict__: # Walk every base in MRO (most-derived-last) so mixin classes # that each declare Properties merge correctly. The previous # "first-base-wins" rule silently dropped Properties from # secondary mixins, which broke kwargs like `position=` on # diamond classes such as ``CollisionShape2D``. inherited: dict[str, Property] = {} for base in reversed(owner.__mro__[1:]): base_props = base.__dict__.get('__properties__') if base_props: inherited.update(base_props) owner.__properties__ = inherited owner.__properties__[name] = self # Class-level fast-path flag: classes with no on_change Properties # skip the __init__-scope queue entirely. if self.on_change is not None: owner._has_on_change_hooks = True
[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.scalar: # 0-d arrays and numpy scalar types pass; ndim >= 1 (including # Vec2/Vec3 which are ndarray subclasses) is rejected. shape = getattr(value, "shape", None) if shape is not None and shape != (): raise TypeError( f"{type(obj).__name__}.{self.name} must be a scalar " f"(got {type(value).__name__} with shape {shape}). " f"For per-axis non-uniform values, a separate API is needed." ) if isinstance(value, list | tuple | dict | set): raise TypeError( f"{type(obj).__name__}.{self.name} must be a scalar " f"(got {type(value).__name__})." ) 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. Hooks fired while ``__init__`` is running are # queued and dispatched after init returns (deduplicated), so user # code always sees a fully constructed object. Post-init, if # ``coalesce=True`` and the owner is attached to a SceneTree, the # hook is enqueued on the tree's ``_pending_coalesced_hooks`` set # and drained once at the end of :meth:`SceneTree.tick`: multiple # writes within one frame collapse to a single call. Tree-less # owners fall back to the synchronous path so pure-logic unit # tests keep working. if changed and self.on_change is not None: if getattr(obj, '_on_change_init', False): obj._on_change_pending.append((self.attr, self.on_change)) else: if self.coalesce: tree = getattr(obj, "_tree", None) if tree is not None: tree._pending_coalesced_hooks.add((obj, self.on_change)) return 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 try_decrement(self, obj, amount: float | int) -> bool: """Atomically decrement the property if it would not go negative. Returns ``True`` and assigns ``current - amount`` when the result is non-negative; returns ``False`` and leaves the value untouched otherwise. Resolves the "currency / mana / HP gate" pattern that every game-with-currency reinvents: Before:: if player.coins >= cost: player.coins -= cost purchase() After:: if Player.coins.try_decrement(player, cost): purchase() Reads-via-``__get__`` so parent-link offsets are respected; writes the raw stored value (no parent contribution) via ``__set__`` so any ``on_change`` hook and ``range``/``enum`` validation still fire. ``amount`` must be numeric and non-negative. """ if not isinstance(amount, int | float): raise TypeError(f"Property.try_decrement amount must be numeric, got {type(amount).__name__}") if amount < 0: raise ValueError(f"Property.try_decrement amount must be >= 0, got {amount}") current = self.__get__(obj) if not isinstance(current, int | float): raise TypeError( f"Property.try_decrement requires a numeric Property; " f"{self.name!r} is {type(current).__name__}" ) if current < amount: return False self.__set__(obj, current - amount) return True
[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
# ============================================================================ # 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 move_first(self, node) -> None: """Move ``node`` to index 0 (drawn first, hit-tested last). No-op if absent.""" if node not in self._list: return self._list.remove(node) self._list.insert(0, node) self._dirty = True
[docs] def move_last(self, node) -> None: """Move ``node`` to the end (drawn last, hit-tested first). No-op if absent.""" if node not in self._list: return self._list.remove(node) self._list.append(node) self._dirty = True
[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]})"