adbflow 0.2.0

Feature-rich Android device automation via ADB

adbflow

Async-first, type-safe Python package for Android device automation via ADB.

adbflow gives you a clean, modern Python API for everything you'd do with adb — UI automation, gestures, app management, file operations, screen capture, and much more. Built on asyncio with full type hints.

Documentation · PyPI · GitHub

---

Features

• Async-first — every operation is async/await with sync wrappers available
• Type-safe — full type hints, passes mypy --strict
• UI automation — find elements by text/ID/class, tap, type, wait for conditions
• Gestures — tap, swipe, drag, pinch, multi-touch
• App management — install, launch, stop, permissions, intents
• File operations — push, pull, ls, sync directories with checksum diffing
• Screen capture — screenshots (bytes, PIL, file), screen recording
• Wait engine — composable conditions with AllOf, AnyOf, Not
• Vision — template matching and color detection (optional: opencv)
• OCR — on-screen text recognition (optional: easyocr)
• Watchers — auto-dismiss dialogs, background UI monitors
• Recorder — record and replay action sequences
• Flow engine — multi-step workflows with conditions and retry logic
• Multi-device — parallel execution across device pools
• Logcat — stream, filter, and capture logs
• Network — WiFi, mobile data, airplane mode, port forwarding, proxy
• Notifications — list, find, wait for notifications
• CLI — adbflow command for common operations

Installation

pip install adbflow

Optional extras

pip install adbflow[vision]   # Template matching, color detection (opencv)
pip install adbflow[ocr]      # Text recognition (EasyOCR)
pip install adbflow[all]      # Everything

Requirements

• Python 3.10+
• ADB installed and on PATH (Android Platform-Tools)
• An Android device with USB Debugging enabled

Quick Start

import asyncio
from adbflow import ADB

async def main():
    adb = ADB()
    device = await adb.device_async()

    # Device info
    model = await device.info.model_async()
    print(f"Connected to {model}")

    # Shell commands
    output = await device.shell_async("getprop ro.product.model")

    # UI automation
    from adbflow.ui import Selector

    element = await device.ui.find_async(
        Selector().text("Settings").clickable()
    )
    if element:
        await element.tap_async()

    # Wait for UI state
    await device.wait_for_text_async("Display", timeout=5.0)

    # Gestures
    from adbflow.utils.types import SwipeDirection
    await device.gestures.swipe_direction_async(SwipeDirection.UP)

    # Screenshots
    png_bytes = await device.screenshot_async()

    # App management
    await device.apps.start_async("com.example.app")
    await device.apps.stop_async("com.example.app")

asyncio.run(main())

Sync API

Every _async method has a sync wrapper for scripts that don't need asyncio:

from adbflow import ADB

adb = ADB()
device = adb.device()
output = device.shell("getprop ro.product.model")
print(output)

Modules

Module	Access	Description
UI	device.ui	Find elements, tap, type, wait
Gestures	device.gestures	Tap, swipe, drag, pinch, keys
Apps	device.apps	Install, launch, stop, permissions, intents
Files	device.files	Push, pull, ls, sync, backup
Media	device.media	Screenshots, screen recording, audio
Network	device.network	WiFi, mobile data, airplane, port forwarding
Logcat	device.logcat	Stream, filter, capture, crash detection
Notifications	device.notifications	List, find, wait for notifications
Wait	device.wait_for_text_async()	Composable wait conditions
Vision	device.vision	Template matching, color detection
OCR	device.ocr	Text recognition via EasyOCR
Watchers	device.watchers	Auto-dismiss dialogs
Recorder	device.recorder	Record and replay actions
Parallel	DevicePool	Multi-device execution
Flow	Flow	Multi-step workflow engine

Examples

UI Automation

from adbflow.ui import Selector

# Find and tap
button = await device.ui.find_async(
    Selector().resource_id("com.example:id/submit").clickable()
)
await button.tap_async()

# Wait for element
element = await device.ui.wait_for_async(
    Selector().text("Welcome"), timeout=10.0
)

# Type text
field = await device.ui.find_async(
    Selector().resource_id("com.example:id/search")
)
await field.tap_async()
await field.text_input_async("hello world")

Wait Conditions

from adbflow.wait import wait_for, TextVisible, any_of, not_

# Wait for success or error
await wait_for(
    any_of(TextVisible(device, "Success"), TextVisible(device, "Error")),
    timeout=10.0
)

# Wait for loading to disappear
await wait_for(not_(TextVisible(device, "Loading...")), timeout=15.0)

Multi-Device

from adbflow.parallel import DevicePool

pool = DevicePool(devices)

async def screenshot_all(device):
    model = await device.info.model_async()
    await device.media.screenshot.capture_to_file_async(f"{model}.png")

results = await pool.run_async(screenshot_all)

Flow Engine

from adbflow.flow import Flow
from adbflow.utils.types import ErrorStrategy

flow = Flow(device)
flow.add_step("open", open_app)
flow.add_step("login", login, retries=2)
flow.add_step("verify", verify, on_error=ErrorStrategy.SKIP)
result = await flow.run_async()

CLI

adbflow devices              # List connected devices
adbflow shell "ls /sdcard"   # Run shell command
adbflow screenshot out.png   # Take screenshot
adbflow install app.apk      # Install APK
adbflow info                 # Device info
adbflow logcat               # Stream logs
adbflow record -o rec.json   # Record actions
adbflow play rec.json        # Replay actions

Documentation

Full documentation is available at shahadh7.github.io/adbflow.

• Installation
• Quickstart
• UI Automation Guide
• API Reference

Contributing

git clone https://github.com/Shahadh7/adbflow.git
cd adbflow
python -m venv .venv && source .venv/bin/activate
pip install -e ".[all,dev]"

pytest                    # Run tests
ruff check src/           # Lint
mypy src/adbflow/         # Type check

See the Contributing guide for details.

License

MIT — see LICENSE for details.