ddp_utils.browser.observers.page¶

import ddp_utils.browser.observers.page

Observe page and selector changes through the unified browser transport.

PageObserver compares a page or selector fingerprint around a bounded operation. GlobalObserver adds callbacks, URL events, and multi-selector monitoring without introducing a second browser abstraction.

Examples

Observe one results container:

from ddp_utils.browser.observers import PageObserver

observer = PageObserver(browser, selector="div#results")
with observer:
    browser.find("button", text="Search").click()
changed = observer.wait_for_change(timeout=15)

from ddp_utils.browser.observers import GlobalObserver

observer = GlobalObserver(browser)
observer.on_change("div#results", handle_results)
observer.on_url_change(handle_navigation)
observer.start()
observer.stop()

class ddp_utils.browser.observers.page.PageObserver(driver: Any, *, selector: str | None = None, check_interval: float = 0.3, ready_timeout: float = 5.0)¶

Bases: object

Detect a document or CSS-scoped text change by polling fingerprints.

Parameters

Name	Type	Description
driver	Any	Unified browser facade or Selenium-compatible driver.
selector	Optional[str]	CSS selector to fingerprint, or `None` for a lightweight document fingerprint.
check_interval	float	Seconds between fingerprint polls.
ready_timeout	float	Default seconds allowed for document readiness.

Examples

Observe a results container around an action:

with PageObserver(browser, selector="#results") as observer:
    browser.find("button", text="Search").click()
changed = observer.wait_for_change(timeout=15)

Initialize a page fingerprint observer.

Parameters

Name	Type	Description
driver	Any	Selenium or SeleniumBase WebDriver.
selector	Optional[str]	CSS scope to observe, or `None` for the document.
check_interval	float	Poll interval in seconds; must be positive.
ready_timeout	float	Maximum ready-state wait; cannot be negative.

Raises

Exception	Description
ValueError	If an interval or timeout is invalid.

Examples

Configure a fast selector observer:

observer = PageObserver(
    browser,
    selector="#results",
    check_interval=0.1,
    ready_timeout=3.0,
)

start() → PageObserver¶

Capture the current page fingerprint as the observation baseline.

Returns

Type	Description
PageObserver	This active observer for fluent use.

Examples

Start and retain the observer in one expression:

observer = PageObserver(browser).start()

stop() → None¶

Disable further change checks without discarding captured state.

Examples

Stop polling after a bounded workflow:

observer.stop()

reset() → None¶

Replace the observation baseline with the current page fingerprint.

Examples

Accept the current content as the new baseline:

observer.reset()

wait_for_change(timeout: float = 30.0, *, wait_for_ready: bool = True, ready_timeout: float | None = None) → bool¶

Poll until observed content changes or the deadline expires.

Parameters

Name	Type	Description
timeout	float	Maximum change wait in seconds; cannot be negative.
wait_for_ready	bool	Wait for `document.readyState == 'complete'`.
ready_timeout	float \| None	Ready-state timeout, defaulting to the constructor value.

Returns

Type	Description
bool	`True` once a change is detected, independently of ready-state success.

Raises

Exception	Description
ValueError	If `timeout` or `ready_timeout` is negative.

Examples

Wait for a change and subsequent document readiness:

changed = observer.wait_for_change(
    timeout=20,
    wait_for_ready=True,
    ready_timeout=5,
)

wait_for_ready(timeout: float | None = None) → bool¶

Wait for the document ready state to become complete.

Parameters

Name	Type	Description
timeout	float \| None	Maximum wait in seconds, or the configured default.

Returns

Type	Description
bool	`True` when the document is ready.

Raises

Exception	Description
ValueError	If the resolved timeout is negative.

Examples

Bound the readiness wait independently:

ready = observer.wait_for_ready(timeout=2)

has_changed() → bool¶

Check once whether observed content differs from the baseline.

Returns

Type	Description
bool	True after the first detected change; false while inactive, unchanged, unavailable, or unreadable.

Examples

Perform a non-blocking check after an action:

if observer.has_changed():
    process_results()

is_ready() → bool¶

Check the current document readiness without waiting.

Returns

Type	Description
bool	True only when JavaScript reports `document.readyState` as `complete`; transport errors produce false.

Examples

Skip work until the current document is complete:

if observer.is_ready():
    inspect_page()

is_active() → bool¶

Return whether this observer is currently watching for page changes.

Returns

Type	Description
bool	True between `start()` and `stop()`.

Examples

Guard an optional wait:

if observer.is_active():
    observer.wait_for_change()

get_last_change_time() → float | None¶

Return the timestamp of the most recently detected change, if any.

Returns

Type	Description
float \| None	Unix timestamp recorded at first detection, or `None` before a change has been detected.

Examples

Include detection time in a diagnostic record:

changed_at = observer.get_last_change_time()

class ddp_utils.browser.observers.page.ChangeInfo(selector: str | None, old_fp: str | None, new_fp: str | None, url: str = '')¶

Bases: object

Record the before/after fingerprints for one selector change.

Parameters

Name	Type	Description
selector	Optional[str]	Observed CSS selector, or `None` for document scope.
old_fp	Optional[str]	Previous content fingerprint.
new_fp	Optional[str]	Replacement content fingerprint.
url	str	URL active when the change was detected.

Examples

Describe a selector transition:

info = ChangeInfo("#results", old_fp, new_fp, browser.url)

Initialize a fixed-shape change record with detection time.

Parameters

Name	Type	Description
selector	Optional[str]	Observed CSS selector, or `None` for document scope.
old_fp	Optional[str]	Fingerprint captured before the change.
new_fp	Optional[str]	Fingerprint captured after the change.
url	str	URL active at detection time.

