162 lines
6.0 KiB
Python
162 lines
6.0 KiB
Python
"""
|
|
Event Log
|
|
Platform-agnostic event logging for recording device interactions.
|
|
"""
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
|
|
class EventLog:
|
|
"""
|
|
事件日志类
|
|
|
|
用于记录设备交互事件，包括开始状态、结束状态、性能分析等。
|
|
"""
|
|
|
|
def __init__(self, device, app, event, profiling_method=None, tag=None):
|
|
"""
|
|
初始化事件日志
|
|
|
|
:param device: 设备对象
|
|
:param app: 应用对象
|
|
:param event: 事件对象
|
|
:param profiling_method: 性能分析方法
|
|
:param tag: 日志标签
|
|
"""
|
|
self.device = device
|
|
self.app = app
|
|
self.event = event
|
|
if tag is None:
|
|
from datetime import datetime
|
|
tag = datetime.now().strftime("%Y-%m-%d_%H%M%S")
|
|
self.tag = tag
|
|
|
|
self.from_state = None
|
|
self.to_state = None
|
|
self.event_str = None
|
|
|
|
self.profiling_method = profiling_method
|
|
self.trace_remote_file = "/data/local/tmp/event.trace"
|
|
self.is_profiling = False
|
|
self.sampling = None
|
|
|
|
# sampling feature was added in Android 5.0 (API level 21)
|
|
self.sampling = None
|
|
if profiling_method is not None and str(profiling_method) != "full":
|
|
# Check device capability safely
|
|
if hasattr(device, 'get_sdk_version') and device.get_sdk_version() >= 21:
|
|
try:
|
|
self.sampling = int(profiling_method)
|
|
except:
|
|
pass
|
|
|
|
def to_dict(self):
|
|
"""序列化为字典"""
|
|
return {
|
|
"tag": self.tag,
|
|
"event": self.event.to_dict(),
|
|
"start_state": self.from_state.state_str if self.from_state else None,
|
|
"stop_state": self.to_state.state_str if self.to_state else None,
|
|
"event_str": self.event_str
|
|
}
|
|
|
|
def save2dir(self, output_dir=None):
|
|
"""保存事件到目录"""
|
|
if output_dir is None:
|
|
if self.device.output_dir is None:
|
|
return
|
|
else:
|
|
output_dir = os.path.join(self.device.output_dir, "events")
|
|
try:
|
|
if not os.path.exists(output_dir):
|
|
os.makedirs(output_dir)
|
|
event_json_file_path = "%s/event_%s.json" % (output_dir, self.tag)
|
|
with open(event_json_file_path, "w") as f:
|
|
json.dump(self.to_dict(), f, indent=2)
|
|
except Exception as e:
|
|
self.device.logger.error("Saving event to dir failed, %s" % e)
|
|
|
|
def save_views(self, output_dir=None):
|
|
"""保存视图"""
|
|
views = self.event.get_views()
|
|
if views and self.from_state and hasattr(self.from_state, 'save_view_img'):
|
|
for view_dict in views:
|
|
self.from_state.save_view_img(view_dict=view_dict, output_dir=output_dir)
|
|
|
|
def is_start_event(self):
|
|
"""检查是否是启动事件"""
|
|
from .abstract_input_event import EventType
|
|
if hasattr(self.event, 'event_type') and self.event.event_type == EventType.INTENT:
|
|
intent_cmd = getattr(self.event, 'intent', '')
|
|
if intent_cmd and hasattr(self.app, 'get_package_name'):
|
|
if "start" in str(intent_cmd) and self.app.get_package_name() in str(intent_cmd):
|
|
return True
|
|
return False
|
|
|
|
def start(self):
|
|
"""开始发送事件"""
|
|
if hasattr(self.device, '_last_state') and self.device._last_state is not None:
|
|
self.from_state = self.device._last_state
|
|
else:
|
|
print("Warning: No last state available, using current state as start state.")
|
|
self.from_state = self.device.get_current_state()
|
|
self.start_profiling()
|
|
self.event_str = self.event.get_event_str(self.from_state)
|
|
print(f"[DroidBot] Action: {self.event_str}") # 添加打印以确保可见
|
|
self.device.logger.info("Action: %s" % self.event_str)
|
|
self.device.send_event(self.event)
|
|
|
|
def start_profiling(self):
|
|
"""开始性能分析"""
|
|
if self.profiling_method is None:
|
|
return
|
|
if self.is_profiling:
|
|
return
|
|
|
|
# 尝试使用设备提供的统一接口
|
|
if hasattr(self.device, 'start_profiling'):
|
|
success = self.device.start_profiling(self.trace_remote_file, self.sampling)
|
|
if success:
|
|
self.is_profiling = True
|
|
return
|
|
|
|
# 处理启动事件的特殊情况 (Legacy Android support)
|
|
# 如果是启动事件，并且应用尚未运行，可能需要修改 Intent
|
|
if self.is_start_event():
|
|
if hasattr(self.app, 'get_start_with_profiling_intent'):
|
|
start_intent = self.app.get_start_with_profiling_intent(self.trace_remote_file, self.sampling)
|
|
if hasattr(self.event, 'intent') and hasattr(start_intent, 'get_cmd'):
|
|
self.event.intent = start_intent.get_cmd()
|
|
self.is_profiling = True
|
|
|
|
def stop(self):
|
|
"""结束发送事件"""
|
|
self.stop_profiling()
|
|
self.to_state = self.device.get_current_state()
|
|
if hasattr(self.device, '_last_state'):
|
|
self.device._last_state = self.to_state
|
|
self.save2dir()
|
|
self.save_views()
|
|
|
|
def stop_profiling(self, output_dir=None):
|
|
"""停止性能分析"""
|
|
if self.profiling_method is None:
|
|
return
|
|
if not self.is_profiling:
|
|
return
|
|
|
|
if output_dir is None:
|
|
if self.device.output_dir is None:
|
|
return
|
|
else:
|
|
output_dir = os.path.join(self.device.output_dir, "events")
|
|
if not os.path.exists(output_dir):
|
|
os.makedirs(output_dir)
|
|
event_trace_local_path = "%s/event_trace_%s.trace" % (output_dir, self.tag)
|
|
|
|
# 尝试使用设备提供的统一接口
|
|
if hasattr(self.device, 'stop_profiling'):
|
|
self.device.stop_profiling(self.trace_remote_file, event_trace_local_path)
|
|
|