3309a47e10
- ADB client (connect, tap, swipe, screencap, app control) - Vision module (in-memory screenshot + OpenCV template matching) - Action loader (auto-discovers .py files in actions/) - Engine (screenshot → detect → act loop with YAML playbook) - 12 built-in actions: tap, click, wait, swipe, find, find_and_click, wait_until, click_seq, keyevent, app, log - Example custom action template - Debug screenshot tool for template creation - Detailed plan.md with 20 actions across 9 phases - Sample playbook with 10 steps
120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
"""
|
|
ADB client — communicates with BlueStacks emulator via Android Debug Bridge.
|
|
All screenshot data stays in memory (never written to disk).
|
|
"""
|
|
import subprocess
|
|
import time
|
|
import logging
|
|
|
|
logger = logging.getLogger("sos.adb")
|
|
|
|
|
|
class AdbError(Exception):
|
|
"""Raised when an ADB command fails."""
|
|
pass
|
|
|
|
|
|
class AdbClient:
|
|
"""Low-level ADB wrapper over a single device."""
|
|
|
|
def __init__(self, host="127.0.0.1", port=5555):
|
|
self.host = host
|
|
self.port = port
|
|
self.serial = None # set by connect()
|
|
|
|
# ── connection ────────────────────────────────────────────
|
|
|
|
def connect(self) -> str:
|
|
"""Connect to the emulator. Returns device serial."""
|
|
out = self._raw(f"connect {self.host}:{self.port}")
|
|
self.serial = f"{self.host}:{self.port}"
|
|
lines = [l.strip() for l in out.splitlines() if l.strip()]
|
|
logger.info("ADB connect: %s", lines[-1] if lines else out)
|
|
return self.serial
|
|
|
|
def disconnect(self):
|
|
"""Disconnect from the emulator."""
|
|
if self.serial:
|
|
self._raw(f"disconnect {self.serial}")
|
|
|
|
def is_connected(self) -> bool:
|
|
"""Check if device shows as 'device' in adb devices."""
|
|
try:
|
|
out = self._raw("devices")
|
|
return self.serial in out and "device" in out.split(self.serial)[1][:20]
|
|
except Exception:
|
|
return False
|
|
|
|
# ── screenshots (pure memory — no files) ──────────────────
|
|
|
|
def screencap(self) -> bytes:
|
|
"""
|
|
Capture screen as raw PNG bytes, fully in memory.
|
|
Returns PNG bytes ready for OpenCV / PIL.
|
|
"""
|
|
raw = self._exec_out(f"shell screencap -p")
|
|
if not raw:
|
|
raise AdbError("screencap returned empty bytes — device disconnected?")
|
|
return raw
|
|
|
|
# ── input actions ─────────────────────────────────────────
|
|
|
|
def tap(self, x: int, y: int):
|
|
"""Tap at screen coordinates (x, y)."""
|
|
self._shell(f"input tap {x} {y}")
|
|
|
|
def swipe(self, x1: int, y1: int, x2: int, y2: int, duration_ms: int = 300):
|
|
"""Swipe from (x1,y1) to (x2,y2) over duration_ms."""
|
|
self._shell(f"input swipe {x1} {y1} {x2} {y2} {duration_ms}")
|
|
|
|
def keyevent(self, keycode: int):
|
|
"""Send a key event (e.g. 4 = BACK, 3 = HOME)."""
|
|
self._shell(f"input keyevent {keycode}")
|
|
|
|
def text(self, text: str):
|
|
"""Type text (only ASCII safe; use with care)."""
|
|
# escape special chars for shell
|
|
safe = text.replace(" ", "%s").replace("'", "\\'")
|
|
self._shell(f"input text '{safe}'")
|
|
|
|
# ── app management ────────────────────────────────────────
|
|
|
|
def force_stop(self, package: str):
|
|
"""Force-stop an app by package name."""
|
|
self._shell(f"am force-stop {package}")
|
|
|
|
def launch_app(self, package: str, activity: str = None):
|
|
"""Launch an app. If activity is omitted, monkey-launch it."""
|
|
if activity:
|
|
self._shell(f"am start -n {package}/{activity}")
|
|
else:
|
|
self._shell(f"monkey -p {package} -c android.intent.category.LAUNCHER 1")
|
|
|
|
# ── helpers ───────────────────────────────────────────────
|
|
|
|
def _shell(self, cmd: str) -> str:
|
|
"""Run 'adb -s <serial> shell <cmd>'. Returns stdout decoded."""
|
|
if not self.serial:
|
|
raise AdbError("Not connected — call connect() first")
|
|
return self._raw(f"-s {self.serial} shell {cmd}")
|
|
|
|
def _exec_out(self, cmd: str) -> bytes:
|
|
"""Run 'adb -s <serial> exec-out <cmd>'. Returns raw bytes (for images)."""
|
|
if not self.serial:
|
|
raise AdbError("Not connected — call connect() first")
|
|
full = ["adb", "-s", self.serial, "exec-out"] + cmd.split()
|
|
logger.debug("ADB exec-out: %s", " ".join(full))
|
|
r = subprocess.run(full, capture_output=True, timeout=30)
|
|
if r.returncode != 0:
|
|
raise AdbError(f"ADB exec-out failed: {r.stderr.decode(errors='replace')}")
|
|
return r.stdout
|
|
|
|
def _raw(self, cmd: str) -> str:
|
|
"""Run 'adb <cmd>'. Returns decoded stdout."""
|
|
full = ["adb"] + cmd.split()
|
|
logger.debug("ADB: %s", " ".join(full))
|
|
r = subprocess.run(full, capture_output=True, text=True, timeout=30)
|
|
if r.returncode != 0:
|
|
raise AdbError(f"ADB error: {r.stderr.strip()}")
|
|
return r.stdout
|