Source code for simvx.core.testing.input_sim
"""InputSimulator -- simulate keyboard/mouse/touch input for headless testing.
Each public method does three things in one call:
1. **State injection** -- writes into the ``Input`` state the tree being driven
owns, matching the bytes the platform adapter would write (so
``Input.is_mouse_button_pressed()`` and friends see the press).
2. **Scene-tree event propagation** -- posts a ``TreeInputEvent`` to the
active ``SceneTree`` so ``@on_input`` decorators fire.
3. **UI-tree event propagation** -- posts a ``UIInputEvent`` via
``tree.ui_input(...)`` so ``Control._on_gui_input`` handlers fire.
If no ``SceneTree`` is active (pure logic tests with no tree), steps 2 and 3
are silently skipped.
The keyboard path runs through :class:`InputRouter`, the same object every
window backend feeds, so a simulated key has the shape a real one does:
modifiers and the auto-repeat ``echo`` flag included, and the UI reached as
well as ``@on_input``. The state object is resolved per call from the tree, so
a tree built with ``isolated_input=True`` receives its own input rather than
the process-wide default.
"""
from collections import deque
from ..events import TreeInputEvent
from ..input import Input, JoyAxis, JoyButton, Key, MouseButton
from ..input.router import InputRouter
from ..input.state import _JOY_AXIS_CODES, _JOY_BUTTON_CODES
from ..scene_tree import SceneTree
__all__ = ["InputSimulator"]
# 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.
The target SceneTree is either bound at construction time (when a test
explicitly knows which tree to drive, as ``UITestHarness`` does) or
resolved lazily via ``SceneTree.current()`` on each call (so simple
one-tree scenarios work without plumbing).
Usage:
from simvx.core.input import Key
sim = InputSimulator()
sim.press_key(Key.SPACE)
runner.advance_frames(1)
sim.release_key(Key.SPACE)
"""
# Class-level pending-release queues. ``tap_key`` / ``tap_gamepad`` append
# here so the release fires on the NEXT frame boundary (preserving the
# ``is_action_just_pressed`` edge for the current frame and the
# ``is_action_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 ``InputSimulator()`` ad-hoc without bookkeeping.
_pending_releases: deque[_PendingKey] = deque()
_pending_gamepad_releases: deque[int] = deque()
def __init__(self, tree: SceneTree | None = None):
# If ``tree`` is ``None``, ``_tree()`` falls back to
# ``SceneTree.current()``: the most recently-activated tree. Pass
# an explicit tree when multiple trees coexist (e.g. editor
# scenarios where opening a scene tab creates a new tree).
self._bound_tree: SceneTree | None = tree
# One router per tree driven, rebuilt when the resolved tree changes:
# a router binds its state object at construction, so a cached one must
# not outlive the tree it was built for.
self._router_for: tuple[SceneTree | None, 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_gamepad_releases.clear()
def _tree(self) -> SceneTree | None:
return self._bound_tree if self._bound_tree is not None else SceneTree.current()
def _router(self) -> InputRouter:
"""The router driving the current tree.
Rebuilt whenever the resolved tree changes, so an unbound simulator
follows ``SceneTree.current()`` per call the way its own contract says
it does, and the router it uses always holds that tree's input state.
"""
tree = self._tree()
cached = self._router_for
if cached is None or cached[0] is not tree:
cached = (tree, InputRouter(tree))
self._router_for = cached
return cached[1]
def _input(self):
"""The ``Input`` state this simulator writes.
Resolved from the tree per call: an ``isolated_input=True`` tree owns an
instance that is not the process-wide default, and a test driving one
must read back what it wrote. Falls back to the ambient proxy only when
there is no tree at all.
"""
tree = self._tree()
return tree.input if tree is not None else 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: Key | int, *, echo: bool = False) -> None:
"""Simulate a key press. Accepts Key enum or int.
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 it is not propagated as an ``@on_input`` event, because
a game action fires once per physical press.
"""
ctrl, shift, alt, meta = self._mods()
self._router().key(int(key), True, echo=echo, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def release_key(self, key: Key | int) -> None:
"""Simulate a key release."""
ctrl, shift, alt, meta = self._mods()
self._router().key(int(key), False, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def tap_key(self, key: Key | int) -> None:
"""Press now, schedule release for the next frame boundary.
Pressing AND releasing in the same Python tick lights up both
``is_action_just_pressed`` AND ``is_action_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 ``is_action_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
``is_action_just_pressed`` on the first iteration and
``is_action_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((int(key), ctrl, shift, alt, meta))
[docs]
@classmethod
def flush_pending_releases(cls) -> None:
"""Release every key and gamepad button queued by a tap since the last flush.
Writes the release into the ``Input`` state AND propagates a
:class:`TreeInputEvent` to the active ``SceneTree`` so polling
(``is_action_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_gamepad_releases:
return
keys = list(cls._pending_releases)
buttons = list(cls._pending_gamepad_releases)
cls._pending_releases.clear()
cls._pending_gamepad_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)
# ------------------------------------------------------------------ mouse
def _mouse_button(self, button: MouseButton | int, pressed: bool) -> None:
inp = self._input()
inp._on_mouse_button(int(button), pressed)
tree = self._tree()
if tree is None:
return
mb = button if isinstance(button, MouseButton) else MouseButton(int(button))
ctrl, shift, alt, meta = self._mods()
tree.propagate_input(
TreeInputEvent(
"mouse_button",
mouse_button=mb,
pressed=pressed,
position=inp._mouse_pos,
ctrl=ctrl,
shift=shift,
alt=alt,
meta=meta,
)
)
tree.ui_input(mouse_pos=inp._mouse_pos, button=mb, pressed=pressed, ctrl=ctrl, shift=shift, alt=alt, meta=meta)
[docs]
def press_mouse(
self, button: MouseButton | int = MouseButton.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: MouseButton | int = MouseButton.LEFT) -> None:
"""Simulate mouse button release."""
self._mouse_button(button, False)
[docs]
def click(self, position: tuple[float, float], button: MouseButton | int = MouseButton.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)."""
inp = self._input()
old = inp._mouse_pos
inp._on_mouse_move(x, y)
tree = self._tree()
if tree is not None:
tree.propagate_input(
TreeInputEvent(
"mouse_motion",
position=(x, y),
delta=(x - old[0], y - old[1]),
)
)
tree.ui_input(mouse_pos=(x, y), button=None, pressed=False)
[docs]
def scroll(self, dx: float = 0.0, dy: float = -1.0) -> None:
"""Simulate scroll wheel. dy < 0 = scroll down, dy > 0 = scroll up."""
inp = self._input()
inp._scroll_delta = (dx, dy)
tree = self._tree()
if tree is not None:
tree.propagate_input(
TreeInputEvent(
"scroll",
position=inp._mouse_pos,
delta=(dx, dy),
)
)
if dy > 0:
tree.ui_input(mouse_pos=inp._mouse_pos, key="scroll_up", pressed=True)
elif dy < 0:
tree.ui_input(mouse_pos=inp._mouse_pos, key="scroll_down", pressed=True)
# --------------------------------------------------------------- gamepad
[docs]
def press_gamepad(self, button: JoyButton | int) -> None:
"""Simulate a gamepad button press (single gamepad, no pad_id)."""
self._input()._on_joy_button(int(button), True)
[docs]
def release_gamepad(self, button: JoyButton | int) -> None:
"""Simulate a gamepad button release (single gamepad, no pad_id)."""
self._input()._on_joy_button(int(button), False)
[docs]
def tap_gamepad(self, button: JoyButton | int) -> None:
"""Press now, schedule the release for the next frame boundary.
The gamepad 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_gamepad(button)
type(self)._pending_gamepad_releases.append(int(button))
[docs]
def set_gamepad_axis(self, axis: JoyAxis | int, value: float) -> None:
"""Set a gamepad axis in the engine's convention (single gamepad, 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(int(axis), value)
[docs]
def set_gamepad(
self,
pad_id: int = 0,
*,
buttons: dict[str, bool] | None = None,
axes: dict[str, float] | None = None,
) -> None:
"""Publish one pad's whole state, the way a platform adapter does per frame.
The ergonomic helpers above write the typed ``JoyButton`` / ``JoyAxis``
state the ``InputMap`` binding resolver reads. This writes the per-pad
string-keyed state too, so the ``is_gamepad_pressed(0, "a")`` and
``get_gamepad_axis(0, "lt")`` polling API is drivable from a test
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
:meth:`Input.get_connected_gamepads`, 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: Which 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}``.
"""
inp = self._input()
merged_buttons = dict.fromkeys(_JOY_BUTTON_CODES, False)
merged_buttons.update(inp._gamepad_buttons.get(pad_id, {}))
merged_axes = dict.fromkeys(_JOY_AXIS_CODES, 0.0)
merged_axes.update(inp._gamepad_axes.get(pad_id, {}))
for given, known, kind in ((buttons, _JOY_BUTTON_CODES, "button"), (axes, _JOY_AXIS_CODES, "axis")):
unknown = sorted(set(given or ()) - set(known))
if unknown:
raise ValueError(f"unknown gamepad {kind} name(s) {unknown}; known: {sorted(known)}")
merged_buttons.update({k: bool(v) for k, v in (buttons or {}).items()})
merged_axes.update({k: float(v) for k, v in (axes or {}).items()})
inp._update_gamepad(pad_id, merged_buttons, merged_axes, source="simulated")
# ------------------------------------------------------------------ touch
def _touch(self, finger_id: int, action: int, position: tuple[float, float], pressure: float) -> None:
self._input()._update_touch(finger_id, action, position[0], position[1], pressure)
tree = self._tree()
if tree is not None:
tree.touch_input(finger_id, action, position[0], position[1])
[docs]
def touch_down(self, finger_id: int = 0, position: tuple[float, float] = (0, 0), pressure: float = 1.0) -> None:
"""Simulate a touch press (finger down)."""
self._touch(finger_id, 0, position, pressure)
[docs]
def touch_move(self, finger_id: int = 0, position: tuple[float, float] = (0, 0), pressure: float = 1.0) -> None:
"""Simulate a touch move (finger drag)."""
self._touch(finger_id, 2, position, pressure)
[docs]
def touch_up(self, finger_id: int = 0, position: tuple[float, float] = (0, 0)) -> None:
"""Simulate a touch release (finger up)."""
self._touch(finger_id, 1, position, 0.0)
# ------------------------------------------------------------------ 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_gamepad`` 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_gamepad_releases.clear()