refactor: 重构基础控件 (#1)

Author: MistEO

.gitignore

@@ -161,3 +161,7 @@ cython_debug/
 
 # nicegui
 .nicegui/
+
+# maafw
+config
+debug

main.py

@@ -2,4 +2,4 @@
 
 from source.webpage import index
 
-ui.run(title="Maa Debugger", storage_secret="maadbg")
+ui.run(title="Maa Debugger", storage_secret="maadbg")#, root_path="/proxy/8080")

requirements.txt

@@ -1,4 +1,7 @@
+### from MaaFw
 numpy
 Pillow
 
+### from MaaDebugger
 nicegui
+asyncer

source/control/runner.py

@@ -0,0 +1,120 @@
+from pathlib import Path
+from typing import List, Optional
+from asyncer import asyncify
+from PIL import Image
+
+import sys
+
+async def import_maa(binding_dir: Path, bin_dir: Path) -> bool:
+    if not binding_dir.exists():
+        print("Binding directory does not exist")
+        return False
+
+    if not bin_dir.exists():
+        print("Bin dir does not exist")
+        return False
+
+    binding_dir = str(binding_dir)
+    if binding_dir not in sys.path:
+        sys.path.insert(0, binding_dir)
+
+    try:
+        from maa.library import Library
+        from maa.toolkit import Toolkit
+    except ModuleNotFoundError as err:
+        print(err)
+        return False
+
+    version = await asyncify(Library.open)(bin_dir)
+    if not version:
+        print("Failed to open MaaFramework")
+        return False
+
+    print(f"Import MAA successfully, version: {version}")
+
+    Toolkit.init_option("./")
+
+    return True
+
+
+async def detect_adb() -> List["AdbDevice"]:
+    from maa.toolkit import Toolkit
+
+    return await Toolkit.adb_devices()
+
+
+resource = None
+controller = None
+instance = None
+
+
+async def connect_adb(path: Path, address: str) -> bool:
+    global controller
+
+    from maa.controller import AdbController
+
+    controller = AdbController(path, address)
+    connected = await controller.connect()
+    if not connected:
+        print(f"Failed to connect {path} {address}")
+        return False
+
+    return True
+
+
+async def load_resource(dir: Path) -> bool:
+    global resource
+
+    from maa.resource import Resource
+
+    if not resource:
+        resource = Resource()
+
+    return resource.clear() and await resource.load(dir)
+
+
+async def run_task(entry: str, param: dict = {}) -> bool:
+    global controller, resource, instance
+
+    from maa.instance import Instance
+
+    if not instance:
+        instance = Instance()
+
+    instance.bind(resource, controller)
+    if not instance.inited:
+        print("Failed to init MaaFramework instance")
+        return False
+
+    return await instance.run_task(entry, param)
+
+
+async def stop_task():
+    global instance
+
+    if not instance:
+        return
+
+    await instance.stop()
+
+
+async def screencap() -> Optional[Image]:
+    global controller
+    if not controller:
+        return None
+
+    im = await controller.screencap()
+    if im is None:
+        return None
+
+    pil = Image.fromarray(im)
+    b, g, r = pil.split()
+    return Image.merge("RGB", (r, g, b))
+
+
+async def click(x, y) -> None:
+    global controller
+    if not controller:
+        return None
+
+    await controller.click(x, y)

source/interaction/maafw.py

@@ -0,0 +1,30 @@
+import asyncio
+import threading
+
+import source.interaction.maafw as maafw
+
+
+class Screenshotter(threading.Thread):
+    def __init__(self):
+        super().__init__()
+        self.source = None
+        self.active = False
+
+    def __del__(self):
+        self.active = False
+        self.source = None
+
+    def run(self):
+        while self.active:
+            im = asyncio.run(maafw.screencap())
+            if not im:
+                continue
+
+            self.source = im
+
+    def start(self):
+        self.active = True
+        super().start()
+
+    def stop(self):
+        self.active = False

source/webpage/components/screenshotter.py

@@ -0,0 +1,33 @@
+from nicegui import ui
+from enum import Enum, auto
+
+
+class Status(Enum):
+    PENDING = auto()
+    RUNNING = auto()
+    SUCCESS = auto()
+    FAILURE = auto()
+
+
+class StatusIndicator:
+    def __init__(self, target_object, target_name):
+        self._label = ui.label().bind_text_from(
+            target_object,
+            target_name,
+            backward=lambda s: StatusIndicator._text_backward(s),
+        )
+
+    def label(self):
+        return self._label
+
+    @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 "❌"