Add PII screenshot redaction + uiautomator2 Android widget tree + iOS XCUITest backend

Author: JE-Chen

je_auto_control/__init__.py

@@ -55,6 +55,14 @@
     HealEvent, HealEventLog, HealOutcome, SelfHealError,
     default_heal_log, self_heal_click, self_heal_locate,
 )
+# Screenshot PII redaction (blur regions before VLM upload / audit log).
+from je_auto_control.utils.redaction import (
+    POLICY_MODERATE, POLICY_OFF, POLICY_STRICT,
+    RedactionEngine, RedactionPolicy, RedactionResult,
+    default_policy as default_redaction_policy,
+    policy_from_name as redaction_policy_from_name,
+    redact_png_bytes,
+)
 # WebRunner bridge (headless: optional je_web_runner dependency)
 from je_auto_control.utils.webrunner_bridge import (
     WebRunnerBridgeError, is_webrunner_available, list_webrunner_commands,
@@ -475,6 +483,11 @@ def start_autocontrol_gui(*args, **kwargs):
     # Self-healing locator (image → VLM fallback)
     "HealEvent", "HealEventLog", "HealOutcome", "SelfHealError",
     "default_heal_log", "self_heal_click", "self_heal_locate",
+    # Screenshot redaction (PII blur)
+    "POLICY_MODERATE", "POLICY_OFF", "POLICY_STRICT",
+    "RedactionEngine", "RedactionPolicy", "RedactionResult",
+    "default_redaction_policy", "redaction_policy_from_name",
+    "redact_png_bytes",
     # WebRunner bridge (browser automation via je_web_runner)
     "WebRunnerBridgeError", "is_webrunner_available",
     "list_webrunner_commands", "run_webrunner_action",

je_auto_control/android/__init__.py

@@ -32,7 +32,18 @@
 from je_auto_control.android.adb_client import (
     AdbClient, AdbError, AdbNotAvailable, AndroidDevice,
 )
+from je_auto_control.android.client import (
+    UIAutomatorDevice, UIAutomatorUnavailableError,
+    default_ui_device, reset_default_ui_device,
+)
+from je_auto_control.android.find import (
+    ElementNotFoundError, click_element, dump_hierarchy, find_element,
+)
 
 __all__ = [
     "AdbClient", "AdbError", "AdbNotAvailable", "AndroidDevice",
+    "ElementNotFoundError",
+    "UIAutomatorDevice", "UIAutomatorUnavailableError",
+    "click_element", "default_ui_device", "dump_hierarchy",
+    "find_element", "reset_default_ui_device",
 ]

je_auto_control/android/client.py

@@ -0,0 +1,91 @@
+"""Thin lazy wrapper around ``uiautomator2.Device``.
+
+The ADB-based path in :mod:`adb_client` handles tap / swipe / text /
+screenshot via raw ``adb shell`` commands. ``uiautomator2`` adds what
+``adb shell`` cannot: a live widget tree, blocking ``wait`` for an
+element, and bounding-rect introspection. We keep it in a separate
+class so the cheap adb-only path stays available when the daemon
+isn't installed.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Any, Optional
+
+
+class UIAutomatorUnavailableError(RuntimeError):
+    """Raised when the ``uiautomator2`` SDK or a target device is missing."""
+
+
+class UIAutomatorDevice:
+    """Adapter around ``uiautomator2.Device`` with lazy connection.
+
+    Construct with an optional ``serial`` (the adb device serial as
+    reported by ``adb devices``). When omitted, ``uiautomator2``
+    selects the first attached device. The underlying
+    ``uiautomator2.Device`` is built on first attribute access so
+    importing this module never triggers an adb scan.
+    """
+
+    def __init__(self, serial: Optional[str] = None,
+                 handle: Optional[Any] = None) -> None:
+        self._serial = serial
+        self._handle = handle
+        self._lock = threading.Lock()
+
+    @property
+    def serial(self) -> Optional[str]:
+        return self._serial
+
+    @property
+    def handle(self) -> Any:
+        """Return the underlying ``uiautomator2.Device`` instance.
+
+        Lazily connects on first call. Subsequent calls reuse the
+        handle so the daemon-side session survives across operations.
+        """
+        return self._resolve_handle()
+
+    def _resolve_handle(self) -> Any:
+        with self._lock:
+            if self._handle is not None:
+                return self._handle
+            try:
+                import uiautomator2 as u2
+            except ImportError as error:
+                raise UIAutomatorUnavailableError(
+                    "uiautomator2 not installed. "
+                    "`pip install uiautomator2` and ensure adb sees the "
+                    "device (`adb devices`).",
+                ) from error
+            try:
+                self._handle = u2.connect(self._serial)
+            except (OSError, RuntimeError, ValueError) as error:
+                raise UIAutomatorUnavailableError(
+                    f"could not connect to Android device "
+                    f"{self._serial or '(default)'}: {error}",
+                ) from error
+            return self._handle
+
+
+_DEFAULT_DEVICE: Optional[UIAutomatorDevice] = None
+
+
+def default_ui_device() -> UIAutomatorDevice:
+    """Process-wide default :class:`UIAutomatorDevice` (lazy-built)."""
+    global _DEFAULT_DEVICE
+    if _DEFAULT_DEVICE is None:
+        _DEFAULT_DEVICE = UIAutomatorDevice()
+    return _DEFAULT_DEVICE
+
+
+def reset_default_ui_device() -> None:
+    """Clear the process-wide default — used by tests between cases."""
+    global _DEFAULT_DEVICE
+    _DEFAULT_DEVICE = None
+
+
+__all__ = [
+    "UIAutomatorDevice", "UIAutomatorUnavailableError",
+    "default_ui_device", "reset_default_ui_device",
+]

je_auto_control/android/find.py

@@ -0,0 +1,104 @@
+"""Element lookup over the uiautomator2 widget tree.
+
+The Android equivalent of the macOS accessibility-tree locator:
+callers describe a widget by ``text`` / ``resource_id`` /
+``description`` / ``class_name``, and the helper returns the
+bounding rect or taps it. The thin :func:`dump_hierarchy` is
+exposed so test code can snapshot the live UI tree.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Tuple
+
+from je_auto_control.android.client import (
+    UIAutomatorDevice, default_ui_device,
+)
+
+
+class ElementNotFoundError(LookupError):
+    """Raised when no widget on screen matches the supplied selector."""
+
+
+def _build_query(handle: Any,
+                 text: Optional[str],
+                 resource_id: Optional[str],
+                 description: Optional[str],
+                 class_name: Optional[str]) -> Any:
+    """Translate the public kwargs into uiautomator2's chained selector."""
+    selectors: Dict[str, Any] = {}
+    if text is not None:
+        selectors["text"] = text
+    if resource_id is not None:
+        selectors["resourceId"] = resource_id
+    if description is not None:
+        selectors["description"] = description
+    if class_name is not None:
+        selectors["className"] = class_name
+    if not selectors:
+        raise ValueError(
+            "at least one of text/resource_id/description/class_name "
+            "is required",
+        )
+    return handle(**selectors)
+
+
+def find_element(text: Optional[str] = None,
+                 resource_id: Optional[str] = None,
+                 description: Optional[str] = None,
+                 class_name: Optional[str] = None,
+                 *, timeout_s: float = 5.0,
+                 device: Optional[UIAutomatorDevice] = None,
+                 ) -> Tuple[int, int, int, int]:
+    """Return the matched widget's bounding rect ``(x1, y1, x2, y2)``."""
+    handle = (device or default_ui_device()).handle
+    query = _build_query(handle, text, resource_id, description, class_name)
+    if not query.wait(timeout=float(timeout_s)):
+        raise ElementNotFoundError(
+            f"no widget matched selectors text={text!r} "
+            f"resource_id={resource_id!r} description={description!r} "
+            f"class_name={class_name!r}",
+        )
+    info = query.info
+    bounds = info.get("bounds") or {}
+    return (
+        int(bounds.get("left", 0)),
+        int(bounds.get("top", 0)),
+        int(bounds.get("right", 0)),
+        int(bounds.get("bottom", 0)),
+    )
+
+
+def click_element(text: Optional[str] = None,
+                  resource_id: Optional[str] = None,
+                  description: Optional[str] = None,
+                  class_name: Optional[str] = None,
+                  *, timeout_s: float = 5.0,
+                  device: Optional[UIAutomatorDevice] = None,
+                  ) -> Tuple[int, int]:
+    """Tap the matched widget; return the click-centre ``(x, y)``.
+
+    Uses the uiautomator2 handle for the tap rather than ``adb shell
+    input tap`` so the daemon notices the press synchronously and
+    can update its event queue.
+    """
+    bounds = find_element(
+        text=text, resource_id=resource_id, description=description,
+        class_name=class_name, timeout_s=timeout_s, device=device,
+    )
+    cx = (bounds[0] + bounds[2]) // 2
+    cy = (bounds[1] + bounds[3]) // 2
+    handle = (device or default_ui_device()).handle
+    handle.click(int(cx), int(cy))
+    return (int(cx), int(cy))
+
+
+def dump_hierarchy(*, device: Optional[UIAutomatorDevice] = None) -> str:
+    """Return the device's current widget tree as an XML string."""
+    handle = (device or default_ui_device()).handle
+    return str(handle.dump_hierarchy())
+
+
+__all__ = [
+    "ElementNotFoundError", "click_element", "dump_hierarchy",
+    "find_element",
+]

je_auto_control/ios/__init__.py

@@ -0,0 +1,36 @@
+"""iOS device backend (WebDriverAgent / facebook-wda).
+
+Typical usage::
+
+    from je_auto_control.ios import tap, swipe, find_element, screenshot
+
+    tap(180, 800)
+    swipe(180, 1000, 180, 200)
+    find_element(name="Sign in")
+    screenshot("/tmp/phone.png")
+
+WebDriverAgent must be running on the device — see the Facebook
+WDA README for installation. ``facebook-wda`` is an optional pip
+dependency that loads lazily so importing this module on a non-Mac
+host does not fail.
+"""
+from je_auto_control.ios.client import (
+    IOSDevice, IOSUnavailableError,
+    default_ios_device, reset_default_ios_device,
+)
+from je_auto_control.ios.find import (
+    ElementNotFoundError, click_element, dump_source, find_element,
+)
+from je_auto_control.ios.input import (
+    long_press, press_key, swipe, tap, type_text,
+)
+from je_auto_control.ios.screen import screen_size, screenshot
+
+
+__all__ = [
+    "ElementNotFoundError", "IOSDevice", "IOSUnavailableError",
+    "click_element", "default_ios_device", "dump_source",
+    "find_element", "long_press", "press_key",
+    "reset_default_ios_device", "screen_size", "screenshot", "swipe",
+    "tap", "type_text",
+]

je_auto_control/ios/client.py

@@ -0,0 +1,94 @@
+"""Thin wrapper around ``facebook-wda`` (WebDriverAgent client).
+
+WebDriverAgent (WDA) is the iOS automation server that Appium and
+``facebook-wda`` both talk to. The Python client connects over HTTP
+to a WDA instance running on a real device or the iOS Simulator.
+We import it lazily so the package stays usable on Windows / Linux
+hosts where iOS automation isn't possible.
+
+Setup outside this module:
+
+* Install WDA on the device (``xcodebuild test`` from Facebook's
+  WebDriverAgent project, then run ``iproxy`` to forward 8100).
+* ``pip install facebook-wda``.
+* Point ``IOSDevice(url=...)`` at ``http://127.0.0.1:8100`` (or the
+  WDA hub URL when going through Appium / Sauce Labs).
+"""
+from __future__ import annotations
+
+import threading
+from typing import Any, Optional
+
+
+class IOSUnavailableError(RuntimeError):
+    """Raised when the ``wda`` SDK is missing or the device can't be reached."""
+
+
+class IOSDevice:
+    """Adapter around ``wda.Client`` with a lazy connection.
+
+    ``url`` is the WebDriverAgent HTTP endpoint
+    (default ``http://localhost:8100``). ``handle`` lets tests inject
+    a fake client without ever loading the real SDK.
+    """
+
+    DEFAULT_URL = "http://localhost:8100"
+
+    def __init__(self, url: Optional[str] = None,
+                 handle: Optional[Any] = None) -> None:
+        self._url = url or self.DEFAULT_URL
+        self._handle = handle
+        self._lock = threading.Lock()
+
+    @property
+    def url(self) -> str:
+        return self._url
+
+    @property
+    def handle(self) -> Any:
+        """Return the underlying ``wda.Client`` instance (lazy)."""
+        return self._resolve_handle()
+
+    def _resolve_handle(self) -> Any:
+        with self._lock:
+            if self._handle is not None:
+                return self._handle
+            try:
+                import wda
+            except ImportError as error:
+                raise IOSUnavailableError(
+                    "facebook-wda not installed. "
+                    "`pip install facebook-wda` and run WebDriverAgent "
+                    "on the target device (see the Facebook WDA "
+                    "project README).",
+                ) from error
+            try:
+                self._handle = wda.Client(self._url)
+            except (OSError, RuntimeError, ValueError) as error:
+                raise IOSUnavailableError(
+                    f"could not reach WebDriverAgent at {self._url}: {error}",
+                ) from error
+            return self._handle
+
+
+_DEFAULT_DEVICE: Optional[IOSDevice] = None
+
+
+def default_ios_device() -> IOSDevice:
+    """Process-wide default :class:`IOSDevice` (lazy-built)."""
+    global _DEFAULT_DEVICE
+    if _DEFAULT_DEVICE is None:
+        _DEFAULT_DEVICE = IOSDevice()
+    return _DEFAULT_DEVICE
+
+
+def reset_default_ios_device() -> None:
+    """Clear the process-wide default — used by tests between cases."""
+    global _DEFAULT_DEVICE
+    _DEFAULT_DEVICE = None
+
+
+__all__ = [
+    "IOSDevice", "IOSUnavailableError",
+    "default_ios_device", "reset_default_ios_device",
+]