trycua--cua/libs/python/cua-sandbox/cua_sandbox/interfaces/mobile.py

"""Mobile interface — touch, gestures, and hardware keys via ADB, backed by a Transport."""

from __future__ import annotations

import asyncio

from cua_sandbox.transport.base import Transport


class Mobile:
    """Mobile (Android) touch and hardware-key control.

    All touch coordinates are in screen pixels.  Single-touch methods use
    ``input tap/swipe`` via ``adb shell``.  Multi-touch gestures delegate to
    the ``multitouch_gesture`` transport action, which uses ``adb root`` +
    MT Protocol B ``sendevent`` for reliable injection on both local and cloud
    transports.
    """

    def __init__(self, transport: Transport):
        self._t = transport

    async def _shell(self, command: str) -> None:
        await self._t.send("shell", command=command)

    # ── Touch ─────────────────────────────────────────────────────────────

    async def tap(self, x: int, y: int) -> None:
        await self._shell(f"input tap {x} {y}")

    async def long_press(self, x: int, y: int, duration_ms: int = 1000) -> None:
        await self._shell(f"input swipe {x} {y} {x} {y} {duration_ms}")

    async def double_tap(self, x: int, y: int, delay: float = 0.1) -> None:
        await self._shell(f"input tap {x} {y}")
        await asyncio.sleep(delay)
        await self._shell(f"input tap {x} {y}")

    async def type_text(self, text: str) -> None:
        encoded = text.replace(" ", "%s")
        await self._shell(f"input text '{encoded}'")

    # ── Gestures ──────────────────────────────────────────────────────────

    async def swipe(self, x1: int, y1: int, x2: int, y2: int, duration_ms: int = 300) -> None:
        await self._shell(f"input swipe {x1} {y1} {x2} {y2} {duration_ms}")

    async def scroll_up(self, x: int, y: int, distance: int = 600, duration_ms: int = 400) -> None:
        await self._shell(f"input swipe {x} {y + distance} {x} {y} {duration_ms}")

    async def scroll_down(
        self, x: int, y: int, distance: int = 600, duration_ms: int = 400
    ) -> None:
        await self._shell(f"input swipe {x} {y} {x} {y + distance} {duration_ms}")

    async def scroll_left(
        self, x: int, y: int, distance: int = 400, duration_ms: int = 300
    ) -> None:
        await self._shell(f"input swipe {x + distance} {y} {x} {y} {duration_ms}")

    async def scroll_right(
        self, x: int, y: int, distance: int = 400, duration_ms: int = 300
    ) -> None:
        await self._shell(f"input swipe {x} {y} {x + distance} {y} {duration_ms}")

    async def fling(self, x1: int, y1: int, x2: int, y2: int) -> None:
        await self.swipe(x1, y1, x2, y2, duration_ms=80)

    async def _get_screen_size(self) -> tuple[int, int]:
        """Return ``(width, height)`` in screen pixels via ``wm size``."""
        result = await self._t.send("shell", command="wm size")
        output = result.get("stdout", "").strip()
        for line in output.splitlines():
            if "size" in line.lower():
                dims = line.split(":")[-1].strip()
                w, h = dims.split("x")
                return int(w), int(h)
        return 32768, 32768  # fallback: 1:1 mapping

    async def gesture(
        self,
        *finger_paths: tuple[int, int],
        duration_ms: int = 400,
        steps: int = 0,
    ) -> None:
        """Inject an arbitrary N-finger gesture via MT Protocol B ``sendevent``.

        Each positional argument is a sequence of ``(x, y)`` waypoints for one
        finger.  Pass an even number of ``(x, y)`` tuples and they are paired
        into start/end positions per finger::

            # two-finger pinch-out
            await mobile.gesture(
                (cx - 20, cy), (cx - 200, cy),   # finger 0
                (cx + 20, cy), (cx + 200, cy),   # finger 1
            )

        Delegates to the ``multitouch_gesture`` transport action, which uses
        ``adb root`` + MT Protocol B ``sendevent`` for reliable injection on
        both local ADB and cloud HTTP transports.

        Args:
            *finger_paths: Alternating start/end ``(x, y)`` tuples, two per finger.
                           Must have an even length >= 4 (at least 2 fingers × 2 points).
            duration_ms:   Total gesture duration in milliseconds.
            steps:         Interpolation steps (0 = auto: duration_ms // 20, min 5).
        """
        if len(finger_paths) < 4 or len(finger_paths) % 2 != 0:
            raise ValueError(
                "gesture() requires an even number of (x, y) tuples, two per finger "
                f"(start + end).  Got {len(finger_paths)} tuples."
            )

        # Pair waypoints into per-finger (start, end) tuples
        fingers_pairs: list[tuple[tuple[int, int], tuple[int, int]]] = []
        for i in range(0, len(finger_paths), 2):
            fingers_pairs.append((finger_paths[i], finger_paths[i + 1]))

        sw, sh = await self._get_screen_size()
        n_steps = steps if steps > 0 else max(5, duration_ms // 20)

        # Structured params for the server-side multitouch_gesture action
        fingers_payload = [{"start": list(start), "end": list(end)} for start, end in fingers_pairs]
        await self._t.send(
            "multitouch_gesture",
            fingers=fingers_payload,
            screen_w=sw,
            screen_h=sh,
            duration_ms=duration_ms,
            steps=n_steps,
        )

    async def pinch_in(self, cx: int, cy: int, spread: int = 300, duration_ms: int = 400) -> None:
        """Pinch-in (zoom out) with two real simultaneous fingers."""
        await self.gesture(
            (cx - spread, cy),
            (cx - 20, cy),
            (cx + spread, cy),
            (cx + 20, cy),
            duration_ms=duration_ms,
        )

    async def pinch_out(self, cx: int, cy: int, spread: int = 300, duration_ms: int = 400) -> None:
        """Pinch-out (zoom in) with two real simultaneous fingers."""
        await self.gesture(
            (cx - 20, cy),
            (cx - spread, cy),
            (cx + 20, cy),
            (cx + spread, cy),
            duration_ms=duration_ms,
        )

    # ── Hardware keys ─────────────────────────────────────────────────────

    async def key(self, keycode: int) -> None:
        await self._shell(f"input keyevent {keycode}")

    async def home(self) -> None:
        await self.key(3)

    async def back(self) -> None:
        await self.key(4)

    async def recents(self) -> None:
        await self.key(187)

    async def power(self) -> None:
        await self.key(26)

    async def volume_up(self) -> None:
        await self.key(24)

    async def volume_down(self) -> None:
        await self.key(25)

    async def enter(self) -> None:
        await self.key(66)

    async def backspace(self) -> None:
        await self.key(67)

    # ── System ────────────────────────────────────────────────────────────

    async def wake(self) -> None:
        await self.key(224)

    async def notifications(self) -> None:
        await self._shell("service call statusbar 1")

    async def close_notifications(self) -> None:
        await self._shell("service call statusbar 2")