xcantloadx
/
kotones-auto-assistant
mirror of https://github.com/XcantloadX/kotones-auto-assistant
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor(core): 重构 Device 与 Impl 的创建方式 

现在允许 Impl 存在构造参数，并允许下游脚本传递参数给 Impl。
This commit is contained in:
XcantloadX 2025-06-10 12:46:51 +08:00
parent b46d69a22b
commit 2fc9ad5200
18 changed files with 559 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
kotonebot/backend/context/context.py
View File

@ -46,12 +46,12 @@ from kotonebot.backend.color import (
from kotonebot.backend.ocr import (
Ocr, OcrResult, OcrResultList, jp, en, StringMatchFunction
)
from kotonebot.client.factory import create_device
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.factory import DeviceImpl
from kotonebot.client import DeviceImpl
from kotonebot.backend.preprocessor import PreprocessorProtocol
from kotonebot.primitives import Rect
@ -774,13 +774,46 @@ class Context(Generic[T]):
self.__debug = ContextDebug(self)
self.__config = ContextConfig[T](self, config_path, config_type)
ip = self.config.current.backend.adb_ip
port = self.config.current.backend.adb_port
# TODO: 处理链接失败情况
if screenshot_impl is None:
screenshot_impl = self.config.current.backend.screenshot_impl
logger.info(f'Using "{screenshot_impl}" as screenshot implementation')
self.__device = ContextDevice(device or create_device(f'{ip}:{port}', screenshot_impl))
if device is None:
# 根据不同的 impl 类型创建设备
if screenshot_impl in ['adb', 'adb_raw', 'uiautomator2']:
from kotonebot.client.implements.adb import AdbImplConfig
ip = self.config.current.backend.adb_ip
port = self.config.current.backend.adb_port
config = AdbImplConfig(
addr=f'{ip}:{port}',
connect=True,
disconnect=True,
device_serial=None,
timeout=180
)
impl = cast(AdbBasedImpl, screenshot_impl) # make pylance happy
device = create_device(impl, config)
elif screenshot_impl == 'windows':
from kotonebot.client.implements.windows import WindowsImplConfig
config = WindowsImplConfig(
window_title=getattr(self.config.current.backend, 'windows_window_title', 'gakumas'),
ahk_exe_path=getattr(self.config.current.backend, 'windows_ahk_path', None)
)
device = create_device(screenshot_impl, config)
elif screenshot_impl == 'remote_windows':
from kotonebot.client.implements.remote_windows import RemoteWindowsImplConfig
ip = self.config.current.backend.adb_ip
port = self.config.current.backend.adb_port
config = RemoteWindowsImplConfig(
host=ip,
port=port
)
device = create_device(screenshot_impl, config)
else:
raise ValueError(f'Unsupported screenshot implementation: {screenshot_impl}')
self.__device = ContextDevice(device)
def inject(
self,

5
kotonebot/client/__init__.py
View File

@ -1,5 +1,8 @@
from .device import Device
from .factory import create_device, DeviceImpl
from .registration import create_device, DeviceImpl
# 确保所有实现都被注册
from . import implements # noqa: F401
__all__ = [
'Device',

92
kotonebot/client/factory.py
View File

@ -1,92 +0,0 @@
import logging
from typing import Literal
from .implements.adb import AdbImpl
from .implements.adb_raw import AdbRawImpl
from .implements.windows import WindowsImpl
from .implements.remote_windows import RemoteWindowsImpl
from .implements.uiautomator2 import UiAutomator2Impl
from .device import Device, AndroidDevice, WindowsDevice
from adbutils import adb
DeviceImpl = Literal['adb', 'adb_raw', 'uiautomator2', 'windows', 'remote_windows']
logger = logging.getLogger(__name__)
def create_device(
addr: str,
impl: DeviceImpl,
*,
connect: bool = True,
disconnect: bool = True,
device_serial: str | None = None,
timeout: float = 180,
) -> Device:
"""
根据指定的实现方式创建 Device 实例。
:param addr: 设备地址，如 `127.0.0.1:5555`。
仅当通过无线方式连接 Android 设备，或者使用 `remote_windows` 时有效。
:param impl: 实现方式。
:param connect: 是否在创建时连接设备，默认为 True。
仅对 ADB-based 的实现方式有效。
:param disconnect: 是否在连接前先断开设备，默认为 True。
仅对 ADB-based 的实现方式有效。
:param device_serial: 设备序列号，默认为 None。
若为非 None，则当存在多个设备时通过该值判断是否为目标设备。
仅对 ADB-based 的实现方式有效。
:param timeout: 连接超时时间，默认为 180 秒。
仅对 ADB-based 的实现方式有效。
"""
if impl in ['adb', 'adb_raw', 'uiautomator2']:
if disconnect:
logger.debug('adb disconnect %s', addr)
adb.disconnect(addr)
if connect:
logger.debug('adb connect %s', addr)
result = adb.connect(addr)
if 'cannot connect to' in result:
raise ValueError(result)
serial = device_serial or addr
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 {addr} not found")
d = d[0]
device = AndroidDevice(d)
if impl == 'adb':
device._command = AdbImpl(device)
device._touch = AdbImpl(device)
device._screenshot = AdbImpl(device)
elif impl == 'adb_raw':
device._command = AdbRawImpl(device)
device._touch = AdbRawImpl(device)
device._screenshot = AdbRawImpl(device)
elif impl == 'uiautomator2':
device._command = UiAutomator2Impl(device)
device._touch = UiAutomator2Impl(device)
device._screenshot = UiAutomator2Impl(device)
elif impl == 'windows':
device = WindowsDevice()
device._touch = WindowsImpl(device)
device._screenshot = WindowsImpl(device)
elif impl == 'remote_windows':
# For remote_windows, addr should be in the format 'host:port'
if ':' not in addr:
raise ValueError(f"Invalid address format for remote_windows: {addr}. Expected format: 'host:port'")
host, port_str = addr.split(':', 1)
try:
port = int(port_str)
except ValueError:
raise ValueError(f"Invalid port in address: {port_str}")
device = WindowsDevice()
remote_impl = RemoteWindowsImpl(device, host, port)
device._touch = remote_impl
device._screenshot = remote_impl
else:
raise ValueError(f"Unsupported device implementation: {impl}")
return device

3
kotonebot/client/host/__init__.py
View File

@ -1,10 +1,11 @@
from .protocol import HostProtocol, Instance
from .protocol import HostProtocol, Instance, AdbHostConfig, WindowsHostConfig, RemoteWindowsHostConfig
from .custom import CustomInstance, create as create_custom
from .mumu12_host import Mumu12Host, Mumu12Instance
from .leidian_host import LeidianHost, LeidianInstance
__all__ = [
'HostProtocol', 'Instance',
'AdbHostConfig', 'WindowsHostConfig', 'RemoteWindowsHostConfig',
'CustomInstance', 'create_custom',
'Mumu12Host', 'Mumu12Instance',
'LeidianHost', 'LeidianInstance'

29
kotonebot/client/host/custom.py
View File

@ -1,19 +1,22 @@
import os
import subprocess
from psutil import process_iter
from .protocol import HostProtocol, Instance
from typing import Optional, ParamSpec, TypeVar, TypeGuard
from .protocol import Instance, AdbHostConfig
from typing import ParamSpec, TypeVar, cast
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
logger = logging.getLogger(__name__)
P = ParamSpec('P')
T = TypeVar('T')
class CustomInstance(Instance):
class CustomInstance(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
@ -65,6 +68,26 @@ class CustomInstance(Instance):
def refresh(self):
pass
@override
def create_device(self, impl: DeviceImpl, 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}')
def __repr__(self) -> str:
return f'CustomInstance(#{self.id}# at "{self.exe_path}" with {self.adb_ip}:{self.adb_port})'

33
kotonebot/client/host/leidian_host.py
View File

@ -1,13 +1,16 @@
import os
import subprocess
from typing import cast
from functools import lru_cache
from typing_extensions import override
from kotonebot import logging
from kotonebot.client import DeviceImpl, create_device
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.util import Countdown, Interval
from .protocol import HostProtocol, Instance, copy_type
from .protocol import HostProtocol, Instance, copy_type, AdbHostConfig
logger = logging.getLogger(__name__)
@ -18,7 +21,7 @@ else:
"""Stub for read_reg on non-Windows platforms."""
return default
class LeidianInstance(Instance):
class LeidianInstance(Instance[AdbHostConfig]):
@copy_type(Instance.__init__)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@ -70,16 +73,24 @@ class LeidianInstance(Instance):
return result.strip() == 'running'
@override
def create_device(self, impl: DeviceImpl, *, timeout: float = 180) -> Device:
def create_device(self, impl: DeviceImpl, host_config: AdbHostConfig) -> Device:
"""为雷电模拟器实例创建 Device。"""
if self.adb_port is None:
raise ValueError("ADB port is not set and is required.")
return create_device(
addr=f'{self.adb_ip}:{self.adb_port}',
impl=impl,
device_serial=self.adb_name,
connect=False,
timeout=timeout
)
# 为 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}')
class LeidianHost(HostProtocol):
@staticmethod

28
kotonebot/client/host/mumu12_host.py
View File

@ -2,13 +2,15 @@ import os
import json
import subprocess
from functools import lru_cache
from typing import Any
from typing import Any, cast
from typing_extensions import override
from kotonebot import logging
from kotonebot.client import DeviceImpl, Device
from kotonebot.client.registration import AdbBasedImpl, create_device
from kotonebot.client.implements.adb import AdbImplConfig
from kotonebot.util import Countdown, Interval
from .protocol import HostProtocol, Instance, copy_type
from .protocol import HostProtocol, Instance, copy_type, AdbHostConfig
logger = logging.getLogger(__name__)
@ -19,7 +21,7 @@ else:
"""Stub for read_reg on non-Windows platforms."""
return default
class Mumu12Instance(Instance):
class Mumu12Instance(Instance[AdbHostConfig]):
@copy_type(Instance.__init__)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@ -70,6 +72,26 @@ class Mumu12Instance(Instance):
def running(self) -> bool:
return self.is_android_started
@override
def create_device(self, impl: DeviceImpl, host_config: AdbHostConfig) -> Device:
"""为MuMu12模拟器实例创建 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 MuMu12: {impl}')
class Mumu12Host(HostProtocol):
@staticmethod
@lru_cache(maxsize=1)

57
kotonebot/client/host/protocol.py
View File

@ -1,14 +1,15 @@
import time
import socket
from abc import ABC, abstractmethod
from typing_extensions import ParamSpec, Concatenate
from typing import Callable, TypeVar, Generic, Protocol, runtime_checkable, Type, Any
from typing import Callable, TypeVar, Protocol, Any, Generic
from dataclasses import dataclass
from adbutils import adb, AdbTimeout, AdbError
from adbutils._device import AdbDevice
from kotonebot import logging
from kotonebot.client import Device, create_device, DeviceImpl
from kotonebot.client import Device, DeviceImpl
from kotonebot.util import Countdown, Interval
logger = logging.getLogger(__name__)
@ -17,6 +18,27 @@ _T = TypeVar("_T")
def copy_type(_: _T) -> Callable[[Any], _T]:
return lambda x: x
# --- 定义专用的 HostConfig 数据类 ---
@dataclass
class AdbHostConfig:
"""由外部为基于 ADB 的主机提供的配置。"""
timeout: float = 180
@dataclass
class WindowsHostConfig:
"""由外部为 Windows 实现提供配置。"""
window_title: str
ahk_exe_path: str
@dataclass
class RemoteWindowsHostConfig:
"""由外部为远程 Windows 实现提供配置。"""
host: str
port: int
# --- 使用泛型改造 Instance 协议 ---
T_HostConfig = TypeVar("T_HostConfig")
def tcp_ping(host: str, port: int, timeout: float = 1.0) -> bool:
"""
通过 TCP ping 检查主机和端口是否可达。
@ -36,7 +58,11 @@ def tcp_ping(host: str, port: int, timeout: float = 1.0) -> bool:
return False
class Instance(ABC):
class Instance(Generic[T_HostConfig], ABC):
"""
代表一个可运行环境的实例（如一个模拟器）。
使用泛型来约束 create_device 方法的配置参数类型。
"""
def __init__(self,
id: str,
name: str,
@ -68,7 +94,7 @@ class Instance(ABC):
启动模拟器实例。
"""
raise NotImplementedError()
@abstractmethod
def stop(self):
"""
@ -80,21 +106,16 @@ class Instance(ABC):
def running(self) -> bool:
raise NotImplementedError()
def create_device(self, impl: DeviceImpl, *, timeout: float = 180) -> Device:
@abstractmethod
def create_device(self, impl: DeviceImpl, host_config: T_HostConfig) -> Device:
"""
创建 Device 实例，可用于控制模拟器系统。
:return: Device 实例
根据实现名称和类型化的主机配置创建设备。
:param impl: 设备实现的名称。
:param host_config: 一个类型化的数据对象，包含创建所需的所有外部配置。
:return: 配置好的 Device 实例。
"""
if self.adb_port is None:
raise ValueError("ADB port is not set and is required.")
return create_device(
addr=f'{self.adb_ip}:{self.adb_port}',
impl=impl,
device_serial=self.adb_name,
connect=True,
timeout=timeout
)
raise NotImplementedError()
def wait_available(self, timeout: float = 180):
logger.info('Starting to wait for emulator %s(127.0.0.1:%d) to be available...', self.name, self.adb_port)

6
kotonebot/client/implements/__init__.py
View File

@ -0,0 +1,6 @@
# 导入所有内置实现，以触发它们的 @register_impl 装饰器
from . import adb # noqa: F401
from . import adb_raw # noqa: F401
from . import remote_windows # noqa: F401
from . import uiautomator2 # noqa: F401
from . import windows # noqa: F401

43
kotonebot/client/implements/adb.py
View File

@ -6,11 +6,22 @@ import cv2
import numpy as np
from cv2.typing import MatLike
from ..device import Device
from ..device import Device, AndroidDevice
from ..protocol import Commandable, Touchable, Screenshotable
from ..registration import register_impl, ImplConfig
from dataclasses import dataclass
logger = logging.getLogger(__name__)
# 定义配置模型
@dataclass
class AdbImplConfig(ImplConfig):
addr: str
connect: bool = True
disconnect: bool = True
device_serial: str | None = None
timeout: float = 180
class AdbImpl(Commandable, Touchable, Screenshotable):
def __init__(self, device: Device):
self.device = device
@ -66,3 +77,33 @@ class AdbImpl(Commandable, 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}")
# 编写并注册创建函数
@register_impl('adb', config_model=AdbImplConfig)
def create_adb_device(config: AdbImplConfig) -> Device:
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 = AdbImpl(device)
device._command = impl
device._touch = impl
device._screenshot = impl
return device

37
kotonebot/client/implements/adb_raw.py
View File

@ -11,8 +11,9 @@ import numpy as np
from cv2.typing import MatLike
from adbutils._utils import adb_path
from .adb import AdbImpl
from ..device import Device
from .adb import AdbImpl, AdbImplConfig
from ..device import Device, AndroidDevice
from ..registration import register_impl
from kotonebot import logging
logger = logging.getLogger(__name__)
@ -156,4 +157,34 @@ 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
return data
# 编写并注册创建函数
@register_impl('adb_raw', config_model=AdbImplConfig)
def create_adb_raw_device(config: AdbImplConfig) -> Device:
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 = AdbRawImpl(device)
device._command = impl
device._touch = impl
device._screenshot = impl
return device

20
kotonebot/client/implements/remote_windows.py
View File

@ -14,6 +14,7 @@ import xmlrpc.server
from typing import Literal, cast, Any, Tuple
from functools import cached_property
from threading import Thread
from dataclasses import dataclass
import cv2
import numpy as np
@ -22,10 +23,17 @@ 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
logger = logging.getLogger(__name__)
# 定义配置模型
@dataclass
class RemoteWindowsImplConfig(ImplConfig):
host: str = "localhost"
port: int = 8000
def _encode_image(image: MatLike) -> str:
"""Encode an image as a base64 string."""
success, buffer = cv2.imencode('.png', image)
@ -54,7 +62,7 @@ class RemoteWindowsServer:
self.port = port
self.server = None
self.device = WindowsDevice()
self.impl = WindowsImpl(self.device)
self.impl = WindowsImpl(self.device, window_title='gakumas', ahk_exe_path=None)
self.device._screenshot = self.impl
self.device._touch = self.impl
@ -180,6 +188,16 @@ class RemoteWindowsImpl(Touchable, Screenshotable):
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
if __name__ == "__main__":
import argparse

34
kotonebot/client/implements/uiautomator2.py
View File

@ -6,8 +6,10 @@ import uiautomator2 as u2
from cv2.typing import MatLike
from kotonebot import logging
from ..device import Device
from ..device import Device, AndroidDevice
from ..protocol import Screenshotable, Commandable, Touchable
from ..registration import register_impl
from .adb import AdbImplConfig
logger = logging.getLogger(__name__)
@ -83,3 +85,33 @@ 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:
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 = UiAutomator2Impl(device)
device._command = impl
device._touch = impl
device._screenshot = impl
return device

51
kotonebot/client/implements/windows.py
View File

@ -2,6 +2,7 @@ from ctypes import windll
from typing import Literal
from importlib import resources
from functools import cached_property
from dataclasses import dataclass
import cv2
import win32ui
@ -10,14 +11,21 @@ import numpy as np
from ahk import AHK, MsgBoxIcon
from cv2.typing import MatLike
from ..device import Device
from ..device import Device, WindowsDevice
from ..protocol import Commandable, Touchable, Screenshotable
from ..registration import register_impl, ImplConfig
# 1. 定义配置模型
@dataclass
class WindowsImplConfig(ImplConfig):
window_title: str
ahk_exe_path: str
class WindowsImpl(Touchable, Screenshotable):
def __init__(self, device: Device):
def __init__(self, device: Device, window_title: str, ahk_exe_path: str):
self.__hwnd: int | None = None
# TODO: 硬编码路径
self.ahk = AHK(executable_path=str(resources.files('kaa.res.bin') / 'AutoHotkey.exe'))
self.window_title = window_title
self.ahk = AHK(executable_path=ahk_exe_path)
self.device = device
# 设置 DPI aware，否则高缩放显示器上返回的坐标会错误
@ -55,9 +63,9 @@ class WindowsImpl(Touchable, Screenshotable):
@property
def hwnd(self) -> int:
if self.__hwnd is None:
self.__hwnd = win32gui.FindWindow(None, 'gakumas')
self.__hwnd = win32gui.FindWindow(None, self.window_title)
if self.__hwnd is None or self.__hwnd == 0:
raise RuntimeError('Failed to find window')
raise RuntimeError(f'Failed to find window: {self.window_title}')
return self.__hwnd
def __client_rect(self) -> tuple[int, int, int, int]:
@ -73,8 +81,8 @@ class WindowsImpl(Touchable, Screenshotable):
return win32gui.ClientToScreen(hwnd, (x, y))
def screenshot(self) -> MatLike:
if not self.ahk.win_is_active('gakumas'):
self.ahk.win_activate('gakumas')
if not self.ahk.win_is_active(self.window_title):
self.ahk.win_activate(self.window_title)
hwnd = self.hwnd
# TODO: 需要检查下面这些 WinAPI 的返回结果
@ -130,7 +138,7 @@ class WindowsImpl(Touchable, Screenshotable):
return 720, 1280
def detect_orientation(self) -> None | Literal['portrait'] | Literal['landscape']:
pos = self.ahk.win_get_position('gakumas')
pos = self.ahk.win_get_position(self.window_title)
if pos is None:
return None
w, h = pos.width, pos.height
@ -147,24 +155,37 @@ class WindowsImpl(Touchable, Screenshotable):
if y == 0:
y = 2
x, y = int(x / self.scale_ratio), int(y / self.scale_ratio)
if not self.ahk.win_is_active('gakumas'):
self.ahk.win_activate('gakumas')
if not self.ahk.win_is_active(self.window_title):
self.ahk.win_activate(self.window_title)
self.ahk.click(x, y)
def swipe(self, x1: int, y1: int, x2: int, y2: int, duration: float | None = None) -> None:
if not self.ahk.win_is_active('gakumas'):
self.ahk.win_activate('gakumas')
if not self.ahk.win_is_active(self.window_title):
self.ahk.win_activate(self.window_title)
x1, y1 = int(x1 / self.scale_ratio), int(y1 / self.scale_ratio)
x2, y2 = int(x2 / self.scale_ratio), int(y2 / self.scale_ratio)
# 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
from time import sleep
device = Device()
impl = WindowsImpl(device)
impl = WindowsImpl(device, window_title='gakumas', ahk_exe_path=str(resources.files('kaa.res.bin') / 'AutoHotkey.exe'))
device._screenshot = impl
device._touch = impl
device.swipe_scaled(0.5, 0.8, 0.5, 0.2)

89
kotonebot/client/registration.py Normal file
View File

@ -0,0 +1,89 @@
from dataclasses import dataclass
from typing import TypeVar, Callable, Dict, Type, Any, overload, Literal, cast, TYPE_CHECKING
from ..errors import KotonebotError
from .device import Device
if TYPE_CHECKING:
from .implements.adb import AdbImplConfig
from .implements.remote_windows import RemoteWindowsImplConfig
from .implements.windows import WindowsImplConfig
AdbBasedImpl = Literal['adb', 'adb_raw', 'uiautomator2']
DeviceImpl = str | AdbBasedImpl | Literal['windows', 'remote_windows']
# --- 核心类型定义 ---
class ImplRegistrationError(KotonebotError):
"""与 impl 注册相关的错误"""
pass
@dataclass
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: ...
# 函数的实际实现
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()

6
kotonebot/config/base_config.py
View File

@ -3,7 +3,7 @@ from typing import Generic, TypeVar, Literal
from pydantic import BaseModel, ConfigDict
from kotonebot.client.factory import DeviceImpl
from kotonebot.client import DeviceImpl
T = TypeVar('T')
BackendType = Literal['custom', 'mumu12', 'leidian', 'dmm']
@ -44,6 +44,10 @@ class BackendConfig(ConfigBaseModel):
"""模拟器 exe 文件路径"""
emulator_args: str = ""
"""模拟器启动时的命令行参数"""
windows_window_title: str = 'gakumas'
"""Windows 截图方式的窗口标题"""
windows_ahk_path: str | None = None
"""Windows 截图方式的 AutoHotkey 可执行文件路径，为 None 时使用默认路径"""
class PushConfig(ConfigBaseModel):
"""推送配置。"""

36
kotonebot/kaa/main/dmm_host.py
View File

@ -1,11 +1,18 @@
from importlib import resources
from typing_extensions import override
from kotonebot.client import Device, create_device
from kotonebot.client import DeviceImpl, Device
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.host import HostProtocol, Instance
from kotonebot.client.host.protocol import WindowsHostConfig, RemoteWindowsHostConfig
DmmHostConfigs = WindowsHostConfig | RemoteWindowsHostConfig
# TODO: 可能应该把 start_game 和 end_game 里对启停的操作移动到这里来
class DmmInstance(Instance):
class DmmInstance(Instance[DmmHostConfigs]):
def __init__(self):
super().__init__('dmm', 'gakumas')
@ -30,10 +37,25 @@ class DmmInstance(Instance):
raise NotImplementedError()
@override
def create_device(self, impl: DeviceImpl, *, timeout: float = 180) -> Device:
if impl not in ['windows', 'remote_windows']:
raise ValueError(f'Unsupported device implementation: {impl}')
return create_device('', impl, timeout=timeout)
def create_device(self, impl: DeviceImpl, host_config: DmmHostConfigs) -> Device:
if impl == 'windows':
assert isinstance(host_config, WindowsHostConfig)
config = WindowsImplConfig(
window_title=host_config.window_title,
ahk_exe_path=host_config.ahk_exe_path
)
return create_device(impl, config)
elif impl == 'remote_windows':
assert isinstance(host_config, RemoteWindowsHostConfig)
config = RemoteWindowsImplConfig(
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):
instance = DmmInstance()

153
kotonebot/kaa/main/kaa.py
View File

@ -1,21 +1,30 @@
import io
import os
import logging
import importlib.metadata
import traceback
import zipfile
import logging
import traceback
import importlib.metadata
from datetime import datetime
import cv2
from importlib import resources
from typing_extensions import override
from .dmm_host import DmmHost
import cv2
from ...client import Device
from kotonebot.ui import user
from kotonebot import KotoneBot
from ..kaa_context import _set_instance
from .dmm_host import DmmHost, DmmInstance
from ..common import BaseConfig, upgrade_config
from kotonebot.client.host import Mumu12Host, LeidianHost
from kotonebot.config.base_config import UserConfig
from kotonebot.client.host import (
Mumu12Host, LeidianHost, Mumu12Instance,
LeidianInstance, CustomInstance
)
from kotonebot.client.host.protocol import (
Instance, AdbHostConfig, WindowsHostConfig,
RemoteWindowsHostConfig
)
# 初始化日志
log_formatter = logging.Formatter('[%(asctime)s][%(levelname)s][%(name)s] %(message)s')
@ -104,23 +113,27 @@ class Kaa(KotoneBot):
raise ValueError('Backend instance is not set.')
_set_instance(self.backend_instance)
@override
def _on_create_device(self) -> Device:
def __get_backend_instance(self, config: UserConfig) -> Instance:
"""
根据配置获取或创建 Instance。
:param config: 用户配置对象
:return: 后端实例
"""
from kotonebot.client.host import create_custom
from kotonebot.config.manager import load_config
# HACK: 硬编码
config = load_config(self.config_path, type=self.config_type)
config = config.user_configs[0]
logger.info('Checking backend...')
logger.info(f'Querying for backend: {config.backend.type}')
if config.backend.type == 'custom':
exe = config.backend.emulator_path
self.backend_instance = create_custom(
instance = create_custom(
adb_ip=config.backend.adb_ip,
adb_port=config.backend.adb_port,
adb_name=config.backend.adb_emulator_name,
exe_path=exe,
emulator_args=config.backend.emulator_args
)
# 对于 custom 类型，需要额外验证模拟器路径
if config.backend.check_emulator:
if exe is None:
user.error('「检查并启动模拟器」已开启但未配置「模拟器 exe 文件路径」。')
@ -128,42 +141,96 @@ class Kaa(KotoneBot):
if not os.path.exists(exe):
user.error('「模拟器 exe 文件路径」对应的文件不存在！请检查路径是否正确。')
raise FileNotFoundError(f'Emulator executable not found: {exe}')
if not self.backend_instance.running():
logger.info('Starting custom backend...')
self.backend_instance.start()
logger.info('Waiting for custom backend to be available...')
self.backend_instance.wait_available()
else:
logger.info('Custom backend "%s" already running.', self.backend_instance)
return instance
elif config.backend.type == 'mumu12':
if config.backend.instance_id is None:
raise ValueError('MuMu12 instance ID is not set.')
self.backend_instance = Mumu12Host.query(id=config.backend.instance_id)
if self.backend_instance is None:
instance = Mumu12Host.query(id=config.backend.instance_id)
if instance is None:
raise ValueError(f'MuMu12 instance not found: {config.backend.instance_id}')
if not self.backend_instance.running():
logger.info('Starting MuMu12 backend...')
self.backend_instance.start()
logger.info('Waiting for MuMu12 backend to be available...')
self.backend_instance.wait_available()
else:
logger.info('MuMu12 backend "%s" already running.', self.backend_instance)
return instance
elif config.backend.type == 'leidian':
if config.backend.instance_id is None:
raise ValueError('Leidian instance ID is not set.')
self.backend_instance = LeidianHost.query(id=config.backend.instance_id)
if self.backend_instance is None:
instance = LeidianHost.query(id=config.backend.instance_id)
if instance is None:
raise ValueError(f'Leidian instance not found: {config.backend.instance_id}')
if not self.backend_instance.running():
logger.info('Starting Leidian backend...')
self.backend_instance.start()
logger.info('Waiting for Leidian backend to be available...')
self.backend_instance.wait_available()
else:
logger.info('Leidian backend "%s" already running.', self.backend_instance)
return instance
elif config.backend.type == 'dmm':
self.backend_instance = DmmHost.instance
return DmmHost.instance
else:
raise ValueError(f'Unsupported backend type: {config.backend.type}')
assert self.backend_instance is not None, 'Backend instance is not set.'
return self.backend_instance.create_device(config.backend.screenshot_impl)
def __ensure_instance_running(self, instance: Instance, config: UserConfig):
"""
确保 Instance 正在运行。
:param instance: 后端实例
:param config: 用户配置对象
"""
# DMM 实例不需要启动，直接返回
if isinstance(instance, DmmInstance):
logger.info('DMM backend does not require startup.')
return
# 对所有需要启动的后端（custom, mumu, leidian）使用统一逻辑
if config.backend.check_emulator and not instance.running():
logger.info(f'Starting backend "{instance}"...')
instance.start()
logger.info(f'Waiting for backend "{instance}" to be available...')
instance.wait_available()
else:
logger.info(f'Backend "{instance}" already running or check is disabled.')
@override
def _on_create_device(self) -> Device:
"""
创建设备。
"""
from kotonebot.config.manager import load_config
# 步骤1：加载配置
config = load_config(self.config_path, type=self.config_type)
user_config = config.user_configs[0] # HACK: 硬编码
# 步骤2：获取实例
self.backend_instance = self.__get_backend_instance(user_config)
if self.backend_instance is None:
raise RuntimeError(f"Failed to find instance for backend '{user_config.backend.type}'")
# 步骤3：确保实例运行
self.__ensure_instance_running(self.backend_instance, user_config)
# 步骤4：准备 HostConfig 并创建 Device
impl_name = user_config.backend.screenshot_impl
if isinstance(self.backend_instance, DmmInstance):
if impl_name == 'windows':
ahk_path = str(resources.files('kaa.res.bin') / 'AutoHotkey.exe')
host_conf = WindowsHostConfig(
window_title='gakumas',
ahk_exe_path=ahk_path
)
elif impl_name == 'remote_windows':
host_conf = RemoteWindowsHostConfig(
host=user_config.backend.adb_ip,
port=user_config.backend.adb_port
)
else:
raise ValueError(f"Impl of '{impl_name}' is not supported on DMM.")
return self.backend_instance.create_device(impl_name, host_conf)
# 统一处理所有基于 ADB 的后端
elif isinstance(self.backend_instance, (CustomInstance, Mumu12Instance, LeidianInstance)):
if impl_name in ['adb', 'adb_raw', 'uiautomator2']:
host_conf = AdbHostConfig(timeout=180)
return self.backend_instance.create_device(impl_name, host_conf)
else:
raise ValueError(f"{user_config.backend.type} backend does not support implementation '{impl_name}'")
else:
raise TypeError(f"Unknown instance type: {type(self.backend_instance)}")