Merge pull request #618 from AirtestProject/dev

1.1.2 版本更新

airtest/__init__.py

@@ -0,0 +1 @@
+from airtest.utils.version import __version__

airtest/aircv/aircv.py

@@ -6,27 +6,32 @@
 import numpy as np
 from .error import FileNotExistError
 from six import PY2, PY3
-from airtest.aircv.utils import cv2_2_pil
+from airtest.aircv.utils import cv2_2_pil, compress_image
 
 
-def imread(filename):
+def imread(filename, flatten=False):
     """根据图片路径，将图片读取为cv2的图片处理格式."""
     if not os.path.isfile(filename):
         raise FileNotExistError("File not exist: %s" % filename)
+
+    # choose image readin mode: cv2.IMREAD_UNCHANGED=-1, cv2.IMREAD_GRAYSCALE=0, cv2.IMREAD_COLOR=1,
+    readin_mode = cv2.IMREAD_GRAYSCALE if flatten else cv2.IMREAD_COLOR
+
     if PY3:
-        img = cv2.imdecode(np.fromfile(filename, dtype=np.uint8), cv2.IMREAD_UNCHANGED)
+        img = cv2.imdecode(np.fromfile(filename, dtype=np.uint8), readin_mode)
     else:
         filename = filename.encode(sys.getfilesystemencoding())
-        img = cv2.imread(filename, 1)
+        img = cv2.imread(filename, readin_mode)
+
     return img
 
 
-def imwrite(filename, img):
+def imwrite(filename, img, quality=10):
     """写出图片到本地路径，压缩"""
     if PY2:
         filename = filename.encode(sys.getfilesystemencoding())
     pil_img = cv2_2_pil(img)
-    pil_img.save(filename, quality=10, optimize=True)
+    compress_image(pil_img, filename, quality, 1200, 1200)
 
 
 def show(img, title="show_img", test_flag=False):

airtest/aircv/error.py

@@ -51,4 +51,3 @@ class NoMatchPointError(BaseError):
 
 class MatchResultCheckError(BaseError):
     """Exception raised for errors 0 keypoint found in the input images."""
-

airtest/aircv/utils.py

@@ -70,9 +70,12 @@ def cv2_2_pil(cv2_image):
     return pil_im
 
 
-def compress_image(pil_img, path, w=300, h=300):
+def compress_image(pil_img, path, quality, max_width=300, max_height=300):
     '''
         生成缩略图
     '''
-    pil_img.thumbnail((w, h))
-    pil_img.save(path, quality=30)
+    pil_img.thumbnail((max_width, max_height), Image.ANTIALIAS)
+    quality = int(round(quality))
+    if quality <= 0 or quality >= 100:
+        raise Exception("SNAPSHOT_QUALITY (" + str(quality) + ") should be an integer in the range [1,99]")
+    pil_img.save(path, quality=quality, optimize=True)

airtest/cli/__main__.py

@@ -14,6 +14,9 @@ def main(argv=None):
     elif args.action == "run":
         from airtest.cli.runner import run_script
         run_script(args)
+    elif args.action == "version":
+        from airtest.utils.version import show_version
+        show_version()
     else:
         ap.print_help()
 

airtest/cli/parser.py

@@ -7,7 +7,9 @@
 
 def get_parser():
     ap = argparse.ArgumentParser()
-    subparsers = ap.add_subparsers(dest="action", help="run/info/report")
+    subparsers = ap.add_subparsers(dest="action", help="version/run/info/report")
+    # subparser version
+    subparsers.add_parser("version", help="show version and exit")
     # subparser run
     ap_run = subparsers.add_parser("run", help="run script")
     runner_parser(ap_run)
@@ -26,6 +28,7 @@ def runner_parser(ap=None):
     ap.add_argument("script", help="air path")
     ap.add_argument("--device", help="connect dev by uri string, e.g. Android:///", nargs="?", action="append")
     ap.add_argument("--log", help="set log dir, default to be script dir", nargs="?", const=True)
