From f01e0224cb912c2a8b1de6e10e39ba78fbaa4702 Mon Sep 17 00:00:00 2001 From: XcantloadX <3188996979@qq.com> Date: Tue, 24 Jun 2025 14:34:55 +0800 Subject: [PATCH] =?UTF-8?q?refactor(core):=20=E7=BB=84=E8=A3=85=20Device?= =?UTF-8?q?=20=E6=94=B9=E7=94=A8=20recipe=20=E6=96=B9=E6=A1=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原来组装 Device 的代码放在每个 Impl 文件下实现，通过 @register_impl 装饰器注册组装函数，然后通过统一接口 组装。现在将所有组装代码移动到了 Host 实现下，Impl 实现 只需要实现自身。 --- README.md | 3 + kotonebot/backend/context/context.py | 2 - kotonebot/client/__init__.py | 3 +- kotonebot/client/host/adb_common.py | 91 +++++++++++++++++++ kotonebot/client/host/custom.py | 48 +++++----- kotonebot/client/host/leidian_host.py | 37 +++----- kotonebot/client/host/mumu12_host.py | 57 ++++++------ kotonebot/client/host/protocol.py | 5 +- kotonebot/client/host/windows_common.py | 55 +++++++++++ kotonebot/client/implements/adb.py | 45 +-------- kotonebot/client/implements/adb_raw.py | 11 +-- .../client/implements/nemu_ipc/nemu_ipc.py | 39 ++++---- kotonebot/client/implements/remote_windows.py | 22 ++--- kotonebot/client/implements/uiautomator2.py | 10 +- kotonebot/client/implements/windows.py | 17 +--- kotonebot/client/registration.py | 70 -------------- kotonebot/config/base_config.py | 4 +- kotonebot/kaa/main/dmm_host.py | 45 +++------ kotonebot/kaa/main/kaa.py | 8 +- 19 files changed, 271 insertions(+), 301 deletions(-) create mode 100644 kotonebot/client/host/adb_common.py create mode 100644 kotonebot/client/host/windows_common.py diff --git a/README.md b/README.md index bc172f6..40e4aa9 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,9 @@ kaa 的开发主要用到了以下开源项目： ](https://github.com/AllenHeartcore/GkmasObjectManager)：用于提取游戏图像资源，以 GPLv3 协议开源。 * [gakumasu-diff](https://github.com/vertesan/gakumasu-diff)：游戏数据。 +kaa 的开发还参考了以下开源项目： +* [EmulatorExtras](https://github.com/MaaXYZ/EmulatorExtras)：MuMu 与雷电模拟器的截图与控制接口定义。 +* [blue_archive_auto_script](https://github.com/pur1fying/blue_archive_auto_script)：MuMu 与雷电模拟器的截图与控制接口的 Python 实现，以及各模拟器的控制实现。 ## 免责声明 **请在使用本项目前仔细阅读以下内容。使用本脚本将带来包括但不限于账号被封禁的风险。** diff --git a/kotonebot/backend/context/context.py b/kotonebot/backend/context/context.py index fb67b78..e3d5a3d 100644 --- a/kotonebot/backend/context/context.py +++ b/kotonebot/backend/context/context.py @@ -46,12 +46,10 @@ from kotonebot.backend.color import ( from kotonebot.backend.ocr import ( Ocr, OcrResult, OcrResultList, jp, en, StringMatchFunction ) -from kotonebot.client.registration import AdbBasedImpl, create_device from kotonebot.config.manager import load_config, save_config from kotonebot.config.base_config import UserConfig from kotonebot.backend.core import Image, HintBox from kotonebot.errors import KotonebotWarning -from kotonebot.client import DeviceImpl from kotonebot.backend.preprocessor import PreprocessorProtocol from kotonebot.primitives import Rect diff --git a/kotonebot/client/__init__.py b/kotonebot/client/__init__.py index 3bb8695..7c79ecc 100644 --- a/kotonebot/client/__init__.py +++ b/kotonebot/client/__init__.py @@ -1,11 +1,10 @@ from .device import Device -from .registration import create_device, DeviceImpl +from .registration import DeviceImpl # 确保所有实现都被注册 from . import implements # noqa: F401 __all__ = [ 'Device', - 'create_device', 'DeviceImpl', ] \ No newline at end of file diff --git a/kotonebot/client/host/adb_common.py b/kotonebot/client/host/adb_common.py new file mode 100644 index 0000000..b2643e8 --- /dev/null +++ b/kotonebot/client/host/adb_common.py @@ -0,0 +1,91 @@ +from abc import ABC +from typing import Literal, TypeVar +from typing_extensions import assert_never + +from adbutils import adb +from adbutils._device import AdbDevice +from kotonebot import logging +from kotonebot.client.device import AndroidDevice +from .protocol import Instance, AdbHostConfig, Device + +logger = logging.getLogger(__name__) +AdbRecipes = Literal['adb', 'adb_raw', 'uiautomator2'] + +def connect_adb( + ip: str, + port: int, + connect: bool = True, + disconnect: bool = True, + timeout: float = 180, + device_serial: str | None = None +) -> AdbDevice: + """ + 创建 ADB 连接。 + """ + if disconnect: + logger.debug('adb disconnect %s:%d', ip, port) + adb.disconnect(f'{ip}:{port}') + if connect: + logger.debug('adb connect %s:%d', ip, port) + result = adb.connect(f'{ip}:{port}') + if 'cannot connect to' in result: + raise ValueError(result) + serial = device_serial or f'{ip}:{port}' + logger.debug('adb wait for %s', serial) + adb.wait_for(serial, timeout=timeout) + devices = adb.device_list() + logger.debug('adb device_list: %s', devices) + d = [d for d in devices if d.serial == serial] + if len(d) == 0: + raise ValueError(f"Device {serial} not found") + d = d[0] + return d + +class CommonAdbCreateDeviceMixin(ABC): + """ + 通用 ADB 创建设备的 Mixin。 + 该 Mixin 定义了创建 ADB 设备的通用接口。 + """ + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + # 下面的属性只是为了让类型检查通过，无实际实现 + self.adb_ip: str + self.adb_port: int + self.adb_name: str + + def create_device(self, recipe: AdbRecipes, config: AdbHostConfig) -> Device: + """ + 创建 ADB 设备。 + """ + connection = connect_adb( + self.adb_ip, + self.adb_port, + connect=True, + disconnect=True, + timeout=config.timeout, + device_serial=self.adb_name + ) + d = AndroidDevice(connection) + match recipe: + case 'adb': + from kotonebot.client.implements.adb import AdbImpl + impl = AdbImpl(connection) + d._screenshot = impl + d._touch = impl + d.commands = impl + case 'adb_raw': + from kotonebot.client.implements.adb_raw import AdbRawImpl + impl = AdbRawImpl(connection) + d._screenshot = impl + d._touch = impl + d.commands = impl + case 'uiautomator2': + from kotonebot.client.implements.uiautomator2 import UiAutomator2Impl + from kotonebot.client.implements.adb import AdbImpl + impl = UiAutomator2Impl(d) + d._screenshot = impl + d._touch = impl + d.commands = AdbImpl(connection) + case _: + assert_never(f'Unsupported ADB recipe: {recipe}') + return d diff --git a/kotonebot/client/host/custom.py b/kotonebot/client/host/custom.py index aca4f18..05391dc 100644 --- a/kotonebot/client/host/custom.py +++ b/kotonebot/client/host/custom.py @@ -1,22 +1,21 @@ import os import subprocess from psutil import process_iter -from .protocol import Instance, AdbHostConfig -from typing import ParamSpec, TypeVar, cast +from .protocol import Instance, AdbHostConfig, HostProtocol +from typing import ParamSpec, TypeVar from typing_extensions import override from kotonebot import logging -from kotonebot.client import DeviceImpl -from kotonebot.client.device import Device -from kotonebot.client.registration import AdbBasedImpl, create_device -from kotonebot.client.implements.adb import AdbImplConfig +from kotonebot.client import Device +from .adb_common import AdbRecipes, CommonAdbCreateDeviceMixin logger = logging.getLogger(__name__) +CustomRecipes = AdbRecipes P = ParamSpec('P') T = TypeVar('T') -class CustomInstance(Instance[AdbHostConfig]): +class CustomInstance(CommonAdbCreateDeviceMixin, Instance[AdbHostConfig]): def __init__(self, exe_path: str | None, emulator_args: str = "", *args, **kwargs): super().__init__(*args, **kwargs) self.exe_path: str | None = exe_path @@ -69,24 +68,12 @@ class CustomInstance(Instance[AdbHostConfig]): pass @override - def create_device(self, impl: DeviceImpl, host_config: AdbHostConfig) -> Device: + def create_device(self, impl: CustomRecipes, host_config: AdbHostConfig) -> Device: """为自定义实例创建 Device。""" if self.adb_port is None: raise ValueError("ADB port is not set and is required.") - # 为 ADB 相关的实现创建配置 - if impl in ['adb', 'adb_raw', 'uiautomator2']: - config = AdbImplConfig( - addr=f'{self.adb_ip}:{self.adb_port}', - connect=True, - disconnect=True, - device_serial=self.adb_name, - timeout=host_config.timeout - ) - impl = cast(AdbBasedImpl, impl) # make pylance happy - return create_device(impl, config) - else: - raise ValueError(f'Unsupported device implementation for Custom: {impl}') + return super().create_device(impl, host_config) def __repr__(self) -> str: return f'CustomInstance(#{self.id}# at "{self.exe_path}" with {self.adb_ip}:{self.adb_port})' @@ -99,6 +86,25 @@ def _type_check(ins: Instance) -> CustomInstance: def create(exe_path: str | None, adb_ip: str, adb_port: int, adb_name: str | None, emulator_args: str = "") -> CustomInstance: return CustomInstance(exe_path, emulator_args=emulator_args, id='custom', name='Custom', adb_ip=adb_ip, adb_port=adb_port, adb_name=adb_name) +class CustomHost(HostProtocol[CustomRecipes]): + @staticmethod + def installed() -> bool: + # Custom instances don't have a specific installation requirement + return True + + @staticmethod + def list() -> list[Instance]: + # Custom instances are created manually, not discovered + return [] + + @staticmethod + def query(*, id: str) -> Instance | None: + # Custom instances are created manually, not discovered + return None + + @staticmethod + def recipes() -> 'list[CustomRecipes]': + return ['adb', 'adb_raw', 'uiautomator2'] if __name__ == '__main__': ins = create(r'C:\Program Files\BlueStacks_nxt\HD-Player.exe', '127.0.0.1', 5555, '**emulator-name**') diff --git a/kotonebot/client/host/leidian_host.py b/kotonebot/client/host/leidian_host.py index 99a186b..564fc12 100644 --- a/kotonebot/client/host/leidian_host.py +++ b/kotonebot/client/host/leidian_host.py @@ -1,18 +1,17 @@ import os import subprocess -from typing import cast +from typing import Literal from functools import lru_cache from typing_extensions import override from kotonebot import logging -from kotonebot.client import DeviceImpl -from kotonebot.client.device import Device -from kotonebot.client.registration import AdbBasedImpl, create_device -from kotonebot.client.implements.adb import AdbImplConfig +from kotonebot.client import Device from kotonebot.util import Countdown, Interval from .protocol import HostProtocol, Instance, copy_type, AdbHostConfig +from .adb_common import AdbRecipes, CommonAdbCreateDeviceMixin logger = logging.getLogger(__name__) +LeidianRecipes = AdbRecipes if os.name == 'nt': from ...interop.win.reg import read_reg @@ -21,7 +20,7 @@ else: """Stub for read_reg on non-Windows platforms.""" return default -class LeidianInstance(Instance[AdbHostConfig]): +class LeidianInstance(CommonAdbCreateDeviceMixin, Instance[AdbHostConfig]): @copy_type(Instance.__init__) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -64,35 +63,23 @@ class LeidianInstance(Instance[AdbHostConfig]): it = Interval(5) while not cd.expired() and not self.running(): it.wait() + self.refresh() if not self.running(): raise TimeoutError(f'Leidian instance "{self.name}" is not available.') @override def running(self) -> bool: - result = LeidianHost._invoke_manager(['isrunning', '--index', str(self.index)]) - return result.strip() == 'running' + return self.is_running @override - def create_device(self, impl: DeviceImpl, host_config: AdbHostConfig) -> Device: + def create_device(self, impl: LeidianRecipes, host_config: AdbHostConfig) -> Device: """为雷电模拟器实例创建 Device。""" if self.adb_port is None: raise ValueError("ADB port is not set and is required.") - # 为 ADB 相关的实现创建配置 - if impl in ['adb', 'adb_raw', 'uiautomator2']: - config = AdbImplConfig( - addr=f'{self.adb_ip}:{self.adb_port}', - connect=False, # 雷电模拟器不需要 adb connect - disconnect=False, - device_serial=self.adb_name, - timeout=host_config.timeout - ) - impl = cast(AdbBasedImpl, impl) # make pylance happy - return create_device(impl, config) - else: - raise ValueError(f'Unsupported device implementation for Leidian: {impl}') + return super().create_device(impl, host_config) -class LeidianHost(HostProtocol): +class LeidianHost(HostProtocol[LeidianRecipes]): @staticmethod @lru_cache(maxsize=1) def _read_install_path() -> str | None: @@ -197,6 +184,10 @@ class LeidianHost(HostProtocol): return instance return None + @staticmethod + def recipes() -> 'list[LeidianRecipes]': + return ['adb', 'adb_raw', 'uiautomator2'] + if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s') print(LeidianHost._read_install_path()) diff --git a/kotonebot/client/host/mumu12_host.py b/kotonebot/client/host/mumu12_host.py index c08955f..aa69056 100644 --- a/kotonebot/client/host/mumu12_host.py +++ b/kotonebot/client/host/mumu12_host.py @@ -2,19 +2,17 @@ import os import json import subprocess from functools import lru_cache -from typing import Any, cast +from typing import Any, Literal from typing_extensions import override from kotonebot import logging -from kotonebot.client import DeviceImpl, Device +from kotonebot.client import Device from kotonebot.client.device import AndroidDevice -from kotonebot.client.registration import AdbBasedImpl, create_device -from kotonebot.client.implements.adb import AdbImpl, AdbImplConfig, _create_adb_device_base +from kotonebot.client.implements.adb import AdbImpl from kotonebot.client.implements.nemu_ipc import NemuIpcImpl, NemuIpcImplConfig from kotonebot.util import Countdown, Interval from .protocol import HostProtocol, Instance, copy_type, AdbHostConfig - -logger = logging.getLogger(__name__) +from .adb_common import AdbRecipes, CommonAdbCreateDeviceMixin, connect_adb if os.name == 'nt': from ...interop.win.reg import read_reg @@ -23,7 +21,10 @@ else: """Stub for read_reg on non-Windows platforms.""" return default -class Mumu12Instance(Instance[AdbHostConfig]): +logger = logging.getLogger(__name__) +MuMu12Recipes = AdbRecipes | Literal['nemu_ipc'] + +class Mumu12Instance(CommonAdbCreateDeviceMixin, Instance[AdbHostConfig]): @copy_type(Instance.__init__) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @@ -75,45 +76,35 @@ class Mumu12Instance(Instance[AdbHostConfig]): return self.is_android_started @override - def create_device(self, impl: DeviceImpl, host_config: AdbHostConfig) -> Device: + def create_device(self, impl: MuMu12Recipes, host_config: AdbHostConfig) -> Device: """为MuMu12模拟器实例创建 Device。""" if self.adb_port is None: raise ValueError("ADB port is not set and is required.") - # 新增对 nemu_ipc 的支持 if impl == 'nemu_ipc': + # NemuImpl nemu_path = Mumu12Host._read_install_path() if not nemu_path: raise RuntimeError("无法找到 MuMu12 的安装路径。") - nemu_config = NemuIpcImplConfig(mumu_root_folder=nemu_path, instance_id=int(self.id)) + nemu_config = NemuIpcImplConfig(nemu_folder=nemu_path, instance_id=int(self.id)) + nemu_impl = NemuIpcImpl(nemu_config) + # AdbImpl + adb_impl = AdbImpl(connect_adb( + self.adb_ip, + self.adb_port, + timeout=host_config.timeout, + device_serial=self.adb_name + )) device = AndroidDevice() - nemu_impl = NemuIpcImpl(device, nemu_config) device._screenshot = nemu_impl device._touch = nemu_impl + device.commands = adb_impl - # 组装命令部分 (Adb) - adb_config = AdbImplConfig(addr=f'{self.adb_ip}:{self.adb_port}', timeout=host_config.timeout) - # adb_impl = AdbImpl(device, adb_config) - _d = _create_adb_device_base(adb_config, AdbImpl) - device.commands = _d.commands - return device - - # 为 ADB 相关的实现创建配置 - if impl in ['adb', 'adb_raw', 'uiautomator2']: - config = AdbImplConfig( - addr=f'{self.adb_ip}:{self.adb_port}', - connect=True, - disconnect=True, - device_serial=self.adb_name, - timeout=host_config.timeout - ) - impl = cast(AdbBasedImpl, impl) # make pylance happy - return create_device(impl, config) else: - raise ValueError(f'Unsupported device implementation for MuMu12: {impl}') + return super().create_device(impl, host_config) -class Mumu12Host(HostProtocol): +class Mumu12Host(HostProtocol[MuMu12Recipes]): @staticmethod @lru_cache(maxsize=1) def _read_install_path() -> str | None: @@ -210,6 +201,10 @@ class Mumu12Host(HostProtocol): if instance.id == id: return instance return None + + @staticmethod + def recipes() -> 'list[MuMu12Recipes]': + return ['adb', 'adb_raw', 'uiautomator2', 'nemu_ipc'] if __name__ == '__main__': logging.basicConfig(level=logging.DEBUG, format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(funcName)s] [%(lineno)d] %(message)s') diff --git a/kotonebot/client/host/protocol.py b/kotonebot/client/host/protocol.py index 2e88937..8d170a4 100644 --- a/kotonebot/client/host/protocol.py +++ b/kotonebot/client/host/protocol.py @@ -195,7 +195,8 @@ class Instance(Generic[T_HostConfig], ABC): def __repr__(self) -> str: return f'{self.__class__.__name__}(name="{self.name}", id="{self.id}", adb="{self.adb_ip}:{self.adb_port}"({self.adb_name}))' -class HostProtocol(Protocol): +Recipe = TypeVar('Recipe', bound=str) +class HostProtocol(Generic[Recipe], Protocol): @staticmethod def installed() -> bool: ... @@ -205,6 +206,8 @@ class HostProtocol(Protocol): @staticmethod def query(*, id: str) -> Instance | None: ... + @staticmethod + def recipes() -> 'list[Recipe]': ... if __name__ == '__main__': pass diff --git a/kotonebot/client/host/windows_common.py b/kotonebot/client/host/windows_common.py new file mode 100644 index 0000000..3c21063 --- /dev/null +++ b/kotonebot/client/host/windows_common.py @@ -0,0 +1,55 @@ +from abc import ABC +from typing import Literal +from typing_extensions import assert_never + +from kotonebot import logging +from kotonebot.client.device import WindowsDevice +from .protocol import Device, WindowsHostConfig, RemoteWindowsHostConfig + +logger = logging.getLogger(__name__) +WindowsRecipes = Literal['windows', 'remote_windows'] + +# Windows 相关的配置类型联合 +WindowsHostConfigs = WindowsHostConfig | RemoteWindowsHostConfig + +class CommonWindowsCreateDeviceMixin(ABC): + """ + 通用 Windows 创建设备的 Mixin。 + 该 Mixin 定义了创建 Windows 设备的通用接口。 + """ + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + + def create_device(self, recipe: WindowsRecipes, config: WindowsHostConfigs) -> Device: + """ + 创建 Windows 设备。 + """ + match recipe: + case 'windows': + if not isinstance(config, WindowsHostConfig): + raise ValueError(f"Expected WindowsHostConfig for 'windows' recipe, got {type(config)}") + from kotonebot.client.implements.windows import WindowsImpl + d = WindowsDevice() + impl = WindowsImpl( + device=d, + window_title=config.window_title, + ahk_exe_path=config.ahk_exe_path + ) + d._screenshot = impl + d._touch = impl + return d + case 'remote_windows': + if not isinstance(config, RemoteWindowsHostConfig): + raise ValueError(f"Expected RemoteWindowsHostConfig for 'remote_windows' recipe, got {type(config)}") + from kotonebot.client.implements.remote_windows import RemoteWindowsImpl + d = WindowsDevice() + impl = RemoteWindowsImpl( + device=d, + host=config.host, + port=config.port + ) + d._screenshot = impl + d._touch = impl + return d + case _: + assert_never(f'Unsupported Windows recipe: {recipe}') diff --git a/kotonebot/client/implements/adb.py b/kotonebot/client/implements/adb.py index e87fce0..e3dc7c0 100644 --- a/kotonebot/client/implements/adb.py +++ b/kotonebot/client/implements/adb.py @@ -9,7 +9,7 @@ from adbutils._device import AdbDevice as AdbUtilsDevice from ..device import AndroidDevice from ..protocol import AndroidCommandable, Touchable, Screenshotable -from ..registration import register_impl, ImplConfig +from ..registration import ImplConfig from dataclasses import dataclass logger = logging.getLogger(__name__) @@ -83,46 +83,3 @@ class AdbImpl(AndroidCommandable, Touchable, Screenshotable): if duration is not None: logger.warning("Swipe duration is not supported with AdbDevice. Ignoring duration.") self.adb.shell(f"input touchscreen swipe {x1} {y1} {x2} {y2}") - - -def _create_adb_device_base(config: AdbImplConfig, impl_class: type) -> AndroidDevice: - """ - 通用的 ADB 设备创建工厂函数。 - 其他任意基于 ADB 的 Impl 可以直接复用这个函数。 - - :param config: ADB 实现配置 - :param impl_class: 实现类或工厂函数。构造函数接收 adb_connection 参数。 - """ - from adbutils import adb - - if config.disconnect: - logger.debug('adb disconnect %s', config.addr) - adb.disconnect(config.addr) - if config.connect: - logger.debug('adb connect %s', config.addr) - result = adb.connect(config.addr) - if 'cannot connect to' in result: - raise ValueError(result) - serial = config.device_serial or config.addr - logger.debug('adb wait for %s', serial) - adb.wait_for(serial, timeout=config.timeout) - devices = adb.device_list() - logger.debug('adb device_list: %s', devices) - d = [d for d in devices if d.serial == serial] - if len(d) == 0: - raise ValueError(f"Device {config.addr} not found") - d = d[0] - - device = AndroidDevice(d) - impl = impl_class(d) - device._touch = impl - device._screenshot = impl - device.commands = impl - - return device - - -@register_impl('adb', config_model=AdbImplConfig) -def create_adb_device(config: AdbImplConfig) -> AndroidDevice: - """AdbImpl 工厂函数""" - return _create_adb_device_base(config, AdbImpl) diff --git a/kotonebot/client/implements/adb_raw.py b/kotonebot/client/implements/adb_raw.py index 985d6b6..cdaf9e4 100644 --- a/kotonebot/client/implements/adb_raw.py +++ b/kotonebot/client/implements/adb_raw.py @@ -11,9 +11,8 @@ import numpy as np from cv2.typing import MatLike from adbutils._utils import adb_path -from .adb import AdbImpl, AdbImplConfig, _create_adb_device_base +from .adb import AdbImpl from adbutils._device import AdbDevice as AdbUtilsDevice -from ..registration import register_impl from kotonebot import logging logger = logging.getLogger(__name__) @@ -157,10 +156,4 @@ class AdbRawImpl(AdbImpl): logger.verbose(f"adb raw screenshot wait time: {time.time() - start_time:.4f}s") data = self.__data self.__data = None - return data - - -@register_impl('adb_raw', config_model=AdbImplConfig) -def create_adb_raw_device(config: AdbImplConfig): - """AdbRawImpl 工厂函数""" - return _create_adb_device_base(config, AdbRawImpl) \ No newline at end of file + return data \ No newline at end of file diff --git a/kotonebot/client/implements/nemu_ipc/nemu_ipc.py b/kotonebot/client/implements/nemu_ipc/nemu_ipc.py index 264fb17..710b991 100644 --- a/kotonebot/client/implements/nemu_ipc/nemu_ipc.py +++ b/kotonebot/client/implements/nemu_ipc/nemu_ipc.py @@ -11,8 +11,9 @@ from cv2.typing import MatLike from ...device import AndroidDevice, Device from ...protocol import Touchable, Screenshotable -from ...registration import register_impl, ImplConfig +from ...registration import ImplConfig from .external_renderer_ipc import ExternalRendererIpc +from kotonebot.errors import KotonebotError logger = logging.getLogger(__name__) @@ -22,16 +23,23 @@ class NemuIpcIncompatible(Exception): pass -class NemuIpcError(Exception): +class NemuIpcError(KotonebotError): """调用 IPC 过程中发生错误""" pass @dataclass class NemuIpcImplConfig(ImplConfig): - """nemu_ipc 设备实现的配置模型，mumu_root_folder 为 MuMu 根目录""" - mumu_root_folder: str + r"""nemu_ipc 能力的配置模型。 + + 参数说明： + nemu_folder: MuMu12 根目录（如 F:\Apps\Netease\MuMuPlayer-12.0）。 + instance_id: 模拟器实例 ID。 + display_id: 目标显示器 ID，默认为 0（主显示器）。 + """ + nemu_folder: str instance_id: int + display_id: int = 0 class NemuIpcImpl(Touchable, Screenshotable): @@ -39,23 +47,22 @@ class NemuIpcImpl(Touchable, Screenshotable): 利用 MuMu12 提供的 external_renderer_ipc.dll 进行截图与触摸控制。 """ - def __init__(self, device: Device, config: NemuIpcImplConfig): - self.device = device + def __init__(self, config: NemuIpcImplConfig): self.config = config self.__width: int = 0 self.__height: int = 0 self.__connected: bool = False self._connect_id: int = 0 - self.display_id: int = 0 + self.display_id: int = config.display_id """ 显示器 ID。`0` 表示主显示器。 如果没有启用「后台保活」功能，一般为主显示器。 """ - self.nemu_folder = config.mumu_root_folder + self.nemu_folder = config.nemu_folder # --------------------------- DLL 封装 --------------------------- - self._ipc = ExternalRendererIpc(config.mumu_root_folder) + self._ipc = ExternalRendererIpc(config.nemu_folder) logger.info("ExternalRendererIpc initialized and DLL loaded") @property @@ -222,16 +229,4 @@ class NemuIpcImpl(Touchable, Screenshotable): # 最终抬起 self._ipc.input_touch_up(self._connect_id, self.display_id) - sleep(0.01) - -# ------------------------------------------------------------------ -# 工厂方法 -# ------------------------------------------------------------------ -@register_impl("nemu_ipc", config_model=NemuIpcImplConfig) -def create_nemu_ipc_device(config: NemuIpcImplConfig): - """创建一个 AndroidDevice，并挂载 NemuIpcImpl。""" - device = AndroidDevice() - impl = NemuIpcImpl(device, config) - device._touch = impl - device._screenshot = impl - return device \ No newline at end of file + sleep(0.01) \ No newline at end of file diff --git a/kotonebot/client/implements/remote_windows.py b/kotonebot/client/implements/remote_windows.py index 72ef042..ef75e15 100644 --- a/kotonebot/client/implements/remote_windows.py +++ b/kotonebot/client/implements/remote_windows.py @@ -23,8 +23,8 @@ from cv2.typing import MatLike from kotonebot import logging from ..device import Device, WindowsDevice from ..protocol import Touchable, Screenshotable -from ..registration import register_impl, ImplConfig -from .windows import WindowsImpl, WindowsImplConfig, create_windows_device +from ..registration import ImplConfig +from .windows import WindowsImpl, WindowsImplConfig logger = logging.getLogger(__name__) @@ -63,7 +63,11 @@ class RemoteWindowsServer: self.port = port self.server = None self.device = WindowsDevice() - self.impl = create_windows_device(windows_impl_config) + self.impl = WindowsImpl( + WindowsDevice(), + ahk_exe_path=windows_impl_config.ahk_exe_path, + window_title=windows_impl_config.window_title + ) self.device._screenshot = self.impl self.device._touch = self.impl @@ -186,14 +190,4 @@ class RemoteWindowsImpl(Touchable, Screenshotable): def swipe(self, x1: int, y1: int, x2: int, y2: int, duration: float | None = None) -> None: """Swipe from (x1, y1) to (x2, y2) on the remote server.""" if not self.proxy.swipe(x1, y1, x2, y2, duration): - raise RuntimeError(f"Failed to swipe from ({x1}, {y1}) to ({x2}, {y2})") - - -# 编写并注册创建函数 -@register_impl('remote_windows', config_model=RemoteWindowsImplConfig) -def create_remote_windows_device(config: RemoteWindowsImplConfig) -> Device: - device = WindowsDevice() - remote_impl = RemoteWindowsImpl(device, config.host, config.port) - device._touch = remote_impl - device._screenshot = remote_impl - return device \ No newline at end of file + raise RuntimeError(f"Failed to swipe from ({x1}, {y1}) to ({x2}, {y2})") \ No newline at end of file diff --git a/kotonebot/client/implements/uiautomator2.py b/kotonebot/client/implements/uiautomator2.py index 25e840c..cf8a348 100644 --- a/kotonebot/client/implements/uiautomator2.py +++ b/kotonebot/client/implements/uiautomator2.py @@ -8,8 +8,6 @@ from cv2.typing import MatLike from kotonebot import logging from ..device import Device from ..protocol import Screenshotable, Commandable, Touchable -from ..registration import register_impl -from .adb import AdbImplConfig, _create_adb_device_base logger = logging.getLogger(__name__) @@ -84,10 +82,4 @@ class UiAutomator2Impl(Screenshotable, Commandable, Touchable): """ 滑动屏幕 """ - self.u2_client.swipe(x1, y1, x2, y2, duration=duration or 0.1) - - -@register_impl('uiautomator2', config_model=AdbImplConfig) -def create_uiautomator2_device(config: AdbImplConfig) -> Device: - """UiAutomator2Impl 工厂函数""" - return _create_adb_device_base(config, UiAutomator2Impl) + self.u2_client.swipe(x1, y1, x2, y2, duration=duration or 0.1) \ No newline at end of file diff --git a/kotonebot/client/implements/windows.py b/kotonebot/client/implements/windows.py index 3cb0ee1..1ebd15b 100644 --- a/kotonebot/client/implements/windows.py +++ b/kotonebot/client/implements/windows.py @@ -13,7 +13,7 @@ from cv2.typing import MatLike from ..device import Device, WindowsDevice from ..protocol import Commandable, Touchable, Screenshotable -from ..registration import register_impl, ImplConfig +from ..registration import ImplConfig # 1. 定义配置模型 @dataclass @@ -167,21 +167,6 @@ class WindowsImpl(Touchable, Screenshotable): # TODO: 这个 speed 的单位是什么？ self.ahk.mouse_drag(x2, y2, from_position=(x1, y1), coord_mode='Client', speed=10) - -# 3. 编写并注册创建函数 -@register_impl('windows', config_model=WindowsImplConfig) -def create_windows_device(config: WindowsImplConfig) -> Device: - device = WindowsDevice() - impl = WindowsImpl( - device, - window_title=config.window_title, - ahk_exe_path=config.ahk_exe_path - ) - device._touch = impl - device._screenshot = impl - return device - - if __name__ == '__main__': from ..device import Device device = Device() diff --git a/kotonebot/client/registration.py b/kotonebot/client/registration.py index 09a566f..a34131a 100644 --- a/kotonebot/client/registration.py +++ b/kotonebot/client/registration.py @@ -22,73 +22,3 @@ class ImplRegistrationError(KotonebotError): class ImplConfig: """所有设备实现配置模型的名义上的基类，便于类型约束。""" pass - -T_Config = TypeVar("T_Config", bound=ImplConfig) - -# 定义两种创建者函数类型 -CreatorWithConfig = Callable[[Any], Device] -CreatorWithoutConfig = Callable[[], Device] - -# --- 底层 API: 公开的注册表 --- - -# 注册表结构: {'impl_name': (创建函数, 配置模型类 或 None)} -DEVICE_CREATORS: Dict[str, tuple[Callable[..., Device], Type[ImplConfig] | None]] = {} - - -def register_impl(name: str, config_model: Type[ImplConfig] | None = None) -> Callable[..., Any]: - """ - 一个统一的装饰器，用于向 DEVICE_CREATORS 注册表中注册一个设备实现。 - - :param name: 实现的名称 (e.g., 'windows', 'adb') - :param config_model: (可选) 与该实现关联的 dataclass 配置模型 - """ - def decorator(creator_func: Callable[..., Device]) -> Callable[..., Device]: - # if name in DEVICE_CREATORS: - # raise ImplRegistrationError(f"实现 '{name}' 已被注册。") - DEVICE_CREATORS[name] = (creator_func, config_model) - return creator_func - return decorator - - -# --- 高层 API: 带 overload 的便利函数 --- - -# 为需要配置的已知 impl 提供 overload -@overload -def create_device(impl_name: Literal['windows'], config: 'WindowsImplConfig') -> Device: ... - -@overload -def create_device(impl_name: Literal['remote_windows'], config: 'RemoteWindowsImplConfig') -> Device: ... - -@overload -def create_device(impl_name: AdbBasedImpl, config: 'AdbImplConfig') -> Device: ... - -# 新增 nemu_ipc overload -@overload -def create_device(impl_name: Literal['nemu_ipc'], config: 'NemuIpcImplConfig') -> Device: ... - -# 函数的实际实现 -def create_device(impl_name: DeviceImpl, config: ImplConfig | None = None) -> Device: - """ - 根据名称和可选的配置对象，统一创建设备。 - """ - creator_tuple = DEVICE_CREATORS.get(impl_name) - if not creator_tuple: - raise ImplRegistrationError(f"未找到名为 '{impl_name}' 的实现。") - - creator_func, registered_config_model = creator_tuple - - # 情况 A: 实现需要配置 - if registered_config_model is not None: - creator_with_config = cast(CreatorWithConfig, creator_func) - if config is None: - raise ValueError(f"实现 '{impl_name}' 需要一个配置对象，但传入的是 None。") - if not isinstance(config, registered_config_model): - raise TypeError(f"为 '{impl_name}' 传入的配置类型错误，应为 '{registered_config_model.__name__}'，实际为 '{type(config).__name__}'。") - return creator_with_config(config) - - # 情况 B: 实现无需配置 - else: - creator_without_config = cast(CreatorWithoutConfig, creator_func) - if config is not None: - print(f"提示：实现 '{impl_name}' 无需配置，但你提供了一个配置对象，它将被忽略。") - return creator_without_config() diff --git a/kotonebot/config/base_config.py b/kotonebot/config/base_config.py index 62b2c49..bd3f64d 100644 --- a/kotonebot/config/base_config.py +++ b/kotonebot/config/base_config.py @@ -3,10 +3,10 @@ from typing import Generic, TypeVar, Literal from pydantic import BaseModel, ConfigDict -from kotonebot.client import DeviceImpl T = TypeVar('T') BackendType = Literal['custom', 'mumu12', 'leidian', 'dmm'] +DeviceRecipes = Literal['adb', 'adb_raw', 'uiautomator2', 'windows', 'remote_windows', 'nemu_ipc'] class ConfigBaseModel(BaseModel): model_config = ConfigDict(use_attribute_docstrings=True) @@ -27,7 +27,7 @@ class BackendConfig(ConfigBaseModel): 雷电模拟器需要设置正确的模拟器名，否则 自动启动模拟器 功能将无法正常工作。 其他功能不受影响。 """ - screenshot_impl: DeviceImpl = 'adb' + screenshot_impl: DeviceRecipes = 'adb' """ 截图方法。暂时推荐使用【adb】截图方式。 diff --git a/kotonebot/kaa/main/dmm_host.py b/kotonebot/kaa/main/dmm_host.py index 8282d1b..4d66c27 100644 --- a/kotonebot/kaa/main/dmm_host.py +++ b/kotonebot/kaa/main/dmm_host.py @@ -1,18 +1,15 @@ -from importlib import resources from typing_extensions import override -from kotonebot.client import Device, DeviceImpl -from kotonebot.client.registration import create_device -from kotonebot.client.implements.windows import WindowsImplConfig -from kotonebot.client.implements.remote_windows import RemoteWindowsImplConfig +from kotonebot.client import Device from kotonebot.client.host import HostProtocol, Instance -from kotonebot.client.host.protocol import WindowsHostConfig, RemoteWindowsHostConfig +from kotonebot.client.host.windows_common import WindowsRecipes, WindowsHostConfigs, CommonWindowsCreateDeviceMixin -DmmHostConfigs = WindowsHostConfig | RemoteWindowsHostConfig +DmmHostConfigs = WindowsHostConfigs +DmmRecipes = WindowsRecipes # TODO: 可能应该把 start_game 和 end_game 里对启停的操作移动到这里来 -class DmmInstance(Instance[DmmHostConfigs]): +class DmmInstance(CommonWindowsCreateDeviceMixin, Instance[DmmHostConfigs]): def __init__(self): super().__init__('dmm', 'gakumas') @@ -37,31 +34,11 @@ class DmmInstance(Instance[DmmHostConfigs]): raise NotImplementedError() @override - def create_device(self, impl: DeviceImpl, host_config: DmmHostConfigs) -> Device: - if impl == 'windows': - assert isinstance(host_config, WindowsHostConfig) - win_config = WindowsImplConfig( - window_title=host_config.window_title, - ahk_exe_path=host_config.ahk_exe_path - ) - return create_device(impl, win_config) + def create_device(self, impl: DmmRecipes, host_config: DmmHostConfigs) -> Device: + """为 DMM 实例创建 Device。""" + return super().create_device(impl, host_config) - elif impl == 'remote_windows': - assert isinstance(host_config, RemoteWindowsHostConfig) - config = RemoteWindowsImplConfig( - windows_impl_config=WindowsImplConfig( - window_title=host_config.windows_host_config.window_title, - ahk_exe_path=host_config.windows_host_config.ahk_exe_path - ), - host=host_config.host, - port=host_config.port - ) - return create_device(impl, config) - - else: - raise ValueError(f'Unsupported impl for DMM: {impl}') - -class DmmHost(HostProtocol): +class DmmHost(HostProtocol[DmmRecipes]): instance = DmmInstance() """DmmInstance 单例。""" @@ -77,3 +54,7 @@ class DmmHost(HostProtocol): @staticmethod def query(*, id: str) -> Instance | None: raise NotImplementedError() + + @staticmethod + def recipes() -> 'list[DmmRecipes]': + return ['windows', 'remote_windows'] diff --git a/kotonebot/kaa/main/kaa.py b/kotonebot/kaa/main/kaa.py index a9085f1..ccb51e6 100644 --- a/kotonebot/kaa/main/kaa.py +++ b/kotonebot/kaa/main/kaa.py @@ -1,5 +1,6 @@ import io import os +from typing import Any, cast import zipfile import logging import traceback @@ -9,8 +10,6 @@ from typing_extensions import override import cv2 -from kotonebot.client.implements.nemu_ipc import NemuIpcImplConfig - from ...client import Device from kotonebot.ui import user from kotonebot import KotoneBot @@ -234,7 +233,10 @@ class Kaa(KotoneBot): elif isinstance(self.backend_instance, (CustomInstance, Mumu12Instance, LeidianInstance)): if impl_name in ['adb', 'adb_raw', 'uiautomator2'] or (impl_name == 'nemu_ipc' and isinstance(self.backend_instance, Mumu12Instance)): host_conf = AdbHostConfig(timeout=180) - return self.backend_instance.create_device(impl_name, host_conf) + return self.backend_instance.create_device( + cast(Any, impl_name), # :( + host_conf + ) else: raise ValueError(f"{user_config.backend.type} backend does not support implementation '{impl_name}'")