autool/DroidBot/core/abstract_device.py

"""
Abstract Device Base Class
Platform-agnostic device interface that all platform implementations must inherit.
"""
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any, List, Tuple
import logging


class AbstractDevice(ABC):
    """
    所有设备平台的抽象基类
    
    定义了设备操作的标准接口，包括：
    - 设备连接/断开
    - 状态获取
    - 事件发送
    - 屏幕操作
    - 应用管理
    """
    
    def __init__(self, output_dir: Optional[str] = None):
        """
        初始化设备基类
        
        :param output_dir: 输出目录路径
        """
        self.logger = logging.getLogger(self.__class__.__name__)
        self.output_dir = output_dir
        self.connected = False
        self.display_info = None
    
    # ==================== 连接管理 ====================
    
    @abstractmethod
    def set_up(self) -> None:
        """设置设备连接前的准备工作"""
        pass
    
    @abstractmethod
    def connect(self) -> bool:
        """
        连接到设备
        
        :return: 连接是否成功
        """
        pass
    
    @abstractmethod
    def disconnect(self) -> None:
        """断开设备连接"""
        pass
    
    @abstractmethod
    def tear_down(self) -> None:
        """清理设备资源"""
        pass
    
    @abstractmethod
    def check_connectivity(self) -> bool:
        """
        检查设备连接状态
        
        :return: 是否已连接
        """
        pass
    
    # ==================== 状态获取 ====================

    @abstractmethod
    def check_network(self, host: str = "8.8.8.8") -> bool:
        """
        检查设备内部网络是否连通
        
        :param host: 测试目标主机
        :return: True 如果网络可用, False 否则
        """
        pass    
    
    @abstractmethod
    def get_current_state(self) -> 'AbstractDeviceState':
        """
        获取当前设备状态
        
        :return: 设备状态对象
        """
        pass
    
    @abstractmethod
    def get_display_info(self, refresh: bool = False) -> Dict[str, Any]:
        """
        获取显示信息
        
        :param refresh: 是否刷新缓存
        :return: 包含 width, height, density 等信息的字典
        """
        pass
    
    def get_width(self, refresh: bool = False) -> int:
        """获取屏幕宽度"""
        display_info = self.get_display_info(refresh=refresh)
        return display_info.get("width", 0)
    
    def get_height(self, refresh: bool = False) -> int:
        """获取屏幕高度"""
        display_info = self.get_display_info(refresh=refresh)
        return display_info.get("height", 0)
    
    # ==================== 事件发送 ====================
    
    @abstractmethod
    def send_event(self, event: 'AbstractInputEvent') -> bool:
        """
        发送输入事件到设备
        
        :param event: 输入事件对象
        :return: 是否发送成功
        """
        pass
    
    # ==================== 屏幕操作 ====================
    
    @abstractmethod
    def take_screenshot(self, path: str) -> bool:
        """
        截取屏幕
        
        :param path: 截图保存路径
        :return: 是否成功
        """
        pass
    
    @abstractmethod
    def unlock(self) -> None:
        """解锁屏幕"""
        pass
    
    # ==================== 应用管理 ====================
    
    @abstractmethod
    def is_foreground(self) -> bool:
        """
        检查被测应用是否在前台
        
        :return: 是否在前台
        """
        pass
    
    @abstractmethod
    def start_app(self) -> bool:
        """
        启动被测应用
        
        :return: 是否成功
        """
        pass
    
    def install_app(self) -> bool:
        """
        安装应用（可选接口，移动平台适用）
        
        :return: 是否成功
        :raises NotImplementedError: 如果平台不支持
        """
        raise NotImplementedError("This platform does not support app installation")
    
    def uninstall_app(self) -> bool:
        """
        卸载应用（可选接口，移动平台适用）
        
        :return: 是否成功
        :raises NotImplementedError: 如果平台不支持
        """
        raise NotImplementedError("This platform does not support app uninstallation")
    
    @abstractmethod
    def pull_back_to_app(self) -> bool:
        """
        将被测应用拉回前台
        
        :return: 是否成功
        """
        pass

    def get_redirect_target_info(self) -> Optional[Dict[str, Optional[str]]]:
        """
        获取应用离开前台后的目标信息（平台特定，可选接口）

        返回格式:
            {
                "target": 跳转目标标识符,
                "type": "app_store" | "launcher" | "other" | "unknown"
            }

        Android 可返回包名及分类；其他平台默认不实现，返回 None。
        """
        pass
    
    @property
    @abstractmethod
    def app_identifier(self) -> str:
        """
        获取被测应用的唯一标识符
        
        :return: 应用标识符（Android: package_name, Windows: window_class/exe_path）
        """
        pass
    
    # ==================== 设备信息 ====================
    
    @abstractmethod
    def get_platform_name(self) -> str:
        """
        获取平台名称
        
        :return: 平台名称 (如 'android', 'windows')
        """
        pass

    
    # ==================== 可选功能: 性能分析 ====================
    def run_initial_setup(self) -> bool:
        """
        使用 GuiAgent 处理应用的初始化任务（填写初始信息、进入游戏等）
        此方法在 InputPolicy.start() 调用，只执行一次，且目前只用在cv模式
        """
        pass

    def start_profiling(self, trace_file: str, sampling: Optional[int] = None) -> bool:
        """
        启动性能分析(可选接口,平台特定实现)
        
        :param trace_file: 跟踪文件路径
        :param sampling: 采样间隔(可选)
        :return: 是否成功启动
        """
        return False  # 默认不支持
    
    def stop_profiling(self, trace_file: str, output_path: str) -> bool:
        """
        停止性能分析(可选接口,平台特定实现)
        
        :param trace_file: 跟踪文件路径
        :param output_path: 输出路径
        :return: 是否成功停止
        """
        return False  # 默认不支持
    
    def get_traffic_domains(self, remote_dir: str) -> Optional[str]:
        """
        获取最新的流量域名日志文件内容(可选接口)
        
        :param remote_dir: 远程日志目录
        :return: 日志内容或 None
        """
        return None
    
    @property
    def captured_traffic_dir(self) -> Optional[str]:
        """
        获取捕获的流量日志存储目录
        
        :return: 目录路径或 None
        """
        return None
    
    def is_traffic_capture_running(self) -> bool:
        """
        检查流量抓包工具是否正在运行(可选接口)
        
        :return: True 如果正在运行
        """
        raise NotImplementedError("This platform does not support traffic capture")
    
    def restart_traffic_capture(self, package_name: str) -> None:
        """
        重启流量抓包工具(可选接口)
        
        :param package_name: 需要抓包的应用标识符
        """
        raise NotImplementedError("This platform does not support traffic capture")