Examples

Build a record delivered to observer callbacks:

info = ChangeInfo("#status", "old", "new", "https://example.test")

class ddp_utils.browser.observers.page.GlobalObserver(driver: Any, *, poll_interval: float = 0.5, ready_timeout: float = 5.0)¶

Bases: object

Poll multiple CSS selectors and the current URL on one worker thread.

Parameters

Name	Type	Description
driver	Any	Unified browser facade or Selenium-compatible driver.
poll_interval	float	Seconds between background polling passes.
ready_timeout	float	Default readiness wait after a detected change.

Examples

Watch a selector and navigation for the lifetime of a block:

with GlobalObserver(browser) as observer:
    observer.on_change("#results", handle_results)
    observer.on_url_change(handle_navigation)

Initialize callback registries and background-worker state.

Parameters

Name	Type	Description
driver	Any	Unified browser facade or Selenium-compatible driver.
poll_interval	float	Seconds between background polling passes.
ready_timeout	float	Seconds allowed for readiness waits.

Examples

Configure a quarter-second observer poll:

observer = GlobalObserver(
    browser,
    poll_interval=0.25,
    ready_timeout=4.0,
)

on_change(selector: str, callback: Callable[[ChangeInfo], None], *, once: bool = False) → GlobalObserver¶

Parameters

Name	Type	Description
selector	str	CSS selector whose normalized text is fingerprinted.
callback	Callable[[ChangeInfo], None]	Function receiving the resulting `ChangeInfo`.
once	bool	Remove this callback after its first invocation.

Returns

Type	Description
GlobalObserver	This observer for fluent registration.

Examples

observer.on_change("#results", handle_results, once=True)

on_any_change(callback: Callable[[ChangeInfo], None]) → GlobalObserver¶

Parameters

Name	Type	Description
callback	Callable[[ChangeInfo], None]	Function receiving every selector `ChangeInfo`.

Returns

Type	Description
GlobalObserver	This observer for fluent registration.

Examples

Collect all selector changes in one audit stream:

observer.on_any_change(audit_change)

on_url_change(callback: Callable[[str], None]) → GlobalObserver¶

Parameters

Name	Type	Description
callback	Callable[[str], None]	Function receiving each newly observed URL.

Returns

Type	Description
GlobalObserver	This observer for fluent registration.

Examples

Track client-side navigation:

observer.on_url_change(record_url)

on_error(callback: Callable[[Exception], None]) → GlobalObserver¶

Parameters

Name	Type	Description
callback	Callable[[Exception], None]	Function receiving uncaught polling-loop exceptions.

Returns

Type	Description
GlobalObserver	This observer for fluent registration.

Examples

Route observer failures into project diagnostics:

observer.on_error(report_observer_error)

watch(selector: str) → GlobalObserver¶

Add a CSS selector to the monitored set.

Parameters

Name	Type	Description
selector	str	CSS selector to fingerprint without a custom callback.

Returns

Type	Description
GlobalObserver	This observer for fluent configuration.

Examples

observer.watch("#results")

start() → GlobalObserver¶

Snapshot current state and start the background poller idempotently.

Returns

Type	Description
GlobalObserver	This running observer.

Examples

Start after all callbacks are registered:

observer.on_change("#results", handler).start()

stop(timeout: float = 3.0) → None¶

Stop the background observer and wait for its worker.

Parameters

Name	Type	Description
timeout	float	Maximum seconds to wait for the worker thread to exit.

Examples

Stop cleanly before closing the browser:

observer.stop(timeout=2.0)

wait_for_selector_change(selector: str, timeout: float = 30.0, *, wait_for_ready: bool = True) → ChangeInfo | None¶

Wait for a selector event and return its latest change record.

Parameters

Name	Type	Description
selector	str	CSS selector to watch, registered automatically if needed.
timeout	float	Maximum seconds to wait for its event.
wait_for_ready	bool	Wait for document readiness after the event.

Returns

Type	Description
ChangeInfo \| None	Latest `ChangeInfo`, or `None` when the wait expires.

Examples

Wait for asynchronous result replacement:

info = observer.wait_for_selector_change("#results", timeout=20)

wait_for_url_change(timeout: float = 30.0) → str | None¶

Wait for navigation and return the new URL.

Parameters

Name	Type	Description
timeout	float	Maximum seconds to wait for navigation.

Returns

Type	Description
str \| None	Newly observed URL, or `None` when the wait expires.

Examples

Wait for client-side navigation:

new_url = observer.wait_for_url_change(timeout=10)

wait_for_ready(timeout: float | None = None) → bool¶

Block until the page is considered ready, or the timeout elapses.

Parameters

Name	Type	Description
timeout	float \| None	Maximum wait in seconds, or the configured default.

Returns

Type	Description
bool	True when readiness becomes complete; false at the deadline.

Examples

Wait after a navigation callback:

ready = observer.wait_for_ready(timeout=5)

get_last_change(selector: str) → ChangeInfo | None¶

Return the last recorded change info for a given selector, if any.

Parameters

Name	Type	Description
selector	str	Registered CSS selector.

Returns

Type	Description
ChangeInfo \| None	Latest record for the selector, or `None` before any change.

Examples

Inspect the most recently detected result change:

info = observer.get_last_change("#results")

reset(selector: str | None = None) → None¶

Reset one selector or every observation baseline.

Parameters

Name	Type	Description
selector	str \| None	CSS selector to reset, or `None` for every registered selector.

Examples

Accept every current selector state as the new baseline:

observer.reset()