purewhiter--mobilegym
wehub-resource-sync
861 行
32 KiB
Python
861 行
32 KiB
Python
"""
|
|
RealDeviceEnv - ADB-based real device environment.
|
|
|
|
This module provides a lightweight ADB interface for controlling real Android devices
|
|
or standard emulators. It supports visual-based agents (VLM) through screenshots and
|
|
basic touch interactions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import shlex
|
|
import subprocess
|
|
from typing import Any, Optional, Tuple, TYPE_CHECKING
|
|
|
|
from bench_env.logger import get_logger
|
|
from bench_env.env.base import Action, ActionType, BaseMobileEnv, Observation, StepResult
|
|
|
|
if TYPE_CHECKING:
|
|
from bench_env.task.base import BaseTask
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# App name -> package name mapping (from Open-AutoGLM)
|
|
# Supports both Chinese names and English aliases
|
|
APP_PACKAGES: dict[str, str] = {
|
|
# ==================== Mobile-Gym Simulator Apps ====================
|
|
# WeChat
|
|
"微信": "com.tencent.mm",
|
|
"WeChat": "com.tencent.mm",
|
|
"wechat": "com.tencent.mm",
|
|
# RedBook (小红书)
|
|
"小红书": "com.xingin.xhs",
|
|
"RedBook": "com.xingin.xhs",
|
|
"redbook": "com.xingin.xhs",
|
|
"xiaohongshu": "com.xingin.xhs",
|
|
# Alipay (支付宝)
|
|
"支付宝": "com.eg.android.AlipayGphone",
|
|
"Alipay": "com.eg.android.AlipayGphone",
|
|
"alipay": "com.eg.android.AlipayGphone",
|
|
# Bilibili (B站)
|
|
"哔哩哔哩": "tv.danmaku.bili",
|
|
"bilibili": "tv.danmaku.bili",
|
|
"Bilibili": "tv.danmaku.bili",
|
|
"B站": "tv.danmaku.bili",
|
|
# Map (地图) - Google Maps
|
|
"Map": "com.google.android.apps.maps",
|
|
"map": "com.google.android.apps.maps",
|
|
"Google Maps": "com.google.android.apps.maps",
|
|
"googlemaps": "com.google.android.apps.maps",
|
|
# 高德地图
|
|
"高德地图": "com.autonavi.minimap",
|
|
"amap": "com.autonavi.minimap",
|
|
"gaode": "com.autonavi.minimap",
|
|
# QQ Music (QQ音乐)
|
|
"QQ音乐": "com.tencent.qqmusic",
|
|
"QQMusic": "com.tencent.qqmusic",
|
|
"qqmusic": "com.tencent.qqmusic",
|
|
# Spotify
|
|
"Spotify": "com.spotify.music",
|
|
"spotify": "com.spotify.music",
|
|
# Tencent Meeting (腾讯会议)
|
|
"腾讯会议": "com.tencent.wemeet.app",
|
|
"TencentMeeting": "com.tencent.wemeet.app",
|
|
"tencent_meeting": "com.tencent.wemeet.app",
|
|
"tencentmeeting": "com.tencent.wemeet.app",
|
|
"wemeet": "com.tencent.wemeet.app",
|
|
# WeChat Reading (微信读书)
|
|
"微信读书": "com.tencent.weread",
|
|
"WechatReading": "com.tencent.weread",
|
|
"wechat_reading": "com.tencent.weread",
|
|
"wechatreading": "com.tencent.weread",
|
|
"weread": "com.tencent.weread",
|
|
# X (Twitter)
|
|
"X": "com.twitter.android",
|
|
"x": "com.twitter.android",
|
|
"Twitter": "com.twitter.android",
|
|
"twitter": "com.twitter.android",
|
|
# Weather (天气)
|
|
"天气": "com.miui.weather2",
|
|
"Weather": "com.miui.weather2",
|
|
"weather": "com.miui.weather2",
|
|
# Notes (备忘录)
|
|
"备忘录": "com.miui.notes",
|
|
"Notes": "com.miui.notes",
|
|
"notes": "com.miui.notes",
|
|
# Gallery (相册)
|
|
"相册": "com.miui.gallery",
|
|
"Gallery": "com.miui.gallery",
|
|
"gallery": "com.miui.gallery",
|
|
# FileManager (文件管理)
|
|
"文件管理": "com.android.fileexplorer",
|
|
"FileManager": "com.android.fileexplorer",
|
|
"file_manager": "com.android.fileexplorer",
|
|
"filemanager": "com.android.fileexplorer",
|
|
"Files": "com.android.fileexplorer",
|
|
# Calculator (计算器)
|
|
"计算器": "com.miui.calculator",
|
|
"Calculator": "com.miui.calculator",
|
|
"calculator": "com.miui.calculator",
|
|
# Browser (浏览器)
|
|
"浏览器": "com.android.browser",
|
|
"Browser": "com.android.browser",
|
|
"browser": "com.android.browser",
|
|
# Calendar (日历)
|
|
"日历": "com.android.calendar",
|
|
"Calendar": "com.android.calendar",
|
|
"calendar": "com.android.calendar",
|
|
# SMS (短信)
|
|
"短信": "com.android.mms",
|
|
"Messages": "com.android.mms",
|
|
"sms": "com.android.mms",
|
|
|
|
# ==================== Other Common Apps ====================
|
|
# Social & Messaging
|
|
"QQ": "com.tencent.mobileqq",
|
|
"微博": "com.sina.weibo",
|
|
"weibo": "com.sina.weibo",
|
|
"Telegram": "org.telegram.messenger",
|
|
"WhatsApp": "com.whatsapp",
|
|
# E-commerce
|
|
"淘宝": "com.taobao.taobao",
|
|
"taobao": "com.taobao.taobao",
|
|
"京东": "com.jingdong.app.mall",
|
|
"jd": "com.jingdong.app.mall",
|
|
"拼多多": "com.xunmeng.pinduoduo",
|
|
"pinduoduo": "com.xunmeng.pinduoduo",
|
|
"Temu": "com.einnovation.temu",
|
|
"eBay": "com.ebay.mobile",
|
|
"ebay": "com.ebay.mobile",
|
|
# Lifestyle & Social
|
|
"豆瓣": "com.douban.frodo",
|
|
"douban": "com.douban.frodo",
|
|
"知乎": "com.zhihu.android",
|
|
"zhihu": "com.zhihu.android",
|
|
"Reddit": "com.reddit.frontpage",
|
|
"Quora": "com.quora.android",
|
|
# Maps & Navigation
|
|
"百度地图": "com.baidu.BaiduMap",
|
|
"baidumap": "com.baidu.BaiduMap",
|
|
"Google Maps": "com.google.android.apps.maps",
|
|
# Food & Services
|
|
"美团": "com.sankuai.meituan",
|
|
"meituan": "com.sankuai.meituan",
|
|
"大众点评": "com.dianping.v1",
|
|
"饿了么": "me.ele",
|
|
"eleme": "me.ele",
|
|
"肯德基": "com.yek.android.kfc.activitys",
|
|
"McDonald": "com.mcdonalds.app",
|
|
# Travel
|
|
"携程": "ctrip.android.view",
|
|
"ctrip": "ctrip.android.view",
|
|
"铁路12306": "com.MobileTicket",
|
|
"12306": "com.MobileTicket",
|
|
"railway12306": "com.MobileTicket",
|
|
"去哪儿": "com.Qunar",
|
|
"滴滴出行": "com.sdu.didi.psnger",
|
|
"didi": "com.sdu.didi.psnger",
|
|
"Booking": "com.booking",
|
|
"Expedia": "com.expedia.bookings",
|
|
# Video & Entertainment
|
|
"抖音": "com.ss.android.ugc.aweme",
|
|
"douyin": "com.ss.android.ugc.aweme",
|
|
"TikTok": "com.zhiliaoapp.musically",
|
|
"tiktok": "com.zhiliaoapp.musically",
|
|
"快手": "com.smile.gifmaker",
|
|
"kuaishou": "com.smile.gifmaker",
|
|
"腾讯视频": "com.tencent.qqlive",
|
|
"爱奇艺": "com.qiyi.video",
|
|
"iqiyi": "com.qiyi.video",
|
|
"优酷视频": "com.youku.phone",
|
|
"youku": "com.youku.phone",
|
|
# Music & Audio
|
|
"网易云音乐": "com.netease.cloudmusic",
|
|
"netease_music": "com.netease.cloudmusic",
|
|
"喜马拉雅": "com.ximalaya.ting.android",
|
|
"ximalaya": "com.ximalaya.ting.android",
|
|
# Productivity
|
|
"飞书": "com.ss.android.lark",
|
|
"feishu": "com.ss.android.lark",
|
|
"Gmail": "com.google.android.gm",
|
|
# Browser
|
|
"Chrome": "com.android.chrome",
|
|
"chrome": "com.android.chrome",
|
|
# System
|
|
"Settings": "com.android.settings",
|
|
"settings": "com.android.settings",
|
|
"设置": "com.android.settings",
|
|
"Clock": "com.android.deskclock",
|
|
"Contacts": "com.android.contacts",
|
|
"Google Play Store": "com.android.vending",
|
|
# AI & Tools
|
|
"豆包": "com.larus.nova",
|
|
"doubao": "com.larus.nova",
|
|
"Duolingo": "com.duolingo",
|
|
# Reading
|
|
"番茄小说": "com.dragon.read",
|
|
# News
|
|
"腾讯新闻": "com.tencent.news",
|
|
"今日头条": "com.ss.android.article.news",
|
|
"toutiao": "com.ss.android.article.news",
|
|
}
|
|
|
|
|
|
def _package_to_app_name(package: str) -> Optional[str]:
|
|
"""Convert package name to app display name."""
|
|
for name, pkg in APP_PACKAGES.items():
|
|
if pkg == package or package.startswith(pkg):
|
|
return name
|
|
return None
|
|
|
|
|
|
class ActionHandler:
|
|
"""Base class for action handlers (Real Device)."""
|
|
def __init__(self, env: "RealDeviceEnv"):
|
|
self.env = env
|
|
|
|
async def execute(self, action: Action) -> None:
|
|
raise NotImplementedError
|
|
|
|
|
|
class ClickHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
x, y = self.env._parse_point(action.data.get("point"))
|
|
await self.env._tap(x, y)
|
|
|
|
|
|
class TypeHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
value = str(action.data.get("value", ""))
|
|
point = action.data.get("point")
|
|
clear = bool(action.data.get("clear", False))
|
|
if point:
|
|
x, y = self.env._parse_point(point)
|
|
await self.env._tap(x, y)
|
|
if clear:
|
|
await self.env._clear_input()
|
|
await self.env._type_text(value)
|
|
|
|
|
|
class DoubleTapHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
x, y = self.env._parse_point(action.data.get("point"))
|
|
await self.env._tap(x, y)
|
|
await asyncio.sleep(0.08)
|
|
await self.env._tap(x, y)
|
|
|
|
|
|
class LongPressHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
x, y = self.env._parse_point(action.data.get("point"))
|
|
duration = int(action.data.get("duration", 800))
|
|
await self.env._long_press(x, y, duration_ms=duration)
|
|
|
|
|
|
class SwipeHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
x1, y1 = self.env._parse_point(action.data.get("point1"))
|
|
x2, y2 = self.env._parse_point(action.data.get("point2"))
|
|
duration = int(action.data.get("duration", 300))
|
|
await self.env._swipe(x1, y1, x2, y2, duration_ms=duration)
|
|
|
|
|
|
class BackHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
await self.env._press_key("KEYCODE_BACK")
|
|
|
|
|
|
class HomeHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
await self.env._press_key("KEYCODE_HOME")
|
|
|
|
|
|
class RecentHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
await self.env._press_key("KEYCODE_APP_SWITCH")
|
|
|
|
|
|
class EnterHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
await self.env._press_key("KEYCODE_ENTER")
|
|
|
|
|
|
class DragHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
x1, y1 = self.env._parse_point(action.data.get("point1"))
|
|
x2, y2 = self.env._parse_point(action.data.get("point2"))
|
|
duration = int(action.data.get("duration", 2000))
|
|
await self.env._swipe(x1, y1, x2, y2, duration_ms=duration)
|
|
|
|
|
|
class WaitHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
secs = float(action.data.get("value", 1.0))
|
|
await asyncio.sleep(max(0.0, secs))
|
|
|
|
|
|
class AwakeHandler(ActionHandler):
|
|
async def execute(self, action: Action) -> None:
|
|
app_name = str(action.data.get("value", "")).strip()
|
|
if app_name:
|
|
await self.env._launch_app(app_name)
|
|
|
|
|
|
class RealDeviceEnv(BaseMobileEnv):
|
|
"""
|
|
ADB-based real Android device environment.
|
|
|
|
This is a lightweight implementation for real device support.
|
|
|
|
Limitations:
|
|
- Text Input: Uses YADB for Chinese/Unicode input (auto-installed on first run).
|
|
Falls back to `adb shell input text` for ASCII-only if YADB fails.
|
|
- Observation: Visual only (screenshot). No XML/ViewHierarchy dumping.
|
|
- App Launch: Basic `monkey` launch.
|
|
- State injection: Not supported (no set_state, no answer_sheet app).
|
|
|
|
Example usage (once implemented):
|
|
env = RealDeviceEnv(
|
|
device_serial="emulator-5554", # or real device serial
|
|
adb_path="~/Android/Sdk/platform-tools/adb",
|
|
)
|
|
env.start()
|
|
obs = env.reset(task)
|
|
...
|
|
"""
|
|
|
|
# No JSON state mutation on a real device — grounded-mode answer_sheet
|
|
# injection and similar state-seeding hooks must be skipped.
|
|
supports_state_injection: bool = False
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
device_serial: Optional[str] = None,
|
|
adb_path: str = "adb",
|
|
physical_size: Tuple[int, int] = (1080, 2400),
|
|
coord_space: str = "norm_0_1000",
|
|
delay_after_action: float = 1.0,
|
|
):
|
|
"""
|
|
Initialize RealDeviceEnv.
|
|
|
|
Args:
|
|
device_serial: Device serial number (from `adb devices`)
|
|
adb_path: Path to adb executable
|
|
physical_size: Device screen size (width, height)
|
|
coord_space: Coordinate space ("norm_0_1000", "norm_0_1", or "physical")
|
|
delay_after_action: Delay after each action in seconds
|
|
"""
|
|
self.device_serial = device_serial
|
|
self.adb_path = adb_path
|
|
self.physical_width, self.physical_height = physical_size
|
|
self.coord_space = coord_space
|
|
self.delay_after_action = delay_after_action
|
|
|
|
self._task: Optional[BaseTask] = None
|
|
self._step_count = 0
|
|
self._done = False
|
|
self._agent_message: Optional[str] = None
|
|
self._agent_answer: Optional[str] = None
|
|
# package -> "package/MainActivity", resolved lazily via pm resolve-activity.
|
|
# Empty string = resolution failed previously (skip and fall back to monkey).
|
|
self._main_activity_cache: dict[str, str] = {}
|
|
|
|
from bench_env.env.stopwatch import StopWatch
|
|
self.stopwatch = StopWatch()
|
|
|
|
# Initialize handlers
|
|
self._handlers: dict[ActionType, ActionHandler] = {
|
|
ActionType.CLICK: ClickHandler(self),
|
|
ActionType.DOUBLE_TAP: DoubleTapHandler(self),
|
|
ActionType.LONG_PRESS: LongPressHandler(self),
|
|
ActionType.TYPE: TypeHandler(self),
|
|
ActionType.SWIPE: SwipeHandler(self),
|
|
ActionType.DRAG: DragHandler(self),
|
|
ActionType.BACK: BackHandler(self),
|
|
ActionType.HOME: HomeHandler(self),
|
|
ActionType.RECENT: RecentHandler(self),
|
|
ActionType.ENTER: EnterHandler(self),
|
|
ActionType.WAIT: WaitHandler(self),
|
|
ActionType.AWAKE: AwakeHandler(self),
|
|
}
|
|
|
|
async def start(self) -> "RealDeviceEnv":
|
|
"""Connect to device and initialize required tools."""
|
|
# Check connectivity
|
|
out = await self._adb("shell", "getprop", "ro.product.model")
|
|
if not out:
|
|
raise RuntimeError(f"Failed to connect to device {self.device_serial}")
|
|
logger.info(f"Connected to {out.strip()}")
|
|
|
|
# Initialize YADB for Chinese text input
|
|
await self._init_yadb()
|
|
|
|
return self
|
|
|
|
async def _init_yadb(self) -> None:
|
|
"""
|
|
Initialize YADB (Yet Another Debug Bridge) for Chinese text input.
|
|
|
|
Checks if yadb is installed on device, if not, pushes it from local project.
|
|
"""
|
|
logger.debug("Checking YADB installation...")
|
|
|
|
# Check if yadb exists and has correct md5
|
|
YADB_MD5 = "29a0cd3b3adea92350dd5a25594593df"
|
|
result = await self._adb("shell", "md5sum", "/data/local/tmp/yadb")
|
|
|
|
if YADB_MD5 in result:
|
|
logger.info("YADB is already installed on the device")
|
|
return
|
|
|
|
logger.info("YADB not found or outdated on device, need to install...")
|
|
|
|
# Find yadb file in project
|
|
import os
|
|
bench_env_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
project_root = os.path.dirname(bench_env_dir)
|
|
yadb_paths = [
|
|
os.path.join(bench_env_dir, "yadb"), # bench_env/yadb (preferred)
|
|
os.path.join(project_root, "gelab-zero", "yadb"),
|
|
os.path.join(project_root, "yadb"),
|
|
]
|
|
|
|
yadb_path = None
|
|
for path in yadb_paths:
|
|
if os.path.exists(path):
|
|
yadb_path = path
|
|
logger.debug(f"Found YADB at: {yadb_path}")
|
|
break
|
|
|
|
if not yadb_path:
|
|
logger.warning("YADB not found in any of the expected paths:")
|
|
for path in yadb_paths:
|
|
logger.warning(f" - {path}")
|
|
logger.warning("Chinese text input may not work. "
|
|
"Please manually push yadb to /data/local/tmp/yadb")
|
|
return
|
|
|
|
# Push yadb to device
|
|
logger.info(f"Pushing YADB from {yadb_path} to device...")
|
|
push_result = await self._adb_push(yadb_path, "/data/local/tmp/yadb")
|
|
if not push_result:
|
|
logger.error("Failed to push YADB to device")
|
|
return
|
|
logger.info(f"YADB pushed: {push_result.strip()}")
|
|
|
|
# Set executable permission
|
|
logger.debug("Setting YADB executable permission...")
|
|
chmod_result = await self._adb("shell", "chmod", "+x", "/data/local/tmp/yadb")
|
|
|
|
# Verify installation
|
|
verify_result = await self._adb("shell", "md5sum", "/data/local/tmp/yadb")
|
|
if YADB_MD5 in verify_result:
|
|
logger.info("YADB installed and verified successfully")
|
|
else:
|
|
logger.warning(f"YADB installed but MD5 mismatch. Expected: {YADB_MD5}, Got: {verify_result.strip()}")
|
|
|
|
async def _adb_push(self, local_path: str, remote_path: str) -> str:
|
|
"""Push a file to the device."""
|
|
cmd = [self.adb_path]
|
|
if self.device_serial:
|
|
cmd.extend(["-s", self.device_serial])
|
|
cmd.extend(["push", local_path, remote_path])
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
if proc.returncode != 0:
|
|
logger.error(f"[ADB Push Error] {stderr.decode()}")
|
|
return ""
|
|
# Modern `adb push` writes progress / "1 file pushed" to stderr and
|
|
# leaves stdout empty; returning stdout here made callers that check
|
|
# `if not push_result:` treat a successful push as a failure. Return
|
|
# a non-empty sentinel on success and log the stderr diagnostics.
|
|
stderr_text = stderr.decode().strip()
|
|
if stderr_text:
|
|
logger.debug(f"[ADB Push] {stderr_text}")
|
|
return "ok"
|
|
|
|
async def close(self) -> None:
|
|
"""Disconnect from device."""
|
|
pass
|
|
|
|
async def reset(self, app_ids: list[str] | None = None) -> None:
|
|
"""Reset for a new episode.
|
|
|
|
`app_ids` is accepted for signature parity with MobileGymEnv but
|
|
ignored: on real devices we always force-stop every package listed in
|
|
APP_PACKAGES so the next episode starts from a fully clean background,
|
|
regardless of which apps the task declares.
|
|
"""
|
|
self._step_count = 0
|
|
self._done = False
|
|
self._agent_message = None
|
|
self._agent_answer = None
|
|
|
|
await self._press_key("KEYCODE_HOME")
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Force-stop every package listed in APP_PACKAGES so each episode starts from
|
|
# a clean background. We only touch apps we explicitly benchmark — system
|
|
# components (Launcher, IME, SystemUI) are left untouched.
|
|
for pkg in {p for p in APP_PACKAGES.values() if "." in p}:
|
|
await self._adb("shell", "am", "force-stop", pkg)
|
|
await asyncio.sleep(0.3)
|
|
|
|
# Second HOME press: most launchers scroll back to the first
|
|
# page on a second HOME while already on the launcher.
|
|
await self._press_key("KEYCODE_HOME")
|
|
await asyncio.sleep(0.1)
|
|
|
|
async def step(self, action: Action) -> StepResult:
|
|
self._step_count += 1
|
|
action_type = action.action_type
|
|
|
|
if action_type == ActionType.ANSWER:
|
|
value = str(action.data.get("value", ""))
|
|
self._agent_answer = value
|
|
logger.info(f"ANSWER: {value}")
|
|
return StepResult(
|
|
await self._get_observation(),
|
|
False,
|
|
{"action_type": ActionType.ANSWER, "answer": value},
|
|
)
|
|
|
|
if action_type == ActionType.COMPLETE:
|
|
self._done = True
|
|
self._agent_message = action.data.get("return", "")
|
|
return StepResult(await self._get_observation(), True,
|
|
{"stop_reason": ActionType.COMPLETE, "message": self._agent_message})
|
|
|
|
if action_type == ActionType.ABORT:
|
|
self._done = True
|
|
self._agent_message = action.data.get("value", "")
|
|
return StepResult(await self._get_observation(), True,
|
|
{"stop_reason": ActionType.ABORT, "message": self._agent_message})
|
|
|
|
handler = self._handlers.get(action_type)
|
|
if handler:
|
|
await handler.execute(action)
|
|
else:
|
|
logger.warning(f"Unknown action: {action_type}")
|
|
|
|
await asyncio.sleep(self.delay_after_action)
|
|
return StepResult(await self._get_observation(), False, {})
|
|
|
|
async def get_observation(self) -> Observation:
|
|
"""Public observation API (avoid calling private _get_observation)."""
|
|
return await self._get_observation()
|
|
|
|
async def get_state(self, required_apps: list[str] | None = None) -> dict[str, Any]:
|
|
"""
|
|
Get current environment state.
|
|
|
|
NOTE: This lightweight real-device env is visual-only; structured app/os
|
|
state is not available yet, so we return an empty dict.
|
|
"""
|
|
return {}
|
|
|
|
async def open_app(
|
|
self,
|
|
app_name: str,
|
|
timeout_ms: int = 8000,
|
|
wait_stable: bool = False,
|
|
) -> None:
|
|
"""
|
|
Open an application.
|
|
|
|
For real devices, `app_name` is treated as a package name (or "package/activity").
|
|
|
|
Args:
|
|
app_name: App ID or package name
|
|
timeout_ms: Timeout (ignored for ADB)
|
|
wait_stable: Wait for app to stabilize (for real devices, we just add a delay)
|
|
"""
|
|
# Ignore timeout_ms for now; ADB calls are best-effort.
|
|
if not app_name:
|
|
return
|
|
if "/" in app_name:
|
|
pkg, act = app_name.split("/", 1)
|
|
await self._launch_app(pkg, act)
|
|
else:
|
|
await self._launch_app(app_name)
|
|
|
|
# For real devices, wait a bit for app to stabilize
|
|
if wait_stable:
|
|
await asyncio.sleep(2.0)
|
|
|
|
async def go_home(self) -> None:
|
|
"""Press HOME key to return to launcher."""
|
|
await self._press_key("KEYCODE_HOME")
|
|
await asyncio.sleep(1.0)
|
|
|
|
@property
|
|
def agent_message(self) -> Optional[str]:
|
|
return self._agent_message
|
|
|
|
@property
|
|
def agent_answer(self) -> Optional[str]:
|
|
return self._agent_answer
|
|
|
|
def get_device_size(self) -> Tuple[int, int]:
|
|
return self.physical_width, self.physical_height
|
|
|
|
# ==================== ADB Command Helpers ====================
|
|
|
|
async def _adb(self, *args: str) -> str:
|
|
"""Execute ADB command asynchronously."""
|
|
cmd = [self.adb_path]
|
|
if self.device_serial:
|
|
cmd.extend(["-s", self.device_serial])
|
|
cmd.extend(args)
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
if proc.returncode != 0:
|
|
logger.error(f"[ADB Error] cmd={cmd} err={stderr.decode()}")
|
|
return ""
|
|
return stdout.decode()
|
|
|
|
async def _get_observation(self) -> Observation:
|
|
png_bytes = await self._screenshot()
|
|
|
|
# Get current app info
|
|
current_app = await self._get_current_app()
|
|
|
|
# NOTE: XML dump / ViewHierarchy is not implemented in this lightweight version.
|
|
# This environment currently supports VLM-based agents (Visual Only).
|
|
return Observation(
|
|
screenshot_bytes=png_bytes,
|
|
route={"app": current_app} if current_app else {},
|
|
state={},
|
|
step_idx=self._step_count,
|
|
)
|
|
|
|
async def _screenshot(self) -> bytes:
|
|
"""Capture screenshot via ADB."""
|
|
# Use exec-out screencap -p for raw PNG stream (faster than shell)
|
|
cmd = [self.adb_path]
|
|
if self.device_serial:
|
|
cmd.extend(["-s", self.device_serial])
|
|
cmd.extend(["exec-out", "screencap", "-p"])
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
return stdout
|
|
|
|
def _parse_point(self, point: Any) -> Tuple[int, int]:
|
|
if point is None:
|
|
return self.physical_width // 2, self.physical_height // 2
|
|
if isinstance(point, (list, tuple)) and len(point) >= 2:
|
|
x, y = float(point[0]), float(point[1])
|
|
else:
|
|
return self.physical_width // 2, self.physical_height // 2
|
|
|
|
if self.coord_space == "norm_0_1000":
|
|
x = x / 1000.0 * self.physical_width
|
|
y = y / 1000.0 * self.physical_height
|
|
elif self.coord_space == "norm_0_1":
|
|
x = x * self.physical_width
|
|
y = y * self.physical_height
|
|
|
|
return int(x), int(y)
|
|
|
|
async def _tap(self, x: int, y: int) -> None:
|
|
await self._adb("shell", "input", "tap", str(x), str(y))
|
|
|
|
async def _long_press(self, x: int, y: int, duration_ms: int = 800) -> None:
|
|
await self._adb("shell", "input", "swipe", str(x), str(y), str(x), str(y), str(duration_ms))
|
|
|
|
async def _clear_input(self) -> None:
|
|
"""Clear current input field via Ctrl+A then Delete."""
|
|
await self._adb("shell", "input", "keyevent", "KEYCODE_MOVE_HOME")
|
|
await self._adb("shell", "input", "keyevent", "--longpress", "KEYCODE_SHIFT_LEFT", "KEYCODE_MOVE_END")
|
|
await self._adb("shell", "input", "keyevent", "KEYCODE_DEL")
|
|
|
|
async def _swipe(self, x1: int, y1: int, x2: int, y2: int, duration_ms: int = 300) -> None:
|
|
await self._adb("shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(duration_ms))
|
|
|
|
async def _type_text(self, text: str) -> None:
|
|
"""
|
|
Type text into the currently focused input field.
|
|
|
|
Uses YADB for all text input (supports Chinese and Unicode).
|
|
Falls back to basic `adb shell input text` for ASCII-only if YADB fails.
|
|
"""
|
|
logger.debug(f"Typing text: {text[:30]}{'...' if len(text) > 30 else ''}")
|
|
|
|
# Always try YADB first (supports both ASCII and Unicode)
|
|
success = await self._type_via_yadb(text)
|
|
|
|
if success:
|
|
logger.debug("Text input via YADB successful")
|
|
else:
|
|
# Fallback to basic input for ASCII text
|
|
is_ascii = all(ord(c) < 128 for c in text)
|
|
if is_ascii:
|
|
logger.debug("YADB failed, falling back to basic adb input text")
|
|
escaped = shlex.quote(text)
|
|
await self._adb("shell", "input", "text", escaped)
|
|
else:
|
|
logger.warning(f"Failed to input non-ASCII text (YADB not available): {text[:20]}...")
|
|
|
|
async def _type_via_yadb(self, text: str) -> bool:
|
|
"""
|
|
Type text using YADB (Yet Another Debug Bridge).
|
|
|
|
Supports both ASCII and Chinese/Unicode text.
|
|
Requires yadb to be pushed to /data/local/tmp/yadb on the device.
|
|
"""
|
|
try:
|
|
# adb shell 会在设备端再次解析命令；这里必须保留空格为同一个参数。
|
|
shell_text = shlex.quote(text)
|
|
result = await self._adb(
|
|
"shell",
|
|
"app_process",
|
|
"-Djava.class.path=/data/local/tmp/yadb",
|
|
"/data/local/tmp",
|
|
"com.ysbing.yadb.Main",
|
|
"-keyboard",
|
|
shell_text
|
|
)
|
|
# YADB returns empty on success, check for errors
|
|
if "error" in result.lower() or "exception" in result.lower():
|
|
logger.debug(f"YADB returned error: {result}")
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
logger.debug(f"YADB input exception: {e}")
|
|
return False
|
|
|
|
async def _press_key(self, keycode: str) -> None:
|
|
await self._adb("shell", "input", "keyevent", keycode)
|
|
|
|
async def _launch_app(self, app_or_package: str, activity: Optional[str] = None) -> None:
|
|
"""
|
|
Launch an app by name or package.
|
|
|
|
Args:
|
|
app_or_package: App name (e.g., "微信", "wechat") or package name (e.g., "com.tencent.mm")
|
|
activity: Optional activity name
|
|
"""
|
|
package = self._resolve_package(app_or_package)
|
|
|
|
if activity:
|
|
# Explicit activity requested — honor it verbatim.
|
|
await self._adb("shell", "am", "start", "-n", f"{package}/{activity}")
|
|
return
|
|
|
|
# Prefer `am start --activity-clear-task` so the previous Task/Recents entry
|
|
# is wiped and the app opens from its MainActivity — avoids state carry-over
|
|
# between episodes (e.g. resuming the last opened book in WeChat Reading).
|
|
#
|
|
# Pass `-a MAIN -c LAUNCHER` explicitly so entry activities that inspect the
|
|
# incoming intent (e.g. system Messaging's MmsTabActivity) treat this as a
|
|
# launcher click and land on the default home screen, instead of
|
|
# dispatching to a secondary flow (like "compose new message").
|
|
component = await self._resolve_main_activity(package)
|
|
if component:
|
|
await self._adb(
|
|
"shell", "am", "start",
|
|
"-a", "android.intent.action.MAIN",
|
|
"-c", "android.intent.category.LAUNCHER",
|
|
"--activity-clear-task",
|
|
"-n", component,
|
|
)
|
|
return
|
|
|
|
# Fallback: standard LAUNCHER intent via monkey (does NOT clear the task).
|
|
await self._adb(
|
|
"shell", "monkey", "-p", package,
|
|
"-c", "android.intent.category.LAUNCHER", "1",
|
|
)
|
|
|
|
async def _resolve_main_activity(self, package: str) -> str:
|
|
"""Resolve `package/MainActivity` via `cmd package resolve-activity`.
|
|
|
|
Returns the component string usable with `am start -n`, or empty string
|
|
if resolution fails (caller should fall back to monkey).
|
|
"""
|
|
if package in self._main_activity_cache:
|
|
return self._main_activity_cache[package]
|
|
|
|
output = await self._adb(
|
|
"shell", "cmd", "package", "resolve-activity",
|
|
"--brief", "-c", "android.intent.category.LAUNCHER", package,
|
|
)
|
|
component = ""
|
|
for line in output.splitlines():
|
|
line = line.strip()
|
|
# Skip the priority/preferredOrder header line — the component line
|
|
# contains a slash and no '=' assignments.
|
|
if "/" in line and "=" not in line and line.startswith(package + "/"):
|
|
component = line
|
|
break
|
|
self._main_activity_cache[package] = component
|
|
return component
|
|
|
|
def _resolve_package(self, app_or_package: str) -> str:
|
|
"""
|
|
Resolve app name to package name.
|
|
|
|
Checks APP_PACKAGES mapping first, otherwise returns the input as-is
|
|
(assuming it's already a package name).
|
|
"""
|
|
# Direct match in APP_PACKAGES
|
|
if app_or_package in APP_PACKAGES:
|
|
return APP_PACKAGES[app_or_package]
|
|
|
|
# Case-insensitive match
|
|
lower_name = app_or_package.lower()
|
|
for name, pkg in APP_PACKAGES.items():
|
|
if name.lower() == lower_name:
|
|
return pkg
|
|
|
|
# Check if it's already a package name (contains dots)
|
|
if "." in app_or_package:
|
|
return app_or_package
|
|
|
|
logger.warning(f"Unknown app '{app_or_package}', using as package name directly")
|
|
return app_or_package
|
|
|
|
async def _get_current_app(self) -> Optional[str]:
|
|
"""
|
|
Get the currently focused app name via ADB.
|
|
|
|
Uses `dumpsys window` to find mCurrentFocus/mFocusedApp,
|
|
then matches package name against APP_PACKAGES mapping.
|
|
|
|
Returns:
|
|
App display name (e.g., "微信") or None if not recognized.
|
|
"""
|
|
output = await self._adb("shell", "dumpsys", "window")
|
|
if not output:
|
|
return None
|
|
|
|
# Parse window focus info
|
|
for line in output.split("\n"):
|
|
if "mCurrentFocus" in line or "mFocusedApp" in line:
|
|
# Extract package name from line like:
|
|
# mCurrentFocus=Window{...com.tencent.mm/com.tencent.mm.ui.LauncherUI...}
|
|
for app_name, package in APP_PACKAGES.items():
|
|
if package in line:
|
|
return app_name
|
|
|
|
return None
|