watch_detections

Part of Reusable workflows in the Python API reference.

Related guidance: shipped-tooling.

Runnable examples: watch_detections.py.

from picogrid_ecn_client.workflows import watch_detections

Collect detection events, with zero meaning an intentionally unbounded watch.

async def watch_detections(
    client: ECNClient,
    *,
    integration: str | None = None,
    limit: int = 0,
    on_event: Callable[[EntityEvent], None] | None = None,
) -> WatchEntitiesResult

Parameters

Name	Type	Default	Description
client	ECNClient	—	Configured SDK client used to open the entity stream.
integration	str | None	None	Optional exact integration name to observe.
limit	int	0	Maximum events to collect; zero watches until interrupted while retaining only the most recent configured watcher buffer.
on_event	Callable[[EntityEvent], None] | None	None	Optional synchronous callback invoked for every observed event.

Returns

Type	Description
WatchEntitiesResult	All events from a positive-limit watch, or the most recently retained events from an unbounded watch.

Raises

Exception	Description
ValidationError	If limit is negative or integration is empty.
ECNClientError	If the stream cannot be opened or consumed.

Source · workflows/observe.py