feat: 支持 maafw 2.0

README.md

@@ -2,7 +2,7 @@
 
 ## Requirement
 
-Python >= 3.11
+Python >= 3.9
 
 ## Installation
 

pyproject.toml

@@ -3,10 +3,10 @@ name = "MaaDebugger"
 version = "v0.0.1"
 description = "MaaDebugger"
 authors = [{ name = "MaaXYZ" }]
-dependencies = ["MaaFw>=1.7.3", "nicegui"]
+dependencies = ["MaaFw>=2.0.0", "nicegui"]
 readme = "README.md"
 urls = { Homepage = "https://github.com/MaaXYZ/MaaDebugger" }
-requires-python = ">=3.11"
+requires-python = ">=9"
 scripts = { MaaDebugger = "MaaDebugger.main:main" }
 
 [build-system]

requirements.txt

@@ -1,2 +1,2 @@
-maafw>=1.7.3
+maafw>=2.0.0
 nicegui

src/MaaDebugger/maafw/__init__.py

@@ -1,30 +1,33 @@
-import asyncio
-from dataclasses import dataclass
+import re
+from asyncify import asyncify
 from pathlib import Path
 from typing import Callable, List, Optional
 
 from maa.controller import AdbController, Win32Controller
-from maa.instance import Instance
+from maa.tasker import Tasker, RecognitionDetail
 from maa.resource import Resource
-from maa.toolkit import Toolkit
+from maa.toolkit import Toolkit, AdbDevice, DesktopWindow
 from PIL import Image
 
 from ..utils import cvmat_to_image
 
 
 class MaaFW:
+
+    tasker: Tasker
+
     def __init__(
         self,
         on_list_to_recognize: Callable = None,
         on_miss_all: Callable = None,
         on_recognition_result: Callable = None,
     ):
         Toolkit.init_option("./")
-        Instance.set_debug_message(True)
+        Tasker.set_debug_message(True)
 
         self.resource = None
         self.controller = None
-        self.instance = None
+        self.tasker = None
 
         self.screenshotter = Screenshotter(self.screencap)
 