+    ap.add_argument("--compress", required=False, type=int, choices=range(1, 100), help="set snapshot quality, 1-99", default=10)
     ap.add_argument("--recording", help="record screen when running", nargs="?", const=True)
     return ap
 

airtest/cli/runner.py

@@ -127,7 +127,7 @@ def setup_by_args(args):
     # guess project_root to be basedir of current .air path
     project_root = os.path.dirname(args.script) if not ST.PROJECT_ROOT else None
 
-    auto_setup(dirpath, devices, args.log, project_root)
+    auto_setup(dirpath, devices, args.log, project_root, args.compress)
 
 
 def run_script(parsed_args, testcase_cls=AirtestCase):

airtest/core/android/adb.py

@@ -9,14 +9,14 @@
 import subprocess
 import threading
 
-from six import PY3, text_type, binary_type
+from six import PY3, text_type, binary_type, raise_from
 from six.moves import reduce
 
 from airtest.core.android.constant import (DEFAULT_ADB_PATH, IP_PATTERN,
-                                           SDK_VERISON_NEW)
+                                           SDK_VERISON_ANDROID7)
 from airtest.core.error import (AdbError, AdbShellError, AirtestError,
                                 DeviceConnectionError)
-from airtest.utils.compat import decode_path
+from airtest.utils.compat import decode_path, raisefrom, proc_communicate_timeout, SUBPROCESS_FLAG
 from airtest.utils.logger import get_logger
 from airtest.utils.nbsp import NonBlockingStreamReader
 from airtest.utils.retry import retries
@@ -33,9 +33,11 @@ class ADB(object):
     status_offline = "offline"
     SHELL_ENCODING = "utf-8"
 
-    def __init__(self, serialno=None, adb_path=None, server_addr=None):
+    def __init__(self, serialno=None, adb_path=None, server_addr=None, display_id=None, input_event=None):
         self.serialno = serialno
         self.adb_path = adb_path or self.builtin_adb_path()
+        self.display_id = display_id
+        self.input_event = input_event
         self._set_cmd_options(server_addr)
         self.connect()
         self._sdk_version = None
@@ -148,18 +150,20 @@ def start_cmd(self, cmds, device=True):
             cmds,
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
+            stderr=subprocess.PIPE,
+            creationflags=SUBPROCESS_FLAG
         )
         return proc
 
-    def cmd(self, cmds, device=True, ensure_unicode=True):
+    def cmd(self, cmds, device=True, ensure_unicode=True, timeout=None):
         """
         Run the adb command(s) in subprocess and return the standard output
 
         Args:
             cmds: command(s) to be run
             device: if True, the device serial number must be specified by -s serialno argument
             ensure_unicode: encode/decode unicode of standard outputs (stdout, stderr)
+            timeout: timeout in seconds
 
         Raises:
             DeviceConnectionError: if any error occurs when connecting the device
@@ -170,7 +174,10 @@ def cmd(self, cmds, device=True, ensure_unicode=True):
 
         """
         proc = self.start_cmd(cmds, device)
-        stdout, stderr = proc.communicate()
+        if timeout:
+            stdout, stderr = proc_communicate_timeout(proc, timeout)
+        else:
+            stdout, stderr = proc.communicate()
 
         if ensure_unicode:
             stdout = stdout.decode(get_std_encoding(sys.stdout))
@@ -287,14 +294,10 @@ def wait_for_device(self, timeout=5):
             None
 
         """
-        proc = self.start_cmd("wait-for-device")
-        timer = threading.Timer(timeout, proc.kill)
-        timer.start()
-        ret = proc.wait()
-        if ret == 0:
-            timer.cancel()
-        else:
-            raise DeviceConnectionError("device not ready")
+        try:
+            self.cmd("wait-for-device", timeout=timeout)
+        except RuntimeError as e:
+            raisefrom(DeviceConnectionError, "device not ready", e)
 
     def start_shell(self, cmds):
         """
