Watchers and Automation

class Device:
    @cached_property
    def watcher(self) -> Watcher:
        """Access device watcher for automated UI responses"""

class Watcher:
    def when(self, selector: Dict[str, Any]) -> WatcherItem:
        """
        Create watcher for UI element.

        Parameters:
        - selector: UI element selector criteria

        Returns:
        WatcherItem for configuring response
        """

    def remove(self, name: str = None):
        """
        Remove watcher by name.

        Parameters:
        - name: Watcher name to remove, None removes all
        """

    def list(self) -> List[str]:
        """List all active watcher names"""

    def run(self):
        """Run watchers once to check for triggers"""

    def start(self, interval: float = 2.0):
        """
        Start background watcher monitoring.

        Parameters:
        - interval: Check interval in seconds
        """

    def stop(self):
        """Stop background watcher monitoring"""

d = u2.connect()

# Create watchers for common dialogs
d.watcher.when(text="OK").click()
d.watcher.when(text="Allow").click()
d.watcher.when(text="Cancel").click()

# Handle permission dialogs
d.watcher.when(textContains="permission").click(text="Allow")
d.watcher.when(textContains="Location").click(text="Grant")

# Start background monitoring
d.watcher.start(interval=1.0)  # Check every second

# Your automation code here
d.app_start("com.example.app")
# Watchers automatically handle popups

# Stop monitoring when done
d.watcher.stop()

# Remove specific watchers
d.watcher.remove("OK_watcher")
d.watcher.remove()  # Remove all

class Device:
    def watch_context(self, autostart: bool = True, builtin: bool = False) -> WatchContext:
        """
        Create watch context for advanced automation.

        Parameters:
        - autostart: Start monitoring automatically
        - builtin: Use built-in system watchers

        Returns:
        WatchContext for configuring watchers
        """

class WatchContext:
    def __init__(self, d: Device, builtin: bool = False):
        """Initialize watch context"""

    def when(self, xpath: str) -> WatchContextItem:
        """
        Create XPath-based watcher.

        Parameters:
        - xpath: XPath expression to watch for

        Returns:
        WatchContextItem for configuring response
        """

    def start(self):
        """Start watch context monitoring"""

    def stop(self):
        """Stop watch context monitoring"""

    def wait_stable(self, timeout: float = 10.0) -> bool:
        """
        Wait for UI to become stable (no watcher triggers).

        Parameters:
        - timeout: Maximum wait time

        Returns:
        bool: True if UI became stable
        """

d = u2.connect()

# Create watch context
with d.watch_context() as ctx:
    # XPath-based watchers
    ctx.when('//*[@text="OK"]').click()
    ctx.when('//*[@text="Skip"]').click()
    ctx.when('//android.widget.Button[contains(@text, "Allow")]').click()
    
    # Handle ads and interruptions
    ctx.when('//*[@resource-id="close_ad"]').click()
    ctx.when('//*[contains(@text, "Ad")]/..//*[@text="✕"]').click()
    
    # Run automation
    d.app_start("com.example.app")
    
    # Wait for UI to stabilize
    ctx.wait_stable(timeout=30)
    
    # Continue with automation
    d.click(100, 200)

# Context automatically stops when exiting

class WatcherItem:
    def click(self, **kwargs) -> WatcherItem:
        """Click matching element with optional selector refinement"""

    def press(self, key: str) -> WatcherItem:
        """Press hardware key when element appears"""

    def call(self, func) -> WatcherItem:
        """Call custom function when element appears"""

class WatchContextItem:
    def click(self) -> WatchContextItem:
        """Click the matched XPath element"""

    def press(self, key: str) -> WatchContextItem:
        """Press hardware key when XPath matches"""

    def call(self, func) -> WatchContextItem:
        """Call custom function when XPath matches"""

d = u2.connect()

# Custom watcher functions
def handle_popup():
    print("Popup appeared, handling...")
    d.press("back")
    d.press("back")

def skip_ad():
    print("Ad detected, skipping...")
    d.click(0.9, 0.1)  # Top-right corner

# Configure watcher responses
d.watcher.when(text="Update").press("back")
d.watcher.when(textContains="popup").call(handle_popup)
d.watcher.when(resourceId="ad_container").call(skip_ad)

# XPath watcher with custom function
with d.watch_context() as ctx:
    ctx.when('//*[@text="Rate App"]').call(lambda: d.press("back"))
    ctx.when('//android.widget.VideoView').call(skip_ad)
    
    # Run automation with watchers active
    d.app_start("com.social.app")
    ctx.wait_stable()

d = u2.connect()

# Enable built-in system watchers
with d.watch_context(builtin=True) as ctx:
    # Built-in watchers handle:
    # - ANR (Application Not Responding) dialogs
    # - Permission dialogs
    # - System update notifications
    # - Crash dialogs
    
    # Your automation code
    d.app_start("com.potentially.unstable.app")
    
    # Built-in watchers automatically handle system interruptions
    ctx.wait_stable(timeout=60)

d = u2.connect()

# List active watchers
watchers = d.watcher.list()
print(f"Active watchers: {watchers}")

# Manual watcher execution
d.watcher.run()  # Check all watchers once

# Monitor watcher triggers
import logging
u2.enable_pretty_logging(logging.DEBUG)

# Create watchers with names for tracking
d.watcher.when(text="OK").click().name("ok_handler")
d.watcher.when(text="Cancel").press("back").name("cancel_handler")

# Start monitoring with logging
d.watcher.start(interval=0.5)

# Run automation - watcher activity will be logged
d.app_start("com.example.app")
time.sleep(10)

d.watcher.stop()

d = u2.connect()

def robust_app_automation(package_name):
    """Robust app automation with comprehensive error handling"""
    
    with d.watch_context(builtin=True) as ctx:
        # Handle common interruptions
        ctx.when('//*[@text="OK"]').click()
        ctx.when('//*[@text="Allow"]').click()
        ctx.when('//*[@text="Skip"]').click()
        ctx.when('//*[@text="Not now"]').click()
        ctx.when('//*[contains(@text, "Update")]').click(text="Later")
        
        # Handle ads
        ctx.when('//*[@resource-id="close_button"]').click()
        ctx.when('//*[@text="✕"]').click()
        
        # Launch app with session monitoring
        with d.session(package_name) as session:
            ctx.wait_stable(timeout=30)
            
            # App-specific automation
            yield session  # Allow caller to perform actions
            
            # Verify app is still running
            if not session.running():
                print("App crashed, restarting...")
                session.restart()
                ctx.wait_stable(timeout=15)

# Usage
for session in robust_app_automation("com.example.app"):
    # Perform automation with robust error handling
    session.click(100, 200)
    session.send_keys("test input")
    
    # UI interactions protected by watchers
    session(text="Submit").click()
    
    # Wait and continue
    time.sleep(5)

d = u2.connect()

# Monitor toasts alongside watchers
with d.watch_context() as ctx:
    # Handle error dialogs
    ctx.when('//*[@text="Error"]').click(text="OK")
    ctx.when('//*[@text="Failed"]').press("back")
    
    # Start app
    d.app_start("com.example.app")
    
    # Monitor for success/error messages
    ctx.wait_stable()
    
    # Check toast messages
    toast = d.last_toast
    if toast and "error" in toast.lower():
        print(f"Error toast detected: {toast}")
        d.press("back")
    elif toast and "success" in toast.lower():
        print(f"Success toast: {toast}")
    
    d.clear_toast()