's'
This commit is contained in:
eric
423
src/android_control.py
Normal file
423
src/android_control.py
Normal file
@@ -0,0 +1,423 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import xml.etree.ElementTree as ET
|
||||
from typing import Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
|
||||
def adb_available() -> bool:
|
||||
return shutil.which("adb") is not None
|
||||
|
||||
|
||||
async def _run_cmd(
|
||||
command: list[str],
|
||||
*,
|
||||
timeout: float = 20.0,
|
||||
capture_binary: bool = False,
|
||||
) -> tuple[int, str | bytes, str]:
|
||||
proc = await asyncio.create_subprocess_exec(
|
||||
*command,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
try:
|
||||
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
|
||||
except asyncio.TimeoutError:
|
||||
proc.kill()
|
||||
await proc.wait()
|
||||
return 124, b"" if capture_binary else "", "Command timed out"
|
||||
|
||||
if capture_binary:
|
||||
return proc.returncode or 0, stdout or b"", (stderr or b"").decode("utf-8", "replace").strip()
|
||||
return (
|
||||
proc.returncode or 0,
|
||||
(stdout or b"").decode("utf-8", "replace").strip(),
|
||||
(stderr or b"").decode("utf-8", "replace").strip(),
|
||||
)
|
||||
|
||||
|
||||
def _adb_prefix(serial: Optional[str]) -> list[str]:
|
||||
command = ["adb"]
|
||||
if serial:
|
||||
command.extend(["-s", serial])
|
||||
return command
|
||||
|
||||
|
||||
async def _adb_text(serial: Optional[str], *args: str, timeout: float = 20.0) -> tuple[int, str, str]:
|
||||
code, out, err = await _run_cmd(_adb_prefix(serial) + list(args), timeout=timeout)
|
||||
assert isinstance(out, str)
|
||||
return code, out, err
|
||||
|
||||
|
||||
async def _adb_binary(serial: Optional[str], *args: str, timeout: float = 20.0) -> tuple[int, bytes, str]:
|
||||
code, out, err = await _run_cmd(
|
||||
_adb_prefix(serial) + list(args),
|
||||
timeout=timeout,
|
||||
capture_binary=True,
|
||||
)
|
||||
assert isinstance(out, bytes)
|
||||
return code, out, err
|
||||
|
||||
|
||||
def _parse_device_line(line: str) -> Optional[dict]:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("List of devices attached"):
|
||||
return None
|
||||
parts = line.split()
|
||||
if len(parts) < 2:
|
||||
return None
|
||||
|
||||
serial = parts[0]
|
||||
state = parts[1]
|
||||
extras: dict[str, str] = {}
|
||||
for token in parts[2:]:
|
||||
if ":" not in token:
|
||||
continue
|
||||
key, value = token.split(":", 1)
|
||||
extras[key] = value
|
||||
|
||||
return {
|
||||
"serial": serial,
|
||||
"state": state,
|
||||
"usb": extras.get("usb", ""),
|
||||
"product": extras.get("product", ""),
|
||||
"model": extras.get("model", ""),
|
||||
"device": extras.get("device", ""),
|
||||
"transport_id": extras.get("transport_id", ""),
|
||||
"raw": line,
|
||||
}
|
||||
|
||||
|
||||
async def list_android_devices() -> list[dict]:
|
||||
if not adb_available():
|
||||
return []
|
||||
code, out, _ = await _adb_text(None, "devices", "-l")
|
||||
if code != 0:
|
||||
return []
|
||||
devices: list[dict] = []
|
||||
for raw in out.splitlines():
|
||||
parsed = _parse_device_line(raw)
|
||||
if parsed:
|
||||
devices.append(parsed)
|
||||
return devices
|
||||
|
||||
|
||||
async def _shell(serial: str, command: str, timeout: float = 20.0) -> str:
|
||||
code, out, err = await _adb_text(serial, "shell", command, timeout=timeout)
|
||||
if code != 0:
|
||||
raise RuntimeError(err or out or "ADB shell command failed")
|
||||
return out or err
|
||||
|
||||
|
||||
async def _get_display_size(serial: str) -> tuple[int, int]:
|
||||
output = await _shell(serial, "wm size | tail -n 1")
|
||||
match = re.search(r"(\d+)x(\d+)", output)
|
||||
if not match:
|
||||
return 1080, 2400
|
||||
return int(match.group(1)), int(match.group(2))
|
||||
|
||||
|
||||
async def android_device_overview(serial: str) -> dict:
|
||||
if not serial:
|
||||
raise ValueError("Missing Android device serial")
|
||||
|
||||
props = {
|
||||
"model": "getprop ro.product.model",
|
||||
"brand": "getprop ro.product.brand",
|
||||
"android_version": "getprop ro.build.version.release",
|
||||
"sdk": "getprop ro.build.version.sdk",
|
||||
}
|
||||
|
||||
async def grab(name: str, cmd: str) -> tuple[str, str]:
|
||||
try:
|
||||
return name, (await _shell(serial, cmd, timeout=10.0)).strip()
|
||||
except RuntimeError as exc:
|
||||
return name, str(exc)
|
||||
|
||||
results = dict(await asyncio.gather(*(grab(key, cmd) for key, cmd in props.items())))
|
||||
|
||||
battery = await _shell(
|
||||
serial,
|
||||
"dumpsys battery | grep -E 'level|status|powered' | sed 's/^[[:space:]]*//'",
|
||||
timeout=10.0,
|
||||
)
|
||||
focus = await _shell(
|
||||
serial,
|
||||
"(dumpsys window windows | grep -E 'mCurrentFocus' | tail -n 1) || "
|
||||
"(dumpsys activity activities | grep -E 'mResumedActivity' | tail -n 1)",
|
||||
timeout=10.0,
|
||||
)
|
||||
resolution = await _shell(serial, "wm size | tail -n 1", timeout=10.0)
|
||||
rotation = await _shell(serial, "settings get system user_rotation", timeout=10.0)
|
||||
screen_on = await _shell(serial, "dumpsys power | grep -E 'Display Power|mHoldingDisplaySuspendBlocker'", timeout=10.0)
|
||||
|
||||
return {
|
||||
"serial": serial,
|
||||
"model": results.get("model", ""),
|
||||
"brand": results.get("brand", ""),
|
||||
"android_version": results.get("android_version", ""),
|
||||
"sdk": results.get("sdk", ""),
|
||||
"battery": battery,
|
||||
"focus": focus,
|
||||
"resolution": resolution,
|
||||
"rotation": rotation,
|
||||
"screen_state": screen_on,
|
||||
}
|
||||
|
||||
|
||||
def _bounds_to_rect(bounds: str) -> Optional[tuple[int, int, int, int]]:
|
||||
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds.strip())
|
||||
if not match:
|
||||
return None
|
||||
left, top, right, bottom = (int(match.group(i)) for i in range(1, 5))
|
||||
if right <= left or bottom <= top:
|
||||
return None
|
||||
return left, top, right, bottom
|
||||
|
||||
|
||||
def _node_label(node: ET.Element) -> str:
|
||||
text = (node.attrib.get("text") or "").strip()
|
||||
desc = (node.attrib.get("content-desc") or "").strip()
|
||||
resource_id = (node.attrib.get("resource-id") or "").strip()
|
||||
class_name = (node.attrib.get("class") or "").strip()
|
||||
if text:
|
||||
return text
|
||||
if desc:
|
||||
return desc
|
||||
if resource_id:
|
||||
return resource_id.split("/")[-1]
|
||||
return class_name.split(".")[-1] if class_name else "node"
|
||||
|
||||
|
||||
async def android_packages(serial: str, query: str = "", limit: int = 60) -> list[dict]:
|
||||
if not serial:
|
||||
raise ValueError("Missing Android device serial")
|
||||
|
||||
output = await _shell(serial, "pm list packages", timeout=25.0)
|
||||
needle = query.strip().lower()
|
||||
results: list[dict] = []
|
||||
for raw in output.splitlines():
|
||||
package = raw.strip()
|
||||
if package.startswith("package:"):
|
||||
package = package[len("package:") :]
|
||||
if not package:
|
||||
continue
|
||||
if needle and needle not in package.lower():
|
||||
continue
|
||||
results.append(
|
||||
{
|
||||
"package": package,
|
||||
"label": package.split(".")[-1],
|
||||
}
|
||||
)
|
||||
if len(results) >= max(1, min(limit, 200)):
|
||||
break
|
||||
return results
|
||||
|
||||
|
||||
async def android_ui_nodes(serial: str, limit: int = 80) -> list[dict]:
|
||||
if not serial:
|
||||
raise ValueError("Missing Android device serial")
|
||||
|
||||
xml_text = ""
|
||||
for dump_path in ("/data/local/tmp/live-ui.xml", "/sdcard/live-ui.xml"):
|
||||
try:
|
||||
xml_text = await _shell(
|
||||
serial,
|
||||
f"uiautomator dump {dump_path} >/dev/null 2>&1 && cat {dump_path}",
|
||||
timeout=25.0,
|
||||
)
|
||||
except RuntimeError:
|
||||
xml_text = ""
|
||||
if xml_text.strip().startswith("<?xml"):
|
||||
break
|
||||
xml_text = xml_text.strip()
|
||||
if not xml_text.startswith("<?xml"):
|
||||
raise RuntimeError("Failed to dump Android UI hierarchy")
|
||||
|
||||
try:
|
||||
root = ET.fromstring(xml_text)
|
||||
except ET.ParseError as exc:
|
||||
raise RuntimeError(f"Unable to parse Android UI dump: {exc}") from exc
|
||||
|
||||
nodes: list[dict] = []
|
||||
for node in root.iter("node"):
|
||||
bounds = (node.attrib.get("bounds") or "").strip()
|
||||
rect = _bounds_to_rect(bounds)
|
||||
if not rect:
|
||||
continue
|
||||
|
||||
text = (node.attrib.get("text") or "").strip()
|
||||
desc = (node.attrib.get("content-desc") or "").strip()
|
||||
resource_id = (node.attrib.get("resource-id") or "").strip()
|
||||
clickable = node.attrib.get("clickable") == "true"
|
||||
focusable = node.attrib.get("focusable") == "true"
|
||||
scrollable = node.attrib.get("scrollable") == "true"
|
||||
enabled = node.attrib.get("enabled") != "false"
|
||||
if not (text or desc or resource_id or clickable or focusable or scrollable):
|
||||
continue
|
||||
|
||||
left, top, right, bottom = rect
|
||||
center_x = left + (right - left) // 2
|
||||
center_y = top + (bottom - top) // 2
|
||||
nodes.append(
|
||||
{
|
||||
"label": _node_label(node),
|
||||
"text": text,
|
||||
"content_desc": desc,
|
||||
"resource_id": resource_id,
|
||||
"class_name": (node.attrib.get("class") or "").strip(),
|
||||
"package": (node.attrib.get("package") or "").strip(),
|
||||
"bounds": bounds,
|
||||
"center_x": center_x,
|
||||
"center_y": center_y,
|
||||
"clickable": clickable,
|
||||
"focusable": focusable,
|
||||
"scrollable": scrollable,
|
||||
"enabled": enabled,
|
||||
}
|
||||
)
|
||||
|
||||
def sort_key(item: dict) -> tuple[int, int, int]:
|
||||
return (
|
||||
0 if item["clickable"] else 1,
|
||||
0 if item["text"] or item["content_desc"] else 1,
|
||||
len(item["label"]),
|
||||
)
|
||||
|
||||
nodes.sort(key=sort_key)
|
||||
return nodes[: max(1, min(limit, 200))]
|
||||
|
||||
|
||||
async def android_screenshot(serial: str) -> bytes:
|
||||
if not serial:
|
||||
raise ValueError("Missing Android device serial")
|
||||
code, data, err = await _adb_binary(serial, "exec-out", "screencap", "-p", timeout=20.0)
|
||||
if code != 0 or not data:
|
||||
raise RuntimeError(err or "Failed to capture screenshot")
|
||||
return data.replace(b"\r\n", b"\n")
|
||||
|
||||
|
||||
def _normalize_text(text: str) -> str:
|
||||
normalized = text.replace(" ", "%s")
|
||||
normalized = normalized.replace("&", r"\&")
|
||||
normalized = normalized.replace("(", r"\(")
|
||||
normalized = normalized.replace(")", r"\)")
|
||||
return normalized
|
||||
|
||||
|
||||
async def android_action(action: str, serial: str, payload: Optional[dict] = None) -> dict:
|
||||
if not serial:
|
||||
raise ValueError("Missing Android device serial")
|
||||
if not adb_available():
|
||||
raise RuntimeError("adb is not installed")
|
||||
payload = payload or {}
|
||||
|
||||
if action == "key":
|
||||
keycode = str(payload.get("keycode", "")).strip()
|
||||
if not keycode:
|
||||
raise ValueError("Missing keycode")
|
||||
await _shell(serial, f"input keyevent {keycode}")
|
||||
return {"message": f"Sent keycode {keycode}"}
|
||||
|
||||
if action == "text":
|
||||
text = str(payload.get("text", ""))
|
||||
if not text:
|
||||
raise ValueError("Missing text")
|
||||
await _shell(serial, f"input text '{_normalize_text(text)}'")
|
||||
return {"message": "Text sent to device"}
|
||||
|
||||
if action == "open_url":
|
||||
url = str(payload.get("url", "")).strip()
|
||||
if not url:
|
||||
raise ValueError("Missing url")
|
||||
await _shell(
|
||||
serial,
|
||||
f"am start -a android.intent.action.VIEW -d '{quote(url, safe=':/?&=%#.-_')}'",
|
||||
)
|
||||
return {"message": f"Opened URL: {url}"}
|
||||
|
||||
if action == "launch_package":
|
||||
package = str(payload.get("package", "")).strip()
|
||||
if not package:
|
||||
raise ValueError("Missing package")
|
||||
await _shell(serial, f"monkey -p '{package}' -c android.intent.category.LAUNCHER 1")
|
||||
return {"message": f"Launched package: {package}"}
|
||||
|
||||
if action == "shell":
|
||||
command = str(payload.get("command", "")).strip()
|
||||
if not command:
|
||||
raise ValueError("Missing shell command")
|
||||
output = await _shell(serial, command, timeout=30.0)
|
||||
return {"message": "Shell command finished", "output": output}
|
||||
|
||||
if action == "tap":
|
||||
x = int(payload.get("x", 0))
|
||||
y = int(payload.get("y", 0))
|
||||
await _shell(serial, f"input tap {x} {y}")
|
||||
return {"message": f"Tapped at {x}, {y}"}
|
||||
|
||||
if action == "double_tap":
|
||||
x = int(payload.get("x", 0))
|
||||
y = int(payload.get("y", 0))
|
||||
await _shell(serial, f"input tap {x} {y} && sleep 0.08 && input tap {x} {y}")
|
||||
return {"message": f"Double tapped at {x}, {y}"}
|
||||
|
||||
if action == "long_press":
|
||||
x = int(payload.get("x", 0))
|
||||
y = int(payload.get("y", 0))
|
||||
duration = int(payload.get("duration", 700))
|
||||
await _shell(serial, f"input swipe {x} {y} {x} {y} {duration}")
|
||||
return {"message": f"Long pressed at {x}, {y} for {duration}ms"}
|
||||
|
||||
if action == "swipe":
|
||||
direction = str(payload.get("direction", "")).strip().lower()
|
||||
width, height = await _get_display_size(serial)
|
||||
mid_x = max(width // 2, 1)
|
||||
mid_y = max(height // 2, 1)
|
||||
mapping = {
|
||||
"up": (mid_x, int(height * 0.75), mid_x, int(height * 0.28)),
|
||||
"down": (mid_x, int(height * 0.28), mid_x, int(height * 0.75)),
|
||||
"left": (int(width * 0.75), mid_y, int(width * 0.25), mid_y),
|
||||
"right": (int(width * 0.25), mid_y, int(width * 0.75), mid_y),
|
||||
}
|
||||
if direction not in mapping:
|
||||
raise ValueError("Unsupported swipe direction")
|
||||
start_x, start_y, end_x, end_y = mapping[direction]
|
||||
await _shell(serial, f"input swipe {start_x} {start_y} {end_x} {end_y} 240")
|
||||
return {"message": f"Swiped {direction}"}
|
||||
|
||||
if action == "wake":
|
||||
await _shell(serial, "input keyevent KEYCODE_WAKEUP")
|
||||
return {"message": "Wake command sent"}
|
||||
|
||||
if action == "rotate":
|
||||
target = str(payload.get("rotation", "0")).strip()
|
||||
if target not in {"0", "1", "2", "3"}:
|
||||
raise ValueError("Rotation must be one of 0,1,2,3")
|
||||
await _shell(
|
||||
serial,
|
||||
"settings put system accelerometer_rotation 0 && "
|
||||
f"settings put system user_rotation {target}",
|
||||
)
|
||||
return {"message": f"Rotation set to {target}"}
|
||||
|
||||
if action == "force_stop":
|
||||
package = str(payload.get("package", "")).strip()
|
||||
if not package:
|
||||
raise ValueError("Missing package")
|
||||
await _shell(serial, f"am force-stop '{package}'")
|
||||
return {"message": f"Force-stopped: {package}"}
|
||||
|
||||
if action == "clear_app":
|
||||
package = str(payload.get("package", "")).strip()
|
||||
if not package:
|
||||
raise ValueError("Missing package")
|
||||
output = await _shell(serial, f"pm clear '{package}'", timeout=25.0)
|
||||
return {"message": f"Cleared app data: {package}", "output": output}
|
||||
|
||||
raise ValueError(f"Unsupported Android action: {action}")
|
||||
Reference in New Issue
Block a user