"""InputMap: the flattened result of the active input contexts, plus the override layer.
``InputMap`` is not a registry a game writes into and reads back. It is what
the active :class:`~.contexts.InputContext` stack currently means: the
contexts are flattened into one binding map when the stack changes, and every
query reads that map rather than walking the stack.
Two things write bindings and they are not the same thing:
- **A game declares** them, on a node's ``input_context`` (a plain dict, or an
:class:`~.contexts.InputContext` with a name or ``hides``), or in the
project's settings. :meth:`_InputMap.add_action` declares on the tree's own
context, for code that builds its entries as it runs.
- **A player rebinds** them, which writes the **override layer**:
:meth:`_InputMap.rebind`. An override is persistent, survives every rebuild
of the stack, and is reapplied to each flatten, so a rebound key stays
rebound across a scene change, a pause and a vehicle being entered.
:meth:`_InputMap.export` and :meth:`_InputMap.load` persist it.
"""
from __future__ import annotations
import contextvars
import difflib
import logging
import weakref
from collections.abc import Callable, Mapping
from contextlib import contextmanager
from dataclasses import dataclass, replace
from enum import StrEnum
from typing import TYPE_CHECKING, Any, cast
from ..signals import Signal
from .actions import Action, ActionKind, Axis1D, Axis2D, Source, SourceSpec, to_source
from .contexts import _DECLARED_NAMES, InputContext, _RootContext
from .controls import DeviceAxis, DeviceButton, DeviceControl, DeviceType, Key, name_to_keys
from .engine import ENGINE_CONTEXT
from .events import InputBinding, key_combo_to_binding
from .gestures import recognised_in
from .spelling import source_from_data, source_to_data
if TYPE_CHECKING:
from ..node import Node
from .listener import ControlInput
from .state import _Input
[docs]
def to_binding(b: InputBinding | DeviceButton | DeviceAxis | str) -> InputBinding:
"""Coerce one written or bare binding spelling into an :class:`InputBinding`.
The one place a binding spelling is understood, so what an ``InputMap``
accepts is also what an ``Axis1D`` component, a ``simvx.toml`` and a rebind
file accept. A control is its own binding; a written control is its
``str()``, ``"mouse:right"``, ``"tablet:button(2)"`` or
``"pointer:secondary"``; a key may also be written by its combo name,
``"space"`` or ``"shift+tab"``.
"""
if isinstance(b, InputBinding):
return b
if isinstance(b, DeviceButton | DeviceAxis):
return InputBinding(b)
if isinstance(b, str):
# Try name_to_keys lookup first (handles "space", "escape", etc.)
keys = name_to_keys(b)
if keys:
return InputBinding(keys[0])
# A key's constant name ("LEFT_SHIFT").
found = getattr(Key, b.upper(), None)
if isinstance(found, DeviceButton):
return InputBinding(found)
# A control's own spelling: a device type, then a name or an index.
if ":" in b:
control = DeviceControl.parse(b)
if not isinstance(control, DeviceButton | DeviceAxis):
raise ValueError(f"{b!r} is a stick: bind it as a whole, not through a written binding")
return InputBinding(control)
# Last: a modifier combo, e.g. "shift+tab" or "ctrl+alt+delete", so a
# key whose own name contains "+" still resolves to itself first.
combo = key_combo_to_binding(b)
if combo is not None:
return combo
raise ValueError(f"Cannot resolve input binding from string {b!r}")
raise TypeError(f"Cannot create InputBinding from {type(b).__name__}")
log = logging.getLogger(__name__)
#: The persisted override layer's format version, written by :meth:`_InputMap.export`.
REBIND_FORMAT_VERSION = 6
[docs]
class BindingRole(StrEnum):
"""Which part of one source a rebind replaces.
An action's value comes from its sources, and a source is a whole value on
its own: a button, an :class:`~.actions.Axis1D` with two halves, or an
:class:`~.actions.Axis2D` with four directions. A rebind screen shows one
row per part, and this is what that row is called.
"""
#: The source itself, which is a plain binding: a button action's key.
WHOLE = "whole"
#: An :class:`~.actions.Axis1D`'s two halves.
LOW = "low"
HIGH = "high"
#: An :class:`~.actions.Axis2D`'s four directions.
UP = "up"
DOWN = "down"
LEFT = "left"
RIGHT = "right"
_ROLES_1D = (BindingRole.LOW, BindingRole.HIGH)
_ROLES_2D = (BindingRole.UP, BindingRole.DOWN, BindingRole.LEFT, BindingRole.RIGHT)
[docs]
def roles_of(source: Source) -> tuple[BindingRole, ...]:
"""The parts of *source* a player can rebind separately."""
if isinstance(source, Axis2D):
return _ROLES_2D
if isinstance(source, Axis1D):
return _ROLES_1D
return (BindingRole.WHOLE,)
[docs]
@dataclass(frozen=True, slots=True)
class ActionExplanation:
"""What :meth:`_InputMap.explain` finds out about one name.
Attributes:
name: The name asked about.
declared: Whether anything declares it: a context, the project or the engine.
active: Whether it can fire right now.
kind: Its type, or None for a name nothing declares.
sources: Its sources as declared, rebinds applied; a hidden one's as
it will read when it comes back.
bindings: What those sources resolve to, device binding by device binding.
context: The declaration in force, or the one it will come back as
while hidden: a node's context, the tree's, the project's or the engine's.
node: The node whose context that is, or None for the tree, the
project and the engine.
hidden: Whether a context above takes it away right now.
hidden_by: The context doing so, or None.
shadowed: The declarations beneath the one in force, which it overrides.
rebinds: The player's overrides that apply to it.
"""
name: str
declared: bool
active: bool
kind: ActionKind | None
sources: tuple[Source, ...]
bindings: tuple[InputBinding, ...]
context: InputContext | None
node: Node | None
hidden: bool
hidden_by: InputContext | None
shadowed: tuple[InputContext, ...]
rebinds: tuple[Rebind, ...]
[docs]
@dataclass(frozen=True, slots=True)
class Rebind:
"""One override: what a player put in place of what the game declared.
Addressed by ``(action, slot, role)``, which is exactly the row a rebind
screen draws. *slot* is the position in the action's source list, so
rebinding the keyboard half of a movement action leaves the stick alone,
and *role* is the part within that source, so rebinding one direction of a
movement action leaves the other three.
An override is reapplied to every flatten. Where the active contexts no
longer define that slot and role, it is inert rather than dropped, so
entering a vehicle and leaving it again gives the rebound key back.
Attributes:
action: Which entry: an action's or a device alias's name, or an
engine entry such as ``Controller.A``.
binding: What to put there (a binding, or another entry's name), or
``None`` to leave the part unbound.
slot: Which source of the action, in declaration order.
role: Which part of that source.
player: Whose rebind it is, or ``None`` for everybody's. A
player-scoped override gives that player their own copy of the
action and leaves every other player reading the declared one.
"""
action: str | DeviceControl
binding: InputBinding | str | None
slot: int = 0
role: BindingRole = BindingRole.WHOLE
player: int | None = None
[docs]
@property
def key(self) -> tuple[str | DeviceControl, int, BindingRole, int | None]:
"""What this override is addressed by, which is what replaces an earlier one."""
return (self.action, self.slot, self.role, self.player)
def _keeping_units(old: object, new: InputBinding | str) -> InputBinding | str:
"""*new* in the units the binding it replaces was in: a rebind moves the control, not the scale or the rate.
A new binding that states units of its own keeps them.
"""
if not isinstance(old, InputBinding) or not isinstance(new, InputBinding):
return new
if new.scale != 1.0 or new.rate is not None:
return new
return replace(new, scale=old.scale, rate=old.rate)
def _apply_role(source: Source, role: BindingRole, binding: InputBinding | str | None) -> Source | None:
"""*source* with *role* replaced by *binding*, or None where it does not have that role."""
if role is BindingRole.WHOLE:
if not isinstance(source, InputBinding | str):
return None
# A bare binding is the unbound state: it names no device and reads at
# rest everywhere, so clearing a part needs no second spelling.
return _keeping_units(source, binding) if binding is not None else InputBinding()
old = getattr(source, role.value, None)
part: dict[str, Any] = {role.value: _keeping_units(old, binding) if binding is not None else InputBinding()}
if isinstance(source, Axis1D) and role in _ROLES_1D:
return replace(source, **part)
if isinstance(source, Axis2D) and role in _ROLES_2D:
return replace(source, **part)
return None
class _InputMap:
"""The flattened active contexts, and the override layer over them.
One per scene tree. Read it for what an action is bound to right now;
write to it to declare a root action (:meth:`add_action`) or to record a
player's rebind (:meth:`rebind`).
**Layers, bottom to top:** the engine's own entries (``Controller.A`` and the
rest, implicit), the project's entries from its settings (shared by every
tree and read live, so the editor's input-map dialog reaches a game it is
hosting), then the tree's contexts (the imperative :meth:`add_action`
context first, then every node's ``input_context`` in activation order),
then the player's rebinds. A hosted tree built on its host reads the
host's map in place of the project layer, which already carries it.
"""
def __init__(self, base: _InputMap | None = None):
#: The map underneath this one, whose entries this one is laid over:
#: a hosted tree's host. ``None`` for a tree that starts from the
#: engine's and the project's entries alone, which is the default.
self._base = base
#: The maps laid over this one, weakly, so a registration here
#: re-flattens them. A tree that has gone does not hold its map here.
self._derived: weakref.WeakSet[_InputMap] = weakref.WeakSet()
if base is not None:
base._derived.add(self)
#: The base of the stack: what :meth:`add_action` declares, for the
#: tree's lifetime, never deactivated and never hidden below itself.
self._root = _RootContext()
#: Contexts bottom to top, each with the node that holds it. The base
#: has no node, which is what makes it the floor.
self._stack: list[tuple[InputContext, Node | None]] = [(self._root, None)]
#: Which of those the last flatten counted: an entry is False while
#: its node cannot receive input. Kept so a switch that deafens a node
#: can tell whether the map has to be rebuilt at all.
self._active: tuple[bool, ...] = (True,)
#: The flattened result. Every action query reads this and nothing
#: walks the stack.
self._actions: dict[str, Action] = {}
#: The actions a context hides, as they will read when they come back.
self._hidden: dict[str, Action] = {}
#: The same three tables before their names were expanded: what the
#: contexts and the overrides say, which is what a rebind screen shows
#: and what a map laid over this one builds on.
self._declared: dict[str, Action] = {}
self._hidden_declared: dict[str, Action] = {}
self._remaps_declared: dict[DeviceControl, Action] = {}
#: Engine entries that no longer resolve to their default raw control,
#: each to the raw control it resolves to now (None for unbound), and
#: the other way round. Empty is the usual case and the fast path: the
#: resolver renames a press only when something here says to.
self._redirects: dict[DeviceControl, DeviceControl | None] = {}
self._redirected_from: dict[DeviceControl, tuple[DeviceControl, ...]] = {}
#: Engine entries a player defined for one device, by its id: the names
#: an unrecognised controller has once they are mapped, and a per-pad
#: fix on a recognised one. ``Controller.A`` on that unit resolves here
#: first, everywhere else as the map says.
self._unit_redirects: dict[int, dict[DeviceControl, DeviceControl]] = {}
self._unit_redirected_from: dict[int, dict[DeviceControl, tuple[DeviceControl, ...]]] = {}
#: Which context each flattened name came from, which is what an
#: ``ActionEvent`` reports and what says who shadowed whom.
self._origin: dict[str, InputContext] = {}
# The continuous actions' names, kept alongside the map so the resolve
# phase visits only those: it runs once per event, and a game with no
# analogue action then pays one empty loop rather than a scan of every
# action it declared.
self._continuous: list[str] = []
#: Every name any context has declared while it was active, mapped to
#: the type that declaration gave it. The names are what separates
#: "inactive right now" from "not a thing": a query for the first
#: answers at rest, one for the second raises. The kind is what makes
#: that rest value the action's own type, so a hidden movement action
#: reads as a zero vector rather than as False.
#:
#: **Where two declarations of one name disagree about the type, the
#: topmost live one on the stack decides**, and a name the stack no
#: longer declares at all keeps the type it last had. So the rest value
#: is the type the action will have when it comes back, and it follows
#: the stack rather than the order the contexts happened to be visited.
self._known: dict[str, ActionKind] = {}
#: The override layer, addressed by ``Rebind.key``.
self._overrides: dict[tuple[str | DeviceControl, int, BindingRole, int | None], Rebind] = {}
#: Per-player copies of an action, built only where a player-scoped
#: override exists. Empty is the fast path and the usual case.
self._player_actions: dict[int, dict[str, Action]] = {}
#: The stack index of the topmost ``hides=ALL`` context, or 0 for
#: none. A broadcast handler whose nearest context sits below this is
#: not reached (see ``SceneTree._dispatch_input``).
self._hide_floor: int = 0
#: Bumped on every flatten, so a cached answer about the stack knows
#: whether it is still good.
self._generation: int = 0
#: The recognised controls (a gesture, the wheel) some binding names,
#: which is what turns recognising them on.
self._recognised: frozenset[DeviceControl] = frozenset()
#: Fires after the flattened bindings change: a rebind, a context
#: arriving or leaving, an action added. What a prompt listens to.
self.bindings_changed = Signal()
#: The state these bindings resolve against. Set by the scene tree
#: that owns both; the release-on-change pass needs it.
self._state: _Input | None = None
#: Pad sections of a loaded rebind file whose guid no pad currently
#: reports. Kept so an export round-trips them, and resolved if that
#: controller is plugged in later.
#: Device sections of a loaded rebind file whose device is not
#: connected: pinned bindings that wait for it, and round-trip meanwhile.
self._pending_devices: dict[str, list[dict[str, Any]]] = {}
self._watching_devices = False
_project.maps.add(self)
# Usable the moment it exists: the project's (or the host's) entries
# are laid now rather than waiting for the first declaration.
self._flatten()
@staticmethod
def _check_alone(context: InputContext) -> None:
"""Refuse *context* where its entries cannot resolve among themselves: the project layer's check."""
_expand(context._actions, {}, context._actions, {})
def _declared_where(self) -> dict[str, str]:
"""Which context declares each name on the stack, for a refusal that names it."""
where: dict[str, str] = {}
for index, (context, _node) in enumerate(self._stack):
label = context.name or f"#{index}"
for name in context._actions:
where[name] = label
return where
# -- The context stack ------------------------------------------------
@property
def context_stack(self) -> tuple[InputContext, ...]:
"""The active contexts, bottom to top. The first is always the root's.
A context whose node cannot receive input is not here: it is still
held for the node's lifetime, so it comes back when the node wakes,
but while the node is deaf the context declares nothing and hides
nothing.
"""
return tuple(context for (context, _node), live in zip(self._stack, self._active, strict=True) if live)
def _activate(self, context: InputContext, node: Node | None) -> None:
"""Push *context* on top of the stack. Called when *node* enters the tree.
Refused, and taken off again, where what it declares cannot be
resolved against what is already here: a name it uses that nothing
declares, a loop, or a type that does not fit.
"""
def undo() -> None:
if self._stack and self._stack[-1] == (context, node):
del self._stack[-1]
self._reindex()
with self._changing(undo):
self._stack.append((context, node))
self._reindex()
def _deactivate(self, context: InputContext, node: Node | None) -> None:
"""Take *context* off the stack. Called when *node* leaves it."""
for index in range(len(self._stack) - 1, 0, -1):
if self._stack[index] == (context, node):
del self._stack[index]
self._reindex()
self._flatten()
return
def _refresh_active(self) -> None:
"""Re-flatten when a node going deaf or waking changed which contexts count.
A context is live while its node can receive input, which the node's
``update_mode``, the tree's ``paused`` and a script error all move
without anything entering or leaving the tree. Those switches call
here rather than the flatten walking the tree per frame: the cost is
one cached read per context, and a tree whose only context is the
root's pays a length check.
"""
if len(self._stack) <= 1:
return
if self._live_mask() != self._active:
self._flatten()
def _live_mask(self) -> tuple[bool, ...]:
"""Which stack entries can hear input right now. The base always can."""
return tuple(node is None or node._can_receive_input() for _context, node in self._stack)
def _reindex(self) -> None:
"""Record each context-holding node's stack index on the node itself.
A handler's rank under a hiding context is its nearest context-holding
ancestor's index, and walking up to find it is O(depth) only while
something is hiding. Writing the index here keeps that walk from also
having to search the stack.
"""
for index, (_context, node) in enumerate(self._stack):
if node is not None:
node._input_context_index = index
# -- The flatten ------------------------------------------------------
def _flatten(self) -> None:
"""Rebuild the map from the stack, reapply the overrides, and expand every name.
Bottom to top: each context first takes what it hides and then adds
what it declares, so a name in two contexts resolves to the higher one
and ``hides`` reaches only what is below.
A context whose node cannot receive input is skipped: the switch that
stops a node processing is the switch that deafens it, and a deaf
node's context declares nothing and hides nothing. It keeps its place
in the stack, so waking the node puts it back where it was rather than
on top of whatever arrived meanwhile.
Then the names are expanded: an entry naming another takes that
entry's sources, and an engine entry redefined by a game or rebound
by a player is resolved to the raw control it now stands for. The
resolver reads the expanded table and never follows a name.
An entry names only entries at or beneath its own context, so a change
that takes a context away can orphan nothing and never fails. A change
that adds one (a context arriving, a declaration, a rebind) is refused
on a cycle, a name the entry cannot see or a type mismatch, and the
caller undoes it.
Measured against a stack far deeper than a game builds, this is a few
tens of microseconds, which is why it is a rebuild rather than an
incremental edit: a game opening and closing a context every frame
pays less for the whole rebuild than the bookkeeping an incremental
one would need.
"""
previous = self._actions
previous_redirects = self._redirects
# The player's shared rebinds, folded into an entry as it arrives, so
# the entry every level resolves against is already the rebound one.
shared: dict[str, list[Rebind]] = {}
for override in self._overrides.values():
if isinstance(override.action, str) and override.player is None:
shared.setdefault(override.action, []).append(override)
def fold(action: Action) -> Action:
for override in shared.get(str(action.name), ()):
edited = self._rebound(action, override)
if edited is not None:
action = edited
return action
flat: dict[str, Action] = {}
remaps: dict[DeviceControl, Action] = {}
origin: dict[str, InputContext] = {}
# What a context above took away, as it will read when it comes back.
hidden: dict[str, Action] = {}
# What an entry may name: the entries declared at its own level or
# beneath it, deaf contexts included, since a paused context is still
# there to come back. Each entry keeps the pool it was declared into.
resolvable: dict[str, Action] = {}
pools: dict[int, dict[str, Action]] = {}
known = dict(self._known)
# The engine's own entries, at the bottom of every map; a hidden name
# the engine declares comes back as the engine declared it.
engine = {name: fold(action) for name, action in ENGINE_CONTEXT._actions.items()}
base = self._base
if base is not None:
# The host's map, read live: its entries (the project's among them)
# are here until a context above takes or shadows them.
floor = {name: fold(action) for name, action in base._declared.items()}
remaps.update(base._remaps_declared)
origin.update(dict.fromkeys(floor, base._root))
known.update(base._known)
else:
# The project's entries, from its settings: the layer under every
# tree's own, read live.
project = _project.context
floor = dict(engine)
origin.update(dict.fromkeys(engine, ENGINE_CONTEXT))
known.update((name, action.kind) for name, action in engine.items())
floor.update((name, fold(action)) for name, action in project._actions.items())
remaps.update(project._remaps)
origin.update(dict.fromkeys(project._actions, project))
known.update((name, action.kind) for name, action in project._actions.items())
flat.update(floor)
resolvable.update(floor)
floor_pool = dict(resolvable)
for action in floor.values():
pools[id(action)] = floor_pool
hide_floor = 0
active = self._live_mask()
for index, (context, _node) in enumerate(self._stack):
arriving = {name: fold(action) for name, action in context._actions.items()}
if not active[index]:
# On the stack and not heard from. Its names are still things
# this tree has a context for, so a query for one answers at
# rest rather than raising at a typo it is not, and an entry
# above may still name them. Its types give way to any live
# declaration of the same name.
for name, action in context._actions.items():
known.setdefault(name, action.kind)
resolvable.update(arriving)
level = dict(resolvable)
for action in arriving.values():
pools[id(action)] = level
continue
if context._hides_all:
hide_floor = index
hidden.update((name, action) for name, action in flat.items() if action is not engine.get(name))
flat = dict(engine)
origin = dict.fromkeys(engine, ENGINE_CONTEXT)
resolvable = dict(engine)
# A game's redefinitions of engine entries are the game's, and
# go with the rest; the engine entries under them stay.
remaps = {}
elif context._hides:
for name in context._hides:
if name in flat and flat[name] is not engine.get(name):
hidden[name] = flat.pop(name)
origin.pop(name, None)
resolvable.pop(name, None)
if name in engine:
flat[name] = resolvable[name] = engine[name]
origin[name] = ENGINE_CONTEXT
flat.update(arriving)
resolvable.update(arriving)
level = dict(resolvable)
for action in arriving.values():
pools[id(action)] = level
remaps.update(context._remaps)
for name in context._actions:
hidden.pop(name, None)
origin.update(dict.fromkeys(context._actions, context))
# Recorded from the context rather than from the flattened result,
# so a name a context above hides keeps the type the context that
# declared it gave it; bottom to top, so the topmost declaration
# wins where two disagree.
for name, action in context._actions.items():
known[name] = action.kind
players: dict[int, dict[str, Action]] = {}
unit_remaps: dict[int, dict[DeviceControl, Action]] = {}
if self._overrides:
self._apply_overrides(flat, hidden, remaps, players, unit_remaps)
redirects, redirected_from = _resolve_redirects(remaps)
unit_redirects: dict[int, dict[DeviceControl, DeviceControl]] = {}
unit_redirected_from: dict[int, dict[DeviceControl, tuple[DeviceControl, ...]]] = {}
for unit, own in unit_remaps.items():
forward, reverse = _resolve_unit_redirects(own)
unit_redirects[unit], unit_redirected_from[unit] = forward, reverse
declared_where = self._declared_where()
expanded = _expand(flat, pools, flat, declared_where)
# A hidden entry answers what it will read when it comes back, from the
# level it was declared at.
shelved = _expand(hidden, pools, {**hidden, **flat}, declared_where) if hidden else {}
player_actions = {
player: _expand(own, pools, {**flat, **own}, declared_where) for player, own in players.items()
}
for name, action in expanded.items():
known[name] = action.kind
# Nothing above can raise from here on, so the map changes as a whole.
self._known = known
self._active = active
self._origin = origin
self._hide_floor = hide_floor
self._declared = flat
self._hidden_declared = hidden
self._remaps_declared = remaps
self._actions = expanded
self._hidden = shelved
self._player_actions = player_actions
self._redirects = redirects
self._redirected_from = redirected_from
unit_changed = unit_redirects != self._unit_redirects
self._unit_redirects = unit_redirects
self._unit_redirected_from = unit_redirected_from
self._continuous = [name for name, action in self._actions.items() if action.kind is not ActionKind.BUTTON]
self._recognised = frozenset(
found
for actions in (expanded, *player_actions.values())
for action in actions.values()
for b in action.bindings
for control in (b.control, *b.held)
for found in recognised_in(control)
)
self._generation += 1
self._settle(previous, redirects_changed=redirects != previous_redirects or unit_changed)
self.bindings_changed.emit()
for derived in tuple(self._derived):
# What this map resolves to is the floor of theirs, so a change
# here is a change there. One level deep: a tree's map is nobody's
# base.
derived._flatten()
@contextmanager
def _changing(self, undo: Callable[[], None]):
"""Run one change that adds to the map, undoing it if the map it makes is refused.
The change is made, the map is flattened, and a refusal
(a cycle, an unknown name, a type mismatch) undoes the change and
re-raises, so a map is never left holding a configuration it refused.
"""
try:
yield
self._flatten()
except Exception:
undo()
self._flatten()
raise
def _apply_overrides(
self,
flat: dict[str, Action],
hidden: dict[str, Action],
remaps: dict[DeviceControl, Action],
players: dict[int, dict[str, Action]],
unit_remaps: dict[int, dict[DeviceControl, Action]],
) -> None:
"""Lay the override layer over the freshly flattened tables, in place.
An override that names a player gets that player their own copy of the
action, so two people can hold different keys for the same name; one
that names nobody edits the shared copy. An override of an engine
entry redefines it for everyone, whether or not a game redefined it
first, and a clean branch keeps it: a pad with two buttons the wrong
way round is the player's hardware wherever they are in the game. One
pinned to a unit, ``Controller.A`` on one pad, defines the entry for
that device alone.
"""
for override in self._overrides.values():
key = override.action
if isinstance(key, DeviceControl) and key.unit is not None:
if isinstance(override.binding, InputBinding) and override.binding.control is not None:
unit_remaps.setdefault(key.unit, {})[key.on(None)] = Action(
key.on(None), sources=[InputBinding(override.binding.control.on(None))]
)
continue
if isinstance(key, DeviceControl):
current = remaps.get(key) or Action(key, sources=[InputBinding(cast(DeviceButton | DeviceAxis, key))])
edited = self._rebound(current, override)
if edited is not None:
remaps[key] = edited
continue
if override.player is None:
continue # folded into the entry as it arrived, hidden or not
declared = flat.get(key)
if declared is None:
continue
table = players.setdefault(override.player, {})
edited = self._rebound(table.get(key, declared), override)
if edited is not None:
table[key] = edited
@staticmethod
def _rebound(action: Action, override: Rebind) -> Action | None:
"""*action* with one part replaced, or None where it has no such part."""
if not 0 <= override.slot < len(action.sources):
return None
source = _apply_role(action.sources[override.slot], override.role, override.binding)
if source is None:
return None
edited = Action(action.name, sources=list(action.sources))
edited.sources[override.slot] = source
if not isinstance(action.name, DeviceControl):
edited.refresh()
return edited
def _settle(self, previous: dict[str, Action], *, redirects_changed: bool = False) -> None:
"""Release what the change took away, and press nothing it brought in.
An action whose definition the change left alone keeps its held state:
a dialog opening while the player walks does not make them let go of
the key. An action that is gone, or that resolves to different
bindings, closes with a real release edge; and whatever is held at
this moment does not count towards the new definition until the player
presses it again, so a key held across a pause reads unpressed on
resume.
"""
state = self._state
if state is None or not state._anything_held():
return
changed: list[str] = []
arrived: list[Action] = []
for name, was in previous.items():
still = self._actions.get(name)
# An engine entry resolving elsewhere changes what every action
# bound through it means, whatever its own sources say.
if still is None or still.sources != was.sources or redirects_changed:
changed.append(name)
for name, now in self._actions.items():
before = previous.get(name)
if before is None or before.sources != now.sources or redirects_changed:
arrived.append(now)
if not changed and not arrived:
return
state._release_actions(changed, previous)
for action in arrived:
state._mark_stale(action.bindings)
# -- Declaring actions: the root context ------------------------------
def add_action(
self,
name: str,
bindings: list[SourceSpec] | SourceSpec | None = None,
*,
_quiet: bool = False,
):
"""Register a named action on the root context, with optional initial bindings.
It writes the tree's own context, the base of the stack, which lasts as
long as the tree: a node's ``input_context`` shadows the name while the
node is in the tree and gives it back when it goes.
It is not where a player's rebind goes; that is :meth:`rebind`.
Convenience: passing bare controls (``Key.SPACE``, ``Controller.A``) auto-wraps
them. A string names another entry, an action or a device alias, and
the action takes whatever that entry resolves to: ``add_action("pause",
["start", Key.ESCAPE])``. A name nothing declares, a loop and a type
that does not fit are refused here, naming the problem.
**The action's type follows from its bindings.** Buttons alone give a
``bool``; an :class:`~.actions.Axis1D` or a bare
axis such as ``Controller.LEFT_TRIGGER`` gives a ``float``; an
:class:`~.actions.Axis2D` or a stick such as ``Controller.LEFT_STICK`` gives
a ``Vec2``. Each entry is one source that produces a whole value on its
own, and when several are active the largest magnitude wins.
The canonical declaration is a node's ``input_context = {...}``, which
comes and goes with the node; this is for entries built as the game
runs.
Registering from a node callback lands in the tree's own map, because
every tree entry point that runs node code opens the tree's input span:
a tick, for ``on_update`` and ``on_fixed_update``; a mount, for
``on_enter_tree`` and ``on_ready``, whether the node arrives through
``set_root``, ``change_scene``, ``add_singleton`` or ``add_child``; a
teardown, for ``on_exit_tree``; event dispatch, for ``@on_input`` and UI
handlers such as ``on_gui_input``, a focus or hover signal, or a
button's ``pressed`` signal; and drawing, for ``on_draw``.
The ``_quiet`` flag is internal; it suppresses the overwrite warning.
"""
if not _quiet:
existing = self._root._actions.get(name)
if existing is not None:
log.warning("Input action %r overwritten (had %s bindings)", name, len(existing.bindings))
log.debug("InputMap.add_action(%r, %s)", name, bindings)
action = Action(name)
if bindings is not None and not isinstance(bindings, (list, tuple)):
bindings = [bindings]
if bindings:
action.sources.extend(to_source(b) for b in bindings)
if name in action.names:
raise ValueError(f"input entry {name!r} names itself: {name!r} -> {name!r}")
action.refresh()
self._declare(action)
def remove_action(self, name: str):
"""Remove a named action and all its bindings from the root context."""
if name in self._root._actions:
self._root._drop(name)
self._flatten()
def add_binding(self, name: str, binding: SourceSpec):
"""Declare one more default source for a root action, creating the action if it does not exist.
This is the game declaring what an action is, as ``add_action`` does,
and it edits the game's own context. A player's change goes through
:meth:`rebind`, the override layer, which survives every rebuild of the
stack, is saved by :meth:`export` and resets with :meth:`clear_rebinds`.
The action's type follows its sources, so adding a stick to an action
that had only keys widens it to a ``Vec2`` and adding a button to one
that already has a stick is refused, naming the fix.
"""
existing = self._root._actions.get(name)
action = Action(name, sources=[*existing.sources] if existing is not None else [])
action.sources.append(to_source(binding))
action.refresh()
self._declare(action)
def _declare(self, *actions: Action) -> None:
"""Put *actions* on the root context as one change, taking them off again if the map refuses it.
One change rather than one per entry, so entries declared together may
name each other in any order.
"""
was = {str(action.name): self._root._actions.get(str(action.name)) for action in actions}
known = {name: self._known.get(name) for name in was}
def undo() -> None:
for name, previous in was.items():
if previous is None:
self._root._drop(name)
else:
self._root._set(previous)
if known[name] is None:
self._known.pop(name, None)
else:
self._known[name] = known[name] # type: ignore[assignment]
with self._changing(undo):
for action in actions:
self._root._set(action)
self._known[str(action.name)] = action.kind
def remove_binding(self, name: str, binding: InputBinding):
"""Take one declared default source off a root action.
For the game declaring its defaults; a player's change goes through
:meth:`rebind` (``rebind(name, None, slot=...)`` to unbind one).
"""
action = self._root._actions.get(name)
if action is None:
return
try:
action.sources.remove(binding)
except ValueError:
return
action.refresh()
self._flatten()
# -- Reading what is bound now ----------------------------------------
def explain(self, name: str) -> ActionExplanation:
"""Why *name* reads what it reads: every fact the map used, as data.
Where it is declared and by which node, what its sources are and what
they resolve to, whether a context above hides it or it shadows a
declaration beneath, and which of the player's rebinds apply. What a
debugger panel, a rebind screen's tooltip or a test's failure message
shows, rather than something to reconstruct from the stack by hand.
A name nothing declares answers with ``declared`` False and empty
fields; asking never raises.
"""
declarers: list[tuple[InputContext, Node | None, bool]] = []
for (level, holder), live in zip(self._stack, self._active, strict=True):
if name in level._actions:
declarers.append((level, holder, live))
base_context: InputContext | None = None
if name in _project.context._actions and self._base is None:
base_context = _project.context
elif name in ENGINE_CONTEXT._actions:
base_context = ENGINE_CONTEXT
context = self._origin.get(name)
node = next((n for c, n, _live in declarers if c is context), None)
active = name in self._actions
hidden = not active and name in self._hidden
hidden_by: InputContext | None = None
if hidden:
origin_rank = max((i for i, (c, _n) in enumerate(self._stack) if name in c._actions), default=-1)
for index in range(len(self._stack) - 1, origin_rank, -1):
above, _ = self._stack[index]
if self._active[index] and (above._hides_all or name in above._hides):
hidden_by = above
break
in_force = context if context is not None else (declarers[-1][0] if declarers else base_context)
shadowed = tuple(c for c, _n, live in declarers if live and c is not in_force)
if base_context is not None and base_context is not in_force:
shadowed = (base_context, *shadowed)
return ActionExplanation(
name=name,
declared=self.declares(name) or base_context is not None,
active=active,
kind=self._known.get(name),
sources=tuple(self.get_sources(name)),
bindings=tuple(self.get_bindings(name)),
context=in_force,
node=node,
hidden=hidden,
hidden_by=hidden_by,
shadowed=shadowed,
rebinds=tuple(r for r in self._overrides.values() if r.action == name),
)
def get_bindings(self, name: str) -> list[InputBinding]:
"""The device bindings behind an action, flattened (empty list if unknown).
A composite source contributes the bindings it is made of, so a prompt,
a rebind screen and the event resolver all see one flat list of device
inputs whatever spelling declared them. :meth:`get_sources` is the
unflattened view, which is what says how they compose.
An action a context hides answers what it will be bound to when it
comes back, as :meth:`kind` answers its type: the menu that hides the
game's actions is the screen that shows and rebinds them. Whether it
can fire right now is :meth:`has_action`.
"""
action = self._actions.get(name)
if action is None:
action = self._hidden.get(name)
return action.bindings if action is not None else []
def get_sources(self, name: str) -> list[Source]:
"""The sources behind an entry as declared, each producing a whole value on its own.
A name in them is left as the name, which is what a rebind screen's
action table shows (``"pause"`` is ``["start", Key.ESCAPE]``);
:meth:`get_bindings` is what they resolve to. A hidden action answers
with what it will have when it comes back, rebinds included.
"""
action = self._declared.get(name)
if action is None:
action = self._hidden_declared.get(name)
return action.sources if action is not None else []
def resolve(self, entry: DeviceControl) -> DeviceButton | DeviceAxis | None:
"""The raw control an engine entry resolves to in this map, or None where it is unbound.
``resolve(Controller.A)`` is ``Controller.button(0)`` until something
redefines it; a raw control is its own. What a device table on a
rebind screen shows beside each named control.
"""
from .layouts import default_raw
if not isinstance(entry, DeviceControl):
raise TypeError(
f"resolve() takes an engine entry such as Controller.A, got {entry!r}; "
"get_bindings() reads a named entry"
)
bare = entry.on(None)
own = self._unit_redirects.get(entry.unit, {}) if entry.unit is not None else {}
if bare in own:
mapped = own[bare].on(entry.unit)
return mapped if isinstance(mapped, DeviceButton | DeviceAxis) else None
if entry.unit is not None and not _names_its_controls(entry.unit):
return None
raw = self._redirects[bare] if bare in self._redirects else default_raw(bare)
return raw if isinstance(raw, DeviceButton | DeviceAxis) else None
def aliases(self, raw: DeviceControl) -> tuple[str | DeviceControl, ...]:
"""Every entry that resolves directly to the raw control *raw*, in declaration order.
The engine entries first (``Controller.A`` for pad button 0, unless
something redefined it), then the game's own entries whose sources name
*raw* itself, in stack order and declaration order. An entry reaching
*raw* only through another name is that name's alias and is not
listed, so an action bound to ``Controller.A`` is not an alias of pad
button 0. Empty when nothing resolves to it.
Every one rather than a chosen one, because the map genuinely can hold
several and which to show is the rebind screen's decision: the first,
all of them, or the raw control's own label when there are none.
"""
from .controls import _NAMED
from .layouts import default_raw
if not isinstance(raw, DeviceControl):
raise TypeError(f"aliases() takes a raw control such as Controller.button(0), got {raw!r}")
bare = raw.on(None)
if bare.device is not DeviceType.KEYBOARD and bare.index is None:
raise TypeError(
f"aliases() takes a raw control, and {raw!r} is an entry: resolve({raw!r}) says what it resolves to"
)
whole = bare.whole if isinstance(bare, DeviceAxis) else bare
found: list[str | DeviceControl] = []
if bare.device is not DeviceType.KEYBOARD:
# On one device, what the player mapped there comes first, and an
# unrecognised controller has nothing else: its button 3 is not the
# name a recognised pad gives button 3.
own = self._unit_redirects.get(raw.unit, {}) if raw.unit is not None else {}
named = raw.unit is None or _names_its_controls(raw.unit)
for control in _NAMED.get(bare.device, {}).values():
if type(control) is not type(whole):
continue
if control in own:
target: DeviceControl | None = own[control]
elif not named:
continue
else:
target = self._redirects[control] if control in self._redirects else default_raw(control)
if target == whole:
found.append(control)
def names(control: object) -> bool:
if not isinstance(control, DeviceControl):
return False
if control.unit is not None and raw.unit is not None and control.unit != raw.unit:
return False
mine = control.on(None)
if mine == bare:
return True
if not isinstance(mine, DeviceAxis):
return False
# A whole axis is its two halves: an entry on either half, or on the
# whole of it, names the axis; a half asked for is named by itself
# and by the whole.
return mine == whole if bare is not whole else mine.whole == whole
for name, action in self._declared.items():
for source in action.sources:
parts: tuple[InputBinding | str, ...] = (
(source,) if isinstance(source, InputBinding) else () if isinstance(source, str) else source.parts
)
if any(isinstance(part, InputBinding) and names(part.control) for part in parts):
found.append(name)
break
return tuple(found)
def kind(self, name: str) -> ActionKind:
"""What type *name*'s value has.
An action a context hid answers the type its own declaration gave it,
which is the type it reads as at rest and the type it will have when it
comes back. A name no context has ever declared is a button action.
"""
action = self._actions.get(name)
if action is not None:
return action.kind
return self._known.get(name, ActionKind.BUTTON)
def has_action(self, name: str) -> bool:
"""Whether *name* is bound right now, by any active context."""
return name in self._actions
def declares(self, name: str) -> bool:
"""Whether *name* is an action this game has, active right now or not.
What separates a shadowed or hidden action, which reads as at rest,
from a misspelling, which raises.
"""
return name in self._known
@property
def actions(self) -> list[str]:
"""The action names bound right now."""
return list(self._actions)
def _unknown_action(self, name: str, query: str) -> KeyError:
"""The error a query for a name no context declares raises.
Answering False for ever is the worse failure: a misspelled action is
then a control that silently does nothing, on every machine, for the
life of the game.
"""
near = difflib.get_close_matches(name, sorted(self._known), n=3, cutoff=0.6)
hint = f" Did you mean {' or '.join(repr(n) for n in near)}?" if near else ""
return KeyError(
f"Input.{query}({name!r}): no input context declares an action called {name!r}.{hint} "
"Declare it on a node's input_context, in the project's settings, or with "
"InputMap.add_action. An action that is declared but not active right now reads "
"as at rest rather than raising."
)
def clear(self):
"""Remove every root action, every override and every remembered name.
The contexts nodes hold are not cleared: they belong to those nodes and
leave with them.
"""
self._root._clear()
self._overrides.clear()
self._known.clear()
self._pending_devices.clear()
self._flatten()
# -- The override layer -----------------------------------------------
@property
def rebinds(self) -> tuple[Rebind, ...]:
"""Every override in force, in the order they were made."""
return tuple(self._overrides.values())
def rebind(
self,
entry: str | DeviceControl,
binding: InputBinding | DeviceButton | DeviceAxis | ControlInput | str | None,
*,
slot: int = 0,
role: BindingRole | str = BindingRole.WHOLE,
player: int | None = None,
) -> Rebind:
"""Put *binding* in place of what an entry was declared as, and keep it there.
This is where a player's rebind screen writes, at any level of the map:
- an action, ``rebind("jump", Controller.B)``;
- a device alias, ``rebind("throttle", Controller.axis(3))``;
- an engine entry, ``rebind(Controller.A, Controller.button(1))``, after
which everything using ``Controller.A`` hears pad button 1: every
action, every handler, every role it belongs to;
- an engine entry on one device, ``rebind(Controller.A,
Controller.button(3).on(pad))``, which is how a controller no
database recognised gets its names: ``Controller.A`` then means that
pad's button 3 and nobody else's, and the mapping is saved with the
rest, filed under the device.
The override survives every rebuild of the context stack and is
reapplied to each flatten, so it outlives a scene change, a pause and
a vehicle being entered; it does not edit the
:class:`~.contexts.InputContext` the game wrote, which is frozen.
An action no active context declares right now can still be rebound:
a settings screen at the title rebinds the driving controls before any
car exists. The override is kept, dormant, and applies when a context
declaring the name arrives. Only a name no context anywhere has
declared is refused, with a ``KeyError`` naming the near misses,
because that is a misspelling rather than an action still to come. A
rebind that would leave the active map unresolvable (a loop, an
unknown name as a source, a type that does not fit) is refused and
not recorded.
Args:
entry: Which entry: a name, or an engine entry's control.
binding: What to put there: what an
:class:`~.listener.InputListener` heard, a control, a binding,
another entry's name, or ``None`` to leave it unbound. An
engine entry takes a raw control or another engine entry,
meaning the control the backend calls that.
What a listener heard is the one-line form a rebind screen
writes, ``InputMap.rebind(target, heard)``, and this applies
the rules a screen would otherwise carry: an action takes the
press with the modifiers and buttons held for it, so a chord is
bound whole; an engine entry takes the whole axis whichever
way it was pushed; and a press on a controller no database
recognised is pinned to that controller, whose button 3 is not
anybody else's.
slot: Which of the entry's sources, in declaration order. A
movement action declared as ``[Axis2D.WASD, Controller.LEFT_STICK]``
has the keyboard at slot 0 and the stick at slot 1, so
rebinding one leaves the other alone.
role: Which part of that source:
:attr:`BindingRole.WHOLE` for a button, ``LOW``/``HIGH`` for an
:class:`~.actions.Axis1D`, ``UP``/``DOWN``/``LEFT``/``RIGHT``
for an :class:`~.actions.Axis2D`. This is what makes rebinding
one direction of a movement action expressible.
player: Whose rebind it is. ``None``, the default, is everybody's;
a player id gives that player their own copy of the action and
leaves the rest reading what the game declared. An engine
entry is the hardware and has no per-player copy.
Returns:
The :class:`Rebind` recorded, which is also what
:meth:`export` writes.
"""
from .listener import ControlInput
if isinstance(binding, ControlInput):
binding = _heard(entry, binding)
record = self._rebind_record(entry, binding, slot=slot, role=role, player=player)
return self._record((record,))[0]
def swap(self, first: str | DeviceControl, second: str | DeviceControl) -> tuple[Rebind, Rebind]:
"""Rebind each of two entries to what the other resolves to, as one change.
The backwards-pad case, ``InputMap.swap(Controller.A, Controller.B)``, after
which every action, handler and role using either follows. Two device
aliases or two single-source actions swap the same way. One change and
one release edge rather than two, so nothing reads the halfway map in
which both entries resolve to the same control.
"""
if isinstance(first, DeviceControl) != isinstance(second, DeviceControl):
raise TypeError(f"swap() takes two engine entries or two named entries, got {first!r} and {second!r}")
if isinstance(first, DeviceControl) and isinstance(second, DeviceControl):
a, b = self.resolve(first), self.resolve(second)
records = (
self._rebind_record(first, b, slot=0, role=BindingRole.WHOLE, player=None),
self._rebind_record(second, a, slot=0, role=BindingRole.WHOLE, player=None),
)
else:
one, two = self.get_sources(str(first)), self.get_sources(str(second))
if (
len(one) != 1
or len(two) != 1
or not isinstance(one[0], InputBinding | str)
or not isinstance(two[0], InputBinding | str)
):
raise TypeError(
f"swap({first!r}, {second!r}) swaps two entries with one control each; "
f"rebind their parts with rebind(..., slot=, role=) instead"
)
records = (
self._rebind_record(first, two[0], slot=0, role=BindingRole.WHOLE, player=None),
self._rebind_record(second, one[0], slot=0, role=BindingRole.WHOLE, player=None),
)
first_record, second_record = self._record(records)
return first_record, second_record
def _rebind_record(
self,
entry: str | DeviceControl,
binding: InputBinding | DeviceButton | DeviceAxis | str | None,
*,
slot: int,
role: BindingRole | str,
player: int | None,
) -> Rebind:
"""Validate one rebind into the :class:`Rebind` it records."""
from .contexts import remap_entry
if isinstance(binding, str):
value: InputBinding | str | None = binding
elif binding is None:
value = None
else:
value = to_binding(binding)
if isinstance(entry, DeviceControl):
if player is not None:
raise TypeError(f"{entry!r} is an engine entry, the hardware, and has no per-player copy")
if slot != 0 or BindingRole(role) is not BindingRole.WHOLE:
raise TypeError(f"{entry!r} is an engine entry with one control: rebind it whole")
entry, value = _for_one_unit(entry, value)
if isinstance(value, InputBinding) and value.control is not None:
remap_entry(entry.on(None), [replace(value, control=value.control.on(None))])
elif value is not None:
remap_entry(entry.on(None), [value])
elif not isinstance(entry, str) or not entry:
raise TypeError(f"rebind() takes an entry's name or an engine entry such as Controller.A, got {entry!r}")
elif entry not in self._known and entry not in _DECLARED_NAMES:
near = difflib.get_close_matches(entry, sorted(set(self._known) | _DECLARED_NAMES), n=3, cutoff=0.6)
hint = f" Did you mean {' or '.join(repr(n) for n in near)}?" if near else ""
raise KeyError(
f"rebind({entry!r}, ...): no input context anywhere declares {entry!r}.{hint} A rebind "
"of an action whose context is not active yet is kept and applies when it arrives; "
"a name nothing declares is a misspelling."
)
elif isinstance(value, InputBinding):
value = self._by_name(value)
return Rebind(action=entry, binding=value, slot=slot, role=BindingRole(role), player=player)
def _by_name(self, binding: InputBinding) -> InputBinding:
"""*binding* with a raw control replaced by the engine entry that names it now, where one does.
What an action row of a rebind screen stores when the listener hands
it the raw control a player pressed. An action names entries, so the
binding follows a later fix in the device table (the player swapping
A and B moves it too) and prints as the button's name; a control no
engine entry names is bound raw. The half, the motion and the unit
carry over.
"""
control = binding.control
if control is None or control.device is DeviceType.KEYBOARD or control.index is None:
return binding
# aliases() answers for the unit: an unrecognised controller's button 3
# is named only where the player mapped it.
names = [name for name in self.aliases(control) if isinstance(name, DeviceButton | DeviceAxis)]
if not names:
return binding
named = names[0]
if isinstance(control, DeviceAxis) and isinstance(named, DeviceAxis):
if control.half:
named = named.positive if control.half > 0 else named.negative
if control.is_delta:
named = named.delta
return replace(binding, control=named.on(control.unit))
def _record(self, records: tuple[Rebind, ...]) -> tuple[Rebind, ...]:
"""Lay *records* over the override layer as one change, undone if the map refuses it."""
before = {record.key: self._overrides.get(record.key) for record in records}
def undo() -> None:
for key, was in before.items():
if was is None:
self._overrides.pop(key, None)
else:
self._overrides[key] = was
with self._changing(undo):
for record in records:
self._overrides[record.key] = record
return records
def clear_rebind(
self,
entry: str | DeviceControl,
*,
slot: int = 0,
role: BindingRole | str = BindingRole.WHOLE,
player: int | None = None,
) -> None:
"""Drop one override, giving that part back to what the game declared."""
if self._overrides.pop((entry, slot, BindingRole(role), player), None) is not None:
self._flatten()
def clear_rebinds(self) -> None:
"""Drop every override: reset to defaults, as a settings screen spells it."""
if self._overrides:
self._overrides.clear()
self._flatten()
def export(self) -> dict[str, Any]:
"""The override layer as plain data, ready for JSON, TOML or a save slot.
**A binding pinned to one device is filed under that device's guid,
never under its runtime id.** An id is the order devices were plugged
in, so a rebinding keyed on one follows the slot rather than the
controller: the player's rebound cockpit would land on somebody else's
pad after a USB reorder, and a wrong match is worse than a lost one.
A device that reports no guid, which some backends cannot supply, gets
no section: its rebinds stay session-only and a warning says so.
Everything not pinned to a device -- keys, mouse buttons, and a pad
binding that names no unit -- goes in the general section and always
persists.
"""
general: list[dict[str, Any]] = []
devices: dict[str, dict[str, Any]] = {
guid: {"bindings": [dict(row) for row in rows]} for guid, rows in self._pending_devices.items()
}
lost = 0
for override in self._overrides.values():
row = _rebind_to_row(override)
binding = override.binding
device_id = (
binding.control.unit if isinstance(binding, InputBinding) and binding.control is not None else None
)
if device_id is None:
general.append(row)
continue
guid = self._device_guid(device_id)
if not guid:
lost += 1
continue
devices.setdefault(guid, {}).setdefault("bindings", []).append(row)
if lost:
log.warning(
"%d rebind(s) are pinned to a device that reports no guid and "
"were not exported: a runtime id is the order devices were plugged in, so saving one "
"would restore it onto whichever device happens to hold that id next. They stay in "
"force for this session.",
lost,
)
data: dict[str, Any] = {"version": REBIND_FORMAT_VERSION, "bindings": general}
if devices:
data["devices"] = devices
return data
def load(self, data: Mapping[str, Any]) -> None:
"""Restore an override layer written by :meth:`export`, replacing the current one.
A device section is matched to the device reporting that guid and its
bindings are pinned to whatever id that device has now. A section whose
device is not plugged in is kept as it was, so it round-trips through
the next :meth:`export` and comes into force if that device is
connected later.
"""
version = data.get("version")
if version != REBIND_FORMAT_VERSION:
raise ValueError(
f"input rebind data is version {version!r}, and this engine writes and reads "
f"version {REBIND_FORMAT_VERSION}. Export it again from a build that wrote it."
)
unknown = sorted(set(data) - {"version", "bindings", "devices"})
if unknown:
raise ValueError(
f"input rebind data has unknown sections {unknown}; it holds 'version', 'bindings' "
"and 'devices'. Export it again from this build."
)
overrides: dict[tuple[str | DeviceControl, int, BindingRole, int | None], Rebind] = {}
pending: dict[str, list[dict[str, Any]]] = {}
for row in data.get("bindings") or ():
entry = _rebind_from_row(row)
overrides[entry.key] = entry
for guid, section in (data.get("devices") or {}).items():
rows = section.get("bindings") or []
device_id = self._device_for_guid(guid)
if device_id is None:
if rows:
pending[guid] = [dict(row) for row in rows]
continue
for row in rows:
entry = _rebind_from_row(row, device_id=device_id)
overrides[entry.key] = entry
was_overrides, was_pending = dict(self._overrides), dict(self._pending_devices)
def undo() -> None:
self._overrides, self._pending_devices = was_overrides, was_pending
with self._changing(undo):
self._overrides, self._pending_devices = overrides, pending
self._watch_devices()
@staticmethod
def _device_guid(device_id: int | None) -> str:
from .devices import store
rec = store.device_records.get(device_id) if device_id is not None else None
return rec.guid if rec is not None else ""
@staticmethod
def _device_for_guid(guid: str) -> int | None:
from .devices import store
if not guid:
return None
for rec in store.device_records.values():
if rec.connected and rec.guid == guid:
return rec.id
return None
def _watch_devices(self) -> None:
"""Bring a pending device section into force when its device is announced."""
if self._watching_devices or not self._pending_devices:
return
from .devices import store
self._watching_devices = True
# Weakly, because the device store is the process's and this map is a
# tree's: a strong subscription would keep a finished tree, every node
# whose context is on this stack, alive for the life of the process.
ref = weakref.ref(self)
def on_connected(device: Any) -> None:
table = ref()
if table is not None:
table._on_device_connected(device)
store.device_connected.connect(on_connected)
def _on_device_connected(self, device: Any) -> None:
"""A device a loaded section named has turned up: pin its rebinds to it."""
rows = self._pending_devices.pop(device.guid, None) if device.guid else None
if not rows:
return
for row in rows:
entry = _rebind_from_row(row, device_id=device.id)
self._overrides[entry.key] = entry
self._flatten()
def _heard(entry: str | DeviceControl, heard: ControlInput) -> InputBinding:
"""What a rebind of *entry* to the input a listener *heard* binds.
An engine entry is a whole control and holds nothing, so an axis half is
its whole axis; an action takes the press with what was held for it. A
control on a controller no database recognised is pinned to it.
"""
control = heard.control
unit = heard.device_id if not _names_its_controls(heard.device_id) else None
if isinstance(entry, DeviceControl):
whole = control.whole if isinstance(control, DeviceAxis) else control
return InputBinding(whole.on(unit))
held = tuple(h.on(unit) if unit is not None and h.device is control.device else h for h in heard.held)
return InputBinding(control.on(unit), held=held)
def _for_one_unit(
entry: DeviceControl, binding: InputBinding | str | None
) -> tuple[DeviceControl, InputBinding | str | None]:
"""An engine entry's rebind addressed by the unit it is for, when it is for one.
``rebind(Controller.A, Controller.button(3).on(pad))`` and
``rebind(Controller.A.on(pad), Controller.button(3))`` are the same
mapping: the entry is recorded on the unit, so two pads map independently,
and the binding on it, so the rebind file files it under the device.
"""
control = binding.control if isinstance(binding, InputBinding) else None
unit = entry.unit if entry.unit is not None else control.unit if control is not None else None
if unit is None:
return entry, binding
if control is not None and control.unit not in (None, unit):
raise TypeError(f"{entry!r} is mapped on one device, and {control!r} is on another")
if isinstance(binding, InputBinding) and control is not None:
binding = replace(binding, control=control.on(unit))
return entry.on(unit), binding
def _resolve_unit_redirects(
remaps: Mapping[DeviceControl, Action],
) -> tuple[dict[DeviceControl, DeviceControl], dict[DeviceControl, tuple[DeviceControl, ...]]]:
"""What each engine entry mapped on one device resolves to there, and the reverse."""
forward: dict[DeviceControl, DeviceControl] = {}
reverse: dict[DeviceControl, list[DeviceControl]] = {}
for key, action in remaps.items():
(source,) = action.sources
control = source.control if isinstance(source, InputBinding) else None
if control is None:
continue
target: DeviceControl = control
if target.index is None:
# Another engine entry means the control the backend calls that.
from .layouts import default_raw
target = default_raw(target) or target
forward[key] = target
reverse.setdefault(target, []).append(key)
return forward, {raw: tuple(keys) for raw, keys in reverse.items()}
def _names_its_controls(device_id: int) -> bool:
"""Whether the engine's named controls mean anything on *device_id*: False for an unrecognised controller."""
from .devices import store
pad = store.pads.get(device_id)
return pad is None or pad.recognised
def _rebind_to_row(override: Rebind) -> dict[str, Any]:
"""One override as plain data, with only what differs from the defaults.
A named entry is filed as ``"action"`` and an engine entry as ``"control"``
in its written spelling; the binding is written as ``simvx.toml`` writes a
source (:mod:`.spelling`), a reference to an entry as ``{"name": ...}``.
"""
entry = override.action
row: dict[str, Any] = {"control": str(entry)} if isinstance(entry, DeviceControl) else {"action": entry}
binding = override.binding
row["binding"] = None if binding is None else source_to_data(binding)
if override.slot:
row["slot"] = override.slot
if override.role is not BindingRole.WHOLE:
row["role"] = str(override.role)
if override.player is not None:
row["player"] = override.player
return row
def _rebind_from_row(row: Mapping[str, Any], *, device_id: int | None = None) -> Rebind:
"""One override read back, repinned to the device a section named."""
data = row.get("binding")
binding: InputBinding | str | None = None
if data is not None:
read = source_from_data(data, device_id=device_id)
if not isinstance(read, InputBinding | str):
raise ValueError(f"input rebind data binds {data!r}, which is not one control or one entry's name")
if isinstance(read, InputBinding) and read.control is not None and device_id is not None:
from .devices import store
rec = store.device_records.get(device_id)
if rec is not None and rec.type is not read.control.device:
raise ValueError(
f"input rebind data files {data!r} under a {rec.type.value}, which has no such control"
)
binding = read
entry: str | DeviceControl
if "control" in row:
entry = DeviceControl.parse(str(row["control"]))
elif "action" in row:
entry = str(row["action"])
else:
raise ValueError(f"input rebind data has a row naming no entry: {dict(row)!r}")
if isinstance(entry, DeviceControl):
entry, binding = _for_one_unit(entry, binding)
return Rebind(
action=entry,
binding=binding,
slot=int(row.get("slot", 0)),
role=BindingRole(row.get("role", BindingRole.WHOLE)),
player=None if row.get("player") is None else int(row["player"]),
)
def _expand(
targets: Mapping[str, Action],
pools: Mapping[int, Mapping[str, Action]],
default_pool: Mapping[str, Action],
declared_where: Mapping[str, str],
) -> dict[str, Action]:
"""*targets* with every name in them expanded, each through the entries visible at its own level.
An entry naming another takes that entry's sources; a composite's part
naming one takes the single control it resolves to. An entry may name only
entries declared in its own context or beneath it, never one from a context
above that can leave, so taking a context off the stack can orphan nothing.
A loop, a name the entry cannot see and a type mismatch each raise, naming
the entries involved.
"""
if not any(action.names for action in targets.values()):
return dict(targets)
done: dict[int, Action] = {}
def resolve(action: Action, trail: tuple[str, ...]) -> Action:
found = done.get(id(action))
if found is not None:
return found
name = str(action.name)
if name in trail:
loop = " -> ".join(repr(step) for step in (*trail[trail.index(name) :], name))
raise ValueError(f"input entries name each other in a loop: {loop}")
if not action.names:
done[id(action)] = action
return action
pool = pools.get(id(action), default_pool)
inner = (*trail, name)
def named(target: str) -> Action:
found_there = pool.get(target)
if found_there is not None:
return resolve(found_there, inner)
above = declared_where.get(target)
if above is not None:
raise KeyError(
f"input entry {name!r} names {target!r}, which is not visible where {name!r} is declared: "
f"the context {above!r} declares it, above {name!r} or behind a context that hides it. An "
"entry may name only entries in its own context or beneath it, since a context above can "
"leave. Declare it at or below, or move the entry up."
)
near = difflib.get_close_matches(target, [str(key) for key in pool], n=3, cutoff=0.6)
hint = f" Did you mean {' or '.join(repr(n) for n in near)}?" if near else ""
spelled = _written_key(target)
if spelled is not None:
# A string in a source list is an entry's name; a key is its control.
hint += f" A key is written as its control: {spelled}."
raise KeyError(f"input entry {name!r} names {target!r}, which no context at or below it declares.{hint}")
sources: list[Source] = []
for source in action.sources:
if isinstance(source, str):
sources.extend(named(source).sources)
elif isinstance(source, InputBinding):
sources.append(source)
else:
sources.append(_expand_parts(source, named, owner=name))
expanded = Action(action.name, sources=sources)
expanded.refresh()
done[id(action)] = expanded
return expanded
return {name: resolve(action, ()) for name, action in targets.items()}
def _written_key(text: str) -> str | None:
"""How *text* is spelt in Python, where it is a key or a key combo written as a string; else None."""
try:
binding = to_binding(text)
except TypeError, ValueError:
return None
control = binding.control
if not isinstance(control, DeviceButton) or control.device is not DeviceType.KEYBOARD:
return None
return " + ".join(repr(control) for control in (*binding.held, control))
def _expand_parts(source: Axis1D | Axis2D, resolve: Callable[[str], Action], *, owner: str) -> Axis1D | Axis2D:
"""*source* with each part that names an entry replaced by the one control that entry resolves to."""
fields = ("low", "high") if isinstance(source, Axis1D) else ("up", "down", "left", "right")
changes: dict[str, InputBinding] = {}
for field_name in fields:
part = getattr(source, field_name)
if not isinstance(part, str):
continue
resolved = resolve(part).sources
if len(resolved) != 1 or not isinstance(resolved[0], InputBinding):
what = f"{len(resolved)} sources" if len(resolved) != 1 else f"a {type(resolved[0]).__name__}"
raise TypeError(
f"input entry {owner!r}: {type(source).__name__}({field_name}={part!r}) takes one control, "
f"and {part!r} resolves to {what}. Name an entry with a single control there, or bind "
f"the control itself."
)
changes[field_name] = resolved[0]
return replace(source, **changes) if changes else source # type: ignore[arg-type]
def _resolve_redirects(
remaps: Mapping[DeviceControl, Action],
) -> tuple[dict[DeviceControl, DeviceControl | None], dict[DeviceControl, tuple[DeviceControl, ...]]]:
"""Which engine entries resolve somewhere other than their default, and where.
A redefinition names a raw control (``Controller.button(1)``) or another
engine entry (``Controller.B``), and the second means the control the backend
calls that, not whatever that entry has been redefined to: so
``{Controller.A: [Controller.B], Controller.B: [Controller.A]}`` is the swap it reads
as, and engine entries can never name each other in a loop.
"""
from .layouts import default_raw
redirects: dict[DeviceControl, DeviceControl | None] = {}
reverse: dict[DeviceControl, list[DeviceControl]] = {}
for key, action in remaps.items():
(source,) = action.sources
target = source.control if isinstance(source, InputBinding) else None
raw = default_raw(target) if target is not None else None
if raw is not None and raw == default_raw(key):
continue
redirects[key] = raw
if raw is not None:
reverse.setdefault(raw, []).append(key)
return redirects, {raw: tuple(keys) for raw, keys in reverse.items()}
class _ProjectLayer:
"""The project's entries: the layer under every tree's own, from its settings."""
def __init__(self) -> None:
self.context = InputContext(name="project")
#: Every map that exists, so a change to the project reaches them all.
self.maps: weakref.WeakSet[_InputMap] = weakref.WeakSet()
def set(self, entries: Mapping[str | DeviceControl, list[SourceSpec] | SourceSpec]) -> None:
"""Make *entries* the project's, refusing a set that cannot resolve on its own."""
context = InputContext(dict(entries), name="project")
_InputMap._check_alone(context)
self.context = context
for table in tuple(self.maps):
table._flatten()
_project = _ProjectLayer()
_active_input_map: contextvars.ContextVar[_InputMap | None] = contextvars.ContextVar("_active_input_map", default=None)
class _InputMapProxy:
"""Proxy that delegates all access to the active _InputMap for the current context."""
__slots__ = ()
#: The class this proxy stands in for, for ``tools/api_surface.py``: the
#: package root publishes this object, and the guard freezes the target's
#: public methods under the exported name rather than one opaque line.
_api_surface_class = _InputMap
def __getattr__(self, name: str):
return getattr(current_input_map(), name)
def __setattr__(self, name: str, value):
setattr(current_input_map(), name, value)
def __repr__(self) -> str:
try:
return repr(current_input_map())
except RuntimeError:
return "InputMap(no tree)"
InputMap = _InputMapProxy()
__all__ = [
"ActionExplanation",
"BindingRole",
"InputMap",
"Rebind",
"REBIND_FORMAT_VERSION",
"roles_of",
"set_active_input_map",
"to_binding",
]