Source code for simvx.core.testing.input_sim
"""InputSimulator -- simulate keyboard, mouse, touch and pen input for headless testing.
Each public method does three things in one call:
Keyboard, mouse, fingers, a pen and scroll all go through :class:`InputRouter`, the same
object every window backend feeds, so one ``sim.click(pos)`` writes the polled
state, reaches the UI and fires ``@on_input`` in exactly the order and shape a
real click does: modifiers and the auto-repeat ``echo`` flag included. A
simulator that built its own events would be a second delivery path, and the
difference between it and the real one is precisely what a test is there to
catch.
If no ``SceneTree`` is active (pure logic tests with no tree), the router
writes device state and delivers nothing, which is what a state-only test
wants. The state object is resolved per call from the tree, so what the
simulator presses lands in the delivered record of the tree under test and
its actions resolve from it.
"""
from collections import deque
from typing import cast
from ..input import (
ControllerSource,
DeviceButton,
DeviceType,
Input,
Mouse,
Pointer,
PointerPhase,
PointerTool,
)
from ..input import controller as _controller
from ..input.controls import DeviceAxis, key_code
from ..input.devices import ControllerDevice
from ..input.router import InputRouter
from ..node import Node
from ..scene_tree import SceneTree
__all__ = ["InputSimulator"]
def _code(key: DeviceButton | int) -> int:
"""A key's platform code, from the control or the code itself."""
if isinstance(key, int):
return key
code = key_code(key)
if code is None:
raise TypeError(f"{key!r} is not a key")
return code
# Modifier state carried alongside an event: ctrl, shift, alt, meta.
_Mods = tuple[bool, bool, bool, bool]
# Queued key releases carry the modifier state their press carried, so a chord
# released on a later frame is still the chord.
_PendingKey = tuple[int, bool, bool, bool, bool]
[docs]
class InputSimulator:
"""Simulate input events for headless testing.
Drives the engine the same way real platform adapters do: state writes,
scene-tree event propagation, and UI-tree event propagation, so a
single ``sim.click(pos)`` call fires polling state, ``@on_input``
decorators, and ``Control.on_gui_input`` in lockstep. Keys go through
:class:`InputRouter` itself, so a simulated key is indistinguishable from
one a window backend produced.
It drives one tree, named at construction: input belongs to a tree, and a
simulator that guessed would write into whichever tree an earlier test
left current.
Usage:
from simvx.core.input import Key, Mouse
sim = InputSimulator(runner.tree)
sim.press_key(Key.SPACE)
runner.advance_frames(1)
sim.release_key(Key.SPACE)
"""
# Class-level pending-release queues. ``tap_key`` / ``tap_controller`` append
# here so the release fires on the NEXT frame boundary (preserving the
# ``just_pressed`` edge for the current frame and the
# ``just_released`` edge for the next). Drained by
# ``flush_pending_releases()``: ``SceneRunner.advance_frames`` invokes it
# at the end of every iteration (after ``Input._new_frame``). Class-level
# so callers can spin up a simulator ad hoc without bookkeeping.
_pending_releases: deque[_PendingKey] = deque()
_pending_controller_releases: deque[DeviceButton | int] = deque()
def __init__(self, target: SceneTree | Node):
"""Drive *target*: a tree, or a node whose tree is resolved on first use.
A node may be handed over before it is in a tree, so a simulator can
be built beside a scene that ``app.run_headless`` has yet to mount;
using it while the node is in no tree raises.
"""
if not isinstance(target, SceneTree | Node):
raise TypeError(
f"InputSimulator drives one tree: InputSimulator(runner.tree) or InputSimulator(scene_root), "
f"got {target!r}"
)
self._target = target
self._bound: SceneTree | None = target if isinstance(target, SceneTree) else None
#: The window this simulator publishes through, built on first use.
self._window: InputRouter | None = None
# Drop any tap releases left over from a previous test/instance that
# crashed before its release frames advanced. The deques are
# class-level (so `SceneRunner.advance_frames` can drain them without
# holding a sim reference), which means stale entries would
# otherwise fire on the first frame this new instance drives.
cls = type(self)
cls._pending_releases.clear()
cls._pending_controller_releases.clear()
@property
def _bound_tree(self) -> SceneTree:
"""The tree this simulator drives, resolved from its node the first time it is needed."""
if self._bound is None:
node = self._target
assert isinstance(node, Node)
tree = node.tree
if tree is None:
raise RuntimeError(
f"InputSimulator was given {node!r}, which is in no tree yet: use it once the scene is "
"mounted (from on_frame under run_headless, or after SceneRunner.load)"
)
self._bound = tree
return self._bound
def _tree(self) -> SceneTree:
return self._bound_tree
[docs]
@property
def router(self) -> InputRouter:
"""The router this simulator publishes through: its window.
A pointer belongs to its source, so a test that presses through the
simulator and then wants that window to lose focus has to use the
same one. Building a second router models a second window, which
correctly cancels nothing the first one is holding.
"""
return self._router()
def _router(self) -> InputRouter:
"""The router driving this simulator's tree."""
if self._window is None:
self._window = InputRouter(self._bound_tree)
return self._window
def _input(self):
"""The tree's ``Input`` state, which this simulator writes and a test reads back."""
return self._bound_tree.input
def _mods(self) -> _Mods:
"""Modifier state as a window backend would report it with an event.
Derived from the keys currently held rather than passed in, so
``sim.press_key(Key.LEFT_CONTROL)`` followed by ``sim.press_key(Key.S)``
produces the ctrl-flagged event a real ctrl+S produces, with no
bookkeeping asked of the caller.
"""
keys = self._input()._keys
return (
bool(keys.get("ctrl")),
bool(keys.get("shift")),
bool(keys.get("alt")),
bool(keys.get("super")),
)
# ------------------------------------------------------------------ keys
[docs]
def press_key(self, key: DeviceButton | int, *, echo: bool = False) -> None:
"""Simulate a key press. Accepts a key or its platform code.
The modifier flags on the event come from the keys currently held, so a
chord is spelled by pressing its modifier first. ``echo=True`` marks an
auto-repeat press: it reaches the UI, so a held key keeps repeating in a
text field, and in the broadcast only an ``@on_input`` key handler
declared with ``repeat=True`` hears it, because a game action fires
once per physical press.
"""
ctrl, shift, alt, meta = self._mods()
self._router().key(_code(key), True, echo=echo, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def release_key(self, key: DeviceButton | int) -> None:
"""Simulate a key release."""
ctrl, shift, alt, meta = self._mods()
self._router().key(_code(key), False, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def tap_key(self, key: DeviceButton | int) -> None:
"""Press now, schedule release for the next frame boundary.
Pressing AND releasing in the same Python tick lights up both
``just_pressed`` AND ``just_released`` on
the same frame, which breaks edge-triggered game logic
(PyDew Valley and the Tier-1 Balatro port both hit this).
Instead, press immediately so the current frame's
``tree.tick()`` sees ``just_pressed``, then queue the
release for the next frame boundary.
``SceneRunner.advance_frames`` drains the queue at the end of
every iteration (after ``Input._new_frame``), so the typical
flow ``sim.tap_key(); runner.advance_frames(2)`` observes
``just_pressed`` on the first iteration and
``just_released`` on the second. Callers driving
frames outside SceneRunner can flush manually via
:meth:`flush_pending_releases`.
"""
ctrl, shift, alt, meta = self._mods()
self.press_key(key)
type(self)._pending_releases.append((_code(key), ctrl, shift, alt, meta))
[docs]
@classmethod
def flush_pending_releases(cls) -> None:
"""Release every key and controller button queued by a tap since the last flush.
Writes the release into the ``Input`` state AND propagates a
:class:`KeyEvent` to the active ``SceneTree`` so polling
(``just_released``) and ``@on_input`` decorators both
observe the release edge.
Snapshots the queues before iterating so a release handler that
re-queues a release does not extend the current drain.
"""
if not cls._pending_releases and not cls._pending_controller_releases:
return
keys = list(cls._pending_releases)
buttons = list(cls._pending_controller_releases)
cls._pending_releases.clear()
cls._pending_controller_releases.clear()
tree = SceneTree.current()
router = InputRouter(tree)
for key_int, ctrl, shift, alt, meta in keys:
router.key(key_int, False, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
state = tree.input if tree is not None else Input
for button in buttons:
state._on_joy_button(button, False)
router.flush_joy_events()
# ------------------------------------------------------------------ mouse
def _mouse_button(self, button: DeviceButton | int, pressed: bool) -> None:
ctrl, shift, alt, meta = self._mods()
self._router().mouse_button(button, pressed, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def press_mouse(self, button: DeviceButton | int = Mouse.LEFT, position: tuple[float, float] | None = None) -> None:
"""Simulate mouse button press, optionally at a position."""
if position is not None:
self.move_mouse(position[0], position[1])
self._mouse_button(button, True)
[docs]
def release_mouse(self, button: DeviceButton | int = Mouse.LEFT) -> None:
"""Simulate mouse button release."""
self._mouse_button(button, False)
[docs]
def click(self, position: tuple[float, float], button: DeviceButton | int = Mouse.LEFT) -> None:
"""Click at a screen position (press + release)."""
self.press_mouse(button, position)
self.release_mouse(button)
[docs]
def move_mouse(self, x: float, y: float) -> None:
"""Move the mouse cursor to (x, y), and deliver the motion now.
The engine coalesces motion into one event per frame, so a backend's
samples pile up until the frame's input phase spends them. One
simulated move is one frame's worth of motion, and a test that has
just moved the pointer expects it to have arrived, so the flush
happens here rather than leaving every caller to step a frame for it.
"""
router = self._router()
router.mouse_motion(x, y)
router.flush_motion()
[docs]
def scroll(self, dx: float = 0.0, dy: float = -1.0) -> None:
"""Simulate scroll wheel. dy < 0 = scroll down, dy > 0 = scroll up."""
self._router().scroll(dx, dy)
# --------------------------------------------------------------- controller
[docs]
def press_controller(self, button: DeviceButton | int) -> None:
"""Simulate a controller button press on one simulated pad: ``Controller.A``, or the pad's index."""
self._input()._on_joy_button(button, True)
self._router().flush_joy_events()
[docs]
def release_controller(self, button: DeviceButton | int) -> None:
"""Simulate a controller button release on the same pad."""
self._input()._on_joy_button(button, False)
self._router().flush_joy_events()
[docs]
def tap_controller(self, button: DeviceButton | int) -> None:
"""Press now, schedule the release for the next frame boundary.
The controller twin of :meth:`tap_key`, and it exists for the same reason:
pressing and releasing in one tick means no frame ever observes the
button held, so a handler reading held state never runs and both edges
land together.
"""
self.press_controller(button)
type(self)._pending_controller_releases.append(button)
[docs]
def set_controller_axis(self, axis: DeviceAxis | int, value: float) -> None:
"""Set a controller axis in the engine's convention (single controller, no pad_id).
Sticks run -1.0 to 1.0 with +y down; triggers run 0.0 released to 1.0
fully pulled. Values arrive already normalised, as they do from a real
backend, so a simulated resting trigger is 0.0 and not -1.0.
"""
self._input()._on_joy_axis(axis, value)
self._router().flush_joy_events()
[docs]
def set_controller(
self,
pad_id: int = 0,
*,
buttons: dict[str, bool] | dict[int, bool] | None = None,
axes: dict[str, float] | dict[int, float] | None = None,
recognised: bool = True,
name: str | None = None,
guid: str = "",
) -> ControllerDevice:
"""Publish one pad's whole state, the way a platform adapter does per frame, and return it.
The ergonomic helpers above drive one nameless pad. This publishes a
pad of its own, the backend's *pad_id*, and returns the
:class:`~simvx.core.input.devices.ControllerDevice` ``Input.controllers``
holds for it, so a test can exercise local multiplayer, a pad arriving
and a pad leaving without reaching past the public surface.
The named entries merge into the pad's current snapshot; a pad this
simulator has not seen starts neutral (every button up, every axis at
zero). Unknown names raise, since a typo would otherwise read as a
button nothing ever presses.
The pad counts as connected from here on and reads back through
``Input.controllers``, with its
:attr:`~simvx.core.input.enums.ControllerSource.SIMULATED` source, but a
real window backend's poll will not prune it: a test's pad does not
vanish because the machine running the test has no controller plugged
in.
Args:
pad_id: The backend's id for the pad to publish.
buttons: Button names to set, e.g. ``{"a": True}``. See the standard
set both desktop backends report.
axes: Axis names to set, e.g. ``{"lt": 0.5}``.
recognised: False publishes a controller no database recognised,
a flight stick say, whose buttons and axes are keyed by the
driver's index (``{3: True}``) because it has no names.
name: What the device calls itself; ``"Simulated pad <id>"`` by default.
guid: The device's guid, which the player's saved mapping is filed under.
"""
inp = self._input()
pad = inp._devices.pad(pad_id)
label = name if name is not None else f"Simulated pad {pad_id}"
if not recognised:
for given, kind in ((buttons, "button"), (axes, "axis")):
named = sorted(str(k) for k in (given or {}) if not isinstance(k, int))
if named:
raise ValueError(f"an unrecognised controller's {kind}s are keyed by index, got {named}")
inp._update_controller(
pad_id,
{},
{},
source=ControllerSource.SIMULATED,
name=label,
guid=guid,
recognised=False,
raw_buttons={int(k): bool(v) for k, v in (buttons or {}).items()},
raw_axes={int(k): float(v) for k, v in (axes or {}).items()},
)
self._router().flush_joy_events()
return cast(ControllerDevice, inp._devices.objects[pad.id])
merged_buttons = {name: index in pad.held for name, index in _controller.BUTTON_INDEX.items()}
merged_axes = {name: pad.axis_values.get(index) or 0.0 for name, index in _controller.AXIS_INDEX.items()}
known_buttons, known_axes = _controller.BUTTON_INDEX, _controller.AXIS_INDEX
for given, known, kind in ((buttons, known_buttons, "button"), (axes, known_axes, "axis")):
unknown = sorted(str(k) for k in set(given or ()) - set(known))
if unknown:
raise ValueError(f"unknown controller {kind} name(s) {unknown}; known: {sorted(known)}")
merged_buttons.update({str(k): bool(v) for k, v in (buttons or {}).items()})
merged_axes.update({str(k): float(v) for k, v in (axes or {}).items()})
inp._update_controller(
pad_id,
merged_buttons,
merged_axes,
source=ControllerSource.SIMULATED,
name=label,
guid=guid or None,
)
self._router().flush_joy_events()
device: ControllerDevice = inp._devices.objects[pad.id]
return device
# ------------------------------------------------------------------ fingers
def _touch(
self, finger_id: int, phase: PointerPhase, position: tuple[float, float], pressure: float | None
) -> None:
router = self._router()
router.touch(finger_id, phase, position[0], position[1], pressure)
# One simulated move is one frame's worth of motion, delivered now, as
# :meth:`move_mouse` does: a test that moved a finger expects it moved.
router.flush_motion()
[docs]
def pointer_id(self, kind: DeviceType | str, backend_id: int = 0, *, device_id: int = 0) -> int | None:
"""The pointer id the router gave a simulated finger or pen, while it is live.
Pointer ids are opaque, so a test that put finger 0 down asks here
which id ``Input.pointers`` and the events carry for it.
"""
return self._router().pointer_id(kind, backend_id, device_id=device_id)
[docs]
def touch_down(
self, finger_id: int = 0, position: tuple[float, float] = (0, 0), pressure: float | None = None
) -> int:
"""Put a finger on the glass, through the router, and return its pointer id.
``pressure`` left unset is a screen that cannot measure it, which
reads as 1.0 in contact. The id is fresh each time a finger touches
down, as it is on a real screen.
"""
self._touch(finger_id, PointerPhase.DOWN, position, pressure)
pointer = self.pointer_id(DeviceType.TOUCH, finger_id)
assert pointer is not None
return pointer
[docs]
def touch_move(
self, finger_id: int = 0, position: tuple[float, float] = (0, 0), pressure: float | None = None
) -> None:
"""Drag a finger that is down."""
self._touch(finger_id, PointerPhase.MOVE, position, pressure)
[docs]
def touch_up(self, finger_id: int = 0, position: tuple[float, float] = (0, 0)) -> None:
"""Lift a finger."""
self._touch(finger_id, PointerPhase.UP, position, None)
[docs]
def touch_cancel(self, finger_id: int = 0, position: tuple[float, float] = (0, 0)) -> None:
"""Have the system take a finger away mid-gesture, as an edge swipe does."""
self._touch(finger_id, PointerPhase.CANCEL, position, None)
# ------------------------------------------------------------------ pen
[docs]
def pen(
self,
phase: PointerPhase | str,
position: tuple[float, float],
*,
pen_id: int = 0,
button: DeviceButton | int = 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,
) -> int | None:
"""One pen sample through the router; returns the pen's pointer id while it is in range.
``MOVE`` with nothing pressed is the pen hovering; ``DOWN`` and ``UP``
with the primary button are the tip touching and lifting, and with
``Pen.BARREL_1`` or an index a barrel button; ``ENTER`` and
``LEAVE`` bring it into and out of range. An axis left as None is one
the simulated pen does not measure. The sample is delivered at once,
as :meth:`move_mouse` delivers a move.
"""
router = self._router()
router.pen(
pen_id,
phase,
position[0],
position[1],
button=button,
pressure=pressure,
tilt=tilt,
twist=twist,
distance=distance,
tangential_pressure=tangential_pressure,
tool=tool,
tool_id=tool_id,
)
router.flush_motion()
return self.pointer_id(DeviceType.PEN, pen_id)
# ------------------------------------------------------------------ misc
[docs]
def reset(self) -> None:
"""Reset all input state to defaults.
Also drains the class-level tap queues so a ``tap_key`` or
``tap_controller`` from an earlier test cannot fire a stale release the
first time the next test advances a frame.
"""
self._input()._reset()
cls = type(self)
cls._pending_releases.clear()
cls._pending_controller_releases.clear()