auto-save 2026-04-01 09:03 (+8, ~2)
This commit is contained in:
3
src/executor/__init__.py
Normal file
3
src/executor/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from .adb_executor import ADBExecutor
|
||||
|
||||
__all__ = ["ADBExecutor"]
|
||||
109
src/executor/adb_executor.py
Normal file
109
src/executor/adb_executor.py
Normal file
@@ -0,0 +1,109 @@
|
||||
"""L5 - Action Execution via ADB
|
||||
|
||||
Translates structured actions into ADB commands and executes them on device.
|
||||
Coordinates are normalized (0-1), converted to device pixels at execution time.
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
from config import settings
|
||||
|
||||
|
||||
@dataclass
|
||||
class Action:
|
||||
"""A single GUI action to execute."""
|
||||
type: str # tap, swipe, type, long_press, back, home, scroll, wait
|
||||
x: float = 0.0 # normalized x (0-1)
|
||||
y: float = 0.0 # normalized y (0-1)
|
||||
text: str = "" # for type action
|
||||
x2: float = 0.0 # for swipe end
|
||||
y2: float = 0.0 # for swipe end
|
||||
duration: int = 300 # ms, for long_press and swipe
|
||||
|
||||
|
||||
class ADBExecutor:
|
||||
"""Execute actions on Android device via ADB."""
|
||||
|
||||
def __init__(self, capture):
|
||||
self.capture = capture
|
||||
self.adb = settings.adb_path
|
||||
self.serial = settings.device_serial
|
||||
|
||||
def _adb_cmd(self, *args: str) -> list[str]:
|
||||
cmd = [self.adb]
|
||||
if self.serial:
|
||||
cmd.extend(["-s", self.serial])
|
||||
cmd.extend(args)
|
||||
return cmd
|
||||
|
||||
def _run(self, *args: str):
|
||||
cmd = self._adb_cmd(*args)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(f"ADB command failed: {' '.join(cmd)}\n{result.stderr}")
|
||||
return result.stdout
|
||||
|
||||
def _to_pixels(self, x: float, y: float) -> tuple[int, int]:
|
||||
"""Convert normalized (0-1) coordinates to device pixels."""
|
||||
w, h = self.capture.get_resolution()
|
||||
return int(x * w), int(y * h)
|
||||
|
||||
def execute(self, action: Action) -> str:
|
||||
"""Execute a single action and return a description of what was done."""
|
||||
match action.type:
|
||||
case "tap":
|
||||
px, py = self._to_pixels(action.x, action.y)
|
||||
self._run("shell", "input", "tap", str(px), str(py))
|
||||
desc = f"tap ({px}, {py})"
|
||||
|
||||
case "long_press":
|
||||
px, py = self._to_pixels(action.x, action.y)
|
||||
self._run("shell", "input", "swipe",
|
||||
str(px), str(py), str(px), str(py), str(action.duration))
|
||||
desc = f"long_press ({px}, {py}) {action.duration}ms"
|
||||
|
||||
case "swipe":
|
||||
px1, py1 = self._to_pixels(action.x, action.y)
|
||||
px2, py2 = self._to_pixels(action.x2, action.y2)
|
||||
self._run("shell", "input", "swipe",
|
||||
str(px1), str(py1), str(px2), str(py2), str(action.duration))
|
||||
desc = f"swipe ({px1},{py1}) → ({px2},{py2})"
|
||||
|
||||
case "type":
|
||||
# Escape special characters for ADB
|
||||
escaped = action.text.replace(" ", "%s").replace("&", "\\&")
|
||||
self._run("shell", "input", "text", escaped)
|
||||
desc = f"type '{action.text}'"
|
||||
|
||||
case "back":
|
||||
self._run("shell", "input", "keyevent", "KEYCODE_BACK")
|
||||
desc = "back"
|
||||
|
||||
case "home":
|
||||
self._run("shell", "input", "keyevent", "KEYCODE_HOME")
|
||||
desc = "home"
|
||||
|
||||
case "scroll":
|
||||
# Scroll direction: swipe center screen
|
||||
px, py = self._to_pixels(0.5, 0.5)
|
||||
if action.y < 0: # scroll up
|
||||
self._run("shell", "input", "swipe",
|
||||
str(px), str(py - 300), str(px), str(py + 300), "300")
|
||||
desc = "scroll up"
|
||||
else: # scroll down
|
||||
self._run("shell", "input", "swipe",
|
||||
str(px), str(py + 300), str(px), str(py - 300), "300")
|
||||
desc = "scroll down"
|
||||
|
||||
case "wait":
|
||||
time.sleep(action.duration / 1000)
|
||||
desc = f"wait {action.duration}ms"
|
||||
|
||||
case _:
|
||||
raise ValueError(f"Unknown action type: {action.type}")
|
||||
|
||||
# Wait for UI to settle after action
|
||||
time.sleep(settings.action_delay)
|
||||
return desc
|
||||
Reference in New Issue
Block a user