360 lines
14 KiB
Python
360 lines
14 KiB
Python
"""
|
||
Android Device State Implementation
|
||
Concrete implementation of AbstractDeviceState for Android devices.
|
||
"""
|
||
import copy
|
||
import os
|
||
from typing import Optional, Dict, Any, List, Set
|
||
|
||
from ...core.abstract_device_state import AbstractDeviceState
|
||
from ...core.abstract_input_event import EventType
|
||
from ...exceptions import FATAL_EXCEPTIONS
|
||
|
||
|
||
class AndroidDeviceState(AbstractDeviceState):
|
||
"""
|
||
Android 设备状态的具体实现
|
||
"""
|
||
|
||
def __init__(self, device, views, foreground_activity, activity_stack,
|
||
background_services, tag=None, screenshot_path=None):
|
||
"""
|
||
初始化 Android 设备状态
|
||
|
||
:param device: AndroidDevice 实例
|
||
:param views: 视图列表
|
||
:param foreground_activity: 前台 Activity
|
||
:param activity_stack: Activity 栈
|
||
:param background_services: 后台服务
|
||
:param tag: 状态标签
|
||
:param screenshot_path: 截图路径
|
||
"""
|
||
super().__init__(device, tag, screenshot_path)
|
||
|
||
self._foreground_activity = foreground_activity
|
||
self.activity_stack = activity_stack if isinstance(activity_stack, list) else []
|
||
self.background_services = background_services
|
||
|
||
# 解析视图
|
||
self._views = self._parse_views(views)
|
||
self._view_tree = {}
|
||
self._assemble_view_tree(self._view_tree, self._views)
|
||
self._generate_view_strs()
|
||
|
||
# 计算状态标识
|
||
self._state_str = self._get_state_str()
|
||
self._structure_str = self._get_content_free_state_str()
|
||
self._search_content = self._get_search_content()
|
||
|
||
# ==================== 属性实现 ====================
|
||
|
||
@property
|
||
def search_content(self) -> str:
|
||
return self._search_content
|
||
|
||
@property
|
||
def views(self) -> List[Dict[str, Any]]:
|
||
return self._views
|
||
|
||
@property
|
||
def view_tree(self) -> Dict[str, Any]:
|
||
return self._view_tree
|
||
|
||
@property
|
||
def state_str(self) -> str:
|
||
return self._state_str
|
||
|
||
@property
|
||
def structure_str(self) -> str:
|
||
return self._structure_str
|
||
|
||
@property
|
||
def foreground_page(self) -> Optional[str]:
|
||
"""实现抽象接口 - 返回当前页面标识符（Android 上就是 foreground_activity）"""
|
||
return self._foreground_activity
|
||
|
||
|
||
def _parse_views(self, raw_views) -> List[Dict[str, Any]]:
|
||
"""解析原始视图数据，确保符合 ViewDict 格式"""
|
||
views = []
|
||
if not raw_views or len(raw_views) == 0:
|
||
return views
|
||
for view_dict in raw_views:
|
||
# 添加来源标识
|
||
view_dict['source'] = 'accessibility'
|
||
views.append(view_dict)
|
||
return views
|
||
|
||
|
||
def _assemble_view_tree(self, root_view, views) -> None:
|
||
"""组装视图树"""
|
||
if not len(self._view_tree):
|
||
if not len(views):
|
||
return
|
||
self._view_tree = copy.deepcopy(views[0])
|
||
self._assemble_view_tree(self._view_tree, views)
|
||
else:
|
||
children = list(enumerate(root_view.get("children", [])))
|
||
if not len(children):
|
||
return
|
||
for i, j in children:
|
||
if j < len(self._views):
|
||
root_view["children"][i] = copy.deepcopy(self._views[j])
|
||
self._assemble_view_tree(root_view["children"][i], views)
|
||
|
||
def _generate_view_strs(self) -> None:
|
||
"""生成视图字符串标识"""
|
||
for view_dict in self._views:
|
||
self._get_view_str(view_dict)
|
||
|
||
# ==================== 状态标识 ====================
|
||
|
||
def _get_state_str(self) -> str:
|
||
"""获取状态唯一标识"""
|
||
state_str_raw = self._get_state_str_raw()
|
||
from .utils import md5
|
||
return md5(state_str_raw)
|
||
|
||
def _get_state_str_raw(self) -> str:
|
||
"""获取原始状态字符串"""
|
||
if self.device.humanoid is not None:
|
||
import json
|
||
from xmlrpc.client import ServerProxy
|
||
proxy = ServerProxy("http://%s/" % self.device.humanoid)
|
||
return proxy.render_view_tree(json.dumps({
|
||
"view_tree": self._view_tree,
|
||
"screen_res": [self.device.display_info["width"],
|
||
self.device.display_info["height"]]
|
||
}))
|
||
else:
|
||
view_signatures = set()
|
||
for view in self._views:
|
||
view_signature = self._get_view_signature(view)
|
||
if view_signature:
|
||
view_signatures.add(view_signature)
|
||
return "%s{%s}" % (self._foreground_activity, ",".join(sorted(view_signatures)))
|
||
|
||
def _get_content_free_state_str(self) -> str:
|
||
"""获取内容无关的状态标识"""
|
||
view_signatures = set()
|
||
for view in self._views:
|
||
view_signature = self._get_content_free_view_signature(view)
|
||
if view_signature:
|
||
view_signatures.add(view_signature)
|
||
state_str = "%s{%s}" % (self._foreground_activity, ",".join(sorted(view_signatures)))
|
||
import hashlib
|
||
return hashlib.md5(state_str.encode('utf-8')).hexdigest()
|
||
|
||
def _get_search_content(self) -> str:
|
||
"""获取搜索内容"""
|
||
words = [
|
||
",".join(self._get_property_from_all_views("resource_id")),
|
||
",".join(self._get_property_from_all_views("text"))
|
||
]
|
||
return "\n".join(words)
|
||
|
||
def _get_property_from_all_views(self, property_name: str) -> Set[str]:
|
||
"""从所有视图获取属性值"""
|
||
property_values = set()
|
||
for view in self._views:
|
||
property_value = self._safe_dict_get(view, property_name, None)
|
||
if property_value:
|
||
property_values.add(property_value)
|
||
return property_values
|
||
|
||
# ==================== 视图签名 ====================
|
||
|
||
@staticmethod
|
||
def _get_view_signature(view_dict: Dict[str, Any]) -> Optional[str]:
|
||
"""获取视图签名"""
|
||
if 'signature' in view_dict:
|
||
return view_dict['signature']
|
||
|
||
view_text = AndroidDeviceState._safe_dict_get(view_dict, 'text', "None")
|
||
if view_text is None or len(view_text) > 50:
|
||
view_text = "None"
|
||
|
||
signature = "[class]%s[resource_id]%s[text]%s[%s,%s,%s]" % (
|
||
AndroidDeviceState._safe_dict_get(view_dict, 'class_name', "None"),
|
||
AndroidDeviceState._safe_dict_get(view_dict, 'resource_id', "None"),
|
||
view_text,
|
||
AndroidDeviceState._key_if_true(view_dict, 'enabled'),
|
||
AndroidDeviceState._key_if_true(view_dict, 'checked'),
|
||
AndroidDeviceState._key_if_true(view_dict, 'selected')
|
||
)
|
||
view_dict['signature'] = signature
|
||
return signature
|
||
|
||
@staticmethod
|
||
def _get_content_free_view_signature(view_dict: Dict[str, Any]) -> Optional[str]:
|
||
"""获取内容无关的视图签名"""
|
||
if 'content_free_signature' in view_dict:
|
||
return view_dict['content_free_signature']
|
||
|
||
content_free_signature = "[class]%s[resource_id]%s" % (
|
||
AndroidDeviceState._safe_dict_get(view_dict, 'class_name', "None"),
|
||
AndroidDeviceState._safe_dict_get(view_dict, 'resource_id', "None")
|
||
)
|
||
view_dict['content_free_signature'] = content_free_signature
|
||
return content_free_signature
|
||
|
||
def _get_view_str(self, view_dict: Dict[str, Any]) -> str:
|
||
"""获取视图字符串"""
|
||
if 'view_str' in view_dict:
|
||
return view_dict['view_str']
|
||
|
||
view_signature = self._get_view_signature(view_dict)
|
||
parent_strs = []
|
||
for parent_id in self.get_all_ancestors(view_dict):
|
||
parent_strs.append(self._get_view_signature(self._views[parent_id]))
|
||
parent_strs.reverse()
|
||
|
||
child_strs = []
|
||
for child_id in self.get_all_children(view_dict):
|
||
child_strs.append(self._get_view_signature(self._views[child_id]))
|
||
child_strs.sort()
|
||
|
||
view_str = "Activity:%s\nSelf:%s\nParents:%s\nChildren:%s" % (
|
||
self._foreground_activity, view_signature,
|
||
"//".join(parent_strs), "||".join(child_strs)
|
||
)
|
||
import hashlib
|
||
view_str = hashlib.md5(view_str.encode('utf-8')).hexdigest()
|
||
view_dict['view_str'] = view_str
|
||
return view_str
|
||
|
||
# ==================== 辅助方法 ====================
|
||
|
||
@staticmethod
|
||
def _key_if_true(view_dict: Dict[str, Any], key: str) -> str:
|
||
return key if (key in view_dict and view_dict[key]) else ""
|
||
|
||
@staticmethod
|
||
def _safe_dict_get(view_dict: Dict[str, Any], key: str, default=None):
|
||
value = view_dict.get(key, None)
|
||
return value if value is not None else default
|
||
|
||
def get_all_ancestors(self, view_dict: Dict[str, Any]) -> List[int]:
|
||
"""获取所有祖先节点 ID"""
|
||
result = []
|
||
parent_id = self._safe_dict_get(view_dict, 'parent', -1)
|
||
if 0 <= parent_id < len(self._views):
|
||
result.append(parent_id)
|
||
result += self.get_all_ancestors(self._views[parent_id])
|
||
return result
|
||
|
||
def get_all_children(self, view_dict: Dict[str, Any]) -> Set[int]:
|
||
"""获取所有子节点 ID"""
|
||
children = self._safe_dict_get(view_dict, 'children')
|
||
if not children:
|
||
return set()
|
||
children = set(children)
|
||
for child in list(children):
|
||
if child < len(self._views):
|
||
children_of_child = self.get_all_children(self._views[child])
|
||
children = children.union(children_of_child)
|
||
return children
|
||
|
||
# ==================== 抽象方法实现 ====================
|
||
|
||
def get_possible_input(self) -> List:
|
||
"""获取可能的输入事件"""
|
||
if self._possible_events:
|
||
return [] + self._possible_events
|
||
|
||
from .android_input_event import (
|
||
AndroidTouchEvent, AndroidLongTouchEvent,
|
||
AndroidScrollEvent, AndroidSetTextEvent
|
||
)
|
||
|
||
possible_events = []
|
||
enabled_view_ids = []
|
||
touch_exclude_view_ids = set()
|
||
|
||
for view_dict in self._views:
|
||
if (self._safe_dict_get(view_dict, 'enabled') and
|
||
self._safe_dict_get(view_dict, 'visible') and
|
||
self._safe_dict_get(view_dict, 'resource_id') not in
|
||
['android:id/navigationBarBackground', 'android:id/statusBarBackground']):
|
||
enabled_view_ids.append(view_dict['temp_id'])
|
||
|
||
for view_id in enabled_view_ids:
|
||
if self._safe_dict_get(self._views[view_id], 'clickable'):
|
||
possible_events.append(AndroidTouchEvent(view=self._views[view_id]))
|
||
touch_exclude_view_ids.add(view_id)
|
||
touch_exclude_view_ids = touch_exclude_view_ids.union(
|
||
self.get_all_children(self._views[view_id])
|
||
)
|
||
|
||
# 添加滚动事件
|
||
possible_events.append(AndroidScrollEvent(direction="up"))
|
||
possible_events.append(AndroidScrollEvent(direction="up"))
|
||
possible_events.append(AndroidScrollEvent(direction="up"))
|
||
possible_events.append(AndroidScrollEvent(direction="down"))
|
||
|
||
for view_id in enabled_view_ids:
|
||
if self._safe_dict_get(self._views[view_id], 'checkable'):
|
||
possible_events.append(AndroidTouchEvent(view=self._views[view_id]))
|
||
touch_exclude_view_ids.add(view_id)
|
||
|
||
for view_id in enabled_view_ids:
|
||
if self._safe_dict_get(self._views[view_id], 'long_clickable'):
|
||
possible_events.append(AndroidLongTouchEvent(view=self._views[view_id]))
|
||
|
||
for view_id in enabled_view_ids:
|
||
if self._safe_dict_get(self._views[view_id], 'editable'):
|
||
possible_events.append(AndroidSetTextEvent(
|
||
view=self._views[view_id], text="cat"
|
||
))
|
||
touch_exclude_view_ids.add(view_id)
|
||
|
||
for view_id in enabled_view_ids:
|
||
if view_id in touch_exclude_view_ids:
|
||
continue
|
||
children = self._safe_dict_get(self._views[view_id], 'children')
|
||
if children and len(children) > 0:
|
||
continue
|
||
possible_events.append(AndroidTouchEvent(view=self._views[view_id]))
|
||
|
||
self._possible_events = possible_events
|
||
return [] + possible_events
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""序列化为字典"""
|
||
return {
|
||
'tag': self.tag,
|
||
'state_str': self._state_str,
|
||
'state_str_content_free': self._structure_str,
|
||
'foreground_activity': self._foreground_activity,
|
||
'activity_stack': self.activity_stack,
|
||
'background_services': self.background_services,
|
||
'width': self.width,
|
||
'height': self.height,
|
||
'views': self._views
|
||
}
|
||
|
||
def get_app_page_depth(self) -> int:
|
||
"""获取应用页面深度 - 实现抽象接口"""
|
||
if not self.device._app:
|
||
return 0 # 没有指定应用时，返回0（在前台）
|
||
|
||
package_name = self.device.app_identifier
|
||
if self._foreground_activity and package_name in self._foreground_activity:
|
||
return 0
|
||
|
||
try:
|
||
app_pid = self.device.get_app_pid(package_name)
|
||
if app_pid:
|
||
return 0
|
||
except FATAL_EXCEPTIONS:
|
||
raise
|
||
except Exception as e:
|
||
self.device.logger.error(f"Exception checking app PID: {e}")
|
||
return -1 # 非致命异常，返回默认深度
|
||
|
||
depth = 0
|
||
for activity_str in self.activity_stack:
|
||
if package_name in activity_str:
|
||
return depth
|
||
depth += 1
|
||
return -1 |