refactor(core): 将 Commandable 分离为 WindowsCommandable 与 AndroidCommandable

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)

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)
+    adb_imp = AdbRawImpl(d)
-    dd._command = adb_imp
     dd._touch = adb_imp
     dd._screenshot = adb_imp
+    dd.commands = adb_imp
     # dd._screenshot = MinicapScreenshotImpl(dd)
     # dd._screenshot = UiAutomator2Impl(dd)
 

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 ..device import AndroidDevice
-from ..protocol import Commandable, Touchable, Screenshotable
+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):
+class AdbImpl(AndroidCommandable, Touchable, Screenshotable):
-    def __init__(self, device: Device):
+    def __init__(self, adb_connection: AdbUtilsDevice):
-        self.device = device
+        self.adb = adb_connection
-        self.adb = device.adb
 
     @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)
+    impl = impl_class(d)
-    device._command = impl
     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)

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):
+    def __init__(self, adb_connection: AdbUtilsDevice):
-        super().__init__(device)
+        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)

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'): ...

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__")
 

kotonebot/tools/mirror.py

@@ -297,19 +297,27 @@ class DeviceMirrorFrame(wx.Frame):
             self.log("请输入包名")
             return
         try:
-            self.device.launch_app(package_name)
+            # 使用新的 API 通过 commands 属性访问平台特定方法
-            self.log(f"启动APP: {package_name}")
+            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()
+            # 使用新的 API 通过 commands 属性访问平台特定方法
-            if package:
+            if hasattr(self.device, 'commands') and hasattr(self.device.commands, 'current_package'):
-                self.log(f"前台APP: {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)
+    # 使用新的 API
-    adb_imp = AdbImpl(dd)
+    dd = AndroidDevice(d)
-    dd._command = adb_imp
+    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)