@@ -347,7 +350,7 @@ def shell(self, cmd):
             command output
 
         """
-        if self.sdk_version < SDK_VERISON_NEW:
+        if self.sdk_version < SDK_VERISON_ANDROID7:
             # for sdk_version < 25, adb shell do not raise error
             # https://stackoverflow.com/questions/9379400/adb-error-codes
             cmd = split_cmd(cmd) + [";", "echo", "---$?---"]
@@ -683,7 +686,10 @@ def snapshot(self):
             command output (stdout)
 
         """
-        raw = self.cmd('shell screencap -p', ensure_unicode=False)
+        if self.display_id:
+            raw = self.cmd('shell screencap -d {0} -p'.format(self.display_id), ensure_unicode=False)
+        else:
+            raw = self.cmd('shell screencap -p', ensure_unicode=False)
         return raw.replace(self.line_breaker, b"\n")
 
     # PEP 3113 -- Removal of Tuple Parameter Unpacking
@@ -818,7 +824,7 @@ def line_breaker(self):
 
         """
         if not self._line_breaker:
-            if self.sdk_version >= SDK_VERISON_NEW:
+            if self.sdk_version >= SDK_VERISON_ANDROID7:
                 line_breaker = os.linesep
             else:
                 line_breaker = '\r' + os.linesep
@@ -1475,14 +1481,15 @@ def get_display_of_all_screen(self, info):
         windows = output.split("Window #")
         offsetx, offsety, x, y = info['width'], info['height'], 0, 0
         package = self._search_for_current_package(output)
-        for w in windows:
-            if "package=%s" % package in w:
-                arr = re.findall(r'Frames: containing=\[(\d+\.?\d*),(\d+\.?\d*)]\[(\d+\.?\d*),(\d+\.?\d*)]', w)
-                if len(arr) >= 1 and len(arr[0]) == 4:
-                    offsetx, offsety, x, y = float(arr[0][0]), float(arr[0][1]), float(arr[0][2]), float(arr[0][3])
-                    if info["orientation"] in [1, 3]:
-                        offsetx, offsety, x, y = offsety, offsetx, y, x
-                    x, y = x - offsetx, y - offsety
+        if package:
+            for w in windows:
+                if "package=%s" % package in w:
+                    arr = re.findall(r'Frames: containing=\[(\d+\.?\d*),(\d+\.?\d*)]\[(\d+\.?\d*),(\d+\.?\d*)]', w)
+                    if len(arr) >= 1 and len(arr[0]) == 4:
+                        offsetx, offsety, x, y = float(arr[0][0]), float(arr[0][1]), float(arr[0][2]), float(arr[0][3])
+                        if info["orientation"] in [1, 3]:
+                            offsetx, offsety, x, y = offsety, offsetx, y, x
+                        x, y = x - offsetx, y - offsety
         return {
             "offset_x": offsetx,
             "offset_y": offsety,
@@ -1497,10 +1504,15 @@ def _search_for_current_package(self, ret):
         Returns:
             package name if exists else ""
         """
-        packageRE = re.compile('\s*mCurrentFocus=Window{.* ([A-Za-z0-9_.]+)/[A-Za-z0-9_.]+}')
-        m = packageRE.findall(ret)
-        if m:
-            return m[-1]
+        try:
+            packageRE = re.compile('\s*mCurrentFocus=Window{.* ([A-Za-z0-9_.]+)/[A-Za-z0-9_.]+}')
+            m = packageRE.findall(ret)
+            if m:
+                return m[-1]
+            else:
+                return self.get_top_activity()[0]
+        except Exception as e:
+            print("[Error] Cannot get current top activity")
         return ""
 
 

airtest/core/android/android.py

@@ -8,7 +8,8 @@
 from airtest.utils.logger import get_logger
 from airtest.core.device import Device
 from airtest.core.android.ime import YosemiteIme
-from airtest.core.android.constant import CAP_METHOD, TOUCH_METHOD, IME_METHOD, ORI_METHOD, SDK_VERISON_NEW
+from airtest.core.android.constant import CAP_METHOD, TOUCH_METHOD, IME_METHOD, ORI_METHOD,\
+    SDK_VERISON_ANDROID7, SDK_VERISON_ANDROID10
 from airtest.core.android.adb import ADB
 from airtest.core.android.minicap import Minicap
 from airtest.core.android.minitouch import Minitouch
