添加python利用interface运行任务

.gitignore

@@ -446,3 +446,6 @@ install
 
 # Tools
 tools/ImageCropper/**/*.png
+
+# PYTHON venv
+piclivenv/

pypicli/NotificationHandler/MwaNotificationHandler.py

@@ -0,0 +1,48 @@
+from maa.notification_handler import NotificationHandler, NotificationType
+
+
+#自定义通知处理器类
+class MwaNotificationHandler(NotificationHandler):
+    # 处理资源加载的通知
+    def on_resource_loading(
+        self,
+        noti_type: NotificationType,
+        detail: NotificationHandler.ResourceLoadingDetail,
+    ):
+        print(f"资源加载中: {noti_type}, {detail}")
+
+    # 处理控制器动作的通知
+    def on_controller_action(
+        self,
+        noti_type: NotificationType,
+        detail: NotificationHandler.ControllerActionDetail,
+    ):
+        print(f"控制器动作: {noti_type}, {detail}")
+
+    # 处理任务器任务的通知
+    def on_tasker_task(
+        self, noti_type: NotificationType, detail: NotificationHandler.TaskerTaskDetail
+    ):
+        print(f"任务器任务: {noti_type}, {detail}")
+
+    # 处理任务的下一个列表的通知
+    def on_task_next_list(
+        self,
+        noti_type: NotificationType,
+        detail: NotificationHandler.TaskNextListDetail,
+    ):
+        print(f"下一个任务: {noti_type}, {detail}")
+
+    # 处理任务识别的通知
+    def on_task_recognition(
+        self,
+        noti_type: NotificationType,
+        detail: NotificationHandler.TaskRecognitionDetail,
+    ):
+        print(f"任务识别: {noti_type}, {detail}")
+
+    # 处理任务动作的通知
+    def on_task_action(
+        self, noti_type: NotificationType, detail: NotificationHandler.TaskActionDetail
+    ):
+        print(f"任务动作: {noti_type}, {detail}")

pypicli/app.py

@@ -0,0 +1,48 @@
+import json
+from maa.toolkit import Toolkit
+from maa.resource import Resource
+from maa.controller import AdbController
+from maa.tasker import Tasker
+
+from NotificationHandler.MwaNotificationHandler import MwaNotificationHandler
+
+
+def tasker_init():
+    user_path = "./"
+    Toolkit.init_option(user_path)
+
+    resource = Resource()
+    resource.set_cpu()
+
+    res_job = resource.post_path("./resource_picli/base")
+    res_job.wait()
+
+    adb_devices = Toolkit.find_adb_devices()
+    if not adb_devices:
+        print("No ADB device found.")
+        exit()
+
+    # for demo, we just use the first device
+    device = adb_devices[0]
+    controller = AdbController(
+        adb_path=device.adb_path,
+        address=device.address,
+        screencap_methods=device.screencap_methods,
+        input_methods=device.input_methods,
+        config=device.config
+    )
+    controller.post_connection().wait()
+
+    # tasker = Tasker()
+    tasker = Tasker(notification_handler=MwaNotificationHandler())
+    tasker.bind(resource, controller)
+
+    if not tasker.inited:
+        print("Failed to init MAA.")
+        exit()
+    return tasker 
+
+
+if __name__ == "__main__":
+    # main()
+    pass

pypicli/config/maa_option.json

@@ -0,0 +1,7 @@
+{
+    "logging": true,
+    "recording": false,
+    "save_draw": false,
+    "show_hit_draw": false,
+    "stdout_level": 2
+}

pypicli/deal_tasks/deal_task_list.py

