Source code for simvx.core.input.router
"""Platform-neutral input routing.
Every backend -- desktop windowing (GLFW/SDL3) and the browser runtime -- feeds
its events through :class:`InputRouter`. :meth:`InputRouter.publish` is the one
door input comes through, and the five phases it runs are the whole of what an
input *means* to the engine:
1. **Device state**, always written, so a poll answers honestly whoever else
claimed the event.
2. **Targeted**, the delivery that belongs to one node: the UI's focus and
hit-test route.
3. **Resolve**, the active bindings turned into action edges.
4. **Broadcast**, every matching ``@on_input`` handler in reverse tree order,
stopping at the first claim.
5. **Notify**: ``tree.input_delivered.emit(event, claimed_by)``, for everything that watches.
Backends translate their own wire format into the normalised arguments of the
entry points below -- engine :class:`Key` codes, a device's own button and axis
indices, ``pressed`` booleans -- and do nothing else; each entry point builds
one typed event and publishes it. A button is reported by index, and the
device's layout (:mod:`~simvx.core.input.layouts`) names it: a backend whose
platform numbers a device its own way declares that layout
(:meth:`InputRouter.declare_layout`) rather than renumbering. Keeping the phases in one place is what stops the
backends drifting apart: a handler kind that works on the desktop works in the
browser because both reach it through this module.
It is also what lets a thin backend be a correct one. A backend states the
event's identity and its payload; the router fills in what it left out, so the
event a handler receives carries a time, a key label and the modifier state
whether or not the platform reported them. What the platform did supply is
declared once, through :mod:`simvx.core.input.capabilities`, so a game can ask
what its input source can do rather than inspecting events.
"""
from __future__ import annotations
import logging
from collections.abc import Collection, Sequence
from itertools import count
from time import monotonic
from typing import TYPE_CHECKING
from .capabilities import current as _current_capabilities
from .capabilities import declare as _declare_capabilities
from .controls import (
SELF_MODIFIER,
DeviceAxis,
DeviceButton,
DeviceControl,
DeviceType,
Mouse,
Pointer,
_name_for_code,
key_code,
key_for_code,
)
from .devices import AxisReport, InputDevice, check_axes, wrapped_delta
from .drag import DEFAULT_DRAG_THRESHOLD
from .enums import (
ControllerSource,
InputMethod,
PenAxis,
PointerPhase,
PointerTool,
PointerType,
)
from .events import (
AxisEvent,
ButtonEvent,
InputEvent,
KeyEvent,
PointerEvent,
ScrollEvent,
TextEvent,
)
from .gestures import GestureRecognition, gesture_axis_event, gesture_event
from .layouts import DeviceLayout
from .state import Input as _ambient_input
from .state import _Input
if TYPE_CHECKING:
from ..node import Node
from ..scene_tree import SceneTree
from .devices import _DeviceRecord
log = logging.getLogger(__name__)
#: Source ids, handed out one per router. Zero is reserved for an event
#: nobody attributed, which is what a hand-built event carries.
_source_ids = count(1)
#: Pointer ids for everything but the mouse, which keeps 0. Opaque and never
#: reused: a finger gets a fresh one each time it touches down, so a capture
#: left over from the last finger cannot catch the next, and two touchscreens
#: that both report finger 0 get two ids.
_pointer_ids = count(1)
#: A press this soon after the previous one continues the click run, so a
#: second press inside it is a double click. Seconds.
CLICK_INTERVAL = 0.4
#: How far a pointer may travel between two presses and still have them count
#: as one run, in logical pixels, by pointer type. A finger lands somewhere
#: near where it meant to and a pen sits between the two.
CLICK_SLOP = {PointerType.MOUSE: 5.0, PointerType.PEN: 8.0, PointerType.TOUCH: 25.0}
#: How far a stick must leave centre before it counts as the player using the
#: pad, for ``Input.input_method``. Deliberately looser than a binding's own
#: deadzone: a worn stick that rests off-centre must not claim the player picked
#: the controller up, and a real push clears this easily.
AXIS_METHOD_DEADZONE = 0.5
class _PointerTrack:
"""What the router remembers about one pointer between its events.
One press, its moment and its place, and what the events since have made
of it. Click runs and drag thresholds are resolved here, once, so every
consumer reads one answer off the event instead of each widget keeping a
press position and a clock of its own and disagreeing about the slop.
"""
__slots__ = (
"button",
"press_pos",
"press_time",
"click_count",
"dragging",
"down",
"pointer_type",
"drag_threshold",
)
def __init__(self) -> None:
self.button: DeviceButton | None = None
self.press_pos: tuple[float, float] = (0.0, 0.0)
self.press_time: float = 0.0
self.click_count: int = 0
self.dragging: bool = False
self.down: bool = False
self.pointer_type: PointerType = PointerType.MOUSE
#: The travel that makes this press a drag, read from the node that
#: took the press once its delivery says who that is.
self.drag_threshold: float = DEFAULT_DRAG_THRESHOLD[PointerType.MOUSE]
class _Contact:
"""What the router holds about one finger or pen between its events.
The buttons it has down, which decide its contact, and the pose it last
reported, so a cancel or a coalesced move can be published with the
pressure and tilt the pointer really had. A pending move waits here until
the frame's motion is flushed, one per pointer, so two fingers moving in
one frame are two events and neither swallows the other.
"""
__slots__ = (
"pointer_type",
"key",
"primary_bit",
"mask",
"pressure",
"tilt",
"twist",
"distance",
"tangential_pressure",
"tool",
"tool_id",
"contact_size",
"device_id",
"pending",
)
def __init__(self, pointer_type: PointerType, device_id: int, key: tuple, primary_bit: int) -> None:
self.pointer_type = pointer_type
#: ``(kind, device id, backend id)``: what the backend calls this
#: pointer, which the router's id table is keyed by.
self.key = key
#: The bit of ``mask`` that is contact: the primary role's, from the
#: device's layout.
self.primary_bit = primary_bit
self.mask = 0
#: The measured pressure, or None where the hardware cannot measure.
self.pressure: float | None = None
self.tilt: tuple[float, float] = (0.0, 0.0)
self.twist = 0.0
self.distance = 0.0
self.tangential_pressure = 0.0
self.tool = PointerTool.TIP
self.tool_id = 0
self.contact_size: tuple[float, float] = (0.0, 0.0)
self.device_id = device_id
#: A move not yet delivered, as ``(x, y, time)``.
self.pending: tuple[float, float, float] | None = None
def _declare_pen_axes(
pressure: float | None,
tilt: tuple[float, float] | None,
twist: float | None,
distance: float | None,
tangential_pressure: float | None,
) -> None:
"""The pen axes a sample proves the hardware measures, declared once each.
A backend passes a value for what the platform measured and None for what
it did not, the way it passes a time or leaves it out, so a measurement
arriving is the proof. The guard is the point: this runs on every pen
sample, and once the axes are known it is five comparisons.
"""
known = _current_capabilities().pen_axes
found = {
axis
for axis, value in (
(PenAxis.PRESSURE, pressure),
(PenAxis.TILT, tilt),
(PenAxis.TWIST, twist),
(PenAxis.DISTANCE, distance),
(PenAxis.TANGENTIAL, tangential_pressure),
)
if value is not None and axis not in known
}
if found:
_declare_capabilities(pen_axes=frozenset(found))
def _complete_pointer(inp: _Input, event: PointerEvent) -> None:
"""Name *event*'s button through its device's layout, and make it agree with itself.
The entry points build a consistent event; one published straight through
the door, by a test, a replay or an agent, may state only its phase and a
button, named, by index or as a role. So everything the layout decides is
decided here, once, for both: the device the event came from, the button
as the control the layout calls it, and which bit is contact. A press
holds the button it presses and a release does not, so the mask is
completed from the button, and ``contact`` follows from the mask. Pressure
then follows contact: none out of contact, and in contact a pressure left
at nothing is one nobody measured, which reads 1.0.
"""
device_type = DeviceType(event.pointer_type.value)
store = inp._devices
rec = store.record_for(device_type, event.device_id)
event.device_id = rec.id
layout = store.layout_of(rec)
primary = layout.primary
event.contact_mask = 0 if primary is None else 1 << primary
button = event.button
if button is not None:
index = layout.button_index(button)
if index is None:
raise ValueError(
f"{device_type.value} {event.pointer_id}'s layout names no {button!r}; its names are "
f"{sorted(layout.buttons)}. Name the button by index, such as "
f"{device_type.value.title()}.button(n), or override the layout"
)
event.button = layout.button_control(index)
bit = 1 << index
if event.phase is PointerPhase.DOWN:
event.buttons |= bit
elif event.phase is PointerPhase.UP or event.phase is PointerPhase.CANCEL:
event.buttons &= ~bit
if not event.contact:
event.pressure = 0.0
elif event.pressure == 0.0:
event.pressure = 1.0
def _complete_device_event(inp: _Input, event: ButtonEvent | AxisEvent) -> None:
"""Name a tablet button or axis event through its device's layout, as a pointer's is.
A press or a motion built by hand may name the control or its index;
published, it carries the control the layout calls that index.
"""
store = inp._devices
if isinstance(event, ButtonEvent):
device_type = event.button.device
rec = store.record_for(device_type, event.device_id)
event.device_id = rec.id
layout = store.layout_of(rec)
index = layout.button_index(event.button)
if index is None:
raise ValueError(
f"{device_type.value}'s layout names no {event.button!r}; its names are {sorted(layout.buttons)}"
)
event.button = layout.button_control(index)
return
device_type = event.axis.device
rec = store.record_for(device_type, event.device_id)
event.device_id = rec.id
layout = store.layout_of(rec)
index = layout.axis_index(event.axis)
if index is None:
raise ValueError(
f"{device_type.value}'s layout names no axis {event.axis!r}; its names are {sorted(layout.axes)}"
)
event.axis = layout.axis_control(index)
def _index_of(inp: _Input, event: PointerEvent | ButtonEvent) -> int:
"""The device's own index for a published event's button, through the layout it was named by."""
store = inp._devices
rec = store.device_records.get(event.device_id)
button = event.button
assert button is not None
index = store.layout_of(rec).button_index(button) if rec is not None else button.index
assert index is not None
return index
def _write_mouse_button(inp: _Input, index: int, pressed: bool) -> None:
"""One mouse button, by index, into the device record and the string-keyed state."""
name = f"mouse_{index + 1}"
if pressed:
if not inp._keys.get(name):
inp._keys_just_pressed[name] = True
inp._keys[name] = True
else:
inp._keys[name] = False
inp._keys_just_released[name] = True
inp._on_mouse_button(index, pressed)
def _moved_past(a: tuple[float, float], b: tuple[float, float], limit: float) -> bool:
"""Whether *a* and *b* are more than *limit* logical pixels apart."""
dx = a[0] - b[0]
dy = a[1] - b[1]
return (dx * dx + dy * dy) > (limit * limit)
def _event_time(supplied: float | None) -> float:
"""The time to stamp on an event, and the capability a supplied one proves.
A backend with a platform event time converts it onto the monotonic clock
and passes it here; one without passes None and the event is timed at
receipt, which is accurate to the frame rather than to the input.
A time that arrives declares ``event_timestamps``, and a time that does not
declares nothing. The absence of a time is not evidence the source has
none: a synthetic event, a driver stamping zero and a simulated key all
look the same as a backend with no clock, so withdrawing the capability on
one would report whatever published last instead of what the source can do.
"""
if supplied is None:
return monotonic()
_declare_capabilities(event_timestamps=True)
return supplied
def _declare_pointer_facts(event: PointerEvent) -> None:
"""The device facts a pointer event proves, declared once each.
A finger proves there is a touchscreen, a pen proves there is a pen, a
pointer that moves or crosses a boundary without a button down proves the
device can hover, and a locked pointer proves the platform can lock one.
Each is one-way, like every capability: an event that does not prove a fact
says nothing about it, because a mouse event on a machine with a
touchscreen is not evidence the touchscreen left.
Whether the *primary* pointer is coarse is not decided here. A laptop with
a touchscreen delivers primary touches and still has a cursor, so the
answer comes from the platform's own query (a ``pointer: coarse`` media
query in the browser, the windowing library elsewhere) and the backend
declares it.
The guard is the point: this runs on every pointer event, and once the
facts a pointer can prove are known it costs one global read and four
comparisons.
"""
known = _current_capabilities()
if known.touch and known.pen and known.hover and known.pointer_lock:
return
kind = event.pointer_type
_declare_capabilities(
touch=kind is PointerType.TOUCH,
pen=kind is PointerType.PEN,
hover=event.buttons == 0 and event.phase in (PointerPhase.MOVE, PointerPhase.ENTER, PointerPhase.LEAVE),
pointer_lock=event.locked,
)
def _note_pointer_method(state: _Input, event: PointerEvent) -> None:
"""Switch the input method if this pointer event is the player reaching for a device.
A touch press is touch; a mouse or pen press is keyboard and mouse, because
both are a deliberate act aimed at something. Motion is not: a cursor sits
where it was left and a window moving under it reports a move, so mouse
motion switches only once it has travelled
:attr:`~._Input.mouse_method_threshold`. Crossing a boundary is hover and
never switches at all.
"""
phase = event.phase
if phase is PointerPhase.DOWN:
state._note_input_method(
InputMethod.TOUCH if event.pointer_type is PointerType.TOUCH else InputMethod.KEYBOARD_MOUSE
)
elif phase is PointerPhase.MOVE and event.pointer_type is not PointerType.TOUCH:
state._note_mouse_travel(*event.delta)
def _note_controller_method(state: _Input, event: ButtonEvent | AxisEvent) -> None:
"""Switch to controller, but only for a pad the machine actually has.
The on-screen pad publishes under a virtual source so a game binding
``Controller.A`` gets touch play with no further code. If its presses
switched the method it would hide itself the moment it was used, which is
the one thing it must not do. A simulated pad is excluded for the same
reason in reverse: a test driving one is not a player picking one up.
"""
pad = state._devices.pads.get(event.device_id)
if pad is None or pad.source is not ControllerSource.DEVICE:
return
if isinstance(event, AxisEvent) and abs(event.value or 0.0) < AXIS_METHOD_DEADZONE:
return
state._note_input_method(InputMethod.CONTROLLER)
def _claim_slots(event: InputEvent, inp: _Input) -> tuple[tuple, ...]:
"""The device elements *event* names, as the keys a claim is filed under.
The same keys :meth:`_Input._binding_slot` files a stale mark under, so a
claim on the press reaches every binding reading that element. A pad
button is filed twice, under the pad it came from and under "any pad",
because a binding may name either and both are the same physical press.
Empty for anything with no element behind it: motion, the wheel, text.
Those carry no action and there is nothing to withhold. A pointer, tablet
or pad button is filed under every name it put in front of the bindings,
and under each of those on its own device, so a control that takes a
barrel press withholds the barrel binding and not the right click.
"""
if isinstance(event, KeyEvent):
return (("key", _index_key(event)),)
if isinstance(event, (PointerEvent, ButtonEvent)):
if event.button is None:
return ()
names = inp._binding_names(event)
device = event.device_id
return tuple(("button", name) for name in names) + tuple(("button", (device, name)) for name in names)
return ()
def _index_key(event: KeyEvent) -> int:
"""A key event's platform code."""
code = key_code(event.key)
assert code is not None
return code
def _is_press(event: InputEvent) -> bool:
"""Whether *event* opens a press, as opposed to closing one or being neither."""
if isinstance(event, KeyEvent):
return event.pressed
if isinstance(event, PointerEvent):
return event.phase is PointerPhase.DOWN
if isinstance(event, ButtonEvent):
return event.pressed
return False
def _button_mask(state: _Input) -> int:
"""The held mouse buttons as a bitmask, one bit per index in the mouse's numbering.
Read off the device record, because an event's ``buttons`` field says what
the machine is holding rather than what any one tree was given.
"""
mask = 0
for code in state._devices.mouse_buttons_pressed:
mask |= 1 << code
return mask
def _resolve_modifiers(
state: _Input, ctrl: bool | None, shift: bool | None, alt: bool | None, meta: bool | None
) -> tuple[bool, bool, bool, bool]:
"""The four modifier flags an event carries, filling in what it did not state.
A window backend reports what it saw at the moment of the event, and those
flags pass through unchanged. Anything left unstated -- a wheel notch, a
pure move, a touch, an event a test synthesised -- falls back to ``state``'s
held keys, which is the same fact arriving a different way. That fallback is
what lets every event kind carry modifiers, not only the ones a backend
reports them for.
The Super / Command key is held under the name ``"super"`` and reported on
an event as ``meta``: one key, in the spelling each layer already uses.
"""
held = state._keys
return (
held.get("ctrl", False) if ctrl is None else bool(ctrl),
held.get("shift", False) if shift is None else bool(shift),
held.get("alt", False) if alt is None else bool(alt),
held.get("super", False) if meta is None else bool(meta),
)
[docs]
class InputRouter:
"""Routes normalised input events into engine state, UI, and ``@on_input``.
Args:
tree: The :class:`SceneTree` receiving UI and ``@on_input`` dispatch.
``None`` routes to input state only, which is what a headless
backend or a state-only test wants.
input_state: Input state override, for a caller that owns an instance
no tree holds.
A router that was given a tree writes that tree's own :class:`_Input`, not
the ambient proxy. The state write happens before the tree opens its input
span, so resolving the ambient proxy here would land the delivered record
in whichever pair was active for the caller and the tree the event was
meant for would resolve none of its actions. The device record underneath
is one for the process and takes the same write whoever makes it.
"""
__slots__ = (
"tree",
"source_id",
"_input",
"_primary",
"_contacts",
"_claimed_contacts",
"_release_picked",
"_departed",
"_ids",
"_moving",
"_pointers",
"_last_tap",
"_sample_prev",
"_recognition",
"_pending_move",
"_pending_locked",
"_window_focused",
"__weakref__",
)
def __init__(self, tree: SceneTree | None = None, *, input_state: _Input | None = None):
self.tree = tree
#: Which input source this is. A window is a source and a screen is
#: not, so one router per window and a fresh id per router; it is
#: stamped on everything published here, and it is what the pointer
#: position, the hover and the capture table are kept per.
self.source_id = next(_source_ids)
#: Whether the platform says this window has keyboard focus; assumed until told otherwise.
self._window_focused = True
if input_state is not None:
self._input = input_state
elif tree is not None:
self._input = tree.input
else:
self._input = _ambient_input
# The finger or pen that holds primacy: the one standing in for the
# mouse. Taken by a press when nobody holds it and given up when that
# contact ends, so primacy moves on a press and never mid-gesture.
self._primary: int | None = None
# Every finger on the glass and pen in range, by pointer id.
self._contacts: dict[int, _Contact] = {}
# ``(pointer id, button)`` of the finger and pen presses the targeted
# phase claimed, each with the slots the claim was filed under, so
# each release inherits its press's decision and a lift after a scene
# change can still let the claim go.
self._claimed_contacts: dict[tuple[int, DeviceButton | None], tuple[tuple, ...]] = {}
# The node whose ``on_picked`` claimed the release being delivered, if
# one did: it stops every later handler when nobody claimed the press.
self._release_picked: Node | None = None
# The pointer id each ``(kind, device id, backend id)`` has now: a
# finger's for as long as it is down, a pen's while it is in range.
self._ids: dict[tuple, int] = {}
# The fingers and pens that were touching when the scene changed, by
# pointer id, each with whether it was driving the mouse. Their lift
# belongs to no scene, but what they wrote must still be let go of.
self._departed: dict[int, bool] = {}
# The fingers and pens with a move waiting for the flush, in the order
# they first moved this frame.
self._moving: list[int] = []
# Click runs and drag thresholds, one entry per pointer that has
# pressed. Created on the first press rather than per pointer id, so a
# mouse-only session holds one entry.
self._pointers: dict[int, _PointerTrack] = {}
# A finger is a new pointer id each time it lands, so a run of taps
# spans pointers: the last finger to lift carries the run to the next
# one, and its entry leaves ``_pointers`` when it lifts.
self._last_tap: _PointerTrack | None = None
# Where each pointer's last motion sample this frame left it, so the
# next sample's delta runs from there; cleared when the frame's merged
# motion is delivered, which starts the count again from the record.
self._sample_prev: dict[int, tuple[float, float]] = {}
# The gesture recogniser, made the first time a finger moves while
# the tree names a gesture; None in a game that names none.
self._recognition: GestureRecognition | None = None
# The mouse's motion this frame, waiting to be delivered as one event:
# position and time. Absolute motion keeps only where the pointer got
# to, because the delta is the distance from where the last flush left
# the recorded position; locked motion has no position to keep, so its
# steps are summed.
self._pending_move: tuple[float, float, float] | None = None
self._pending_locked: tuple[float, float, float] | None = None
if tree is not None:
# The tree keeps every router feeding it, weakly, so a scene change
# can reach back and clear the pointers that were in flight.
tree._routers.add(self)
@property
def _pointer(self):
"""This source's pointer in the tree's input.
A router is one window, and a pointer belongs to its window: read
through here rather than through the source-less spellings, so two
windows feeding one tree never re-base one another's delta. Resolved
per call because ``Input._reset`` drops the records.
"""
return self._input._pointer(self.source_id)
def _to_viewport(self, x: float, y: float) -> tuple[float, float]:
"""A point the backend measured on the glass, in the scene's own space.
The identity under ``EXPAND``, which is every backend's default and so
the path that must cost nothing: one attribute read and a branch. Under
``FIT`` it undoes the letterbox, and it is applied here, at the door,
rather than by anything downstream. That is what lets
``PointerEvent.position``, ``Input.mouse_position``, the per-source
recorded position, ``tree.screen_size`` and every ``Control`` rect all
be the same units: the conversion happens once and nothing after it
knows a letterbox exists.
A point in a bar converts to a coordinate outside the design rect
rather than being clamped or dropped, so it hits nothing by ordinary
geometry. Dropping it would be wrong: the pointer really is there, and
the device record has to stay honest about where.
"""
tree = self.tree
return tree.window_to_viewport(x, y) if tree is not None else (x, y)
def _to_viewport_delta(self, dx: float, dy: float) -> tuple[float, float]:
"""Motion measured on the glass, as motion in the scene's own space.
The companion to :meth:`_to_viewport` for a locked pointer, which has
a delta and no position to take the difference of.
"""
tree = self.tree
if tree is None:
return (dx, dy)
scale = tree.content_scale
return (dx, dy) if scale == 1.0 else (dx / scale, dy / scale)
[docs]
def reset(self) -> None:
"""Drop the pointer state this router was tracking.
Called by :meth:`SceneTree.change_scene
<simvx.core.scene_tree.SceneTree.change_scene>`: the fingers and pen
in flight were on the scene that just left, and a new scene that
inherited them would answer a lift it never saw pressed. Their records
go from ``Input.pointers``, and each one still touching is kept as
departed: its lift reaches nobody, and lets go of the mouse's primary it
was standing in for, exactly as a mouse button held across the change
is released by its own lift. A claim a press made is kept too, because
its release inherits it whichever scene is running, as a mouse
button's claim does on the input state.
"""
emulating = self._input.emulate_mouse
self._departed.update(
(pid, emulating and pid == self._primary) for pid, contact in self._contacts.items() if contact.mask
)
departed = self._departed
self._ids = {key: pid for key, pid in self._ids.items() if pid in departed}
self._primary = None
self._contacts.clear()
self._moving.clear()
self._pointers.clear()
self._last_tap = None
self._sample_prev.clear()
self._pending_move = None
self._pending_locked = None
self._input._drop_pointers(self.source_id)
[docs]
def flush_motion(self) -> None:
"""Deliver the frame's motion as one event, and forget it.
**A motion event carries the frame's accumulated delta**, which is
what a camera or a drag wants: a 1000Hz mouse and a 125Hz one move the
view by the same amount in the same frame, and a handler runs once
rather than sixteen times. The samples are not lost, they are summed,
because the delta a frame's motion means is the total.
Called at the top of the frame's input phase, and by anything that
needs the position to be true before it acts: a button press carries
where the pointer is, so the motion that got it there is spent first.
Every finger and pen is its own pointer with its own pending move, so
two fingers dragging in one frame are two events. A handler that needs
every sample as it arrives, a drawing or handwriting canvas under a
pen, declares ``@on_input(PointerEvent, samples=True)``: it was handed
each one with its own position, pressure and time as it came in, and
hears no merged move; the rest of the tree hears only this.
"""
self._sample_prev.clear()
locked = self._pending_locked
if locked is not None:
self._pending_locked = None
dx, dy, stamp = locked
self.publish(
PointerEvent(
phase=PointerPhase.MOVE,
position=self._pointer.position,
delta=(dx, dy),
buttons=_button_mask(self._input),
locked=True,
time=stamp,
)
)
pending = self._pending_move
if pending is not None:
self._pending_move = None
self.publish(self._motion_event(*pending))
moving = self._moving
if moving:
self._moving = []
for pointer_id in moving:
contact = self._contacts.get(pointer_id)
if contact is None or contact.pending is None:
continue
x, y, stamp = contact.pending
contact.pending = None
self.publish(self._contact_event(pointer_id, contact, PointerPhase.MOVE, (x, y), stamp))
if self._recognition is not None:
self._recognition.frame()
def _motion_event(self, x: float, y: float, stamp: float) -> PointerEvent:
"""Absolute mouse motion to *x*, *y*, with the delta since the last one delivered."""
inp = self._input
pointer = self._pointer
old = pointer.position
# Discontinuity (capture toggle / warp): re-baseline, so the event
# reports no motion. The flag itself is spent in phase one, by the
# state write that acts on it.
delta = (0.0, 0.0) if pointer.discard_next_delta else (x - old[0], y - old[1])
return PointerEvent(
phase=PointerPhase.MOVE,
position=(x, y),
delta=delta,
buttons=_button_mask(inp),
time=stamp,
)
def _contact_event(
self,
pointer_id: int,
contact: _Contact,
phase: PointerPhase,
position: tuple[float, float],
stamp: float,
*,
button: int | None = None,
mods: tuple[bool, bool, bool, bool] = (False, False, False, False),
) -> PointerEvent:
"""The event a finger or pen is, from what the router holds about it.
The delta runs from where the last delivered event left the pointer,
which is its record in ``Input.pointers``. Pressure is honest: 0.0 out
of contact, the measurement in contact, and 1.0 in contact only on
hardware that cannot measure. A hovering pen that nobody holds primacy
against is primary, so it drives the cursor; a finger is primary only
from its own press.
"""
prior = self._input._pointer(self.source_id).pointers.get(pointer_id)
delta = (position[0] - prior.position[0], position[1] - prior.position[1]) if prior is not None else (0.0, 0.0)
mask = contact.mask
touching = bool(mask & contact.primary_bit)
primary = self._primary
ctrl, shift, alt, meta = mods
return PointerEvent(
phase=phase,
pointer_id=pointer_id,
pointer_type=contact.pointer_type,
is_primary=primary == pointer_id
or (primary is None and contact.pointer_type is PointerType.PEN and not touching),
position=position,
delta=delta,
button=None if button is None else DeviceButton(DeviceType(contact.pointer_type.value), index=button),
buttons=mask,
contact_mask=contact.primary_bit,
pressure=(1.0 if contact.pressure is None else contact.pressure) if touching else 0.0,
tilt=contact.tilt,
twist=contact.twist,
distance=contact.distance,
tangential_pressure=contact.tangential_pressure,
tool=contact.tool,
tool_id=contact.tool_id,
contact_size=contact.contact_size if touching else (0.0, 0.0),
device_id=contact.device_id,
time=stamp,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
[docs]
def is_key_down(self, code: int) -> bool:
"""Whether *code* is currently held, for backends that must infer key repeat.
The device record: a repeat is a fact about the key under the
player's finger, not about which tree the last press reached.
"""
return code in self._input._devices.keys_pressed
# -- The door ---------------------------------------------------------
[docs]
def publish(self, event: InputEvent) -> Node | None:
"""Deliver one input event. The only way input enters the engine.
Backends, the simulator, AI agents, on-screen controls and replays all
arrive here, and the entry points below are translation layers that
build an event and call this. Input is pumped once per frame, before
the physics tick, and each event dispatches synchronously inside that
pump.
**A publish made during dispatch is queued and drained in order after
the current event completes.** An on-screen controller publishing a button
from inside a pointer's targeted phase, and a recogniser publishing a
gesture from motion, are the cases; queueing is what keeps the frame's
events in arrival order rather than nesting one inside another. The
queue belongs to the tree, because several routers can feed one tree
and a second queue would let a nested publish overtake.
**A key repeat runs the targeted phase only.** An echo reaches the
focused widget, so a held key still repeats in a text field and a held
direction still walks a menu, and it never resolves to an action, so a
game action fires once per physical press. The broadcast offers it only
to the key handlers that asked for repeats, ``repeat=True``.
Returns the node that claimed the event, or None when nobody did or
when it was queued behind the event being dispatched. Everything that
watches every event, a recorder or a debugger, connects to the tree's
``input_delivered`` instead of asking the event.
"""
tree = self.tree
# Stamped here rather than at each entry point, so an event published
# straight through the door carries its source too. A forwarded event
# is re-stamped by the router that forwards it, which is right: the
# host is the source as far as the tree it forwards into is concerned.
event.source_id = self.source_id
if (self._pending_move is not None or self._pending_locked is not None or self._moving) and not (
isinstance(event, PointerEvent) and event.phase is PointerPhase.MOVE
):
# Anything that is not itself motion arrives after the motion that
# preceded it, whoever published it. The entry points that read the
# pointer position flush before they build their event; this is the
# backstop for an event published straight through the door.
self.flush_motion()
if tree is None:
# No tree to deliver to, but the device is still the device: a
# state-only router exists precisely so a headless backend keeps
# ``Input`` honest.
self._write_device_state(event)
return None
if tree._input_dispatching:
tree._input_queue.append((self, event))
return None
tree._input_dispatching = True
try:
claimant = self._deliver(event)
queue = tree._input_queue
while queue:
router, queued = queue.popleft()
router._deliver(queued)
finally:
tree._input_dispatching = False
tree._input_queue.clear()
# A handler that opened or closed a context has release edges
# waiting behind it; they belong to this frame, not the next.
tree._flush_action_edges()
return claimant
def _resolve_pointer(self, event: PointerEvent) -> None:
"""Fill in what only a run of events can say: the click count and the drag.
Both questions are asked by nearly every widget that takes a pointer,
and both need the previous press to answer. Answering them here means
one clock and one slop for the whole engine: a double click is the same
gesture in a text field, a file list and a game, and what counts as a
drag rather than a click that wobbled does not depend on which widget
the pointer landed on.
The slop is per pointer type, because a finger lands a good deal
further from where it meant to than a mouse does, and a pen sits
between them.
"""
track = self._pointers.get(event.pointer_id)
if track is None:
track = self._pointers[event.pointer_id] = _PointerTrack()
phase = event.phase
if phase is PointerPhase.DOWN:
# A finger landing continues the run of the last finger to lift,
# unless another finger is still down: two together are a
# two-finger tap, not a double tap.
previous = track
if event.pointer_type is PointerType.TOUCH and track.click_count == 0:
fingers_down = any(t.down and t.pointer_type is PointerType.TOUCH for t in self._pointers.values())
if self._last_tap is not None and not fingers_down:
previous = self._last_tap
same_run = (
previous.button == event.button
and previous.click_count > 0
and (event.time - previous.press_time) <= CLICK_INTERVAL
and not _moved_past(event.position, previous.press_pos, CLICK_SLOP.get(event.pointer_type, 5.0))
)
track.click_count = previous.click_count + 1 if same_run else 1
track.button = event.button
track.press_pos = event.position
track.press_time = event.time
track.dragging = False
track.down = True
track.pointer_type = event.pointer_type
tree = self.tree
track.drag_threshold = (
tree.drag_threshold.get(event.pointer_type, DEFAULT_DRAG_THRESHOLD[event.pointer_type])
if tree is not None
else DEFAULT_DRAG_THRESHOLD[event.pointer_type]
)
event.click_count = track.click_count
elif phase is PointerPhase.MOVE:
if track.down and not track.dragging:
track.dragging = _moved_past(event.position, track.press_pos, track.drag_threshold)
event.dragging = track.dragging
elif phase is PointerPhase.UP or phase is PointerPhase.CANCEL:
if track.down:
event.dragging = track.dragging
if track.button == event.button:
event.click_count = track.click_count
track.down = False
if phase is PointerPhase.CANCEL:
# The gesture did not complete, so it starts no click run: a
# cancelled press followed by a real one is one click, not two.
track.click_count = 0
if track.pointer_type is PointerType.TOUCH:
self._pointers.pop(event.pointer_id, None)
self._last_tap = track
[docs]
def cancel_pointers(self) -> None:
"""Abandon every press in flight, telling whoever holds one.
A press that is interrupted rather than completed is a cancel, not a
release: the button does not fire, the slider does not commit and no
release edge is published, but the node holding the pointer is told
that the gesture is over so it can stop dragging. Silence is what left
a splitter believing it was still being dragged after the window lost
focus, and the next pass of the cursor over it moved the divider.
"""
# Every finger and pen in contact first, through the door: each is a
# pointer of its own whose gesture has ended, so whoever holds it hears
# the cancel on its own id and ``Input.pointers`` records the contact
# as taken away rather than lifted.
for pointer_id in [pid for pid, contact in self._contacts.items() if contact.mask]:
self._end_contact(pointer_id, PointerPhase.CANCEL)
tree = self.tree
if tree is None:
return
# Only this window's captures: a router speaks for one source, and
# another window's pointer is not its to take back.
for pointer_id in tree._ui.source_captures(self.source_id):
tree.cancel_capture(pointer_id, self._cancel_event(pointer_id))
for track in self._pointers.values():
track.down = False
track.click_count = 0
self._last_tap = None
def _cancel_event(self, pointer_id: int, position: tuple[float, float] | None = None) -> PointerEvent:
"""The cancel closing whatever press *pointer_id* had in flight.
The button it names is the one the press was made with, because a
widget reads the cancel on the same branch it reads the release. The
router that saw the press knows it; one that did not, because several
routers can feed one tree, falls back to the device record, which is
the only place a press this router never saw is recorded.
"""
track = self._pointers.get(pointer_id)
button = track.button if track is not None and track.down else None
if button is None:
held = self._input._devices.mouse_buttons_pressed
if held:
button = DeviceButton(DeviceType.MOUSE, index=min(held))
return PointerEvent(
phase=PointerPhase.CANCEL,
pointer_id=pointer_id,
source_id=self.source_id,
pointer_type=track.pointer_type if track is not None else PointerType.MOUSE,
is_primary=pointer_id == 0,
position=position if position is not None else (track.press_pos if track is not None else (0.0, 0.0)),
button=button,
dragging=track.dragging if track is not None else False,
)
def _deliver(self, event: InputEvent) -> Node | None:
"""One event through the five phases, and then the fifth: tell the tree who claimed it.
Returns the node that claimed it, or None when nobody did.
"""
claimant = self._deliver_phases(event)
tree = self.tree
if tree is not None:
if tree._has_repeat_handlers and isinstance(event, KeyEvent) and not event.echo:
# Who took a key's press owns its auto-repeats in the broadcast,
# as the claimant of a press owns them in the UI and in capture.
if event.pressed and claimant is not None:
tree._repeat_keys[event.key] = claimant
else:
tree._repeat_keys.pop(event.key, None)
if isinstance(event, PointerEvent) and event.phase is PointerPhase.DOWN:
self._settle_drag_threshold(tree, event, claimant)
tree.input_delivered.emit(event, claimant)
return claimant
def _settle_drag_threshold(self, tree: SceneTree, event: PointerEvent, claimant: Node | None) -> None:
"""Read the press's drag threshold from the node that took it: its captor, else its claimant.
What counts as a drag belongs to whoever the gesture belongs to, so a
card table that asks more travel of a finger gets it for its own
presses and nowhere else. A press nobody took keeps the tree's.
"""
track = self._pointers.get(event.pointer_id)
if track is None:
return
owner = tree._ui._captor_for(event.source_id, event.pointer_id) or claimant
if owner is not None and owner.tree is tree:
track.drag_threshold = owner.drag_threshold_for(event.pointer_type)
def _deliver_phases(self, event: InputEvent) -> Node | None:
"""Phases one to four for one event, outside the re-entrancy guard; returns the claimant."""
if isinstance(event, PointerEvent):
_complete_pointer(self._input, event)
# Phase zero: what the event means in the run it belongs to, before
# anything is told about it, so the UI and the broadcast read the
# same click count and the same drag off one resolution.
self._resolve_pointer(event)
elif isinstance(event, (ButtonEvent, AxisEvent)):
_complete_device_event(self._input, event)
self._write_device_state(event) # 1
tree = self.tree
if tree is not None and tree._input_listeners:
# A listener hearing the next input takes it ahead of the targeted
# phase, and whole: no widget, action or handler hears it. The
# press is claimed as any press is, so its release inherits the
# claim whether or not the listener is still there to hear it.
listener = tree.listen_input(event)
if listener is not None:
self._claimed(event, listener)
return listener
if tree is not None and tree._has_capture_handlers:
# The capture phase: root-first, ahead of every control, and a claim
# is the whole event, its release included.
captor = tree.capture_input(event)
if captor is not None:
self._claimed(event, captor)
return captor
claimant = self._deliver_targeted(event) # 2
picked, self._release_picked = self._release_picked, None
if isinstance(event, KeyEvent) and event.echo:
# A repeat runs the targeted phase and stops: a held key repeats in
# a text field and walks a menu, and a game action fires once per
# physical press. Only a key handler that asked, ``repeat=True``,
# hears one in the broadcast.
if claimant is None and tree is not None and tree._has_repeat_handlers:
return tree._broadcast_repeat(event)
return claimant
if self._claimed(event, claimant):
return claimant
tree = self.tree
if picked is not None:
# A release whose press nobody claimed, taken by a release pick, is
# claimed as a broadcast handler claims it: no handler hears it, and
# the action its press opened still closes with its edge.
if tree is not None:
tree._broadcast_input(event, handlers=False)
return picked
if tree is not None:
return tree._broadcast_input(event) # 3 and 4
return None
def _claimed(self, event: InputEvent, claimant: Node | None) -> bool:
"""Whether the targeted phase took *event*, and record what that means.
A claim here is the whole of the event: no action state, no edge and
no broadcast. The control that swallowed a press owns what follows
from it, which is why the on-screen button that opens a menu no longer
also fires the weapon bound to the same mouse button.
**The claim is decided at the press and the release inherits it.** A
press nobody claimed has already opened an action, so letting a widget
that happened to be under the pointer at release time claim the
release would strand that action open for ever. A press that *was*
claimed carries its decision to the release that closes it, wherever
that release lands. The one exception is a release pick that claims
a release whose press nobody claimed: :meth:`_deliver_phases` stops
every later handler for it, as a broadcast claim does, and leaves the
action layer to close the action with its edge.
Polled device state is never touched: ``is_mouse_button_pressed``
reports the hardware, and that is the honest-and-routed split.
"""
inp = self._input
slots = _claim_slots(event, inp)
# A finger or a pen, whose press claim this router carries itself.
contact = event if isinstance(event, PointerEvent) and event.pointer_type is not PointerType.MOUSE else None
if _is_press(event):
if claimant is None:
return False
inp._claim_slots(slots)
if contact is not None:
self._claimed_contacts[(contact.pointer_id, contact.button)] = slots
elif contact is not None and contact.phase in (PointerPhase.UP, PointerPhase.CANCEL):
# Its release belongs to its press, whoever holds it now, exactly
# as a mouse button's does through its slot.
claimed = self._claimed_contacts.pop((contact.pointer_id, contact.button), None)
if claimed is None:
return False
inp._drop_claims(claimed)
elif slots:
if not inp._slots_claimed(slots):
return False
# The press is over, so the claim on the element is spent. The
# stale mark is not: it goes at the end of the frame, once the
# element is no longer held, so this release reads at rest too.
inp._drop_claims(slots)
elif claimant is None:
# Nothing pairing a release with a press: the wheel and a typed
# character are each complete on their own, so the targeted
# phase's answer is the whole decision.
return False
return True
def _write_device_state(self, event: InputEvent) -> None:
"""Phase one: what the event says the device is doing, recorded as fact.
Never gated by a claim. A broadcast claim stops later handlers and
withdraws nothing from ``Input``: to make an input vanish from the
game, use focus, an overlay or the node's update mode, not consumption.
A discontinuity in the platform's own stream -- a warp, a capture
toggle -- is resolved before the event exists, by the entry point that
owns ``_discard_next_mouse_delta``, so what arrives here is already the
motion that happened.
"""
inp = self._input
if isinstance(event, KeyEvent):
if not event.echo:
if event.pressed:
inp._note_input_method(InputMethod.KEYBOARD_MOUSE)
# _on_key keeps the typed and string-keyed state in lock-step,
# just_pressed included, so this is the whole state update.
inp._on_key(_index_key(event), event.pressed, event.device_id)
elif isinstance(event, PointerEvent):
_declare_pointer_facts(event)
_note_pointer_method(inp, event)
# A pointer act makes its source the one the source-less queries
# answer for: a press carries a position even when nothing moved.
inp._last_source = event.source_id
inp._write_pointer(event)
phase = event.phase
if event.pointer_type is not PointerType.MOUSE:
self._write_contact_buttons(inp, event)
if event.is_primary and inp.emulate_mouse:
self._emulate_mouse(inp, event)
elif phase is PointerPhase.MOVE:
if event.locked:
# A locked pointer has no meaningful screen position, so it
# contributes delta only and the stored position stands.
pointer = inp._pointer(event.source_id)
pointer.delta = (
pointer.delta[0] + event.delta[0],
pointer.delta[1] + event.delta[1],
)
else:
# The state layer owns absolute motion: the delta
# accumulator and the warp re-baseline are one fact each,
# written in one place rather than copied here.
inp._on_mouse_move(*event.position, source_id=event.source_id)
elif event.button is not None and (phase is PointerPhase.DOWN or phase is PointerPhase.UP):
# A press lands where it says it did, even one published
# straight through the door with no motion before it.
if not event.locked:
inp._pointer(event.source_id).position = event.position
index = _index_of(inp, event)
pressed = phase is PointerPhase.DOWN
_write_mouse_button(inp, index, pressed)
inp._on_button(("mouse", event.device_id, index), inp._binding_names(event), pressed, event.device_id)
elif isinstance(event, ScrollEvent):
inp._last_source = event.source_id
wheel = inp._pointer(event.source_id)
wheel.scroll = (wheel.scroll[0] + event.scroll[0], wheel.scroll[1] + event.scroll[1])
elif isinstance(event, ButtonEvent):
if event.pressed and event.button.device is DeviceType.CONTROLLER:
_note_controller_method(inp, event)
index = _index_of(inp, event)
inp._on_button(
("device", event.device_id, index), inp._binding_names(event), event.pressed, event.device_id
)
elif isinstance(event, AxisEvent):
if event.axis.device is DeviceType.CONTROLLER:
_note_controller_method(inp, event)
inp._on_device_axis(inp._axis_forms(event), event.value, event.delta, event.device_id)
# The device record for a pad is written by the poll that found it, not
# here: a backend polls a pad rather than receiving a callback, and the
# state layer records what changed and queues the events this publishes
# (see ``flush_joy_events``). What the event writes is the *delivered*
# record of the tree it reaches, which is how a hosted game hears a pad
# its host forwarded and no pad its host kept. The poll already
# delivered it to the tree that polled, and both writes are the same
# value, so arriving twice records one edge.
@staticmethod
def _write_contact_buttons(inp: _Input, event: PointerEvent) -> None:
"""A finger's or a pen's buttons into this tree's delivered record, by name.
A press adds the names it answers to; anything this pointer held that
the event's mask no longer holds goes, with a release edge on a lift
and none on a cancel or a pen carried out of range, because that press
never completed. Every button the pointer held goes with its contact,
so a barrel held as the pen lifts is not left pressed for ever.
"""
phase = event.phase
pointer_id = event.pointer_id
if phase is PointerPhase.DOWN:
if event.button is not None:
inp._on_button(
("pointer", pointer_id, _index_of(inp, event)), inp._binding_names(event), True, event.device_id
)
return
if phase is PointerPhase.MOVE or phase is PointerPhase.ENTER:
return
held = event.buttons
gone = [
press
for press in inp._button_presses
if press[0] == "pointer" and press[1] == pointer_id and not held & (1 << press[2])
]
for press in gone:
if phase is PointerPhase.UP:
inp._on_button(press, (), False)
else:
inp._abandon_button(press)
@staticmethod
def _emulate_mouse(inp: _Input, event: PointerEvent) -> None:
"""The mouse state the primary finger or pen stands in for.
State only: ``mouse_position`` and, for the pointer's contact and
nothing else, the mouse's primary button in the device record. The
delivered record already took the mouse's primary names with the
contact's own (see :func:`_complete_pointer`), so every binding on the
mouse's primary hears it. No mouse event is made, so a handler and a
control hear the finger once, as itself. A finger lands where it
lands, so a press moves the recorded position without counting as
travel; a cancel lets the button go with no release edge, because the
gesture never completed.
"""
phase = event.phase
if phase is PointerPhase.MOVE:
inp._on_mouse_move(*event.position, source_id=event.source_id)
return
if event.button is None or not event.contact_mask & (1 << _index_of(inp, event)):
return
primary = inp._mouse_layout().primary
if primary is None:
return
if phase is PointerPhase.DOWN or phase is PointerPhase.UP:
inp._pointer(event.source_id).position = event.position
_write_mouse_button(inp, primary, phase is PointerPhase.DOWN)
elif phase is PointerPhase.CANCEL:
inp._abandon_mouse_button(primary)
def _deliver_targeted(self, event: InputEvent) -> Node | None:
"""Phase two: the delivery that belongs to one node rather than the tree.
The focus owner for a key; for a pointer, the captor, else the control
under it, else the nearest pickable node under the ray.
Returns the node that took the event, or ``None`` when nothing here
wanted it. That answer is the claim: :meth:`_claimed` turns it into no
action state, no edge and no broadcast, so a widget that swallows a
press really does swallow it.
"""
tree = self.tree
if tree is None:
return None
if isinstance(event, KeyEvent):
return tree.ui_key(event)
if isinstance(event, TextEvent):
return tree.ui_text(event)
if isinstance(event, PointerEvent):
taken = tree.ui_pointer(event)
return taken if taken is not None else self._pick(tree, event)
if isinstance(event, ScrollEvent):
# The UI receives the same object the tree does: a wheel notch is a
# scroll, not a key named after one.
return tree.ui_scroll(event)
if (
isinstance(event, (ButtonEvent, AxisEvent))
and (event.button.device if isinstance(event, ButtonEvent) else event.axis.device) is DeviceType.CONTROLLER
):
# The pad is a navigation device like any other, so it reaches the
# focus route here rather than only the broadcast. Without this a
# player holding a controller cannot leave a title screen.
return tree.ui_pad(event)
return None
def _pick(self, tree: SceneTree, event: PointerEvent) -> Node | None:
"""Cast the pointer's ray and deliver ``on_picked``, the last targeted step.
Picking is on the event path rather than polled after the tick, so it
happens for the press it belongs to: every pointer picks, not only the
mouse; every button picks, not only the left one; the source event is
on the hit; and every way of running a scene picks the same, whether
the frame loop, the browser runtime or a bare
:class:`~simvx.core.testing.SceneRunner` is driving it.
A locked pointer has no screen position to cast through. Every press
picks; a release picks only while a node in the tree declares an
``@on_picked(..., edge="release")`` handler, so a tree that never asks
casts no second ray. A handler that wants the whole gesture takes the
pointer events themselves.
Returns the picked node only when its ``on_picked`` claimed the press.
A pick that merely happened does not: clicking a barrel usually means
firing at it as well as touching it, and ``on_picked`` claims the same
way every other handler does, by saying so.
A release is never returned, because a release belongs to whoever
claimed its press. A release pick that claims is kept aside instead,
and :meth:`_deliver_phases` lets it stop every later handler only
when nobody claimed the press.
"""
if event.locked or event.button is None:
return None
if event.phase is PointerPhase.DOWN:
node, claimed = tree._cast(event.position, button=event.button, source=event)
return node if claimed else None
if event.phase is PointerPhase.UP and tree._release_pickers:
node, claimed = tree._cast(event.position, button=event.button, source=event, pressed=False)
if claimed:
self._release_picked = node
return None
# -- Keyboard ---------------------------------------------------------
[docs]
def key(
self,
code: int | DeviceButton,
pressed: bool,
*,
echo: bool = False,
ctrl: bool | None = None,
shift: bool | None = None,
alt: bool | None = None,
meta: bool | None = None,
time: float | None = None,
label: str | None = None,
device_id: int = 0,
) -> None:
"""Route a key event.
Args:
code: The platform code of a **physical** key position, in the
engine's table (``Key``), or the key itself. A backend whose
native codes are layout keysyms converts first. A code the
table does not name arrives as ``Key.button(code)``.
pressed: True on press, False on release.
echo: True for an auto-repeat press. Repeats run the targeted phase
only, so a held key still repeats in a text field while a game
action fires once per physical press.
ctrl, shift, alt, meta: Modifier state at event time. A backend
that does not report them leaves them unset and the held-key
state answers instead.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
label: What this key prints under the player's layout. Left unset,
it falls back to the engine's US table, so a rebind screen
shows the physical name rather than nothing.
device_id: The backend's id for the keyboard; 0 is the aggregate device.
"""
if not isinstance(code, int):
resolved = key_code(code)
if resolved is None:
raise TypeError(f"{code!r} is not a key")
code = resolved
mods = _resolve_modifiers(self._input, ctrl, shift, alt, meta)
# A modifier key's OWN event reports itself from the event, never from
# state and never from the backend. Both of those answer "before or
# after?" and disagree: the held-key fallback is read before the state
# write, so a Shift press used to arrive carrying ``shift=False`` and
# its release ``shift=True``, the inverse of the truth. A handler
# filtered on a modifier-bound action then matched the press and never
# the release. ``pressed`` is the answer and needs no ordering.
self_mod = SELF_MODIFIER.get(code)
if self_mod is not None:
mods = mods[:self_mod] + (pressed,) + mods[self_mod + 1 :]
ctrl, shift, alt, meta = mods
stamp = _event_time(time)
keyboard = self._input._devices.device(DeviceType.KEYBOARD, device_id)
self.publish(
KeyEvent(
key=key_for_code(code),
pressed=pressed,
echo=echo,
label=_name_for_code(code) if label is None else label,
device_id=keyboard.id,
time=stamp,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
)
[docs]
def char(self, text: str, *, time: float | None = None) -> None:
"""Route a typed character (text entry only; never a game action).
The modifier flags every event carries are filled from the held-key
state, as they are on a pointer or a wheel notch that reports none of
its own. No backend reports them on a character, and a handler that
reads ``event.ctrl`` off one is owed the truth rather than a standing
False.
Args:
text: The characters produced.
time: The platform's event time on the monotonic clock. Left unset,
the event is timed at receipt.
"""
ctrl, shift, alt, meta = _resolve_modifiers(self._input, None, None, None, None)
self.publish(TextEvent(text=text, time=_event_time(time), ctrl=ctrl, shift=shift, alt=alt, meta=meta))
# -- Mouse ------------------------------------------------------------
[docs]
def mouse_button(
self,
button: int | DeviceButton,
pressed: bool,
*,
ctrl: bool | None = None,
shift: bool | None = None,
alt: bool | None = None,
meta: bool | None = None,
time: float | None = None,
device_id: int = 0,
) -> None:
"""Route a mouse button event.
A pen is not a mouse and does not come through here: it is a pointer
of its own (:meth:`pen`), which stands in for the mouse while it is
primary.
Args:
button: The mouse's own index, as the backend reports it, or a
control such as ``Mouse.RIGHT``, which resolves through the
mouse's layout. A backend whose platform
numbers the mouse its own way (the DOM puts middle before
right) declares that layout rather than renumbering.
pressed: True on press, False on release.
ctrl, shift, alt, meta: Modifier state at event time, carried to
the UI as well as to ``@on_input``, so a widget can implement a
Shift-click. A caller that does not report them leaves them
unset and the held-key state answers instead.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
device_id: Which physical mouse; 0 is the aggregate one.
"""
# The press carries where the pointer is, so the motion that got it
# there is delivered first: a click straight after a move must not land
# at the position the move came from.
self.flush_motion()
ctrl, shift, alt, meta = _resolve_modifiers(self._input, ctrl, shift, alt, meta)
inp = self._input
rec = inp._devices.device(DeviceType.MOUSE, device_id)
if isinstance(button, DeviceButton):
index = inp._devices.layout_of(rec).button_index(button)
if index is None:
raise ValueError(f"the mouse's layout names no {button!r}")
else:
index = int(button)
# The mask an event carries is the state once this event has happened,
# and phase one has not run yet, so fold this button in by hand.
bit = 1 << index
mask = _button_mask(inp)
mask = (mask | bit) if pressed else (mask & ~bit)
self.publish(
PointerEvent(
phase=PointerPhase.DOWN if pressed else PointerPhase.UP,
button=DeviceButton(DeviceType.MOUSE, index=index),
buttons=mask,
position=self._pointer.position,
device_id=rec.id,
time=_event_time(time),
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
)
[docs]
def mouse_motion(self, x: float, y: float, *, time: float | None = None) -> None:
"""Route absolute mouse motion, in logical (screen) pixels.
HiDPI needs no scaling here: cursor coordinates and ``tree.screen_size``
are both logical units even when the framebuffer is larger.
Coalesced: the sample is remembered and the frame delivers one motion
event carrying the total, which is what a camera or a drag wants and
what keeps a 1000Hz mouse from running every handler sixteen times a
frame. Only where the pointer got to is kept, because the delta is
measured from where the last delivery left the recorded position, so
the sum falls out. A handler declared ``@on_input(PointerEvent,
samples=True)`` is handed each sample as it arrives instead.
Args:
x, y: Position in logical (screen) pixels.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
"""
x, y = self._to_viewport(x, y)
stamp = _event_time(time)
tree = self.tree
if tree is not None and tree._has_sample_handlers:
pointer = self._pointer
prev = self._sample_prev.get(0)
if prev is None:
prev = (x, y) if pointer.discard_next_delta else pointer.position
self._sample_prev[0] = (x, y)
self._sample(
PointerEvent(
phase=PointerPhase.MOVE,
position=(x, y),
delta=(x - prev[0], y - prev[1]),
buttons=_button_mask(self._input),
time=stamp,
)
)
self._pending_move = (x, y, stamp)
[docs]
def mouse_motion_relative(self, dx: float, dy: float, *, time: float | None = None) -> None:
"""Route relative pointer motion under pointer lock.
A locked pointer has no meaningful screen position, so this contributes
delta only: the UI is not notified and the stored position stands.
Coalesced like absolute motion, and here the steps really are summed,
because there is no position to read the total off.
Args:
dx, dy: Motion since the previous event, in logical pixels.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
"""
inp = self._input
pointer = self._pointer
if pointer.discard_next_delta:
pointer.discard_next_delta = False
return
dx, dy = self._to_viewport_delta(dx, dy)
stamp = _event_time(time)
tree = self.tree
if tree is not None and tree._has_sample_handlers:
self._sample(
PointerEvent(
phase=PointerPhase.MOVE,
position=pointer.position,
delta=(dx, dy),
buttons=_button_mask(inp),
locked=True,
time=stamp,
)
)
pending = self._pending_locked
if pending is None:
self._pending_locked = (dx, dy, stamp)
else:
self._pending_locked = (pending[0] + dx, pending[1] + dy, stamp)
[docs]
def scroll(self, dx: float, dy: float, *, time: float | None = None) -> None:
"""Route a scroll event. Positive ``dy`` scrolls up.
Args:
dx, dy: The scroll offset.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
"""
# The notch carries where the pointer is, the same as a press does.
self.flush_motion()
ctrl, shift, alt, meta = _resolve_modifiers(self._input, None, None, None, None)
stamp = _event_time(time)
taken = self.publish(
ScrollEvent(
position=self._pointer.position,
scroll=(dx, dy),
time=stamp,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
)
if dy and taken is None and Mouse.WHEEL in self._recognising():
# The wheel as an axis, for the bindings and handlers that name it:
# a scroll a control took (a list scrolling) moves none of them.
rec = self._input._devices.device(DeviceType.MOUSE, 0)
self.publish(AxisEvent(axis=Mouse.WHEEL, value=None, delta=dy, device_id=rec.id, time=stamp))
# -- Controller ----------------------------------------------------------
[docs]
def flush_joy_events(self) -> None:
"""Publish the pad changes recorded since the last flush.
Every other device reaches the engine from the callback that carried
it. A pad has no callback: a backend polls it and publishes one
snapshot per frame, so the state layer records what changed and this
turns those changes into events, a :class:`ButtonEvent` or an
:class:`AxisEvent` carrying the control the pad's layout names. A
backend that polls pads calls this once per poll, after the
transaction closes.
Without it a pad drives polled state only, and neither a handler
filtered on a pad control nor one on an action with a pad binding
fires.
"""
inp = self._input
pending = inp._devices.pending_joy_events
if not pending:
return
# Taken before publishing: an event nobody can receive is still spent,
# and leaving it queued would deliver it to the next tree.
inp._devices.pending_joy_events = []
# Published whether or not there is a tree. ``publish`` writes the
# device-state phase for a treeless router precisely so ``Input`` stays
# honest headless, and the input method is one of those facts: a pad
# pressed with no scene mounted is still the player holding a pad.
# A pad is polled rather than delivered, so the poll's own moment is
# the best time there is for every change it found.
stamp = monotonic()
ctrl, shift, alt, meta = _resolve_modifiers(inp, None, None, None, None)
store = inp._devices
for kind, pad_id, index, value in pending:
pad = store.pads.get(pad_id)
if pad is None:
continue
layout = store.layout_of(pad)
event: InputEvent
if kind == "button":
event = ButtonEvent(
button=layout.button_control(index),
pressed=value > 0.0,
device_id=pad_id,
time=stamp,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
else:
previous = pad.axes_prev.get(index, value)
event = AxisEvent(
axis=layout.axis_control(index),
value=value,
delta=value - previous,
device_id=pad_id,
time=stamp,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
self.publish(event)
# -- Fingers and pens -------------------------------------------------
[docs]
def touch(
self,
finger_id: int,
phase: PointerPhase | str,
x: float,
y: float,
pressure: float | None = None,
*,
contact_size: tuple[float, float] | None = None,
time: float | None = None,
device_id: int = 0,
) -> None:
"""Route one finger's press, move, lift or cancel.
Every finger is a pointer of its own, with an id the router allocates
each time it touches down, published through :meth:`publish` like the
mouse: a handler taking a
``PointerEvent`` hears each finger once, as a finger, with the
pressure the screen reported. The first finger down while no pointer
holds primacy is primary until it lifts, and while ``Input.emulate_mouse``
is on it also drives ``mouse_position`` and the mouse's primary, as state,
so a game written for a mouse plays under a thumb.
A ``CANCEL`` is the system taking the finger away mid-gesture: a phone
edge swipe, a notification, the browser claiming the gesture. The
finger is gone exactly as a lift is, but the gesture did not complete:
whoever held it hears a cancel rather than a release, and the left
button the primary finger held goes up with no release edge, so a
charge shot is dropped rather than fired.
Args:
finger_id: The backend's id for the finger, unique while it is down.
phase: ``DOWN``, ``MOVE``, ``UP`` or ``CANCEL``.
x, y: Position in logical (screen) pixels.
pressure: What the screen measured, 0.0 to 1.0, or None where it
cannot measure, which reads as 1.0 in contact.
contact_size: The contact's width and height in logical pixels,
where the platform reports it.
time: The platform's event time on the monotonic clock. Left unset,
the event is stamped at receipt.
device_id: The backend's id for the touchscreen; 0 is the
aggregate device. Two screens that both call a finger 0 are
two pointers.
Raises:
ValueError: for a phase a finger cannot have. A finger has no
hover, so ``ENTER`` and ``LEAVE`` are refused.
"""
phase = PointerPhase(phase)
if phase is PointerPhase.ENTER or phase is PointerPhase.LEAVE:
raise ValueError(f"a finger cannot {phase.name}: it has no hover. Use DOWN, MOVE, UP or CANCEL")
rec = self._input._devices.device(DeviceType.TOUCH, device_id)
key = (DeviceType.TOUCH, rec.id, finger_id)
pointer_id = self._ids.get(key)
position = self._to_viewport(x, y)
stamp = _event_time(time)
if pressure is not None and not _current_capabilities().touch_pressure:
_declare_capabilities(touch_pressure=True)
if phase is not PointerPhase.MOVE:
# The moves already waiting carry the pose they were made with, so
# they go before this sample changes it.
self.flush_motion()
recognition = self._recognition
if recognition is None and self._recognising():
recognition = self._recognition = GestureRecognition(self)
if recognition is not None and phase is not PointerPhase.DOWN:
# Before the router, so a finger whose contact a gesture cancelled
# is still followed to its lift.
recognition.touch(key, pointer_id, phase, position, stamp)
if pointer_id is not None and self._let_go_departed(pointer_id, phase):
return
contact = self._contacts.get(pointer_id) if pointer_id is not None else None
if phase is PointerPhase.DOWN:
if contact is not None and contact.mask:
return # a second press of a finger already down says nothing new
# A finger touching down is a new pointer, whatever the backend
# calls it: its previous touch is over.
pointer_id = self._ids[key] = next(_pointer_ids)
contact = self._contacts[pointer_id] = self._new_contact(PointerType.TOUCH, rec, key)
elif contact is None or pointer_id is None:
# A move or a lift for a finger this router never saw go down: it
# was pressed on a scene that has since left, or before the router
# existed, and there is nothing here for it to finish.
return
contact.pressure = pressure
if contact_size is not None:
contact.contact_size = self._to_viewport_delta(*contact_size)
if phase is PointerPhase.MOVE:
self._move_contact(pointer_id, contact, position, stamp)
elif phase is PointerPhase.DOWN:
primary = self._input._devices.layout_of(rec).primary
self._begin_contact(pointer_id, contact, 0 if primary is None else primary, position, stamp)
if recognition is not None:
recognition.touch(key, pointer_id, phase, position, stamp)
else:
self._end_contact(pointer_id, phase, position=position, stamp=stamp)
# -- Gestures -----------------------------------------------------------
def _recognising(self) -> frozenset[DeviceControl]:
"""The recognised controls this router's tree names, by a binding, a handler or ``touch_secondary``."""
tree = self.tree
if tree is None:
return frozenset()
named: frozenset[DeviceControl] = tree.input_map._recognised | tree._recognised_by_handlers
secondary = self._input.touch_secondary
return named | {secondary} if secondary is not None else named
def _gesture_button(
self, pointer_id: int, device_id: int, button: DeviceButton, pressed: bool, position: tuple[float, float]
) -> None:
"""Publish a gesture button going down or up on the finger that made it."""
store = self._input._devices
rec = store.device_records.get(device_id)
index = store.layout_of(rec).button_index(button) if rec is not None else None
if index is None:
return # a layout that names no such button has none to press
bit = 1 << index
contact = self._contacts.get(pointer_id)
held = contact.mask & ~bit if contact is not None else 0
event = gesture_event(pointer_id, device_id, button, pressed, position, (held | bit) if pressed else held)
if contact is not None:
contact.mask = event.buttons
self.publish(event)
def _gesture_axis(self, axis: DeviceAxis, motion: float, device_id: int) -> None:
"""Publish a gesture axis's motion this frame."""
self.publish(gesture_axis_event(axis, motion, device_id))
[docs]
def pen(
self,
pen_id: int,
phase: PointerPhase | str,
x: float,
y: float,
*,
button: int | DeviceButton = Pointer.PRIMARY,
pressure: float | None = None,
tilt: tuple[float, float] | None = None,
twist: float | None = None,
distance: float | None = None,
tangential_pressure: float | None = None,
tool: PointerTool | str = PointerTool.TIP,
tool_id: int = 0,
time: float | None = None,
device_id: int = 0,
ctrl: bool | None = None,
shift: bool | None = None,
alt: bool | None = None,
meta: bool | None = None,
) -> None:
"""Route one pen sample: hovering, touching, a barrel button, or leaving.
A pen is a pointer of its own, with an id the router allocates when it
is first seen and keeps while it is in range. It hovers: a ``MOVE``
while nothing is pressed is the pen in the air, with no contact and a
pressure of 0.0, and ``ENTER`` and ``LEAVE`` bracket its time in the
digitiser's range. ``DOWN`` and ``UP`` with the primary button are the
tip touching and lifting; with another button they are a barrel button
pressed while the pen is wherever it is. A hovering pen nobody holds
primacy against is primary, and a pen that touches down while nobody
holds it takes primacy, so under ``Input.emulate_mouse`` a pen drives
the cursor as it always has.
A value left as None is one the platform did not measure: it reads as
the neutral value on the event, and the one that arrives declares the
axis in ``Input.capabilities.pen_axes``. **Turning the pen over ends
the stroke**: a change of ``tool`` while touching lifts the old tool
and presses the new one, so a handler never sees the tool change
inside a contact.
Args:
pen_id: The backend's id for the stylus, unique while it is in range.
phase: ``ENTER``, ``MOVE``, ``DOWN``, ``UP``, ``CANCEL`` or ``LEAVE``.
x, y: Position in logical (screen) pixels, sub-pixel.
button: Which button a ``DOWN`` or ``UP`` is about: the pen's own
index as the backend reports it, or a control resolved through
the pen's layout. The default is the tip.
pressure: 0.0 to 1.0, or None where the pen cannot measure it.
tilt: The barrel's tilt from upright as ``(x, y)`` in radians.
twist: The barrel's rotation about its axis in radians, ``[0, 2pi)``.
distance: Height above the surface while hovering, 0 to 1.
tangential_pressure: The airbrush wheel or barrel squeeze, -1 to 1.
tool: Which end is working: ``TIP``, ``ERASER``, ``AIRBRUSH`` or ``PUCK``.
tool_id: The physical tool's serial, where the platform reports one.
time: The platform's event time on the monotonic clock.
device_id: The backend's id for the tablet; 0 is the aggregate
device.
ctrl, shift, alt, meta: Modifier state at event time, where the
platform reports it.
"""
phase = PointerPhase(phase)
tool = PointerTool(tool)
rec = self._input._devices.device(DeviceType.PEN, device_id)
layout = self._input._devices.layout_of(rec)
if isinstance(button, DeviceButton):
index = layout.button_index(button)
if index is None:
raise ValueError(f"the pen's layout names no {button!r}; its names are {sorted(layout.buttons)}")
else:
index = int(button)
key = (DeviceType.PEN, rec.id, pen_id)
pointer_id = self._ids.get(key)
if pointer_id is None:
pointer_id = self._ids[key] = next(_pointer_ids)
position = self._to_viewport(x, y)
stamp = _event_time(time)
_declare_pen_axes(pressure, tilt, twist, distance, tangential_pressure)
if phase is not PointerPhase.MOVE:
# The moves already waiting carry the pose they were made with, so
# they go before this sample changes it.
self.flush_motion()
if self._let_go_departed(pointer_id, phase):
return
contact = self._contacts.get(pointer_id)
if contact is None:
contact = self._contacts[pointer_id] = self._new_contact(PointerType.PEN, rec, key)
if tool is not contact.tool and contact.mask & contact.primary_bit:
# The pen was turned over mid-stroke. The stroke the old end made
# ends here and the new end's begins, so no contact ever changes
# its tool.
self._end_contact(pointer_id, PointerPhase.UP, position=position, stamp=stamp)
contact = self._contacts[pointer_id] = self._new_contact(PointerType.PEN, rec, key)
if phase is PointerPhase.MOVE:
phase = PointerPhase.DOWN
index = layout.primary if layout.primary is not None else 0
contact.pressure = pressure
contact.tilt = tilt if tilt is not None else (0.0, 0.0)
contact.twist = twist if twist is not None else 0.0
contact.distance = distance if distance is not None else 0.0
contact.tangential_pressure = tangential_pressure if tangential_pressure is not None else 0.0
contact.tool = tool
contact.tool_id = tool_id
mods = _resolve_modifiers(self._input, ctrl, shift, alt, meta)
if phase is PointerPhase.MOVE:
self._move_contact(pointer_id, contact, position, stamp)
elif phase is PointerPhase.DOWN:
self._begin_contact(pointer_id, contact, index, position, stamp, mods)
elif phase is PointerPhase.UP:
self._release_button(pointer_id, contact, index, position, stamp, mods)
elif phase is PointerPhase.ENTER:
self.flush_motion()
self.publish(self._contact_event(pointer_id, contact, phase, position, stamp, mods=mods))
else:
self._end_contact(pointer_id, phase, position=position, stamp=stamp)
def _let_go_departed(self, pointer_id: int, phase: PointerPhase) -> bool:
"""Whether *pointer_id* belongs to a scene that left, releasing what it drove when it lifts.
Nothing is published for it: the scene it was pressed on is gone. The
lift writes the mouse primary's release when the pointer was standing
in for the mouse, and a cancel lets the button go with no release
edge, as a cancel always does.
"""
if pointer_id not in self._departed:
return False
if phase is PointerPhase.MOVE or phase is PointerPhase.ENTER:
return True
drove = self._departed.pop(pointer_id)
inp = self._input
for claimed_key in [k for k in self._claimed_contacts if k[0] == pointer_id]:
inp._drop_claims(self._claimed_contacts.pop(claimed_key))
# What the pointer put in the delivered record goes with it: a lift
# lets go with a release edge, anything else with none.
for press in [p for p in inp._button_presses if p[0] == "pointer" and p[1] == pointer_id]:
if phase is PointerPhase.UP:
inp._on_button(press, (), False)
else:
inp._abandon_button(press)
self._ids = {key: pid for key, pid in self._ids.items() if pid != pointer_id}
if phase is PointerPhase.DOWN:
return False # a fresh press on a reused id is the new scene's own
if drove:
primary = inp._mouse_layout().primary
if primary is not None:
if phase is PointerPhase.UP:
_write_mouse_button(inp, primary, False)
else:
inp._abandon_mouse_button(primary)
return True
def _move_contact(self, pointer_id: int, contact: _Contact, position: tuple[float, float], stamp: float) -> None:
"""A finger or pen moving: handed to the tree's sample handlers now, and coalesced into the frame's flush."""
tree = self.tree
if tree is not None and tree._has_sample_handlers:
sample = self._contact_event(pointer_id, contact, PointerPhase.MOVE, position, stamp)
prev = self._sample_prev.get(pointer_id)
if prev is not None:
sample.delta = (position[0] - prev[0], position[1] - prev[1])
self._sample_prev[pointer_id] = position
self._sample(sample)
if contact.pending is None:
self._moving.append(pointer_id)
contact.pending = (position[0], position[1], stamp)
def _sample(self, event: PointerEvent) -> None:
"""Hand one motion sample, as it arrived, to the ``samples=True`` handlers of this router's tree.
A sample is not published: it writes no device state, reaches no
control and resolves no action, because the frame's merged event does
all of that once. It carries its own position, axes and time, a delta
from the sample before it, and whether the press it belongs to has
become a drag by now.
"""
tree = self.tree
if tree is None:
return
event.source_id = self.source_id
_complete_pointer(self._input, event)
track = self._pointers.get(event.pointer_id)
if track is not None and track.down:
event.dragging = track.dragging or _moved_past(event.position, track.press_pos, track.drag_threshold)
tree._deliver_samples(event)
def _new_contact(self, pointer_type: PointerType, rec: _DeviceRecord, key: tuple) -> _Contact:
"""A record for a finger or pen, with its contact bit from its device's layout."""
primary = self._input._devices.layout_of(rec).primary
return _Contact(pointer_type, rec.id, key, 0 if primary is None else 1 << primary)
def _begin_contact(
self,
pointer_id: int,
contact: _Contact,
button: int,
position: tuple[float, float],
stamp: float,
mods: tuple[bool, bool, bool, bool] | None = None,
) -> None:
"""A button going down on a finger or pen; the primary one is contact.
The motion that got the pointer here is delivered first, so the press
lands where it happened. A contact taken while nobody holds primacy
takes it.
"""
self.flush_motion()
bit = 1 << button
if contact.mask & bit:
return
contact.mask |= bit
if bit == contact.primary_bit and self._primary is None:
self._primary = pointer_id
if mods is None:
mods = _resolve_modifiers(self._input, None, None, None, None)
event = self._contact_event(pointer_id, contact, PointerPhase.DOWN, position, stamp, button=button, mods=mods)
self.publish(event)
def _release_button(
self,
pointer_id: int,
contact: _Contact,
button: int,
position: tuple[float, float],
stamp: float,
mods: tuple[bool, bool, bool, bool],
) -> None:
"""A pen's button coming up; the primary one is the tip leaving the surface."""
self.flush_motion()
bit = 1 << button
if not contact.mask & bit:
return
event = self._contact_event(pointer_id, contact, PointerPhase.UP, position, stamp, button=button, mods=mods)
contact.mask &= ~bit
event.buttons = contact.mask
if bit == contact.primary_bit:
event.pressure = 0.0
event.contact_size = (0.0, 0.0)
if self._primary == pointer_id:
self._primary = None
self.publish(event)
def _end_contact(
self,
pointer_id: int,
phase: PointerPhase,
*,
position: tuple[float, float] | None = None,
stamp: float | None = None,
) -> None:
"""A finger lifting or cancelled, a pen cancelled or leaving range.
Every button the pointer held goes with it. A lift is the ordinary
release of the contact; a cancel names the button it abandons, so a
widget reads it on the branch it reads a release, and carries no
release edge. A pen that leaves range stops being tracked, and a finger
that lifts or is cancelled is gone. Primacy, if this pointer held it,
is free for the next press.
"""
contact = self._contacts.get(pointer_id)
if contact is None:
return
self.flush_motion()
if position is None:
prior = self._input._pointer(self.source_id).pointers.get(pointer_id)
position = prior.position if prior is not None else (0.0, 0.0)
if stamp is None:
stamp = monotonic()
held = contact.mask
# The contact's own bit when it is held, else the lowest button held.
if held & contact.primary_bit:
button: int | None = contact.primary_bit.bit_length() - 1
else:
button = (held & -held).bit_length() - 1 if held else None
mods = _resolve_modifiers(self._input, None, None, None, None)
event = self._contact_event(pointer_id, contact, phase, position, stamp, button=button, mods=mods)
contact.mask = 0
event.buttons = 0
event.pressure = 0.0
event.contact_size = (0.0, 0.0)
if self._primary == pointer_id:
self._primary = None
if contact.pointer_type is PointerType.TOUCH or phase is PointerPhase.LEAVE:
del self._contacts[pointer_id]
self._ids.pop(contact.key, None)
if phase is PointerPhase.LEAVE:
event.button = None
if held:
# A pen carried out of range while still pressing never
# released anything: whoever held it hears the press end
# without completing, and then that the pen has gone.
cancel = self._contact_event(pointer_id, contact, PointerPhase.CANCEL, position, stamp, button=button)
cancel.is_primary = event.is_primary
self.publish(cancel)
self.publish(event)
[docs]
def pointer_id(self, kind: DeviceType | str, backend_id: int = 0, *, device_id: int = 0) -> int | None:
"""The pointer id this router gave the backend's finger or pen *backend_id*, while it is live.
None for one it is not tracking: a finger that has lifted, a pen out of
range. The mouse is always 0 and is not asked for here.
"""
kind = DeviceType(kind)
engine = self._input._devices.device_ids.get((kind, device_id))
return self._ids.get((kind, engine, backend_id)) if engine is not None else None
# -- Devices, and the buttons and axes with no position ----------------
[docs]
def declare_layout(self, layout: DeviceLayout) -> None:
"""Name every device of *layout*'s type this backend reports, by this layout.
For a backend whose platform numbers a kind of device its own way: the
browser reports the mouse in DOM order, middle before right, and a pen
in the W3C numbering, where the barrel is 2. The backend passes its own
indices and declares what they are called; the engine's own table for
the type answers otherwise.
"""
self._input._devices.type_layouts[layout.device] = layout
[docs]
def connect_device(
self,
device_type: DeviceType | str,
*,
device_id: int = 0,
name: str = "",
guid: str = "",
layout: DeviceLayout | None = None,
buttons: int | None = None,
axes: Sequence[AxisReport | str] = (),
wraps: Collection[int] = (),
) -> InputDevice:
"""Announce a device the backend can enumerate: a pen, a tablet pad, a second touchscreen.
A pointer that arrives unannounced is registered under its kind's
layout, so announcing is for what an event cannot say: a name and a
guid for a rebind screen and a saved layout, how many buttons there
are, and what each axis reports.
Args:
device_type: What type of device.
device_id: The backend's id for it.
name: What the driver calls it.
guid: Its model's stable id, or ``""`` where the backend has none;
a pinned rebind persists by it.
layout: The names the backend knows for its indices, or None for
the type's.
buttons: How many buttons it has, or None to count what the
layout names.
axes: What each axis reports, in index order.
wraps: The indices of the axes whose position wraps at 1.0, as a
ring's does.
Returns:
The device, as ``Input.devices`` holds it.
"""
kind = DeviceType(device_type)
if layout is not None and layout.device is not kind:
raise ValueError(f"a {kind.value} was announced with a {layout.device.value} layout")
reports, wrapping = check_axes(axes, wraps)
store = self._input._devices
rec = store.device(kind, device_id)
rec.name = name
rec.guid = guid
rec.layout = layout
rec.button_count = buttons
rec.axis_reports = reports
rec.wraps = wrapping
device = store.device_objects[rec.id]
store.device_connected.emit(device)
return device
[docs]
def disconnect_device(self, device_type: DeviceType | str, *, device_id: int = 0) -> None:
"""The device went: every button it held goes up with no release edge.
Its record answers as a device with nothing held until the frame
boundary, and then leaves ``Input.devices``.
"""
kind = DeviceType(device_type)
store = self._input._devices
rec = store.disconnect_device(kind, device_id)
if rec is None:
return
rec.held.clear()
inp = self._input
for press in [p for p in inp._button_presses if p[0] == "device" and p[1] == rec.id]:
inp._abandon_button(press)
[docs]
def device_button(
self,
device_type: DeviceType | str,
button: int | DeviceButton,
pressed: bool,
*,
device_id: int = 0,
time: float | None = None,
) -> None:
"""Route a button with no position: a tablet's express key.
Args:
device_type: What type of device.
button: The device's own index, or a :class:`~.buttons.Button`
resolved through its layout.
pressed: True on press, False on release.
device_id: The backend's id for the device.
time: The platform's event time on the monotonic clock.
"""
kind = DeviceType(device_type)
store = self._input._devices
rec = store.device(kind, device_id)
layout = store.layout_of(rec)
index = layout.button_index(button) if isinstance(button, DeviceButton) else int(button)
if index is None:
raise ValueError(f"{kind.value}'s layout names no {button!r}; its names are {sorted(layout.buttons)}")
if pressed == (index in rec.held):
return
# The device record is the machine's, written once here rather than
# by every tree the event reaches.
if pressed:
rec.held.add(index)
else:
rec.held.discard(index)
self.publish(
ButtonEvent(
button=DeviceButton(kind, index=index),
pressed=pressed,
device_id=rec.id,
time=_event_time(time),
)
)
[docs]
def device_axis(
self,
device_type: DeviceType | str,
axis: int,
*,
value: float | None = None,
delta: float | None = None,
device_id: int = 0,
time: float | None = None,
) -> None:
"""Route an axis with no position: a tablet's ring or strip.
Pass what the hardware reported and nothing else. An axis announced as
:attr:`~.buttons.AxisReport.ABSOLUTE` takes *value* and the delta is
derived, the short way round one that wraps; a
:attr:`~.buttons.AxisReport.RELATIVE` one takes *delta* and has no
position, which reads None rather than a number nobody measured; one
reporting both takes both. An axis the backend never announced is
taken as absolute.
Raises:
ValueError: when the report does not match what the axis was
announced to report, which is a backend defect worth hearing
about rather than a value to guess at.
"""
kind = DeviceType(device_type)
store = self._input._devices
rec = store.device(kind, device_id)
index = int(axis)
report = rec.axis_reports[index] if index < len(rec.axis_reports) else AxisReport.ABSOLUTE
if report is AxisReport.ABSOLUTE:
if value is None or delta is not None:
raise ValueError(f"{kind.value} axis {index} reports a position: pass value= and not delta=")
before = rec.axis_values.get(index)
if before is None:
delta = 0.0
elif index in rec.wraps:
delta = wrapped_delta(value, before)
else:
delta = value - before
elif report is AxisReport.RELATIVE:
if delta is None or value is not None:
raise ValueError(f"{kind.value} axis {index} reports motion only: pass delta= and not value=")
elif value is None or delta is None:
raise ValueError(f"{kind.value} axis {index} reports a position and motion: pass both")
rec.axis_values[index] = value
rec.axis_deltas[index] = rec.axis_deltas.get(index, 0.0) + delta
self.publish(
AxisEvent(
axis=DeviceAxis(kind, index=index),
value=value,
delta=delta,
device_id=rec.id,
time=_event_time(time),
)
)
# -- Window focus -----------------------------------------------------
[docs]
def focus_lost(self) -> None:
"""Release everything the window was holding when it lost focus.
The release of a key or button held at the moment focus moves away is
delivered to whatever took the focus, so without this the engine goes
on believing it is held: the player alt-tabs mid-sprint and comes back
to a character running into a wall.
The two device kinds part company here, for the same reason a
cancelled touch does. A key goes back up through the ordinary key path,
so its release edge, its tree event and its UI handling all fire
exactly as a real release would: the key genuinely is not held any
more, and a text field repeating a character has to stop. A pointer
does not: its press was interrupted rather than completed, so the held
state simply goes away with no release edge, and whoever holds the
pointer is sent a cancel instead of a release: the click does not
happen, and the drag stops.
Idempotent, which matters because a browser fires ``blur`` and
``visibilitychange`` for the same switch away.
"""
inp = self._input
self._set_window_focus(False)
for code in sorted(inp._devices.keys_pressed):
self.key(code, False)
# Before the held buttons go: the cancel names the button whose press
# it closes, and a router that did not see that press reads it off the
# held state, which this is about to clear.
self.cancel_pointers()
inp._abandon_held_pointers()
[docs]
def focus_gained(self) -> None:
"""The window has the operating system's keyboard focus again.
Nothing is held on the way back in, so this only records it: a window
without keyboard focus moves no focus by hover, and this one may again.
"""
self._set_window_focus(True)
[docs]
@property
def window_focused(self) -> bool:
"""Whether this window has the operating system's keyboard focus, as the platform last said."""
return self._window_focused
def _set_window_focus(self, focused: bool) -> None:
self._window_focused = focused
tree = self.tree
if tree is not None:
tree._ui.window_focus(self.source_id, focused)
__all__ = ["InputRouter"]