feat: 允许上传文件作为待匹配图片

Author: weinibuliu

src/MaaDebugger/maafw/__init__.py

@@ -1,25 +1,44 @@
 import re
+import io
 from pathlib import Path
 from typing import Callable, List, Optional, Tuple, Union
 
 from asyncify import asyncify
 from PIL import Image
-from maa.controller import AdbController, Win32Controller
+from maa.controller import AdbController, Win32Controller, CustomController
 from maa.context import Context, ContextEventSink
 from maa.tasker import Tasker, RecognitionDetail
 from maa.resource import Resource, ResourceEventSink
 from maa.toolkit import Toolkit, AdbDevice, DesktopWindow
 from maa.agent_client import AgentClient
 from maa.library import Library
 from maa.event_sink import NotificationType
+import numpy as np
 
 from ..utils import cvmat_to_image
 
 
+class MyCustomController(CustomController):
+    def __init__(self, img_bytes: bytes):
+        super().__init__()
+
+        img = Image.open(io.BytesIO(img_bytes))
+        self.ndarray = np.array(img)
+
+    def connect(self) -> bool:
+        return True
+
+    def request_uuid(self) -> str:
+        return "0"
+
+    def screencap(self) -> np.ndarray:
+        return self.ndarray
+
+
 class MaaFW:
 
     resource: Optional[Resource]
-    controller: Union[AdbController, Win32Controller, None]
+    controller: Union[AdbController, Win32Controller, CustomController, None]
     tasker: Optional[Tasker]
     agent: Optional[AgentClient]
     context_event_sink: Optional[ContextEventSink]
@@ -89,6 +108,12 @@ def connect_win32hwnd(
 
         return True, None
 
+    def connect_custom_controller(self, img_bytes) -> Tuple[bool, Optional[str]]:
+        self.controller = MyCustomController(img_bytes)
+        self.controller.post_connection().wait()
+
+        return True, None
+
     @asyncify
     def load_resource(self, dir: List[Path]) -> Tuple[bool, Optional[str]]:
         if not self.resource:
@@ -149,7 +174,20 @@ def run_task(
         if not AgentClient().register_sink(self.resource, self.controller, self.tasker):
             return False, "Failed to register Agent sink."
 
-        return self.tasker.post_task(entry, pipeline_override).wait().succeeded, None
+        if isinstance(self.controller, CustomController):
+            # disable action
+            pipeline_override.update(
+                {entry: {"action": {"type": "DoNothing"}, "next": []}}
+            )
+            return (
+                self.tasker.post_task(entry, pipeline_override).wait().succeeded,
+                None,
+            )
+        else:
+            return (
+                self.tasker.post_task(entry, pipeline_override).wait().succeeded,
+                None,
+            )
 
     @asyncify
     def stop_task(self) -> None:
@@ -176,6 +214,9 @@ def click(self, x, y) -> bool:
         if not self.controller:
             return False
 
+        if isinstance(self.controller, CustomController):
+            return False
+
         return self.controller.post_click(x, y).wait().succeeded
 
     @asyncify

src/MaaDebugger/webpage/index_page/master_control.py

@@ -39,6 +39,7 @@ def connect_control():
     with ui.tabs() as tabs:
         adb = ui.tab("Adb")
         win32 = ui.tab("Win32")
+        custom = ui.tab("Custom")
 
     tab_panels = ui.tab_panels(tabs, value="Adb").bind_value(STORAGE, "controller_type")
     with tab_panels:
@@ -49,6 +50,10 @@ def connect_control():
             with ui.row(align_items="center").classes("w-full"):
                 connect_win32_control()
 
+        with ui.tab_panel(custom):
+            with ui.row(align_items="center").classes("w-full"):
+                connect_custom_control()
+
     os_type = system.get_os_type()
     if os_type != system.OSTypeEnum.Windows:
         win32.disable()
@@ -301,6 +306,22 @@ def on_change_hwnd_select(value: Optional[str]):
         hwnd_input.value = value
 
 
+def connect_custom_control():
+    def on_upload(e):
+        GlobalStatus.ctrl_connecting = Status.RUNNING
+        try:
+            maafw.connect_custom_controller(e.content.read())
+        except:
+            GlobalStatus.ctrl_connecting = Status.FAILED
+            ui.notify("Failed to load image.", position="bottom-right", type="negative")
+            return
+
+        GlobalStatus.ctrl_connecting = Status.SUCCEEDED
+
+    StatusIndicator(GlobalStatus, "ctrl_connecting")
+    ui.upload(auto_upload=True, on_upload=lambda e: on_upload(e))
+
+
 def screenshot_control():
     with (
         ui.row()