"""Signal and Connection classes: observable event dispatcher.
Leaf module: it imports only ``diagnostics`` eagerly, and reads
``Node.dev_warnings`` through a deferred import. Used by Property descriptors,
Selection, UndoStack, EventBus, and any node-attached signal.
"""
import inspect
import types
import weakref
from collections.abc import Callable
from .diagnostics import warn_dev
# (callback code, signal types) pairs already reported by ``Signal._validate_arity``.
_arity_warned: set[tuple] = set()
[docs]
class Connection:
"""Handle returned by ``Signal.connect()``, which disconnects the callback it holds.
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")
_weak: weakref.WeakMethod[types.MethodType] | None
_fn: Callable | None
def __init__(self, signal: Signal, fn: Callable):
self._signal = signal
if isinstance(fn, types.MethodType) and hasattr(owner := fn.__self__, "_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
self._release_owner()
def _release_owner(self) -> None:
"""Drop this connection from its node's ``_outgoing_connections``."""
if self._weak is None or (method := self._weak()) is None:
return
owner = method.__self__
if hasattr(owner, "_outgoing_connections") and self in owner._outgoing_connections:
owner._outgoing_connections.remove(self)
def _target(self) -> Callable | None:
"""The connected callable, or None once a weakly held one is collected."""
return self._weak() if self._weak is not None else self._fn
[docs]
@property
def connected(self) -> bool:
return self._connected
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})"
[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
declaration itself, which no instance ever emits.
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.emit(50) # only p1 listeners fire
``emit`` is the one way to fire a signal. To forward one signal into
another, connect the other's ``emit``: ``a.connect(b.emit)``.
"""
__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 __set_name__(self, owner, name):
self._name = name
[docs]
def __get__(self, obj: object, objtype: type | None = None) -> Signal:
if obj is None or self._name is None:
return self
sig: Signal | None = 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 once if *fn* cannot accept the number of args this signal emits.
A development diagnostic: :meth:`connect` calls it only under
``Node.dev_warnings``. Reported once per callback function and signal
signature, attributed to the game's ``connect`` call.
"""
key = (getattr(getattr(fn, "__func__", fn), "__code__", fn), self._types)
if key in _arity_warned:
return
try:
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:
_arity_warned.add(key)
type_names = ", ".join(t.__name__ if isinstance(t, type) else repr(t) for t in self._types)
warn_dev(
f"Signal({type_names}) connected to {fn!r}, which accepts at most {max_params} "
f"positional args; the signal emits {n_types}. Give it a parameter per argument, "
"or *args to ignore them."
)
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. To forward into another signal,
pass its bound ``emit``; a ``Signal`` itself is refused.
once: If True, auto-disconnect after first emit.
Raises:
TypeError: ``fn`` is a ``Signal`` rather than a callable.
"""
if isinstance(fn, Signal):
source, target = self._name or "signal", fn._name or "other"
raise TypeError(
f"{source}.connect({target}): connect takes a callable, and a Signal is not one; "
f"to forward {source} into {target}, pass {target}.emit: {source}.connect({target}.emit)"
)
if self._types:
from .node import Node # node imports this module
if Node.dev_warnings:
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
# Matching with == lets a freshly bound method find its connection.
# Each match is marked disconnected as well as removed, so a removal
# made mid-emit takes effect before the emit reaches it.
for c in [c for c in self._callbacks if c._target() == fn_or_conn]:
c.disconnect()
def _connection_for(self, fn: Callable) -> Connection | None:
"""The live connection whose callable equals ``fn``, if any."""
for c in self._callbacks:
if c._connected and c._target() == fn:
return c
return None
[docs]
def emit(self, *args, **kwargs):
"""Emit the signal, calling all connected callbacks with the given arguments.
The loop runs over a snapshot, so a callback that connects another
callback mid-emit does not reach it in this emit, and one that
disconnects a later callback stops it at once: disconnecting removes
the connection from ``_callbacks`` and marks it, and the loop skips a
marked one. A connection whose weak target was garbage-collected is
marked as it is reached and pruned after the loop.
"""
dead = False
for cb in self._callbacks[:]:
if not cb._connected:
continue
weak = cb._weak
if weak is None:
cb._fn(*args, **kwargs) # type: ignore[misc]
continue
method = weak()
if method is None:
cb._connected = False
dead = True
continue
method(*args, **kwargs)
if dead:
self._callbacks[:] = [c for c in self._callbacks if c._connected]
[docs]
def clear(self):
"""Remove all connected callbacks."""
self._callbacks.clear()
[docs]
def disconnect_from_module(self, module_name: str) -> int:
"""Drop all connections whose callback is defined in ``module_name``.
Used by the hot-reload system to release references to old-module
code (lambdas, free functions, methods) before reload, so the old
module can be garbage-collected.
Returns the number of connections dropped.
"""
def _module_of(c: Connection) -> str | None:
target = c._target()
return getattr(target, "__module__", None) if target is not None else None
dropped = [c for c in self._callbacks if _module_of(c) == module_name]
for c in dropped:
c.disconnect()
return len(dropped)
[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)})"