observe_mesh_data

Part of Reusable workflows in the Python API reference.

Related guidance: shipped-tooling.

Runnable examples: observe_mesh_data.py.

from picogrid_ecn_client.workflows import observe_mesh_data

Observe routed entity and location families until either consumer completes.

async def observe_mesh_data(
    client: ECNClient,
    *,
    integrations: Collection[str],
    limit: int = 0,
    on_event: Callable[[EntityEvent | LocationEvent], None] | None = None,
) -> MeshObservationResult

Parameters

Name	Type	Default	Description
client	ECNClient	—	Configured SDK client used to open both public streams.
integrations	Collection[str]	—	Exact integration names allowed on both streams.
limit	int	0	Maximum events per consumer; zero watches until interrupted while retaining only the most recent configured watcher buffer per consumer.
on_event	Callable[[EntityEvent | LocationEvent], None] | None	None	Optional synchronous callback invoked for every observed event.

Returns

Type	Description
MeshObservationResult	All events from positive-limit consumers, or the most recently retained events from unbounded consumers, before either consumer completed.

Raises

Exception	Description
ValidationError	If limit is negative, integrations is empty, or an integration name is empty.
ECNClientError	If either stream cannot be opened or consumed.

Source · workflows/observe.py