From 16a267de79fcc70b2872fc6ab37b9434fffa85cd Mon Sep 17 00:00:00 2001 From: XcantloadX <3188996979@qq.com> Date: Wed, 11 Jun 2025 08:49:57 +0800 Subject: [PATCH] =?UTF-8?q?refactor(core):=20=E5=B0=86=20Commandable=20?= =?UTF-8?q?=E5=88=86=E7=A6=BB=E4=B8=BA=20WindowsCommandable=20=E4=B8=8E=20?= =?UTF-8?q?AndroidCommandable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kotonebot/backend/context/context.py | 4 +-- kotonebot/client/device.py | 44 ++++++++++++++------------ kotonebot/client/implements/adb.py | 32 ++++++++++++------- kotonebot/client/implements/adb_raw.py | 8 ++--- kotonebot/client/protocol.py | 13 ++++++++ kotonebot/debug_entry.py | 1 + kotonebot/tools/mirror.py | 36 +++++++++++++-------- 7 files changed, 86 insertions(+), 52 deletions(-) diff --git a/kotonebot/backend/context/context.py b/kotonebot/backend/context/context.py index 527b5a1..5660832 100644 --- a/kotonebot/backend/context/context.py +++ b/kotonebot/backend/context/context.py @@ -747,13 +747,13 @@ class ContextDevice(Device): return img def __getattribute__(self, name: str) -> Any: - if name in ['_device', 'screenshot']: + if name in ['_device', 'screenshot', 'of_android', 'of_windows']: return object.__getattribute__(self, name) else: return getattr(self._device, name) def __setattr__(self, name: str, value: Any): - if name in ['_device', 'screenshot']: + if name in ['_device', 'screenshot', 'of_android', 'of_windows']: return object.__setattr__(self, name, value) else: return setattr(self._device, name, value) diff --git a/kotonebot/client/device.py b/kotonebot/client/device.py index 2478e8b..52894d4 100644 --- a/kotonebot/client/device.py +++ b/kotonebot/client/device.py @@ -11,7 +11,7 @@ from adbutils._device import AdbDevice as AdbUtilsDevice from ..backend.debug import result from kotonebot.backend.core import HintBox from kotonebot.primitives import Rect, Point, is_point -from .protocol import ClickableObjectProtocol, Commandable, Touchable, Screenshotable +from .protocol import ClickableObjectProtocol, Commandable, Touchable, Screenshotable, AndroidCommandable, WindowsCommandable logger = logging.getLogger(__name__) @@ -71,7 +71,6 @@ class Device: 横屏时为 'landscape'，竖屏时为 'portrait'。 """ - self._command: Commandable self._touch: Touchable self._screenshot: Screenshotable @@ -90,12 +89,6 @@ class Device: def adb(self, value: AdbUtilsDevice) -> None: self._adb = value - def launch_app(self, package_name: str) -> None: - """ - 根据包名启动 app - """ - self._command.launch_app(package_name) - @overload def click(self) -> None: """ @@ -306,17 +299,6 @@ class Device: """ return self._screenshot.screen_size - def current_package(self) -> str | None: - """ - 获取前台 APP 的包名。 - - :return: 前台 APP 的包名。如果获取失败，则返回 None。 - :exception: 如果设备不支持此功能，则抛出 NotImplementedError。 - """ - ret = self._command.current_package() - logger.debug("current_package: %s", ret) - return ret - def detect_orientation(self) -> Literal['portrait', 'landscape'] | None: """ 检测当前设备方向并设置 `self.orientation` 属性。 @@ -330,10 +312,30 @@ class AndroidDevice(Device): def __init__(self, adb_connection: AdbUtilsDevice | None = None) -> None: super().__init__('android') self._adb: AdbUtilsDevice | None = adb_connection + self.commands: AndroidCommandable + + def current_package(self) -> str | None: + """ + 获取前台 APP 的包名。 + + :return: 前台 APP 的包名。如果获取失败，则返回 None。 + :exception: 如果设备不支持此功能，则抛出 NotImplementedError。 + """ + ret = self.commands.current_package() + logger.debug("current_package: %s", ret) + return ret + + def launch_app(self, package_name: str) -> None: + """ + 根据包名启动 app + """ + self.commands.launch_app(package_name) + class WindowsDevice(Device): def __init__(self) -> None: super().__init__('windows') + self.commands: WindowsCommandable if __name__ == "__main__": @@ -346,10 +348,10 @@ if __name__ == "__main__": d = adb.device_list()[-1] d.shell("dumpsys activity top | grep ACTIVITY | tail -n 1") dd = AndroidDevice(d) - adb_imp = AdbRawImpl(dd) - dd._command = adb_imp + adb_imp = AdbRawImpl(d) dd._touch = adb_imp dd._screenshot = adb_imp + dd.commands = adb_imp # dd._screenshot = MinicapScreenshotImpl(dd) # dd._screenshot = UiAutomator2Impl(dd) diff --git a/kotonebot/client/implements/adb.py b/kotonebot/client/implements/adb.py index cb898d8..e87fce0 100644 --- a/kotonebot/client/implements/adb.py +++ b/kotonebot/client/implements/adb.py @@ -5,9 +5,10 @@ from typing_extensions import override import cv2 import numpy as np from cv2.typing import MatLike +from adbutils._device import AdbDevice as AdbUtilsDevice -from ..device import Device, AndroidDevice -from ..protocol import Commandable, Touchable, Screenshotable +from ..device import AndroidDevice +from ..protocol import AndroidCommandable, Touchable, Screenshotable from ..registration import register_impl, ImplConfig from dataclasses import dataclass @@ -22,10 +23,9 @@ class AdbImplConfig(ImplConfig): device_serial: str | None = None timeout: float = 180 -class AdbImpl(Commandable, Touchable, Screenshotable): - def __init__(self, device: Device): - self.device = device - self.adb = device.adb +class AdbImpl(AndroidCommandable, Touchable, Screenshotable): + def __init__(self, adb_connection: AdbUtilsDevice): + self.adb = adb_connection @override def launch_app(self, package_name: str) -> None: @@ -47,6 +47,10 @@ class AdbImpl(Commandable, Touchable, Screenshotable): package = activity.split('/')[0] return package + def adb_shell(self, cmd: str) -> str: + """执行 ADB shell 命令""" + return cast(str, self.adb.shell(cmd)) + @override def detect_orientation(self): # 判断方向：https://stackoverflow.com/questions/10040624/check-if-device-is-landscape-via-adb @@ -61,7 +65,9 @@ class AdbImpl(Commandable, Touchable, Screenshotable): def screen_size(self) -> tuple[int, int]: ret = cast(str, self.adb.shell("wm size")).strip('Physical size: ') spiltted = tuple(map(int, ret.split("x"))) - landscape = self.device.orientation == 'landscape' + # 检测当前方向 + orientation = self.detect_orientation() + landscape = orientation == 'landscape' spiltted = tuple(sorted(spiltted, reverse=landscape)) if len(spiltted) != 2: raise ValueError(f"Invalid screen size: {ret}") @@ -79,13 +85,13 @@ class AdbImpl(Commandable, Touchable, Screenshotable): self.adb.shell(f"input touchscreen swipe {x1} {y1} {x2} {y2}") -def _create_adb_device_base(config: AdbImplConfig, impl_factory: type) -> Device: +def _create_adb_device_base(config: AdbImplConfig, impl_class: type) -> AndroidDevice: """ 通用的 ADB 设备创建工厂函数。 其他任意基于 ADB 的 Impl 可以直接复用这个函数。 :param config: ADB 实现配置 - :param impl_factory: 实现类工厂函数，接收 device 参数并返回实现实例 + :param impl_class: 实现类或工厂函数。构造函数接收 adb_connection 参数。 """ from adbutils import adb @@ -106,15 +112,17 @@ def _create_adb_device_base(config: AdbImplConfig, impl_factory: type) -> Device if len(d) == 0: raise ValueError(f"Device {config.addr} not found") d = d[0] + device = AndroidDevice(d) - impl = impl_factory(device) - device._command = impl + 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) -> Device: +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 df9f048..985d6b6 100644 --- a/kotonebot/client/implements/adb_raw.py +++ b/kotonebot/client/implements/adb_raw.py @@ -12,7 +12,7 @@ from cv2.typing import MatLike from adbutils._utils import adb_path from .adb import AdbImpl, AdbImplConfig, _create_adb_device_base -from ..device import Device +from adbutils._device import AdbDevice as AdbUtilsDevice from ..registration import register_impl from kotonebot import logging @@ -28,8 +28,8 @@ done """ class AdbRawImpl(AdbImpl): - def __init__(self, device: Device): - super().__init__(device) + def __init__(self, adb_connection: AdbUtilsDevice): + super().__init__(adb_connection) self.__worker: Thread | None = None self.__process: subprocess.Popen | None = None self.__data: MatLike | None = None @@ -161,6 +161,6 @@ class AdbRawImpl(AdbImpl): @register_impl('adb_raw', config_model=AdbImplConfig) -def create_adb_raw_device(config: AdbImplConfig) -> Device: +def create_adb_raw_device(config: AdbImplConfig): """AdbRawImpl 工厂函数""" return _create_adb_device_base(config, AdbRawImpl) \ No newline at end of file diff --git a/kotonebot/client/protocol.py b/kotonebot/client/protocol.py index 4691901..6dad0f4 100644 --- a/kotonebot/client/protocol.py +++ b/kotonebot/client/protocol.py @@ -28,6 +28,19 @@ class Commandable(Protocol): def launch_app(self, package_name: str) -> None: ... def current_package(self) -> str | None: ... +@runtime_checkable +class AndroidCommandable(Protocol): + """定义 Android 平台的特定命令""" + def launch_app(self, package_name: str) -> None: ... + def current_package(self) -> str | None: ... + def adb_shell(self, cmd: str) -> str: ... + +@runtime_checkable +class WindowsCommandable(Protocol): + """定义 Windows 平台的特定命令""" + def get_foreground_window(self) -> tuple[int, str]: ... + def exec_command(self, command: str) -> tuple[int, str, str]: ... + @runtime_checkable class Screenshotable(Protocol): def __init__(self, device: 'Device'): ... diff --git a/kotonebot/debug_entry.py b/kotonebot/debug_entry.py index 820b4ea..1dfae1b 100644 --- a/kotonebot/debug_entry.py +++ b/kotonebot/debug_entry.py @@ -24,6 +24,7 @@ def run_script(script_path: str) -> None: config_path = './config.json' kaa_instance = Kaa(config_path) init_context(config_type=BaseConfig, target_device=kaa_instance._on_create_device()) + kaa_instance._on_after_init_context() manual_context().begin() runpy.run_module(module_name, run_name="__main__") diff --git a/kotonebot/tools/mirror.py b/kotonebot/tools/mirror.py index d274545..1d93d11 100644 --- a/kotonebot/tools/mirror.py +++ b/kotonebot/tools/mirror.py @@ -297,19 +297,27 @@ class DeviceMirrorFrame(wx.Frame): self.log("请输入包名") return try: - self.device.launch_app(package_name) - self.log(f"启动APP: {package_name}") + # 使用新的 API 通过 commands 属性访问平台特定方法 + if hasattr(self.device, 'commands') and hasattr(self.device.commands, 'launch_app'): + self.device.commands.launch_app(package_name) + self.log(f"启动APP: {package_name}") + else: + self.log("当前设备不支持启动APP功能") except Exception as e: self.log(f"启动APP失败: {e}") def on_get_current_app(self, event): """获取前台APP""" try: - package = self.device.current_package() - if package: - self.log(f"前台APP: {package}") + # 使用新的 API 通过 commands 属性访问平台特定方法 + if hasattr(self.device, 'commands') and hasattr(self.device.commands, 'current_package'): + package = self.device.commands.current_package() + if package: + self.log(f"前台APP: {package}") + else: + self.log("未获取到前台APP") else: - self.log("获取前台APP失败") + self.log("当前设备不支持获取前台APP功能") except Exception as e: self.log(f"获取前台APP失败: {e}") @@ -326,19 +334,21 @@ def show_device_mirror(device: Device): if __name__ == "__main__": # 测试代码 + from kotonebot.client.device import AndroidDevice from kotonebot.client.implements.adb import AdbImpl from kotonebot.client.implements.uiautomator2 import UiAutomator2Impl from adbutils import adb - + print("server version:", adb.server_version()) adb.connect("127.0.0.1:5555") print("devices:", adb.device_list()) d = adb.device_list()[-1] - - dd = Device(d) - adb_imp = AdbImpl(dd) - dd._command = adb_imp + + # 使用新的 API + dd = AndroidDevice(d) + adb_imp = AdbImpl(d) # 直接传入 adb 连接 dd._touch = adb_imp - dd._screenshot = UiAutomator2Impl(dd) - + dd._screenshot = UiAutomator2Impl(dd) # UiAutomator2Impl 可能还需要 device 对象 + dd.commands = adb_imp # 设置 Android 特定命令 + show_device_mirror(dd)