Files
agentic-mobile-control/driver/wda_driver.py
T

156 lines
5.0 KiB
Python

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from core.errors import DeviceOfflineError, DriverError
from driver.base import Driver
@dataclass(frozen=True)
class WDADriverConfig:
server_url: str = "http://127.0.0.1:4723"
platform_name: str = "iOS"
automation_name: str = "XCUITest"
device_name: str | None = None
udid: str | None = None
wda_local_port: int | None = None
no_reset: bool = True
extra_capabilities: dict[str, Any] = field(default_factory=dict)
class WDADriver(Driver):
def __init__(self, config: WDADriverConfig | None = None) -> None:
self.config = config or WDADriverConfig()
self._client: Any | None = None
def connect(self) -> None:
try:
from appium import webdriver
from appium.options.ios import XCUITestOptions
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.wda_local_port:
capabilities["wdaLocalPort"] = self.config.wda_local_port
options = XCUITestOptions().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: tap", {"x": x, "y": y})
except Exception as exc:
raise DriverError("tap 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()
try:
client.execute_script(
"mobile: dragFromToForDuration",
{
"fromX": start_x,
"fromY": start_y,
"toX": end_x,
"toY": end_y,
"duration": duration_ms / 1000,
},
)
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: pressButton", {"name": "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