feat(core): 新增远程 Windows 截图控制方式

This commit is contained in:
XcantloadX 2025-04-25 20:18:53 +08:00
parent 1c8621f026
commit 15003b198c
6 changed files with 241 additions and 5 deletions

View File

@ -4,12 +4,13 @@ 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']
DeviceImpl = Literal['adb', 'adb_raw', 'uiautomator2', 'windows', 'remote_windows']
def create_device(
addr: str,
@ -40,4 +41,18 @@ def create_device(
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
return device

View File

@ -0,0 +1,197 @@
"""
Remote Windows implementation using XML-RPC.
This module provides:
1. RemoteWindowsImpl - Client implementation that connects to a remote Windows machine
2. RemoteWindowsServer - Server implementation that exposes a WindowsImpl instance via XML-RPC
"""
import io
import base64
import logging
import xmlrpc.client
import xmlrpc.server
from typing import Literal, cast, Any, Tuple
from functools import cached_property
from threading import Thread
import cv2
import numpy as np
from cv2.typing import MatLike
from kotonebot import logging
from ..device import Device, WindowsDevice
from ..protocol import Touchable, Screenshotable
from .windows import WindowsImpl
logger = logging.getLogger(__name__)
def _encode_image(image: MatLike) -> str:
"""Encode an image as a base64 string."""
success, buffer = cv2.imencode('.png', image)
if not success:
raise RuntimeError("Failed to encode image")
return base64.b64encode(buffer.tobytes()).decode('ascii')
def _decode_image(encoded_image: str) -> MatLike:
"""Decode a base64 string to an image."""
buffer = base64.b64decode(encoded_image)
image = cv2.imdecode(np.frombuffer(buffer, np.uint8), cv2.IMREAD_COLOR)
if image is None:
raise RuntimeError("Failed to decode image")
return image
class RemoteWindowsServer:
"""
XML-RPC server that exposes a WindowsImpl instance.
This class wraps a WindowsImpl instance and exposes its methods via XML-RPC.
"""
def __init__(self, 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)
self.device._screenshot = self.impl
self.device._touch = self.impl
def start(self):
"""Start the XML-RPC server."""
self.server = xmlrpc.server.SimpleXMLRPCServer(
(self.host, self.port),
logRequests=True,
allow_none=True
)
self.server.register_instance(self)
logger.info(f"Starting RemoteWindowsServer on {self.host}:{self.port}")
self.server.serve_forever()
def start_in_thread(self):
"""Start the XML-RPC server in a separate thread."""
thread = Thread(target=self.start, daemon=True)
thread.start()
return thread
# Screenshotable methods
def screenshot(self) -> str:
"""Take a screenshot and return it as a base64-encoded string."""
try:
image = self.impl.screenshot()
return _encode_image(image)
except Exception as e:
logger.error(f"Error taking screenshot: {e}")
raise
def get_screen_size(self) -> tuple[int, int]:
"""Get the screen size."""
return self.impl.screen_size
def detect_orientation(self) -> str | None:
"""Detect the screen orientation."""
return self.impl.detect_orientation()
# Touchable methods
def click(self, x: int, y: int) -> bool:
"""Click at the given coordinates."""
try:
self.impl.click(x, y)
return True
except Exception as e:
logger.error(f"Error clicking at ({x}, {y}): {e}")
return False
def swipe(self, x1: int, y1: int, x2: int, y2: int, duration: float | None = None) -> bool:
"""Swipe from (x1, y1) to (x2, y2)."""
try:
self.impl.swipe(x1, y1, x2, y2, duration)
return True
except Exception as e:
logger.error(f"Error swiping from ({x1}, {y1}) to ({x2}, {y2}): {e}")
return False
# Other methods
def get_scale_ratio(self) -> float:
"""Get the scale ratio."""
return self.impl.scale_ratio
def ping(self) -> bool:
"""Check if the server is alive."""
return True
class RemoteWindowsImpl(Touchable, Screenshotable):
"""
Client implementation that connects to a remote Windows machine via XML-RPC.
This class implements the same interfaces as WindowsImpl but forwards all
method calls to a remote server.
"""
def __init__(self, device: Device, host="localhost", port=8000):
"""Initialize the client with the given device, host, and port."""
self.device = device
self.host = host
self.port = port
self.proxy = xmlrpc.client.ServerProxy(
f"http://{host}:{port}/",
allow_none=True
)
# Test connection
try:
if not self.proxy.ping():
raise ConnectionError(f"Failed to connect to RemoteWindowsServer at {host}:{port}")
logger.info(f"Connected to RemoteWindowsServer at {host}:{port}")
except Exception as e:
raise ConnectionError(f"Failed to connect to RemoteWindowsServer at {host}:{port}: {e}")
@cached_property
def scale_ratio(self) -> float:
"""Get the scale ratio from the remote server."""
return cast(float, self.proxy.get_scale_ratio())
@property
def screen_size(self) -> tuple[int, int]:
"""Get the screen size from the remote server."""
return cast(Tuple[int, int], self.proxy.get_screen_size())
def detect_orientation(self) -> None | Literal['portrait'] | Literal['landscape']:
"""Detect the screen orientation from the remote server."""
return cast(None | Literal['portrait'] | Literal['landscape'], self.proxy.detect_orientation())
def screenshot(self) -> MatLike:
"""Take a screenshot from the remote server."""
encoded_image = cast(str, self.proxy.screenshot())
return _decode_image(encoded_image)
def click(self, x: int, y: int) -> None:
"""Click at the given coordinates on the remote server."""
if not self.proxy.click(x, y):
raise RuntimeError(f"Failed to click at ({x}, {y})")
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})")
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")

