predicate-runtime-python/sentience/wait.py at 2482735a5a25d537fa2c9a67680b5f1b300a500f · PredicateSystems/predicate-runtime-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
"""
Wait functionality - wait_for element matching selector
"""

import time

from .browser import SentienceBrowser
from .models import WaitResult, SnapshotOptions
from .query import find
from .snapshot import snapshot


def wait_for(
    browser: SentienceBrowser,
    selector: str | dict,
    timeout: float = 10.0,
    interval: float | None = None,
    use_api: bool | None = None,
) -> WaitResult:
    """
    Wait for element matching selector to appear

    Args:
        browser: SentienceBrowser instance
        selector: String DSL or dict query
        timeout: Maximum time to wait (seconds)
        interval: Polling interval (seconds). If None, auto-detects:
                  - 0.25s for local extension (use_api=False, fast)
                  - 1.5s for remote API (use_api=True or default, network latency)
        use_api: Force use of server-side API if True, local extension if False.
                 If None, uses API if api_key is set, otherwise uses local extension.

    Returns:
        WaitResult
    """
    # Auto-detect optimal interval based on API usage
    if interval is None:
        # Determine if using API
        will_use_api = use_api if use_api is not None else (browser.api_key is not None)
        if will_use_api:
            interval = 1.5  # Longer interval for API calls (network latency)
        else:
            interval = 0.25  # Shorter interval for local extension (fast)

    start_time = time.time()

    while time.time() - start_time < timeout:
        # Take snapshot (may be local extension or remote API)
        snap = snapshot(browser, SnapshotOptions(use_api=use_api))

        # Try to find element
        element = find(snap, selector)

        if element:
            duration_ms = int((time.time() - start_time) * 1000)
            return WaitResult(
                found=True,
                element=element,
                duration_ms=duration_ms,
                timeout=False,
            )

        # Wait before next poll
        time.sleep(interval)

    # Timeout
    duration_ms = int((time.time() - start_time) * 1000)
    return WaitResult(
        found=False,
        element=None,
        duration_ms=duration_ms,
        timeout=True,
    )