chore.更新 C、Python 集成示例

This commit is contained in:
MistEO
2022-03-12 23:41:30 +08:00
parent 328daa7e03
commit 59637500e0
6 changed files with 311 additions and 348 deletions

202
src/Python/asst.py Normal file
View File

@@ -0,0 +1,202 @@
import ctypes
import pathlib
import platform
import json
from typing import Union, Dict, List, Any, Type
from enum import Enum, unique, auto
JSON = Union[Dict[str, Any], List[Any], int, str, float, bool, Type[None]]
class Asst:
CallBackType = ctypes.CFUNCTYPE(
None, ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p)
"""
回调函数,使用实例可参照 my_callback
:params:
``param1 message``: 消息类型
``param2 details``: json string
``param3 arg``: 自定义参数
"""
@staticmethod
def load(path: str) -> bool:
"""
加载 dll 及资源
:params:
``path``: DLL及资源所在文件夹路径
"""
if platform.system().lower() == 'windows':
Asst.__libpath = pathlib.Path(path) / 'MeoAssistant.dll'
Asst.__lib = ctypes.WinDLL(str(Asst.__libpath))
else:
Asst.__libpath = pathlib.Path(path) / 'libMeoAssistant.so'
Asst.__lib = ctypes.CDLL(str(Asst.__libpath))
Asst.__set_lib_properties()
return Asst.__lib.AsstLoadResource(path.encode('utf-8'))
def __init__(self, callback: CallBackType = None, arg=None):
"""
:params:
``callback``: 回调函数
``arg``: 自定义参数
"""
if callback:
self.__ptr = Asst.__lib.AsstCreateEx(callback, arg)
else:
self.__ptr = Asst.__lib.AsstCreate()
def __del__(self):
Asst.__lib.AsstDestroy(self.__ptr)
self.__ptr = None
def connect(self, adb_path: str, address: str, config: str = 'General'):
"""
连接设备
:params:
``adb_path``: adb 程序的路径
``address``: adb 地址+端口
``config``: adb 配置,可参考 resource/config.json
:return: 是否连接成功
"""
return Asst.__lib.AsstConnect(self.__ptr,
adb_path.encode('utf-8'), address.encode('utf-8'), config.encode('utf-8'))
TaskId = int
def append_task(self, type_name: str, params: JSON = {}) -> TaskId:
"""
添加任务
:params:
``type_name``: 任务类型,请参考 docs/集成文档.md
``params``: 任务参数,请参考 docs/集成文档.md
:return: 任务 ID, 可用于 set_task_params 接口
"""
return Asst.__lib.AsstAppendTask(self.__ptr, type_name.encode('utf-8'), json.dumps(params).encode('utf-8'))
def set_task_params(self, task_id: TaskId, params: JSON) -> bool:
"""
动态设置任务参数
:params:
``task_id``: 任务 ID, 使用 append_task 接口的返回值
``params``: 任务参数,同 append_task 接口,请参考 docs/集成文档.md
:return: 是否成功
"""
return Asst.__lib.AsstSetTaskParams(self.__ptr, task_id, json.dumps(params).encode('utf-8'))
def start(self) -> bool:
"""
开始任务
:return: 是否成功
"""
return Asst.__lib.AsstStart(self.__ptr)
def stop(self) -> bool:
"""
停止并清空所有任务
:return: 是否成功
"""
Asst.__lib.AsstStop.restype = ctypes.c_bool
Asst.__lib.AsstStop.argtypes = (ctypes.c_void_p,)
return Asst.__lib.AsstStop(self.__ptr)
@staticmethod
def log(level: str, message: str) -> None:
'''
打印日志
:params:
``level``: 日志等级标签
``message``: 日志内容
'''
Asst.__lib.AsstLog(level.encode('utf-8'), message.encode('utf-8'))
def get_version(self) -> str:
"""
获取DLL版本号
:return: 版本号
"""
return Asst.__lib.AsstGetVersion().decode('utf-8')
@staticmethod
def __set_lib_properties():
Asst.__lib.AsstLoadResource.restype = ctypes.c_bool
Asst.__lib.AsstLoadResource.argtypes = (
ctypes.c_char_p,)
Asst.__lib.AsstCreate.restype = ctypes.c_void_p
Asst.__lib.AsstCreate.argtypes = ()
Asst.__lib.AsstCreateEx.restype = ctypes.c_void_p
Asst.__lib.AsstCreateEx.argtypes = (
ctypes.c_void_p, ctypes.c_void_p,)
Asst.__lib.AsstDestroy.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstConnect.restype = ctypes.c_bool
Asst.__lib.AsstConnect.argtypes = (
ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p,)
Asst.__lib.AsstAppendTask.restype = ctypes.c_int
Asst.__lib.AsstAppendTask.argtypes = (
ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p)
Asst.__lib.AsstSetTaskParams.restype = ctypes.c_bool
Asst.__lib.AsstSetTaskParams.argtypes = (
ctypes.c_void_p, ctypes.c_int, ctypes.c_char_p)
Asst.__lib.AsstStart.restype = ctypes.c_bool
Asst.__lib.AsstStart.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstGetVersion.restype = ctypes.c_char_p
Asst.__lib.AsstLog.restype = None
Asst.__lib.AsstLog.argtypes = (
ctypes.c_char_p, ctypes.c_char_p)
@unique
class Message(Enum):
"""
回调消息
请参考 docs/回调消息.md
"""
InternalError = 0
InitFailed = auto()
ConnectionInfo = auto()
AllTasksCompleted = auto()
TaskChainError = 10000
TaskChainStart = auto()
TaskChainCompleted = auto()
TaskChainExtraInfo = auto()
SubTaskError = 20000
SubTaskStart = auto()
SubTaskCompleted = auto()
SubTaskExtraInfo = auto()

