179 lines
6.3 KiB
Python
179 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Interaction Controller - Injects mouse clicks into a running Ruffle window
|
|
at specified timestamps using xdotool.
|
|
"""
|
|
|
|
import logging
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
|
|
log = logging.getLogger("interaction")
|
|
|
|
|
|
@dataclass
|
|
class ClickEvent:
|
|
"""A single click interaction to perform."""
|
|
t: float # Time (seconds from start of recording) to fire this click
|
|
x: int # X coordinate relative to the window
|
|
y: int # Y coordinate relative to the window
|
|
label: str = "" # Human-readable label
|
|
button: int = 1 # Mouse button (1=left, 2=middle, 3=right)
|
|
double: bool = False # Whether to double-click
|
|
|
|
|
|
class InteractionController:
|
|
"""
|
|
Schedules and dispatches click events into a running Ruffle window.
|
|
|
|
Usage
|
|
-----
|
|
ctrl = InteractionController(window_id="0x1234abc", clicks=[...])
|
|
ctrl.start(recording_start_time=time.monotonic())
|
|
# ... recording happens ...
|
|
ctrl.stop()
|
|
"""
|
|
|
|
def __init__(self, window_id: str, clicks: List[dict]):
|
|
self.window_id = window_id
|
|
self.clicks = [
|
|
ClickEvent(
|
|
t=c["t"],
|
|
x=c["x"],
|
|
y=c["y"],
|
|
label=c.get("label", ""),
|
|
button=c.get("button", 1),
|
|
double=c.get("double", False),
|
|
)
|
|
for c in sorted(clicks, key=lambda c: c["t"])
|
|
]
|
|
self._thread: Optional[threading.Thread] = None
|
|
self._stop_event = threading.Event()
|
|
self._recording_start: float = 0.0
|
|
|
|
def start(self, recording_start_time: float):
|
|
"""Begin the click scheduler. Call immediately after recording starts."""
|
|
self._recording_start = recording_start_time
|
|
self._stop_event.clear()
|
|
self._thread = threading.Thread(target=self._schedule_clicks, daemon=True)
|
|
self._thread.start()
|
|
log.info(f"InteractionController started with {len(self.clicks)} click(s).")
|
|
|
|
def stop(self):
|
|
"""Cancel any pending clicks."""
|
|
self._stop_event.set()
|
|
if self._thread:
|
|
self._thread.join(timeout=2.0)
|
|
|
|
def wait_until_done(self):
|
|
"""Block until all clicks have been dispatched."""
|
|
if self._thread:
|
|
self._thread.join()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
def _schedule_clicks(self):
|
|
for click in self.clicks:
|
|
if self._stop_event.is_set():
|
|
break
|
|
elapsed = time.monotonic() - self._recording_start
|
|
wait = click.t - elapsed
|
|
if wait > 0:
|
|
# Sleep in small increments so we can respond to stop events
|
|
deadline = time.monotonic() + wait
|
|
while time.monotonic() < deadline:
|
|
if self._stop_event.is_set():
|
|
return
|
|
time.sleep(min(0.05, deadline - time.monotonic()))
|
|
|
|
if not self._stop_event.is_set():
|
|
self._fire_click(click)
|
|
|
|
def _fire_click(self, click: ClickEvent):
|
|
"""Execute a click using xdotool."""
|
|
try:
|
|
# Move mouse to position within window
|
|
subprocess.run(
|
|
[
|
|
"xdotool", "mousemove",
|
|
"--window", self.window_id,
|
|
str(click.x), str(click.y),
|
|
],
|
|
check=True,
|
|
timeout=2.0,
|
|
)
|
|
time.sleep(0.05) # Brief settle time
|
|
|
|
# Fire the click(s)
|
|
click_cmd = ["xdotool", "click", "--window", self.window_id, str(click.button)]
|
|
if click.double:
|
|
subprocess.run(click_cmd, check=True, timeout=2.0)
|
|
time.sleep(0.1)
|
|
subprocess.run(click_cmd, check=True, timeout=2.0)
|
|
|
|
log.info(
|
|
f" Click fired: label='{click.label}' "
|
|
f"t={click.t:.2f}s x={click.x} y={click.y} "
|
|
f"btn={click.button} double={click.double}"
|
|
)
|
|
except subprocess.CalledProcessError as e:
|
|
log.error(f"xdotool click failed: {e}")
|
|
except FileNotFoundError:
|
|
log.error("xdotool not found. Install with: sudo dnf install xdotool")
|
|
except subprocess.TimeoutExpired:
|
|
log.error("xdotool click timed out.")
|
|
|
|
|
|
class InteractionMapper:
|
|
"""
|
|
Helper for discovering interaction points interactively.
|
|
Runs Ruffle in the foreground and records the user's clicks + timestamps
|
|
so they can be saved to a config file.
|
|
"""
|
|
|
|
def __init__(self, window_id: str):
|
|
self.window_id = window_id
|
|
self._recorded: list = []
|
|
self._start_time: float = 0.0
|
|
self._running = False
|
|
|
|
def start_recording(self):
|
|
"""
|
|
Listen for clicks in the Ruffle window via xdotool key event tracking.
|
|
Note: This is a best-effort approach — for accurate mapping, use the
|
|
interactive GUI tool (gui.py) which shows a live overlay.
|
|
"""
|
|
self._start_time = time.monotonic()
|
|
self._running = True
|
|
log.info("Interaction recording started. Press Ctrl+C to stop.")
|
|
try:
|
|
while self._running:
|
|
# Poll for click position using xdotool
|
|
result = subprocess.run(
|
|
["xdotool", "getmouselocation", "--window", self.window_id],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=1.0,
|
|
)
|
|
# This is polled, not event-driven — use GUI for real click capture
|
|
time.sleep(0.1)
|
|
except KeyboardInterrupt:
|
|
self._running = False
|
|
log.info("Interaction recording stopped.")
|
|
|
|
def get_recorded(self) -> list:
|
|
return self._recorded
|
|
|
|
def record_click_at(self, x: int, y: int, label: str = ""):
|
|
"""Manually register a click event (called from GUI or external hook)."""
|
|
t = time.monotonic() - self._start_time
|
|
entry = {"t": round(t, 2), "x": x, "y": y, "label": label or f"click_{len(self._recorded)+1}"}
|
|
self._recorded.append(entry)
|
|
log.info(f"Recorded interaction: {entry}")
|
|
return entry
|