refactor(core): 将启动 remote_server 的逻辑移动到 kaa cli 中

kotonebot/client/host/protocol.py

@@ -33,6 +33,7 @@ class WindowsHostConfig:
 @dataclass
 class RemoteWindowsHostConfig:
     """由外部为远程 Windows 实现提供配置。"""
+    windows_host_config: WindowsHostConfig
     host: str
     port: int
 

kotonebot/client/implements/remote_windows.py

@@ -24,13 +24,14 @@ from kotonebot import logging
 from ..device import Device, WindowsDevice
 from ..protocol import Touchable, Screenshotable
 from ..registration import register_impl, ImplConfig
-from .windows import WindowsImpl
+from .windows import WindowsImpl, WindowsImplConfig, create_windows_device
 
 logger = logging.getLogger(__name__)
 
 # 定义配置模型
 @dataclass
 class RemoteWindowsImplConfig(ImplConfig):
+    windows_impl_config: WindowsImplConfig
     host: str = "localhost"
     port: int = 8000
 
@@ -56,13 +57,13 @@ class RemoteWindowsServer:
     This class wraps a WindowsImpl instance and exposes its methods via XML-RPC.
     """
 
-    def __init__(self, host="localhost", port=8000):
+    def __init__(self, windows_impl_config: WindowsImplConfig, host="localhost", port=8000):
         """Initialize the server with the given host and port."""
         self.host = host
         self.port = port
         self.server = None
         self.device = WindowsDevice()
-        self.impl = WindowsImpl(self.device, window_title='gakumas', ahk_exe_path=None)
+        self.impl = create_windows_device(windows_impl_config)
         self.device._screenshot = self.impl
         self.device._touch = self.impl
 
@@ -195,21 +196,4 @@ def create_remote_windows_device(config: RemoteWindowsImplConfig) -> Device:
     remote_impl = RemoteWindowsImpl(device, config.host, config.port)
     device._touch = remote_impl
     device._screenshot = remote_impl
-    return device
-
-
-if __name__ == "__main__":
-    import argparse
-
-    parser = argparse.ArgumentParser(description="Remote Windows XML-RPC Server")
-    parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
-    parser.add_argument("--port", type=int, default=8000, help="Port to bind to")
-    args = parser.parse_args()
-
-    logging.basicConfig(level=logging.INFO)
-
-    server = RemoteWindowsServer(args.host, args.port)
-    try:
-        server.start()
-    except KeyboardInterrupt:
-        logger.info("Server stopped by user")
+    return device

kotonebot/client/implements/windows.py

@@ -185,7 +185,9 @@ def create_windows_device(config: WindowsImplConfig) -> Device:
 if __name__ == '__main__':
     from ..device import Device
     device = Device()
-    impl = WindowsImpl(device, window_title='gakumas', ahk_exe_path=str(resources.files('kaa.res.bin') / 'AutoHotkey.exe'))
+    # 在测试环境中直接使用默认路径
+    ahk_path = str(resources.files('kaa.res.bin') / 'AutoHotkey.exe')
+    impl = WindowsImpl(device, window_title='gakumas', ahk_exe_path=ahk_path)
     device._screenshot = impl
     device._touch = impl
     device.swipe_scaled(0.5, 0.8, 0.5, 0.2)

kotonebot/kaa/main/cli.py

@@ -1,13 +1,15 @@
 import os
 import sys
-import runpy
 import logging
 import argparse
 import importlib.metadata
 from datetime import datetime
 
 from .kaa import Kaa
+from ..util.paths import get_ahk_path
+from kotonebot.client.implements.windows import WindowsImplConfig
 from kotonebot.backend.context import tasks_from_id, task_registry
+from kotonebot.client.implements.remote_windows import RemoteWindowsServer
 
 version = importlib.metadata.version('ksaa')
 
@@ -90,9 +92,12 @@ def task_list() -> int:
 def remote_server() -> int:
     args = psr.parse_args()
     try:
-        # 使用runpy运行remote_windows.py模块
-        sys.argv = ['remote_windows.py', f'--host={args.host}', f'--port={args.port}']
-        runpy.run_module('kotonebot.client.implements.remote_windows', run_name='__main__')
+        ahk_path = get_ahk_path()
+        server = RemoteWindowsServer(WindowsImplConfig(window_title='gakumas', ahk_exe_path=ahk_path), args.host, args.port)
+        server.start()
+        return 0
+    except KeyboardInterrupt:
+        print("Server stopped by user")
         return 0
     except Exception as e:
         print(f'Error starting remote server: {e}')

kotonebot/kaa/main/dmm_host.py

@@ -40,15 +40,19 @@ class DmmInstance(Instance[DmmHostConfigs]):
     def create_device(self, impl: DeviceImpl, host_config: DmmHostConfigs) -> Device:
         if impl == 'windows':
             assert isinstance(host_config, WindowsHostConfig)
-            config = WindowsImplConfig(
+            win_config = WindowsImplConfig(
                 window_title=host_config.window_title,
                 ahk_exe_path=host_config.ahk_exe_path
             )
-            return create_device(impl, config)
+            return create_device(impl, win_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
             )

kotonebot/kaa/main/kaa.py

@@ -5,7 +5,6 @@ import logging
 import traceback
 import importlib.metadata
 from datetime import datetime
-from importlib import resources
 from typing_extensions import override
 
 import cv2
@@ -13,6 +12,7 @@ import cv2
 from ...client import Device
 from kotonebot.ui import user
 from kotonebot import KotoneBot
+from ..util.paths import get_ahk_path
 from ..kaa_context import _set_instance
 from .dmm_host import DmmHost, DmmInstance
 from ..common import BaseConfig, upgrade_config
@@ -210,13 +210,18 @@ class Kaa(KotoneBot):
 
         if isinstance(self.backend_instance, DmmInstance):
             if impl_name == 'windows':
-                ahk_path = str(resources.files('kaa.res.bin') / 'AutoHotkey.exe')
+                ahk_path = get_ahk_path()
                 host_conf = WindowsHostConfig(
                     window_title='gakumas',
                     ahk_exe_path=ahk_path
                 )
             elif impl_name == 'remote_windows':
+                ahk_path = get_ahk_path()
                 host_conf = RemoteWindowsHostConfig(
+                    windows_host_config=WindowsHostConfig(
+                        window_title='gakumas',
+                        ahk_exe_path=ahk_path
+                    ),
                     host=user_config.backend.adb_ip,
                     port=user_config.backend.adb_port
                 )

kotonebot/kaa/tasks/start_game.py

@@ -156,8 +156,8 @@ def windows_launch():
             logger.info('Gakumas Localify disabled.')
     
     from ahk import AHK
-    from importlib import resources
-    ahk_path = str(resources.files('kaa.res.bin') / 'AutoHotkey.exe')
+    from kotonebot.kaa.util.paths import get_ahk_path
+    ahk_path = get_ahk_path()
     ahk = AHK(executable_path=ahk_path)
 
     if ahk.find_window(title='gakumas', title_match_mode=3): # 3=精确匹配

kotonebot/kaa/util/paths.py

@@ -1,5 +1,6 @@
 import os
 from typing import cast
+from importlib import resources
 
 from kotonebot.kaa import resources as res
 
@@ -15,4 +16,8 @@ def cache(path: str) -> str:
     return p
 
 def resource(path: str) -> str:
-    return os.path.join(RESOURCE, path)
+    return os.path.join(RESOURCE, path)
+
+def get_ahk_path() -> str:
+    """获取 AutoHotkey 可执行文件路径"""
+    return str(resources.files('kaa.res.bin') / 'AutoHotkey.exe')