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

Refactor: Make MaaTouch inputs synchronous

Browse Source
This commit is contained in:
LmeSzinc 2024-04-28 23:27:23 +08:00
parent 03d99313db
commit 0fb6a96691
4 changed files with 170 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
bin/MaaTouch/maatouchsync Normal file
View File

Binary file not shown.

4
module/config/config_manual.py
View File

@ -91,8 +91,8 @@ class ManualConfig:
SCRCPY_FILEPATH_LOCAL = './bin/scrcpy/scrcpy-server-v1.20.jar'
SCRCPY_FILEPATH_REMOTE = '/data/local/tmp/scrcpy-server-v1.20.jar'
MAATOUCH_FILEPATH_LOCAL = './bin/MaaTouch/maatouch'
MAATOUCH_FILEPATH_REMOTE = '/data/local/tmp/maatouch'
MAATOUCH_FILEPATH_LOCAL = './bin/MaaTouch/maatouchsync'
MAATOUCH_FILEPATH_REMOTE = '/data/local/tmp/maatouchsync'
"""
module.campaign.gems_farming

117
module/device/method/maatouch.py
View File

@ -1,5 +1,6 @@
import socket
import threading
import time
from functools import wraps
from adbutils.errors import AdbError
@ -8,7 +9,7 @@ from module.base.decorator import cached_property, del_cached_property, has_cach
from module.base.timer import Timer
from module.base.utils import *
from module.device.connection import Connection
from module.device.method.minitouch import CommandBuilder, insert_swipe
from module.device.method.minitouch import CommandBuilder, insert_swipe, Command
from module.device.method.utils import RETRY_TRIES, handle_adb_error, retry_sleep
from module.exception import RequestHumanTakeover
from module.logger import logger
@ -38,6 +39,15 @@ def retry(func):
def init():
self.adb_reconnect()
del_cached_property(self, '_maatouch_builder')
# MaaTouchSyncTimeout
# Probably because adb server was killed
except MaaTouchSyncTimeout as e:
logger.error(e)
def init():
self.adb_reconnect()
del_cached_property(self, '_maatouch_builder')
self.reset_maatouch()
# Emulator closed
except ConnectionAbortedError as e:
logger.error(e)
@ -79,7 +89,12 @@ def retry(func):
class MaatouchBuilder(CommandBuilder):
def __init__(self, device, contact=0, handle_orientation=False):
def __init__(
self,
device,
contact=0,
handle_orientation=False,
):
"""
Args:
device (MaaTouch):
@ -90,11 +105,21 @@ class MaatouchBuilder(CommandBuilder):
def send(self):
return self.device.maatouch_send(builder=self)
def send_sync(self, mode=2):
return self.device.maatouch_send_sync(builder=self, mode=mode)
def end(self):
self.device.sleep(self.DEFAULT_DELAY)
class MaaTouchNotInstalledError(Exception):
pass
class MaaTouchSyncTimeout(Exception):
pass
class MaaTouch(Connection):
"""
Control method that implements the same as scrcpy and has an interface similar to minitouch.
@ -121,6 +146,8 @@ class MaaTouch(Connection):
del self._maatouch_init_thread
self._maatouch_init_thread = None
# Return an empty builder
self._maatouch_builder.clear()
return self._maatouch_builder
def early_maatouch_init(self):
@ -174,7 +201,7 @@ class MaaTouch(Connection):
# CLASSPATH=/data/local/tmp/maatouch app_process / com.shxyke.MaaTouch.App
stream = self.adb_shell(
['CLASSPATH=/data/local/tmp/maatouch', 'app_process', '/', 'com.shxyke.MaaTouch.App'],
[f'CLASSPATH={self.config.MAATOUCH_FILEPATH_REMOTE}', 'app_process', '/', 'com.shxyke.MaaTouch.App'],
stream=True,
recvall=False
)
@ -226,6 +253,8 @@ class MaaTouch(Connection):
# _, pid = out.split(" ")
# self._maatouch_pid = pid
# Timeout 2s for sync
stream.settimeout(2)
logger.info(
"MaaTouch stream connected"
)
@ -237,11 +266,56 @@ class MaaTouch(Connection):
def maatouch_send(self, builder: MaatouchBuilder):
content = builder.to_minitouch()
# logger.info("send operation: {}".format(content.replace("\n", "\\n")))
logger.info("send operation: {}".format(content.replace("\n", "\\n")))
byte_content = content.encode('utf-8')
self._maatouch_stream.sendall(byte_content)
self._maatouch_stream.recv(0)
self.sleep(self.maatouch_builder.delay / 1000 + builder.DEFAULT_DELAY)
self.sleep(builder.delay / 1000 + builder.DEFAULT_DELAY)
builder.clear()
def maatouch_send_sync(self, builder: MaatouchBuilder, mode=2):
# Set inject mode to the last command
for command in builder.commands[::-1]:
if command.operation in ['r', 'd', 'm', 'u']:
command.mode = mode
break
# add maatouch sync command: 's <timestamp>\n'
timestamp = str(int(time.time() * 1000))
builder.commands.insert(0, Command(
's', text=timestamp
))
# Send
content = builder.to_maatouch_sync()
logger.info("send operation: {}".format(content.replace("\n", "\\n")))
byte_content = content.encode('utf-8')
self._maatouch_stream.sendall(byte_content)
self._maatouch_stream.recv(0)
# Wait until operations finished
start = time.time()
socket_out = self._maatouch_stream.makefile()
max_trial = 3
for n in range(3):
try:
out = socket_out.readline()
except socket.timeout as e:
raise MaaTouchSyncTimeout(str(e))
out = out.strip()
logger.info(out)
if out == timestamp:
break
if out == 'Killed':
raise MaaTouchNotInstalledError('MaaTouch died, probably because version incompatible')
if n == max_trial - 1:
raise MaaTouchSyncTimeout('Too many incorrect sync response')
time.sleep(0.001)
logger.info(f'Delay: {builder.delay}')
logger.info(f'Waiting control {time.time() - start}')
self.sleep(builder.DEFAULT_DELAY)
builder.clear()
def maatouch_install(self):
@ -257,30 +331,31 @@ class MaaTouch(Connection):
builder = self.maatouch_builder
builder.down(x, y).commit()
builder.up().commit()
builder.send()
builder.send_sync()
@retry
def long_click_maatouch(self, x, y, duration=1.0):
duration = int(duration * 1000)
builder = self.maatouch_builder
builder.down(x, y).commit().wait(duration)
builder.down(x, y).wait(duration).commit()
builder.up().commit()
builder.send()
builder.send_sync()
@retry
def swipe_maatouch(self, p1, p2):
points = insert_swipe(p0=p1, p3=p2)
builder = self.maatouch_builder
builder.down(*points[0]).commit()
builder.send()
builder.down(*points[0]).wait(10).commit()
builder.send_sync()
for point in points[1:]:
builder.move(*point).commit().wait(10)
builder.send()
builder.move(*point).wait(10)
builder.commit()
builder.send_sync()
builder.up().commit()
builder.send()
builder.send_sync()
@retry
def drag_maatouch(self, p1, p2, point_random=(-10, -10, 10, 10)):
@ -289,16 +364,22 @@ class MaaTouch(Connection):
points = insert_swipe(p0=p1, p3=p2, speed=20)
builder = self.maatouch_builder
builder.down(*points[0]).commit()
builder.send()
builder.down(*points[0]).commit().wait(10)
builder.send_sync()
for point in points[1:]:
builder.move(*point).commit().wait(10)
builder.send()
builder.send_sync()
builder.move(*p2).commit().wait(140)
builder.move(*p2).commit().wait(140)
builder.send()
builder.send_sync()
builder.up().commit()
builder.send()
builder.send_sync()
@retry
def reset_maatouch(self):
builder = self.maatouch_builder
builder.reset().commit()
builder.send_sync()

81
module/device/method/minitouch.py
View File

@ -100,7 +100,9 @@ class Command:
x: int = 0,
y: int = 0,
ms: int = 10,
pressure: int = 100
pressure: int = 100,
mode: int = 0,
text: str = ''
):
"""
See https://github.com/openstf/minitouch#writable-to-the-socket
@ -112,6 +114,8 @@ class Command:
y:
ms:
pressure:
mode:
text:
"""
self.operation = operation
self.contact = contact
@ -119,6 +123,8 @@ class Command:
self.y = y
self.ms = ms
self.pressure = pressure
self.mode = mode
self.text = text
def to_minitouch(self) -> str:
"""
@ -139,6 +145,36 @@ class Command:
else:
return ''
def to_maatouch_sync(self):
if self.operation == 'c':
return f'{self.operation}\n'
elif self.operation == 'r':
if self.mode:
return f'{self.operation} {self.mode}\n'
else:
return f'{self.operation}\n'
elif self.operation == 'd':
if self.mode:
return f'{self.operation} {self.contact} {self.x} {self.y} {self.pressure} {self.mode}\n'
else:
return f'{self.operation} {self.contact} {self.x} {self.y} {self.pressure}\n'
elif self.operation == 'm':
if self.mode:
return f'{self.operation} {self.contact} {self.x} {self.y} {self.pressure} {self.mode}\n'
else:
return f'{self.operation} {self.contact} {self.x} {self.y} {self.pressure}\n'
elif self.operation == 'u':
if self.mode:
return f'{self.operation} {self.contact} {self.mode}\n'
else:
return f'{self.operation} {self.ms}\n'
elif self.operation == 'w':
return f'{self.operation} {self.ms}\n'
elif self.operation == 's':
return f'{self.operation} {self.text}\n'
else:
return ''
def to_atx_agent(self, max_x=1280, max_y=720) -> str:
"""
Dict that send to atx-agent, $DEVICE_URL/minitouch
@ -184,7 +220,12 @@ class CommandBuilder:
max_x = 1280
max_y = 720
def __init__(self, device, contact=0, handle_orientation=True):
def __init__(
self,
device,
contact=0,
handle_orientation=True,
):
"""
Args:
device:
@ -230,45 +271,61 @@ class CommandBuilder:
def commit(self):
""" add minitouch command: 'c\n' """
self.commands.append(Command('c'))
self.commands.append(Command(
'c'
))
return self
def reset(self):
def reset(self, mode=0):
""" add minitouch command: 'r\n' """
self.commands.append(Command('r'))
self.commands.append(Command(
'r', mode=mode
))
return self
def wait(self, ms=10):
""" add minitouch command: 'w <ms>\n' """
self.commands.append(Command('w', ms=ms))
self.commands.append(Command(
'w', ms=ms
))
self.delay += ms
return self
def up(self):
def up(self, mode=0):
""" add minitouch command: 'u <contact>\n' """
self.commands.append(Command('u', contact=self.contact))
self.commands.append(Command(
'u', contact=self.contact, mode=mode
))
return self
def down(self, x, y, pressure=100):
def down(self, x, y, pressure=100, mode=0):
""" add minitouch command: 'd <contact> <x> <y> <pressure>\n' """
x, y = self.convert(x, y)
self.commands.append(Command('d', x=x, y=y, contact=self.contact, pressure=pressure))
self.commands.append(Command(
'd', x=x, y=y, contact=self.contact, pressure=pressure, mode=mode
))
return self
def move(self, x, y, pressure=100):
def move(self, x, y, pressure=100, mode=0):
""" add minitouch command: 'm <contact> <x> <y> <pressure>\n' """
x, y = self.convert(x, y)
self.commands.append(Command('m', x=x, y=y, contact=self.contact, pressure=pressure))
self.commands.append(Command(
'm', x=x, y=y, contact=self.contact, pressure=pressure, mode=mode
))
return self
def clear(self):
""" clear current commands """
self.commands = []
self.delay = 0
return self
def to_minitouch(self) -> str:
return ''.join([command.to_minitouch() for command in self.commands])
def to_maatouch_sync(self) -> str:
return ''.join([command.to_maatouch_sync() for command in self.commands])
def to_atx_agent(self) -> List[str]:
return [command.to_atx_agent(self.max_x, self.max_y) for command in self.commands]