@@ -0,0 +1,143 @@
+import os
+from deal_tasks.get_all_tasks import get_all_tasks
+from deal_tasks.get_task_entry import get_all_entry
+
+# 添加任务列表的函数
+def add_task_list(all_task_list,task_option):
+    task_list = []
+
+    while True:
+        os.system("cls")
+        # 显示所有任务
+        print("添加任务")
+        add_task_num = 1
+        if all_task_list:
+            for i in all_task_list:
+                print(add_task_num,".",i,end="\n")
+                add_task_num += 1
+        # 显示已添加的任务
+        print("已添加任务：")
+        added_task_num = 1
+        for i in task_list:
+            try:
+                i = dict(i)
+                if type(i) == dict:
+                    for key,value in i.items():
+                        print(added_task_num,".",key,":",value,end="\n")
+                added_task_num += 1
+            except:
+                print(added_task_num,".",i,end="\n")
+                added_task_num += 1
+        print("\n")
+        # 输入任务编号
+        task_id = input("请输入任务编号,输入0退出：")
+        # 将输入的任务编号转换为整数
+        try:
+            task_id = int(task_id)
+        except:
+            os.system("cls")
+            continue
+        # 判断输入的任务编号是否有效
+        if task_id > len(all_task_list) or task_id < 0:
+            print("输入错误，请重新输入！")
+            os.system("cls")
+        elif task_id == 0:
+            os.system("cls")
+            break
+        else:
+            if task_option[task_id-1] != " ":
+                print(task_option[task_id-1])
+                option_name = get_option_task(task_option[task_id-1])
+                if get_all_entry(option_name) == ([],[],[],[]):
+                    task_list.append({f'{all_task_list[task_id-1]}':f'{option_name}'})
+            else:
+                task_list.append(all_task_list[task_id-1])
+                os.system("cls")
+        """
+        [' ', 
+        ['征集几个'], 
+        ['领取资源、好感、售卖机', '社教模拟仪', '制造物品', '是否喝茶', '是否领取活力值'], 
+        ' ', 
+        ['和合版本'], 
+        ' ', 
+        ' ', 
+        ' ', 
+        ' ', 
+        ['选择关卡-演训', '打哪个-演训', '打几次'], 
+        ['选择关卡-资源', '打哪个-资源', '打几次'], 
+        ' ',
+        ' ',
+        ['选择研学主题', '选择地图(仅老研学)', '是否自行选择角色和buff'], 
+        ['分解的装备的最大品质'], 
+        ' ', 
+        ' ']
+        """
+    return task_list
+
+# 删除任务列表的函数
+def delete_task_list(task_list):
+    os.system("cls")
+    while True:
+        num = 1
+        print("删除任务")
+
+        print("已添加任务：")
+
+        for i in task_list:
+            print(num,".",i,end=",")
+            num += 1
+        print("\n")
+
+        task_id = int(input("请输入任务编号,输入0退出："))
+
+        if task_id > len(task_list) or task_id < 0:
+            print("输入错误，请重新输入！")
+            os.system("cls")
+        elif task_id == 0:
+            break
+        else:
+            task_list.pop(task_id-1)
+            os.system("cls")
+
+    return task_list
+
+# 获取任务选项的函数
+def get_option_task(task_option):
+
+    # task_option = [
+    #             "领取资源、好感、售卖机",
+    #             "社教模拟仪",
+    #             "制造物品",
+    #             "是否喝茶",
+    #             "是否领取活力值"
+    #         ]
+     # ['领取奖励', '不领取']  
+     # ['entry_dispath_one', ['entry_back', 'entry_convened']]
+    task_list = []
+    for task_name in task_option:
+        pipeline_override,all_cases_names,all_cases_pipelines,all_tasks_options = get_all_entry(task_name)
+        os.system("cls")
+        num = 1
+        print(f"{task_name}")
+        for i in all_cases_names:
+            print(num,".",i,end="\n")
+            num += 1
+        # 输入任务编号
+        task_id = input("请输入任务编号,输入0退出：")
+        # 将输入的任务编号转换为整数
+        try:
+            task_id = int(task_id)
+            task_list.append(all_cases_names[task_id-1])
+            os.system("cls")
+        except:
+            os.system("cls")
+            continue
+    return task_list
+
+# 主程序入口
+if __name__ == '__main__':
+    tasks_names, tasks_entry, task_option = get_all_tasks()
+
+    task_list = add_task_list(tasks_names,task_option)
+    print(task_list)
+    # get_option_task()

pypicli/deal_tasks/get_all_tasks.py

@@ -0,0 +1,38 @@
+import codecs
+import os
+import json
+from deal_tasks.get_task_entry import get_all_entry
+
+def get_all_tasks()->list:
+    """
+    Get all single tasks of a project.
+    :param project_name: The name of the project.
+    :return: A list of single tasks.
+    """
+
+    with codecs.open("interface.json", 'r', encoding='utf-8') as f:
+        interface = json.load(f)
+
+    tasks,task_names,task_options = [],[],[]
+    for task in interface.get("task"):
+        tasks.append(task.get("name"))
+        task_names.append(task.get("entry"))
+        if task.get("option"):
+            task_options.append(task.get("option"))
+        else:
+            task_options.append(" ")
+
+    return tasks, task_names, task_options
+
+
+
+if __name__ == '__main__':
+    tasks, task_names, task_options = get_all_tasks()
+    print(task_options)
+    for i in task_options:
+        if i == " ":
+            continue
+        for j in i:
+            print(get_all_entry(j))
+    # print(get_all_option_entry(""))
+

pypicli/deal_tasks/get_task_entry.py

@@ -0,0 +1,51 @@
+import codecs
+import json
+
+
+def get_all_entry(option_name:str):
+    with codecs.open("interface.json", 'r', encoding='utf-8') as f:
+        interface = json.load(f)
+    all_cases, all_cases_names, all_cases_pipelines,all_tasks_options,pipeline_override = {},[],[],[],[]
+    try:
+        for _allcases in interface.get("option").get(option_name).get("cases"):
+            all_cases[_allcases.get("name")] = _allcases.get("pipeline_override")
+        for _cases_name in all_cases.keys():
+            all_cases_names.append(_cases_name)
+        pipeline_override.append(list(list(all_cases.values())[0].keys())[0])
+        for _case in all_cases.values():
+            for _case_key in _case.keys():
+                for _case_value in _case.get(_case_key).values():
+                    all_cases_pipelines.append(_case_value)
+        return pipeline_override,all_cases_names,all_cases_pipelines,all_tasks_options
+    except:
+        for _entry in interface.get("task"):
+            if _entry.get("name") == option_name:
+                all_cases_names.append(_entry.get("name"))
+                all_cases_pipelines.append(_entry.get("entry"))
+                try:
+                    for _option in _entry.get("option"):
+                        # pipeline_override,all_cases_names,all_cases_pipelines,all_tasks_options = get_all_entry(_option)
+                        all_tasks_options.append(_option)
+                except:
+                    pass
+        return pipeline_override,all_cases_names,all_cases_pipelines,all_tasks_options
+
+
+    """
+    {'冬谷币': {'StartTraining': {'next': 'Cash   '}},
+    '教材': {'StartTraining': {'next': 'Experience'}},
+    '装备': {'StartTraining': {'next': 'Weapon'}},
+    '宿卫': {'StartTraining': {'next': 'Defender'}},
+    '构术': {'StartTraining': {'next': 'Caster'}},
+    '远击': {'StartTraining': {'next': 'Ranger'}},
+    '轻锐': {'StartTraining': {'next': 'LightMelee'}},
+    '战略': {'StartTraining': {'next': 'Tactical'}}}
+    """
+
+
+if __name__ == '__main__':
+    # print(get_all_entry("器者征集"))
+    # print(get_all_entry("领取奖励"))
+    print(get_all_entry("社教模拟仪"))
+    # print(get_all_entry("选择关卡-演训"))
+