feat(cli)!: add direct device action commands and replace press_key with press_button

Add `droidrun device` subgroup with 9 commands: screenshot, ui, tap,
swipe, long-press, type, press, apps, start. These bypass the LLM agent
and talk directly to the device driver.

Replace the Android-centric `press_key(keycode)` interface with a
platform-agnostic `press_button(name)` abstraction across all drivers
(Android, iOS, Cloud, Stealth, Recording). Each driver declares its own
`supported_buttons` set and maps button names internally.

Remove `DroidRunCLI` auto-routing that implicitly routed unknown args
to the `run` command.

BREAKING CHANGE: `DeviceDriver.press_key(keycode)` is removed in favor
of `DeviceDriver.press_button(button)`. Macro recordings now emit
`button_press` action type instead of `key_press`.
This commit is contained in:
johnmalek312
2026-03-26 16:06:27 +11:00
parent 8a7603a36c
commit ebd7941f02
12 changed files with 383 additions and 67 deletions
+5 -16
View File
@@ -124,25 +124,14 @@ async def type(
async def system_button(button: str, *, ctx: "ActionContext") -> ActionResult:
"""Press a system button (back, home, or enter)."""
button_map = {"back": 4, "home": 3, "enter": 66}
button_lower = button.lower()
if button_lower not in button_map:
return ActionResult(
success=False,
summary=f"Failed to press {button} button: unknown button. Valid options: back, home, enter",
)
keycode = button_map[button_lower]
key_names = {66: "ENTER", 4: "BACK", 3: "HOME"}
key_name = key_names.get(keycode, str(keycode))
try:
await ctx.driver.press_key(keycode)
return ActionResult(success=True, summary=f"Pressed {key_name} button")
await ctx.driver.press_button(button)
return ActionResult(success=True, summary=f"Pressed {button.upper()} button")
except ValueError as e:
return ActionResult(success=False, summary=str(e))
except Exception as e:
return ActionResult(
success=False, summary=f"Failed to press {key_name} button: {e}"
success=False, summary=f"Failed to press {button} button: {e}"
)
+1 -1
View File
@@ -78,7 +78,7 @@ ATOMIC_ACTION_SIGNATURES = {
},
"description": 'Press a system button, including back, home, and enter. Usage example: {"action": "system_button", "button": "Home"}',
"function": system_button,
"deps": {"press_key"},
"deps": {"press_button"},
},
"swipe": {
"parameters": {
+281
View File
@@ -0,0 +1,281 @@
"""Direct device action CLI commands.
Provides ``droidrun device <action>`` subcommands that bypass the LLM agent
and talk directly to the device driver.
"""
import asyncio
import tempfile
from functools import wraps
from typing import Optional
import click
from async_adbutils import adb
from rich.console import Console
from droidrun.config_manager import ConfigLoader
from droidrun.portal import ensure_portal_ready
from droidrun.tools.driver.android import AndroidDriver
from droidrun.tools.driver.ios import IOSDriver
from droidrun.tools.filters import ConciseFilter
from droidrun.tools.formatters import IndexedFormatter
from droidrun.tools.ui.ios_provider import IOSStateProvider
from droidrun.tools.ui.provider import AndroidStateProvider
console = Console()
def coro(f):
@wraps(f)
def wrapper(*args, **kwargs):
return asyncio.run(f(*args, **kwargs))
return wrapper
def device_options(f):
"""Common device options for all action commands."""
f = click.option(
"--device", "-d", help="Device serial number or IP address", default=None
)(f)
f = click.option(
"--config", "-c", "config_path", help="Path to config file", default=None
)(f)
f = click.option(
"--tcp/--no-tcp", default=None, help="Use TCP communication"
)(f)
f = click.option(
"--ios", is_flag=True, default=False, help="Target iOS device"
)(f)
return f
async def _create_driver(
device: Optional[str],
config_path: Optional[str],
tcp: Optional[bool],
ios: bool,
):
"""Create and connect a device driver based on CLI options."""
config = ConfigLoader.load(config_path)
if device is not None:
config.device.serial = device
if tcp is not None:
config.device.use_tcp = tcp
if ios:
config.device.platform = "ios"
is_ios = config.device.platform.lower() == "ios"
if is_ios:
if not config.device.serial:
raise click.ClickException("iOS device URL required (--device)")
driver = IOSDriver(url=config.device.serial)
await driver.connect()
return driver, True
serial = config.device.serial
if serial is None:
devices = await adb.list()
if not devices:
raise click.ClickException("No connected Android devices found.")
serial = devices[0].serial
if config.device.auto_setup:
device_obj = await adb.device(serial=serial)
await ensure_portal_ready(device_obj, debug=False)
driver = AndroidDriver(serial=serial, use_tcp=config.device.use_tcp)
await driver.connect()
return driver, False
async def _teardown_android(driver):
"""Disable DroidRun keyboard after direct command execution."""
if isinstance(driver, AndroidDriver) and driver.device:
try:
await driver.device.shell(
"ime disable com.droidrun.portal/.input.DroidrunKeyboardIME"
)
except Exception:
pass
# ---------------------------------------------------------------------------
# Click group
# ---------------------------------------------------------------------------
@click.group()
def device_cli():
"""Direct device actions (screenshot, tap, swipe, etc.)."""
pass
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
@device_cli.command()
@device_options
@coro
async def screenshot(device, config_path, tcp, ios):
"""Take a screenshot and print the saved file path to stdout."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
png_bytes = await driver.screenshot()
fd, path = tempfile.mkstemp(prefix="droidrun_", suffix=".png")
import os
os.write(fd, png_bytes)
os.close(fd)
click.echo(path)
finally:
await _teardown_android(driver)
@device_cli.command()
@device_options
@coro
async def ui(device, config_path, tcp, ios):
"""Print the UI accessibility tree with element bounds for targeting."""
driver, is_ios = await _create_driver(device, config_path, tcp, ios)
try:
if is_ios:
provider = IOSStateProvider(driver)
else:
provider = AndroidStateProvider(
driver,
tree_filter=ConciseFilter(),
tree_formatter=IndexedFormatter(),
)
state = await provider.get_state()
click.echo(state.formatted_text)
if state.phone_state:
click.echo(f"\nPhone state: {state.phone_state}")
finally:
await _teardown_android(driver)
@device_cli.command()
@click.argument("x", type=int)
@click.argument("y", type=int)
@device_options
@coro
async def tap(x, y, device, config_path, tcp, ios):
"""Tap at screen coordinates."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
await driver.tap(x, y)
click.echo(f"Tapped ({x}, {y})")
finally:
await _teardown_android(driver)
@device_cli.command("swipe")
@click.argument("x1", type=int)
@click.argument("y1", type=int)
@click.argument("x2", type=int)
@click.argument("y2", type=int)
@click.option("--duration", type=float, default=1.0, show_default=True, help="Duration in seconds")
@device_options
@coro
async def swipe_cmd(x1, y1, x2, y2, duration, device, config_path, tcp, ios):
"""Swipe from (x1, y1) to (x2, y2)."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
await driver.swipe(x1, y1, x2, y2, duration_ms=duration * 1000)
click.echo(f"Swiped ({x1}, {y1}) -> ({x2}, {y2})")
finally:
await _teardown_android(driver)
@device_cli.command("long-press")
@click.argument("x", type=int)
@click.argument("y", type=int)
@device_options
@coro
async def long_press(x, y, device, config_path, tcp, ios):
"""Long press at screen coordinates."""
driver, is_ios = await _create_driver(device, config_path, tcp, ios)
try:
if is_ios:
raise click.ClickException(
"long-press is not supported on iOS"
)
await driver.swipe(x, y, x, y, 1000)
click.echo(f"Long pressed ({x}, {y})")
finally:
await _teardown_android(driver)
@device_cli.command("type")
@click.argument("text")
@click.option("--clear", is_flag=True, default=False, help="Clear field before typing")
@device_options
@coro
async def type_text(text, clear, device, config_path, tcp, ios):
"""Type text into the currently focused field. Use 'tap' first to focus."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
success = await driver.input_text(text, clear)
if success:
click.echo(f"Typed: {text}")
else:
raise click.ClickException("Failed to type text")
finally:
await _teardown_android(driver)
@device_cli.command()
@click.argument("button", type=click.Choice(["back", "home", "enter"], case_sensitive=False))
@device_options
@coro
async def press(button, device, config_path, tcp, ios):
"""Press a system button."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
await driver.press_button(button)
click.echo(f"Pressed {button}")
except ValueError as e:
raise click.ClickException(str(e)) from None
finally:
await _teardown_android(driver)
@device_cli.command()
@click.option(
"--system/--no-system", default=False, help="Include system apps"
)
@device_options
@coro
async def apps(system, device, config_path, tcp, ios):
"""List installed apps."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
app_list = await driver.get_apps(include_system=system)
for app in app_list:
label = app.get("label", "")
package = app.get("package", "")
if label and label != package:
click.echo(f"{package} ({label})")
else:
click.echo(package)
finally:
await _teardown_android(driver)
@device_cli.command()
@click.argument("package")
@device_options
@coro
async def start(package, device, config_path, tcp, ios):
"""Launch an app by package name."""
driver, _ = await _create_driver(device, config_path, tcp, ios)
try:
result = await driver.start_app(package)
click.echo(result)
finally:
await _teardown_android(driver)
+5 -10
View File
@@ -23,6 +23,7 @@ from droidrun import ResultEvent, DroidAgent
from droidrun.log_handlers import CLILogHandler, configure_logging
from droidrun.cli.event_handler import EventHandler
from droidrun.config_manager import ConfigLoader
from droidrun.cli.device_commands import device_cli
from droidrun.macro.cli import macro_cli
from droidrun.portal import (
PORTAL_PACKAGE_NAME,
@@ -257,15 +258,6 @@ async def run_command(
return False
class DroidRunCLI(click.Group):
def parse_args(self, ctx, args):
# If the first arg is not an option and not a known command, treat as 'run'
if args and not args[0].startswith("-") and args[0] not in self.commands:
args.insert(0, "run")
return super().parse_args(ctx, args)
def _print_version(ctx, param, value):
"""Click callback to print version and exit early when --version is passed."""
if not value or ctx.resilient_parsing:
@@ -304,7 +296,7 @@ def _print_version(ctx, param, value):
ctx.exit()
@click.group(cls=DroidRunCLI)
@click.group()
@click.option(
"--version",
is_flag=True,
@@ -690,6 +682,9 @@ async def ping(device: str | None, tcp: bool | None, debug: bool | None):
# Add macro commands as a subgroup
cli.add_command(macro_cli, name="macro")
# Add device action commands as a subgroup
cli.add_command(device_cli, name="device")
@cli.command()
@click.option("--device", "-d", help="Device serial number or IP address", default=None)
+3
View File
@@ -220,6 +220,9 @@ async def _show_dry_run(
elif action_type == "key_press":
key_name = action.get("key_name", "UNKNOWN")
details = f"{key_name}"
elif action_type == "button_press":
button = action.get("button", "unknown")
details = f"{button}"
elif action_type == "wait":
duration = action.get("duration", 1.0)
details = f"{duration}s"
+16 -3
View File
@@ -14,6 +14,9 @@ from droidrun.tools.driver.android import AndroidDriver
logger = logging.getLogger("droidrun-macro")
# Reverse map for legacy key_press macro entries
_KEYCODE_TO_BUTTON = {4: "back", 3: "home", 66: "enter"}
class MacroPlayer:
"""
@@ -133,13 +136,23 @@ class MacroPlayer:
elif action_type == "key_press":
keycode = action.get("keycode", 0)
logger.info(f"🔘 Pressing key: {keycode}")
await driver.press_key(keycode)
button = _KEYCODE_TO_BUTTON.get(keycode)
if button:
logger.info(f"🔘 Pressing button: {button}")
await driver.press_button(button)
else:
logger.warning(f"⚠️ Unknown keycode {keycode}, skipping")
return True
elif action_type == "button_press":
button = action.get("button", "")
logger.info(f"🔘 Pressing button: {button}")
await driver.press_button(button)
return True
elif action_type == "back":
logger.info("⬅️ Pressing back button")
await driver.press_key(4)
await driver.press_button("back")
return True
elif action_type == "wait":
+17 -3
View File
@@ -28,7 +28,7 @@ class AndroidDriver(DeviceDriver):
"tap",
"swipe",
"input_text",
"press_key",
"press_button",
"start_app",
"screenshot",
"get_ui_tree",
@@ -39,6 +39,14 @@ class AndroidDriver(DeviceDriver):
"drag",
}
supported_buttons = {"back", "home", "enter"}
_BUTTON_KEYCODES = {
"back": 4,
"home": 3,
"enter": 66,
}
def __init__(
self,
serial: str | None = None,
@@ -97,9 +105,15 @@ class AndroidDriver(DeviceDriver):
await self.ensure_connected()
return await self.portal.input_text(text, clear)
async def press_key(self, keycode: int) -> None:
async def press_button(self, button: str) -> None:
await self.ensure_connected()
await self.device.keyevent(keycode)
button_lower = button.lower()
if button_lower not in self.supported_buttons:
raise ValueError(
f"Button '{button}' not supported. "
f"Supported: {', '.join(sorted(self.supported_buttons))}"
)
await self.device.keyevent(self._BUTTON_KEYCODES[button_lower])
async def drag(
self,
+6 -2
View File
@@ -24,6 +24,7 @@ class DeviceDriver:
"""
supported: set[str] = set()
supported_buttons: set[str] = set()
# -- lifecycle -----------------------------------------------------------
@@ -61,8 +62,11 @@ class DeviceDriver:
"""
raise NotImplementedError
async def press_key(self, keycode: int) -> None:
"""Send a single key-event."""
async def press_button(self, button: str) -> None:
"""Press a named button (e.g. back, home, enter).
Raises ``ValueError`` if *button* is not in ``supported_buttons``.
"""
raise NotImplementedError
async def drag(
+20 -11
View File
@@ -26,7 +26,7 @@ class CloudDriver(DeviceDriver):
"tap",
"swipe",
"input_text",
"press_key",
"press_button",
"start_app",
"screenshot",
"get_ui_tree",
@@ -35,9 +35,13 @@ class CloudDriver(DeviceDriver):
"list_packages",
}
# MobileRun global action codes (accessibility service)
_GLOBAL_BACK = 1
_GLOBAL_HOME = 2
supported_buttons = {"back", "home", "enter"}
_BUTTON_ACTIONS = {
"back": 1, # global action: BACK
"home": 2, # global action: HOME
"enter": 66, # keycode passthrough
}
def __init__(
self,
@@ -144,16 +148,21 @@ class CloudDriver(DeviceDriver):
)
return True
async def press_key(self, keycode: int) -> None:
# Map Android keycodes to MobileRun global actions where needed
if keycode == 4: # KEYCODE_BACK
await self.global_action(self._GLOBAL_BACK)
elif keycode == 3: # KEYCODE_HOME
await self.global_action(self._GLOBAL_HOME)
async def press_button(self, button: str) -> None:
button_lower = button.lower()
if button_lower not in self.supported_buttons:
raise ValueError(
f"Button '{button}' not supported. "
f"Supported: {', '.join(sorted(self.supported_buttons))}"
)
if button_lower in ("back", "home"):
await self.global_action(self._BUTTON_ACTIONS[button_lower])
else:
await self._call(
self._client.devices.keyboard.key(
self.device_id, key=keycode, **self._display_kw
self.device_id,
key=self._BUTTON_ACTIONS[button_lower],
**self._display_kw,
)
)
+18 -18
View File
@@ -5,7 +5,7 @@ through the same ``DeviceDriver`` interface used by Android.
Known limitations (pre-existing, documented as TODOs):
- ``clear`` parameter in ``input_text`` is ignored
- ``press_key`` only maps HOME; BACK and ENTER have no iOS equivalent
- ``press_button`` only supports HOME; BACK and ENTER have no iOS equivalent
- ``get_date`` not available (no iOS portal endpoint)
- ``drag`` not implemented
- ``get_apps`` returns bundle identifiers, not real app metadata
@@ -45,15 +45,6 @@ SYSTEM_BUNDLE_IDENTIFIERS = [
"com.apple.webapp",
]
# Android keycode → iOS keycode translation.
# Only HOME is mapped; others have no iOS equivalent.
_ANDROID_TO_IOS_KEYCODE = {
3: 0, # HOME
# TODO: 4 (BACK) has no iOS equivalent
# TODO: 66 (ENTER) has no iOS equivalent
}
class IOSDriver(DeviceDriver):
"""iOS device driver communicating via HTTP REST to the iOS portal app."""
@@ -61,7 +52,7 @@ class IOSDriver(DeviceDriver):
"tap",
"swipe",
"input_text",
"press_key",
"press_button",
"start_app",
"screenshot",
"get_ui_tree",
@@ -69,6 +60,12 @@ class IOSDriver(DeviceDriver):
"get_apps",
}
supported_buttons = {"home"}
_BUTTON_IOS_KEYCODES = {
"home": 0,
}
def __init__(
self,
url: str,
@@ -128,13 +125,16 @@ class IOSDriver(DeviceDriver):
)
return resp.status_code == 200
async def press_key(self, keycode: int) -> None:
ios_keycode = _ANDROID_TO_IOS_KEYCODE.get(keycode)
if ios_keycode is None:
# TODO: BACK (4) and ENTER (66) have no iOS equivalent
logger.warning(f"Keycode {keycode} not supported on iOS, ignoring")
return
resp = await self._client.post("/inputs/key", json={"key": ios_keycode})
async def press_button(self, button: str) -> None:
await self.ensure_connected()
button_lower = button.lower()
if button_lower not in self.supported_buttons:
raise ValueError(
f"Button '{button}' not supported on iOS. "
f"Supported: {', '.join(sorted(self.supported_buttons))}"
)
keycode = self._BUTTON_IOS_KEYCODES[button_lower]
resp = await self._client.post("/inputs/key", json={"key": keycode})
resp.raise_for_status()
# -- app management ------------------------------------------------------
+7 -3
View File
@@ -27,6 +27,10 @@ class RecordingDriver:
def supported(self) -> set[str]:
return self.inner.supported
@property
def supported_buttons(self) -> set[str]:
return self.inner.supported_buttons
def __getattr__(self, name: str):
"""Delegate all non-overridden attribute lookups to the inner driver."""
return getattr(self.inner, name)
@@ -62,9 +66,9 @@ class RecordingDriver:
self.log.append({"action_type": "input_text", "text": text, "clear": clear})
return result
async def press_key(self, keycode: int) -> None:
await self.inner.press_key(keycode)
self.log.append({"action_type": "key_press", "keycode": keycode})
async def press_button(self, button: str) -> None:
await self.inner.press_button(button)
self.log.append({"action_type": "button_press", "button": button})
async def start_app(self, package: str, activity: Optional[str] = None) -> str:
result = await self.inner.start_app(package, activity)
+4
View File
@@ -126,6 +126,10 @@ class StealthDriver:
def supported(self) -> set[str]:
return self.inner.supported
@property
def supported_buttons(self) -> set[str]:
return self.inner.supported_buttons
def __getattr__(self, name: str) -> Any:
return getattr(self.inner, name)