通過本篇你講了解到Airtest是如何跟安卓設(shè)備交互的着饥，以及多設(shè)備時(shí)的多機(jī)交互使用犀农。
在之前從Touch接口分析Airtest的圖像識(shí)別中，在圖像識(shí)別獲取到目標(biāo)位置以后宰掉，發(fā)起點(diǎn)擊的操作是通過以下這句：
G.DEVICE.touch(pos, **kwargs)
看一下有那么多個(gè)類里有touch接口呵哨，device、minitouch轨奄、adb孟害、android、win挪拟、linux挨务、ios
另外再翻一下airtest.core.api這個(gè)文件里的其他接口
"""
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)</pre>
可見，這些設(shè)備操作的接口都是通過這個(gè)G.DEVICE，所以這里就是我們要找的Airtest與各類被測(cè)設(shè)備交互的實(shí)現(xiàn)部分了谎柄。
先來(lái)看一下這個(gè)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)</pre>
看這個(gè)add_device的注釋丁侄，傳入的dev是初始化之后的設(shè)備對(duì)象，例如安卓朝巫，ios等鸿摇，然后存放在G.DEVICE和添加到G.DEVICE_LIST列表里。既然是初始化劈猿，那么想必就是要在腳本的最前面的執(zhí)行吧拙吉，所以Airtest新建腳本時(shí)自動(dòng)生成的那句auto_setup應(yīng)該就跟設(shè)備初始化有關(guān)系了，一起去看看揪荣。
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</pre>
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</pre>
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</pre>
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</pre>
由上到下的調(diào)用關(guān)系：auto_setup -> connect_device -> init_device -> add_device
auto_setup接口：依次連接全部設(shè)備筷黔，處理日志晒衩，工程根目錄等事物
connect_device接口：根據(jù)傳入?yún)?shù)uri的解析出其平臺(tái)和序列號(hào)信息憎夷，然后初始化設(shè)備
init_device接口：調(diào)用import_device_cls導(dǎo)入不同的平臺(tái)，初始化設(shè)備對(duì)象原杂，如果DEVICE_LIST列表里沒有該設(shè)備揽乱，則添加設(shè)備
add_device接口：將新連接上的設(shè)備賦值給G.DEVICE名眉，添加到G.DEVICE_LIST
所以在Airtest教程中的“4.3 多機(jī)協(xié)作腳本”講到：
在我們的腳本中，支持通過
set_current接口來(lái)切換當(dāng)前連接的手機(jī)凰棉，因此我們一個(gè)腳本中损拢，是能夠調(diào)用多臺(tái)手機(jī)，編寫出一些復(fù)雜的多機(jī)交互腳本的撒犀。
在命令行運(yùn)行腳本時(shí)福压，只需要將手機(jī)依次使用
--device Android:///添加到命令行中即可，例如：
airtest run untitled.air --device Android:///serialno1 --device Android:///serialno2 --device
在之前的筆記里分析過run_script接口解析命令行參數(shù)中的device會(huì)生成成一個(gè)設(shè)備列表或舞，傳入到auto_setup里就會(huì)遍歷列表逐個(gè)去連接荆姆，所以多設(shè)備交互的操作是：
1.初始化連接所有的設(shè)備——命令行或者是調(diào)用run_script傳入多個(gè)設(shè)備，當(dāng)然也可以直接調(diào)用connect_device映凳、add_device;
2.調(diào)用set_current來(lái)切換當(dāng)前操作的設(shè)備胆筒。
set_current接口很簡(jiǎn)單了，在G.DEVICE_LIST里找出目標(biāo)設(shè)備诈豌，賦值給G.DEVICE仆救，因?yàn)閷?duì)設(shè)備的操作都是通過G.DEVICE的，所以只要換掉G.DEVICE就完成了設(shè)備的切換矫渔⊥危看下源碼：
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</pre>
關(guān)于Airtest的設(shè)備管理的分析大概就是以上這些了，多設(shè)備的交互很簡(jiǎn)單庙洼，不用在具體的操作方法中指定設(shè)備顿痪，而是只用在中間調(diào)用set_current來(lái)完成切換設(shè)備镊辕，例如切換前是A設(shè)備，那么所有的操作都會(huì)指向A設(shè)備蚁袭，切換后則都指向B設(shè)備征懈，這種設(shè)計(jì)也挺省事的。
接下來(lái)再拿android這部分來(lái)看一下airtest是怎么跟設(shè)備交互的揩悄。
從import_device_cls接口里找進(jìn)去
'elif platform == "android": from airtest.core.android.android import Android as cls'
android平臺(tái)的設(shè)備管理在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()</pre>
Android是安卓設(shè)備類受裹，父類是Device,這是一個(gè)基類，只定義了設(shè)備通用接口虏束。android設(shè)備初始化，初始化adb厦章，初始化minicap镇匀、javacap、minitouch袜啃、yosemite汗侵、recorder等組件。
翻一下Android類的接口群发，全都是對(duì)安卓設(shè)備的操作晰韵，基本的一些操作是通過adb完成的，比如：?jiǎn)?dòng)應(yīng)用熟妓，卸載應(yīng)用雪猪，喚醒...
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()</pre>
還有就是用到了其他組件的操作了，比如截圖用到minicap和javacap組件起愈，截圖有四種方式：minicap_stream只恨、minicap、javacap抬虽、adb_snapshot官觅，初始化傳入?yún)?shù)可配置截圖的方式，默認(rèn)是MINICAP_STREAM阐污，截圖之后就是寫入休涤，轉(zhuǎn)換成cv2的格式，處理橫豎屏的轉(zhuǎn)換笛辟。
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: # 當(dāng)前是橫屏功氨，但是圖片是豎的，則旋轉(zhuǎn)隘膘，針對(duì)sdk<=16的機(jī)器
screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
# adb 截圖總是要根據(jù)orientation旋轉(zhuǎn)
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</pre>
輸入字符用到y(tǒng)osemite輸入法疑故，在yosemite初始化時(shí)會(huì)往安卓設(shè)備中安裝一個(gè)叫yosemite的輸入法app，并通過adb命令將設(shè)備的當(dāng)前輸入法切換成yosemite弯菊，yosemite輸入法app有個(gè)廣播接收器纵势，接收到廣播后輸入字符踱阿。
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()
# 更多的輸入用法請(qǐng)見 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))</pre>
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</pre>
所以輸入字符的接口也有兩種方式：yosemite輸入法和adb命令，默認(rèn)是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])
# 游戲輸入時(shí)钦铁，輸入有效內(nèi)容后點(diǎn)擊Enter確認(rèn)软舌，如不需要，enter置為False即可牛曹。
if enter:
self.adb.shell(["input", "keyevent", "ENTER"])</pre>
錄屏用到recorder組件佛点，錄屏是用yosemite這個(gè)app實(shí)現(xiàn)的，pythod這邊只是發(fā)adb命令黎比，簡(jiǎn)單的看一下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)</pre>
源碼位置：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</pre>
點(diǎn)擊、滑動(dòng)等用到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)</pre>
minitouch演闭、minicap有啥不同呢，這是openstf的庫(kù)颓帝，大概是在安卓設(shè)備下放了一個(gè)client米碰，pythod這邊用safesocket發(fā)消息給client，由client執(zhí)行操作购城，詳細(xì)的先不在這里分析了吕座。
android設(shè)備類大致就是這樣了，再往下可以看看adb類瘪板，這個(gè)就只看看發(fā)命令的核心接口吧吴趴。
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</pre>
總結(jié)，Airtest的設(shè)備管理只是用G.DEVICE指向當(dāng)前設(shè)備篷帅，用G.DEVICE_LIST保存全部設(shè)備史侣，所有的操作都通過G.DEVICE轉(zhuǎn)發(fā)，所以改變G.DEVICE即可切換設(shè)備魏身。而安卓設(shè)備的交互則是通過adb命令惊橱，和一些別的庫(kù):yosemete、minitouch箭昵、minicap税朴、javacap。
