refactor(core): 提取三个 adb-based 截图方法的工厂函数的共同部分

kotonebot/client/implements/adb.py

@@ -79,9 +79,14 @@ class AdbImpl(Commandable, Touchable, Screenshotable):
         self.adb.shell(f"input touchscreen swipe {x1} {y1} {x2} {y2}")
 
 
-# 编写并注册创建函数
-@register_impl('adb', config_model=AdbImplConfig)
-def create_adb_device(config: AdbImplConfig) -> Device:
+def _create_adb_device_base(config: AdbImplConfig, impl_factory: type) -> Device:
+    """
+    通用的 ADB 设备创建工厂函数。
+    其他任意基于 ADB 的 Impl 可以直接复用这个函数。
+
+    :param config: ADB 实现配置
+    :param impl_factory: 实现类工厂函数，接收 device 参数并返回实现实例
+    """
     from adbutils import adb
 
     if config.disconnect:
@@ -102,8 +107,14 @@ def create_adb_device(config: AdbImplConfig) -> Device:
         raise ValueError(f"Device {config.addr} not found")
     d = d[0]
     device = AndroidDevice(d)
-    impl = AdbImpl(device)
+    impl = impl_factory(device)
     device._command = impl
     device._touch = impl
     device._screenshot = impl
     return device
+
+
+@register_impl('adb', config_model=AdbImplConfig)
+def create_adb_device(config: AdbImplConfig) -> Device:
+    """AdbImpl 工厂函数"""
+    return _create_adb_device_base(config, AdbImpl)

kotonebot/client/implements/adb_raw.py

@@ -11,8 +11,8 @@ import numpy as np
 from cv2.typing import MatLike
 from adbutils._utils import adb_path
 
-from .adb import AdbImpl, AdbImplConfig
-from ..device import Device, AndroidDevice
+from .adb import AdbImpl, AdbImplConfig, _create_adb_device_base
+from ..device import Device
 from ..registration import register_impl
 from kotonebot import logging
 
@@ -160,31 +160,7 @@ class AdbRawImpl(AdbImpl):
         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
+    """AdbRawImpl 工厂函数"""
+    return _create_adb_device_base(config, AdbRawImpl)

kotonebot/client/implements/uiautomator2.py

@@ -6,10 +6,10 @@ import uiautomator2 as u2
 from cv2.typing import MatLike
 
 from kotonebot import logging
-from ..device import Device, AndroidDevice
+from ..device import Device
 from ..protocol import Screenshotable, Commandable, Touchable
 from ..registration import register_impl
-from .adb import AdbImplConfig
+from .adb import AdbImplConfig, _create_adb_device_base
 
 logger = logging.getLogger(__name__)
 
@@ -87,31 +87,7 @@ 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
+    """UiAutomator2Impl 工厂函数"""
+    return _create_adb_device_base(config, UiAutomator2Impl)