WindyMadman/AzurLaneAutoScript
1
0
Fork 0
mirror of https://github.com/LmeSzinc/AzurLaneAutoScript.git synced 2025-01-08 12:27:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add: Faster screenshot methods ADB_nc and aScreenCap_nc

Browse Source 
- Allow config names when creating Device instances
- Add "Ultra Fast" level in performance test
This commit is contained in:
LmeSzinc 2022-03-23 21:20:01 +08:00
parent 5951d05f90
commit 5ba2f5c5ac
16 changed files with 254 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
config/template.json
View File

@ -1424,8 +1424,10 @@
"Benchmark": {
"Benchmark": {
"AdbScreenshot": true,
"AdbncScreenshot": true,
"Uiautomator2Screenshot": true,
"AscreencapScreenshot": true,
"AscreencapncScreenshot": true,
"AdbClick": true,
"Uiautomator2Click": true,
"MinitouchClick": true,

12
module/config/argument/args.json
View File

@ -24,8 +24,10 @@
"value": "ADB",
"option": [
"ADB",
"ADB_nc",
"uiautomator2",
"aScreenCap"
"aScreenCap",
"aScreenCap_nc"
]
},
"ControlMethod": {
@ -6386,6 +6388,10 @@
"type": "checkbox",
"value": true
},
"AdbncScreenshot": {
"type": "checkbox",
"value": true
},
"Uiautomator2Screenshot": {
"type": "checkbox",
"value": true
@ -6394,6 +6400,10 @@
"type": "checkbox",
"value": true
},
"AscreencapncScreenshot": {
"type": "checkbox",
"value": true
},
"AdbClick": {
"type": "checkbox",
"value": true

4
module/config/argument/argument.yaml
View File

@ -19,7 +19,7 @@ Emulator:
option: [cn, en, jp, tw]
ScreenshotMethod:
value: ADB
option: [ADB, uiautomator2, aScreenCap]
option: [ADB, ADB_nc, uiautomator2, aScreenCap, aScreenCap_nc]
ControlMethod:
value: minitouch
option: [ADB, uiautomator2, minitouch, Hermit]
@ -494,8 +494,10 @@ OpsiDaemon:
RepairShip: true
Benchmark:
AdbScreenshot: true
AdbncScreenshot: true
Uiautomator2Screenshot: true
AscreencapScreenshot: true
AscreencapncScreenshot: true
AdbClick: true
Uiautomator2Click: true
MinitouchClick: true

4
module/config/config_generated.py
View File

@ -21,7 +21,7 @@ class GeneratedConfig:
Emulator_Serial = '127.0.0.1:5555'
Emulator_PackageName = 'com.bilibili.azurlane'
Emulator_Server = 'cn' # cn, en, jp, tw
Emulator_ScreenshotMethod = 'ADB' # ADB, uiautomator2, aScreenCap
Emulator_ScreenshotMethod = 'ADB' # ADB, ADB_nc, uiautomator2, aScreenCap, aScreenCap_nc
Emulator_ControlMethod = 'minitouch' # ADB, uiautomator2, minitouch, Hermit
Emulator_ScreenshotDedithering = False
@ -359,8 +359,10 @@ class GeneratedConfig:
# Group `Benchmark`
Benchmark_AdbScreenshot = True
Benchmark_AdbncScreenshot = True
Benchmark_Uiautomator2Screenshot = True
Benchmark_AscreencapScreenshot = True
Benchmark_AscreencapncScreenshot = True
Benchmark_AdbClick = True
Benchmark_Uiautomator2Click = True
Benchmark_MinitouchClick = True

1
module/config/config_manual.py
View File

@ -61,6 +61,7 @@ class ManualConfig:
module.device
"""
FORWARD_PORT_RANGE = (20000, 21000)
REVERSE_SERVER_PORT = 7903
ASCREENCAP_FILEPATH_LOCAL = './bin/ascreencap'
ASCREENCAP_FILEPATH_REMOTE = '/data/local/tmp/ascreencap'
MINITOUCH_FILEPATH_REMOTE = '/data/local/tmp/minitouch'

22
module/config/i18n/en-US.json
View File

@ -292,17 +292,19 @@
},
"ScreenshotMethod": {
"name": "Screenshot Method",
"help": "Speed: aScreenCap > uiautomator2 ~= ADB\nADB is recommended",
"help": "Speed: aScreenCap_nc > ADB_nc >>> aScreenCap > uiautomator2 ~= ADB\nRun Tools - Performance Test to find the fastest method",
"ADB": "ADB ",
"uiautomator2": "uiautomator2 ",
"aScreenCap": "aScreenCap "
"ADB_nc": "ADB_nc",
"uiautomator2": "uiautomator2",
"aScreenCap": "aScreenCap",
"aScreenCap_nc": "aScreenCap_nc"
},
"ControlMethod": {
"name": "Control Method",
"help": "Speed: minitouch > Hermit >>> uiautomator2 ~= ADB\nminitouch is recommended, hermit is not recommended unless on vmos",
"ADB": "ADB ",
"uiautomator2": "uiautomator2 ",
"minitouch": "minitouch ",
"ADB": "ADB",
"uiautomator2": "uiautomator2",
"minitouch": "minitouch",
"Hermit": "Hermit"
},
"ScreenshotDedithering": {
@ -1846,6 +1848,10 @@
"name": "Test ADB Screenshot",
"help": ""
},
"AdbncScreenshot": {
"name": "Test ADB_nc Screenshot",
"help": ""
},
"Uiautomator2Screenshot": {
"name": "Test Uiautomator2 Screenshot",
"help": ""
@ -1854,6 +1860,10 @@
"name": "Test aScreenCap Screenshot",
"help": ""
},
"AscreencapncScreenshot": {
"name": "Test aScreenCap_nc Screenshot",
"help": ""
},
"AdbClick": {
"name": "Test ADB Click",
"help": ""

12
module/config/i18n/ja-JP.json
View File

@ -294,8 +294,10 @@
"name": "Emulator.ScreenshotMethod.name",
"help": "Emulator.ScreenshotMethod.help",
"ADB": "ADB",
"ADB_nc": "ADB_nc",
"uiautomator2": "uiautomator2",
"aScreenCap": "aScreenCap"
"aScreenCap": "aScreenCap",
"aScreenCap_nc": "aScreenCap_nc"
},
"ControlMethod": {
"name": "Emulator.ControlMethod.name",
@ -1846,6 +1848,10 @@
"name": "Benchmark.AdbScreenshot.name",
"help": "Benchmark.AdbScreenshot.help"
},
"AdbncScreenshot": {
"name": "Benchmark.AdbncScreenshot.name",
"help": "Benchmark.AdbncScreenshot.help"
},
"Uiautomator2Screenshot": {
"name": "Benchmark.Uiautomator2Screenshot.name",
"help": "Benchmark.Uiautomator2Screenshot.help"
@ -1854,6 +1860,10 @@
"name": "Benchmark.AscreencapScreenshot.name",
"help": "Benchmark.AscreencapScreenshot.help"
},
"AscreencapncScreenshot": {
"name": "Benchmark.AscreencapncScreenshot.name",
"help": "Benchmark.AscreencapncScreenshot.help"
},
"AdbClick": {
"name": "Benchmark.AdbClick.name",
"help": "Benchmark.AdbClick.help"

24
module/config/i18n/zh-CN.json
View File

@ -292,17 +292,19 @@
},
"ScreenshotMethod": {
"name": "模拟器截图方案",
"help": "速度: aScreenCap > uiautomator2 ~= ADB\n建议选 ADB",
"ADB": "ADB ",
"uiautomator2": "uiautomator2 ",
"aScreenCap": "aScreenCap "
"help": "速度: aScreenCap_nc > ADB_nc >>> aScreenCap > uiautomator2 ~= ADB\n运行 工具 - 性能测试 以寻找最快的方案",
"ADB": "ADB",
"ADB_nc": "ADB_nc",
"uiautomator2": "uiautomator2",
"aScreenCap": "aScreenCap",
"aScreenCap_nc": "aScreenCap_nc"
},
"ControlMethod": {
"name": "模拟器控制方案",
"help": "速度: minitouch > Hermit >>> uiautomator2 ~= ADB\n建议选minitouch，不建议选Hermit除非在vmos下",
"ADB": "ADB ",
"uiautomator2": "uiautomator2 ",
"minitouch": "minitouch ",
"ADB": "ADB",
"uiautomator2": "uiautomator2",
"minitouch": "minitouch",
"Hermit": "Hermit"
},
"ScreenshotDedithering": {
@ -1846,6 +1848,10 @@
"name": "测试 ADB 截图",
"help": ""
},
"AdbncScreenshot": {
"name": "测试 ADB_nc 截图",
"help": ""
},
"Uiautomator2Screenshot": {
"name": "测试 uiautomator2 截图",
"help": ""
@ -1854,6 +1860,10 @@
"name": "测试 aScreenCap 截图",
"help": ""
},
"AscreencapncScreenshot": {
"name": "测试 aScreenCap_nc 截图",
"help": ""
},
"AdbClick": {
"name": "测试 ADB 点击",
"help": ""

14
module/config/i18n/zh-TW.json
View File

@ -292,10 +292,12 @@
},
"ScreenshotMethod": {
"name": "模擬器截圖方案",
"help": "速度: aScreenCap > uiautomator2 ~= ADB\n建議選 ADB",
"help": "速度: aScreenCap_nc > ADB_nc >>> aScreenCap > uiautomator2 ~= ADB\n運行 工具 - 性能測試 以尋找最快的方案",
"ADB": "ADB",
"ADB_nc": "ADB_nc",
"uiautomator2": "uiautomator2",
"aScreenCap": "aScreenCap"
"aScreenCap": "aScreenCap",
"aScreenCap_nc": "aScreenCap_nc"
},
"ControlMethod": {
"name": "模擬器控制方案",
@ -1846,6 +1848,10 @@
"name": "測試 ADB 截圖",
"help": ""
},
"AdbncScreenshot": {
"name": "測試 ADB 截圖",
"help": "Benchmark.AdbncScreenshot.help"
},
"Uiautomator2Screenshot": {
"name": "測試 uiautomator2 截圖",
"help": ""
@ -1854,6 +1860,10 @@
"name": "測試 aScreenCap 截圖",
"help": ""
},
"AscreencapncScreenshot": {
"name": "測試 aScreenCap_nc 截圖",
"help": ""
},
"AdbClick": {
"name": "測試 ADB 點擊",
"help": ""

6
module/daemon/benchmark.py
View File

@ -68,6 +68,8 @@ class Benchmark(DaemonBase, CampaignUI):
if not isinstance(cost, (float, int)):
return Text(cost, style="bold bright_red")
if cost < 0.12:
return Text('Ultra Fast', style="bold bright_green")
if cost < 0.25:
return Text('Very Fast', style="bright_green")
if cost < 0.45:
@ -132,10 +134,14 @@ class Benchmark(DaemonBase, CampaignUI):
data = []
if self.config.Benchmark_AdbScreenshot:
data.append(['ADB', self.benchmark_test(self.device.screenshot_adb)])
if self.config.Benchmark_AdbncScreenshot:
data.append(['ADB_nc', self.benchmark_test(self.device.screenshot_adb_nc)])
if self.config.Benchmark_Uiautomator2Screenshot:
data.append(['uiautomator2', self.benchmark_test(self.device.screenshot_uiautomator2)])
if self.config.Benchmark_AscreencapScreenshot:
data.append(['aScreenCap', self.benchmark_test(self.device.screenshot_ascreencap)])
if self.config.Benchmark_AscreencapncScreenshot:
data.append(['aScreenCap_nc', self.benchmark_test(self.device.screenshot_ascreencap_nc)])
screenshot = data
data = []

96
module/device/connection.py
View File

@ -1,23 +1,27 @@
import logging
import os
import re
import socket
import subprocess
import time
import adbutils
import uiautomator2 as u2
from adbutils import AdbClient, AdbDevice, ForwardItem
from adbutils import AdbClient, AdbDevice, ForwardItem, ReverseItem, AdbTimeout
from deploy.utils import poor_yaml_read, DEPLOY_CONFIG
from module.base.decorator import cached_property
from module.base.utils import ensure_time
from module.config.config import AzurLaneConfig
from module.device.method.utils import possible_reasons, random_port, del_cached_property
from module.device.method.utils import recv_all, possible_reasons, random_port, del_cached_property
from module.exception import RequestHumanTakeover
from module.logger import logger
class Connection:
config: AzurLaneConfig
serial: str
adb_binary_list = [
'./bin/adb/adb.exe',
'./toolkit/Lib/site-packages/adbutils/binaries/adb.exe',
@ -27,10 +31,14 @@ class Connection:
def __init__(self, config):
"""
Args:
config (AzurLaneConfig):
config (AzurLaneConfig, str): Name of the user config under ./config
"""
logger.hr('Device')
self.config = config
if isinstance(config, str):
self.config = AzurLaneConfig(config, task=None)
else:
self.config = config
self.serial = str(self.config.Emulator_Serial)
if "bluestacks4-hyperv" in self.serial:
self.serial = self.find_bluestacks4_hyperv(self.serial)
@ -192,6 +200,50 @@ class Connection:
result = self.adb.shell(cmd, timeout=10, **kwargs)
return result
@cached_property
def reverse_server(self):
"""
Setup a server on Alas, access it from emulator.
This will bypass adb shell and be faster.
"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self._server_port = self.adb_reverse(f'tcp:{self.config.REVERSE_SERVER_PORT}')
server.bind(('127.0.0.1', self._server_port))
server.listen(5)
logger.info(f'Reverse server listening on {self._server_port}')
return server
def adb_shell_nc(self, cmd, timeout=5, chunk_size=262144):
"""
Args:
cmd (list):
timeout (int):
chunk_size (int): Default to 262144
Returns:
bytes:
"""
# <command> | nc 127.0.0.1 {port}
cmd += ['|', 'nc', '127.0.0.1', self.config.REVERSE_SERVER_PORT]
# Server start listening
server = self.reverse_server
server.settimeout(timeout)
# Client send data, waiting for server accept
_ = self.adb_shell(cmd, stream=True)
try:
# Server accept connection
conn, conn_port = server.accept()
except socket.timeout:
raise AdbTimeout('reverse server accept timeout')
# Server receive data
data = recv_all(conn, chunk_size=chunk_size)
# Server close connection
conn.close()
return data
def adb_exec_out(self, cmd, serial=None):
cmd.insert(0, 'exec-out')
return self.adb_command(cmd, serial)
@ -234,6 +286,27 @@ class Connection:
self.adb.forward(forward.local, forward.remote)
return port
def adb_reverse(self, remote):
port = 0
for reverse in self.adb.reverse_list():
if reverse.remote == remote and reverse.local.startswith('tcp:'):
if not port:
logger.info(f'Reuse reverse: {reverse}')
port = int(reverse.local[4:])
else:
logger.info(f'Remove redundant forward: {reverse}')
self.adb_forward_remove(reverse.local)
if port:
return port
else:
# Create new reverse
port = random_port(self.config.FORWARD_PORT_RANGE)
reverse = ReverseItem(f'tcp:{port}', remote)
logger.info(f'Create reverse: {reverse}')
self.adb.reverse(reverse.local, reverse.remote)
return port
def adb_forward_remove(self, local):
"""
Equivalent to `adb -s <serial> forward --remove <local>`
@ -248,6 +321,20 @@ class Connection:
c.send_command(list_cmd)
c.check_okay()
def adb_reverse_remove(self, local):
"""
Equivalent to `adb -s <serial> reverse --remove <local>`
Args:
local (str): Such as 'tcp:2437'
"""
with self.adb_client._connect() as c:
c.send_command(f"host:transport:{self.serial}")
c.check_okay()
list_cmd = f"reverse:killforward:{local}"
c.send_command(list_cmd)
c.check_okay()
def adb_push(self, local, remote):
"""
Args:
@ -297,6 +384,7 @@ class Connection:
del_cached_property(self, 'hermit_session')
del_cached_property(self, 'minitouch_builder')
del_cached_property(self, 'reverse_server')
def install_uiautomator2(self):
"""

29
module/device/method/adb.py
View File

@ -1,4 +1,5 @@
import re
from functools import wraps
import cv2
import numpy as np
@ -6,12 +7,13 @@ from adbutils.errors import AdbError
from lxml import etree
from module.device.connection import Connection
from module.device.method.utils import possible_reasons, handle_adb_error, RETRY_TRIES, RETRY_DELAY
from module.device.method.utils import recv_all, possible_reasons, handle_adb_error, RETRY_TRIES, RETRY_DELAY
from module.exception import ScriptError, RequestHumanTakeover
from module.logger import logger
def retry(func):
@wraps(func)
def retry_wrapper(self, *args, **kwargs):
"""
Args:
@ -101,15 +103,28 @@ class Adb(Connection):
def screenshot_adb(self):
stream = self.adb_shell(['screencap', '-p'], stream=True)
content = b""
while True:
chunk = stream.read(4096)
if not chunk:
break
content += chunk
content = recv_all(stream)
return self.__process_screenshot(content)
@retry
def screenshot_adb_nc(self):
data = self.adb_shell_nc(['screencap'])
if len(data) < 100:
logger.warning(f'Unexpected screenshot: {data}')
# Load data
header = np.frombuffer(data[0:12], dtype=np.uint32)
channel = 4 # screencap sends an RGBA image
width, height, _ = header # Usually to be 1280, 720, 1
image = np.frombuffer(data, dtype=np.uint8)
shape = image.shape[0]
image = image[shape - width * height * channel:].reshape(height, width, channel)
image = cv2.cvtColor(image, cv2.COLOR_BGRA2BGR)
return image
@retry
def click_adb(self, x, y):
self.adb_shell(['input', 'tap', x, y])

35
module/device/method/ascreencap.py
View File

@ -5,7 +5,7 @@ from adbutils.errors import AdbError
from module.base.utils import *
from module.device.connection import Connection
from module.device.method.utils import handle_adb_error, RETRY_TRIES, RETRY_DELAY
from module.device.method.utils import recv_all, handle_adb_error, RETRY_TRIES, RETRY_DELAY
from module.exception import ScriptError, RequestHumanTakeover
from module.logger import logger
@ -106,28 +106,17 @@ class AScreenCap(Connection):
raise AscreencapError(text)
return byte_array[self.__bytepointer:]
def _ascreencap_execute(self):
stream = self.adb_shell([self.config.ASCREENCAP_FILEPATH_REMOTE, '--pack', '2', '--stdout'], stream=True)
content = b""
while True:
chunk = stream.read(4096)
if not chunk:
break
content += chunk
return self.__process_screenshot(content)
def __load_screenshot(self, screenshot, method):
if method == 0:
pass
return screenshot
elif method == 1:
screenshot = screenshot.replace(b'\r\n', b'\n')
return screenshot.replace(b'\r\n', b'\n')
elif method == 2:
screenshot = screenshot.replace(b'\r\r\n', b'\n')
return screenshot.replace(b'\r\r\n', b'\n')
else:
raise ScriptError(f'Unknown method to load screenshots: {method}')
def __uncompress(self, screenshot):
raw_compressed_data = self._ascreencap_reposition_byte_pointer(screenshot)
# See headers in:
@ -158,6 +147,7 @@ class AScreenCap(Connection):
for method in self.__screenshot_method_fixed:
try:
result = self.__load_screenshot(screenshot, method=method)
result = self.__uncompress(result)
self.__screenshot_method_fixed = [method] + self.__screenshot_method
return result
except lz4.block.LZ4BlockError:
@ -173,11 +163,12 @@ class AScreenCap(Connection):
def screenshot_ascreencap(self):
stream = self.adb_shell([self.config.ASCREENCAP_FILEPATH_REMOTE, '--pack', '2', '--stdout'], stream=True)
content = b""
while True:
chunk = stream.read(4096)
if not chunk:
break
content += chunk
content = recv_all(stream)
return self.__process_screenshot(content)
@retry
def screenshot_ascreencap_nc(self):
data = self.adb_shell_nc([self.config.ASCREENCAP_FILEPATH_REMOTE, '--pack', '2', '--stdout'])
return self.__uncompress(data)

2
module/device/method/hermit.py
View File

@ -7,7 +7,7 @@ from module.base.decorator import cached_property
from module.base.timer import Timer
from module.base.utils import random_rectangle_point, point2str
from module.device.method.adb import Adb
from module.device.method.utils import HierarchyButton, handle_adb_error, del_cached_property, RETRY_TRIES, RETRY_DELAY
from module.device.method.utils import HierarchyButton, handle_adb_error, RETRY_TRIES, RETRY_DELAY
from module.exception import RequestHumanTakeover
from module.logger import logger

29
module/device/method/utils.py
View File

@ -4,6 +4,7 @@ import socket
import uiautomator2 as u2
from lxml import etree
from adbutils import _AdbStreamConnection, AdbTimeout
from module.base.decorator import cached_property
from module.logger import logger
@ -34,6 +35,34 @@ def random_port(port_range):
return new_port
def recv_all(stream, chunk_size=4096) -> bytes:
"""
Args:
stream:
chunk_size:
Returns:
bytes:
Raises:
AdbTimeout
"""
if isinstance(stream, _AdbStreamConnection):
stream = stream.conn
try:
fragments = []
while 1:
chunk = stream.recv(chunk_size)
if chunk:
fragments.append(chunk)
else:
break
return b''.join(fragments)
except socket.timeout:
raise AdbTimeout('adb read timeout')
def possible_reasons(*args):
"""
Show possible reasons

22
module/device/screenshot.py
View File

@ -26,6 +26,16 @@ class Screenshot(Adb, WSA, Uiautomator2, AScreenCap):
_last_save_time = {}
image: np.ndarray
@cached_property
def screenshot_methods(self):
return {
'ADB': self.screenshot_adb,
'ADB_nc': self.screenshot_adb_nc,
'uiautomator2': self.screenshot_uiautomator2,
'aScreenCap': self.screenshot_ascreencap,
'aScreenCap_nc': self.screenshot_ascreencap_nc,
}
@timer
def screenshot(self):
"""
@ -36,13 +46,11 @@ class Screenshot(Adb, WSA, Uiautomator2, AScreenCap):
self._screenshot_interval.reset()
for _ in range(2):
method = self.config.Emulator_ScreenshotMethod
if method == 'aScreenCap':
self.image = self.screenshot_ascreencap()
elif method == 'uiautomator2':
self.image = self.screenshot_uiautomator2()
else:
self.image = self.screenshot_adb()
method = self.screenshot_methods.get(
self.config.Emulator_ScreenshotMethod,
self.screenshot_adb
)
self.image = method()
if self.config.Emulator_ScreenshotDedithering:
# This will take 40-60ms