Watchers

Watchers run in the background, polling the UI hierarchy and automatically responding to conditions — like auto-dismissing dialogs or clicking through popups.

Basic Usage

from adbflow.ui import Selector
from adbflow.watchers import click_watcher, dismiss_watcher

async def watcher_example(device):
    wm = device.watchers

    # Register a watcher that clicks "OK" whenever it appears
    name, selector, action = click_watcher("ok_dismiss", Selector().text("OK"))
    wm.register(name, selector, action)

    # Convenience: auto-dismiss dialogs containing specific text
    name, selector, action = dismiss_watcher("crash_dismiss", "has stopped")
    wm.register(name, selector, action)

    # Start the watcher loop (polls every 1 second)
    await wm.start_async(interval=1.0)

    # ... run your test/automation ...

    # Stop watchers when done
    await wm.stop_async()

Managing Watchers

async def manage_watchers(device):
    wm = device.watchers

    # Register
    name, selector, action = click_watcher("allow_perms", Selector().text("Allow"))
    wm.register(name, selector, action)

    # List registered watchers
    names = wm.list_watchers()
    print(names)  # ["allow_perms"]

    # Check if running
    print(wm.is_running)  # False

    # Start
    await wm.start_async()
    print(wm.is_running)  # True

    # Unregister a specific watcher
    wm.unregister("allow_perms")

    # Stop all
    await wm.stop_async()

Custom Watcher Actions

For more control, register a custom WatcherAction:

from adbflow.watchers.manager import WatcherAction

async def custom_watcher(device):
    wm = device.watchers

    # Define a custom action
    async def log_and_dismiss(d, element):
        text = element.get_text()
        print(f"Watcher triggered: {text}")
        await element.tap_async()

    action = WatcherAction(callback=log_and_dismiss, device=device)
    wm.register("custom", Selector().text_contains("Error"), action)

    await wm.start_async()

Common Patterns

Auto-dismiss permission dialogs

async def auto_permissions(device):
    wm = device.watchers
    name, sel, act = click_watcher("allow", Selector().text("Allow"))
    wm.register(name, sel, act)

    name, sel, act = click_watcher("while_using", Selector().text("While using the app"))
    wm.register(name, sel, act)

    await wm.start_async()

Dismiss crash dialogs

async def auto_crash_dismiss(device):
    wm = device.watchers
    name, sel, act = dismiss_watcher("crash", "has stopped")
    wm.register(name, sel, act)

    name, sel, act = dismiss_watcher("anr", "isn't responding")
    wm.register(name, sel, act)

    await wm.start_async()

Tips

• Keep the polling interval reasonable (1–2 seconds). Too fast wastes resources; too slow may miss transient dialogs.
• Always call stop_async() when done to clean up the background task.
• Watchers are checked in registration order. Register more specific watchers first.
• The watcher loop dumps the UI hierarchy each interval, so it shares the same cache as device.ui.