View File

@ -22,9 +22,11 @@ class BackendConfig(ConfigBaseModel):
雷电模拟器需要设置正确的模拟器名否则 自动启动模拟器 功能将无法正常工作
其他功能不受影响
"""
screenshot_impl: Literal['adb', 'adb_raw', 'uiautomator2', 'windows'] = 'adb'
screenshot_impl: Literal['adb', 'adb_raw', 'uiautomator2', 'windows', 'remote_windows'] = 'adb'
"""
截图方法暂时推荐使用adb截图方式
如果使用 remote_windows需要在 adb_ip 中填写远程 Windows IP 地址 adb_port 中填写远程 Windows 的端口号
"""
check_emulator: bool = False
"""

View File

@ -2,6 +2,7 @@ import sys
import logging
import argparse
import importlib.metadata
import runpy
from .kaa import Kaa
from kotonebot.backend.context import tasks_from_id, task_registry
@ -27,6 +28,11 @@ invoke_psr.add_argument('task_ids', nargs='*', help='Tasks to invoke')
# task list 子命令
list_psr = task_subparsers.add_parser('list', help='List all available tasks')
# remote-server 子命令
remote_server_psr = subparsers.add_parser('remote-server', help='Start the remote Windows server')
remote_server_psr.add_argument('--host', default='0.0.0.0', help='Host to bind to')
remote_server_psr.add_argument('--port', type=int, default=8000, help='Port to bind to')
_kaa: Kaa | None = None
def kaa() -> Kaa:
global _kaa
@ -61,6 +67,17 @@ def task_list() -> int:
print(f' * {task.id}: {task.name}\n {task.description.strip()}')
return 0
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__')
return 0
except Exception as e:
print(f'Error starting remote server: {e}')
return -1
def main():
args = psr.parse_args()
if args.subcommands == 'task':
@ -68,6 +85,8 @@ def main():
sys.exit(task_invoke())
elif args.task_command == 'list':
sys.exit(task_list())
elif args.subcommands == 'remote-server':
sys.exit(remote_server())
elif args.subcommands is None:
kaa().set_log_level(logging.DEBUG)
from .gr import main as gr_main

View File

@ -534,7 +534,7 @@ class KotoneBotUI:
interactive=True
)
screenshot_impl = gr.Dropdown(
choices=['adb', 'adb_raw', 'uiautomator2', 'windows'],
choices=['adb', 'adb_raw', 'uiautomator2', 'windows', 'remote_windows'],
value=self.current_config.backend.screenshot_impl,
label="截图方法",
info=BackendConfig.model_fields['screenshot_impl'].description,

View File

@ -1,5 +1,6 @@
"""启动游戏,领取登录奖励,直到首页为止"""
import os
import ctypes
import logging
from . import R
@ -9,7 +10,7 @@ from kotonebot.util import Countdown, Interval
from .actions.scenes import at_home, goto_home
from .actions.commu import handle_unread_commu
from kotonebot.errors import GameUpdateNeededError
from kotonebot import task, action, sleep, device, image, ocr
from kotonebot import task, action, sleep, device, image, ocr, config
logger = logging.getLogger(__name__)
@ -96,7 +97,9 @@ def windows_launch():
结束状态游戏窗口出现
"""
# 检查管理员权限
import ctypes, os
# TODO: 检查截图类型不应该依赖配置文件,而是直接检查 device 实例
if config.current.backend.screenshot_impl == 'remote_windows':
raise NotImplementedError("Task `start_game` is not supported on remote_windows.")
try:
is_admin = os.getuid() == 0 # type: ignore
except AttributeError: