"""Lifecycle hook decorators for Node subclasses.
A node's per-frame, lifecycle, and input behaviour is defined by methods
named ``on_ready``, ``on_update``, ``on_fixed_update``, ``on_input``,
``on_enter_tree``, ``on_exit_tree``, ``on_draw``, ``on_gui_input`` and
``on_picked``. Subclasses override these directly to define one "primary"
handler per hook.
The decorators in this module mark *additional* methods (with arbitrary
names) as extra handlers for the same hooks, allowing concerns to be
split across methods without forcing one monolithic override::
class Player(Node):
@on_update
def update_animation(self, dt): # additional handler (fires after the override)
self._frame += dt * self.fps
def on_update(self, dt): # primary override (always fires first)
self.position += self.velocity * dt
@on_input("jump", consume=True) # an action's press
def jump(self, event):
self.velocity.y = -300 # claims the event: consume=True and no refusal
@on_input(Key.CTRL + Key.S) # a key, with a modifier held
def save(self, event):
self.tree.save_game()
A decorated handler is declared above a same-named override, or its
decorator is qualified (``@decorators.on_update``): in a class body a name
defined earlier wins, so ``@on_update`` below ``def on_update`` would call
the override rather than the decorator. ``tools/decorator_shadowing.py``
finds the mistake.
The decorators are bare-or-called: ``@on_update`` and ``@on_update()``
are both valid. ``@on_input(...)`` requires the call form when filters
are passed; bare ``@on_input`` is the all-events catch-all.
``@on_gui_input`` and ``@on_picked`` take the same filters for the targeted
phase, the input aimed at one node, and are run by that node rather than
broadcast.
All hook collection happens once in :meth:`Node.__init_subclass__`:
runtime dispatch is a tuple iteration, never a name lookup.
"""
from __future__ import annotations
import difflib
from collections.abc import Callable, Iterable
from dataclasses import replace
from enum import StrEnum
from typing import Any, Protocol, cast
from .input.controls import DeviceAxis, DeviceButton, DeviceControl, DeviceStick, DeviceType
from .input.enums import PointerPhase
from .input.events import (
AxisEvent,
ButtonEvent,
InputBinding,
InputEvent,
KeyEvent,
PickEvent,
PointerEvent,
ScrollEvent,
TextEvent,
)
_InputFilter = dict[str, Any]
class _HookedCallable(Protocol):
"""A callable stamped with a lifecycle-hook marker by the ``@on_<hook>`` decorators."""
_simvx_hook: str
def __call__(self, *args: Any, **kwargs: Any) -> Any: ...
class _InputHookedCallable(_HookedCallable, Protocol):
"""A callable stamped as an input handler, carrying its accumulated filter list."""
_simvx_input_filters: list[_InputFilter]
_LIFECYCLE_HOOK_NAMES = frozenset(
{
"ready",
"update",
"fixed_update",
"enter_tree",
"exit_tree",
"draw",
}
)
def _stamp_simple(fn: Callable[..., Any], hook_name: str) -> Callable[..., Any]:
"""Mark *fn* as an ``on_<hook_name>`` handler, refusing one already marked for another hook."""
existing_hook = getattr(fn, "_simvx_hook", None)
if existing_hook is not None and existing_hook != hook_name:
raise TypeError(
f"{getattr(fn, '__qualname__', fn)!r} is already an on_{existing_hook} handler and cannot also be "
f"an on_{hook_name} one: the two run at different times. Write a method for each"
)
cast(_HookedCallable, fn)._simvx_hook = hook_name
return fn
def _make_simple(hook_name: str) -> Callable[..., Any]:
"""Build a `@on_<hook>` decorator supporting bare and called forms."""
def decorator(*args: Any, **kwargs: Any) -> Any:
if kwargs:
raise TypeError(f"on_{hook_name} takes no keyword arguments")
if len(args) == 1 and callable(args[0]):
return _stamp_simple(args[0], hook_name)
if len(args) == 0:
return lambda fn: _stamp_simple(fn, hook_name)
raise TypeError(f"on_{hook_name} must decorate a function (use @on_{hook_name} or @on_{hook_name}())")
decorator.__name__ = f"on_{hook_name}"
decorator.__qualname__ = f"on_{hook_name}"
return decorator
on_ready = _make_simple("ready")
on_update = _make_simple("update")
on_fixed_update = _make_simple("fixed_update")
on_enter_tree = _make_simple("enter_tree")
on_exit_tree = _make_simple("exit_tree")
on_draw = _make_simple("draw")
# ---------------------------------------------------------------------------
# @on_input: unified filter decorator
# ---------------------------------------------------------------------------
#: Which ``released`` values each edge registers for.
_EDGE_RELEASED: dict[ButtonEdge, tuple[bool, ...]] = {
ButtonEdge.PRESS: (False,),
ButtonEdge.RELEASE: (True,),
ButtonEdge.BOTH: (False, True),
}
#: The pointer phases each edge is, for a pointer button: its press is DOWN and its release UP.
_EDGE_PHASES: dict[ButtonEdge, tuple[PointerPhase, ...]] = {
ButtonEdge.PRESS: (PointerPhase.DOWN,),
ButtonEdge.RELEASE: (PointerPhase.UP,),
ButtonEdge.BOTH: (PointerPhase.DOWN, PointerPhase.UP),
}
#: The device types whose buttons are pointer buttons, routed by phase.
_POINTER_TYPES = frozenset({DeviceType.MOUSE, DeviceType.PEN, DeviceType.TOUCH, DeviceType.POINTER})
_VALID_INPUT_KWARGS = frozenset({"edge", "phase", "exact", "consume", "capture", "samples", "repeat"})
#: Spellings that no longer exist, and the one that replaced each. An unknown
#: keyword raises anyway; naming the replacement is what makes the error useful.
_REPLACED_KWARGS = {
"action": '@on_input("jump"): the action name is the positional target',
"key": "@on_input(Key.S): the key is the positional target",
"button": "@on_input(Pointer.PRIMARY): the button is the positional target",
"joy_button": "@on_input(Controller.A): the button is the positional target",
"joy_axis": "@on_input(Controller.LEFT_X): the axis is the positional target",
"gesture": "@on_input(Touch.PINCH): a gesture is a Touch control, the positional target",
"released": 'edge="release" or edge="both"',
"motion": "@on_input(PointerEvent, phase=PointerPhase.MOVE)",
"scroll": "@on_input(ScrollEvent)",
**dict.fromkeys(
("ctrl", "shift", "alt", "meta"),
"@on_input(Key.CTRL + Key.S): a modifier is a held control, written into the target",
),
}
def _normalise_phases(value: Any, name: str) -> tuple[PointerPhase, ...]:
"""Validate ``phase=`` into a non-empty tuple of :class:`PointerPhase`; ``None`` is every phase."""
if value is None:
return tuple(PointerPhase)
values = tuple(value) if isinstance(value, tuple | list) else (value,)
phases = []
for item in values:
try:
phases.append(PointerPhase(item))
except ValueError:
raise TypeError(f"@{name}(phase=...) takes a PointerPhase or a tuple of them, got {item!r}") from None
if not phases:
raise TypeError(f"@{name}(phase=...) needs at least one PointerPhase")
return tuple(phases)
def _target_kind(target: object, name: str) -> str:
"""What one positional target is: an action, a button, an axis, an event family."""
if isinstance(target, type):
if not issubclass(target, InputEvent):
raise TypeError(f"@{name}'s event family must be an InputEvent subclass, got {target!r}")
return "event_class"
if isinstance(target, str) and not isinstance(target, StrEnum):
if not target:
raise TypeError(f"@{name} needs a non-empty action name")
return "action"
if isinstance(target, DeviceButton):
return "button"
if isinstance(target, InputBinding):
if not isinstance(target.control, DeviceButton):
raise TypeError(
f"@{name}({target!r}) holds controls while pressing a button, and {target.control!r} is not one"
)
return "button"
if isinstance(target, DeviceAxis):
if target.half or target.is_delta:
raise TypeError(
f"@{name}({target!r}) routes on a whole axis: filter on {target.whole!r} and read "
"event.value, or bind the half to an action and filter on that"
)
return "axis"
if isinstance(target, DeviceStick):
return "axis"
raise TypeError(
f"@{name}'s target is an action name, a control (Key.S, Pointer.PRIMARY, Controller.A, "
f"Controller.LEFT_STICK), a chord (Key.CTRL + Key.S) or an InputEvent subclass, got {target!r}"
)
def _refuse_unknown_kwargs(kwargs: dict[str, Any], valid: frozenset[str], name: str) -> None:
"""Raise naming each unknown keyword's replacement or nearest valid spelling."""
unknown = sorted(set(kwargs) - valid)
if not unknown:
return
hints = []
for key in unknown:
if key in _REPLACED_KWARGS:
hints.append(f"{key}= is now {_REPLACED_KWARGS[key]}")
continue
near = difflib.get_close_matches(key, valid, n=1)
if near:
hints.append(f"did you mean {near[0]}=?")
detail = "; ".join(hints) if hints else f"valid keywords are {sorted(valid)}"
raise TypeError(f"@{name} got unexpected keyword(s) {unknown}: {detail}")
def _classify_input_filter(target: Any, kwargs: dict[str, Any], name: str = "on_input") -> dict[str, Any]:
"""Validate one decoration and return the normalised filter it routes by.
One target per decoration, or a tuple of targets of one kind: a handler
that wants two unrelated inputs stacks two decorators, which keeps every
filter a set of routing keys. *name* is the decorator the filter belongs
to, for the messages.
"""
_refuse_unknown_kwargs(kwargs, _VALID_INPUT_KWARGS, name)
event_classes: tuple[type, ...] = ()
targets: tuple[Any, ...] = ()
kind: str | None = None
if target is not None:
targets = tuple(target) if isinstance(target, tuple | list) else (target,)
if not targets:
raise TypeError(f"@{name} needs at least one target in a tuple")
kinds = {_target_kind(t, name) for t in targets}
if len(kinds) > 1:
raise TypeError(f"@{name}'s targets are one kind of input, got {sorted(kinds)}: stack a decorator per kind")
kind = kinds.pop()
if kind == "event_class":
event_classes, targets, kind = targets, (), None
if InputEvent in event_classes and len(event_classes) > 1:
raise TypeError(f"@{name}((InputEvent, ...)) names every event already: write a bare @{name}")
edge_given = "edge" in kwargs
try:
edge = ButtonEdge(kwargs.get("edge", ButtonEdge.PRESS))
except ValueError:
choices = ", ".join(repr(e.value) for e in ButtonEdge)
raise TypeError(f"@{name}(edge=...) takes a ButtonEdge or one of {choices}, got {kwargs['edge']!r}") from None
if edge_given and kind not in ("action", "button"):
raise TypeError(
f"@{name}(edge=...) applies to an action or a button target; "
+ ("an axis has no press to take the edge of" if kind else "name one, or test event.pressed")
)
phases: tuple[PointerPhase, ...] = ()
if "phase" in kwargs:
if kind is not None or (event_classes and event_classes != (PointerEvent,)):
raise TypeError(
f"@{name}(phase=...) filters the PointerEvent family: write @{name}(PointerEvent, "
'phase=...), or name a pointer button and pass edge="release" for its release'
)
phases = _normalise_phases(kwargs["phase"], name)
kind = "pointer"
elif event_classes == (PointerEvent,):
phases = tuple(PointerPhase)
kind = "pointer"
# A button target is the control pressed; what it holds, and whether it
# is exact about modifiers, travels beside it as the binding it came from,
# read exactly as an action's binding is. A target that holds nothing and
# names no exactness constrains nothing, so dispatch skips the test.
exact = kwargs.get("exact")
if exact is not None and not isinstance(exact, bool):
raise TypeError(f"@{name}(exact=...) must be True or False; got {exact!r}")
if "exact" in kwargs and kind != "button":
raise TypeError(f"@{name}(exact=...) applies to a button target: an action's own bindings say what they hold")
chords: tuple[InputBinding | None, ...] = ()
if kind == "button":
bindings = [t if isinstance(t, InputBinding) else InputBinding(t) for t in targets]
if exact is not None:
for binding in bindings:
binding.exact = exact
targets = tuple(binding.control for binding in bindings)
chords = tuple(binding if binding.held or binding.exact is not None else None for binding in bindings)
repeat = kwargs.get("repeat", False)
if not isinstance(repeat, bool):
raise TypeError(f"@{name}(repeat=...) must be True or False; got {repeat!r}")
if "repeat" in kwargs:
if kind != "button" or any(button.device is not DeviceType.KEYBOARD for button in targets):
if not event_classes or kind == "pointer":
why = "An action opens once per physical press, and a pointer never repeats"
elif name == "on_input" and not kwargs.get("capture", False):
why = (
"An ordinary-phase event family never hears a repeat: name the key with repeat=True, or "
"hear every repeat with capture=True or @on_gui_input(KeyEvent) and test event.echo there"
)
else:
why = "An event family hears every event its phase delivers, repeats included, so test event.echo there"
raise TypeError(
f"@{name}(repeat=...) applies to a key target, Key.S or Key.CTRL + Key.S: only a held key "
f"auto-repeats. {why}"
)
if repeat and edge is ButtonEdge.RELEASE:
raise TypeError(
f'@{name}(repeat=True, edge="release"): an auto-repeat is a press, so a release handler '
'never hears one. Pass edge="both", or drop repeat='
)
consume = kwargs.get("consume", False)
if not isinstance(consume, bool):
raise TypeError(f"@{name}(consume=...) must be True or False; got {consume!r}")
capture = kwargs.get("capture", False)
if not isinstance(capture, bool):
raise TypeError(f"@{name}(capture=...) must be True or False; got {capture!r}")
if capture and kind == "action":
raise TypeError(
f"@{name}(capture=True) takes a device input (a key, a button, an axis or an event "
"family): the capture phase runs before any action resolves. Capture the key, "
f"@{name}(Key.F5, capture=True), or handle the action in the ordinary phase"
)
samples = kwargs.get("samples", False)
if not isinstance(samples, bool):
raise TypeError(f"@{name}(samples=...) must be True or False; got {samples!r}")
if samples:
if kind != "pointer" or PointerPhase.MOVE not in phases:
raise TypeError(
f"@{name}(samples=True) hears every motion sample of the PointerEvent family: write "
f"@{name}(PointerEvent, samples=True), or name phase=PointerPhase.MOVE among the phases"
)
if capture:
raise TypeError(f"@{name}(samples=True) runs in the ordinary phase: a sample is never claimed ahead")
if kind is None:
kind = "catch_all" if not event_classes or event_classes == (InputEvent,) else "event_class"
return {
"kind": kind,
"targets": targets,
"event_classes": event_classes if kind == "event_class" else (),
# Only the kinds with an edge carry one, so a filter holds exactly what it routes by.
"released": _EDGE_RELEASED[edge] if kind in ("action", "button") else (),
"phases": phases if kind == "pointer" else _EDGE_PHASES[edge] if kind == "button" else (),
"chords": chords,
"consume": consume,
"capture": capture,
"samples": samples,
"repeat": repeat,
}
def _stamp_input(fn: Callable[..., Any], filt: _InputFilter) -> Callable[..., Any]:
"""Attach an input-handler marker to *fn*, supporting stacked decorators."""
existing_hook = getattr(fn, "_simvx_hook", None)
if existing_hook is not None and existing_hook != "input":
raise TypeError(
f"{getattr(fn, '__qualname__', fn)!r} is already an on_{existing_hook} handler and cannot also be "
"an on_input one: the two receive different events. Write a method for each"
)
stamped = cast(_InputHookedCallable, fn)
stamped._simvx_hook = "input"
existing: list[_InputFilter] | None = getattr(fn, "_simvx_input_filters", None)
if existing is None:
stamped._simvx_input_filters = [filt]
else:
existing.append(filt)
return fn
# ---------------------------------------------------------------------------
# @on_gui_input and @on_picked: the targeted phase
# ---------------------------------------------------------------------------
#: The event classes the targeted GUI phase hands a node: ``GuiInputEvent``.
_GUI_EVENT_CLASSES: tuple[type[InputEvent], ...] = (PointerEvent, ScrollEvent, KeyEvent, TextEvent)
#: The keywords that belong to the broadcast alone, and what to write instead.
_BROADCAST_ONLY_KWARGS = {
"capture": (
"the capture phase is the broadcast's, root-first ahead of every control: use @on_input(..., capture=True)"
),
"samples": "motion samples reach the broadcast alone: use @on_input(PointerEvent, samples=True)",
}
#: What ``@on_picked`` takes: a pick has a button and its edge, and nothing else to route on.
_PICK_KWARGS = frozenset({"edge", "exact", "consume"})
def _refuse_action(target: Any, name: str) -> None:
"""Refuse an action name in a targeted filter: actions resolve after the targeted phase."""
targets = target if isinstance(target, tuple | list) else (target,)
for item in targets:
if isinstance(item, str) and not isinstance(item, StrEnum):
raise TypeError(
f"@{name}({item!r}): actions resolve after the targeted phase, from what no control "
f"claimed, so no targeted handler hears one. Name the control, @{name}(Key.ENTER), "
f"or handle the action in the broadcast, @on_input({item!r})"
)
def _classify_gui_filter(target: Any, kwargs: dict[str, Any]) -> _InputFilter:
"""Validate one ``@on_gui_input`` decoration: ``@on_input``'s grammar, less what never reaches a control."""
name = "on_gui_input"
for key, instead in _BROADCAST_ONLY_KWARGS.items():
if key in kwargs:
raise TypeError(f"@{name}({key}=...) does not apply to the targeted phase: {instead}")
_refuse_action(target, name)
filt = _classify_input_filter(target, kwargs, name)
kind = filt["kind"]
if kind == "axis":
raise TypeError(
f"@{name}({target!r}): an axis never reaches the targeted phase, which hands a control keys, "
f"text, pointers and the wheel. Handle it in the broadcast, @on_input({target!r})"
)
for button in filt["targets"]:
if button.device is not DeviceType.KEYBOARD and button.device not in _POINTER_TYPES:
raise TypeError(
f"@{name}({button!r}): the targeted phase hands a control keys, text, pointers and the "
f"wheel, and a {button.device.value} button is none of them. A pad drives focus through "
f"the ui_* actions; handle the button itself in the broadcast, @on_input({button!r})"
)
for event_class in filt["event_classes"]:
if event_class not in _GUI_EVENT_CLASSES:
raise TypeError(
f"@{name}({event_class.__name__}): the targeted phase hands a control a PointerEvent, "
f"ScrollEvent, KeyEvent or TextEvent; a {event_class.__name__} reaches @on_input alone"
)
return filt
def _classify_pick_filter(target: Any, kwargs: dict[str, Any]) -> _InputFilter:
"""Validate one ``@on_picked`` decoration: the pointer button the pick was cast for."""
name = "on_picked"
if "phase" in kwargs:
raise TypeError(
f"@{name}(phase=...) does not apply: a pick is cast for a pointer button's press or release "
f'and has no motion. Name the button and its edge, @{name}(Pointer.PRIMARY, edge="release")'
)
if "edge" in kwargs and target is None:
raise TypeError(
f'@{name}(edge=...) needs a pointer button to take the edge of, @{name}(Pointer.PRIMARY, edge="release"): '
f"a bare @{name} hears every press"
)
if "repeat" in kwargs:
raise TypeError(f"@{name}(repeat=...) does not apply: a pointer button never auto-repeats")
for key, instead in _BROADCAST_ONLY_KWARGS.items():
if key in kwargs:
raise TypeError(f"@{name}({key}=...) does not apply to a pick: {instead}")
_refuse_unknown_kwargs(kwargs, _PICK_KWARGS, name)
_refuse_action(target, name)
filt = _classify_input_filter(target, kwargs, name)
kind = filt["kind"]
if kind in ("event_class", "pointer"):
raise TypeError(f"@{name}({target!r}): every pick is a PickEvent, so a bare @{name} hears them all")
if kind == "axis":
raise TypeError(f"@{name}({target!r}): a pick is cast for a pointer button, never for an axis")
for button in filt["targets"]:
if button.device not in _POINTER_TYPES:
raise TypeError(
f"@{name}({button!r}): a pick is cast for a pointer button (Pointer.PRIMARY, "
f"Mouse.RIGHT), and a {button.device.value} button never casts one. A key held through "
f"the press is written into the target: Key.CTRL + Pointer.PRIMARY"
)
return filt
[docs]
def on_picked(*args: Any, **kwargs: Any) -> Any:
"""Mark a method as a handler for a 3D pick of this node.
The decorator form of :meth:`Node.on_picked`. A bare ``@on_picked`` hears
every press's pick; a pointer button narrows it to the picks cast for that
button, matched as ``@on_input`` matches a press, so a role hears every
pointer::
@on_picked(Pointer.PRIMARY) # any pointer's main press
@on_picked(Mouse.RIGHT) # the context-menu click
@on_picked((Mouse.MIDDLE, Pen.BARREL_1)) # any of them
@on_picked(Key.SHIFT + Pointer.PRIMARY) # with Shift held, and nothing else
@on_picked(Pointer.PRIMARY, edge="release") # the button coming up over this node
@on_picked # every press's pick
``edge=``, ``exact=`` and ``consume=`` mean what they mean on
``@on_input``. A button's pick hears its press by default; ``"release"``
hears the pick cast where the button comes up, and ``"both"`` either,
told apart by ``event.pressed``. A release casts a ray only while a node
in the tree declares such a handler, and only those handlers hear it: a
bare ``@on_picked`` and the :meth:`Node.on_picked` override hear presses
alone. A release belongs to whoever claimed its press, so when a node
claimed the press, a release pick's claim stops only this node's later
handlers. When nobody claimed the press, a release pick that claims stops
every later handler, as any claim does, so no broadcast or action handler
hears the release; the action the press opened still closes, and
``just_released`` and the physics tick report it.
``phase=`` is refused, as is anything that is not a pointer button.
The handler receives the :class:`~simvx.core.input.events.PickEvent` and
claims it by the rule every input handler answers to: ``True`` claims,
``False`` declines, and ``None`` defers to ``consume=``. For one pick the
decorated handlers run in declaration order, base classes first, until
one claims; the :meth:`Node.on_picked` override runs after them, and only
if none claimed.
"""
if len(args) == 1 and callable(args[0]) and not isinstance(args[0], type) and not kwargs:
return _stamp_targeted(args[0], "picked", _classify_pick_filter(None, {}))
if len(args) > 1:
raise TypeError("@on_picked takes one target, or a tuple of targets of one kind; stack decorators for more")
filt = _classify_pick_filter(args[0] if args else None, kwargs)
def wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
return _stamp_targeted(fn, "picked", filt)
return wrap
def _stamp_targeted(fn: Callable[..., Any], hook: str, filt: _InputFilter) -> Callable[..., Any]:
"""Attach a targeted-hook marker to *fn*, supporting stacked decorators of the same hook."""
existing_hook = getattr(fn, "_simvx_hook", None)
if existing_hook is not None and existing_hook != hook:
raise TypeError(
f"{getattr(fn, '__qualname__', fn)!r} is already an on_{existing_hook} handler and cannot also be "
f"an on_{hook} one: the two receive different events. Write a method for each"
)
stamped = cast(_InputHookedCallable, fn)
stamped._simvx_hook = hook
existing: list[_InputFilter] | None = getattr(fn, "_simvx_input_filters", None)
if existing is None:
stamped._simvx_input_filters = [filt]
else:
existing.append(filt)
return fn
def _naming(event: InputEvent, name: DeviceControl) -> InputEvent:
"""*event* as a handler filtered on *name* receives it: naming the control it asked for.
Every name a press resolves to fires, and each filtered handler is told
the name it asked for, so ``@on_input(Controller.A)`` and
``event.button == Controller.A`` inside it agree even while A is rebound
onto another button. A role is a group rather than a control, so a role's
handler receives the press as it came; so does an unfiltered handler.
A claim on the copy is carried back to the event by the dispatch.
"""
if isinstance(event, AxisEvent):
return event if event.axis is name or not isinstance(name, DeviceAxis) else replace(event, axis=name)
if not isinstance(event, PointerEvent | ButtonEvent | PickEvent) or not isinstance(name, DeviceButton):
return event
if event.button is name or name.device is DeviceType.POINTER:
return event
return replace(event, button=name)
#: One targeted handler's entry: declaration rank, method name, the chord it holds, its ``consume=``.
_TargetedEntry = tuple[int, str, InputBinding | None, bool]
[docs]
class TargetedHandlers:
"""One class's ``@on_gui_input`` or ``@on_picked`` handlers, indexed by the input each routes.
Built once per class at definition, so an event looks up only the
handlers registered for it, under the same routing keys the broadcast
tables use: a key and its edge, a key's auto-repeat, a pointer button (in
any of the forms a press is named by) and its phase, a phase alone, an
event class, and the catch-all. A pick is looked up as a pointer press or
release. ``picks_release`` says whether any handler hears a release pick,
which is what lets the router skip casting a ray for a release nobody asked for.
"""
__slots__ = ("keys", "repeats", "pointer", "classes", "catch_all", "picks_release")
def __init__(self, handlers: Iterable[tuple[str, _InputFilter]]) -> None:
keys: dict[tuple[DeviceButton, bool], list[_TargetedEntry]] = {}
repeats: dict[DeviceButton, list[_TargetedEntry]] = {}
pointer: dict[tuple[DeviceButton | None, PointerPhase], list[_TargetedEntry]] = {}
classes: dict[type, list[_TargetedEntry]] = {}
catch_all: list[_TargetedEntry] = []
for rank, (method_name, filt) in enumerate(handlers):
kind = filt["kind"]
consume = filt["consume"]
if kind == "button":
for button, chord in zip(filt["targets"], filt["chords"], strict=True):
entry = (rank, method_name, chord, consume)
if button.device is DeviceType.KEYBOARD:
for released in filt["released"]:
keys.setdefault((button, released), []).append(entry)
if filt["repeat"]:
repeats.setdefault(button, []).append(entry)
else:
for phase in filt["phases"]:
pointer.setdefault((button, phase), []).append(entry)
elif kind == "pointer":
for phase in filt["phases"]:
pointer.setdefault((None, phase), []).append((rank, method_name, None, consume))
elif kind == "event_class":
for event_class in filt["event_classes"]:
classes.setdefault(event_class, []).append((rank, method_name, None, consume))
else:
catch_all.append((rank, method_name, None, consume))
self.keys = {key: tuple(entries) for key, entries in keys.items()}
self.repeats = {key: tuple(entries) for key, entries in repeats.items()}
self.picks_release = any(phase is PointerPhase.UP for _button, phase in pointer)
self.pointer = {key: tuple(entries) for key, entries in pointer.items()}
self.classes = {key: tuple(entries) for key, entries in classes.items()}
self.catch_all = tuple(catch_all)
[docs]
def matching(
self, event: InputEvent, button_forms: Callable[[InputEvent], Iterable[DeviceButton]]
) -> list[tuple[_TargetedEntry, InputEvent]]:
"""The entries *event* reaches, each with the event as it is delivered, in declaration order.
*button_forms* names a pointer press in every form a filter can name
it by; it is asked only when a pointer-button filter exists. A handler
reached under several keys appears once per key, so the dispatch can
test each one's chord and run the handler under the first that holds.
A key's auto-repeat reaches only the key targets that asked for it
(``repeat=True``), and the families and catch-alls. A pick hears its
press as the ``DOWN`` phase and a release cast as ``UP``, so a button
filter hears the release only when its edge asks, and a release pick
reaches no catch-all.
"""
found: list[tuple[_TargetedEntry, InputEvent]] = []
if isinstance(event, KeyEvent):
entries = self.repeats.get(event.key) if event.echo else self.keys.get((event.key, not event.pressed))
if entries:
found.extend((entry, event) for entry in entries)
elif isinstance(event, PointerEvent | PickEvent):
pointer = self.pointer
if pointer:
phase = (
event.phase
if isinstance(event, PointerEvent)
else (PointerPhase.DOWN if event.pressed else PointerPhase.UP)
)
if event.button is not None:
for form in button_forms(event):
entries = pointer.get((form, phase))
if entries:
delivered = _naming(event, form)
found.extend((entry, delivered) for entry in entries)
entries = pointer.get((None, phase))
if entries:
found.extend((entry, event) for entry in entries)
entries = self.classes.get(type(event))
if entries:
found.extend((entry, event) for entry in entries)
if self.catch_all and not (isinstance(event, PickEvent) and not event.pressed):
found.extend((entry, event) for entry in self.catch_all)
if len(found) > 1:
found.sort(key=lambda pair: pair[0][0])
return found
[docs]
def collect_targeted(cls: type) -> tuple[TargetedHandlers | None, TargetedHandlers | None]:
"""Index the ``@on_gui_input`` and ``@on_picked`` handlers *cls* and its bases declare.
Declaration order across the MRO, base classes first, as
:func:`collect_hooks` gathers the rest. ``None`` for a hook the class
declares no decorated handler for, which is what keeps a node without
one off the targeted path's extra work entirely.
"""
gathered: dict[str, list[tuple[str, _InputFilter]]] = {"gui_input": [], "picked": []}
seen: set[tuple[str, str]] = set()
for base in reversed(cls.__mro__):
for name, attr in base.__dict__.items():
hook = getattr(attr, "_simvx_hook", None)
if hook not in gathered or (hook, name) in seen:
continue
seen.add((hook, name))
if name == f"on_{hook}":
raise TypeError(
f"{cls.__name__}.{name} is the on_{hook} override, the catch-all that runs after the "
f"decorated handlers; decorate a method with another name, or drop the decorator"
)
gathered[hook].extend((name, filt) for filt in getattr(attr, "_simvx_input_filters", ()))
gui, picked = gathered["gui_input"], gathered["picked"]
return (TargetedHandlers(gui) if gui else None, TargetedHandlers(picked) if picked else None)
[docs]
def collect_hooks(
cls: type, primary_methods: Iterable[str]
) -> tuple[dict[str, tuple[str, ...]], tuple[tuple[str, dict[str, Any]], ...]]:
"""Collect lifecycle and input handlers declared on *cls* and its bases.
Walks the MRO most-derived-last (matching ``Property.__set_name__``):
- Decorated methods (``@on_update`` etc.) are gathered in declaration
order across the MRO.
- Same-named primary overrides (``def on_update(self, dt):``) are
treated as implicit primary handlers and pinned to index 0 of their
bucket: users do not need to apply ``@on_update`` to their own
``on_update`` override for it to fire.
- An ``on_input`` override declaring no filter is the catch-all, exactly
as a bare ``@on_input`` is. Input handlers are routed by filter rather
than by bucket order, so it has no index to be pinned to.
Returns ``(hooks, input_handlers)``:
hooks: dict mapping hook name (e.g. ``"update"``) to an ordered
tuple of method names. The primary override, if present, is
always first.
input_handlers: tuple of ``(method_name, filter_dict)`` pairs in
declaration order across the MRO.
"""
hooks: dict[str, list[str]] = {}
input_handlers: list[tuple[str, dict[str, Any]]] = []
seen_input_methods: set[str] = set()
seen_hook_methods: dict[str, set[str]] = {}
primary_set = frozenset(primary_methods)
for base in reversed(cls.__mro__):
if base is object:
continue
for name, attr in base.__dict__.items():
hook = getattr(attr, "_simvx_hook", None)
# Treat same-named primary methods as implicit lifecycle handlers,
# even without an explicit decorator. e.g. `def on_update(self, dt)`.
if hook is None and name in primary_set and callable(attr):
hook = name[3:] # strip the "on_" prefix to get the bucket name
if hook is None or hook in ("gui_input", "picked"):
continue # the targeted hooks are indexed by collect_targeted
if hook == "input":
if name in seen_input_methods:
continue
seen_input_methods.add(name)
filters: Iterable[_InputFilter] = getattr(attr, "_simvx_input_filters", ())
# A method that reached this bucket without filters is an
# undecorated `def on_input(self, event)` override. It means the
# same thing a bare `@on_input` does, so give it the same
# catch-all filter rather than dropping it on the floor.
for filt in filters or (_classify_input_filter(None, {}),):
input_handlers.append((name, filt))
else:
bucket = hooks.setdefault(hook, [])
seen = seen_hook_methods.setdefault(hook, set())
if name in seen:
continue
seen.add(name)
bucket.append(name)
# Pin primary same-named overrides (on_update, on_ready, ...) to index 0.
for primary in primary_methods:
if not primary.startswith("on_"):
continue
hook_name = primary[3:]
ordered = hooks.get(hook_name)
if not ordered or primary not in ordered:
continue
if ordered[0] != primary:
ordered.remove(primary)
ordered.insert(0, primary)
frozen_hooks = {k: tuple(v) for k, v in hooks.items()}
return frozen_hooks, tuple(input_handlers)
__all__ = [
"on_ready",
"on_update",
"on_fixed_update",
"on_input",
"ButtonEdge",
"on_enter_tree",
"on_exit_tree",
"on_draw",
"on_picked",
"on_gui_input",
"collect_hooks",
"collect_targeted",
]