View File

@@ -1,286 +0,0 @@
import ctypes
import pathlib
import platform
from message import Message
class Asst:
CallBackType = ctypes.CFUNCTYPE(
None, ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p)
"""
回调函数使用实例可参照my_callback
:params:
``param1 message``: 消息类型
``param2 details``: json string
``param3 arg``: 自定义参数
"""
@staticmethod
def load(path: str) -> bool:
"""
加载 dll 及资源
:params:
``path``: DLL及资源所在文件夹路径
"""
if platform.system().lower() == 'windows':
Asst.__libpath = pathlib.Path(path) / 'MeoAssistant.dll'
Asst.__lib = ctypes.WinDLL(str(Asst.__libpath))
else:
Asst.__libpath = pathlib.Path(path) / 'libMeoAssistant.so'
Asst.__lib = ctypes.CDLL(str(Asst.__libpath))
Asst.__set_lib_properties()
return Asst.__lib.AsstLoadResource(path.encode('utf-8'))
def __init__(self, callback: CallBackType, arg=None):
"""
:params:
``callback``: 回调函数
``arg``: 自定义参数
"""
self.__ptr = Asst.__lib.AsstCreateEx(callback, arg)
def __del__(self):
Asst.__lib.AsstDestroy(self.__ptr)
def connect(self, adb_path: str, address: str, config: str = 'General'):
"""
连接设备
:params:
``adb_path``: adb 程序的路径
``address``: adb 地址+端口
``config``: adb 配置,可参考 resource/config.json
:return: 是否连接成功
"""
return Asst.__lib.AsstConnect(self.__ptr,
adb_path.encode('utf-8'), address.encode('utf-8'), config.encode('utf-8'))
def append_start_up(self) -> bool:
"""
添加开始唤醒任务
"""
return Asst.__lib.AsstAppendStartUp(self.__ptr)
def append_fight(self, stage: str, max_medicine: int, max_stone: int, max_times: int) -> bool:
"""
添加刷理智任务
:params:
``stage``: 关卡名,为空则不进行选关(需要当前在蓝色开始按钮界面),否则跳转到对应关卡
支持:'''LastBattle''Annihilation''CE-5''AP-5''LS-5''CA-5'
``max_medicine``: 最多吃多少理智药
``max_stone``: 最多吃多少源石
``max_times``: 最多刷多少
"""
return Asst.__lib.AsstAppendFight(self.__ptr, stage.encode('utf-8'), max_medicine, max_stone, max_times)
def append_award(self) -> bool:
"""
添加领取每日奖励任务
"""
return Asst.__lib.AsstAppendAward(self.__ptr)
def append_visit(self) -> bool:
"""
添加访问好友任务
"""
return Asst.__lib.AsstAppendVisit(self.__ptr)
def append_mall(self, with_shopping: bool) -> bool:
"""
添加信用商店任务
:params:
``with_shopping``: 是否信用商店购物
"""
return Asst.__lib.AsstAppendMall(self.__ptr, with_shopping)
def append_infrast(self, work_mode: int, order_list: list, uses_of_drones: str, dorm_threshold: float) -> bool:
"""
添加基建任务
:params:
``work_mode``: 工作模式,当前仅支持传 1
``order_list``: 换班顺序str list例如 [ "Mfg", "Trade", "Control", "Power", "Reception", "Office", "Dorm"]
``uses_of_drones``: 无人机用途,支持以下之一:"_NotUse""Money""SyntheticJade""CombatRecord""PureGold""OriginStone""Chip"
``dorm_threshold``: 宿舍心情阈值,取值 [0, 1.0]
"""
order_byte_list = []
for item in order_list:
order_byte_list.append(item.encode('utf-8'))
order_len = len(order_byte_list)
order_arr = (ctypes.c_char_p * order_len)()
order_arr[:] = order_byte_list
return Asst.__lib.AsstAppendInfrast(self.__ptr, work_mode,
order_arr, order_len,
uses_of_drones.encode('utf-8'), dorm_threshold)
def append_recruit(self, max_times: int, select_level: list, confirm_level: list, need_refresh: bool, use_expedited: bool) -> bool:
"""
添加自动公招任务
:params:
``max_times``: 自动公招次数
``select_level``: 要点击的Tags等级例如3级Tags不选也可以直接确认int list
``confirm_level``: 要确认开始招募的Tags等级int list
``need_refresh``: 是否刷新3级Tags
``use_expedited``: 是否使用加急许可
"""
select_len = len(select_level)
select_arr = (ctypes.c_int * select_len)()
select_arr[:] = select_level
confirm_len = len(confirm_level)
confirm_arr = (ctypes.c_int * confirm_len)()
confirm_arr[:] = confirm_level
return Asst.__lib.AsstAppendRecruit(self.__ptr, max_times,
select_arr, select_len,
confirm_arr, confirm_len,
need_refresh, use_expedited)
def append_roguelike(self, mode: int) -> bool:
"""
添加无限肉鸽任务
:params:
``mode``: 工作模式 :
0 - 尽可能一直往后打
1 - 第一层投资完源石锭就退出
2 - 投资过后再退出,没有投资就继续往后打
"""
return Asst.__lib.AsstAppendRoguelike(self.__ptr, mode)
def start(self) -> bool:
"""
开始任务
"""
return Asst.__lib.AsstStart(self.__ptr)
def start_recruit_calc(self, select_level: list, set_time: bool) -> bool:
"""
进行公招计算
:params:
``select_level``: 要点击的Tags等级int list
``set_time``: 是否要点击9小时
"""
select_len = len(select_level)
select_arr = (ctypes.c_int * select_len)()
select_arr[:] = select_level
return Asst.__lib.AsstStartRecruitCalc(self.__ptr, select_arr, select_len, set_time)
def stop(self) -> bool:
"""
停止并清空所有任务
"""
Asst.__lib.AsstStop.restype = ctypes.c_bool
Asst.__lib.AsstStop.argtypes = (ctypes.c_void_p,)
return Asst.__lib.AsstStop(self.__ptr)
def set_penguin_id(self, id: str) -> bool:
"""
设置企鹅物流ID
:params:
``id``: 企鹅物流ID仅数字部分
"""
return Asst.__lib.AsstSetPenguinId(self.__ptr, id.encode('utf-8'))
@staticmethod
def log(level: str, message: str) -> None:
'''
打印日志
:params:
``level``: 日志等级标签
``message``: 日志内容
'''
Asst.__lib.AsstLog(level.encode('utf-8'), message.encode('utf-8'))
def get_version(self) -> str:
"""
获取DLL版本号
:return: 版本号
"""
return Asst.__lib.AsstGetVersion().decode('utf-8')
@staticmethod
def __set_lib_properties():
Asst.__lib.AsstLoadResource.restype = ctypes.c_bool
Asst.__lib.AsstLoadResource.argtypes = (
ctypes.c_char_p,)
Asst.__lib.AsstCreateEx.restype = ctypes.c_void_p
Asst.__lib.AsstCreateEx.argtypes = (
ctypes.c_void_p, ctypes.c_void_p,)
Asst.__lib.AsstDestroy.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstConnect.restype = ctypes.c_bool
Asst.__lib.AsstConnect.argtypes = (
ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p,)
Asst.__lib.AsstAppendStartUp.restype = ctypes.c_bool
Asst.__lib.AsstAppendStartUp.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstAppendFight.restype = ctypes.c_bool
Asst.__lib.AsstAppendFight.argtypes = (
ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int, ctypes.c_int, ctypes.c_int,)
Asst.__lib.AsstAppendAward.restype = ctypes.c_bool
Asst.__lib.AsstAppendAward.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstAppendVisit.restype = ctypes.c_bool
Asst.__lib.AsstAppendVisit.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstAppendMall.restype = ctypes.c_bool
Asst.__lib.AsstAppendMall.argtypes = (ctypes.c_void_p, ctypes.c_bool)
Asst.__lib.AsstAppendInfrast.restype = ctypes.c_bool
Asst.__lib.AsstAppendInfrast.argtypes = (
ctypes.c_void_p, ctypes.c_int,
ctypes.POINTER(ctypes.c_char_p), ctypes.c_int,
ctypes.c_char_p, ctypes.c_double,)
Asst.__lib.AsstAppendRecruit.restype = ctypes.c_bool
Asst.__lib.AsstAppendRecruit.argtypes = (
ctypes.c_void_p, ctypes.c_int,
ctypes.POINTER(ctypes.c_int), ctypes.c_int,
ctypes.POINTER(ctypes.c_int), ctypes.c_int,
ctypes.c_bool, ctypes.c_bool,)
Asst.__lib.AsstAppendRoguelike.restype = ctypes.c_bool
Asst.__lib.AsstAppendRoguelike.argtypes = (
ctypes.c_void_p, ctypes.c_int,)
Asst.__lib.AsstStart.restype = ctypes.c_bool
Asst.__lib.AsstStart.argtypes = (ctypes.c_void_p,)
Asst.__lib.AsstStartRecruitCalc.restype = ctypes.c_bool
Asst.__lib.AsstStartRecruitCalc.argtypes = (
ctypes.c_void_p, ctypes.POINTER(ctypes.c_int), ctypes.c_int, ctypes.c_bool,)
Asst.__lib.AsstSetPenguinId.restype = ctypes.c_bool
Asst.__lib.AsstSetPenguinId.argtypes = (
ctypes.c_void_p, ctypes.c_char_p)
Asst.__lib.AsstGetVersion.restype = ctypes.c_char_p
Asst.__lib.AsstLog.restype = None
Asst.__lib.AsstLog.argtypes = (
ctypes.c_char_p, ctypes.c_char_p)

