minitouch
pyminitouch
minitouch
minitouch 是 openstf 基于 ndk + android 开发的用于模拟人类点击行为的操作库。这个库以高稳定性、反应快著称,比起adb操作与uiautomator都要更灵敏,被广泛用于android设备的精细操作
pymintouch
该项目将对 minitouch 进行封装,致力于降低使用成本,使这个库能够更好的被利用起来
使用简介
使用minitouch,需要通过adb devices先获取到设备的uuid,也可用如下代码:
import subprocess
import re
def get_device_uuid():
stdout, stderr = subprocess.Popen("adb devices", shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True).communicate()
pattern = r'([a-z-A-Z0-9]*)\s*device$'
uuid_lst = re.findall(pattern, stdout, re.M)
return uuid_lst
- 官方demo
_DEVICE_ID = device_uuid
device = MNTDevice(_DEVICE_ID)
# It's also very important to note that the maximum X and Y coordinates may, but usually do not, match the display size.
# so you need to calculate position by yourself, and you can get maximum X and Y by this way:
print("max x: ", device.connection.max_x)
print("max y: ", device.connection.max_y)
# single-tap
device.tap([(400, 600)])
# multi-tap
device.tap([(400, 400), (600, 600)])
# set the pressure, default == 100
device.tap([(400, 600)], pressure=50)
# long-time-tap
device.tap([(400, 600)], duration=2000)
# and no up at the end. you can continue your actions after that. default to false
device.tap([(400, 600)], duration=2000, no_up=True)
# swipe
device.swipe([(100, 100), (500, 500)])
# of course, with duration and pressure
device.swipe([(100, 100), (400, 400), (200, 400)], duration=500, pressure=50)
# and no down at the beginning or no up at the end.
# you can apply a special action before swipe, to build a complex action.
device.tap([(400, 600)], duration=2000, no_up=True)
device.swipe(
[(400, 600), (400, 400), (200, 400)],
duration=500,
pressure=50,
no_down=True,
no_up=True,
)
device.swipe(
[(200, 400), (400, 400), (400, 600)], duration=500, pressure=50, no_down=True
)
# extra functions ( their names start with 'ext_' )
device.ext_smooth_swipe(
[(100, 100), (400, 400), (200, 400)], duration=500, pressure=50, part=20
)
# stop minitouch
# when it was stopped, minitouch can do nothing for device, including release.
device.stop()
# ---
# In another way, you needn't consider about device's life-cycle.
# context manager will handle it
with safe_device(_DEVICE_ID) as device:
# single-tap
device.tap([(400, 600)])
# multi-tap
device.tap([(400, 400), (600, 600)])
# set the pressure, default == 100
device.tap([(400, 600)], pressure=50)
# ---
# What's more, you can also access low level API for further usage.
with safe_connection(_DEVICE_ID) as connection:
builder = CommandBuilder()
builder.down(0, 400, 400, 50)
builder.commit()
builder.move(0, 500, 500, 50)
builder.commit()
builder.move(0, 800, 400, 50)
builder.commit()
builder.up(0)
builder.commit()
builder.publish(connection)
# if you are using MNTDevice object, replace with this:
# builder.publish(device.connection)
# ---
# Of course, you may want to operate it just like using minitouch itself.
# send raw text to it
_OPERATION = """
d 0 150 150 50\n
c\n
u 0\n
c\n
"""
with safe_connection(_DEVICE_ID) as conn:
conn.send(_OPERATION)
device = MNTDevice(_DEVICE_ID)
device.connection.max_x
、device.connection.max_y
device.tap(points, pressure=100, duration=None, no_up=None)
device.swipe(points, pressure=100, duration=None, no_down=None, no_up=None)
device.stop()
:操作完成后需要断开连接,否则下次执行时会出现异常binding socket: Address already in use. Unable to start server on minitouch- 上下文管理:
with safe_device(_DEVICE_ID) as device
# safe_device源码: @contextmanager def safe_device(device_id): """ use MNTDevice safely """ _device = MNTDevice(device_id) try: yield _device finally: time.sleep(config.DEFAULT_DELAY) _device.stop()
- 更底层的api:
with safe_connection(_DEVICE_ID) as connection
- 发送minitouch原始文本:
with safe_connection(_DEVICE_ID) as conn: conn.send(_OPERATION)
- 至于
c r d m u w
的含义可参看minitouch文档
- 至于
部分源码
主要在connection.py, actions.py, utils.py
三个文件中,在IDE中通过from pyminitouch import connection, actions, utils
跳转查看源码即可
- 这里记录几个我个人可能用的上utils中的函数
is_port_using(port_num) restart_adb() is_device_connected(device_id)
# utils.py
def is_port_using(port_num):
""" if port is using by others, return True. else return False """
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(2)
try:
result = s.connect_ex((config.DEFAULT_HOST, port_num))
# if port is using, return code should be 0. (can be connected)
return result == 0
finally:
s.close()
def restart_adb():
""" restart adb server """
_ADB = config.ADB_EXECUTOR
subprocess.check_call([_ADB, "kill-server"])
subprocess.check_call([_ADB, "start-server"])
def is_device_connected(device_id):
""" return True if device connected, else return False """
_ADB = config.ADB_EXECUTOR
try:
device_name = subprocess.check_output(
[_ADB, "-s", device_id, "shell", "getprop", "ro.product.model"]
)
device_name = (
device_name.decode(config.DEFAULT_CHARSET)
.replace("\n", "")
.replace("\r", "")
)
logger.info("device {} online".format(device_name))
except subprocess.CalledProcessError:
return False
return True