222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
# This file contains the main class of droidbot
|
||
# It can be used after AVD was started, app was installed, and adb had been set up properly
|
||
# By configuring and creating a droidbot instance,
|
||
# droidbot will start interacting with Android in AVD like a human
|
||
import logging
|
||
import os
|
||
import sys
|
||
import pkg_resources
|
||
import shutil
|
||
from threading import Timer
|
||
|
||
from .core import PlatformFactory, Platform
|
||
from .input_manager import InputManager
|
||
|
||
# Import platforms to trigger auto-registration with PlatformFactory
|
||
# This ensures platforms are registered before create_device is called
|
||
from . import platforms # noqa: F401
|
||
|
||
# 从 exceptions 模块导入致命异常（避免循环导入）
|
||
from .exceptions import FATAL_EXCEPTIONS # noqa: F401
|
||
|
||
|
||
class DroidBot(object):
|
||
"""
|
||
The main class of droidbot
|
||
"""
|
||
# this is a single instance class
|
||
instance = None
|
||
|
||
def __init__(self,
|
||
package_name=None,
|
||
device_serial=None,
|
||
is_emulator=False,
|
||
output_dir=None,
|
||
policy_name=None,
|
||
random_input=False,
|
||
event_count=None,
|
||
event_interval=None,
|
||
timeout=None,
|
||
keep_app=None,
|
||
keep_env=False,
|
||
cv_mode=False,
|
||
debug_mode=False,
|
||
profiling_method=None,
|
||
grant_perm=False,
|
||
enable_accessibility_hard=False,
|
||
|
||
humanoid=None,
|
||
ignore_ad=False,
|
||
replay_output=None,
|
||
enable_guiagent=False,
|
||
app_name=None,
|
||
platform="android",
|
||
pcap_callback=None,
|
||
enable_app_block=False,
|
||
**kwargs):
|
||
"""
|
||
initiate droidbot with configurations
|
||
:return:
|
||
"""
|
||
# 注意：日志配置现在由统一的 logging_config 模块在入口文件中管理
|
||
# 不在这里调用 basicConfig，避免覆盖已有配置
|
||
|
||
self.logger = logging.getLogger('DroidBot')
|
||
DroidBot.instance = self
|
||
|
||
self.output_dir = output_dir
|
||
if output_dir is not None:
|
||
if not os.path.isdir(output_dir):
|
||
os.makedirs(output_dir)
|
||
html_index_path = pkg_resources.resource_filename("DroidBot", "resources/index.html")
|
||
stylesheets_path = pkg_resources.resource_filename("DroidBot", "resources/stylesheets")
|
||
target_stylesheets_dir = os.path.join(output_dir, "stylesheets")
|
||
if os.path.exists(target_stylesheets_dir):
|
||
shutil.rmtree(target_stylesheets_dir)
|
||
shutil.copy(html_index_path, output_dir)
|
||
shutil.copytree(stylesheets_path, target_stylesheets_dir)
|
||
|
||
self.timeout = timeout
|
||
self.timer = None
|
||
self.keep_env = keep_env
|
||
self.keep_app = keep_app
|
||
|
||
self.device = None
|
||
self.input_manager = None
|
||
self.enable_accessibility_hard = enable_accessibility_hard
|
||
self.humanoid = humanoid
|
||
self.ignore_ad = ignore_ad
|
||
self.replay_output = replay_output
|
||
self.enable_guiagent = enable_guiagent
|
||
self.app_name = app_name
|
||
self.pcap_callback = pcap_callback
|
||
self.enable_app_block = enable_app_block
|
||
|
||
self.enabled = True
|
||
self._timeout_triggered = False
|
||
self._stopped = False
|
||
|
||
try:
|
||
# Use PlatformFactory to create device
|
||
platform_enum = Platform(platform)
|
||
|
||
# Build platform-specific device arguments
|
||
if platform_enum == Platform.IOS:
|
||
# iOS device arguments
|
||
device_kwargs = {
|
||
'wda_url': kwargs.get('wda_url', 'http://localhost:8100'),
|
||
'bundle_id': package_name, # package_name is bundle_id for iOS
|
||
'output_dir': self.output_dir,
|
||
'cv_mode': cv_mode,
|
||
'udid': device_serial, # device_serial maps to udid for iOS
|
||
'debug_mode': debug_mode,
|
||
}
|
||
elif platform_enum == Platform.WINDOWS:
|
||
# Windows device arguments
|
||
device_kwargs = {
|
||
'window_title': self.app_name, # app_name maps to window_title
|
||
'exe_path': device_serial, # device_serial maps to exe_path
|
||
'steam_game_id': package_name, # package_name maps to steam_game_id
|
||
'output_dir': self.output_dir,
|
||
'cv_mode': cv_mode,
|
||
'debug_mode': debug_mode,
|
||
}
|
||
elif platform_enum == Platform.WEB:
|
||
# Web device arguments
|
||
device_kwargs = {
|
||
'app_path': package_name, # URL作为app_path
|
||
'output_dir': self.output_dir,
|
||
'browser': kwargs.get('browser', 'chrome'),
|
||
'engine': kwargs.get('engine', 'playwright'),
|
||
'headless': kwargs.get('headless', False),
|
||
}
|
||
else:
|
||
# Android device arguments
|
||
device_kwargs = {
|
||
'device_serial': device_serial,
|
||
'is_emulator': is_emulator,
|
||
'output_dir': self.output_dir,
|
||
'app_path': package_name,
|
||
'cv_mode': cv_mode,
|
||
'grant_perm': grant_perm,
|
||
'enable_accessibility_hard': self.enable_accessibility_hard,
|
||
'humanoid': self.humanoid,
|
||
'ignore_ad': ignore_ad,
|
||
**kwargs,
|
||
}
|
||
|
||
self.device = PlatformFactory.create_device(platform_enum, **device_kwargs)
|
||
|
||
self.input_manager = InputManager(
|
||
device=self.device,
|
||
policy_name=policy_name,
|
||
random_input=random_input,
|
||
event_count=event_count,
|
||
event_interval=event_interval,
|
||
profiling_method=profiling_method,
|
||
replay_output=replay_output,
|
||
enable_guiagent=self.enable_guiagent,
|
||
app_name=self.app_name,
|
||
pcap_callback=self.pcap_callback,
|
||
enable_app_block=self.enable_app_block)
|
||
except Exception:
|
||
self.stop()
|
||
raise
|
||
|
||
def start(self):
|
||
"""
|
||
start interacting
|
||
:return:
|
||
"""
|
||
if not self.enabled:
|
||
return
|
||
self.logger.info("Starting DroidBot")
|
||
try:
|
||
if self.timeout > 0:
|
||
self.timer = Timer(self.timeout, self._on_timeout)
|
||
self.timer.start()
|
||
|
||
self.device.set_up()
|
||
|
||
if not self.enabled:
|
||
return
|
||
self.device.connect()
|
||
|
||
if not self.enabled:
|
||
return
|
||
# self.device.install_app() # app is already installed
|
||
|
||
|
||
|
||
if not self.enabled:
|
||
return
|
||
self.input_manager.start()
|
||
|
||
except Exception:
|
||
raise
|
||
finally:
|
||
self.stop()
|
||
self.logger.info("DroidBot Stopped")
|
||
|
||
def _on_timeout(self):
|
||
"""Timer 线程只发出停止信号，避免跨线程关闭 Playwright。"""
|
||
self._timeout_triggered = True
|
||
self.enabled = False
|
||
self.logger.warning(f"DroidBot timeout reached ({self.timeout}s), requesting graceful stop")
|
||
if self.input_manager:
|
||
self.input_manager.stop()
|
||
|
||
def stop(self):
|
||
if self._stopped:
|
||
return
|
||
self._stopped = True
|
||
self.enabled = False
|
||
if self.timer and self.timer.is_alive():
|
||
self.timer.cancel()
|
||
|
||
if self.input_manager:
|
||
self.logger.info("Total steps: %d" % self.input_manager.total_exploring_steps)
|
||
self.input_manager.stop()
|
||
if self.device:
|
||
self.device.disconnect()
|