"""ScriptRecorder -- records user input and generates replayable test scripts."""
import logging
import time
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from ..input import Input, Key, MouseButton, key_to_name
from ..input.state import _active_input, _Input
from ..node import Node
log = logging.getLogger(__name__)
__all__ = ["RecordedEvent", "ScriptRecorder"]
# Threshold for grouping rapid keypresses into type_text() calls
_TYPE_GROUP_THRESHOLD = 0.2 # seconds
# Minimum distance squared for significant mouse movement
_MOVE_THRESHOLD_SQ = 25.0 # 5px
[docs]
@dataclass
class RecordedEvent:
"""A single recorded input event.
``button`` is a ``MouseButton`` enum for ``click`` events, ``None``
otherwise. Codegen emits explicit ``MouseButton.LEFT/RIGHT/MIDDLE`` so
the generated tests stay readable.
"""
timestamp: float # seconds since recording started
event_type: str # "click", "key_press", "key_release", "mouse_move", "text", "scroll"
x: float = 0.0
y: float = 0.0
button: MouseButton | None = None
key: str = ""
char: str = ""
target_path: str = "" # widget path if identifiable
[docs]
class ScriptRecorder(Node):
"""Records user input events and generates replayable test scripts.
Attach to a scene to record interactions. Call stop_recording() to get
Python source code that replays the session using UITestHarness,
DemoRunner steps, or raw InputSimulator API.
Toggle with F9 or programmatically via start/stop_recording().
"""
format: str = "harness" # "harness", "demo_steps", or "raw_input"
def __init__(self, format: str = "harness", **kwargs):
super().__init__(**kwargs)
self.name = self.name or "ScriptRecorder"
self.format = format
self._recording = False
self._events: list[RecordedEvent] = []
self._start_time = 0.0
self._last_mouse_pos: tuple[float, float] = (0.0, 0.0)
# The live Input handlers this recorder shadows while it is hooked. Each
# wrapper calls through to the one it replaced, and a value of None is
# also what says the hook is not installed.
self._orig_on_key: Callable[[int, bool], None] | None = None
self._orig_on_mouse_button: Callable[[int, bool], None] | None = None
self._orig_on_mouse_move: Callable[[float, float], None] | None = None
# The concrete input state the hook was installed on. The ambient
# ``Input`` is a proxy that resolves to whichever state is active at the
# moment it is read, so hooking and unhooking through it can reach two
# different objects: unhooking under a different active state raised and
# left the hook installed, feeding a recorder that had stopped.
self._hooked_input: _Input | None = None
# ------------------------------------------------------------------ public API
[docs]
def start_recording(self) -> None:
"""Start capturing input events."""
if self._recording:
return
self._recording = True
self._events.clear()
self._start_time = time.monotonic()
self._hook_input()
assert self._hooked_input is not None
self._last_mouse_pos = self._hooked_input._mouse_pos
[docs]
def stop_recording(self) -> str:
"""Stop recording and return generated test code in the current format."""
self._recording = False
self._unhook_input()
return self._generate(self.format)
[docs]
def save_recording(self, path: Path, format: str | None = None) -> None:
"""Save recorded script to a file."""
code = self._generate(format or self.format)
Path(path).write_text(code, encoding="utf-8")
[docs]
@property
def is_recording(self) -> bool:
return self._recording
[docs]
@property
def event_count(self) -> int:
return len(self._events)
[docs]
def record_event(self, event: RecordedEvent) -> None:
"""Manually inject a recorded event (for testing the recorder itself)."""
self._events.append(event)
# ------------------------------------------------------------------ lifecycle
[docs]
def on_update(self, dt: float) -> None:
if Input.is_key_just_pressed(Key.F9):
if self._recording:
self.stop_recording()
else:
self.start_recording()
# ------------------------------------------------------------------ input hooking
def _resolve_input(self) -> _Input:
"""The input state this recorder records, resolved once at hook time.
A recorder mounted in a tree records THAT tree's input, which is the
only answer that works for a tree built with ``isolated_input=True``:
the ambient state is a different object there, so hooking it would
record the events of a tree this recorder is not in and none of its
own. A recorder with no tree records whichever state is active now.
"""
tree = self.tree
if tree is None:
return _active_input.get()
state: _Input = tree.input
return state
def _hook_input(self) -> None:
"""Wrap the input state's instance methods to intercept all events."""
state = self._resolve_input()
self._hooked_input = state
self._orig_on_key = state._on_key
self._orig_on_mouse_button = state._on_mouse_button
self._orig_on_mouse_move = state._on_mouse_move
recorder = self
def hooked_on_key(key: int, pressed: bool):
recorder._orig_on_key(key, pressed)
if recorder._recording:
recorder._on_key_event(key, pressed)
def hooked_on_mouse_button(button: int, pressed: bool):
recorder._orig_on_mouse_button(button, pressed)
if recorder._recording:
recorder._on_mouse_button_event(button, pressed)
def hooked_on_mouse_move(x: float, y: float):
recorder._orig_on_mouse_move(x, y)
if recorder._recording:
recorder._on_mouse_move_event(x, y)
# These three shadow the class's own handlers with instance attributes on
# the resolved input state, which is why _unhook_input restores them with
# `del` rather than by writing the originals back. There is no listener
# seam on the input state to register with instead, so the replacement is
# stated here and the checker is told once per line; a `setattr` spelling
# would hide the same act behind a string.
state._on_key = hooked_on_key # type: ignore[method-assign]
state._on_mouse_button = hooked_on_mouse_button # type: ignore[method-assign]
state._on_mouse_move = hooked_on_mouse_move # type: ignore[method-assign]
def _unhook_input(self) -> None:
"""Restore the instance methods on the state the hook was installed on.
The state is the one remembered at hook time, never a fresh resolve:
whatever is active when recording stops has no bearing on where the
wrappers were put.
"""
state = self._hooked_input
if state is None:
return
if self._orig_on_key is not None:
del state._on_key
self._orig_on_key = None
if self._orig_on_mouse_button is not None:
del state._on_mouse_button
self._orig_on_mouse_button = None
if self._orig_on_mouse_move is not None:
del state._on_mouse_move
self._orig_on_mouse_move = None
self._hooked_input = None
# ------------------------------------------------------------------ event capture
def _elapsed(self) -> float:
return time.monotonic() - self._start_time
def _recorded_mouse_pos(self) -> tuple[float, float]:
"""The cursor position of the state being recorded, not of the ambient one."""
state = self._hooked_input
if state is None: # not hooked: whatever is active is the only answer
state = _active_input.get()
return state._mouse_pos
def _on_key_event(self, key: int, pressed: bool) -> None:
ts = self._elapsed()
pos = self._recorded_mouse_pos()
# Resolve printable character
char = ""
try:
_k = Key(key)
# Printable ASCII range (space through tilde) -- single char keys
if Key.SPACE.value <= key <= 126:
char = chr(key) if key != Key.SPACE.value else " "
# Lowercase letters
if Key.A.value <= key <= Key.Z.value:
char = chr(key + 32)
except ValueError:
pass
key_name = key_to_name(Key(key)) if key in Key._value2member_map_ else f"key_{key}"
event_type = "key_press" if pressed else "key_release"
self._events.append(
RecordedEvent(
timestamp=ts,
event_type=event_type,
key=key_name,
char=char,
x=pos[0],
y=pos[1],
)
)
def _on_mouse_button_event(self, button: int, pressed: bool) -> None:
ts = self._elapsed()
mx, my = self._recorded_mouse_pos()
target = self._identify_target(mx, my)
if pressed:
try:
btn_enum = MouseButton(button) if not isinstance(button, MouseButton) else button
except ValueError:
btn_enum = None
self._events.append(
RecordedEvent(
timestamp=ts,
event_type="click",
x=mx,
y=my,
button=btn_enum,
target_path=target,
)
)
# We record both press and release for raw format; click event covers press
def _on_mouse_move_event(self, x: float, y: float) -> None:
# Collapse redundant moves: only record if distance is significant
dx = x - self._last_mouse_pos[0]
dy = y - self._last_mouse_pos[1]
if dx * dx + dy * dy < _MOVE_THRESHOLD_SQ:
return
self._last_mouse_pos = (x, y)
ts = self._elapsed()
self._events.append(
RecordedEvent(
timestamp=ts,
event_type="mouse_move",
x=x,
y=y,
)
)
# ------------------------------------------------------------------ target identification
def _identify_target(self, x: float, y: float) -> str:
"""Walk the scene tree to find the deepest Control containing (x, y)."""
root = self._tree.root if self._tree else None
if root is None:
return ""
return _find_deepest_control(root, x, y)
# ------------------------------------------------------------------ code generation
def _generate(self, fmt: str) -> str:
if fmt == "harness":
return self._gen_harness()
elif fmt == "demo_steps":
return self._gen_demo_steps()
elif fmt == "raw_input":
return self._gen_raw_input()
raise ValueError(f"Unknown format: {fmt!r}. Use 'harness', 'demo_steps', or 'raw_input'.")
def _gen_harness(self) -> str:
lines = [
"# Recorded by ScriptRecorder",
"from simvx.core.input import MouseButton",
]
for group in self._group_events():
if group["type"] == "text":
lines.append(f"harness.type_text({group['text']!r})")
elif group["type"] == "click":
evt = group["event"]
comment = f" # {evt.target_path}" if evt.target_path else ""
btn = _button_literal(evt.button)
lines.append(f"harness.click(({int(evt.x)}, {int(evt.y)}), button={btn}){comment}")
lines.append("harness.tick()")
elif group["type"] == "key_press":
evt = group["event"]
lines.append(f"harness.press_key({evt.key!r})")
lines.append("harness.tick()")
elif group["type"] == "scroll":
evt = group["event"]
lines.append(f"harness.scroll(({int(evt.x)}, {int(evt.y)}))")
return "\n".join(lines) + "\n"
def _gen_demo_steps(self) -> str:
lines = [
"# Recorded by ScriptRecorder",
"from simvx.core.scripted_demo import Click, TypeText, PressKey, Wait, MoveTo",
"from simvx.core.input import Key, MouseButton",
"steps = [",
]
groups = self._group_events()
prev_time = 0.0
for group in groups:
# Insert Wait steps for gaps > 0.3s
gap = group["timestamp"] - prev_time
if gap > 0.3:
lines.append(f" Wait({gap:.1f}),")
prev_time = group.get("end_time", group["timestamp"])
if group["type"] == "text":
lines.append(f" TypeText({group['text']!r}, delay_per_char=0.05),")
elif group["type"] == "click":
evt = group["event"]
comment = f" # {evt.target_path}" if evt.target_path else ""
btn = _button_literal(evt.button)
lines.append(f" Click({int(evt.x)}, {int(evt.y)}, button={btn}),{comment}")
elif group["type"] == "key_press":
evt = group["event"]
lines.append(f" PressKey(Key.{evt.key.upper()}),")
lines.append("]")
return "\n".join(lines) + "\n"
def _gen_raw_input(self) -> str:
lines = [
"# Recorded by ScriptRecorder",
"from simvx.core.testing.input_sim import InputSimulator",
"from simvx.core.input import Key, MouseButton",
"sim = InputSimulator()",
]
for evt in self._events:
if evt.event_type == "click":
comment = f" # {evt.target_path}" if evt.target_path else ""
btn = _button_literal(evt.button)
lines.append(f"sim.click(({int(evt.x)}, {int(evt.y)}), button={btn}){comment}")
elif evt.event_type == "key_press":
lines.append(f"sim.tap_key(Key.{evt.key.upper()})")
elif evt.event_type == "mouse_move":
lines.append(f"sim.move_mouse({int(evt.x)}, {int(evt.y)})")
return "\n".join(lines) + "\n"
# ------------------------------------------------------------------ event grouping
def _group_events(self) -> list[dict]:
"""Group events: consecutive printable key_presses within threshold become type_text."""
groups: list[dict] = []
i = 0
events = self._events
while i < len(events):
evt = events[i]
if evt.event_type == "key_press" and evt.char and len(evt.char) == 1 and evt.char.isprintable():
# Start a text group
chars = [evt.char]
start_time = evt.timestamp
end_time = evt.timestamp
j = i + 1
while j < len(events):
nxt = events[j]
# Skip key_release events within the group
if nxt.event_type == "key_release":
j += 1
continue
if (
nxt.event_type == "key_press"
and nxt.char
and len(nxt.char) == 1
and nxt.char.isprintable()
and nxt.timestamp - end_time <= _TYPE_GROUP_THRESHOLD
):
chars.append(nxt.char)
end_time = nxt.timestamp
j += 1
else:
break
if len(chars) >= 2:
groups.append(
{
"type": "text",
"text": "".join(chars),
"timestamp": start_time,
"end_time": end_time,
}
)
i = j
continue
# Single char -- fall through to key_press
if evt.event_type == "click":
groups.append({"type": "click", "event": evt, "timestamp": evt.timestamp})
elif evt.event_type == "key_press":
groups.append({"type": "key_press", "event": evt, "timestamp": evt.timestamp})
elif evt.event_type == "scroll":
groups.append({"type": "scroll", "event": evt, "timestamp": evt.timestamp})
i += 1
return groups
# ============================================================================
# Internal helpers for ScriptRecorder
# ============================================================================
def _button_literal(button: MouseButton | None) -> str:
"""Render a ``MouseButton`` (or ``None``) as a code-emit-safe literal."""
if button is None:
return "MouseButton.LEFT"
return f"MouseButton.{button.name}"
def _find_deepest_control(node: Node, x: float, y: float) -> str:
"""Walk the tree to find the deepest Control whose global rect contains (x, y)."""
from ..ui.core import Control
best: Control | None = None
best_path: str = ""
def walk(n: Node, path: str) -> None:
nonlocal best, best_path
if isinstance(n, Control):
rx, ry, rw, rh = n.get_global_rect()
if rx <= x < rx + rw and ry <= y < ry + rh:
best = n
best_path = path
for child in n.children:
child_path = f"{path}/{child.name}" if path else child.name
walk(child, child_path)
walk(node, node.name)
return best_path