From 436de69f18834165fc1dc83c4c4f7ac8e84ad20f Mon Sep 17 00:00:00 2001 From: AutumnSound <62000745+soundofautumn@users.noreply.github.com> Date: Sat, 28 May 2022 20:42:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?python=E8=87=AA=E5=8A=A8=E7=94=9F=E6=88=90?= =?UTF-8?q?=E4=BD=9C=E4=B8=9A=E8=84=9A=E6=9C=AC=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 加了点注释 添加了自动调试功能 (连打开mma选作业都省了) --- tools/CopilotDesinger/CopilotDesinger.py | 346 +++++++++++++++++++++++ tools/CopilotDesinger/asst.py | 207 ++++++++++++++ tools/CopilotDesinger/main.py | 57 ++++ 3 files changed, 610 insertions(+) create mode 100644 tools/CopilotDesinger/CopilotDesinger.py create mode 100644 tools/CopilotDesinger/asst.py create mode 100644 tools/CopilotDesinger/main.py diff --git a/tools/CopilotDesinger/CopilotDesinger.py b/tools/CopilotDesinger/CopilotDesinger.py new file mode 100644 index 0000000000..71b0c1159c --- /dev/null +++ b/tools/CopilotDesinger/CopilotDesinger.py @@ -0,0 +1,346 @@ +from __future__ import annotations + +import json + + +# ======================= constants begin ========================= +class Direction: + UP = "上" + DOWN = "下" + RIGHT = "右" + LEFT = "左" + + +class ActionType: + Deploy = "部署" + Skill = "技能" + Retreat = "撤退" + SpeedUp = "二倍速" + BulletTime = "子弹时间" + SkillUsage = "技能用法" + + +class SkillUsageType: + NotAutoUse = 0 # 0 - 不自动使用 默认 + UseWhenOk = 1 # 1 - 好了就用,有多少次用多少次(例如干员 棘刺 3 技能、桃金娘 1 技能等) + UseWhenOkOnce = 2 # 2 - 好了就用,仅使用一次(例如干员 山 2 技能) + AutoUse = 3 # 3 - 自动判断使用时机(画饼.jpg) + + +# ========================= constants end ============================= + +class Document: + def __init__(self, title: str, details: str, title_color: str = "dark", details_color: str = "dark"): + self._details_color = details_color + self._details = details + self._title_color = title_color + self._title = title + + def to_dict(self) -> dict: + return { + "title": self._title, + "title_color": self._title_color, + "details": self._details, + "details_color": self._details_color + } + + +class Requirements: + def __init__(self, elite: int, level: int, skill_level: int, module: int, potentiality: int): + self._potentiality = potentiality + self._module = module + self._skill_level = skill_level + self._level = level + self._elite = elite + + def to_dict(self): + return { + "elite": self._elite, + "level": self._level, + "skill_level": self._skill_level, + "module": self._module, + "potentiality": self._potentiality + } + + +class Action: + def __init__(self, action_type: str, name: str = None, location: tuple = None, direction: str = "无", kills: int = 0, + skill_usage: int = None, pre_delay: int = 0, rear_delay: int = 0, timeout: int = 999999999, + doc: str = None, doc_color: str = None): + self._doc_color = doc_color + self._doc = doc + self._timeout = timeout + self._rear_delay = rear_delay + self._pre_delay = pre_delay + self._skill_usage = skill_usage + self._kills = kills + self._direction = direction + self._name = name + self._location = location + self._action_type = action_type + + def to_dict(self) -> dict: + res = { + "type": self._action_type, + } + if self._kills != 0: + res["kills"] = self._kills + if self._pre_delay != 0: + res["pre_delay"] = self._pre_delay + if self._rear_delay != 0: + res["rear_delay"] = self._rear_delay + if self._action_type == ActionType.Deploy or self._action_type == ActionType.Skill \ + or self._action_type == ActionType.Retreat: + res["name"] = self._name + + if self._action_type == ActionType.Deploy: + res["location"] = self._location + res["direction"] = self._direction + return res + + +class OperatorOrGroup: + def __init__(self, name: str, battle: Battle): + self._battle = battle + self._name = name + self._pre_delay = 0 + self._rear_delay = 0 + self._wait_kills = 0 + + def pre_delay(self, times: int) -> OperatorOrGroup: + self._pre_delay = times + return self + + def rear_delay(self, times: int) -> OperatorOrGroup: + self._rear_delay = times + return self + + def wait_kills(self, kills: int) -> OperatorOrGroup: + self._wait_kills = kills + return self + + def deploy(self, location: tuple, direction: str): + self._battle.add_action(self._add_conditions(Action(ActionType.Deploy, self._name, location, direction))) + + def use_skill(self): + self._battle.add_action(self._add_conditions(Action(ActionType.Skill, self._name))) + + def retreat(self): + self._battle.add_action(self._add_conditions(Action(ActionType.Retreat, self._name))) + + def _add_conditions(self, action: Action) -> Action: + if self._pre_delay != 0: + action._pre_delay = self._pre_delay + if self._rear_delay != 0: + action._rear_delay = self._rear_delay + if self._wait_kills != 0: + action._kills = self._wait_kills + self._clear_conditions() + return action + + def _clear_conditions(self): + self._pre_delay = 0 + self._rear_delay = 0 + self._wait_kills = 0 + + def to_dict(self) -> dict: + # 此处不应该被执行到 + raise Exception("请使用子类方法") + + +class Group(OperatorOrGroup): + def __init__(self, name: str, battle: Battle, operators: tuple): + """ + 请使用 Battle.create_group() + :param name: + :param battle: + :param operators: + """ + super().__init__(name, battle) + self._operators = operators + + def to_dict(self) -> dict: + return { + "name": self._name, + "opers": [operator.to_dict() for operator in self._operators] + } + + +class Operator(OperatorOrGroup): + def __init__(self, name: str, battle: Battle, skill: int, skill_usage: int = SkillUsageType.NotAutoUse, + requirements: Requirements = None): + """ + 仅当要创建群组时才应该被调用 + 正常使用请用 Battle.create_operator() + :param name: + :param battle: + :param skill: + :param skill_usage: + :param requirements: + """ + super().__init__(name, battle) + self._requirements = requirements + self._skill_usage = skill_usage + self._skill = skill + + def skill_usage_change(self, change_to: int): + """ + 改变技能使用的类型 + 由于需要兼容群组类型,有时此方法可能无法通过代码提示调用,手动调用即可 + :param change_to: 需要调整到的 SkillUsageType + :return: + """ + self._battle.add_action(self._add_conditions(Action(ActionType.SkillUsage, skill_usage=change_to))) + + def to_dict(self) -> dict: + res = { + "name": self._name, + "skill": self._skill, + } + if self._skill_usage != SkillUsageType.NotAutoUse: + res["skill_usage"] = self._skill_usage + if self._requirements is not None: + res["requirements"] = self._requirements.to_dict() + return res + + +class Battle: + def __init__(self, stage_name: str, version="v4.0", doc: Document = None): + self._doc = doc + self._stage_name = stage_name + self._version = version + self._operators = [] + self._groups = [] + self._actions = [] + self._wait_times = 0 + self._wait_kills = 0 + self._speedup = False + self._last_save_file_name = "battle.json" + + def create_operator(self, name: str, skill: int, skill_usage: int = SkillUsageType.NotAutoUse, + requirements: Requirements = None) -> Operator: + """ + 创建并添加一个干员 + :param name: 干员名称 + :param skill: 要使用的技能 + :param skill_usage: 技能使用类型 参见 SkillUsageType + :param requirements: 可选 + :return: 新生成的干员 + """ + new_operator = Operator(name, self, skill, skill_usage, requirements) + self._add_operator(new_operator) + return new_operator + + def create_group(self, name: str, *operators: Operator) -> Group: + """ + 创建并添加一个群组 + :param name: 群组名 + :param operators: 所包含的干员,构造的时候 battle 一项置 None 即可 + :return: 新创建的群组 + """ + new_group = Group(name, self, operators) + self._add_group(new_group) + return new_group + + def _add_operator(self, operator: Operator): + self._operators.append(operator) + + def _add_group(self, group: Group): + self._groups.append(group) + + def add_action(self, action: Action): + self._actions.append(action) + + def speed_up(self): + """ + 调整为二倍速 + :return: + """ + if not self._speedup: + self._speed_change() + self._speedup = True + + def speed_down(self): + """ + 调整为一倍速 + :return: + """ + if self._speedup: + self._speed_change() + self._speedup = False + + def _speed_change(self): + self.add_action(Action(ActionType.SpeedUp)) + + def save_as_json(self, file_name: str = None): + """ + 保存为 json 文件,生成在同目录下 + :param file_name: 要保存的文件名 + :return: + """ + if file_name is None: + file_name = self._last_save_file_name + else: + self._last_save_file_name = file_name + with open(file_name, 'w', encoding='utf-8') as f: + json.dump(self.to_dict(), f, ensure_ascii=False) + + def to_dict(self) -> dict: + res = { + "stage_name": self._stage_name, + "minimum_required": self._version, + "opers": [operator.to_dict() for operator in self._operators], + "groups": [group.to_dict() for group in self._groups], + "actions": [action.to_dict() for action in self._actions], + } + if self._doc is not None: + res["doc"] = self._doc.to_dict() + + return res + + def test(self, enable_debug_output: bool = True): + """ + 自动调用 MeoAssistantArknights 并测试生成自动战斗的文件 + 测试的是上次生成的 json 需先调用 save_as_json() + 请自行配置 MeoAssistantArknights 所在地址等 + 同时确保能导入 asst.py + + :param enable_debug_output: 是否开启 debug 输出 + :return: + """ + # MeoAssistantArknights所在的路径 + mma_path = "D:\\Program Files\\MeoAssistantArknights" + # adb的路径 + adb_path = "D:\\Program Files\\MeoAssistantArknights\\platform-tools\\adb.exe" + # adb连接的地址 + adb_address = "127.0.0.1:8873" + try: + from asst import Asst, Message + except ImportError: + print("asst导入失败,请确保asst.py在同目录下,或者下载 " + "https://github.com/MaaAssistantArknights/MaaAssistantArknights/blob/master/src/Python/asst.py") + return + + @Asst.CallBackType + def callback(msg, details, arg): + print(Message(msg), json.loads(details.decode('utf-8')), arg) + + if not Asst.load(mma_path): + print("加载MeoAssistantArknights失败,请确定路径是否正确") + + asst = Asst(callback) if enable_debug_output else Asst() + if asst.connect(adb_path, adb_address): + print('连接成功') + else: + print('连接失败') + print('请检查adb地址或者端口是否有误') + exit() + asst.append_task('Copilot', { + 'stage_name': self._stage_name, + 'filename': self._last_save_file_name, + 'formation': False, + }) + asst.start() + while True: + if "stop" == str(input('>')): + exit() diff --git a/tools/CopilotDesinger/asst.py b/tools/CopilotDesinger/asst.py new file mode 100644 index 0000000000..b822c2529f --- /dev/null +++ b/tools/CopilotDesinger/asst.py @@ -0,0 +1,207 @@ +import ctypes +import os +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: Union[pathlib.Path, str]) -> bool: + """ + 加载 dll 及资源 + + :params: + ``path``: DLL及资源所在文件夹路径 + """ + if platform.system().lower() == 'windows': + Asst.__libpath = pathlib.Path(path) / 'MeoAssistant.dll' + os.environ["PATH"] += os.pathsep + str(path) + Asst.__lib = ctypes.WinDLL(str(Asst.__libpath)) + else: + Asst.__libpath = pathlib.Path(path) / 'libMeoAssistant.so' + os.environ['LD_LIBRARY_PATH'] += os.pathsep + str(path) + Asst.__lib = ctypes.CDLL(str(Asst.__libpath)) + Asst.__set_lib_properties() + + return Asst.__lib.AsstLoadResource(str(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, ensure_ascii=False).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, ensure_ascii=False).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')) + + @staticmethod + def get_version() -> 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() diff --git a/tools/CopilotDesinger/main.py b/tools/CopilotDesinger/main.py new file mode 100644 index 0000000000..a2ccee5299 --- /dev/null +++ b/tools/CopilotDesinger/main.py @@ -0,0 +1,57 @@ +from CopilotDesinger import * + +# 使用样例 渊默行动 18 +# 按道理是能用的,我照着作者的作业抄的,不能用一定是玛丽的问题( + +# 阅读过此文会对使用本程序有很大的帮助(毕竟本来就是为了生成这种json文件的( +# https://github.com/MaaAssistantArknights/MaaAssistantArknights/blob/master/docs/%E6%88%98%E6%96%97%E6%B5%81%E7%A8%8B%E5%8D%8F%E8%AE%AE.md + +# (其实我觉得有个 ide 会点英文 应该是个人都会用,不过我还是稍微写点注释啥的 + +# 创建作业对象 要啥填啥就完事了 别问为什么用中文命名变量 我怕改成英文了自己过会都看不懂了( +危机合约十八 = Battle(stage_name="obt/rune/level_rune_11-01", + version="v4.0", + doc=Document("《全自动危机合约 18》", "作业抄自 https://www.bilibili.com/video/BV13S4y187NW")) + +# 创建要使用的干员(老婆们) +# 第一个参数是名字,第二个是几技能 +史尔特尔 = 危机合约十八.create_operator("史尔特尔", 3) +伊芙利特 = 危机合约十八.create_operator("伊芙利特", 2) +# 鸡翅技能设定为好了就开 +棘刺 = 危机合约十八.create_operator("棘刺", 3, SkillUsageType.UseWhenOk) +# 创建一个群组 +对空高速单体狙 = 危机合约十八.create_group("对空高速单体狙", Operator("克洛丝", None, 1), Operator("能天使", None, 3)) +红 = 危机合约十八.create_operator("红", 2) +蜜莓 = 危机合约十八.create_operator("蜜莓", 1) + +# 开始写作业 QWQ +# 调成二倍速 +危机合约十八.speed_up() + +# 部署鸡翅 部署在(8,6)格子上,方向朝上 +# 不知道哪格是哪格的可以先随便写点然后 save_as_json() 和 test() +# 运行后会在同目录下生成 map.png 的 +棘刺.deploy((8, 6), Direction.UP) +对空高速单体狙.deploy((2, 4), Direction.UP) +伊芙利特.deploy((10, 4), Direction.LEFT) +蜜莓.deploy((10, 5), Direction.LEFT) +# 等待 16 杀后撤退 +对空高速单体狙.wait_kills(16).retreat() +红.wait_kills(18).deploy((3, 5), Direction.LEFT) +史尔特尔.wait_kills(21).pre_delay(1500).deploy((2, 3), Direction.DOWN) +# 使用 42 技能 +史尔特尔.use_skill() +史尔特尔.wait_kills(23).retreat() +红.wait_kills(25).pre_delay(10000).deploy((3, 5), Direction.LEFT) +红.retreat() +史尔特尔.wait_kills(29).rear_delay(1500).deploy((9, 4), Direction.LEFT) +红.pre_delay(1000).deploy((4, 4), Direction.DOWN) +史尔特尔.use_skill() +史尔特尔.wait_kills(34).retreat() +红.pre_delay(5000).deploy((3, 5), Direction.LEFT) + +# print(json.dumps(危机合约十八.to_dict())) +# 保存为json文件 +危机合约十八.save_as_json("YanFengRongDong.json") +# 直接进行测试 要用的需要点进去配置一下 不然直接注释掉也行( +危机合约十八.test() From 46e3e419080e67e9690ec1aa6705a4649b393484 Mon Sep 17 00:00:00 2001 From: AutumnSound Date: Sat, 28 May 2022 22:29:03 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=E9=87=8D=E6=96=B0=E9=85=8D=E7=BD=AEasst?= =?UTF-8?q?=E5=AF=BC=E5=85=A5=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/CopilotDesinger/CopilotDesinger.py | 13 +- tools/CopilotDesinger/asst.py | 207 ----------------------- 2 files changed, 10 insertions(+), 210 deletions(-) delete mode 100644 tools/CopilotDesinger/asst.py diff --git a/tools/CopilotDesinger/CopilotDesinger.py b/tools/CopilotDesinger/CopilotDesinger.py index 71b0c1159c..d4301165a2 100644 --- a/tools/CopilotDesinger/CopilotDesinger.py +++ b/tools/CopilotDesinger/CopilotDesinger.py @@ -317,9 +317,16 @@ class Battle: try: from asst import Asst, Message except ImportError: - print("asst导入失败,请确保asst.py在同目录下,或者下载 " - "https://github.com/MaaAssistantArknights/MaaAssistantArknights/blob/master/src/Python/asst.py") - return + import sys + import pathlib + sys.path.append(str(pathlib.Path.cwd().parent.parent/"src"/"Python")) + try: + from asst import Asst, Message + except ImportError: + print("asst导入失败,请自行解决,或者下载" + " https://github.com/MaaAssistantArknights/MaaAssistantArknights/blob/master/src/Python/asst.py " + "到同目录下") + return @Asst.CallBackType def callback(msg, details, arg): diff --git a/tools/CopilotDesinger/asst.py b/tools/CopilotDesinger/asst.py deleted file mode 100644 index b822c2529f..0000000000 --- a/tools/CopilotDesinger/asst.py +++ /dev/null @@ -1,207 +0,0 @@ -import ctypes -import os -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: Union[pathlib.Path, str]) -> bool: - """ - 加载 dll 及资源 - - :params: - ``path``: DLL及资源所在文件夹路径 - """ - if platform.system().lower() == 'windows': - Asst.__libpath = pathlib.Path(path) / 'MeoAssistant.dll' - os.environ["PATH"] += os.pathsep + str(path) - Asst.__lib = ctypes.WinDLL(str(Asst.__libpath)) - else: - Asst.__libpath = pathlib.Path(path) / 'libMeoAssistant.so' - os.environ['LD_LIBRARY_PATH'] += os.pathsep + str(path) - Asst.__lib = ctypes.CDLL(str(Asst.__libpath)) - Asst.__set_lib_properties() - - return Asst.__lib.AsstLoadResource(str(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, ensure_ascii=False).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, ensure_ascii=False).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')) - - @staticmethod - def get_version() -> 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()