Android Uiautomator2 Python Wrapper源码解析:核心类与方法实现原理
Android Uiautomator2 Python Wrapper源码解析:核心类与方法实现原理
引言:Uiautomator2框架架构概览
Android Uiautomator2 Python Wrapper(简称uiautomator2)是一个基于Python的UI自动化测试框架,它封装了Android原生的Uiautomator2测试框架,提供了简洁易用的API接口。该框架通过ADB(Android Debug Bridge)与Android设备通信,实现对设备的UI元素识别、操作和事件触发。
核心架构组件
uiautomator2框架主要由以下核心组件构成:
- 设备通信层:负责与Android设备建立连接,通过ADB进行通信
- UI元素识别层:解析设备界面结构,提供元素定位功能
- 操作执行层:执行各种UI操作,如点击、滑动、输入等
- 应用管理层:负责应用的启动、停止、安装等生命周期管理
核心类解析
1. Device类:设备控制中心
Device类位于uiautomator2/__init__.py,是整个框架的入口点,封装了设备的所有操作。它继承自_BaseClient和_Device类,提供了设备控制、应用管理、UI操作等核心功能。
核心方法实现原理
设备连接与初始化
def __init__(self, serial: Union[str, adbutils.AdbDevice] = None):
if isinstance(serial, adbutils.AdbDevice):
self._serial = serial.serial
self._dev = serial
else:
self._serial = serial
self._dev = self._wait_for_device()
self._debug = False
BasicUiautomatorServer.__init__(self, self._dev)
Device类的初始化过程主要完成以下工作:
- 处理设备序列号或已存在的ADB设备连接
- 等待设备在线(通过
_wait_for_device方法) - 初始化Uiautomator服务器连接
UI元素定位
Device类通过__call__方法创建UiObject实例,实现UI元素定位:
def __call__(self, **kwargs) -> 'UiObject':
return UiObject(self, Selector(**kwargs))
屏幕截图实现
def screenshot(self, filename: Optional[str] = None, format="pillow", display_id: Optional[int] = None):
if display_id is None:
base64_data = self.jsonrpc.takeScreenshot(1, 80)
if base64_data:
jpg_raw = base64.b64decode(base64_data)
pil_img = Image.open(io.BytesIO(jpg_raw))
else:
pil_img = self._dev.screenshot(display_id=display_id)
else:
pil_img = self._dev.screenshot(display_id=display_id)
if filename:
pil_img.save(filename)
return
return image_convert(pil_img, format)
截图功能实现流程:
- 尝试通过JSONRPC调用获取截图(base64编码)
- 如果失败,使用ADB直接获取截图
- 根据需要保存文件或返回图像对象
2. XPathEntry类:高级元素定位与操作
XPathEntry类位于uiautomator2/xpath.py,提供基于XPath的高级UI元素定位和操作功能。它支持复杂的元素查找和交互逻辑。
核心功能
XPath表达式解析
XPathEntry支持多种简化的XPath表达式格式,通过strict_xpath方法转换为标准XPath:
def strict_xpath(xpath: str) -> str:
if xpath.startswith("@"):
xpath = "//*[@resource-id={!r}]".format(xpath[1:])
elif xpath.startswith("^"):
xpath = "//*[re:match(@text, {0}) or re:match(@content-desc, {0}) or re:match(@resource-id, {0})]".format(
string_quote(xpath)
)
elif xpath.startswith("%") and xpath.endswith("%"):
xpath = "//*[contains(@text, {0}) or contains(@content-desc, {0})]".format(
string_quote(xpath[1:-1])
)
# 其他格式处理...
if not is_xpath_syntax_ok(xpath):
raise XPathError("Invalid xpath", orig_xpath)
return xpath
元素等待机制
def wait(self, timeout=None) -> bool:
deadline = time.time() + (timeout or self._global_timeout)
while True:
if self.exists:
return True
if time.time() > deadline:
return False
time.sleep(0.2)
等待机制通过循环检查元素是否存在,直到超时或元素出现,这是UI自动化中处理异步加载的关键技术。
滑动查找实现
def scroll_to(
self,
xpath: str,
direction: Union[Direction, str] = Direction.FORWARD,
max_swipes=10,
) -> Union["XMLElement", None]:
if direction == Direction.FORWARD:
direction = Direction.UP
elif direction == Direction.BACKWARD:
direction = Direction.DOWN
target = self(xpath)
for i in range(max_swipes):
if target.exists:
self._d.swipe_ext(direction, 0.1) # 防止元素停留在边缘
return target.get_last_match()
self._d.swipe_ext(direction, 0.5)
return False
scroll_to方法实现了在指定方向上滑动屏幕,直到找到目标元素或达到最大滑动次数。
3. UiObject与Selector类:UI元素操作基础
UiObject和Selector类位于uiautomator2/_selector.py,提供了基础的UI元素选择和操作功能。
Selector类:元素选择器
Selector类用于构建UI元素选择条件:
class Selector(dict):
__fields = {
"text": (0x01, None),
"textContains": (0x02, None),
"textMatches": (0x04, None),
"textStartsWith": (0x08, None),
"className": (0x10, None),
# 其他选择条件...
}
def __init__(self, **kwargs):
super(Selector, self).__setitem__(self.__mask, 0)
super(Selector, self).__setitem__(self.__childOrSibling, [])
super(Selector, self).__setitem__(self.__childOrSiblingSelector, [])
for k in kwargs:
self[k] = kwargs[k]
def __setitem__(self, k, v):
if k in self.__fields:
super(Selector, self).__setitem__(k, v)
super(Selector, self).__setitem__(self.__mask, self[self.__mask] | self.__fields[k][0])
else:
raise ReferenceError("%s is not allowed." % k)
Selector通过位掩码(mask)机制跟踪设置的选择条件,便于高效地构建和传递选择器参数。
UiObject类:UI元素操作
UiObject封装了对单个UI元素的所有操作:
点击操作实现
def click(self, timeout=None, offset=None):
self.must_wait(timeout=timeout)
x, y = self.center(offset=offset)
self.session.click(x, y)
点击操作流程:
- 等待元素出现(通过
must_wait方法) - 计算元素中心点坐标(考虑偏移量)
- 调用Device的click方法执行点击
文本输入实现
def set_text(self, text, timeout=None):
self.must_wait(timeout=timeout)
if not text:
return self.jsonrpc.clearTextField(self.selector)
else:
return self.jsonrpc.setText(self.selector, text)
文本输入支持两种操作:清空文本(当text为None时)和设置文本内容。
4. BasicUiautomatorServer类:服务通信核心
BasicUiautomatorServer类位于uiautomator2/core.py,负责与Android设备上的Uiautomator服务通信,是整个框架的通信核心。
服务启动与管理
def start_uiautomator(self):
with self._lock:
self._setup_jar()
if self._process:
if self._process.pool() is not None:
self._process = None
if not self._check_alive():
self._process = launch_uiautomator(self._dev)
self._wait_ready()
服务启动流程:
- 检查并推送u2.jar到设备
- 检查服务是否已运行
- 如未运行,启动Uiautomator服务
- 等待服务就绪
JSONRPC调用实现
def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
try:
return _jsonrpc_call(self._dev, method, params, timeout, self._debug)
except (HTTPError, UiAutomationNotConnectedError) as e:
logger.debug("uiautomator2 is not ok, error: %s", e)
self.stop_uiautomator()
self.start_uiautomator()
return _jsonrpc_call(self._dev, method, params, timeout, self._debug)
JSONRPC调用实现了自动重连机制:当检测到通信错误时,会尝试重启Uiautomator服务并重新发送请求。
关键技术实现原理
1. 元素定位机制
uiautomator2提供了多种元素定位方式,包括属性定位和XPath定位,满足不同场景的需求。
属性定位流程
2. 事件注入机制
uiautomator2支持多种用户事件注入,包括点击、滑动、按键等,这些操作通过不同的机制实现。
点击事件实现
点击事件通过两种方式实现:
- 对于可点击元素,直接调用Uiautomator的click方法
- 通用方式:计算元素位置,注入DOWN和UP事件
# 直接点击实现(元素可点击时)
self.jsonrpc.click(self.selector)
# 坐标点击实现(通用方式)
self.session.click(x, y)
滑动事件实现
def swipe(self, fx, fy, tx, ty, duration: Optional[float] = None, steps: Optional[int] = None):
if duration is not None and steps is not None:
warnings.warn("duration and steps can not be set at the same time, use steps", UserWarning)
duration = None
if duration:
steps = int(duration * 200)
if not steps:
steps = SCROLL_STEPS
rel2abs = self.pos_rel2abs
fx, fy = rel2abs(fx, fy)
tx, ty = rel2abs(tx, ty)
steps = max(2, steps) # step=1 has no swipe effect
with self._operation_delay("swipe"):
return self.jsonrpc.swipe(fx, fy, tx, ty, steps)
滑动实现通过将滑动距离分解为多个小步骤,模拟真实滑动过程,steps参数控制滑动的平滑度。
3. 应用管理机制
uiautomator2提供了完整的应用生命周期管理功能,包括启动、停止、安装、卸载等。
应用启动实现
def app_start(self, package_name: str, activity: Optional[str] = None, wait: bool = False, stop: bool = False, use_monkey: bool = False):
if stop:
self.app_stop(package_name)
if use_monkey or not activity:
self.shell([
'monkey', '-p', package_name, '-c',
'android.intent.category.LAUNCHER', '1'
])
if wait:
self.app_wait(package_name)
return
args = [
'am', 'start', '-a', 'android.intent.action.MAIN', '-c',
'android.intent.category.LAUNCHER',
'-n', f'{package_name}/{activity}'
]
self.shell(args)
if wait:
self.app_wait(package_name)
应用启动支持两种方式:
- 通过monkey命令启动(适用于没有指定activity的情况)
- 通过am start命令直接启动指定的activity
高级功能实现原理
1. 多条件元素定位
uiautomator2支持通过XPathSelector实现多条件组合定位:
class XPathSelector(AbstractSelector):
def __and__(self, value) -> 'XPathSelector':
s = XPathSelector(self)
s._next_xpath = XPathSelector.create(value)
s._operator = Operator.AND
s._parent = self._parent
return s
def __or__(self, value) -> 'XPathSelector':
s = XPathSelector(self)
s._next_xpath = XPathSelector.create(value)
s._operator = Operator.OR
s._parent = self._parent
return s
通过重载&(AND)和|(OR)操作符,实现多条件组合:
# 示例:查找文本为"确定"且类名为"Button"的元素
d.xpath("//*[@text='确定']").xpath("//*[@class='Button']").click()
# 或使用更简洁的语法
d.xpath("//*[@text='确定'] & //*[@class='Button']").click()
2. 观察者模式(Watcher)
uiautomator2实现了观察者模式,用于监控特定UI元素的出现并自动执行操作:
class Watcher:
def __init__(self, d: "uiautomator2.Device"):
self._d = d
self._watchers = {}
self._running = False
self._thread = None
self._lock = threading.Lock()
def when(self, xpath: str) -> "XPathWatcher":
name = str(uuid.uuid4())
watcher = XPathWatcher(self, xpath, name)
self._watchers[name] = watcher
return watcher
def start(self, interval: float = 2.0):
with self._lock:
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._watch_forever, args=(interval,))
self._thread.daemon = True
self._thread.start()
def _watch_forever(self, interval: float):
while self._running:
self.run()
time.sleep(interval)
def run(self, source: Optional[PageSource] = None):
if not self._watchers:
return
source = source or self._d.dump_hierarchy()
self._run_watchers(source)
Watcher工作流程:
- 创建Watcher并设置要监控的XPath表达式
- 启动后台线程定期检查元素
- 当元素出现时,执行预设操作(如点击)
3. 截图与图像识别
uiautomator2集成了图像识别功能,支持基于图像的元素定位和操作:
class Image:
def __init__(self, d: "uiautomator2.Device"):
self._d = d
def match(self, imdata: Union[np.ndarray, str, Image.Image]):
# 实现图像匹配逻辑
pass
def wait(self, imdata, timeout=30.0, threshold=0.9):
# 等待图像出现
pass
def click(self, imdata, timeout=30.0, threshold=0.9):
# 点击图像位置
pass
图像识别使用SSIM(结构相似性指数)算法比较图像相似度:
def compare_ssim(image_a: ImageType, image_b: ImageType, full=False, bounds=None):
# 转换为灰度图像
gray_a = color_bgr2gray(image_a)
gray_b = color_bgr2gray(image_b)
# 计算SSIM
(score, diff) = ssim(gray_a, gray_b, full=True)
if full:
return score, diff
return score
性能优化策略
uiautomator2在设计中采用了多种性能优化策略,确保自动化操作的高效执行。
1. 缓存机制
通过缓存机制减少重复计算和网络请求:
def cache_return(fn):
cache = {}
@functools.wraps(fn)
def inner(*args, **kwargs):
key = (args, frozenset(kwargs.items()))
if key not in cache:
cache[key] = fn(*args, **kwargs)
return cache[key]
return inner
缓存装饰器可应用于频繁调用但结果稳定的方法,如获取设备信息、窗口大小等。
2. 延迟注入与操作合并
通过操作延迟注入机制,减少不必要的等待时间:
@contextlib.contextmanager
def _operation_delay(self, operation_name: str = None):
before, after = self.settings['operation_delay']
# 排除不要求延迟的方法
if operation_name not in self.settings['operation_delay_methods']:
before, after = 0, 0
if before:
logger.debug(f"operation [{operation_name}] pre-delay {before}s")
time.sleep(before)
yield
if after:
logger.debug(f"operation [{operation_name}] post-delay {after}s")
time.sleep(after)
异常处理机制
uiautomator2实现了完善的异常处理机制,确保框架的健壮性和错误的可调试性。
异常体系
框架定义了多种特定异常类型,便于错误分类和处理:
class UiAutomationError(Exception):
"""Base exception for all uiautomator2 errors"""
class ConnectError(UiAutomationError):
"""Raised when device can't be connected"""
class UiObjectNotFoundError(UiAutomationError):
"""Raised when UI object not found"""
class XPathError(UiAutomationError):
"""Raised when xpath is invalid"""
class HTTPError(UiAutomationError):
"""Raised when HTTP request failed"""
自动重连机制
当检测到连接错误时,框架会尝试自动恢复连接:
def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
try:
return _jsonrpc_call(self._dev, method, params, timeout, self._debug)
except (HTTPError, UiAutomationNotConnectedError) as e:
logger.debug("uiautomator2 is not ok, error: %s", e)
self.stop_uiautomator()
self.start_uiautomator()
return _jsonrpc_call(self._dev, method, params, timeout, self._debug)
总结与扩展
uiautomator2 Python Wrapper通过精心设计的类结构和方法实现,为Android UI自动化测试提供了强大而灵活的工具。其核心优势包括:
- 多层次的元素定位:支持Selector API和XPath表达式,满足简单和复杂场景需求
- 丰富的操作支持:覆盖各种用户交互,包括点击、滑动、输入等操作
- 高效的设备通信:优化的JSONRPC通信和自动重连机制
- 完善的异常处理:细致的错误分类和自动恢复机制
扩展方向
基于uiautomator2的核心功能,可以进一步扩展以下高级特性:
- 测试报告生成:集成HTML报告生成功能,记录测试步骤和结果
- 智能等待机制:基于机器学习的动态等待策略,优化等待时间
- 多设备协同:支持多台设备的协同操作,模拟多设备交互场景
- AI辅助定位:结合计算机视觉技术,实现更智能的元素识别
uiautomator2框架的设计充分考虑了可扩展性,通过模块化的结构和清晰的接口定义,为后续功能扩展提供了便利。无论是日常的自动化测试还是复杂的场景模拟,uiautomator2都提供了坚实的技术基础。
更多推荐


所有评论(0)