@@ -33,108 +36,121 @@ def __init__(
         self.on_recognition_result = on_recognition_result
 
     @staticmethod
-    async def detect_adb() -> List["AdbDevice"]:
-        return await Toolkit.adb_devices()
-
-    @dataclass
-    class Window:
-        hwnd: int
-        class_name: str
-        window_name: str
+    @asyncify
+    def detect_adb() -> List[AdbDevice]:
+        return Toolkit.find_adb_devices()
 
     @staticmethod
-    async def detect_win32hwnd(class_regex: str, window_regex: str) -> List[Window]:
-        hwnds = Toolkit.search_window(class_regex, window_regex)
-        windows = []
-        for hwnd in hwnds:
-            class_name = Toolkit.get_class_name(hwnd)
-            window_name = Toolkit.get_window_name(hwnd)
-            windows.append(MaaFW.Window(hwnd, class_name, window_name))
+    @asyncify
+    def detect_win32hwnd(window_regex: str) -> List[DesktopWindow]:
+        windows = Toolkit.find_desktop_windows()
+        result = []
+        for win in windows:
+            if not re.search(window_regex, win.window_name):
+                continue
+
+            result.append(win)
 
-        return windows
+        return result
 
-    async def connect_adb(self, path: Path, address: str, config: dict) -> bool:
+    @asyncify
+    def connect_adb(self, path: Path, address: str, config: dict) -> bool:
         self.controller = AdbController(path, address, config=config)
-        connected = await self.controller.connect()
+        connected = self.controller.post_connection().wait().success()
         if not connected:
             print(f"Failed to connect {path} {address}")
             return False
 
         return True
 
-    async def connect_win32hwnd(
-        self, hwnd: int | str, screencap_type: int, input_type: int
+    @asyncify
+    def connect_win32hwnd(
+        self, hwnd: int | str, screencap_method: int, input_method: int
     ) -> bool:
         if isinstance(hwnd, str):
             hwnd = int(hwnd, 16)
 
         self.controller = Win32Controller(
-            hwnd, screencap_type=screencap_type, touch_type=input_type, key_type=0
+            hwnd, screencap_method=screencap_method, input_method=input_method
         )
-        connected = await self.controller.connect()
+        connected = self.controller.post_connection().wait().success()
         if not connected:
             print(f"Failed to connect {hwnd}")
             return False
 
         return True
 
-    async def load_resource(self, dir: Path) -> bool:
+    @asyncify
+    def load_resource(self, dir: Path) -> bool:
         if not self.resource:
             self.resource = Resource()
 
-        return self.resource.clear() and await self.resource.load(dir)
+        return self.resource.clear() and self.resource.post_path(dir).wait().success()
 
-    async def run_task(self, entry: str, param: dict = {}) -> bool:
-        if not self.instance:
-            self.instance = Instance(callback=self._inst_callback)
+    @asyncify
+    def run_task(self, entry: str, pipeline_override: dict = {}) -> bool:
+        if not self.tasker:
+            self.tasker = Tasker(callback=self._tasker_callback)
 
-        self.instance.bind(self.resource, self.controller)
-        if not self.instance.inited:
+        self.tasker.bind(self.resource, self.controller)
+        if not self.tasker.inited:
             print("Failed to init MaaFramework instance")
             return False
 
-        return await self.instance.run_task(entry, param)
+        return self.tasker.post_pipeline(entry, pipeline_override).wait()
 
-    async def stop_task(self):
-        if not self.instance:
+    @asyncify
+    def stop_task(self):
+        if not self.tasker:
             return
 
-        await self.instance.stop()
+        self.tasker.post_stop().wait()
 
-    async def screencap(self, capture: bool = True) -> Optional[Image.Image]:
+    @asyncify
+    def screencap(self, capture: bool = True) -> Optional[Image.Image]:
         if not self.controller:
             return None
 
-        im = await self.controller.screencap(capture)
+        if capture:
+            self.controller.post_screencap().wait()
+        im = self.controller.cached_image
         if im is None:
             return None
 
         return cvmat_to_image(im)
 
-    async def click(self, x, y) -> None:
+    @asyncify
+    def click(self, x, y) -> None:
         if not self.controller:
             return None
 
-        await self.controller.click(x, y)
+        self.controller.post_click(x, y).wait()
+
+    @asyncify
+    def get_reco_detail(self, reco_id: int) -> Optional[RecognitionDetail]:
+        if not self.tasker:
+            return None
+
+        return self.tasker._get_recognition_detail(reco_id)
 
-    def _inst_callback(self, msg: str, detail: dict, arg):
-        match msg:
-            case "Task.Debug.ListToRecognize":
-                asyncio.run(self.screenshotter.refresh(False))
-                if self.on_list_to_recognize:
-                    self.on_list_to_recognize(detail["pre_hit_task"], detail["list"])
+    def _tasker_callback(self, msg: str, detail: dict, arg):
+        if msg == "Task.Debug.ListToRecognize":
+            self.screenshotter.refresh(False)
+            if self.on_list_to_recognize:
+                self.on_list_to_recognize(detail["current"], detail["list"])
 
-            case "Task.Debug.MissAll":
-                if self.on_miss_all:
-                    self.on_miss_all(detail["pre_hit_task"], detail["list"])
+        elif msg == "Task.Debug.MissAll":
+            if self.on_miss_all:
+                self.on_miss_all(detail["current"], detail["list"])
 
-            case "Task.Debug.RecognitionResult":
-                reco_id = detail["recognition"]["id"]
-                name = detail["name"]
-                hit = detail["recognition"]["hit"]
+        elif msg == "Task.Debug.RecognitionResult":
+            reco = detail["recognition"]
+            reco_id = reco["reco_id"]
+            name = reco["name"]
+            hit = reco["box"] is not None
 
-                if self.on_recognition_result:
-                    self.on_recognition_result(reco_id, name, hit)
+            if self.on_recognition_result:
+                self.on_recognition_result(reco_id, name, hit)
 
 
 # class Screenshotter(threading.Thread):

src/MaaDebugger/webpage/components/status_indicator.py

@@ -23,12 +23,11 @@ def label(self):
 
     @staticmethod
     def _text_backward(status: Status) -> str:
-        match status:
-            case Status.PENDING:
-                return "🟡"
-            case Status.RUNNING:
-                return "👀"
-            case Status.SUCCESS:
-                return "✅"
-            case Status.FAILURE:
-                return "❌"
+        if status == Status.PENDING:
+            return "🟡"
+        elif status == Status.RUNNING:
+            return "👀"
+        elif status == Status.SUCCESS:
+            return "✅"
+        elif status == Status.FAILURE:
+            return "❌"

src/MaaDebugger/webpage/index_page/master_control.py

@@ -1,8 +1,8 @@
-import asyncio
 import json
+import time
 from pathlib import Path
 
-from maa.define import MaaWin32ControllerTypeEnum
+from maa.define import MaaWin32ScreencapMethodEnum, MaaWin32InputMethodEnum
 from nicegui import app, binding, ui
 
 from ...maafw import maafw
@@ -129,7 +129,7 @@ async def on_click_detect():
         devices = await maafw.detect_adb()
         options = {}
         for d in devices:
-            v = (d.adb_path, d.address, d.config)
+            v = (d.adb_path, d.address, str(d.config))
             l = d.name + " " + d.address
             options[v] = l
 
@@ -156,30 +156,21 @@ async def connect_win32_control():
     )
 
     SCREENCAP_DICT = {
-        MaaWin32ControllerTypeEnum.Screencap_GDI: "Screencap_GDI",
-        MaaWin32ControllerTypeEnum.Screencap_DXGI_DesktopDup: "Screencap_DXGI_DesktopDup",
-        MaaWin32ControllerTypeEnum.Screencap_DXGI_FramePool: "Screencap_DXGI_FramePool",
+        MaaWin32ScreencapMethodEnum.GDI: "Screencap_GDI",
+        MaaWin32ScreencapMethodEnum.DXGI_DesktopDup: "Screencap_DXGI_DesktopDup",
+        MaaWin32ScreencapMethodEnum.FramePool: "Screencap_DXGI_FramePool",
     }
     screencap_select = ui.select(
-        SCREENCAP_DICT, value=MaaWin32ControllerTypeEnum.Screencap_DXGI_DesktopDup
+        SCREENCAP_DICT, value=MaaWin32ScreencapMethodEnum.DXGI_DesktopDup
     ).bind_value(app.storage.general, "win32_screencap")
 
     INPUT_DICT = {
-        (
-            MaaWin32ControllerTypeEnum.Touch_SendMessage
-            | MaaWin32ControllerTypeEnum.Key_SendMessage
-        ): "Input_SendMessage",
-        (
-            MaaWin32ControllerTypeEnum.Touch_Seize
-            | MaaWin32ControllerTypeEnum.Key_Seize
-        ): "Input_Seize",
+        MaaWin32InputMethodEnum.SendMessage: "Input_SendMessage",
+        MaaWin32InputMethodEnum.Seize: "Input_Seize",
     }
     input_select = ui.select(
         INPUT_DICT,
-        value=(
-            MaaWin32ControllerTypeEnum.Touch_Seize
-            | MaaWin32ControllerTypeEnum.Key_Seize
-        ),
+        value=MaaWin32InputMethodEnum.Seize,
     ).bind_value(app.storage.general, "win32_input")
 
     ui.button(
@@ -273,8 +264,8 @@ async def screenshot_control():
     async def on_click_image(x, y):
         print(f"on_click_image: {x}, {y}")
         await maafw.click(x, y)
-        await asyncio.sleep(0.2)
-        await on_click_refresh()
+        time.sleep(0.2)
+        on_click_refresh()
 
     async def on_click_refresh():
         await maafw.screenshotter.refresh(True)

src/MaaDebugger/webpage/index_page/runtime_control.py

@@ -25,18 +25,18 @@ def register(self):
         maafw.on_miss_all = self.on_miss_all
         maafw.on_recognition_result = self.on_recognition_result
 
-    def on_list_to_recognize(self, pre_hit, list_to_reco):
+    def on_list_to_recognize(self, current, list_to_reco):
         self.row_len = self.row_len + 1
 
         self.cur_list = list_to_reco
         self.next_reco_index = 0
 
         with self.row:
-            self._add_list(pre_hit, list_to_reco)
+            self._add_list(current, list_to_reco)
 
-    def _add_list(self, pre_hit, list_to_reco):
+    def _add_list(self, current, list_to_reco):
         with ui.list().props("bordered separator"):
-            ui.item_label(pre_hit).props("header").classes("text-bold")
+            ui.item_label(current).props("header").classes("text-bold")
             ui.separator()
 
             for index in range(len(list_to_reco)):
@@ -87,7 +87,7 @@ def on_recognition_result(self, reco_id: int, name: str, hit: bool):
 
         RecoData.data[reco_id] = name, hit
 
-    def on_miss_all(self, pre_hit, list_to_reco):
+    def on_miss_all(self, current, list_to_reco):
         pass
 
 

src/MaaDebugger/webpage/reco_page/__init__.py

@@ -1,17 +1,16 @@
 from typing import Dict, Tuple
 
-from maa.instance import Instance
 from nicegui import ui
 
 from ...utils import cvmat_to_image
-
+from ...maafw import maafw, RecognitionDetail
 
 class RecoData:
     data: Dict[int, Tuple[str, bool]] = {}
 
 
 @ui.page("/reco/{reco_id}")
-def reco_page(reco_id: int):
+async def reco_page(reco_id: int):
     if reco_id == 0 or not reco_id in RecoData.data:
         ui.markdown("## Not Found")
         return
@@ -25,12 +24,12 @@ def reco_page(reco_id: int):
 
     ui.separator()
 
-    details = Instance.query_recognition_detail(reco_id)
+    details: RecognitionDetail = await maafw.get_reco_detail(reco_id)
     if not details:
         ui.markdown("## Not Found")
         return
 
-    ui.markdown(f"#### Hit: {str(details.hit_box)}")
+    ui.markdown(f"#### Hit: {str(details.box)}")
 
     for draw in details.draws:
         ui.image(cvmat_to_image(draw)).props("fit=scale-down")