airtest连接设备黑屏或者运行报错minicap超时问题

前提是各种usb模式都搞完了，连接时黑屏或者一闪而断开连接，或者运行代码提示 mincap time out之类的东西

解决方法,在airtestide连接按钮时，把其中的javacap勾上

纯代码的话在连接时加上连接参数

cap_method=javacap&touch_method=adb

一个完整的python项目例程，里面用到了大部分poco操作以及连接初始，看一遍自然全都会了

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# File : 电商爬虫.py
# Author: DaShenHan&道长-----先苦后甜，任凭晚风拂柳颜------
# Date : 2021-01-10from airtest.core.api import *
from airtest.aircv import *
from airtest.core.android.adb import *
from airtest.core.android.android import *import os
import requests
import base64
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
from html.parser import HTMLParser
from db_connect.sqlite_driver import *
from pprint import pprint
html_parser = HTMLParser()
import datetime
sql3 = init_db(&＃39;result_datas/美菜网商品.db&＃39;)
PACKAGE = "com.meicai.mall" # 美菜网包名
INSTALL_PATH = "app/美菜商城.apk"def re_text(text):text = text.replace(&＃39;\xa5&＃39;,&＃39;￥&＃39;)return html_parser.unescape(text) # html转义字符还原def ocr_login(api_key=&＃39;KbfUHNoabG8Slos64ugqnff4&＃39;,sdk_key=&＃39;28GQ5PNNnwKAXy0BxoBvEHF5xtZs1Alf&＃39;):host = f&＃39;https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id={api_key}&client_secret={sdk_key}&＃39;respOnse= requests.get(host)if response:rep_json = response.json()access_token = rep_json.get(&＃39;access_token&＃39;)print(&＃39;登录成功,获得token:&＃39;,access_token)return access_tokenreturn Nonedef setup_function():# 连接当前设备# HUAWEI honor 10 分辨率：2280*1080# OPPO R15 分辨率：2280*1080# Vivo x21 分辨率：2280*1080# onePlus 5T 分辨率：2160×1080 poco sevices启动不稳定# xiaomi mix2 元素无法识别# onePlus pocoserver无法启动# device = init_device("Android") # 获取设备号adb = ADB()device = Android() # 获取设备号devicesList = adb.devices() # 获取所有设备列表print(&＃39;所有设备列表:&＃39;,devicesList)# connect_device("android:///" + devicesList[1][0]) # 切换手机currentDevice = device.get_default_device()print("现在连接的测试设备:", currentDevice)auto_setup(__file__, logdir=True, devices=[# "android://127.0.0.1:5037/{currentDevice}?cap_method=MINICAP_STREAM&&ori_method=MINICAPORI&&touch_method=MINITOUCH".format(currentDevice=currentDevice),"android://127.0.0.1:5037/{currentDevice}?cap_method=javacap&touch_method=adb".format(currentDevice=currentDevice),])try:device.check_app(PACKAGE) # 检测是否安装了指定的apkprint(&＃39;apk已存在，正在尝试打开...&＃39;)except AirtestError: # 安装应用,是否同意覆盖安装,默认否print(&＃39;检测到未安装指定的apk，正在为你安装，请耐心等待...&＃39;)device.install_app(INSTALL_PATH, False) # 不覆盖安装print(&＃39;apk安装完毕&＃39;)# clear_app(PACKAGE) # 清除数据# uninstall(PACKAGE) # 卸载App# install(INSTALL_PATH) # 安装应用# stop_app(PACKAGE) # 停止应用setup_function() # 连接设备
# api_url = "http://127.0.0.1:8000"
api_url = "http://qianyuan.iask.in:9090"
headers = {&＃39;content-type&＃39;: &＃39;application/x-www-form-urlencoded&＃39;}
poco = AndroidUiautomationPoco(use_airtest_input=True, screenshot_each_action=False)
height = G.DEVICE.display_info[&＃39;height&＃39;]
width = G.DEVICE.display_info[&＃39;width&＃39;]
pixes = [height, width] # 分辨率access_token = ocr_login()if not access_token:raise UserWarning(&＃39;登录失败，程序退出&＃39;)def get_nowtime_hs():dt_ms = datetime.datetime.now().strftime(&＃39;%Y-%m-%d_%H_%M_%S_%f&＃39;) # 含微秒的日期时间，来源 比特量化return dt_msdef ocr_price(base_jpg, access_token,is_high):if is_high:request_url = "https://aip.baidubce.com/rest/2.0/ocr/v1/accurate_basic"else:request_url = "https://aip.baidubce.com/rest/2.0/ocr/v1/general_basic"# with open(&＃39;2815.png.png&＃39;, &＃39;rb&＃39;) as f:# img = f.read()# base_jpg = base64.encodebytes(img)params = {"image": base_jpg}access_token = access_tokenrequest_url = request_url + "?access_token=" + access_tokenheaders = {&＃39;content-type&＃39;: &＃39;application/x-www-form-urlencoded&＃39;}respOnse= requests.post(request_url, data=params, headers=headers)if response:ret = response.json()if ret.get(&＃39;error_msg&＃39;):print(&＃39;百度api接口已达上限，统一返回0&＃39;)price = 0else:result = ret.get(&＃39;words_result&＃39;) or []if len(result) > 0:price = ret.get(&＃39;words_result&＃39;)[0][&＃39;words&＃39;]else:price = &＃39;价格获取失败&＃39;return pricereturn 0def get_price(price_ui,save_name):if "/" in save_name:path = save_name.split(&＃39;/&＃39;)save_path = &＃39;/&＃39;.join(path[:-1])os.makedirs(save_path,exist_ok=True)pos = price_ui.get_position() # 取位置size = price_ui.get_size() # 取尺寸pic_loc = price_ui.get_bounds() # 取边缘pic_loc = [pic_loc[3]*width,pic_loc[0]*height,pic_loc[1]*width,pic_loc[2]*height]pos_value = [pos[0]*width,pos[1]*height]size_value = [size[0]*width,size[1]*width]pic_pos = (pos_value[0] - size_value[0] * 0.5, pos_value[1] - size_value[1] * 0.5, pos_value[0] + size_value[0] * 0.5, pos_value[1] + size_value[1] * 0.5)screen = G.DEVICE.snapshot()local_screen = aircv.crop_image(screen, pic_loc)# local_screen = aircv.crop_image(screen, pic_pos)pil_image = cv2_2_pil(local_screen)pil_image.save(f"{save_name}", quality=99, optimize=True) # 读取截图并识别截图中的文字with open(f&＃39;{save_name}&＃39;, &＃39;rb&＃39;) as f:file = f.read()base_jpg = base64.b64encode(file)encode_jpg = base_jpg.decode()print(&＃39;开始识别价格图片:&＃39;, encode_jpg[-20:-12])price_unit = ocr_price(base_jpg,access_token,False) # 调百度云接口识别文字return encode_jpg,price_unitdef get_category(first=0,secOnd=0,wait_time=3):first_cate = poco(name=&＃39;com.meicai.mall:id/ll_container&＃39;).child(&＃39;com.meicai.mall:id/horizontalScrollView&＃39;).child() # 获取所有大类别。用的时候取第一个first_cate[first].click()first_cate_text = first_cate[first].get_text()first_cate.wait_for_appearance(wait_time)second_cate = poco(name=&＃39;com.meicai.mall:id/lv_second_category&＃39;).offspring(&＃39;com.meicai.mall:id/cate_tv&＃39;)second_cate_text = second_cate[second].get_text()second_cate[second].click()second_cate.wait_for_appearance(wait_time)return first_cate_text,second_cate_textdef swip_ui(ui_ret):ui_pos = ui_ret.get_position()ui_size = ui_ret.get_size()print(ui_pos, ui_size)# pos_detla = ui_pos[1] - 0.27395833333333336 + ui_size[1]*0.5pos_detla = ui_size[1]print(&＃39;需要移动:&＃39;, pos_detla)# ret.swipe(&＃39;up&＃39;)ui_ret.swipe([0, -pos_detla])ui_ret.wait(3)global indexprint(f"下次从{index}开始爬.本节点爬取失败:{ui_ret.get_name()}")def swip_up_one(ui_ret):ui_pos = ui_ret.get_position()ui_size = ui_ret.get_size()ui_ret.swipe([0, -ui_size[1]])time.sleep(1)def swip_up2(ui_ret):ui_size = ui_ret.get_size()ui_ret.swipe([0, -2*ui_size[1]])time.sleep(1)def swip_down_one(ui_ret):ui_size = ui_ret.get_size()ui_ret.swipe([0, ui_size[1]])time.sleep(2) # 下拉刷新def get_now_page(cate_name):global index,count,size_hret_list = poco(name=&＃39;com.meicai.mall:id/lv_goods_list&＃39;).offspring(&＃39;android:id/list&＃39;).offspring(name=&＃39;com.meicai.mall:id/ll_container&＃39;)is_goods = ret_list[0].offspring(&＃39;com.meicai.mall:id/tv_goods_name&＃39;).exists()if len(ret_list) > 2: # 必须三个以上if not is_goods:ret_list = [ret_list[1], ret_list[2]]else:ret_list = [ret_list[0], ret_list[1]]for ret in ret_list:try:is_goods = ret.offspring(&＃39;com.meicai.mall:id/tv_goods_name&＃39;).exists()if not is_goods:print(f"跳过非商品的节点:{ret}")continuenot_Onsale= ret.offspring(&＃39;com.meicai.mall:id/tv_goods_ssu_price&＃39;).exists()name = ret.offspring(&＃39;com.meicai.mall:id/tv_goods_name&＃39;) # 名称ggs = ret.offspring("com.meicai.mall:id/specLabels").children() # 规格复数ggs_texts = [gg.child().get_text() for gg in ggs] # 规格文本guage = ret.offspring("com.meicai.mall:id/tv_goods_multi_gauge") # 瓜葛guage = guage if guage.exists() else ret.offspring(&＃39;com.meicai.mall:id/tv_goods_ssu_unitprice&＃39;)guage_text = guage.get_text() if guage.exists() else "" # 瓜葛文本guage_text = re_text(guage_text)price = ret.offspring(&＃39;com.meicai.mall:id/tv_goods_price&＃39;) if not not_onsale else ret.offspring(&＃39;com.meicai.mall:id/tv_goods_ssu_price&＃39;) # 价格图片price_text = get_price(price, f&＃39;images/{get_nowtime_hs()}.png&＃39;) # 价格文字 图片编码,文字discount = ret.offspring(&＃39;com.meicai.mall:id/tvDiscountsPrices&＃39;) # 折扣discount_text = re_text(discount.get_text()) if discount.exists() else "" # 折扣文本# 销售信息 自营 券满减if not_onsale:sale_info = ret.offspring(&＃39;com.meicai.mall:id/rl_price_container&＃39;).child(&＃39;com.meicai.mall:id/ll_goods_promote_tag&＃39;).child(&＃39;android.widget.LinearLayout&＃39;)else:sale_info = ret.offspring(&＃39;com.meicai.mall:id/ll_price_container&＃39;).child(&＃39;com.meicai.mall:id/ll_goods_promote_tag&＃39;).child(&＃39;android.widget.LinearLayout&＃39;)sale_texts = [sale.child().get_text() for sale in sale_info] # 销售信息文本goods_info = {"seq": f&＃39;{index}&＃39;,&＃39;cate_name&＃39;:cate_name,"name": name.get_text(),"gg": &＃39;&&＃39;.join(ggs_texts),&＃39;guage&＃39;: guage_text,&＃39;discount&＃39;: discount_text,"price": price_text[1],&＃39;sale_info&＃39;: &＃39;&&＃39;.join(sale_texts),}sql3.save(goods_info=goods_info) # 保存到数据库index += 1print("-----------一条完整商品信息为--------------")print(goods_info) # 打印构造的信息# req = requests.post(api_url, data={"mc": goods_info[&＃39;name&＃39;], "gg": goods_info[&＃39;gg&＃39;], "tp": goods_info[&＃39;price&＃39;][1]},headers=headers)# print(req.text)except Exception as e:name = ret.offspring(&＃39;com.meicai.mall:id/tv_goods_name&＃39;)if name.exists():print(name.get_text(),f"爬取失败，开始向上滑动\n{e}")else:print(ret, f"爬取失败，开始向上滑动\n{e}")swip_up_one(ret)swip_up2(ret_list[-1]) # 按列表中最后一个节点向上滑动两格def main(number=300):global index,countindex = count = 1# first, secOnd= get_category()# cate_name = f&＃39;{first}/{second}&＃39;cate_name = f&＃39;未定义&＃39;print(f&＃39;正在爬取 {cate_name} 下所有商品...&＃39;)ui_root = poco(name=&＃39;com.meicai.mall:id/lv_goods_list&＃39;).offspring(&＃39;android:id/list&＃39;).offspring(name=&＃39;com.meicai.mall:id/ll_container&＃39;)global size_h # 全局变量每个菜看板高度size_h = ui_root.get_size()[1]# swip_down_one(ui_root) # 下拉刷新num = 0while num