bluesky_tiled_plugins.streaming

Module Contents

Classes

BlueskyStreamUpdate	A selected Tiled update from a Bluesky data stream.
BlueskyStreamSubscription	Own recursive Tiled subscriptions for selected Bluesky event streams.

Functions

subscribe_to_streams

Subscribe to selected data keys in matching live Bluesky runs.

Data

logger

API

bluesky_tiled_plugins.streaming.logger

‘getLogger(…)’

class bluesky_tiled_plugins.streaming.BlueskyStreamUpdate

A selected Tiled update from a Bluesky data stream.

Parameters

run_uid : str Key of the direct BlueskyRun parent. stream_name : str Name of the Bluesky data stream containing the updated node. data_keys : tuple of str Requested data keys represented by this update. A table update may contain several selected scalar keys; an array update contains one. update : LiveArrayData, LiveArrayRef, or LiveTableData Original Tiled live update. It is retained without decoding until :meth:data is called. sequence : int Native positive per-node Tiled streaming sequence number for update. Use it to correlate updates and detect duplicates or gaps for one node.

run_uid: str

None

stream_name: str

None

data_keys: tuple[str, ...]

None

update: tiled.client.stream.LiveArrayData | tiled.client.stream.LiveArrayRef | tiled.client.stream.LiveTableData

None

sequence: int

0

data() → Any

Decode the original Tiled live update.

Returns

Any Tiled’s decoded array or table representation. When Tiled’s standard table decoder returns a pandas DataFrame, only :attr:data_keys are included. Other Tiled-decoded table representations are returned unchanged.

Notes

Decoding happens only when this method is called. Array references retain Tiled’s normal fetch behavior.

class bluesky_tiled_plugins.streaming.BlueskyStreamSubscription(container: tiled.client.container.Container, streams: collections.abc.Mapping[str, tuple[str, ...] | None] | None, callback: collections.abc.Callable[[bluesky_tiled_plugins.streaming.BlueskyStreamUpdate], None], *, start: int | None, max_size: int, run_filter: collections.abc.Callable[[tiled.client.stream.LiveChildCreated], bool] | None)

Own recursive Tiled subscriptions for selected Bluesky event streams.

property closed: bool

Whether teardown has begun.

Returns

bool True after :meth:disconnect marks this manager closed. Tiled subscriptions may still be completing their blocking teardown.

disconnect() → None

Disconnect every owned Tiled subscription.

Owned data-node subscriptions are disconnected before stream, run, and container subscriptions. This method is idempotent and blocks while Tiled closes sockets and waits for its subscription threads.

Raises

Exception The first error raised while disconnecting an owned Tiled subscription, after teardown is attempted for every owned subscription.

bluesky_tiled_plugins.streaming.subscribe_to_streams(container: tiled.client.container.Container, callback: collections.abc.Callable[[bluesky_tiled_plugins.streaming.BlueskyStreamUpdate], None], *, streams: collections.abc.Mapping[str, str | collections.abc.Iterable[str] | None] | None, metadata_filter: collections.abc.Callable[[collections.abc.Mapping[str, Any]], bool] | None = None, required_specs: str | collections.abc.Iterable[str] | None = None, run_filter: collections.abc.Callable[[tiled.client.stream.LiveChildCreated], bool] | None = None, start: int | None = 0, max_size: int = 1000000) → bluesky_tiled_plugins.streaming.BlueskyStreamSubscription

Subscribe to selected data keys in matching live Bluesky runs.

Parameters

container : tiled.client.container.Container Direct parent of BlueskyRun nodes created by :class:~bluesky_tiled_plugins.TiledWriter. callback : Callable[[BlueskyStreamUpdate], None] Function called for each selected native Tiled live update. streams : Mapping[str, str or iterable of str or None] or None Mapping from Bluesky event-stream name to its data-key selection. A mapping value of None selects every streamable array and table key in that stream. An outer None selects every event stream and all its streamable array and table keys. Ragged and bytes nodes are ignored. metadata_filter : Callable[[Mapping[str, Any]], bool] or None, optional Predicate applied to the raw run-creation update’s persisted metadata["start"] mapping. required_specs : str, iterable of str, or None, optional One spec name or every spec name required for a run to match. run_filter : Callable[[LiveChildCreated], bool] or None, optional Predicate applied to the raw Tiled run-creation update. start : int or None, optional Tiled sequence number supplied to every owned subscription. The default, 0, replays records retained by the streaming cache. None receives only new records. max_size : int, optional Maximum incoming WebSocket message size in bytes. Defaults to 1_000_000.

Returns

BlueskyStreamSubscription A running managed subscription. Use :meth:~BlueskyStreamSubscription.disconnect or a context manager to release it.

Raises

ValueError If streams is an empty mapping or any data-key selection is an explicit empty iterable. Exception Any error raised while Tiled establishes the root subscription.

Notes

The optional filters are evaluated in metadata_filter, required_specs, then run_filter order with short-circuiting AND semantics.

The root container subscription is connected before this function returns. Updates from different data nodes or event streams have no global ordering, and Tiled may invoke the callback concurrently. Co-located selected table columns remain one native table update; this function does not join or align updates across nodes or streams.

With Tiled 0.2.18, the server can stream ragged nodes, but RaggedClient has no subscription API and the client has no ragged live schema or update model. BytesClient likewise has no subscription API, and the server has no bytes live schema or cache emitter and rejects bytes on its single-node streaming route. Both families are ignored here.

Examples

def on_update(update): … print(update.stream_name, update.data()) … subscription = subscribe_to_streams( … container, … on_update, … streams={“baseline”: (“x”, “y”), “primary”: None}, … ) subscription.disconnect()