G.DEVICE.touch(pos, **kwargs)
通过本篇你讲了解到 Airtest 是如何跟安卓设备交互的，以及多设备时的多机交互使用。
在之前从 Touch 接口分析 Airtest 的图像识别中，在图像识别获取到目标位置以后，发起点击的操作是通过以下这句：
G.DEVICE.touch(pos, **kwargs)
看一下有那么多个类里有 touch 接口，device、minitouch、adb、android、win、linux、ios
另外再翻一下 airtest.core.api 这个文件里的其他接口
"""
Device Operations
"""
@logwrap
def shell(cmd):
"""
Start remote shell in the target device and execute the command
:param cmd: command to be run on device, e.g. "ls /data/local/tmp"
:return: the output of the shell cmd
:platforms: Android
"""
return G.DEVICE.shell(cmd)
@logwrap
def start_app(package, activity=None):
"""
Start the target application on device
:param package: name of the package to be started, e.g. "com.netease.my"
:param activity: the activity to start, default is None which means the main activity
:return: None
:platforms: Android, iOS
"""
G.DEVICE.start_app(package, activity)
可见，这些设备操作的接口都是通过这个G.DEVICE，所以这里就是我们要找的 Airtest 与各类被测设备交互的实现部分了。
先来看一下这个 G.DEVICE 是什么
class G(object):
"""Represent the globals variables"""
BASEDIR = []
LOGGER = AirtestLogger(None)
LOGGING = get_logger("airtest.core.api")
SCREEN = None
DEVICE = None
DEVICE_LIST = []
RECENT_CAPTURE = None
RECENT_CAPTURE_PATH = None
CUSTOM_DEVICES = {}
@classmethod
def add_device(cls, dev):
"""
Add device instance in G and set as current device.
Examples:
G.add_device(Android())
Args:
dev: device to init
Returns:
None
"""
cls.DEVICE = dev
cls.DEVICE_LIST.append(dev)
看这个 add_device 的注释，传入的 dev 是初始化之后的设备对象，例如安卓，ios 等，然后存放在 G.DEVICE 和添加到 G.DEVICE_LIST 列表里。既然是初始化，那么想必就是要在脚本的最前面的执行吧，所以 Airtest 新建脚本时自动生成的那句 auto_setup 应该就跟设备初始化有关系了，一起去看看。
def auto_setup(basedir=None, devices=None, logdir=None, project_root=None):
"""
Auto setup running env and try connect android device if not device connected.
"""
if devices:
for dev in devices:
connect_device(dev)
elif not G.DEVICE_LIST:
try:
connect_device("Android:///")
except IndexError:
pass
if basedir:
if os.path.isfile(basedir):
basedir = os.path.dirname(basedir)
if basedir not in G.BASEDIR:
G.BASEDIR.append(basedir)
if logdir:
set_logdir(logdir)
if project_root:
ST.PROJECT_ROOT = project_root
def connect_device(uri):
"""
Initialize device with uri, and set as current device.
:param uri: an URI where to connect to device, e.g. `android://adbhost:adbport/serialno?param=value¶m2=value2`
:return: device instance
:Example:
* ``android:///`` # local adb device using default params
* ``android://adbhost:adbport/1234566?cap_method=javacap&touch_method=adb`` # remote device using custom params
* ``windows:///`` # local Windows application
* ``ios:///`` # iOS device
"""
d = urlparse(uri)
platform = d.scheme
host = d.netloc
uuid = d.path.lstrip("/")
params = dict(parse_qsl(d.query))
if host:
params["host"] = host.split(":")
dev = init_device(platform, uuid, **params)
return dev
def init_device(platform="Android", uuid=None, **kwargs):
"""
Initialize device if not yet, and set as current device.
:param platform: Android, IOS or Windows
:param uuid: uuid for target device, e.g. serialno for Android, handle for Windows, uuid for iOS
:param kwargs: Optional platform specific keyword args, e.g. `cap_method=JAVACAP` for Android
:return: device instance
"""
cls = import_device_cls(platform)
dev = cls(uuid, **kwargs)
for index, instance in enumerate(G.DEVICE_LIST):
if dev.uuid == instance.uuid:
G.LOGGING.warn("Device:%s updated %s -> %s" % (dev.uuid, instance, dev))
G.DEVICE_LIST[index] = dev
break
else:
G.add_device(dev)
return dev
def import_device_cls(platform):
"""lazy import device class"""
platform = platform.lower()
if platform in G.CUSTOM_DEVICES:
cls = G.CUSTOM_DEVICES[platform]
elif platform == "android":
from airtest.core.android.android import Android as cls
elif platform == "windows":
from airtest.core.win.win import Windows as cls
elif platform == "ios":
from airtest.core.ios.ios import IOS as cls
elif platform == "linux":
from airtest.core.linux.linux import Linux as cls
else:
raise RuntimeError("Unknown platform: %s" % platform)
return cls
由上到下的调用关系：auto_setup -> connect_device -> init_device -> add_device
auto_setup 接口：依次连接全部设备，处理日志，工程根目录等事物
connect_device 接口：根据传入参数 uri 的解析出其平台和序列号信息，然后初始化设备
init_device 接口：调用 import_device_cls 导入不同的平台，初始化设备对象，如果 DEVICE_LIST 列表里没有该设备，则添加设备
add_device 接口：将新连接上的设备赋值给 G.DEVICE，添加到 G.DEVICE_LIST
所以在 Airtest 教程中的 “4.3 多机协作脚本” 讲到：
在我们的脚本中，支持通过
set_current接口来切换当前连接的手机，因此我们一个脚本中，是能够调用多台手机，编写出一些复杂的多机交互脚本的。在命令行运行脚本时，只需要将手机依次使用
--device Android:///添加到命令行中即可，例如：airtest run untitled.air --device Android:///serialno1 --device Android:///serialno2 --device
在之前的笔记里分析过 run_script 接口解析命令行参数中的 device 会生成成一个设备列表，传入到 auto_setup 里就会遍历列表逐个去连接，所以多设备交互的操作是：
1.初始化连接所有的设备——命令行或者是调用 run_script 传入多个设备，当然也可以直接调用 connect_device、add_device;
2.调用 set_current 来切换当前操作的设备。
set_current 接口很简单了，在 G.DEVICE_LIST 里找出目标设备，赋值给 G.DEVICE，因为对设备的操作都是通过 G.DEVICE 的，所以只要换掉 G.DEVICE 就完成了设备的切换。看下源码：
def set_current(idx):
"""
Set current active device.
:param idx: uuid or index of initialized device instance
:raise IndexError: raised when device idx is not found
:return: None
:platforms: Android, iOS, Windows
"""
dev_dict = {dev.uuid: dev for dev in G.DEVICE_LIST}
if idx in dev_dict:
current_dev = dev_dict[idx]
elif isinstance(idx, int) and idx < len(G.DEVICE_LIST):
current_dev = G.DEVICE_LIST[idx]
else:
raise IndexError("device idx not found in: %s or %s" % (
list(dev_dict.keys()), list(range(len(G.DEVICE_LIST)))))
G.DEVICE = current_dev
关于 Airtest 的设备管理的分析大概就是以上这些了，多设备的交互很简单，不用在具体的操作方法中指定设备，而是只用在中间调用 set_current 来完成切换设备，例如切换前是 A 设备，那么所有的操作都会指向 A 设备，切换后则都指向 B 设备，这种设计也挺省事的。
接下来再拿 android 这部分来看一下 airtest 是怎么跟设备交互的。
从 import_device_cls 接口里找进去
'elif platform == "android": from airtest.core.android.android import Android as cls'
android 平台的设备管理在 airtest.core.android.android 的 Android 类里
class Android(Device):
"""Android Device Class"""
def __init__(self, serialno=None, host=None,
cap_method=CAP_METHOD.MINICAP_STREAM,
touch_method=TOUCH_METHOD.MINITOUCH,
ime_method=IME_METHOD.YOSEMITEIME,
ori_method=ORI_METHOD.MINICAP,
):
super(Android, self).__init__()
self.serialno = serialno or self.get_default_device()
self.cap_method = cap_method.upper()
self.touch_method = touch_method.upper()
self.ime_method = ime_method.upper()
self.ori_method = ori_method.upper()
# init adb
self.adb = ADB(self.serialno, server_addr=host)
self.adb.wait_for_device()
self.sdk_version = self.adb.sdk_version
self._display_info = {}
self._current_orientation = None
# init components
self.rotation_watcher = RotationWatcher(self.adb)
self.minicap = Minicap(self.adb, ori_function=self.get_display_info)
self.javacap = Javacap(self.adb)
self.minitouch = Minitouch(self.adb, ori_function=self.get_display_info)
self.yosemite_ime = YosemiteIme(self.adb)
self.recorder = Recorder(self.adb)
self._register_rotation_watcher()
Android 是安卓设备类，父类是 Device,这是一个基类，只定义了设备通用接口。android 设备初始化，初始化 adb，初始化 minicap、javacap、minitouch、yosemite、recorder 等组件。
翻一下 Android 类的接口，全都是对安卓设备的操作，基本的一些操作是通过 adb 完成的，比如：启动应用，卸载应用，唤醒...
def start_app(self, package, activity=None):
"""
Start the application and activity
Args:
package: package name
activity: activity name
Returns:
None
"""
return self.adb.start_app(package, activity)
def unlock(self):
"""
Unlock the device
Notes:
Might not work on all devices
Returns:
None
"""
return self.adb.unlock()
还有就是用到了其他组件的操作了，比如截图用到 minicap 和 javacap 组件，截图有四种方式：minicap_stream、minicap、javacap、adb_snapshot，初始化传入参数可配置截图的方式，默认是 MINICAP_STREAM，截图之后就是写入，转换成 cv2 的格式，处理横竖屏的转换。
def snapshot(self, filename=None, ensure_orientation=True):
"""
Take the screenshot of the display. The output is send to stdout by default.
Args:
filename: name of the file where to store the screenshot, default is None which si stdout
ensure_orientation: True or False whether to keep the orientation same as display
Returns:
screenshot output
"""
"""default not write into file."""
if self.cap_method == CAP_METHOD.MINICAP_STREAM:
self.rotation_watcher.get_ready()
screen = self.minicap.get_frame_from_stream()
elif self.cap_method == CAP_METHOD.MINICAP:
screen = self.minicap.get_frame()
elif self.cap_method == CAP_METHOD.JAVACAP:
screen = self.javacap.get_frame_from_stream()
else:
screen = self.adb.snapshot()
# output cv2 object
try:
screen = aircv.utils.string_2_img(screen)
except Exception:
# may be black/locked screen or other reason, print exc for debugging
import traceback
traceback.print_exc()
return None
# ensure the orientation is right
if ensure_orientation and self.display_info["orientation"]:
# minicap screenshots are different for various sdk_version
if self.cap_method in (CAP_METHOD.MINICAP, CAP_METHOD.MINICAP_STREAM) and self.sdk_version <= 16:
h, w = screen.shape[:2] # cvshape是高度在前面!!!!
if w < h: # 当前是横屏，但是图片是竖的，则旋转，针对sdk<=16的机器
screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
# adb 截图总是要根据orientation旋转
elif self.cap_method == CAP_METHOD.ADBCAP:
screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
if filename:
aircv.imwrite(filename, screen)
return screen
输入字符用到 yosemite 输入法，在 yosemite 初始化时会往安卓设备中安装一个叫 yosemite 的输入法 app，并通过 adb 命令将设备的当前输入法切换成 yosemite，yosemite 输入法 app 有个广播接收器，接收到广播后输入字符。
self.yosemite_ime = YosemiteIme(self.adb)
class YosemiteIme(CustomIme):
"""
Yosemite Input Method Class Object
"""
def __init__(self, adb):
super(YosemiteIme, self).__init__(adb, None, YOSEMITE_IME_SERVICE)
self.yosemite = Yosemite(adb)
def start(self):
self.yosemite.get_ready()
super(YosemiteIme, self).start()
def text(self, value):
"""
Input text with Yosemite input method
Args:
value: text to be inputted
Returns:
output form `adb shell` command
"""
if not self.started:
self.start()
# 更多的输入用法请见 https://github.com/macacajs/android-unicode#use-in-adb-shell
value = ensure_unicode(value)
self.adb.shell(u"am broadcast -a ADB_INPUT_TEXT --es msg '{}'".format(value))
def start(self):
"""
Enable input method
Returns:
None
"""
try:
self.default_ime = self.adb.shell("settings get secure default_input_method").strip()
except AdbError:
# settings cmd not found for older phones, e.g. Xiaomi 2A
# /system/bin/sh: settings: not found
self.default_ime = None
self.ime_list = self._get_ime_list()
if self.service_name not in self.ime_list:
if self.apk_path:
self.device.install_app(self.apk_path)
if self.default_ime != self.service_name:
self.adb.shell("ime enable %s" % self.service_name)
self.adb.shell("ime set %s" % self.service_name)
self.started = True
所以输入字符的接口也有两种方式：yosemite 输入法和 adb 命令，默认是 yosemite 输入
def text(self, text, enter=True):
"""
Input text on the device
Args:
text: text to input
enter: True or False whether to press `Enter` key
Returns:
None
"""
if self.ime_method == IME_METHOD.YOSEMITEIME:
self.yosemite_ime.text(text)
else:
self.adb.shell(["input", "text", text])
# 游戏输入时，输入有效内容后点击Enter确认，如不需要，enter置为False即可。
if enter:
self.adb.shell(["input", "keyevent", "ENTER"])
录屏用到 recorder 组件，录屏是用 yosemite 这个 app 实现的，pythod 这边只是发 adb 命令，简单的看一下 start_record 这部分吧，
源码位置：airtest/core/android/android.py
def start_recording(self, *args, **kwargs):
"""
Start recording the device display
Args:
*args: optional arguments
**kwargs: optional arguments
Returns:
None
"""
return self.recorder.start_recording(*args, **kwargs)
源码位置：airtest/core/android/recorder.py
@on_method_ready('install_or_upgrade')
def start_recording(self, max_time=1800, bit_rate=None, vertical=None):
"""
Start screen recording
Args:
max_time: maximum rate value, default is 1800
bit_rate: bit rate value, default is None
vertical: vertical parameters, default is None
Raises:
RuntimeError: if any error occurs while setup the recording
Returns:
None if recording did not start, otherwise True
"""
if getattr(self, "recording_proc", None):
raise AirtestError("recording_proc has already started")
pkg_path = self.adb.path_app(YOSEMITE_PACKAGE)
max_time_param = "-Dduration=%d" % max_time if max_time else ""
bit_rate_param = "-Dbitrate=%d" % bit_rate if bit_rate else ""
if vertical is None:
vertical_param = ""
else:
vertical_param = "-Dvertical=true" if vertical else "-Dvertical=false"
p = self.adb.start_shell('CLASSPATH=%s exec app_process %s %s %s /system/bin %s.Recorder --start-record' %
(pkg_path, max_time_param, bit_rate_param, vertical_param, YOSEMITE_PACKAGE))
nbsp = NonBlockingStreamReader(p.stdout)
while True:
line = nbsp.readline(timeout=5)
if line is None:
raise RuntimeError("start recording error")
if six.PY3:
line = line.decode("utf-8")
m = re.match("start result: Record start success! File path:(.*\.mp4)", line.strip())
if m:
output = m.group(1)
self.recording_proc = p
self.recording_file = output
return True
点击、滑动等用到 minitouch 组件，同样的可选 minitouch 或者是 adb
def touch(self, pos, duration=0.01):
"""
Perform touch event on the device
Args:
pos: coordinates (x, y)
duration: how long to touch the screen
Returns:
None
"""
if self.touch_method == TOUCH_METHOD.MINITOUCH:
pos = self._touch_point_by_orientation(pos)
self.minitouch.touch(pos, duration=duration)
else:
self.adb.touch(pos)
minitouch、minicap 有啥不同呢，这是 openstf 的库，大概是在安卓设备下放了一个 client，pythod 这边用 safesocket 发消息给 client，由 client 执行操作，详细的先不在这里分析了。
android 设备类大致就是这样了，再往下可以看看 adb 类，这个就只看看发命令的核心接口吧。
def start_cmd(self, cmds, device=True):
"""
Start a subprocess with adb command(s)
Args:
cmds: command(s) to be run
device: if True, the device serial number must be specified by `-s serialno` argument
Raises:
RuntimeError: if `device` is True and serialno is not specified
Returns:
a subprocess
"""
if device:
if not self.serialno:
raise RuntimeError("please set serialno first")
cmd_options = self.cmd_options + ['-s', self.serialno]
else:
cmd_options = self.cmd_options
cmds = cmd_options + split_cmd(cmds)
LOGGING.debug(" ".join(cmds))
if not PY3:
cmds = [c.encode(get_std_encoding(sys.stdin)) for c in cmds]
proc = subprocess.Popen(
cmds,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
return proc
总结，Airtest 的设备管理只是用 G.DEVICE 指向当前设备，用 G.DEVICE_LIST 保存全部设备，所有的操作都通过 G.DEVICE 转发，所以改变 G.DEVICE 即可切换设备。而安卓设备的交互则是通过 adb 命令，和一些别的库:yosemete、minitouch、minicap、javacap。
