from __future__ import annotations import math from dataclasses import dataclass, field from typing import Any from core.errors import DeviceOfflineError, DriverError from driver.base import Driver # Android KeyEvent.KEYCODE_HOME. Kept as a literal rather than importing the # full keycode table because this is the only keycode the Driver ABC exposes. _KEYCODE_HOME = 3 # Default drag speed (pixels/second) used by ``swipe`` when the caller's start # and end coordinates collapse to a zero/near-zero distance, where converting # ``duration_ms`` via ``distance / seconds`` would divide by zero. _DEFAULT_DRAG_SPEED_PX_PER_SEC = 2500 @dataclass(frozen=True) class AndroidDriverConfig: server_url: str = "http://127.0.0.1:4723" platform_name: str = "Android" automation_name: str = "UiAutomator2" device_name: str | None = None udid: str | None = None system_port: int | None = None no_reset: bool = True extra_capabilities: dict[str, Any] = field(default_factory=dict) class AndroidDriver(Driver): def __init__(self, config: AndroidDriverConfig | None = None) -> None: self.config = config or AndroidDriverConfig() self._client: Any | None = None def connect(self) -> None: try: from appium import webdriver from appium.options.android import UiAutomator2Options from appium.webdriver.client_config import AppiumClientConfig except ImportError as exc: raise DriverError("Appium Python client is not installed") from exc capabilities: dict[str, Any] = { "platformName": self.config.platform_name, "automationName": self.config.automation_name, "noReset": self.config.no_reset, **self.config.extra_capabilities, } if self.config.device_name: capabilities["deviceName"] = self.config.device_name if self.config.udid: capabilities["udid"] = self.config.udid if self.config.system_port: capabilities["systemPort"] = self.config.system_port options = UiAutomator2Options().load_capabilities(capabilities) client_config = AppiumClientConfig(remote_server_addr=self.config.server_url) try: self._client = webdriver.Remote( options=options, client_config=client_config, ) except Exception as exc: self._client = None raise DeviceOfflineError("device offline") from exc def disconnect(self) -> None: client = self._require_client() try: client.quit() finally: self._client = None def screenshot(self) -> bytes: client = self._require_client() try: return client.get_screenshot_as_png() except Exception as exc: raise DriverError("screenshot failed") from exc def tap(self, x: float, y: float) -> None: client = self._require_client() try: client.execute_script("mobile: clickGesture", {"x": x, "y": y}) except Exception as exc: raise DriverError("tap failed") from exc def long_press(self, x: float, y: float, duration_ms: int = 1200) -> None: # Android ``mobile: longClickGesture`` uses a system-fixed hold # duration; the caller's ``duration_ms`` is intentionally ignored # (see design R2). Verify on real Appium against the running device # driver — see plan task 2. client = self._require_client() try: client.execute_script("mobile: longClickGesture", {"x": x, "y": y}) except Exception as exc: raise DriverError("long press failed") from exc def swipe( self, start_x: float, start_y: float, end_x: float, end_y: float, duration_ms: int = 500, ) -> None: client = self._require_client() speed = _drag_speed(start_x, start_y, end_x, end_y, duration_ms) try: client.execute_script( "mobile: dragGesture", { "startX": start_x, "startY": start_y, "endX": end_x, "endY": end_y, "speed": speed, }, ) except Exception as exc: raise DriverError("swipe failed") from exc def input(self, text: str) -> None: client = self._require_client() try: client.switch_to.active_element.send_keys(text) except Exception as exc: raise DriverError("text input failed") from exc def launch(self, app_id: str) -> None: client = self._require_client() try: client.activate_app(app_id) except Exception as exc: raise DriverError("app launch failed") from exc def terminate(self, app_id: str) -> None: client = self._require_client() try: client.terminate_app(app_id) except Exception as exc: raise DriverError("app terminate failed") from exc def tree(self) -> str: client = self._require_client() try: return client.page_source except Exception as exc: raise DriverError("ui tree retrieval failed") from exc def home(self) -> None: client = self._require_client() try: client.execute_script("mobile: pressKey", {"keycode": _KEYCODE_HOME}) except Exception as exc: raise DriverError("home failed") from exc def lock(self) -> None: client = self._require_client() try: client.lock() except Exception as exc: raise DriverError("lock failed") from exc def unlock(self) -> None: client = self._require_client() try: client.unlock() except Exception as exc: raise DriverError("unlock failed") from exc def _require_client(self) -> Any: if self._client is None: raise DeviceOfflineError("device offline") return self._client def _drag_speed( start_x: float, start_y: float, end_x: float, end_y: float, duration_ms: int, ) -> int: """Convert ``Driver.swipe``'s ``duration_ms`` into a drag speed (px/s). ``mobile: dragGesture`` takes a speed in pixels/second rather than a duration. Convert via ``distance / seconds`` and guard against a zero/near-zero distance that would otherwise divide by zero. """ distance = math.hypot(end_x - start_x, end_y - start_y) seconds = duration_ms / 1000 if seconds <= 0: return _DEFAULT_DRAG_SPEED_PX_PER_SEC if distance < 1.0: return _DEFAULT_DRAG_SPEED_PX_PER_SEC return int(distance / seconds)