"""Input: input state tracker. Instance-based with module-level default."""
import contextvars
import logging
from collections.abc import Callable, Iterable, Iterator, Mapping
from contextlib import contextmanager
from itertools import count
from time import monotonic
from types import MappingProxyType
from typing import TYPE_CHECKING, cast
from ..math.types import Vec2
from ..signals import Signal
from . import capabilities as _capabilities
from . import controller as _controller
from . import controls as _controls_module
from .actions import Action, ActionKind, Axis1D, Axis2D, Source
from .capabilities import InputCapabilities
from .controls import (
MODIFIER_OF,
MODIFIER_SIDES,
DeviceAxis,
DeviceButton,
DeviceControl,
DeviceStick,
DeviceType,
Pointer,
_name_for_code,
key_code,
key_to_name,
roles_of,
)
from .devices import ControllerDevice, InputDevice, _PadRecord, family_for
from .devices import store as _devices
from .enums import (
ControllerFamily,
ControllerSource,
CursorShape,
InputMethod,
MouseCaptureMode,
PointerPhase,
PointerType,
)
from .events import AxisEvent, ButtonEvent, InputBinding, PointerEvent, PointerState
from .gestures import recognised_in
from .layouts import UNRECOGNISED_CONTROLLER_LAYOUT, DeviceLayout, raw_axis, raw_button
from .map import _InputMap
from .players import DEFAULT_PLAYER, InputPlayer
from .tick import ActionState, ControllerSnapshot, InputEdge, TickInput
if TYPE_CHECKING:
from ..scene_tree import SceneTree
log = logging.getLogger(__name__)
# Axis names already reported once as out of range, so a backend that lies
# every frame says so once rather than sixty times a second.
_warned_axis_names: set[str] = set()
#: The answer to "who hears this device" while there is one player, returned
#: rather than rebuilt so the ordinary single-player path allocates nothing.
_SOLO: list[int] = [DEFAULT_PLAYER]
class _PointerSource:
"""One input source's pointer: where it is, and what it has done this frame.
A window is a source and a screen is not, so this is per window rather
than per process or per tree: two windows feeding one tree accumulate
their motion separately, and neither re-bases the other's delta.
"""
__slots__ = ("position", "delta", "scroll", "discard_next_delta", "pointers", "edged")
def __init__(self) -> None:
self.position: tuple[float, float] = (0.0, 0.0)
self.delta: tuple[float, float] = (0.0, 0.0)
self.scroll: tuple[float, float] = (0.0, 0.0)
#: The next motion updates the position baseline and contributes no
#: delta. Set across a discontinuity (capture mode change, focus
#: regain, cursor warp) so the one-off jump is not mistaken for real
#: motion. Consumed by the next motion event.
self.discard_next_delta: bool = False
#: Every live pointer this source has delivered, by pointer id: the
#: mouse, each finger on the glass and a pen in range.
self.pointers: dict[int, PointerState] = {}
#: Whether a pointer here carries an edge or has gone, so the frame
#: boundary visits only a source that has something to spend.
self.edged: bool = False
def _without_edges(state: PointerState) -> PointerState:
"""*state* with its frame's press, release and cancel spent."""
return PointerState(
pointer_id=state.pointer_id,
pointer_type=state.pointer_type,
position=state.position,
contact=state.contact,
is_primary=state.is_primary,
buttons=state.buttons,
pressure=state.pressure,
tilt=state.tilt,
twist=state.twist,
distance=state.distance,
tangential_pressure=state.tangential_pressure,
tool=state.tool,
tool_id=state.tool_id,
contact_size=state.contact_size,
device_id=state.device_id,
)
def _off_rest(value: float | Vec2) -> bool:
"""Whether a continuous action's value has left rest.
A :class:`~simvx.core.math.types.Vec2` is a NumPy array and every array of
more than one element is truthy, so the test has to be written rather than
inferred: a zero vector would otherwise read as pressed.
"""
return value.length() > 0.0 if isinstance(value, Vec2) else value != 0.0
def _at_rest(kind: ActionKind) -> bool | float | Vec2:
"""What an action of *kind* reads when nothing is driving it.
An action a context hid resolves to nothing at all, so its rest value comes
from its declared type rather than from a source: a movement action reads as
a zero vector whether it is hidden or merely untouched, and a game polling
it through a modal keeps doing vector arithmetic on the answer.
"""
if kind is ActionKind.AXIS_2D:
return Vec2(0.0, 0.0)
if kind is ActionKind.AXIS_1D:
return 0.0
return False
class _Input:
"""One tree's delivered record, and the actions resolved from it.
Input is three layers, and this object is the top two. The **device
record** underneath is process-wide (:mod:`simvx.core.input.devices`,
:mod:`simvx.core.input.capabilities`) and answers what the machine is
doing: ``is_key_pressed`` is true while the key is physically down,
whoever the input went to. The **delivered record** is what an instance
here owns: the device elements that actually reached *this* tree, which
for a game hosted inside an editor is what its host forwarded and nothing
else, along with the stale and claimed slots, the open actions, the
players and the contexts. The **action layer** is the binding resolution
on top, turning this tree's delivered record into actions.
Two things follow from the split that do not have to be written. A key
already held when input starts arriving at a tree presses nothing there,
because it was never delivered to it. And an editor running a game does
not share action state with it, while both agree about the hardware: the
device record is one machine.
Actions are declared on an InputMap; query with ``pressed()`` and its
companions. Device queries are the escape hatch and read the layer below:
``is_key_pressed(Key.X)``, ``is_mouse_button_pressed(Mouse.LEFT)``.
"""
#: Check every stored action answer against the resolver it replaced, and
#: raise where they differ. Off by default and for the test suite, which
#: turns it on for a whole run: it makes every action query resolve as
#: well as read, which is exactly the cost the store exists to avoid.
verify_action_state: bool = False
def __init__(self, input_map: _InputMap | None = None):
self.input_map = input_map if input_map is not None else _InputMap()
#: The machine's devices, shared with every other instance.
self._devices = _devices
#: Whether the primary finger or pen also drives the mouse: moving
#: ``mouse_position`` and, for its contact alone, pressing the mouse's
#: primary button, in the device record, in this tree's delivered
#: record and so in every binding on it. A barrel press is not a
#: right click to a mouse-written game. That is what lets a game written for a mouse play
#: under a thumb. It is state only: no mouse event is made from a
#: finger, so nothing is delivered twice, and a handler reads the
#: finger's own ``PointerEvent``. Turn it off in a game that tells a
#: finger from the mouse, and a tap then presses nothing on the left
#: button.
#:
#: Per tree, because every consumer is: the delivered record and the
#: bindings resolved from it, the per-source pointer that
#: ``mouse_position`` reads, and the router that writes them all
#: through this object. An editor and the game it hosts may disagree.
self.emulate_mouse: bool = True
self._touch_secondary: DeviceButton | None = None
#: The step being run, frame or physics tick, in seconds: what a
#: per-second binding (``Controller.RIGHT_TRIGGER.per_second(0.5)``)
#: integrates its held value over.
self._step_dt: float = 0.0
#: The actions a per-second binding drives, which a new step length
#: moves, and the map generation that set was read from.
self._rated: set[str] = set()
self._rated_generation = -1
# --- String-keyed state (written by platform adapters, read by UI system) ---
# Kept alongside the typed-set state below because several layers
# still query it by name (cheaper than keycode lookup for the cases
# below): touching the keycode path would balloon those into
# multi-step enum conversions. The readers are:
# * ui_input.py builds combo_keys ("ctrl+s") from these mods on
# every keyboard event.
# * multiline.py / editor panels check ctrl/shift/alt for
# modifier-aware gestures.
# * play_mode.py snapshots them so the editor can pause/resume
# without leaking key state across the boundary.
# * The "mouse_1"/"mouse_2"/... entries let game code poll the
# mouse without remembering enum codes.
# The typed delivered set (below) is what the binding resolver reads,
# and the device record one layer down is what
# ``Input.is_key_pressed(Key.X)`` answers from.
self._keys: dict[str, bool] = {}
# --- The physics tick ---
#: The snapshot the tick now running was given, and None outside one.
#: Every query tests it, so a tick reads one state and one set of
#: edges however many events arrive while it runs.
self._tick: TickInput | None = None
#: Whether the open tick's snapshot is still the live table, which is
#: the only case the action-state oracle can compare against. False
#: for a snapshot a caller supplied, which is the whole truth for its
#: tick and owes the live record nothing, and False once an event
#: delivered mid-tick has moved the live table on underneath it.
self._tick_verifiable: bool = True
#: Whether the open tick is the first of its frame, which is the one
#: that owns the frame's device edges. A later tick of the same frame
#: sees none, or a press would fire once per tick.
self._tick_owns_edges: bool = False
#: How many ticks have run since the frame boundary, so the first can
#: be told from the rest.
self._ticks_this_frame: int = 0
#: Whether a physics tick has ever run on this state. Until one has,
#: the edges below are spent after a frame rather than held for a tick
#: that may never come: a tree that never ticks, a UI harness or an
#: editor panel, would otherwise accumulate them for ever.
self._ticked: bool = False
#: Action edges no tick has taken yet. A frame that runs no tick, which
#: is most frames at 200 fps against a 60 Hz tick, carries its edges to
#: the next frame's first tick rather than dropping them.
self._pending_edges: list[InputEdge] = []
#: Frame boundaries an edge has waited through before the first tick.
self._tickless_boundaries: int = 0
self._keys_just_pressed: dict[str, bool] = {}
self._keys_just_released: dict[str, bool] = {}
#: One pointer per input source, because a pointer belongs to its
#: source rather than to the process: a window is a source and a
#: screen is not, so two windows onto one tree have two cursors and
#: two sets of motion to accumulate. Keyed by
#: :attr:`InputRouter.source_id`.
self._sources: dict[int, _PointerSource] = {}
#: ``(source id, pointer id)`` of the pens that left range this frame,
#: whose records go at the frame boundary.
self._leaving: set[tuple[int, int]] = set()
#: Which source moved a pointer here last, which is the one the
#: source-less queries answer for. There is only ever one under a
#: single window, which is every game.
self._last_source: int = 0
# --- The action layer ---
# What each action reads, per player, written rather than resolved.
# Three answers in one record because the three queries genuinely
# differ -- a button action bound to a trigger is open or not, reads
# as a bool, and has an analogue strength -- and they are written
# together, so one table is one dict copy when a tick snapshots it.
self._action_state: dict[tuple[int, str], ActionState] = {}
#: Which actions read each device element, so a delivery resolves the
#: handful that changed rather than the whole table.
self._actions_on: dict[tuple, set[str]] = {}
#: The button actions an axis can push part-way, which are the only
#: ones whose strength is not their held state.
self._analogue: set[str] = set()
#: The binding table's generation these were built against. Negative
#: until the first query, so a state nobody asks costs nothing.
self._map_generation: int = -1
# --- The delivered record ---
# The elements that reached this tree, and their edges here. Every
# binding resolves against these and nothing else, which is what makes
# action state a fact about what the game was given rather than about
# what the machine is doing. The honest answers to the same questions
# live one layer down, in ``self._devices``.
self._keys_delivered: set[int] = set()
#: The keys delivered from each keyboard, by its device id, for a key
#: binding pinned to one keyboard with ``Key.SPACE.on(keyboard)``.
self._unit_keys: dict[int, set[int]] = {}
#: The modifier groups the event being dispatched says were held, or
#: None outside a dispatch. The platform's report is the truth while
#: its event resolves; the delivered record answers otherwise.
self._event_modifiers: frozenset[DeviceButton] | None = None
self._unit_keys_down: dict[int, set[int]] = {}
self._unit_keys_up: dict[int, set[int]] = {}
self._keys_delivered_down: set[int] = set()
self._keys_delivered_up: set[int] = set()
# A pointer or tablet button is delivered by name: a press adds every
# name it answers to under its device's layout, so a binding holds
# while its own name is here and never compares numbers across
# devices. Counted, because two devices can hold one name at once
# (the mouse's right and a barrel both hold ``secondary``). A binding
# pinned to one device reads the ``(device id, name)`` keys.
self._buttons_delivered: dict[object, int] = {}
self._buttons_delivered_down: set[object] = set()
self._buttons_delivered_up: set[object] = set()
#: Each press in flight, with the names it added, so its release takes
#: away exactly what it put in even if the layout changed meanwhile.
self._button_presses: dict[tuple, tuple[object, ...]] = {}
# A device axis by name, as its position (None where the hardware
# reports only motion), where it stood before this frame's first
# change, and its motion this frame, summed. A pad's buttons and axes
# are delivered here like a tablet's.
self._axes_delivered: dict[object, float | None] = {}
self._axes_delivered_prev: dict[object, float | None] = {}
self._axis_deltas_delivered: dict[object, float] = {}
# Each device axis a binding may read unpinned, as every device that
# reported it last left it: the unpinned value is the one furthest from
# rest, so two pads on one ``Controller.LEFT_X`` do not fight, and a pad
# resting at zero cannot drown out the one being pushed.
self._axis_sources: dict[object, dict[int, float | None]] = {}
self._capture_mode: MouseCaptureMode = MouseCaptureMode.VISIBLE
self._capture_mode_callback: object = None # Platform sets this to apply capture
# The open platform text-input session: the rect being edited, or None
# when there is none. The callback is the backend's, installed the way
# the capture one is, and is called with the rect or with None.
self._text_input_rect: tuple[float, float, float, float] | None = None
self._text_input_callback: Callable[[tuple[float, float, float, float] | None], None] | None = None
# The pointer's shape the tree last asked the platform for, and the
# backend's callback that shows it, installed the way the capture one is.
self._cursor_shape: CursorShape = CursorShape.ARROW
self._cursor_shape_callback: Callable[[CursorShape], None] | None = None
# Which continuous actions were past their deadzone as of the last
# event the resolve phase saw, per player. Deliberately not the same
# thing as ``_action_state``, which is what they read now: an edge on a
# continuous action is the difference between the two, because a
# stick carries no device edge of its own. Not frame-scoped either,
# since the action stays open for as long as the stick is held.
self._open_at_last_event: dict[int, set[str]] = {DEFAULT_PLAYER: set()}
# --- Players ---
# One implicit player, which owns every device nobody claimed. A game
# that never adds a second never leaves the fast path: ``_solo`` is
# what every binding query tests, and while it holds the resolver is
# the one that was there before players existed.
self._players: dict[int, InputPlayer] = {DEFAULT_PLAYER: InputPlayer(self, DEFAULT_PLAYER)}
self._solo: bool = True
#: Which pads and whether the desk are claimed by a player other than
#: the default one, so the default player's "everything unpaired" is a
#: set test rather than a walk over every player.
self._paired_pads: set[int] = set()
self._paired_desk: bool = False
#: Whose queries are being answered. Set for the duration of a
#: subtree's code by :attr:`Node.input_player`, and read by the
#: binding resolver; the default player answers everywhere else.
self._current_player: int = DEFAULT_PLAYER
# --- Held across a context change ---
#: Bindings that were already held when the context stack changed.
#: They read as at rest until the player lets go and presses again, so
#: a key held across a pause does not walk the character on resume.
#: ``_stale`` is the gate every binding query tests, so a game that
#: never changes a context with a key down pays one attribute read.
self._stale_bindings: set[tuple] = set()
self._stale: bool = False
#: ``(player, action)`` pairs that a context change closed this frame,
#: which is what makes that close a real release edge rather than the
#: action quietly going dark.
self._forced_released: set[tuple[int, str]] = set()
# --- Taken by the targeted phase ---
#: Device elements a control claimed when the press reached it. A
#: claimed press resolves to no action, raises no edge and reaches no
#: broadcast, and so does its release: the claim is decided once, at
#: the press, and the element stays claimed until that press is let
#: go. An empty set is the whole cost for a game with no UI over it.
self._claimed_slots: set[tuple] = set()
# ----------------------------------------------------------------
# Action queries (via InputMap only)
# ----------------------------------------------------------------
def _entry(self, action: str, query: str) -> Action | None:
"""What *action* resolves to for the player being answered for.
None where the name is declared somewhere but no active context binds
it: a shadowed or hidden action reads as at rest. A name no context
anywhere declares raises instead, because answering False for ever
turns a misspelling into a control that silently does nothing.
"""
table = self.input_map
per_player = table._player_actions
if per_player:
# Only where a rebind named a player, which is the co-op case and
# nothing else: one empty-dict test otherwise.
own = per_player.get(self._current_player)
if own is not None:
entry = own.get(action)
if entry is not None:
return entry
entry = table._actions.get(action)
if entry is None and action not in table._known:
raise table._unknown_action(action, query)
return entry
@property
def context_stack(self) -> tuple:
"""The active input contexts, bottom to top.
The first is always the tree's own, which ``InputMap.add_action``
declares and which never deactivates. Read it when an action is not firing
and you want to see which context took its name. A context whose node
cannot receive input, because it is paused or disabled, is not listed:
it is holding nothing at that moment.
"""
return self.input_map.context_stack
# ----------------------------------------------------------------
# The action layer: stored, not derived per query
# ----------------------------------------------------------------
#
# Three answers are kept per (player, action) and written when something
# that could change them happens: a device element delivered to this tree,
# a stale or claimed mark, a player pairing, or a change to the binding
# table. A game polls its actions every frame and receives a handful of
# events, so the cost belongs on the write. Measured over a twenty-action
# table, resolving on read costs 8.8us a frame against 2.6us stored plus
# 0.49us per delivered event, so storing wins below about thirteen input
# events in one frame, which no frame has.
#
# The binding table is the one input that is checked rather than pushed:
# it changes rarely, every mutator reaches ``_InputMap._flatten``, and
# that is the single place its generation is bumped, so one comparison on
# read is both cheaper and safer than a notification per mutator.
def _fresh(self) -> None:
"""Rebuild the stored answers if the binding table has moved on."""
if self._map_generation != self.input_map._generation:
self._rebuild_action_state()
def _read(self, action: str) -> ActionState | None:
"""What *action* reads for the player being answered for, or None.
Inside a physics tick this is that tick's own snapshot, so every query
the tick makes sees one state even if something publishes an event
from inside it. Outside a tick it is the live table, rebuilt first if
the binding table has moved on.
"""
tick = self._tick
if tick is not None:
return tick.actions.get((self._current_player, action))
self._fresh()
return self._action_state.get((self._current_player, action))
@staticmethod
def _binding_element(b: InputBinding) -> tuple:
"""The device element *b* reads, as the key the recompute index uses.
Coarser than :meth:`_binding_slot`, which a stale mark is filed under:
an axis is one element whichever half of its travel a binding names,
because a delivery moves the whole axis and both halves have to be
looked at again.
"""
control = b.control
if control is not None:
code = key_code(control) if control.device is DeviceType.KEYBOARD else None
if code is not None:
return ("key", code)
if isinstance(control, DeviceButton):
return ("button", _button_key(b))
return ("device_axis", _axis_key(b))
return ("none",)
def _index_binding(self, index: dict[tuple, set[str]], b: InputBinding, name: str) -> None:
"""File *name* under every element that can change what *b* reads.
The element the binding names, and the modifier keys it requires: a
binding on ctrl+S reads differently the moment Control goes down, so
Control has to reach it too.
"""
index.setdefault(self._binding_element(b), set()).add(name)
for held in b.held:
for member in held.members:
index.setdefault(self._binding_element(InputBinding(member.on(held.unit))), set()).add(name)
if b.is_exact:
# Exact about modifiers: any of them going down changes the answer.
for code in _MODIFIER_CODES:
index.setdefault(("key", code), set()).add(name)
def _rebuild_action_state(self) -> None:
"""Re-index the binding table and resolve every action against it."""
table = self.input_map
self._map_generation = table._generation
index: dict[tuple, set[str]] = {}
for name, action in table._actions.items():
for b in action.bindings:
self._index_binding(index, b, name)
for own in table._player_actions.values():
for name, action in own.items():
for b in action.bindings:
self._index_binding(index, b, name)
self._actions_on = index
# Which actions can read something other than fully on or fully off.
# Only an axis binding can, so every other action's strength is its
# held state and the recompute skips a second walk of its bindings.
analogue: set[str] = set()
for source in (table._actions, *table._player_actions.values()):
for name, action in source.items():
if any(isinstance(b.control, DeviceAxis) for b in action.bindings):
analogue.add(name)
self._analogue = analogue
self._action_state.clear()
self._recompute(table._actions.keys())
def _recompute(self, names: Iterable[str]) -> None:
"""Resolve *names* for every player and write what each one reads."""
if self._tick is not None:
# A tick is open and the live table has just moved on under it.
# The tick goes on reading its snapshot, which is what a snapshot
# is for, but the oracle can no longer compare the two.
self._tick_verifiable = False
previous = self._current_player
stored = self._action_state
try:
for player_id in self._players:
self._current_player = player_id
for name in names:
key = (player_id, name)
entry = self._entry_if_active(name)
if entry is None:
stored.pop(key, None)
elif entry.kind is ActionKind.BUTTON:
held = False
for b in entry.bindings:
if self._binding_pressed(b):
held = True
break
if name in self._analogue:
strength = 0.0
for b in entry.bindings:
strength = max(strength, self._binding_strength(b))
else:
# Nothing here can read part-way, so how hard the
# action is pushed is whether it is pushed.
strength = 1.0 if held else 0.0
stored[key] = ActionState(held, held, strength)
else:
current = self._continuous(entry)
magnitude = current.length() if isinstance(current, Vec2) else abs(current)
stored[key] = ActionState(_off_rest(current), current, magnitude)
finally:
self._current_player = previous
def _entry_if_active(self, action: str) -> Action | None:
"""What *action* resolves to for the player being answered for, or None.
:meth:`_entry` without the raise: the recompute is walking names the
table itself supplied, so an absent one is a shadowed action rather
than a misspelling.
"""
table = self.input_map
per_player = table._player_actions
if per_player:
own = per_player.get(self._current_player)
if own is not None:
entry = own.get(action)
if entry is not None:
return entry
return table._actions.get(action)
def _set_step_dt(self, dt: float) -> None:
"""Start a step of *dt* seconds: an action a per-second binding drives reads over it."""
if dt == self._step_dt:
return
self._step_dt = dt
table = self.input_map
generation = table._generation
if self._rated_generation != generation:
self._rated = {
name
for name, entry in table._actions.items()
if isinstance(name, str) and any(b.rate is not None for b in entry.bindings)
}
self._rated_generation = generation
if self._rated and self._actions_on:
self._recompute(set(self._rated))
def _touch(self, *elements: tuple) -> None:
"""Something changed on these device elements: resolve what reads them.
The one-element case is the keyboard and the mouse, which is most
input, and it hands the index's own set straight to the recompute
rather than building a union of one.
"""
index = self._actions_on
if not index:
return
if len(elements) == 1:
found = index.get(elements[0])
if found:
self._recompute(found)
return
names: set[str] = set()
for element in elements:
found = index.get(element)
if found:
names |= found
if names:
self._recompute(names)
def _touch_bindings(self, bindings: Iterable[InputBinding]) -> None:
"""The same, for a change that names bindings rather than elements."""
if self._actions_on:
self._touch(*{self._binding_element(b) for b in bindings})
def _touch_slots(self, slots: Iterable[tuple]) -> None:
"""The same, for a change that names stale-mark slots.
A slot carries an axis's direction and deadzone and the element key
does not, so an axis slot is cut back to the element before lookup.
"""
if not self._actions_on:
return
self._touch(*{slot[:2] if slot[0] == "device_axis" else slot for slot in slots})
def pressed(self, action: str) -> bool:
"""Is *action* currently open?
For a button action, whether any input mapped to it is held. For a
continuous one, whether its value is past the deadzone: the same
question in the only form a stick can answer it.
The held-state question, safe to ask from anywhere and as often as you
like: it reads the live state and consumes nothing, so two nodes asking
about the same action both get the answer.
An action a higher context hid or shadowed reads False. A name no
context declares raises, naming the nearest one it knows.
"""
state = self._read(action)
if state is None:
self._entry(action, "pressed")
answer = False
else:
answer = state.pressed
if _Input.verify_action_state and self._tick_verifiable:
self._verify("pressed", action, answer, self._derived_pressed(action))
return answer
def _derived_pressed(self, action: str) -> bool:
"""What the query resolved before it was stored. The oracle's answer."""
entry = self._entry(action, "pressed")
if entry is None:
return False
if entry.kind is not ActionKind.BUTTON:
return _off_rest(self._continuous(entry))
for b in entry.bindings:
if self._binding_pressed(b):
return True
return False
def _verify(self, query: str, action: str, stored: object, derived: object) -> None:
"""Raise unless the stored answer is the one the resolver would give.
On only under :attr:`verify_action_state`, which the test suite turns
on for a whole run. The store is written by every path that can change
an action, and this is what proves none of them was missed.
"""
if not (stored == derived):
raise AssertionError(
f"Input.{query}({action!r}) is stored as {stored!r} and resolves to {derived!r}. "
"Something changed the delivered record, the stale marks, the players or the "
"bindings without telling the action layer."
)
def value(self, action: str) -> bool | float | Vec2:
"""What *action* reads right now, in the type its bindings gave it.
``bool`` for an action bound to buttons alone, ``float`` for one
carrying an :class:`~.actions.Axis1D` or a bare axis, and
:class:`~simvx.core.math.types.Vec2` for one carrying an
:class:`~.actions.Axis2D` or a stick. A game binds movement once and reads it
once, whichever device the player picked up::
velocity = Input.value("move") * self.speed
**When several devices drive one action, the largest magnitude wins.**
A stick resting at ``(0.3, 0.9)`` beats keys nobody is pressing, and
keys held to ``(1, 0)`` beat a stick barely off centre, because
``(1, 0)`` is the longer vector and a pressed key is what the player
meant. Summing overshoots on the diagonal and last-device-wins jitters
under a resting hand; this is the one rule a player can predict.
An action a context hid or shadowed reads as at rest: False, 0.0 or a
zero vector, matching its declared type where it has one. A name no
context declares raises.
"""
state = self._read(action)
if state is None:
self._entry(action, "value")
answer: bool | float | Vec2 = _at_rest(self.input_map.kind(action))
else:
answer = state.value
if _Input.verify_action_state and self._tick_verifiable:
self._verify("value", action, answer, self._derived_value(action))
return answer
def _derived_value(self, action: str) -> bool | float | Vec2:
"""What the query resolved before it was stored. The oracle's answer."""
entry = self._entry(action, "value")
if entry is None:
return _at_rest(self.input_map.kind(action))
if entry.kind is ActionKind.BUTTON:
return self._derived_pressed(action)
return self._continuous(entry)
def _continuous(self, entry: Action, *, before: bool = False) -> float | Vec2:
"""The value of a continuous action, taking the source furthest from rest.
*before* reads the state as it was at the start of this frame, which is
what an edge on a continuous action is measured against: it has no
device edge of its own to carry, so the edge is the value crossing the
deadzone.
A tie goes to a source made of buttons: a key is full deflection by
nature, so a key held against a stick pushed just as far is the
player's deliberate choice, whatever order the sources were declared in.
"""
best: float | Vec2 = Vec2(0.0, 0.0) if entry.kind is ActionKind.AXIS_2D else 0.0
best_magnitude = 0.0
best_digital = False
for source in entry.sources:
candidate = self._source_value(source, before=before)
magnitude = candidate.length() if isinstance(candidate, Vec2) else abs(candidate)
if magnitude > best_magnitude:
best_magnitude, best, best_digital = magnitude, candidate, _digital(source)
elif magnitude == best_magnitude and magnitude > 0.0 and not best_digital and _digital(source):
best, best_digital = candidate, True
return best
def _continuous_edges(self) -> list[tuple[str, bool, int]]:
"""The continuous actions that have opened or closed since the last event.
A button action's edge is the device's own, which survives a press and
a release inside one frame. A continuous action has no such edge: keys,
a stick and a trigger all move one value, so the edge is that value
crossing the deadzone and it can only be found by comparing the action
with itself. That comparison is against the previous *event* rather
than the start of the frame, so two keys pressed in one frame open the
action once rather than twice.
Resolved once per player, because a continuous action is open for
whoever is pushing it: two players on one ``move`` action open and
close it independently, and each change names the player it belongs
to.
Called once per event by the resolve phase, which is the only caller:
the state it updates is what makes the comparison meaningful.
"""
names = self.input_map._continuous
if not names:
return []
changes: list[tuple[str, bool, int]] = []
for player_id in self._players:
open_for = self._open_at_last_event.setdefault(player_id, set())
with self._as_player(player_id):
for name in names:
entry = self._entry(name, "value")
if entry is None:
continue
open_now = _off_rest(self._continuous(entry))
if open_now != (name in open_for):
changes.append((name, open_now, player_id))
if open_now:
open_for.add(name)
else:
open_for.discard(name)
return changes
def _source_value(self, source: Source, *, before: bool) -> float | Vec2:
"""One source's whole contribution, with its own deadzone applied.
A composite's deadzone is radial: the vector is discarded entire when
it is shorter than the deadzone rather than clipped per axis, so a
stick pushed diagonally keeps its direction. A keyboard diagonal is
normalised to length one, so walking diagonally is not faster.
Every source of a 2D action is an :class:`~.actions.Axis2D`, because
:func:`~.actions.kind_for` refuses one that names no direction, so a
source only ever produces the type its action has.
"""
strength = self._binding_strength
if isinstance(source, Axis2D):
up, down, left, right = source.bindings
x = strength(right, before=before) - strength(left, before=before)
y = strength(down, before=before) - strength(up, before=before)
unbounded = _unbounded(source)
if unbounded and source.deadzone:
# The deadzone is the device's, measured before any scale or rate.
rx = strength(right, before=before, units=False) - strength(left, before=before, units=False)
ry = strength(down, before=before, units=False) - strength(up, before=before, units=False)
if (rx * rx + ry * ry) ** 0.5 < source.deadzone:
return Vec2(0.0, 0.0)
else:
length = (x * x + y * y) ** 0.5
if length < source.deadzone:
return Vec2(0.0, 0.0)
if length > 1.0 and not unbounded:
return Vec2(x / length, y / length)
return Vec2(x, y)
if isinstance(source, Axis1D):
low, high = source.bindings
x = strength(high, before=before) - strength(low, before=before)
if _unbounded(source):
if source.deadzone:
raw = strength(high, before=before, units=False) - strength(low, before=before, units=False)
if abs(raw) < source.deadzone:
return 0.0
return x
if abs(x) < source.deadzone:
return 0.0
return max(-1.0, min(1.0, x))
if isinstance(source, str):
return 0.0 # a name the map could not expand reads at rest
# A button in a scalar action pushes it positive.
return self._binding_strength(source, before=before)
# ----------------------------------------------------------------
# The physics tick
# ----------------------------------------------------------------
@property
def tick(self) -> TickInput | None:
"""The snapshot the running physics tick was given, or None outside one.
``on_fixed_update`` reads it through the ordinary queries and rarely
needs the object; take it to record a tick's input, to hand the same
input to a replay, or to read several fields without asking six
questions.
"""
return self._tick
def _snapshot(self) -> TickInput:
"""This tick's input: the device record and the action table, copied.
Copied rather than referenced, for two reasons that both matter. A
reference is not a snapshot: an event published from inside a tick, by
an on-screen control or a recogniser, would change what the rest of
that tick reads. And the value has to stand on its own to be recorded
or handed to a tick by a test.
The copy is three small containers plus one per connected pad, which
is why the action table was collapsed into one record per action: a
twenty-action table is one dict copy.
"""
self._fresh()
machine = self._devices
pads: dict[int, ControllerSnapshot] = {}
if machine.pads:
for pad_id, pad in machine.pads.items():
pads[pad_id] = ControllerSnapshot(
frozenset(pad.held), {index: float(v or 0.0) for index, v in pad.axis_values.items()}
)
pointer = self._last_pointer
edges: tuple[InputEdge, ...] = ()
if self._pending_edges:
edges = tuple(self._pending_edges)
self._pending_edges.clear()
# A delta and a wheel travel are the frame's accumulated change, not
# state, so they are spent by the tick that owns the frame's edges and
# read zero on every later one. A camera driven from a fixed update
# would otherwise turn three times as far on a frame that ran three
# ticks, which is the same defect an edge has and the same answer.
moved = self._tick_owns_edges
# The pointers are one dict copy, and a later tick of the frame reads
# them with the edges spent and the lifted fingers gone, exactly as the
# frame after would: a tap lands on one tick.
live = pointer.pointers
if moved or not pointer.edged:
pointers: Mapping[int, PointerState] = dict(live)
else:
pointers = {
pid: _without_edges(state)
for pid, state in live.items()
if state.contact or not state.just_released or state.pointer_type is not PointerType.TOUCH
}
return TickInput(
keys=frozenset(machine.keys_pressed),
mouse_buttons=frozenset(machine.mouse_buttons_pressed),
controllers=pads,
actions=dict(self._action_state),
mouse_position=pointer.position,
mouse_delta=pointer.delta if moved else (0.0, 0.0),
scroll_delta=pointer.scroll if moved else (0.0, 0.0),
pointers=pointers,
edges=edges,
)
def _record_edge(self, action: str, pressed: bool, player: int, when: float) -> None:
"""An action opened or closed: keep it for the next tick that runs.
Kept until a tick takes it rather than cleared at the frame boundary,
because a frame that ran no tick still happened: at 200 fps against a
60 Hz tick two frames in three run none, and a press on one of those
belongs to the next tick rather than to nobody.
"""
self._pending_edges.append(InputEdge(action, pressed, player, when))
@contextmanager
def _fixed_tick(self, snapshot: TickInput | None = None) -> Iterator[TickInput]:
"""Run a physics tick against *snapshot*, or against one taken now.
The first tick of a frame takes the frame's edges; every later tick of
the same frame sees none, so a press fires once whatever the display
and the tick rate are doing. A supplied snapshot is the whole truth
for its tick and takes nothing from the live record.
"""
previous, was_verifiable, owned = self._tick, self._tick_verifiable, self._tick_owns_edges
machine = self._devices
was_hidden = machine.tick_edges_hidden
self._ticked = True
self._tickless_boundaries = 0
self._ticks_this_frame += 1
if snapshot is None:
self._tick_owns_edges = self._ticks_this_frame == 1
snapshot = self._snapshot()
self._tick_verifiable = True
else:
# Nothing was taken from the device record, so its edges are not
# this tick's to read: what the caller supplied is.
self._tick_owns_edges = False
self._tick_verifiable = False
# A ``ControllerDevice`` is a live view over the process-wide record and has no
# tree to ask, so the one rule reaches it through the record itself.
# Saved and restored rather than assigned, because a host tree can
# step a guest tree from inside its own tick and each owns its first.
machine.tick_edges_hidden = not self._tick_owns_edges
self._tick = snapshot
try:
yield snapshot
finally:
self._tick, self._tick_verifiable, self._tick_owns_edges = previous, was_verifiable, owned
machine.tick_edges_hidden = was_hidden
def _tick_edge(self, action: str, pressed: bool) -> bool:
"""Whether this tick's snapshot carries that edge, for its player."""
player = self._current_player
for edge in self._tick.edges if self._tick is not None else ():
if edge.pressed is pressed and edge.action == action and edge.player == player:
return True
return False
def just_pressed(self, action: str) -> bool:
"""Did any input mapped to *action* go down?
Frame-scoped outside a physics tick and tick-scoped inside one: the
frame's edges are handed to its first tick and no other, so a press
fires once however many ticks the frame runs. Safe in
``on_fixed_update`` and in ``on_update`` alike.
An ``@on_input("jump")`` handler is still the sharper tool for a
discrete action: it runs once per event, before any tick, and is
handed the value and the player the press belonged to.
"""
if self._tick is not None:
self._entry(action, "just_pressed")
return self._tick_edge(action, True)
entry = self._entry(action, "just_pressed")
if entry is None:
return False
if entry.kind is not ActionKind.BUTTON:
return _off_rest(self._continuous(entry)) and not _off_rest(self._continuous(entry, before=True))
for b in entry.bindings:
if self._binding_just_pressed(b):
return True
return False
def just_released(self, action: str) -> bool:
"""Did any input mapped to *action* go up?
Frame-scoped outside a tick and tick-scoped inside one, as
:meth:`just_pressed`.
True also for an action a context change closed, which is what makes
that close a real edge: a game that stops sprinting on the release
sees the release when the pause menu takes the action away.
"""
if self._tick is not None:
self._entry(action, "just_released")
return self._tick_edge(action, False)
if self._forced_released and (self._current_player, action) in self._forced_released:
return True
entry = self._entry(action, "just_released")
if entry is None:
return False
if entry.kind is not ActionKind.BUTTON:
return _off_rest(self._continuous(entry, before=True)) and not _off_rest(self._continuous(entry))
for b in entry.bindings:
if self._binding_just_released(b):
return True
return False
def get_action_strength(self, action: str) -> float:
"""How hard *action* is being pushed, 0.0 to 1.0.
The magnitude of :meth:`value`, so a button action reads 1.0 while it
is held and a continuous one reads how far past the deadzone it is,
whichever device is driving it.
"""
state = self._read(action)
if state is None:
self._entry(action, "get_action_strength")
answer = 0.0
else:
answer = state.strength
if _Input.verify_action_state and self._tick_verifiable:
self._verify("get_action_strength", action, answer, self._derived_strength(action))
return answer
def _derived_strength(self, action: str) -> float:
"""What the query resolved before it was stored. The oracle's answer."""
entry = self._entry(action, "get_action_strength")
if entry is None:
return 0.0
if entry.kind is not ActionKind.BUTTON:
current = self._continuous(entry)
return current.length() if isinstance(current, Vec2) else abs(current)
strength = 0.0
for b in entry.bindings:
strength = max(strength, self._binding_strength(b))
return strength
def get_axis(self, negative_action: str, positive_action: str) -> float:
"""Return axis value from two opposing actions. Range [-1, 1]."""
return self.get_action_strength(positive_action) - self.get_action_strength(negative_action)
def get_vector(self, neg_x: str, pos_x: str, neg_y: str, pos_y: str) -> Vec2:
"""Return a normalized 2D direction vector from four input actions.
Handles diagonal normalization so magnitude never exceeds 1.0.
"""
x = self.get_action_strength(pos_x) - self.get_action_strength(neg_x)
y = self.get_action_strength(pos_y) - self.get_action_strength(neg_y)
v = Vec2(x, y)
ln = v.length()
return v / ln if ln > 1.0 else v
# ----------------------------------------------------------------
# Players
# ----------------------------------------------------------------
@property
def players(self) -> Mapping[int, InputPlayer]:
"""Everyone playing, keyed by player id.
Player 0 always exists and is the default: it owns every device no
other player has been paired to, which is why a single-player game
never has to know this mapping is here. :meth:`add_player` adds the
next one.
Per tree rather than per process, unlike ``Input.controllers``: who is
playing is a fact about the game, while a controller is plugged into
the machine.
"""
return MappingProxyType(self._players)
def add_player(self) -> InputPlayer:
"""Add a player and return it, taking the lowest free id.
Nothing is paired to it yet, so it reads nothing until a device is::
two = Input.add_player()
two.pair(Input.controllers[1])
This is the seam local co-op sits on, not the join flow: deciding
*when* somebody has joined, from a button on an unpaired pad or a
lobby screen, is the game's.
"""
player_id = next(i for i in count() if i not in self._players)
player = self._players[player_id] = InputPlayer(self, player_id)
self._open_at_last_event[player_id] = set()
self._repair_ownership()
return player
def remove_player(self, player: InputPlayer | int) -> None:
"""Remove a player, returning its devices to the unpaired pool.
The default player cannot be removed: something has to own a device
nobody claimed.
"""
player_id = player if isinstance(player, int) else player.id
if player_id == DEFAULT_PLAYER:
raise ValueError(
"player 0 is the default and cannot be removed: it owns every device no other "
"player is paired to. Unpair its devices instead."
)
self._players.pop(player_id, None)
self._open_at_last_event.pop(player_id, None)
self._repair_ownership()
def _repair_ownership(self) -> None:
"""Recompute what the default player owns, after a pairing changed.
Called by every pairing mutation, which is the whole point: the two
sets below are a cache of "claimed by somebody other than player 0",
and the default player's ownership test reads them instead of walking
every player on every binding query.
"""
self._solo = len(self._players) == 1
self._paired_pads = set()
self._paired_desk = False
for player in self._players.values():
if player.is_default:
continue
self._paired_pads |= player._pads
self._paired_desk = self._paired_desk or player._desk
# Who hears which device decides what every action reads, and the
# answers are kept per player, so a pairing resolves the whole table.
if self._actions_on:
self._recompute(self.input_map._actions.keys())
@contextmanager
def _as_player(self, player_id: int) -> Iterator[None]:
"""Answer the action queries for *player_id*, for the duration.
What :attr:`Node.input_player` opens around a subtree's code, so
``Input.pressed("jump")`` inside player two's subtree answers for
player two with no argument.
"""
previous = self._current_player
self._current_player = player_id
try:
yield
finally:
self._current_player = previous
def _owns_desk(self) -> bool:
"""Whether the player being answered for hears the keyboard and mouse."""
player = self._players.get(self._current_player)
return player is not None and player._owns_desk()
def _owns_pad(self, pad_id: int) -> bool:
"""Whether the player being answered for hears *pad_id*."""
player = self._players.get(self._current_player)
return player is not None and player._owns_pad(pad_id)
def _desk_players(self) -> list[int]:
"""Who hears the keyboard and mouse. ``[0]`` while there is one player."""
if self._solo:
return _SOLO
return [p.id for p in self._players.values() if p._owns_desk()]
def _pad_players(self, pad_id: int) -> list[int]:
"""Who hears *pad_id*. ``[0]`` while there is one player."""
if self._solo:
return _SOLO
return [p.id for p in self._players.values() if p._owns_pad(pad_id)]
def _owned_pads(self) -> list[int]:
"""The pads the player being answered for hears, for a binding naming none."""
return [pad_id for pad_id in self._devices.pads if self._owns_pad(pad_id)]
def _binding_heard(self, b: InputBinding) -> bool:
"""Whether the player being answered for hears the device this binding names.
Only asked once a second player exists: with one, the default player
owns everything and the resolver never reaches this. A binding naming
no pad is heard by everyone, because the pad union it resolves against
is itself narrowed to this player's pads.
"""
control = b.control
if control is None:
return True
if control.device is DeviceType.CONTROLLER:
# A pad binding naming no unit is heard by everyone, because what
# it reads is itself narrowed to this player's pads.
return control.unit is None or self._owns_pad(control.unit)
return self._owns_desk()
# ----------------------------------------------------------------
# Device facts
# ----------------------------------------------------------------
@property
def capabilities(self) -> InputCapabilities:
"""What the machine's input devices can do.
A property over one process-wide value rather than per-instance state:
the hardware is the same whichever tree is asking, so an editor and the
game it is running read the same answers. Read-only here, because a
capability is stated by whatever knows it through
:func:`simvx.core.input.capabilities.declare`, and nothing else may
claim one.
"""
return _capabilities.current()
@property
def input_method(self) -> InputMethod:
"""What the player is using right now.
Not what the machine has, which is :attr:`capabilities`: this is the
kind of device the player last acted with, and it is what a prompt
should be drawn for.
**At startup** it is touch when the pointer is coarse, else controller
when a controller is already plugged in, else keyboard and mouse. That
answer is settled on the first read rather than at construction,
because a backend declares its facts after this object exists.
**It then switches** on a key press, a controller button, a controller axis
crossing its deadzone, a touch press, or a mouse or pen press; and on
mouse motion only once the cursor has travelled
:attr:`mouse_method_threshold`. Never on hover alone, so a cursor
resting under a moving window does not claim the player picked up the
mouse.
**Only a device-sourced pad switches it.** The on-screen controller
publishes as a pad so a game binding ``Controller.A`` gets touch play
for free, and if its own presses counted it would switch the method to
CONTROLLER and hide itself under the player's thumb.
**Process-wide**, like :attr:`capabilities` and :attr:`controllers`: the
player is one person holding one thing, so an editor and the game it is
running agree about what that is.
"""
if self._devices.input_method is None:
self._devices.input_method = self._startup_input_method()
return self._devices.input_method
@property
def mouse_method_threshold(self) -> float:
"""How far the mouse must travel before :attr:`input_method` follows it.
In engine units. A nudged desk moves a cursor a pixel or two, and
switching on that would hide an on-screen pad under the player's thumb;
reaching for the mouse moves it much further. Raise it for a game
played on a shaking surface, lower it to nothing to switch on any
motion at all. Process-wide, because the method it governs is.
"""
return self._devices.mouse_method_threshold
@mouse_method_threshold.setter
def mouse_method_threshold(self, value: float) -> None:
self._devices.mouse_method_threshold = float(value)
@property
def controller_connected(self) -> Signal:
"""A pad arrived. Handlers receive the :class:`~.devices.ControllerDevice`,
which is already in :attr:`controllers` by the time this fires.
The signal itself is process-wide, so a handler connected through one
tree hears a pad another tree's poll found: a controller is plugged in
once however many trees are running.
"""
return self._devices.controller_connected
@property
def controller_disconnected(self) -> Signal:
"""A pad left. Handlers receive the :class:`~.devices.ControllerDevice`, which
has already gone from :attr:`controllers` and reads as an untouched pad.
"""
return self._devices.controller_disconnected
@property
def input_method_changed(self) -> Signal:
"""The player picked up a different kind of device. Handlers receive
the new :class:`~.enums.InputMethod`; connect to reskin prompts.
"""
return self._devices.input_method_changed
def prompt(self, action: str, *, device_id: int | None = None) -> str:
"""What to draw for *action* on the device the player is holding.
``"Press {Input.prompt('jump')}"`` reads "Press Space" at a keyboard,
"Press A" on an Xbox pad, "Press Cross" on a PlayStation one and "Press
B" on a Nintendo one, because the engine's button codes are positional
and only the printing moves.
Which binding is labelled follows :attr:`input_method`: a pad binding
while the player is on a pad or a touchscreen (the on-screen pad prints
pad labels), a key or mouse binding otherwise. Where the chosen kind
has no binding the other kind answers, so an action bound to a key
alone still prompts on a controller.
Args:
action: The action to label. An action with no bindings, or one
that was never registered, gives ``""``: a prompt is a piece of
presentation, and raising here would turn a missing label into
a crash on the title screen.
device_id: Which pad to label for. Defaults to the lowest-numbered
connected one, which is the right answer for a single-player
game and the wrong one for player two.
"""
bindings = self.input_map.get_bindings(action)
if not bindings:
return ""
pad_first = self.input_method is not InputMethod.KEYBOARD_MOUSE
def on_pad(b: InputBinding) -> bool:
control = b.control
if control is None or control.device is not DeviceType.CONTROLLER:
return False
return device_id is None or control.unit is None or control.unit == device_id
def on_desk(b: InputBinding) -> bool:
control = b.control
return isinstance(control, DeviceButton) and control.device is not DeviceType.CONTROLLER
for wanted in (on_pad, on_desk) if pad_first else (on_desk, on_pad):
for b in bindings:
if wanted(b):
label = self.label(b, device_id=device_id)
if label:
return label
return ""
def label(
self, control: InputBinding | DeviceControl | Axis1D | Axis2D | str, *, device_id: int | None = None
) -> str:
"""What to print for one control, as the player's hardware names it.
``Input.label(Key.SPACE)`` is "Space", ``Input.label(Mouse.RIGHT)`` the
mouse's right button as its layout names it, ``Input.label(Controller.A)``
"Cross" while a PlayStation pad is connected, and
``Input.label(Controller.LEFT_X.positive)`` "Left Stick Right". A binding
prints its chord too ("Ctrl+S"), and a composed source prints the
stick it is ("Left Stick") or its parts ("W/S/A/D"). A raw control
prints as the hardware counts it ("Button 6", "Axis 3 +"), and an
entry's name prints what it resolves to. This is what a rebind screen
draws for each row, and what :meth:`prompt` draws for the binding it
picks.
Args:
control: A control of any device type, a binding, one of an
entry's sources, or an entry's name.
device_id: Which pad names a pad control. Defaults to the one the
control is pinned to, else the lowest-numbered connected pad.
"""
from .prompts import axis_label, button_label, key_binding_label, pointer_button_label
if isinstance(control, str):
resolved = self.input_map.get_bindings(control)
return self.label(resolved[0], device_id=device_id) if resolved else ""
if isinstance(control, Axis1D | Axis2D):
axes = [part.control if isinstance(part, InputBinding) else None for part in control.parts]
if all(isinstance(a, DeviceAxis) and a.device is DeviceType.CONTROLLER and a.name for a in axes):
names = {axis_label(cast(DeviceAxis, a).whole) for a in axes}
if len(names) == 1:
return names.pop() # the halves of one stick: "Left Stick"
return "/".join(self.label(part, device_id=device_id) for part in control.parts)
one: DeviceControl | None
if isinstance(control, InputBinding):
if control.key_combo:
return key_binding_label(control.key_combo)
one = control.control
else:
one = control
if one is None:
return ""
if one.index is not None and one.device is not DeviceType.KEYBOARD:
# A raw control: what the hardware counts it as, one-based.
if isinstance(one, DeviceAxis):
sign = "" if not one.half else (" +" if one.half > 0 else " -")
return f"Axis {one.index + 1}{sign}"
return f"Button {one.index + 1}"
if one.device is DeviceType.CONTROLLER:
pads = self._devices.objects
pad_id = device_id if device_id is not None else one.unit
pad = pads.get(pad_id) if pad_id is not None else None
if pad is None and pads:
pad = pads[min(pads)]
family = pad.family if pad is not None else ControllerFamily.GENERIC
if isinstance(one, DeviceAxis):
return axis_label(one, family)
if isinstance(one, DeviceStick):
return axis_label(one.x, family) # the stick's own name, "Left Stick"
if isinstance(one, DeviceButton):
return button_label(one.on(None), family)
if isinstance(one, DeviceButton):
if one.device is DeviceType.KEYBOARD:
return key_binding_label(key_to_name(one))
return pointer_button_label(one)
return str(one)
def _startup_input_method(self) -> InputMethod:
"""The method to open with, from what the machine has said so far."""
if _capabilities.current().coarse_pointer:
return InputMethod.TOUCH
if any(pad.source is ControllerSource.DEVICE and pad.connected for pad in self._devices.pads.values()):
return InputMethod.CONTROLLER
return InputMethod.KEYBOARD_MOUSE
def _note_input_method(self, method: InputMethod) -> None:
"""Record that the player just used *method*, and tell anyone watching.
Called from the router's device-state phase, which is never gated by a
claim: a key a menu swallowed was still typed on a keyboard.
"""
if self._devices.input_method is None:
self._devices.input_method = self._startup_input_method()
if method is not InputMethod.KEYBOARD_MOUSE:
# Travel is measured from the last time the mouse was not in
# charge, so a player who put the mouse down and picked up a pad
# has to reach back for it rather than inheriting old motion.
self._devices.mouse_method_travel = 0.0
if self._devices.input_method is method:
return
self._devices.input_method = method
self.input_method_changed.emit(method)
def _note_mouse_travel(self, dx: float, dy: float) -> None:
"""Mouse motion, which switches the method only once it adds up."""
if self.input_method is InputMethod.KEYBOARD_MOUSE:
return
self._devices.mouse_method_travel += abs(dx) + abs(dy)
if self._devices.mouse_method_travel >= self._devices.mouse_method_threshold:
self._note_input_method(InputMethod.KEYBOARD_MOUSE)
# ----------------------------------------------------------------
# Typed key query API
# ----------------------------------------------------------------
def is_key_pressed(self, key: DeviceButton) -> bool:
"""Whether *key* is physically down, whoever the input went to.
The device record, which is one machine: a key held while an editor
runs a game reads True in both, even where the editor forwarded
nothing. To ask what this tree was actually given, ask the action it
is bound to -- that is what the action layer answers.
Inside a physics tick this is the tick's snapshot of the machine, so
the answer cannot change halfway through one tick.
"""
code = _code(key)
tick = self._tick
if tick is not None:
return code in tick.keys
return code in self._devices.keys_pressed
def is_key_just_pressed(self, key: DeviceButton) -> bool:
"""Whether *key* went down this frame, on the machine.
Inside a physics tick, whether it went down on the frame this tick
opens: the frame's device edges belong to its first tick and to no
other, the same rule the action edges follow.
"""
if self._tick is not None and not self._tick_owns_edges:
return False
return _code(key) in self._devices.keys_just_pressed
def is_key_just_released(self, key: DeviceButton) -> bool:
"""Whether *key* came up this frame, on the machine.
Tick-scoped inside a tick, as :meth:`is_key_just_pressed`.
"""
if self._tick is not None and not self._tick_owns_edges:
return False
return _code(key) in self._devices.keys_just_released
# ----------------------------------------------------------------
# Mouse
# ----------------------------------------------------------------
def _mouse_layout(self) -> DeviceLayout:
"""The machine's mouse's layout, overrides applied."""
return self._devices.layout_of(self._devices.device(DeviceType.MOUSE))
def _mouse_index(self, button: DeviceButton) -> int | None:
"""The mouse's own index for *button*, through the mouse's layout.
A role resolves to the mouse's member of it, so ``Pointer.PRIMARY`` is
the left button, wherever the layout puts it; a button of another
device type is not on the mouse at all.
"""
return self._mouse_layout().button_index(button)
def is_mouse_button_pressed(self, button: DeviceButton) -> bool:
"""Whether *button* is physically down on the mouse, whoever the input went to.
The device record, like :meth:`is_key_pressed`, and the tick's
snapshot of it inside a tick. *button* resolves through the mouse's
layout, so ``Mouse.RIGHT`` is whichever index the mouse calls
right.
"""
index = self._mouse_index(button)
if index is None:
return False
tick = self._tick
if tick is not None:
return index in tick.mouse_buttons
return index in self._devices.mouse_buttons_pressed
def is_mouse_button_just_pressed(self, button: DeviceButton) -> bool:
"""Whether *button* went down this frame, on the machine.
Tick-scoped inside a tick, as :meth:`is_key_just_pressed`.
"""
if self._tick is not None and not self._tick_owns_edges:
return False
index = self._mouse_index(button)
return index is not None and index in self._devices.mouse_buttons_just_pressed
def is_mouse_button_just_released(self, button: DeviceButton) -> bool:
"""Whether *button* came up this frame, on the machine.
Tick-scoped inside a tick, as :meth:`is_key_just_pressed`.
"""
if self._tick is not None and not self._tick_owns_edges:
return False
index = self._mouse_index(button)
return index is not None and index in self._devices.mouse_buttons_just_released
@property
def devices(self) -> Mapping[int, InputDevice]:
"""Every device, by the engine's id for it.
The keyboard, the mouse, each touchscreen, each pen and each tablet the
backend announced or an event revealed. A rebind screen lists a
device's :attr:`~.devices.InputDevice.buttons` and prints its
:meth:`~.devices.InputDevice.label`; ``event.device_id`` is the key.
Process-wide, like :attr:`controllers`: a device is plugged into the
machine, not into a tree.
"""
return MappingProxyType(self._devices.device_objects)
def _pointer(self, source_id: int) -> _PointerSource:
"""The pointer belonging to *source_id*, created the first time it moves."""
source = self._sources.get(source_id)
if source is None:
source = self._sources[source_id] = _PointerSource()
return source
@property
def _last_pointer(self) -> _PointerSource:
"""The pointer of the source that moved here last.
The source-less queries answer for it, because with one window, which
is every game, there is nothing else to answer for. A tree fed by two
windows has two cursors and this is the one that moved.
"""
source = self._sources.get(self._last_source)
return source if source is not None else self._pointer(self._last_source)
# The source-less spellings of one source's pointer, which answer for the
# source that moved last. Everything that names no window reads and writes
# through these: a cursor warp, a recorder, the debug overlay. A caller
# that does know which window it means -- the router, which is one per
# window -- goes through ``_pointer`` instead.
@property
def _mouse_pos(self) -> tuple[float, float]:
return self._last_pointer.position
@_mouse_pos.setter
def _mouse_pos(self, value: tuple[float, float]) -> None:
self._last_pointer.position = value
@property
def _mouse_delta(self) -> tuple[float, float]:
return self._last_pointer.delta
@_mouse_delta.setter
def _mouse_delta(self, value: tuple[float, float]) -> None:
self._last_pointer.delta = value
@property
def _scroll_delta(self) -> tuple[float, float]:
return self._last_pointer.scroll
@_scroll_delta.setter
def _scroll_delta(self, value: tuple[float, float]) -> None:
self._last_pointer.scroll = value
@property
def _discard_next_mouse_delta(self) -> bool:
return self._last_pointer.discard_next_delta
@_discard_next_mouse_delta.setter
def _discard_next_mouse_delta(self, value: bool) -> None:
self._last_pointer.discard_next_delta = value
@property
def mouse_position(self) -> Vec2:
"""Where the pointer is, in screen coordinates.
The pointer of the source that moved last. A pointer belongs to its
source: two windows onto one tree have two of them, and this answers
for the one the player last moved. Inside a physics tick it is where
the pointer was when the tick started.
"""
tick = self._tick
x, y = tick.mouse_position if tick is not None else self._last_pointer.position
return Vec2(x, y)
@property
def mouse_delta(self) -> Vec2:
"""How far that pointer has moved this frame."""
tick = self._tick
dx, dy = tick.mouse_delta if tick is not None else self._last_pointer.delta
return Vec2(dx, dy)
@property
def scroll_delta(self) -> tuple[float, float]:
"""Scroll wheel delta this frame (x, y), for the pointer that moved last."""
tick = self._tick
return tick.scroll_delta if tick is not None else self._last_pointer.scroll
@property
def mouse_wheel_y(self) -> float:
"""Vertical scroll-wheel delta this frame.
Scalar shortcut for the common case: most consumers only want
``Input.scroll_delta[1]`` (PirateMaker measured 90% of scroll-delta
uses), and ``Input.mouse_wheel_y`` reads cleaner than ``[1]``
indexing.
"""
return self._last_pointer.scroll[1]
@property
def mouse_wheel_x(self) -> float:
"""Horizontal scroll-wheel delta this frame. Companion to ``mouse_wheel_y``."""
return self._last_pointer.scroll[0]
def set_mouse_capture_mode(self, mode: MouseCaptureMode):
"""Set mouse cursor capture mode. Platform adapter applies the change."""
if mode != self._capture_mode:
# Toggling capture warps / hides / recentres the cursor, so the next
# motion event carries a one-off jump that must not drive the camera.
# Every source, because the mode is the tree's and the caller names
# no window; with one window, which is every game, it is the one.
self.skip_next_mouse_delta()
self._capture_mode = mode
if self._capture_mode_callback:
self._capture_mode_callback(mode)
def begin_text_input(self, rect: tuple[float, float, float, float]) -> None:
"""Open the platform's text-input session over *rect*.
What a platform does with this varies and the engine does not care:
a phone slides its soft keyboard up and scrolls *rect* clear of it, a
desktop IME puts its candidate window beside *rect*, a console opens
its own entry screen. A backend that can do none of those installs no
callback and this changes nothing but the recorded rect.
Args:
rect: Where the text being entered is, as ``(x, y, width, height)``
in engine units, in the space :attr:`SceneTree.screen_size`
describes. Call again with a new rect to move the session,
which is what a field scrolling under the caret needs; that is
a move rather than a re-open, and a keyboard already up stays
up.
Focus drives this on its own: a focused control whose
``wants_text_input`` is True opens a session and closing it follows
the focus away. Call it directly for what focus cannot know about, an
on-screen keyboard button or a swipe gesture opening one.
A backend that can open a keyboard declares
``Input.capabilities.soft_keyboard``. Read that before offering a
player an on-screen keyboard button, since on a desktop the button
would do nothing.
"""
self._text_input_rect = (float(rect[0]), float(rect[1]), float(rect[2]), float(rect[3]))
if self._text_input_callback:
self._text_input_callback(self._text_input_rect)
def end_text_input(self) -> None:
"""Close the platform's text-input session. Idempotent."""
if self._text_input_rect is None:
return
self._text_input_rect = None
if self._text_input_callback:
self._text_input_callback(None)
@property
def touch_secondary(self) -> DeviceButton | None:
"""The finger gesture that is a right click here: ``Touch.LONG_PRESS``, ``Touch.TWO_FINGER_TAP`` or None.
None by default, so a finger never right-clicks unless a game asks.
Set, that gesture is in ``Pointer.SECONDARY``: a binding and a context
menu on the secondary press hear it, recognising it cancels the press
that began it so it is not also a click, and it never fires after the
finger travelled or while a control holds the finger.
"""
return self._touch_secondary
@touch_secondary.setter
def touch_secondary(self, gesture: DeviceButton | None) -> None:
from .controls import Touch
if gesture is not None and gesture not in (Touch.LONG_PRESS, Touch.TWO_FINGER_TAP):
raise ValueError(f"touch_secondary is Touch.LONG_PRESS, Touch.TWO_FINGER_TAP or None; got {gesture!r}")
self._touch_secondary = gesture
@property
def cursor_shape(self) -> CursorShape:
"""The shape the operating system's pointer shows over this tree.
Read-only: it follows ``Control.cursor_shape`` on the control the
mouse or a pen is over, which is where a game sets it.
"""
return self._cursor_shape
def _show_cursor_shape(self, shape: CursorShape) -> None:
"""Record *shape* and hand it to the platform, which shows it."""
self._cursor_shape = shape
if self._cursor_shape_callback:
self._cursor_shape_callback(shape)
@property
def text_input_rect(self) -> tuple[float, float, float, float] | None:
"""The rect of the open text-input session, or None when there is none.
Doubles as the "is one open" question, because the two answers are the
same fact and a game that wants to lay out around a keyboard wants the
rect anyway.
"""
return self._text_input_rect
def skip_next_mouse_delta(self):
"""Drop the next motion event's delta (re-baseline only).
Call across any cursor discontinuity the input layer can't infer on its
own, e.g. a manual cursor warp or a window focus regain under capture.
Applies to every source: the caller names no window, and with one
window, which is every game, there is only the one.
"""
for source in self._sources.values():
source.discard_next_delta = True
def get_mouse_capture_mode(self) -> MouseCaptureMode:
"""Get the current mouse capture mode."""
return self._capture_mode
def is_mouse_captured(self) -> bool:
"""Whether the cursor is currently grabbed / locked to the window.
Returns ``True`` when the active capture mode is anything other than
:attr:`MouseCaptureMode.VISIBLE`: that is, ``HIDDEN``, ``CAPTURED``,
or ``CONFINED``. FPS-style scenes flip this on at start of gameplay
and off when a menu opens; verification harnesses (Q1K3) read it to
confirm pointer-lock state without mode-by-mode comparisons.
"""
return self._capture_mode != MouseCaptureMode.VISIBLE
# ----------------------------------------------------------------
# Controller
# ----------------------------------------------------------------
@property
def controllers(self) -> Mapping[int, ControllerDevice]:
"""The pads readable right now, keyed by the engine's id for each, as in :attr:`devices`.
A pad appears here from the first poll that reports it and goes on the
first poll that does not, so unplugging a controller empties the
mapping and a game can pause rather than run on the last reading
forever. Ids need not be contiguous.
The mapping is a read-only view of live state rather than a copy, and
it only ever changes during the frame's input pump, before any node
code runs: a game iterating it in ``on_update`` sees one frame's pads
throughout, and never one part-way through being torn down. A
:class:`~.devices.ControllerDevice` a game keeps hold of stays usable after its
pad has gone and reports ``connected`` False.
A pad published by ``InputSimulator.set_controller`` counts as connected,
so a headless test sees what a windowed run sees, and the on-screen pad
appears alongside the hardware. Filter on
:attr:`~.devices.ControllerDevice.source` for only the physical ones. A backend
with no controller support reports nothing and prunes nothing: the mapping
stays empty rather than claiming every pad just left.
**Process-wide**, like :attr:`capabilities` and :attr:`input_method`: a
controller is plugged into the machine rather than into a scene tree,
so an editor and the game it is running hold the same pad. What a pad
*presses* is another matter and is per tree: an action resolves from
the delivered record, so a pad button the host did not forward
presses nothing in the game it is hosting.
"""
return MappingProxyType(self._devices.objects)
# ----------------------------------------------------------------
# Internal: called by platform adapters
# ----------------------------------------------------------------
def _on_key(self, key: int, pressed: bool, device_id: int = 0):
"""One key event, into the device record and into this tree's delivered one.
Both layers are written here because both are true of the same event:
the machine is holding the key, and this tree was given it. A tree the
event never reached never runs this, which is the whole of why its
actions stay at rest while ``is_key_pressed`` says the key is down.
The device write is idempotent, so a host and the tree it forwards to
both running it lands the same key in the same set.
Mirrors the press into the string-keyed ``_keys`` state as well, so the
two key-tracking conventions stay in lock-step at this single
synchronisation point. Platform adapters already write both; this makes
synthetic events from ``InputSimulator``, which only reach this typed
path: drive the name-based consumers too (the shortcut listener and
the modifier-aware editor gestures that poll by name). The
string name comes from the same key-name map the adapters use, so
the redundant write from the platform path is idempotent.
"""
name = _name_for_code(key)
machine = self._devices
if pressed:
if key not in machine.keys_pressed:
machine.keys_just_pressed.add(key)
machine.keys_pressed.add(key)
if key not in self._keys_delivered:
self._keys_delivered_down.add(key)
self._pressed_again(("key", key))
self._keys_delivered.add(key)
if device_id:
unit = self._unit_keys.setdefault(device_id, set())
if key not in unit:
self._unit_keys_down.setdefault(device_id, set()).add(key)
unit.add(key)
if not self._keys.get(name):
self._keys_just_pressed[name] = True
self._keys[name] = True
else:
machine.keys_pressed.discard(key)
machine.keys_just_released.add(key)
# A release lands where its press landed: a key this tree was
# never given produces no edge here, which is what keeps a
# focus-loss replay of the machine's keys from raising a release
# in a tree that never saw them go down.
if key in self._keys_delivered:
self._keys_delivered.discard(key)
self._keys_delivered_up.add(key)
if device_id and key in self._unit_keys.get(device_id, ()):
self._unit_keys[device_id].discard(key)
self._unit_keys_up.setdefault(device_id, set()).add(key)
self._keys[name] = False
self._keys_just_released[name] = True
self._touch(("key", key))
def _on_mouse_button(self, button: int, pressed: bool):
"""One mouse button by its index, into the device record.
What the machine is holding, which ``is_mouse_button_pressed``
answers. What this tree was given is :meth:`_on_button`, by name.
"""
machine = self._devices
if pressed:
if button not in machine.mouse_buttons_pressed:
machine.mouse_buttons_just_pressed.add(button)
machine.mouse_buttons_pressed.add(button)
else:
machine.mouse_buttons_pressed.discard(button)
machine.mouse_buttons_just_released.add(button)
def _on_button(
self, press: tuple, names: Iterable[DeviceButton], pressed: bool, device_id: int | None = None
) -> None:
"""One button press or release into this tree's delivered record, by name.
*press* identifies the press, so its release takes away exactly the
names it added: ``("mouse", device id, index)``, ``("pointer", pointer id,
index)`` or ``("device", device id, index)``. *names* are every name
the press answers to under the layout in force; a binding holds while
its own name is among them. A release of a press this tree was never
given changes nothing, as a key's does.
"""
if pressed:
if press in self._button_presses:
return
keys: tuple[object, ...] = tuple(names)
if device_id is not None:
keys += tuple((device_id, name) for name in names)
self._button_presses[press] = keys
held = self._buttons_delivered
for key in keys:
count = held.get(key, 0)
if not count:
self._buttons_delivered_down.add(key)
self._pressed_again(("button", key))
held[key] = count + 1
else:
keys = self._button_presses.pop(press, ())
if not keys:
return
held = self._buttons_delivered
for key in keys:
count = held.get(key, 0) - 1
if count <= 0:
held.pop(key, None)
self._buttons_delivered_up.add(key)
else:
held[key] = count
self._touch(*(("button", key) for key in keys))
def _abandon_button(self, press: tuple) -> None:
"""Drop one press from the delivered record with no release edge."""
keys = self._button_presses.pop(press, ())
if not keys:
return
held = self._buttons_delivered
for key in keys:
count = held.get(key, 0) - 1
if count <= 0:
held.pop(key, None)
else:
held[key] = count
self._touch(*(("button", key) for key in keys))
def _on_device_axis(self, names: Iterable[DeviceAxis], value: float | None, delta: float, device_id: int) -> None:
"""One axis report into this tree's delivered record, under every control it answers to.
Each name is filed twice: on the device that moved it, for a binding
pinned to that unit, and plain, for a binding on any unit. The plain
value is the one furthest from rest among every device that reported
the axis, so two pads on one ``Controller.LEFT_X`` do not fight. A report
that repeats a position already delivered adds no motion, which is what
lets the poll and the event it queued both deliver one pad move.
"""
values = self._axes_delivered
prev = self._axes_delivered_prev
deltas = self._axis_deltas_delivered
touched: list[tuple] = []
for name in names:
sources = self._axis_sources.setdefault(name, {})
repeated = value is not None and device_id in sources and sources[device_id] == value
sources[device_id] = value
known = [v for v in sources.values() if v is not None]
plain = max(known, key=abs) if known else None
for key, now in ((name, plain), ((device_id, name), value)):
if key not in prev:
prev[key] = values.get(key)
values[key] = now
if delta and not repeated:
deltas[key] = deltas.get(key, 0.0) + delta
touched.append(("device_axis", key))
self._touch(*touched)
def _button_forms(self, event: PointerEvent | ButtonEvent) -> tuple[DeviceButton, ...]:
"""Every control a handler filter can name *event*'s button by.
The control the layout names, its index form and every role it is a
member of, so a filter on any of them hears the press, and for the
pointer standing in for the mouse, the mouse's own primary button, so
a game written for a mouse hears a thumb. Unlike :meth:`_binding_names`
this does not narrow roles to the primary pointer: a handler filtered
on a role hears every pointer.
"""
button = event.button
if button is None:
return ()
forms = self._binding_names(event)
if isinstance(event, PointerEvent) and not event.is_primary:
return (*forms, *roles_of(button))
return tuple(forms)
def _axis_forms(self, event: AxisEvent) -> frozenset[DeviceAxis]:
"""The controls a binding can name *event*'s axis by: its raw control and the entries naming it."""
axis = event.axis
rec = self._devices.device_records.get(event.device_id)
layout = self._devices.layout_of(rec) if rec is not None else None
index = layout.axis_index(axis) if layout is not None else axis.index
if layout is None or index is None:
return frozenset({axis})
return self._axis_names(layout, index, event.device_id)
def _axis_names(self, layout: DeviceLayout, index: int, unit: int | None = None) -> frozenset[DeviceAxis]:
"""What a report of the backend's axis *index* puts in front of the bindings.
Its raw control, in the engine's own numbering, and the engine entry
the backend names it by, renamed through this tree's map where an
entry was redefined: an entry resolving elsewhere drops out, and one
resolving here joins.
"""
raw = raw_axis(layout, index)
named = layout.axis_control(index)
names = [named] if named.index is None else []
table = self.input_map
if table._redirects:
names = [name for name in names if name not in table._redirects]
names.extend(name for name in table._redirected_from.get(raw, ()) if isinstance(name, DeviceAxis))
if unit is not None and unit in table._unit_redirects:
own = table._unit_redirects[unit]
names = [name for name in names if name not in own]
names.extend(
name for name in table._unit_redirected_from[unit].get(raw, ()) if isinstance(name, DeviceAxis)
)
return frozenset({raw, *names})
def _raw_control(self, control: DeviceButton | DeviceAxis, device_id: int) -> DeviceButton | DeviceAxis:
"""The raw control behind *control* as device *device_id* reported it.
A key is raw already. A button or an axis the backend named is turned
back into its index in the engine's own numbering, the form a rebind
of an engine entry takes and :meth:`InputMap.aliases` looks up; a half
of an axis stays that half.
"""
if control.device is DeviceType.KEYBOARD:
return control
rec = self._devices.device_records.get(device_id)
if rec is None:
return control
layout = self._devices.layout_of(rec)
if isinstance(control, DeviceAxis):
index = layout.axis_index(control.whole)
if index is None:
return control
raw = raw_axis(layout, index)
if control.half:
raw = raw.positive if control.half > 0 else raw.negative
return raw.delta if control.is_delta else raw
index = layout.button_index(control)
return raw_button(layout, index) if index is not None else control
def _press_names(
self, layout: DeviceLayout, index: int, *, roles: bool = True, unit: int | None = None
) -> frozenset[DeviceButton]:
"""What a press of the backend's button *index* puts in front of the bindings.
Its raw control in the engine's own numbering, the engine entry the
backend names it by and that entry's roles, all renamed through this
tree's map: an entry redefined to resolve elsewhere drops out, one
redefined to resolve here joins, with its roles, and on *unit* an
entry the player mapped on that device alone does the same. Nothing
is renamed, and nothing is looked up, while no entry is redefined.
"""
raw = raw_button(layout, index)
named = layout.button_control(index)
names: list[DeviceButton] = [named] if named.index is None else []
table = self.input_map
redirects = table._redirects
if redirects:
names = [name for name in names if name not in redirects]
names.extend(name for name in table._redirected_from.get(raw, ()) if isinstance(name, DeviceButton))
own: Mapping[DeviceControl, DeviceControl] = {}
if unit is not None and unit in table._unit_redirects:
own = table._unit_redirects[unit]
names = [name for name in names if name not in own]
names.extend(
name for name in table._unit_redirected_from[unit].get(raw, ()) if isinstance(name, DeviceButton)
)
found: set[DeviceButton] = {raw, *names}
if roles:
for name in names:
found.update(roles_of(name))
if self._touch_secondary is not None and self._touch_secondary in names:
found.add(Pointer.SECONDARY)
if redirects or own:
# A role redefined to resolve elsewhere drops out as its members
# do; what the player mapped on this device answers for it here.
found = {name for name in found if own.get(name, redirects.get(name, raw)) == raw}
return frozenset(found)
def _binding_names(self, event: PointerEvent | ButtonEvent) -> frozenset[DeviceButton]:
"""The controls *event*'s press puts in front of the bindings.
The control the layout names, its index form and its roles, less the
roles when the pointer is not the primary one: a role binding models a
single pointer, which is the mouse or whatever stands in for it, so a
second finger does not hold ``Pointer.PRIMARY`` while the first does.
The pointer standing in for the mouse under :attr:`emulate_mouse` also
answers to the mouse's own primary button, for its contact alone.
"""
button = event.button
if button is None:
return frozenset()
store = self._devices
rec = store.device_records.get(event.device_id)
layout = store.layout_of(rec) if rec is not None else None
index = layout.button_index(button) if layout is not None else button.index
secondary_pointer = (
isinstance(event, PointerEvent) and not event.is_primary and event.pointer_type is not PointerType.MOUSE
)
found: set[DeviceButton]
if layout is not None and index is not None:
found = set(self._press_names(layout, index, roles=not secondary_pointer, unit=event.device_id))
else:
found = {button} if secondary_pointer else {button, *roles_of(button)}
if (
isinstance(event, PointerEvent)
and button.device is not DeviceType.MOUSE
and layout is not None
and index is not None
and index == layout.primary
and event.is_primary
and self.emulate_mouse
and not event.locked
):
mouse = self._mouse_layout()
primary = mouse.primary
if primary is not None:
found |= self._press_names(mouse, primary)
return frozenset(found)
def _on_mouse_move(self, x: float, y: float, source_id: int = 0):
"""Called by platform adapter for mouse movement, for one source."""
source = self._pointer(source_id)
self._last_source = source_id
old = source.position
source.position = (x, y)
if source.discard_next_delta:
# Discontinuity (warp / capture toggle): re-baseline, no delta.
source.discard_next_delta = False
else:
# Accumulate within the frame (cleared by _end_frame); multiple
# motion events per frame must sum to the true per-frame delta
# rather than leaving only the last segment. See cursor_pos_callback.
source.delta = (source.delta[0] + (x - old[0]), source.delta[1] + (y - old[1]))
def _install_controller_rumble(self, rumble: Callable[[int, float, float, float], bool] | None) -> None:
"""Wire up how a pad vibrates, and declare that one can.
Called by whatever polls the pads, which is the only layer that knows
whether its library can drive the motors. Installing the callback and
declaring the capability are one act, so
``Input.capabilities.rumble`` cannot say True with nothing behind it.
Passing None tears the seam down without withdrawing the fact, because
a capability is what the machine can do rather than what is wired up
this instant.
"""
self._devices.rumble = rumble
if rumble is not None:
_capabilities.declare(rumble=True)
def _pad_for_write(self, pad_id: int) -> _PadRecord:
"""The pad the backend calls *pad_id*, connected as a simulated pad if it is new.
A backend announces a pad through :meth:`_update_controller`, which names
it and says where it came from. Anything that writes a single button or
axis without that announcement is a test or an on-screen control, so a
pad that appears here is simulated until something says otherwise.
"""
pad = self._devices.pad(pad_id)
if not pad.connected:
pad.name = pad.name or f"Simulated pad {pad_id}"
pad.source = ControllerSource.SIMULATED
self._connect_pad(pad)
return pad
def _connect_pad(self, pad: _PadRecord) -> None:
"""A pad arrived: into ``Input.controllers`` and ``Input.devices``, with the signals."""
pad.connected = True
device = ControllerDevice(self._devices, pad.id)
self._devices.objects[pad.id] = device
self._devices.device_objects[pad.id] = device
_capabilities.declare(controllers=True)
self._notify_connection(pad.id, True)
def _write_pad_button(self, pad: _PadRecord, index: int, pressed: bool) -> None:
"""One button on one pad, with its edge and the event the router will publish.
Only a change records anything. A pad is polled every frame and most of
its buttons are up in every one of them, so recording an edge for each
would fill the just-released set with buttons nobody touched.
"""
if pressed == (index in pad.held):
return
if pressed:
pad.held.add(index)
pad.just_pressed.add(index)
else:
pad.held.discard(index)
pad.just_released.add(index)
self._deliver_pad_button(pad.id, index, pressed)
self._devices.pending_joy_events.append(("button", pad.id, index, 1.0 if pressed else 0.0))
def _write_pad_axis(self, pad: _PadRecord, index: int, value: float) -> None:
"""One axis on one pad, recording the event only when it actually moved."""
previous = pad.axis_values.get(index) or 0.0
if previous != value:
pad.axes_prev.setdefault(index, previous)
pad.axis_values[index] = value
self._deliver_pad_axis(pad.id, index, value)
self._devices.pending_joy_events.append(("axis", pad.id, index, value))
def _deliver_pad_button(self, pad_id: int, index: int, pressed: bool) -> None:
"""One pad button reached this tree, under the controls its layout names.
Called from both ends of the same fact and idempotent: the poll that
found the button delivers it to the tree that polled, and the event the
poll queued delivers it to whichever tree a router publishes it into,
which for the polling tree is the same press arriving twice.
"""
pad = self._devices.pads.get(pad_id)
if pad is None:
return
self._on_button(
("device", pad_id, index),
self._press_names(self._devices.layout_of(pad), index, unit=pad_id),
pressed,
pad_id,
)
def _deliver_pad_axis(self, pad_id: int, index: int, value: float) -> None:
"""One pad axis reached this tree, under the controls its layout names."""
pad = self._devices.pads.get(pad_id)
if pad is None:
return
layout = self._devices.layout_of(pad)
named = layout.axis_control(index)
forms = self._axis_names(layout, index, pad_id)
before = self._axis_sources.get(named, {}).get(pad_id)
self._on_device_axis(forms, value, value - (before or 0.0), pad_id)
def _on_joy_button(self, button: int | DeviceButton, pressed: bool, pad_id: int = 0) -> None:
"""One pad button, by the pad's index or its control, as a test or an on-screen pad writes it."""
pad = self._pad_for_write(pad_id)
index = button if isinstance(button, int) else self._devices.layout_of(pad).button_index(button)
if index is None:
raise ValueError(f"pad {pad_id}'s layout names no {button!r}")
self._write_pad_button(pad, index, pressed)
def _on_joy_axis(self, axis: int | DeviceAxis, value: float, pad_id: int = 0) -> None:
"""One pad axis, by the pad's index or its control, in the engine's convention."""
pad = self._pad_for_write(pad_id)
index = axis if isinstance(axis, int) else self._devices.layout_of(pad).axis_index(axis)
if index is None:
raise ValueError(f"pad {pad_id}'s layout names no {axis!r}")
self._write_pad_axis(pad, index, float(value))
def _update_controller(
self,
pad_id: int,
buttons: dict[str, bool],
axes: dict[str, float],
*,
source: ControllerSource | str = ControllerSource.DEVICE,
name: str | None = None,
guid: str | None = None,
recognised: bool | None = None,
raw_buttons: dict[int, bool] | None = None,
raw_axes: dict[int, float] | None = None,
):
"""Called by platform adapter to update one pad's state each frame.
A pad a mapping database laid out is reported by the engine's standard
names (see :mod:`simvx.core.input.controller`), which are its default
layout's positions. One no database knows is reported by the driver's
own indices, ``raw_buttons`` and ``raw_axes``, and has no names until
the player maps them onto it. A pad has one index space either way.
Args:
pad_id: The backend's id for the pad this snapshot belongs to.
buttons: Held state per standard name, for a mapped pad.
axes: Axis values per standard name, already in the engine's
convention, for a mapped pad.
source: Where the pad came from. A poll transaction prunes only
:attr:`~.enums.ControllerSource.DEVICE` pads, so an injected or
on-screen pad survives a real poll that does not know about it.
name: What the driver calls it, which decides its family.
guid: Its model's stable id, for per-model settings.
recognised: Whether a mapping database laid the pad out.
raw_buttons: The driver's button indices, for an unmapped pad.
raw_axes: The driver's axis indices, for an unmapped pad.
"""
from ..node import Node # node imports the input package
# A backend defect check, so it runs under ``Node.dev_checks`` only.
if Node.dev_checks:
for axis_name in _controller.out_of_range_axes(axes):
if axis_name not in _warned_axis_names:
_warned_axis_names.add(axis_name)
log.warning(
"controller axis %r arrived at %r from source %r, outside the engine's range; "
"the backend should map through simvx.core.input.controller.normalise_axes",
axis_name,
axes[axis_name],
source,
)
pad = self._devices.pad(pad_id)
arriving = not pad.connected
pad.source = ControllerSource(source)
if name is not None:
pad.name = name
if guid is not None:
pad.guid = guid
if recognised is not None:
pad.recognised = recognised
pad.family = family_for(pad.name, pad.guid, recognised=pad.recognised)
pad.layout = None if pad.recognised else UNRECOGNISED_CONTROLLER_LAYOUT
if arriving:
pad.connected = True
# A snapshot says what the whole pad is doing, so a name the backend
# left out is that input at rest rather than that input unchanged: a
# poll is the only thing that reports a release.
if pad.recognised:
for button_name, index in _controller.BUTTON_INDEX.items():
self._write_pad_button(pad, index, bool(buttons.get(button_name, False)))
for axis_name, index in _controller.AXIS_INDEX.items():
self._write_pad_axis(pad, index, float(axes.get(axis_name, 0.0)))
else:
for index, held in (raw_buttons or {}).items():
self._write_pad_button(pad, int(index), bool(held))
for index, value in (raw_axes or {}).items():
self._write_pad_axis(pad, int(index), float(value))
if arriving:
self._connect_pad(pad)
def _notify_connection(self, pad_id: int, connected: bool) -> None:
"""Record a pad arriving or leaving, and fire once the pads are consistent.
Inside a poll transaction the signal waits for the transaction to
close, so a handler enumerating ``Input.controllers`` never sees half a
frame's pads. Outside one there is nothing to wait for.
"""
self._devices.notifications.append((pad_id, connected))
if not self._devices.poll_depth:
self._flush_controller_notifications()
def _flush_controller_notifications(self) -> None:
pending, self._devices.notifications = self._devices.notifications, []
for pad_id, connected in pending:
pad = self._devices.objects.get(pad_id) if connected else ControllerDevice(self._devices, pad_id)
if pad is None:
continue
(self.controller_connected if connected else self.controller_disconnected).emit(pad)
if connected:
self._devices.device_connected.emit(pad)
def _disconnect_pad(self, pad_id: int) -> None:
"""Take a pad away: release what it held, then drop it from the mapping.
The record survives until the frame boundary so the release edges stay
readable for the frame the unplug happened on, and a
:class:`~.devices.ControllerDevice` the game is still holding answers from it.
"""
pad = self._devices.pads.get(pad_id)
if pad is None or not pad.connected:
return
for index in list(pad.held):
self._write_pad_button(pad, index, False)
for index in list(pad.axis_values):
self._write_pad_axis(pad, index, 0.0)
pad.connected = False
self._devices.objects.pop(pad_id, None)
self._devices.device_objects.pop(pad_id, None)
self._notify_connection(pad_id, False)
@contextmanager
def _controller_poll(self) -> Iterator[Callable[..., None]]:
"""One frame's controller poll, as a transaction rather than an add-only write.
Yields a ``report(pad_id, buttons, axes, **facts)`` the adapter calls
once per pad it found, with the backend's id for it and whatever it
knows about the pad (``name``, ``guid``, ``recognised``,
``raw_buttons``, ``raw_axes``). On leaving the block every device pad
that was *not* reported is disconnected, so a controller unplugged
mid-game releases its held buttons through the ordinary edge logic
instead of staying pressed until the process exits.
Reporting and closing the transaction are the same object, so an
adapter cannot record a pad without also declaring the poll complete.
A poll that raises part-way does not prune: half a frame's pads is not
evidence that the rest went away, and the connection signals it
accumulated still fire, because a pad that arrived did arrive.
Only a backend that genuinely enumerates pads may open this. A backend
with no controller support must not, since an empty poll from it would
read as "every pad just left".
"""
seen: set[int] = set()
self._devices.poll_depth += 1
def report(pad_id: int, buttons: dict[str, bool], axes: dict[str, float], **facts: object) -> None:
seen.add(pad_id)
self._update_controller(pad_id, buttons, axes, **facts) # type: ignore[arg-type]
try:
yield report
for pad in list(self._devices.pads.values()):
if pad.source is ControllerSource.DEVICE and pad.connected and pad.backend_id not in seen:
self._disconnect_pad(pad.id)
finally:
self._devices.poll_depth -= 1
if not self._devices.poll_depth:
self._flush_controller_notifications()
# ----------------------------------------------------------------
# Pointers: the mouse, every finger and a pen
# ----------------------------------------------------------------
def _write_pointer(self, event: PointerEvent) -> None:
"""Record what *event* says about its pointer in the table ``pointers`` reads.
Every pointer type lands here, the mouse included, so a game polling
the fingers polls the mouse the same way. A finger that lifts and a pen
that leaves stay one frame with their release edge on them, and go at
the frame boundary.
"""
source = self._pointer(event.source_id)
table = source.pointers
pointer_id = event.pointer_id
before = table.get(pointer_id)
phase = event.phase
contact = event.contact
was_contact = before is not None and before.contact
pressed = contact and not was_contact
released = was_contact and not contact
cancelled = phase is PointerPhase.CANCEL
if pressed or released or cancelled or phase is PointerPhase.LEAVE:
source.edged = True
table[pointer_id] = PointerState(
pointer_id=pointer_id,
pointer_type=event.pointer_type,
position=event.position if not event.locked or before is None else before.position,
contact=contact,
is_primary=event.is_primary,
buttons=event.buttons,
pressure=event.pressure,
tilt=event.tilt,
twist=event.twist,
distance=event.distance,
tangential_pressure=event.tangential_pressure,
tool=event.tool,
tool_id=event.tool_id,
contact_size=event.contact_size,
just_pressed=pressed or (before is not None and before.just_pressed),
just_released=released or cancelled or (before is not None and before.just_released),
cancelled=cancelled or (before is not None and before.cancelled and not pressed),
device_id=event.device_id,
)
if phase is PointerPhase.LEAVE:
# Out of range is gone: the record stays for this frame only, so
# a poll made after the event still finds where it left from.
self._leaving.add((event.source_id, pointer_id))
def _drop_pointers(self, source_id: int) -> None:
"""Forget every finger and pen *source_id* had, silently.
For a scene change, where the contacts in flight belonged to the scene
that left and no lift will ever reach the new one. The mouse stays: it
is still wherever it was.
"""
source = self._sources.get(source_id)
if source is None:
return
for pointer_id in [pid for pid, p in source.pointers.items() if p.pointer_type is not PointerType.MOUSE]:
del source.pointers[pointer_id]
def _spend_pointer_edges(self, source: _PointerSource, source_id: int) -> None:
"""The frame boundary for one source's pointers: spend edges, bury the gone."""
source.edged = False
table = source.pointers
for pointer_id, state in list(table.items()):
gone = (source_id, pointer_id) in self._leaving or (
not state.contact and state.just_released and state.pointer_type is PointerType.TOUCH
)
if gone:
del table[pointer_id]
elif state.just_pressed or state.just_released or state.cancelled:
table[pointer_id] = _without_edges(state)
@property
def pointers(self) -> Mapping[int, PointerState]:
"""Every live pointer, by pointer id: the mouse, each finger and a pen.
A finger appears from its press until the frame after it lifts; a pen
appears while it is in range, hovering or touching; the mouse is 0
once it has done anything. Every other id is opaque, and
``pointer_type`` says what kind of pointer it is. The mouse the primary finger emulates is not in here: that
finger is, once, as itself.
A read-only view of the pointers of the source that moved last, as
``mouse_position`` is. Inside a physics tick it is the tick's
snapshot, and only the frame's first tick sees the press and release
edges, so a tap lands on one tick whatever the framerate::
for p in Input.pointers.values():
if p.just_pressed and p.pointer_type is PointerType.TOUCH:
self.spawn_ripple(p.position, p.pressure)
"""
tick = self._tick
if tick is not None:
return tick.pointers
return MappingProxyType(self._last_pointer.pointers)
def _abandon_mouse_button(self, button: int) -> None:
"""Drop one held mouse button without reporting a release.
For a press that was interrupted rather than finished, so the button
must stop reading as held without an edge that says the player let go.
Both spellings of the state go down together, as they do on a real
release.
"""
self._devices.mouse_buttons_pressed.discard(button)
for press in [p for p in self._button_presses if p[0] == "mouse" and p[2] == button]:
self._abandon_button(press)
self._keys[f"mouse_{button + 1}"] = False
def _abandon_held_pointers(self) -> None:
"""Drop held mouse buttons without reporting any release.
For input that is taken away rather than finished: the window loses
focus while a button is down, so the release is delivered to whatever
took the focus and never reaches this process. The press did not
complete, so nothing lands in the just-released sets and no consumer
sees a release edge. A game that fires a charge shot on release must
not fire one because the player alt-tabbed, and an edge recorded here
would in any case be read late: frames stop while the window is in the
background, so it would surface on the first frame after focus returns.
A finger or a pen in contact is not dropped here: the router cancels
it through the door first, so whoever holds it hears the gesture end.
"""
for button in list(self._devices.mouse_buttons_pressed):
self._abandon_mouse_button(button)
def _clear_device_edges(self) -> None:
"""Drop the frame's edges on the device record, and bury the departed pads.
The device record is process-wide, so its edges belong to the
process's frame: a guest tree running inside another tree's frame
opens a ``devices.hold_frame()`` span and spends only its own
delivered edges, or whichever of the two stepped second would read a
key nobody is holding and a pad nobody is touching.
"""
machine = self._devices
if machine.frame_held:
return
machine.keys_just_pressed.clear()
machine.keys_just_released.clear()
machine.mouse_buttons_just_pressed.clear()
machine.mouse_buttons_just_released.clear()
for pad in machine.pads.values():
pad.clear_edges()
machine.spend_device_frame()
def _new_frame(self):
"""Called by engine at frame start. Clears per-frame state."""
self._keys_delivered_down.clear()
self._keys_delivered_up.clear()
self._unit_keys_down.clear()
self._unit_keys_up.clear()
self._buttons_delivered_down.clear()
self._buttons_delivered_up.clear()
self._spend_axis_motion()
self._clear_device_edges()
self._ticks_this_frame = 0
def _end_frame(self):
"""Called by engine at frame end. Clears all per-frame state."""
self._keys_just_pressed.clear()
self._keys_just_released.clear()
self._keys_delivered_down.clear()
self._keys_delivered_up.clear()
self._unit_keys_down.clear()
self._unit_keys_up.clear()
self._buttons_delivered_down.clear()
self._buttons_delivered_up.clear()
self._spend_axis_motion()
self._clear_device_edges()
# An edge nobody flushed belongs to a frame that is over. Keeping it
# would deliver it against the next frame's state.
if not self._devices.frame_held:
self._devices.pending_joy_events.clear()
# Motion and the wheel are per frame and per source: each window's
# own accumulation is spent here.
for source_id, source in self._sources.items():
source.delta = (0.0, 0.0)
source.scroll = (0.0, 0.0)
if source.edged:
self._spend_pointer_edges(source, source_id)
self._leaving.clear()
self._forced_released.clear()
if self._stale:
self._clear_stale()
# The next frame's first tick is the one that owns its edges.
self._ticks_this_frame = 0
if not self._ticked and self._pending_edges:
# Until a tick has run there is nothing to say one is coming, so
# an edge is kept for one frame and then spent, as a frame-scoped
# edge always was: a tree that never ticks, a UI harness or an
# editor panel, would otherwise hoard them for ever. Once a tick
# has run, a tickless frame is a frame the clock skipped and the
# edge waits for the tick that follows.
self._tickless_boundaries += 1
if self._tickless_boundaries > 1:
self._pending_edges.clear()
self._tickless_boundaries = 0
def _spend_axis_motion(self) -> None:
"""The frame boundary for device axes: this frame's motion is over.
An action bound to a ring's delta reads the frame's turn and goes back
to rest, so the actions reading a delta are resolved again here.
"""
self._axes_delivered_prev.clear()
deltas = self._axis_deltas_delivered
if deltas:
spent = [("device_axis", key) for key in deltas]
deltas.clear()
self._touch(*spent)
def _reset(self):
"""Reset all input state. Useful for testing.
Takes both process-wide records down with it: the device record
(:func:`simvx.core.input.devices.reset`), because a test asking for a
clean input state means a clean machine too, and the process-wide
map, which is the floor under every tree's and so leaks the same way.
The capability flags are the exception and are reset on their own
(:func:`simvx.core.input.capabilities.reset`), since they are declared
once by a backend rather than per pad.
"""
self._keys.clear()
self._keys_just_pressed.clear()
self._keys_just_released.clear()
self._sources.clear()
self._last_source = 0
self._devices.reset()
self._keys_delivered.clear()
self._unit_keys.clear()
self._keys_delivered_down.clear()
self._keys_delivered_up.clear()
self._unit_keys_down.clear()
self._unit_keys_up.clear()
self._buttons_delivered.clear()
self._buttons_delivered_down.clear()
self._buttons_delivered_up.clear()
self._button_presses.clear()
self._axes_delivered.clear()
self._axes_delivered_prev.clear()
self._axis_deltas_delivered.clear()
self._action_state.clear()
self._actions_on.clear()
self._analogue.clear()
self._map_generation = -1
self._axis_sources.clear()
self._capture_mode = MouseCaptureMode.VISIBLE
# The session goes; the seam stays, for the same reason the capture
# callback does: it belongs to whatever window is on screen, not to
# the state a test is clearing.
self._text_input_rect = None
self._cursor_shape = CursorShape.ARROW
self.emulate_mouse = True
self._touch_secondary = None
self._step_dt = 0.0
self._leaving.clear()
self._open_at_last_event.clear()
self._stale_bindings.clear()
self._stale = False
self._forced_released.clear()
self._claimed_slots.clear()
self._tick = None
self._tick_verifiable = True
self._tick_owns_edges = False
self._ticks_this_frame = 0
self._ticked = False
self._pending_edges.clear()
self._tickless_boundaries = 0
self.input_map.clear()
# ----------------------------------------------------------------
# Internal: what a context change does to what is held
# ----------------------------------------------------------------
def _anything_held(self) -> bool:
"""Whether any device is down, which is the only case a context change has work in."""
if self._keys_delivered or self._buttons_delivered:
return True
return any(value for value in self._axes_delivered.values())
@staticmethod
def _binding_slot(b: InputBinding) -> tuple | None:
"""The device element *b* reads, as a key a stale mark can be filed under.
The element rather than the binding, because what has to happen before
the action counts again is that the player lets go of the thing they
are holding, and two bindings on the same key are the same hold.
"""
control = b.control
if control is not None:
code = key_code(control) if control.device is DeviceType.KEYBOARD else None
if code is not None:
return ("key", code)
if isinstance(control, DeviceButton):
return ("button", _button_key(b))
assert isinstance(control, DeviceAxis)
return ("device_axis", _axis_key(b), control.is_delta, control.half >= 0, b.deadzone)
return None
def _mark_stale(self, bindings: Iterable[InputBinding]) -> None:
"""Take what is already held out of *bindings* until it is pressed again.
What keeps a context change from pressing anything: the pause menu
that opened on Escape does not see Escape as its own press, and the
vehicle whose ``move`` arrived under a held W does not drive off.
"""
marked: list[InputBinding] = []
for b in bindings:
slot = self._binding_slot(b)
if slot is not None and self._slot_held(slot) and slot not in self._stale_bindings:
self._stale_bindings.add(slot)
self._stale = True
marked.append(b)
if marked:
self._touch_bindings(marked)
def _slot_held(self, slot: tuple) -> bool:
"""Whether the device element *slot* names is down right now, stale or not."""
kind = slot[0]
if kind == "key":
return slot[1] in self._keys_delivered
if kind == "button":
return slot[1] in self._buttons_delivered
if kind == "device_axis":
_, key, delta, positive, deadzone = slot
value: float | None = self._axis_deltas_delivered.get(key, 0.0) if delta else self._axes_delivered.get(key)
if value is None:
value = self._axis_deltas_delivered.get(key)
if value is None:
return False
return bool(value > deadzone) if positive else bool(value < -deadzone)
return False
def _binding_stale(self, b: InputBinding) -> bool:
"""Whether *b* is held over from before a context change and does not count yet."""
slot = self._binding_slot(b)
return slot is not None and slot in self._stale_bindings
# ----------------------------------------------------------------
# Internal: what the targeted phase takes
# ----------------------------------------------------------------
def _claim_slots(self, slots: Iterable[tuple]) -> None:
"""Record that a control took the press on these device elements.
Two facts follow from one mark. The bindings on the element read at
rest, through the same gate a context change uses, so the action
neither opens nor raises an edge while the player holds the button a
widget swallowed. And the element counts as claimed until the press is
let go, which is what carries the decision from the press to its
release: the claim is made once, where the press landed.
Device-named state is untouched: ``is_mouse_button_pressed`` reports
the hardware whoever took the event.
"""
for slot in slots:
self._claimed_slots.add(slot)
if self._slot_held(slot):
self._stale_bindings.add(slot)
self._stale = True
self._touch_slots(slots)
def _slots_claimed(self, slots: Iterable[tuple]) -> bool:
"""Whether the press on any of these device elements was claimed."""
if not self._claimed_slots:
return False
return any(slot in self._claimed_slots for slot in slots)
def _drop_claims(self, slots: Iterable[tuple]) -> None:
"""Forget the claim on these elements, because the press they held is over.
The stale mark outlives this by design: it is dropped at the end of
the frame, once the element is no longer held, so the release itself
still reads at rest.
"""
if not self._claimed_slots:
return
self._claimed_slots.difference_update(slots)
def _pressed_again(self, slot: tuple) -> None:
"""A fresh press of *slot* ends its stale mark: it is the press the mark waits for.
The mark otherwise lasts until the end of the frame the element is let
go in, so a release and a new press in one frame would carry it on to a
press that nothing claimed and no context change saw coming, and hold
the action at rest for as long as that press is held. A press the
targeted phase goes on to claim is marked again by :meth:`_claim_slots`.
"""
if self._stale and slot in self._stale_bindings:
self._stale_bindings.discard(slot)
self._stale = bool(self._stale_bindings)
def _clear_stale(self) -> None:
"""Drop the marks whose device element the player has let go of."""
kept = {slot for slot in self._stale_bindings if self._slot_held(slot)}
dropped = self._stale_bindings - kept
self._stale_bindings = kept
self._stale = bool(kept)
if dropped:
self._touch_slots(dropped)
def _handover(self, tree: SceneTree | None = None, *, devices: frozenset[DeviceType] | None = None) -> None:
"""Input is leaving this tree: close what it was holding, press nothing.
A host that stops forwarding, because play mode paused or the panel
lost focus, takes the player's hands off this tree without any device
moving. So every open action closes with a real edge, and everything
held at this moment is taken out of the reckoning until the player
lets go and presses again: a key down when the editor paused reads
unpressed on resume rather than walking the character into a wall.
*devices* narrows it to the actions something of those device types
is holding, and marks only those bindings: a host losing focus hands
over the keyboard and the pads and leaves a drag in flight alone.
Deliberately not the window's focus-loss routine, which replays
*device* releases. There the key really did come up. Here it is still
physically down, and device state has to stay honest about that.
"""
actions = self.input_map._actions
if not actions:
return
if devices is None:
self._release_actions(list(actions), actions, tree=tree)
for entry in actions.values():
self._mark_stale(entry.bindings)
return
def of_kind(b: InputBinding) -> bool:
return b.control is not None and b.control.device in devices
names = [
name
for name, entry in actions.items()
if any(of_kind(b) and self._binding_strength(b) > 0.0 for b in entry.bindings)
]
self._release_actions(names, actions, tree=tree)
for entry in actions.values():
self._mark_stale([b for b in entry.bindings if of_kind(b)])
def _release_actions(
self, names: list[str], previous: Mapping[str, Action], *, tree: SceneTree | None = None
) -> None:
"""Close the actions a context change took away, with a real edge each.
Only the ones whose definition changed: an action the change left
alone keeps its held state, so a dialog opening while the player walks
does not make them let go of the key.
The edge is recorded here and delivered by the scene tree, which is
what an ``@on_input("jump", edge="release")`` handler hears. A
caller that knows which tree that is names it; a context change does
not, and the active one answers.
"""
edges: list[tuple[str, int]] = []
for player_id in self._players:
with self._as_player(player_id):
for name in names:
entry = previous.get(name)
if entry is None:
continue
if entry.kind is ActionKind.BUTTON:
open_now = any(self._binding_pressed(b) for b in entry.bindings)
else:
open_now = name in self._open_at_last_event.get(player_id, ())
if not open_now:
continue
edges.append((name, player_id))
self._forced_released.add((player_id, name))
self._record_edge(name, False, player_id, monotonic())
self._open_at_last_event.get(player_id, set()).discard(name)
for name in names:
entry = previous.get(name)
if entry is not None:
self._mark_stale(entry.bindings)
if not edges:
return
# Delivered by the tree the state belongs to. A caller that knows it
# says so; otherwise it is found rather than registered, because
# several trees may share the process-wide default pair and a
# registered sink would be whichever of them was built last.
if tree is None:
from ..scene_tree import SceneTree
tree = SceneTree.current()
if tree is None or tree.input is not self:
return
for name, player_id in edges:
tree._deliver_action_release(name, player_id)
# ----------------------------------------------------------------
# Internal: binding resolution helpers
# ----------------------------------------------------------------
def _binding_held(self, b: InputBinding) -> bool:
"""Whether everything *b* holds is down, and, where it is exact, nothing else among the modifiers.
:attr:`InputBinding.exact` says which: by default a binding that holds
no modifier accepts any modifier state, so an action bound to plain
Space keeps firing while the player holds Shift, and one that holds a
modifier is exact, so Ctrl+S is not Ctrl+Shift+S. The pressed
control's own modifier does not count against it.
"""
if not b.held and not b.exact:
return True
for held in b.held:
if not self._control_down(held):
return False
if b.is_exact:
named = b.modifiers
down = self._modifiers_down()
control = b.control
if isinstance(control, DeviceButton):
down = down - {MODIFIER_OF.get(control.on(None), control.on(None))}
if down != named:
return False
return True
@staticmethod
def _key_in(control: DeviceButton | DeviceAxis, keys: set[int], by_unit: dict[int, set[int]]) -> bool:
"""Whether the key *control* names is in *keys*, or in one keyboard's where it is pinned; a group, any side."""
where = keys if control.unit is None else by_unit.get(control.unit, set())
if not isinstance(control, DeviceButton):
return False
return any(key_code(member) in where for member in control.members)
def _control_down(self, control: DeviceButton) -> bool:
"""Whether *control* is held in this tree's delivered record: a key, a button, or any member of a group."""
if control.device is DeviceType.KEYBOARD:
if control.unit is None and self._event_modifiers is not None and control in MODIFIER_SIDES:
return control in self._event_modifiers
keys = self._keys_delivered if control.unit is None else self._unit_keys.get(control.unit, set())
return any(key_code(member) in keys for member in control.members)
return any(
self._held_in(InputBinding(member.on(control.unit)), self._buttons_delivered) for member in control.members
)
def _modifiers_down(self) -> frozenset[DeviceButton]:
"""The modifier groups held: the dispatching event's report, else either side in the delivered record."""
if self._event_modifiers is not None:
return self._event_modifiers
keys = self._keys_delivered
return frozenset(group for group, codes in _MODIFIER_GROUP_CODES.items() if any(c in keys for c in codes))
@staticmethod
def _axis_beyond_deadzone(b: InputBinding, value: float) -> bool:
"""Whether *value* counts as a press for this axis binding's half of travel."""
if _positive(b):
return value > b.deadzone
return value < -b.deadzone
def _held_in(self, b: InputBinding, table: Mapping[object, object] | set[object]) -> bool:
"""Whether the button *b* names is in *table*, one of the delivered button records.
A pad binding that names no unit reads every pad at once, which is the
right answer for one player and the wrong one for two: with a second
player in the game it reads only the pads this one is paired to.
"""
control = b.control
assert control is not None
if not self._solo and control.device is DeviceType.CONTROLLER and control.unit is None:
return any((pad_id, control) in table for pad_id in self._owned_pads())
return _button_key(b) in table
def _device_axis(self, b: InputBinding, *, before: bool = False) -> float:
"""What the device axis *b* names reads: its position, or with ``delta`` its motion this frame.
An axis that reports only motion has no position, and a binding on
that position reads at rest rather than a number nobody measured.
Before this frame's first change, a delta was nothing.
"""
control = b.control
assert isinstance(control, DeviceAxis)
if not self._solo and control.device is DeviceType.CONTROLLER and control.unit is None:
return max(
(
self._device_axis(InputBinding(control.on(pad_id), deadzone=b.deadzone), before=before)
for pad_id in self._owned_pads()
),
key=abs,
default=0.0,
)
key = _axis_key(b)
if control.is_delta or (key in self._axes_delivered and self._axes_delivered[key] is None):
# An axis that reports motion and no position (the wheel, a
# gesture, a relative ring) reads its motion either way.
return 0.0 if before else self._axis_deltas_delivered.get(key, 0.0)
value = self._axes_delivered.get(key)
if before and key in self._axes_delivered_prev:
value = self._axes_delivered_prev[key]
return 0.0 if value is None else value
def _binding_pressed(self, b: InputBinding) -> bool:
"""Check if a typed binding is currently pressed."""
if self._stale and self._binding_stale(b):
return False
if not self._solo and not self._binding_heard(b):
return False
control = b.control
if control is None:
return False
if control.device is DeviceType.KEYBOARD:
hit = self._key_in(control, self._keys_delivered, self._unit_keys)
elif isinstance(control, DeviceButton):
hit = self._held_in(b, self._buttons_delivered)
else:
hit = self._axis_beyond_deadzone(b, self._device_axis(b))
return hit and self._binding_held(b)
def _binding_just_pressed(self, b: InputBinding) -> bool:
"""Check if a typed binding was just pressed this frame."""
if self._stale and self._binding_stale(b):
return False
if not self._solo and not self._binding_heard(b):
return False
control = b.control
if control is None:
return False
if control.device is DeviceType.KEYBOARD:
hit = self._key_in(control, self._keys_delivered_down, self._unit_keys_down)
elif isinstance(control, DeviceButton):
hit = self._held_in(b, self._buttons_delivered_down)
else:
hit = self._axis_beyond_deadzone(b, self._device_axis(b)) and not self._axis_beyond_deadzone(
b, self._device_axis(b, before=True)
)
return hit and self._binding_held(b)
def _binding_just_released(self, b: InputBinding) -> bool:
"""Check if a typed binding was just released this frame.
A binding held over from before a context change has already had its
release: the change delivered one, so the device coming up later is
not a second edge.
"""
if self._stale and self._binding_stale(b):
return False
if not self._solo and not self._binding_heard(b):
return False
control = b.control
if control is None:
return False
if control.device is DeviceType.KEYBOARD:
hit = self._key_in(control, self._keys_delivered_up, self._unit_keys_up)
elif isinstance(control, DeviceButton):
hit = self._held_in(b, self._buttons_delivered_up)
else:
hit = self._axis_beyond_deadzone(b, self._device_axis(b, before=True)) and not self._axis_beyond_deadzone(
b, self._device_axis(b)
)
return hit and self._binding_held(b)
def _binding_strength(self, b: InputBinding, *, before: bool = False, units: bool = True) -> float:
"""Return analog strength [0, 1] for a binding.
*before* reads the state as it stood at the start of this frame, which
is how a continuous action finds its edge: it carries no device edge of
its own, so the edge is its value crossing the deadzone, and both sides
of that comparison have to be measured the same way. A held input that
went down this frame was up before it; one that came up this frame was
down.
"""
if self._stale and self._binding_stale(b):
return 0.0
if not self._solo and not self._binding_heard(b):
return 0.0
control = b.control
if control is None or not self._binding_held(b):
return 0.0
if control.device is DeviceType.KEYBOARD:
keys = self._key_in
held = (
keys(control, self._keys_delivered, self._unit_keys)
and not keys(control, self._keys_delivered_down, self._unit_keys_down)
if before
else keys(control, self._keys_delivered, self._unit_keys)
) or (before and keys(control, self._keys_delivered_up, self._unit_keys_up))
return (self._in_units(b, 1.0) if units else 1.0) if held else 0.0
if isinstance(control, DeviceButton):
held = (
self._held_in(b, self._buttons_delivered) and not self._held_in(b, self._buttons_delivered_down)
if before
else self._held_in(b, self._buttons_delivered)
) or (before and self._held_in(b, self._buttons_delivered_up))
return (self._in_units(b, 1.0) if units else 1.0) if held else 0.0
if isinstance(control, DeviceAxis):
val = self._device_axis(b, before=before)
if control.half >= 0:
raw = max(0.0, val) if val > b.deadzone else 0.0
else:
raw = max(0.0, -val) if val < -b.deadzone else 0.0
return self._in_units(b, raw) if units else raw
return 0.0
def _in_units(self, b: InputBinding, raw: float) -> float:
"""*raw* in the action's units: times the binding's scale, and over this step when it is a rate."""
if b.rate is not None:
raw *= b.rate * self._step_dt
return raw * b.scale if b.scale != 1.0 else raw
#: The key codes that hold each modifier group, and all of them together.
_MODIFIER_GROUP_CODES: dict[DeviceButton, tuple[int, ...]] = {
group: tuple(sorted(key_code(side) or 0 for side in sides)) for group, sides in MODIFIER_SIDES.items()
}
_MODIFIER_CODES: tuple[int, ...] = tuple(code for codes in _MODIFIER_GROUP_CODES.values() for code in codes)
def _unbounded(source: Axis1D | Axis2D) -> bool:
"""Whether *source* has no unit length to keep: motion (a pan, a wheel) or a value put in other units.
A stick's direction is kept within length one so a diagonal is no
faster; a source that reads motion, or that a scale or a rate has put
into the action's own units, means its magnitude and keeps it.
"""
bindings = [b for b in source.bindings if isinstance(b, InputBinding)]
if any(b.scale != 1.0 or b.rate is not None for b in bindings):
return True
return all(
isinstance(b.control, DeviceAxis) and (b.control.is_delta or bool(recognised_in(b.control))) for b in bindings
)
def _digital(source: Source) -> bool:
"""Whether every control behind *source* is a button, which reads fully on or off."""
bindings: tuple[object, ...] = (
(source,) if isinstance(source, InputBinding) else tuple(getattr(source, "bindings", ()))
)
return bool(bindings) and all(isinstance(b, InputBinding) and isinstance(b.control, DeviceButton) for b in bindings)
def _code(key: DeviceControl) -> int:
"""The platform code of a key, refusing a control that is not one."""
code = key_code(key)
if code is None:
raise TypeError(f"{key!r} is not a key: ask about a key such as Key.SPACE")
return code
def _positive(b: InputBinding) -> bool:
"""Which half of an axis *b* reads: the half its control names, the positive for a whole axis."""
control = b.control
return not isinstance(control, DeviceAxis) or control.half >= 0
def _button_key(b: InputBinding) -> object:
"""The delivered-record key a button binding reads: the control, or the control on one unit."""
control = b.control
assert control is not None
return control if control.unit is None else (control.unit, control.on(None))
def _axis_key(b: InputBinding) -> object:
"""The delivered-record key an axis binding reads: the whole axis, on its unit if it names one."""
control = b.control
assert isinstance(control, DeviceAxis)
return control.whole if control.unit is None else (control.unit, control.whole)
#: The input no tree owns: what a state-only test and a headless backend
#: write. It has a map of its own that no ambient name reaches, because
#: outside a tree there is no map a game could mean.
_default_input = _Input(input_map=_InputMap())
# The default pair know about each other: a context change on the default map
# has to find the held state to release, exactly as a tree's pair do.
_default_input.input_map._state = _default_input
#: None outside a span, which is not the same as the process-wide default:
#: it is what lets the resolution below fall back to the tree in play rather
#: than to a record nothing delivers to.
_active_input: contextvars.ContextVar[_Input | None] = contextvars.ContextVar("_active_input", default=None)
#: Resolved once, lazily: ``simvx.core.input`` is imported long before
#: ``simvx.core.scene_tree`` and must not import it at module level.
_scene_tree_class: type[SceneTree] | None = None
def _touch_secondary_in_play() -> DeviceButton | None:
"""The tree in play's ``touch_secondary``, for ``Pointer.SECONDARY`` membership."""
return current_input()._touch_secondary
_controls_module._touch_secondary = _touch_secondary_in_play
#: What the ambient ``Input`` answers only for a tree: its actions, players,
#: contexts and map. With no tree it raises rather than answering from an
#: input nothing delivers to, which would read at rest for ever; the device
#: record (keys, buttons, pads, capabilities) is the machine's and still answers.
_TREE_ONLY = frozenset(
{
"pressed",
"just_pressed",
"just_released",
"value",
"get_action_strength",
"get_axis",
"get_vector",
"prompt",
"context_stack",
"players",
"add_player",
"remove_player",
"tick",
"input_map",
}
)
class _InputProxy:
"""Proxy that delegates all access to the ``_Input`` the caller is reading.
Existing code using ``from simvx.core import Input; Input.pressed(...)``
continues to work: the proxy routes to whichever instance
:func:`current_input` resolves.
"""
__slots__ = ()
#: The class this proxy stands in for. ``tools/api_surface.py`` reads it and
#: freezes that class's public methods under the exported name: without it,
#: the package root publishes an opaque object and every method reachable
#: through ``Input`` is outside the guard that exists to catch a rename.
_api_surface_class = _Input
def __getattr__(self, name: str):
state = current_input()
if state is _default_input and name in _TREE_ONLY:
raise RuntimeError(
f"Input.{name} answers for a scene tree's actions, and no tree exists yet. Ask it from "
"a node, or of a tree in hand, tree.input; device facts such as Input.is_key_pressed "
"and Input.devices answer without one."
)
return getattr(state, name)
def __setattr__(self, name: str, value):
setattr(current_input(), name, value)
def __delattr__(self, name: str):
delattr(current_input(), name)
def __repr__(self) -> str:
return repr(current_input())
# Declared as the class it stands in for. A type checker cannot see through
# ``__getattr__``, so without this every consumer of the ambient ``Input`` holds
# an opaque object: passing it where an ``_Input`` is wanted is an error, and
# reading a member off it is unchecked. Naming the class once here is the same
# claim ``_api_surface_class`` already makes, and it is the claim the proxy is
# built to keep -- it forwards every attribute, including the private ones the
# router and the UI seam read.
Input: _Input = cast(_Input, _InputProxy())