How to use airtest - 10 common examples

def run_snapshot():
    global serialno
    if not serialno and platform=='android':
        print F
        ds = airtest.getDevices()
        print ds
        if len(ds) == 1:
            serialno=ds[0][0]
        else:
            sys.exit("too many devices, don't know which you want")
    assert serialno != None
    print 'snapshot', serialno
    app = airtest.connect(serialno, device=platform)
    app.globalSet(dict(tmpdir='.'))
    print 'ROTATION:', rotation
    app.globalSet(dict(rotation=rotation))
    app.takeSnapshot('screen.png')
    print 'screenshot save to "screen.png"'
    #if platform == 'android':

def wait(self, imgfile, seconds=20):
        '''
        Wait until some picture exists
        @return position when imgfile shows
        '''
        interval = 1
        max_retry = int(seconds/interval)
        pt = base.wait_until(self.find, args=(imgfile,), interval=interval, max_retry=max_retry)
        if not pt:
            raise RuntimeError('wait fails')
        self._last_point = pt
        return pt

import os
import platform
import time
import threading
import json

import cv2
import aircv as ac

from . import base
from . import proto
from . import patch
from .image import sift as imtsift
from .image import template as imttemplate

log = base.getLogger('devsuit')

class DeviceSuit(object):
    def __init__(self, devtype, dev, logfile='log/airtest.log'):
        # print 'DEVSUIT_SERIALNO:', phoneno
        self.dev = dev
        # self.appname = appname
        self._devtype = devtype
        self._inside_depth = 0

        # default image search extentension and 
        self._image_exts = ['.jpg', '.png']
        self._image_dirs = ['.', 'image']

        self._rotation = None # 0,1,2,3
        self._tmpdir = 'tmp'
        self._click_timeout = 20.0 # if icon not found in this time, then panic

def getCpu(serialno, package):
    '''
    @param package(string): android package name
    @return float: the cpu usage
    '''
    command = 'adb -s %s shell dumpsys cpuinfo' % serialno
    cpu_info = base.check_output(command).splitlines()
    try:
        xym_cpu = filter(lambda x: package in x, cpu_info)[0].split()[0]
        cpu = float(xym_cpu[:-1])
        log.info("cpu_info:%s" % cpu)
        return cpu
    except IndexError:
        log.error("cpu_info error")
        return 0

'''

import json
import os
import re
import subprocess
import string
import StringIO
from functools import partial
from .adb.adbclient import AdbClient

from .. import patch, base
from .. import proto

DEBUG = os.getenv("AIRDEBUG")=="true"
log = base.getLogger('android')

__dir__ = os.path.dirname(os.path.abspath(__file__))

def str2any(s):
    if s.isdigit():
        return int(s)
    if s.lower() == 'true' or s.lower() == 'false':
        return s.lower() == 'true'
    return s

class Monitor(object):
    def __init__(self, serialno, pkgname):
        self._sno = serialno 
        self._pkg = pkgname
        def _adb(*args):
            return subprocess.check_output(['adb', '-s', self._sno] + list(args))

def ConnectRemoteADB(self):
		sText = self.TextAddr.text()
		sCmd = sText.split()[-2]
		sAddr = sText.split()[-1]
		if sCmd == 'connect':
			ADB(serialno=sAddr)
		elif sCmd == 'disconnect':
			ADB(serialno=sAddr).disconnect()
		else:
			QtWidgets.QMessageBox.information(self, "提示", self.tr("请输入正确指令！（connect or disconnect）"))
		self.RefreshADB()

p = self.adb.start_shell('CLASSPATH=%s exec app_process /system/bin %s.Recorder --stop-record' % (pkg_path, YOSEMITE_PACKAGE))
        p.wait()
        self.recording_proc = None
        if is_interrupted:
            return
        for line in p.stdout.readlines():
            if line is None:
                break
            if six.PY3:
                line = line.decode("utf-8")
            m = re.match("stop result: Stop ok! File path:(.*\.mp4)", line.strip())
            if m:
                self.recording_file = m.group(1)
                self.adb.pull(self.recording_file, output)
                return True
        raise AirtestError("start_recording first")

package: package name

        Raises:
            AdbShellError: if any adb error occurs
            AirtestError: if package is not found on the device

        Returns:
            path to the package

        """
        try:
            output = self.shell(['pm', 'path', package])
        except AdbShellError:
            output = ""
        if 'package:' not in output:
            raise AirtestError('package not found, output:[%s]' % output)
        return output.split("package:")[1].strip()

'''

import os
import time

from airtest import image
from airtest import patch
from airtest import base
from airtest import jsonlog

from com.dtmilano.android.viewclient import ViewClient 
from com.dtmilano.android.viewclient import adbclient

DEBUG = os.getenv("DEBUG")=="true"
 
log = base.getLogger('android')

def getMem(serialno, package):
    '''
    @param package(string): android package name
    @return float: the memory, unit MB
    '''
    command = 'adb -s %s shell dumpsys meminfo' % serialno
    mem_info = base.check_output(command).splitlines()
    try:
        xym_mem = filter(lambda x: package in x, mem_info)[0].split()[0]
        mem = float(xym_mem) / 1024
        log.info("mem_info:%s" % mem)
        return mem
    except IndexError:
        log.error("mem_info error")
        return 0

    @logwrap
    def find_element_by_xpath(self, xpath):
        """
        Find the web element by xpath.

        Args:
            xpath: find the element by xpath.
        Returns:
            Web element of current page.
        """
        web_element = self.loop_find_element(super(WebChrome, self).find_element_by_xpath, xpath)
        # web_element = super(WebChrome, self).find_element_by_xpath(xpath)
        log_res = self._gen_screen_log(web_element)
        return Element(web_element, log_res)