diff --git a/.gitignore b/.gitignore index 470d849..7b20a98 100644 --- a/.gitignore +++ b/.gitignore @@ -161,3 +161,7 @@ cython_debug/ # nicegui .nicegui/ + +# maafw +config +debug diff --git a/main.py b/main.py index 709dca3..82156fe 100644 --- a/main.py +++ b/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") diff --git a/requirements.txt b/requirements.txt index 64a2d14..cd793d1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,7 @@ +### from MaaFw numpy Pillow +### from MaaDebugger nicegui +asyncer \ No newline at end of file diff --git a/source/control/runner.py b/source/control/runner.py deleted file mode 100644 index e6ae55e..0000000 --- a/source/control/runner.py +++ /dev/null @@ -1,91 +0,0 @@ -from pathlib import Path -import sys - -specified_binding_dir = None -specified_bin_dir = None - -if len(sys.argv) > 2: - specified_binding_dir = Path(sys.argv[1]) - specified_bin_dir = Path(sys.argv[2]) - print( - f"specified binding_dir: {specified_binding_dir}, bin_dir: {specified_bin_dir}" - ) - -latest_install_dir = None -latest_adb_path = None -latest_adb_address = None -latest_resource_dir = None - -resource = None -controller = None -instance = None - - -async def run_task( - install_dir: Path, adb_path: Path, adb_address: str, resource_dir: Path, task: str -): - binding_dir = ( - specified_binding_dir - and specified_binding_dir - or (install_dir / "binding" / "Python") - ) - if not binding_dir.exists(): - return "Binding directory does not exist" - - binding_dir = str(binding_dir) - if binding_dir not in sys.path: - sys.path.insert(0, binding_dir) - - from maa.library import Library - from maa.resource import Resource - from maa.controller import AdbController - from maa.instance import Instance - from maa.toolkit import Toolkit - - global latest_install_dir, latest_adb_path, latest_adb_address - - if latest_install_dir != install_dir: - bin_dir = specified_bin_dir and specified_bin_dir or (install_dir / "bin") - version = Library.open(bin_dir) - if not version: - return "Failed to open MaaFramework" - print(f"MaaFw Version: {version}") - - latest_install_dir = install_dir - - Toolkit.init_option("./") - - global resource, controller, instance - - if latest_adb_path != adb_path or latest_adb_address != adb_address: - controller = AdbController(adb_path, adb_address) - connected = await controller.connect() - if not connected: - return "Failed to connect to ADB" - - latest_adb_path = adb_path - latest_adb_address = adb_address - - if not resource: - resource = Resource() - - # reload every time - loaded = resource.clear() and await resource.load(resource_dir) - if not loaded: - return "Failed to load resource" - - if not instance: - instance = Instance() - - instance.bind(resource, controller) - inited = instance.inited - if not inited: - return "Failed to init MaaFramework instance" - - ret = await instance.run_task(task, {}) - return f"Task returned: {ret}" - - -async def stop_task(): - if instance: - await instance.stop() diff --git a/source/interaction/maafw.py b/source/interaction/maafw.py new file mode 100644 index 0000000..f9f63b3 --- /dev/null +++ b/source/interaction/maafw.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) diff --git a/source/webpage/components/screenshotter.py b/source/webpage/components/screenshotter.py new file mode 100644 index 0000000..fa10990 --- /dev/null +++ b/source/webpage/components/screenshotter.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 diff --git a/source/webpage/components/status_indicator.py b/source/webpage/components/status_indicator.py new file mode 100644 index 0000000..765766b --- /dev/null +++ b/source/webpage/components/status_indicator.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 "❌" diff --git a/source/webpage/index.py b/source/webpage/index.py index 3a98e8b..c738f02 100644 --- a/source/webpage/index.py +++ b/source/webpage/index.py @@ -1,66 +1,295 @@ -from nicegui import app, ui +from nicegui import app, ui, binding from pathlib import Path -from source.control.runner import run_task, stop_task +import source.interaction.maafw as maafw + +from .components.status_indicator import Status, StatusIndicator +from .components.screenshotter import Screenshotter + + +binding.MAX_PROPAGATION_TIME = 1 +ui.dark_mode() # auto dark mode @ui.page("/") async def index(): - maafw_install_dir_input = ( + + with ui.row().style("align-items: center;"): + await import_maa_control() + + ui.separator() + + with ui.row(): + with ui.column(): + with ui.row().style("align-items: center;"): + await connect_adb_control() + with ui.row().style("align-items: center;"): + await load_resource_control() + with ui.row().style("align-items: center;"): + await run_task_control() + + with ui.column(): + await screenshot_control() + + ui.separator() + + +class GlobalStatus: + maa_importing: Status = Status.PENDING + adb_connecting: Status = Status.PENDING + adb_detecting: Status = Status.PENDING # not required + res_loading: Status = Status.PENDING + task_running: Status = Status.PENDING + + +screenshotter = Screenshotter() + + +async def import_maa_control(): + + StatusIndicator(GlobalStatus, "maa_importing") + + pybinding_input = ( + ui.input( + "MaaFramework Python Binding Directory", + placeholder="eg: C:/Downloads/MAA-win-x86_64/binding/Python", + ) + .props("size=60") + .bind_value(app.storage.general, "maa_pybinding") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s != Status.SUCCESS + ) + ) + bin_input = ( ui.input( - "MaaFramework Release Directory", - placeholder="eg: C:/Downloads/MAA-win-x86_64", + "MaaFramework Binary Directory", + placeholder="eg: C:/Downloads/MAA-win-x86_64/bin", ) .props("size=60") - .bind_value(app.storage.general, "maafw_install_dir") + .bind_value(app.storage.general, "maa_bin") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s != Status.SUCCESS + ) + ) + + import_button = ui.button( + "Import", on_click=lambda: on_click_import() + ).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s != Status.SUCCESS ) + async def on_click_import(): + GlobalStatus.maa_importing = Status.RUNNING + + if not pybinding_input.value or not bin_input.value: + GlobalStatus.maa_importing = Status.FAILURE + return + + imported = await maafw.import_maa( + Path(pybinding_input.value), Path(bin_input.value) + ) + if not imported: + GlobalStatus.maa_importing = Status.FAILURE + return + + GlobalStatus.maa_importing = Status.SUCCESS + + +async def connect_adb_control(): + StatusIndicator(GlobalStatus, "adb_connecting") + adb_path_input = ( - ui.input("ADB Path", placeholder="eg: C:/adb.exe") + ui.input( + "ADB Path", + placeholder="eg: C:/adb.exe", + ) .props("size=60") .bind_value(app.storage.general, "adb_path") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) ) - adb_address_input = ( - ui.input("ADB Address", placeholder="eg: 127.0.0.1:5555") - .props("size=60") + ui.input( + "ADB Address", + placeholder="eg: 127.0.0.1:5555", + ) + .props("size=30") .bind_value(app.storage.general, "adb_address") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + ) + ui.button( + "Connect", + on_click=lambda: on_click_connect(), + ).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + + ui.button( + "Detect", + on_click=lambda: on_click_detect(), + ).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + + devices_select = ( + ui.select({}, on_change=lambda e: on_change_devices_select(e)) + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + .bind_visibility_from( + GlobalStatus, + "adb_detecting", + backward=lambda s: s == Status.SUCCESS, + ) ) - resource_dir_input = ( - ui.input("Resource Directory", placeholder="eg: C:/M9A/assets/resource") + StatusIndicator(GlobalStatus, "adb_detecting").label().bind_visibility_from( + GlobalStatus, + "adb_detecting", + backward=lambda s: s == Status.RUNNING or s == Status.FAILURE, + ) + + async def on_click_connect(): + GlobalStatus.adb_connecting = Status.RUNNING + + if not adb_path_input.value or not adb_address_input.value: + GlobalStatus.adb_connecting = Status.FAILURE + return + + connected = await maafw.connect_adb( + Path(adb_path_input.value), adb_address_input.value + ) + if not connected: + GlobalStatus.adb_connecting = Status.FAILURE + return + + GlobalStatus.adb_connecting = Status.SUCCESS + GlobalStatus.adb_detecting = Status.PENDING + + screenshotter.start() + + async def on_click_detect(): + GlobalStatus.adb_detecting = Status.RUNNING + + devices = await maafw.detect_adb() + options = {} + for d in devices: + v = (d.adb_path, d.address) + l = d.name + " " + d.address + options[v] = l + + devices_select.options = options + devices_select.update() + if not options: + GlobalStatus.adb_detecting = Status.FAILURE + + devices_select.value = next(iter(options)) + GlobalStatus.adb_detecting = Status.SUCCESS + + def on_change_devices_select(e): + adb_path_input.value = str(e.value[0]) + adb_address_input.value = e.value[1] + + +async def screenshot_control(): + with ui.card().tight(): + ui.interactive_image( + cross="green", + on_mouse=lambda e: on_click_image(int(e.image_x), int(e.image_y)), + ).bind_source_from(screenshotter, "source").style( + "height: 200px;" + ).bind_visibility_from( + GlobalStatus, "adb_connecting", backward=lambda s: s == Status.SUCCESS + ) + + async def on_click_image(x, y): + print(f"on_click_image: {x}, {y}") + await maafw.click(x, y) + + +async def load_resource_control(): + StatusIndicator(GlobalStatus, "res_loading") + + dir_input = ( + ui.input( + "Resource Directory", + placeholder="eg: C:/M9A/assets/resource/base", + ) .props("size=60") .bind_value(app.storage.general, "resource_dir") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) ) - with ui.row(): - task_input = ( - ui.input("Task", placeholder="Enter the task") - .props("size=30") - .bind_value(app.storage.general, "task") - ) + ui.button( + "Load", + on_click=lambda: on_click_load(), + ).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + + async def on_click_load(): + GlobalStatus.res_loading = Status.RUNNING + + if not dir_input.value: + GlobalStatus.res_loading = Status.FAILURE + return - run_button = ui.button("Run", on_click=on_run_button_click) - stop_button = ui.button("Stop", on_click=on_stop_button_click) + loaded = await maafw.load_resource(Path(dir_input.value)) + if not loaded: + GlobalStatus.res_loading = Status.FAILURE + return + GlobalStatus.res_loading = Status.SUCCESS -async def on_run_button_click(): - maafw_install_dir = Path(app.storage.general.get("maafw_install_dir")) - adb_path = Path(app.storage.general.get("adb_path")) - adb_address = app.storage.general.get("adb_address") - resource_dir = Path(app.storage.general.get("resource_dir")) - task = app.storage.general.get("task") - print( - f"on_run_button_click: maafw_install_dir: {maafw_install_dir}, adb_path: {adb_path}, adb_address: {adb_address}, resource_dir: {resource_dir}, task: {task}" +async def run_task_control(): + StatusIndicator(GlobalStatus, "task_running") + + entry_input = ( + ui.input( + "Task Entry", + placeholder="eg: StartUp", + ) + .props("size=30") + .bind_value(app.storage.general, "task_entry") + .bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) ) - message = await run_task( - maafw_install_dir, adb_path, adb_address, resource_dir, task + ui.button("Start", on_click=lambda: on_click_start()).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS + ) + ui.button("Stop", on_click=lambda: on_click_stop()).bind_enabled_from( + GlobalStatus, "maa_importing", backward=lambda s: s == Status.SUCCESS ) - ui.notify(message) + async def on_click_start(): + screenshotter.stop() + + GlobalStatus.task_running = Status.RUNNING + + if not entry_input.value: + GlobalStatus.task_running = Status.FAILURE + return + + run = await maafw.run_task(entry_input.value) + if not run: + GlobalStatus.task_running = Status.FAILURE + return + + GlobalStatus.task_running = Status.SUCCESS + + async def on_click_stop(): + stopped = await maafw.stop_task() + if not stopped: + GlobalStatus.task_running = Status.FAILURE + return -async def on_stop_button_click(): - await stop_task() - ui.notify("Task stopped") + GlobalStatus.task_running = Status.PENDING + screenshotter.start()