Subscribe to Live Bluesky Stream Data

subscribe_to_streams delivers native Tiled live updates for selected data keys across one or more named Bluesky event streams. The callback receives a BlueskyStreamUpdate; call update.data() to decode its payload.

from bluesky_tiled_plugins import BlueskyStreamUpdate, subscribe_to_streams


def on_update(update: BlueskyStreamUpdate) -> None:
    batch = update.data()
    print(update.run_uid, update.stream_name, update.data_keys, update.sequence)
    print(batch)


with subscribe_to_streams(
    container,
    on_update,
    streams={"baseline": ("x", "y"), "primary": "temperature"},
):
    # Continue producing or consuming data while the subscription is active.
    ...

container is the direct parent Container of the BlueskyRun nodes. Leaving the with block disconnects every subscription created for that container, runs, event streams, and selected data nodes. Call disconnect() directly when a context manager does not fit the application’s lifecycle.

Configure the producer and server

Use clients that address the same parent Container for subscribe_to_streams and TiledWriter.

from bluesky_tiled_plugins import TiledWriter

writer = TiledWriter(container, batch_size=1)

batch_size=1 makes each Event document available for live delivery. The Tiled server must have Streaming cache configuration; follow the deployment guide to enable it. The default start=0 replays only records retained by that cache, not the full historical container.

After a successful Stop, TiledWriter closes every writer-owned data node, event-stream container, and run container from leaves to root. Tiled delivers each node’s pending updates before its normal stream closure, and subscribe_to_streams releases that completed branch while keeping the root container subscription live for later runs. Producers other than TiledWriter must close their own Tiled streams to receive the same automatic cleanup.

Select streams and data keys

streams is required. Pass None to select every future event stream and every streamable array and table key in accepted runs. This is deliberately explicit because it may include high-volume data.

with subscribe_to_streams(container, on_update, streams=None):
    ...

For a narrower subscription, pass a mapping from each event-stream name to one key, an iterable of keys, or None. Omit a stream to ignore it; a mapping value of None selects every streamable key in that named stream.

with subscribe_to_streams(
    container,
    on_update,
    streams={"baseline": None, "primary": ("temperature", "pressure")},
):
    ...

Ragged and bytes nodes are not delivered. Supporting either family requires upstream Tiled development.

Run metadata and sequence

Updates carry run_uid, not a Start-document field. Retrieve a deep, mutable, callback-owned copy of persisted metadata from the direct-parent container:

run_metadata = container[update.run_uid].metadata_copy()[0]
start_document = run_metadata.get("start", {})

update.sequence is Tiled’s native positive streaming sequence number for the node. Use it to correlate delivery and detect duplicates or gaps for one node. Selected arrays and table columns can be different nodes, so their sequences are independent.

Native update batches

For selected scalar columns stored together in one Tiled table, update.data() returns one decoded native table batch with just those columns.

Selected arrays and columns stored in separate tables arrive as independent callbacks. The API does not synthesize array updates from table columns, buffer or join updates across nodes, or invent row-alignment and missing-value behavior.

Updates from different data nodes or event streams have no global ordering, and Tiled may invoke the shared callback concurrently. Synchronize any mutable state owned by that callback.

Filter runs

subscribe_to_streams evaluates three optional filters before opening any run, event-stream, or data-node WebSocket. metadata_filter receives the raw run-creation update’s Start document, required_specs requires every named spec, and run_filter receives the complete raw LiveChildCreated update.

from tiled.client.stream import LiveChildCreated


def accepts_run(update: LiveChildCreated) -> bool:
    return update.key not in ignored_run_uids


subscription = subscribe_to_streams(
    container,
    on_update,
    streams={"baseline": "x", "primary": None},
    metadata_filter=lambda start: start.get("proposal") == "calibration",
    required_specs=("XAS_Calib", "Calibration"),
    run_filter=accepts_run,
)

The filters use short-circuiting AND semantics in metadata_filter, required_specs, then run_filter order. None means no condition. A false result skips that run entirely.

Readiness and lifetime

The root container WebSocket is connected before subscribe_to_streams returns. Queue Server code can establish the subscription before it submits its first plan; future run descendants cannot exist until TiledWriter creates their nodes.

When a producer closes a Tiled stream, the manager releases that completed descendant after its callbacks drain. Closure means no more WebSocket updates; it does not guarantee that a concurrent REST read can immediately observe every write.