@@ -27,24 +28,30 @@ def __init__(self, serialno=None, host=None,
                  touch_method=TOUCH_METHOD.MINITOUCH,
                  ime_method=IME_METHOD.YOSEMITEIME,
                  ori_method=ORI_METHOD.MINICAP,
-                 ):
+                 display_id=None,
+                 input_event=None):
         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()
+        self.display_id = display_id
+        self.input_event = input_event
         # init adb
-        self.adb = ADB(self.serialno, server_addr=host)
+        self.adb = ADB(self.serialno, server_addr=host, display_id=self.display_id, input_event=self.input_event)
         self.adb.wait_for_device()
         self.sdk_version = self.adb.sdk_version
+        # Android10 temporary solution, using adbtouch instead of minitouch
+        if self.sdk_version >= SDK_VERISON_ANDROID10 and self.touch_method != TOUCH_METHOD.ADBTOUCH:
+            self.touch_method = TOUCH_METHOD.ADBTOUCH
         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.minicap = Minicap(self.adb, ori_function=self.get_display_info, display_id=self.display_id)
         self.javacap = Javacap(self.adb)
-        self.minitouch = Minitouch(self.adb, ori_function=self.get_display_info)
+        self.minitouch = Minitouch(self.adb, ori_function=self.get_display_info,input_event=self.input_event)
         self.yosemite_ime = YosemiteIme(self.adb)
         self.recorder = Recorder(self.adb)
         self._register_rotation_watcher()
@@ -63,7 +70,12 @@ def get_default_device(self):
 
     @property
     def uuid(self):
-        return self.serialno
+        ult = [self.serialno]
+        if self.display_id:
+            ult.append(self.display_id)
+        if self.input_event:
+            ult.append(self.input_event)
+        return "_".join(ult)
 
     def list_app(self, third_only=False):
         """
@@ -202,13 +214,14 @@ def uninstall_app(self, package):
         """
         return self.adb.uninstall_app(package)
 
-    def snapshot(self, filename=None, ensure_orientation=True):
+    def snapshot(self, filename=None, ensure_orientation=True, quality=10):
         """
         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 is stdout
             ensure_orientation: True or False whether to keep the orientation same as display
+            quality: The image quality, integer in range [1, 99]
 
         Returns:
             screenshot output
@@ -241,10 +254,10 @@ def snapshot(self, filename=None, ensure_orientation=True):
                 if w < h:  # 当前是横屏，但是图片是竖的，则旋转，针对sdk<=16的机器
                     screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
             # adb 截图总是要根据orientation旋转，但是SDK版本大于等于25(Android7.1以后)无需额外旋转
-            elif self.cap_method == CAP_METHOD.ADBCAP and self.sdk_version <= SDK_VERISON_NEW:
+            elif self.cap_method == CAP_METHOD.ADBCAP and self.sdk_version <= SDK_VERISON_ANDROID7:
                 screen = aircv.rotate(screen, self.display_info["orientation"] * 90, clockwise=False)
         if filename:
-            aircv.imwrite(filename, screen)
+            aircv.imwrite(filename, screen, quality)
         return screen
 
     def shell(self, *args, **kwargs):

airtest/core/android/constant.py

@@ -2,6 +2,7 @@
 import os
 import re
 import sys
+import subprocess
 from airtest.utils.compat import decode_path
 
 THISPATH = decode_path(os.path.dirname(os.path.realpath(__file__)))
@@ -14,7 +15,9 @@
     "Linux-armv7l": os.path.join(STATICPATH, "adb", "linux_arm", "adb"),
 }
 DEFAULT_ADB_SERVER = ('127.0.0.1', 5037)
-SDK_VERISON_NEW = 24
+SDK_VERISON_ANDROID7 = 24
+# Android 10 SDK version
+SDK_VERISON_ANDROID10 = 29
 DEBUG = True
 STFLIB = os.path.join(STATICPATH, "stf_libs")
 ROTATIONWATCHER_APK = os.path.join(STATICPATH, "apks", "RotationWatcher.apk")