python环境：
python3.8
1: 安装beeware
beeware教程：https://docs.beeware.org/en/latest/tutorial/tutorial-2.html
2: 安装python模块uiautomator2
3：测试代码
代码结构
app.py
"""
My first application
"""
from toga.style import Pack
from .dy_dianzan import AndroidTest
from toga.style.pack import COLUMN, ROW
import toga, os, logging
from .u2_server.init import Initer
class dy(toga.App):
def startup(self):
"""
Construct and show the Toga application.
Usually, you would add your application to a main content box.
We then create a main window (with a name matching the app), and
show the main window.
"""
main_box = toga.Box(style=Pack(direction=COLUMN))
log_dir = os.path.dirname(os.path.abspath(__file__))
name_label = toga.Label(
log_dir,
style=Pack(padding=(0, 5))
)
btn_init = toga.Button(
"(首次安装)初始化服务....",
on_press=self.uiautomator_server,
style=Pack(padding=5)
)
button = toga.Button(
"自动打开ATX应用",
on_press=self.say_hello,
style=Pack(padding=5)
)
main_box.add(btn_init)
main_box.add(button)
main_box.add(name_label)
self.main_window = toga.MainWindow(title=self.formal_name)
self.main_window.content = main_box
self.main_window.show()
def say_hello(self, widget):
try:
print("1111")
# 链接uiautomator2 server
my_android = AndroidTest()
my_android.connection_by_wifi("http://127.0.0.1:7912")
# 打开ATX应用
my_android.open_app_by_pkg_name("com.github.uiautomator")
except Exception as e:
print("error: %s" % str(e))
def uiautomator_server(self, widget):
"""
首次使用，初始化uiautomator-server服务
:param widget:
:return:
"""
logger = init_logging("install.log")
logger.info("Install minicap, minitouch")
init = Initer()
init.install()
# result = init.check_install()
pass
def init_logging(log_name):
# log_dir = "/data/local/"
log_dir = os.path.dirname(os.path.abspath(__file__))
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_path = os.path.join(log_dir, log_name)
logger = logging.getLogger()
logger.setLevel(level=logging.DEBUG) # default log level
handler = logging.FileHandler(log_path)
handler.setLevel(logging.DEBUG) # 创建一个handler，用于写入日志文件
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def main():
return dy()
dy_dianzan.py
import uiautomator2 as u2
class AndroidTest(object):
device = None # 已连接的手机对象
# wait_timeout = 300.0 # 链接超时时间
def _get_pkg_name(self):
"""获取当前打开的app的packag_name和activity信息"""
pkg_name = self.device.app_current()
return pkg_name
def connection_by_wifi(self, ip):
"""通过同局域网的wifi链接手机"""
self.device = u2.connect_wifi(ip)
print(self.device.device_info)
def connection_by_num(self, num):
"""通过usb链接手机后通过手机序列号连接手机"""
self.device = u2.connect(num)
# self.device.wait_timeout = self.wait_timeout
print(self.device.device_info)
def open_app_by_pkg_name(self, pkg_name=""):
"""通过pkg名打开app"""
if not pkg_name:
pkg_name = self._get_pkg_name()
self.device.app_clear(pkg_name)
self.device.app_start(pkg_name)
if __name__ == "__main__":
# my_android = AndroidTest()
# my_android.connection_by_num("")
#
# sg = u"/Users/wb-fcj414969/PycharmProjects/douyin/pkg/app-debug.apk"
# my_android.device.push(sg, "/data/")
# my_android.install_app("/data/app-debug.apk")
#
# my_android.open_app("ATX")
pass
u2_server/init.py
# coding: utf-8
import datetime
import os
import shutil
import tarfile
import re
import typing
import progress.bar
import requests
import logging
import hashlib
from subprocess import Popen, PIPE, STDOUT
from retry import retry
from dataclasses import dataclass, asdict
from uiautomator2.version import (__apk_version__, __atx_agent_version__,
__jar_version__, __version__)
from uiautomator2.utils import natualsize
# appdir = os.path.join(os.path.expanduser("~"), '.uiautomator2')
GITHUB_BASEURL = "https://github.com/openatx"
tmp_dir = os.path.dirname(os.path.abspath(__file__))
def init_logging(log_name):
log_dir = os.path.join(tmp_dir, "logs")
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_path = os.path.join(log_dir, log_name)
logger = logging.getLogger()
logger.setLevel(level=logging.DEBUG) #default log level
handler = logging.FileHandler(log_path)
handler.setLevel(logging.DEBUG) #创建一个handler，用于写入日志文件
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
@dataclass
class AppInfo:
package_name: str
version_name: typing.Optional[str]
version_code: typing.Optional[int]
flags: str
first_install_time: datetime.datetime
last_update_time: datetime.datetime
signature: str
path: str
class DownloadBar(progress.bar.PixelBar):
message = "Downloading"
suffix = '%(current_size)s/%(total_size)s'
width = 10
@property
def total_size(self):
return natualsize(self.max)
@property
def current_size(self):
return natualsize(self.index)
def gen_cachepath(url: str) -> str:
filename = os.path.basename(url)
storepath = os.path.join(
tmp_dir, "cache",
filename.replace(" ", "_") + "-" +
hashlib.sha224(url.encode()).hexdigest()[:10], filename)
return storepath
def cache_download(url, filename=None, timeout=None, storepath=None, logger=None):
""" return downloaded filepath """
# check cache
if not filename:
filename = os.path.basename(url)
if not storepath:
storepath = gen_cachepath(url)
storedir = os.path.dirname(storepath)
if not os.path.isdir(storedir):
os.makedirs(storedir)
if logger: logger.info("cache_download:%s storepath:%s" % (url, storepath))
if os.path.exists(storepath) and os.path.getsize(storepath) > 0:
if logger: logger.info("Use cached assets: %s", storepath)
return storepath
if logger: logger.info("Download %s", url)
# download from url
headers = {
'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2',
'Connection': 'keep-alive',
'Origin': 'https://github.com',
'User-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/73.0.3683.86 Safari/537.36'
} # yapf: disable
r = requests.get(url, stream=True, headers=headers, timeout=None)
r.raise_for_status()
file_size = int(r.headers.get("Content-Length"))
bar = DownloadBar(filename, max=file_size)
with open(storepath + '.part', 'wb') as f:
chunk_length = 16 * 1024
while 1:
buf = r.raw.read(chunk_length)
if not buf:
break
f.write(buf)
bar.next(len(buf))
bar.finish()
assert file_size == os.path.getsize(storepath + ".part") # may raise FileNotFoundError
move_path = os.path.join(tmp_dir, os.path.basename(storepath))
shutil.move(storepath + '.part', move_path)
if logger: logger.info("move: %s->%s" % (storepath + '.part', move_path))
return move_path
def mirror_download(url: str, filename=None, logger=None):
"""
Download from mirror, then fallback to origin url
"""
# storepath = "/data/local/tmp/"
storepath = gen_cachepath(url)
if not filename:
filename = os.path.basename(url)
github_host = "https://github.com"
if url.startswith(github_host):
mirror_url = "https://tool.appetizer.io" + url[len(
github_host):] # mirror of github
try:
return cache_download(mirror_url,
filename,
timeout=60,
storepath=storepath, logger=logger)
except (requests.RequestException, FileNotFoundError,
AssertionError) as e:
if logger: logger.info("download error from mirror(%s), use origin source", e)
return cache_download(url, filename, storepath=storepath, logger=logger)
def app_uiautomator_apk_urls():
ret = []
for name in ["app-uiautomator.apk", "app-uiautomator-test.apk"]:
ret.append((name, "".join([
GITHUB_BASEURL, "/android-uiautomator-server/releases/download/",
__apk_version__, "/", name
])))
return ret
class Initer():
def __init__(self):
pass
logger = init_logging("install.log")
self.logger = logger
self.sdk = self.getprop('ro.build.version.sdk')
self.logger.info("sdk: %s" % str(self.sdk))
self.abi = self.getprop('ro.product.cpu.abi')
self.logger.info("abi：%s " % str(self.abi))
self.pre = self.getprop('ro.build.version.preview_sdk')
self.logger.info("pre: %s" % str(self.pre))
self.arch = self.getprop('ro.arch')
self.logger.info("arch:%s" % str(self.arch))
abi_list = self.getprop('ro.product.cpu.abilist')
self.logger.info("abi_list:%s" % abi_list)
self.abis = (self.getprop('ro.product.cpu.abilist').strip() or self.abi).split(",")
# self.logger.info("abis: %s" % ",".join(self.abis))
self.__atx_listen_addr = "127.0.0.1:7912"
self.logger.info("end init")
# self.logger.debug("Initial device %s", device)
# print("uiautomator2 version: %s", __version__)
def getprop(self, prop: str) -> str:
return self.shell(['getprop', prop])[0].strip()
@property
def atx_agent_path(self):
atx_agent_path = os.path.join(tmp_dir, "atx-agent")
if not os.path.exists(atx_agent_path):
os.makedirs(atx_agent_path)
return atx_agent_path
def shell(self, commands: list) -> list:
u"""
执行shell命令
:param commands: shell 命令
:return:
"""
self.logger.info("shell:%s" % " ".join(commands))
p = Popen(commands, shell=False, stderr=STDOUT, stdout=PIPE)
result = []
for line in iter(p.stdout.readline, b''):
self.logger.info("shell stdout->%s" % line.decode().strip())
result.append(line.decode().strip())
p.stdout.close()
try:
p.wait(timeout=120)
except Exception as e:
self.logger.error(str(e))
p.kill()
if not result: result.append("")
return result
# @property
# def jar_urls(self):
# """
# Returns:
# iter([name, url], [name, url])
# """
# for name in ['bundle.jar', 'uiautomator-stub.jar']:
# yield (name, "".join([
# GITHUB_BASEURL,
# "/android-uiautomator-jsonrpcserver/releases/download/",
# __jar_version__, "/", name
# ]))
#
@property
def atx_agent_url(self):
files = {
'armeabi-v7a': 'atx-agent_{v}_linux_armv7.tar.gz',
'arm64-v8a': 'atx-agent_{v}_linux_arm64.tar.gz',
'armeabi': 'atx-agent_{v}_linux_armv6.tar.gz',
'x86': 'atx-agent_{v}_linux_386.tar.gz',
'x86_64': 'atx-agent_{v}_linux_386.tar.gz',
}
name = None
for abi in self.abis:
name = files.get(abi)
if name:
break
if not name:
raise Exception(
"arch(%s) need to be supported yet, please report an issue in github"
% self.abis)
return GITHUB_BASEURL + '/atx-agent/releases/download/%s/%s' % (
__atx_agent_version__, name.format(v=__atx_agent_version__))
@property
def minicap_urls(self):
"""
binary from https://github.com/openatx/stf-binaries
only got abi: armeabi-v7a and arm64-v8a
"""
base_url = GITHUB_BASEURL + \
"/stf-binaries/raw/0.3.0/node_modules/@devicefarmer/minicap-prebuilt/prebuilt/"
sdk = self.sdk
yield base_url + self.abi + "/lib/android-" + sdk + "/minicap.so"
yield base_url + self.abi + "/bin/minicap"
@property
def minitouch_url(self):
self.logger.info(''.join([
GITHUB_BASEURL + "/stf-binaries",
"/raw/0.3.0/node_modules/@devicefarmer/minitouch-prebuilt/prebuilt/",
str(self.abi) + "/bin/minitouch"]))
return ''.join([
GITHUB_BASEURL + "/stf-binaries",
"/raw/0.3.0/node_modules/@devicefarmer/minitouch-prebuilt/prebuilt/",
self.abi + "/bin/minitouch"
])
@retry(tries=2)
def push_url(self, url, dest=None, mode=0o755, tgz=False, extract_name=None):
# yapf: disable
self.logger.info("push_url %s %s" % (url, dest))
path = mirror_download(url, filename=os.path.basename(url), logger=self.logger)
if tgz:
tar = tarfile.open(path, 'r:gz')
path = os.path.join(os.path.dirname(path), extract_name)
tar.extract(extract_name,
os.path.dirname(path)) # zlib.error may raise
self.logger.info("path %s" % path)
return path
def app_info(self, package_name: str) -> typing.Optional[AppInfo]:
"""
Get app info
Returns:
None or AppInfo
"""
output = self.shell(['pm', 'path', package_name])[0]
if "package:" not in output:
return None
apk_path = output.split(":", 1)[-1].strip()
output = self.shell(['dumpsys', 'package', package_name])
m = re.compile(r'versionName=(?P<name>[^\s]+)').search(output)
version_name = m.group('name') if m else ""
if version_name == "null": # Java dumps "null" for null values
version_name = None
m = re.compile(r'versionCode=(?P<code>\d+)').search(output)
version_code = m.group('code') if m else ""
version_code = int(version_code) if version_code.isdigit() else None
m = re.search(r'PackageSignatures\{.*?\[(.*)\]\}', output)
signature = m.group(1) if m else None
if not version_name and signature is None:
return None
m = re.compile(r"pkgFlags=\[\s*(.*)\s*\]").search(output)
pkgflags = m.group(1) if m else ""
pkgflags = pkgflags.split()
time_regex = r"[-\d]+\s+[:\d]+"
m = re.compile(f"firstInstallTime=({time_regex})").search(output)
first_install_time = datetime.datetime.strptime(
m.group(1), "%Y-%m-%d %H:%M:%S") if m else None
m = re.compile(f"lastUpdateTime=({time_regex})").search(output)
last_update_time = datetime.datetime.strptime(
m.group(1).strip(), "%Y-%m-%d %H:%M:%S") if m else None
app_info = AppInfo(package_name=package_name,
version_name=version_name,
version_code=version_code,
flags=pkgflags,
first_install_time=first_install_time,
last_update_time=last_update_time,
signature=signature,
path=apk_path)
return app_info
def package_info(self, package_name: str) -> typing.Union[dict, None]:
"""
version_code might be empty
Returns:
None or dict(version_name, version_code, signature)
"""
self.logger.info("package_info")
app_info = self.app_info(package_name)
if app_info is None:
return app_info
return asdict(app_info)
def is_apk_outdated(self):
"""
If apk signature mismatch, the uiautomator test will fail to start
command: am instrument -w -r -e debug false \
-e class com.github.uiautomator.stub.Stub \
com.github.uiautomator.test/android.support.test.runner.AndroidJUnitRunner
java.lang.SecurityException: Permission Denial: \
starting instrumentation ComponentInfo{com.github.uiautomator.test/android.support.test.runner.AndroidJUnitRunner} \
from pid=7877, uid=7877 not allowed \
because package com.github.uiautomator.test does not have a signature matching the target com.github.uiautomator
"""
apk_debug = self.package_info("com.github.uiautomator")
apk_debug_test = self.package_info("com.github.uiautomator.test")
self.logger.info("apk-debug package-info: %s", apk_debug)
self.logger.info("apk-debug-test package-info: %s", apk_debug_test)
if not apk_debug or not apk_debug_test:
return True
if apk_debug['version_name'] != __apk_version__:
self.logger.info("package com.github.uiautomator version %s, latest %s", apk_debug['version_name'],
__apk_version__)
return True
if apk_debug['signature'] != apk_debug_test['signature']:
# On vivo-Y67 signature might not same, but signature matched.
# So here need to check first_install_time again
max_delta = datetime.timedelta(minutes=3)
if abs(apk_debug['first_install_time'] -
apk_debug_test['first_install_time']) > max_delta:
self.logger.info("package com.github.uiautomator does not have a signature matching the target "
"com.github.uiautomator")
return True
return False
def is_atx_agent_outdated(self):
"""
Returns:
bool
"""
agent_version = self.shell([self.atx_agent_path, "version"])[0]
self.logger.info("atx agent version %s" % agent_version)
if agent_version == "dev":
self.logger.info("skip version check for atx-agent dev")
return False
# semver major.minor.patch
try:
real_ver = list(map(int, agent_version.split(".")))
want_ver = list(map(int, __atx_agent_version__.split(".")))
except ValueError:
return True
self.logger.info("Real version: %s, Expect version: %s", real_ver, want_ver)
if real_ver[:2] != want_ver[:2]:
return True
return real_ver[2] < want_ver[2]
def _install_uiautomator_apks(self):
""" use uiautomator 2.0 to run uiautomator test
通常在连接USB数据线的情况下调用
"""
self.logger.info("pm uninstall com.github.uiautomator")
self.shell(["pm", "uninstall", "com.github.uiautomator"])
self.logger.info("pm uninstall com.github.uiautomator.test")
self.shell(["pm", "uninstall", "com.github.uiautomator.test"])
self.logger.info("pm install com.github.uiautomator, com.github.uiautomator")
for filename, url in app_uiautomator_apk_urls():
path = self.push_url(url, mode=0o644)
self.shell(["pm", "install", "-r", "-t", path])
self.logger.info(" ".join(["pm", "install", "-r", "-t", path]))
def setup_atx_agent(self):
# stop atx-agent first
self.push_url(self.atx_agent_url, tgz=True, extract_name="atx-agent")
# self.shell([self.atx_agent_path, "server", "--stop"])
# if self.is_atx_agent_outdated():
# self.logger.info("Install atx-agent %s", __atx_agent_version__)
# self.push_url(self.atx_agent_url, tgz=True, extract_name="atx-agent")
self.logger.info("strat agent server")
self.shell([self.atx_agent_path, 'server', '--nouia', '-d', "--addr", self.__atx_listen_addr])
self.logger.info("Check atx-agent version")
self.check_atx_agent_version()
@retry(
(requests.ConnectionError, requests.ReadTimeout, requests.HTTPError),
delay=.5,
tries=10)
def check_atx_agent_version(self):
version = requests.get("http://127.0.0.1:7912/version").text.strip()
self.logger.info("atx-agent version %s", version)
wlan_ip = requests.get("http://127.0.0.1:7912/wlan/ip").text.strip()
self.logger.info("device wlan ip: %s", wlan_ip)
return version
def install(self):
"""
TODO: push minicap and minitouch from tgz file
"""
pass
# self.logger.info("Install minicap, minitouch")
# self.push_url(self.minitouch_url)
# if self.abi == "x86":
# self.logger.info("abi:x86 not supported well, skip install minicap")
# elif int(self.sdk) > 30:
# self.logger.info("Android R (sdk:30) has no minicap resource")
# else:
# for url in self.minicap_urls:
# self.push_url(url)
# self.logger.info("end download minitouch")
# self._install_jars() # disable jars
self.logger.info("install uiautomator uiautomator.test")
if self.is_apk_outdated():
self.logger.info("Install com.github.uiautomator, com.github.uiautomator.test %s", __apk_version__)
self._install_uiautomator_apks()
else:
self.logger.info("Already installed com.github.uiautomator apks")
self.logger.info("end install uiautomator uiautomator.test")
self.setup_atx_agent()
self.logger.info("Successfully init")
if __name__ == "__main__":
pass
init = Initer()
print(init.install())
标签：info,name,python,GUI,beeware,version,path,logger,self From： https://www.cnblogs.com/fuchenjie/p/17382899.html