feat(actions): implement ActionExecutor with xdotool and pyautogui backends
This commit is contained in:
parent
0dcc25164d
commit
51a22cbe47
2 changed files with 144 additions and 0 deletions
96
merlin/actions/executor.py
Normal file
96
merlin/actions/executor.py
Normal file
|
|
@ -0,0 +1,96 @@
|
||||||
|
"""
|
||||||
|
ActionExecutor — translate action name strings into OS input events.
|
||||||
|
|
||||||
|
Backends:
|
||||||
|
xdotool — Linux X11; requires xdotool package installed
|
||||||
|
pyautogui — cross-platform fallback
|
||||||
|
auto — detects xdotool at init, falls back to pyautogui
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import subprocess
|
||||||
|
from typing import Literal
|
||||||
|
|
||||||
|
# Suppress pyautogui DISPLAY errors on headless systems
|
||||||
|
if not os.environ.get("DISPLAY"):
|
||||||
|
os.environ["DISPLAY"] = ":0"
|
||||||
|
|
||||||
|
import pyautogui
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
Backend = Literal["xdotool", "pyautogui", "auto"]
|
||||||
|
|
||||||
|
_XDOTOOL_CLICK: dict[str, list[str]] = {
|
||||||
|
"left_click": ["xdotool", "click", "1"],
|
||||||
|
"right_click": ["xdotool", "click", "3"],
|
||||||
|
"double_click": ["xdotool", "click", "--repeat", "2", "1"],
|
||||||
|
"middle_click": ["xdotool", "click", "2"],
|
||||||
|
"scroll_up": ["xdotool", "click", "4"],
|
||||||
|
"scroll_down": ["xdotool", "click", "5"],
|
||||||
|
}
|
||||||
|
|
||||||
|
_XDOTOOL_KEY: dict[str, str] = {
|
||||||
|
"key_enter": "Return",
|
||||||
|
"key_escape": "Escape",
|
||||||
|
"key_space": "space",
|
||||||
|
"key_tab": "Tab",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class ActionExecutor:
|
||||||
|
def __init__(self, backend: Backend = "auto") -> None:
|
||||||
|
if backend == "auto":
|
||||||
|
self._backend: Backend = (
|
||||||
|
"xdotool" if shutil.which("xdotool") else "pyautogui"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
self._backend = backend
|
||||||
|
|
||||||
|
def execute(self, action: str) -> None:
|
||||||
|
"""Execute a named action. Silently ignores unknown action names."""
|
||||||
|
try:
|
||||||
|
if self._backend == "xdotool":
|
||||||
|
self._xdotool(action)
|
||||||
|
else:
|
||||||
|
self._pyautogui(action)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.warning("ActionExecutor error for action=%r: %s", action, exc)
|
||||||
|
|
||||||
|
def _xdotool(self, action: str) -> None:
|
||||||
|
if action in _XDOTOOL_CLICK:
|
||||||
|
subprocess.run(_XDOTOOL_CLICK[action], capture_output=True)
|
||||||
|
elif action in _XDOTOOL_KEY:
|
||||||
|
subprocess.run(
|
||||||
|
["xdotool", "key", _XDOTOOL_KEY[action]], capture_output=True
|
||||||
|
)
|
||||||
|
elif action.startswith("drag_start"):
|
||||||
|
subprocess.run(["xdotool", "mousedown", "1"], capture_output=True)
|
||||||
|
elif action.startswith("drag_end"):
|
||||||
|
subprocess.run(["xdotool", "mouseup", "1"], capture_output=True)
|
||||||
|
else:
|
||||||
|
logger.warning("xdotool: unknown action %r", action)
|
||||||
|
|
||||||
|
def _pyautogui(self, action: str) -> None:
|
||||||
|
if action == "left_click":
|
||||||
|
pyautogui.click()
|
||||||
|
elif action == "right_click":
|
||||||
|
pyautogui.rightClick()
|
||||||
|
elif action == "double_click":
|
||||||
|
pyautogui.doubleClick()
|
||||||
|
elif action == "scroll_up":
|
||||||
|
pyautogui.scroll(3)
|
||||||
|
elif action == "scroll_down":
|
||||||
|
pyautogui.scroll(-3)
|
||||||
|
elif action == "key_enter":
|
||||||
|
pyautogui.press("enter")
|
||||||
|
elif action == "key_escape":
|
||||||
|
pyautogui.press("escape")
|
||||||
|
elif action == "key_space":
|
||||||
|
pyautogui.press("space")
|
||||||
|
else:
|
||||||
|
logger.warning("pyautogui: unknown action %r", action)
|
||||||
48
tests/test_actions/test_executor.py
Normal file
48
tests/test_actions/test_executor.py
Normal file
|
|
@ -0,0 +1,48 @@
|
||||||
|
from unittest.mock import patch, MagicMock
|
||||||
|
from merlin.actions.executor import ActionExecutor
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.subprocess")
|
||||||
|
def test_left_click_calls_xdotool(mock_sub):
|
||||||
|
ex = ActionExecutor(backend="xdotool")
|
||||||
|
ex.execute("left_click")
|
||||||
|
mock_sub.run.assert_called_once_with(["xdotool", "click", "1"], capture_output=True)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.subprocess")
|
||||||
|
def test_right_click_calls_xdotool(mock_sub):
|
||||||
|
ex = ActionExecutor(backend="xdotool")
|
||||||
|
ex.execute("right_click")
|
||||||
|
mock_sub.run.assert_called_once_with(["xdotool", "click", "3"], capture_output=True)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.subprocess")
|
||||||
|
def test_double_click_calls_xdotool(mock_sub):
|
||||||
|
ex = ActionExecutor(backend="xdotool")
|
||||||
|
ex.execute("double_click")
|
||||||
|
mock_sub.run.assert_called_once_with(
|
||||||
|
["xdotool", "click", "--repeat", "2", "1"], capture_output=True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.subprocess")
|
||||||
|
def test_key_enter(mock_sub):
|
||||||
|
ex = ActionExecutor(backend="xdotool")
|
||||||
|
ex.execute("key_enter")
|
||||||
|
mock_sub.run.assert_called_once_with(
|
||||||
|
["xdotool", "key", "Return"], capture_output=True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.subprocess")
|
||||||
|
def test_unknown_action_does_not_raise(mock_sub):
|
||||||
|
ex = ActionExecutor(backend="xdotool")
|
||||||
|
ex.execute("nonexistent_action")
|
||||||
|
mock_sub.run.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
@patch("merlin.actions.executor.pyautogui")
|
||||||
|
def test_pyautogui_fallback_left_click(mock_pag):
|
||||||
|
ex = ActionExecutor(backend="pyautogui")
|
||||||
|
ex.execute("left_click")
|
||||||
|
mock_pag.click.assert_called_once()
|
||||||
Loading…
Reference in a new issue