View File

@@ -1,33 +0,0 @@
from enum import Enum, unique, auto
@unique
class Message(Enum):
"""
回调消息
请参考 https://github.com/MistEO/MeoAssistantArknights/wiki/%E5%9B%9E%E8%B0%83%E6%B6%88%E6%81%AF
"""
InternalError = 0
InitFailed = auto()
ConnectionError = auto()
AllTasksCompleted = auto()
TaskChainError = 10000
TaskChainStart = auto()
TaskChainCompleted = auto()
TaskChainExtraInfo = auto()
SubTaskError = 20000
SubTaskStart = auto()
SubTaskCompleted = auto()
SubTaskExtraInfo = auto()

View File

@@ -1,23 +1,27 @@
import json
import pathlib
from interface import Asst, Message
from asst import Asst, Message
@Asst.CallBackType
def my_callback(msg, details, arg):
m = Message(msg)
d = json.loads(details.decode('utf-8'))
print(m, d, arg)
if __name__ == "__main__":
@Asst.CallBackType
def my_callback(msg, details, arg):
m = Message(msg)
d = json.loads(details.decode('utf-8'))
print(m, d, arg)
# 请设置为存放 dll 文件及资源的路径
path: str = (pathlib.Path.cwd()).__str__()
Asst.load(path=path)
asst = Asst(callback=my_callback)
print('version', asst.get_version())
# 若需要获取详细执行信息,请传入 callback 参数
# 例如 asst = Asst(callback=my_callback)
asst = Asst()
if asst.connect('adb', '127.0.0.1:5555'):
print('连接成功')
@@ -25,12 +29,34 @@ if __name__ == "__main__":
print('连接失败')
exit()
asst.append_fight("CE-5", 0, 0, 1)
# asst.append_infrast(1, [ "Mfg", "Trade", "Control", "Power", "Reception", "Office", "Dorm"], "Money", 0.3)
# asst.append_award()
asst.start()
# 任务及参数请参考 docs/集成文档.md
# 详细参数请参考 docs/集成文档.md
# asst.start_recruit_calc([4, 5, 6], True)
asst.append_task('StartUp')
asst.append_task('Fight', {
'stage': 'LastBattle',
# 'penguin_id': '1234567'
})
asst.append_task('Recruit', {
'select': [4],
'confirm': [3, 4],
'times': 4
})
asst.append_task('Infrast', {
'facility': [
"Mfg", "Trade", "Control", "Power", "Reception", "Office", "Dorm"
],
'drones': "Money"
})
asst.append_task('Visit')
asst.append_task('Mall', {
'shopping': True,
'shopping': ['家具', ''],
'is_black_list': True
})
asst.append_task('Award')
asst.start()
while True:
input('>')

View File

@@ -1,7 +1,7 @@
"""
function: 定时启动 maa
author: Black Cat Bon
version: v2.0.1
version: v2.1
"""
from importlib.resources import path
@@ -15,7 +15,7 @@ import pathlib
import sys
import re
from interface import Asst, Message
from asst import Asst, Message
def get_now_time():
@@ -26,14 +26,31 @@ def job():
# print("I'm running on thread %s" % threading.current_thread())
print("现在时间:" + datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
asst.append_fight("LastBattle", 0, 0, 999)
asst.append_recruit(3, [4, 5], [3, 4, 5], True, False)
asst.append_infrast(1, ["Mfg", "Trade", "Control",
"Power", "Reception", "Office", "Dorm"], "Money", 0.1)
asst.append_visit()
asst.append_mall(True)
asst.append_award()
# asst.set_penguin_id('1234567')
# 详细参数请参考 docs/集成文档.md
# asst.append_task('StartUp')
asst.append_task('Fight', {
'stage': 'LastBattle',
# 'penguin_id': '1234567'
})
asst.append_task('Recruit', {
'select': [4],
'confirm': [3, 4],
'times': 4
})
asst.append_task('Infrast', {
'facility': [
"Mfg", "Trade", "Control", "Power", "Reception", "Office", "Dorm"
],
'drones': "Money"
})
asst.append_task('Visit')
asst.append_task('Mall', {
'shopping': True,
'shopping': ['家具', ''],
'is_black_list': True
})
asst.append_task('Award')
asst.start()
@@ -114,7 +131,9 @@ if __name__ == "__main__":
path: str = (pathlib.Path.cwd()).__str__()
Asst.load(path)
asst = Asst(callback=my_callback)
# 若需要获取详细执行信息,请传入 callback 参数
# 例如 asst = Asst(callback=my_callback)
asst = Asst()
print('version', asst.get_version())