Accessing Saved Data

Bluesky itself does not manage how you access persisted collected data, but this is a common question so we cover the standard workflow here. This section assumes you already have Databroker configured, as covered earlier in this tutorial. You are not required to use Databroker with Bluesky; it is simply a convenient tool for capturing both metadata and data generated by the RunEngine.

To retrieve saved data, you reference the dataset (called a "run") by its unique ID, which is returned by RunEngine when the data colection completes.

from bluesky.plans import count
from ophyd.sim import example_detector
from databroker import Broker

# Connect to the test Databroker instance
data_db = Broker.named('temp')

# Subscribe to capture all captured data into Databroker
RE.subscribe(data_db.insert)

# Execute a count plan and capture the returned unique run ID
run_uid, = RE(count([example_detector], num_points=3))

# Fetch run metadata by UID
run_header = data_db[run_uid]
# A more convenient option: fetch the most recently completed run
run_header = data_db[-1]  # Refers to 1 run prior, i.e. the latest run

This example assumes your plan generates exactly one run, which is typical for simple plans like count(). In general, a single plan can produce multiple runs, so it will return multiple UIDs. Querying Databroker with this list of UIDs will return a list of headers instead of a single entry.

all_run_uids = RE(your_custom_plan(...))
all_run_headers = data_db[all_run_uids]  # Returns a list of Header objects

Most primary experimental data is stored in the "primary" stream, which is accessible directly from the header object. For more details on working with retrieved data, refer to the official Databroker documentation.

Simple Custom Workflows

Create a Reusable Plan Variant with partial

If you consistently use the same set of detectors and want to avoid repeating the detector list in every count() call, you can create a custom variant of the built-in count plan using Python's built-in functools.partial().

from bluesky import RunEngine
from functools import partial
from bluesky.plans import count
from ophyd.sim import example_detector
from bluesky.callbacks.best_effort import BestEffortCallback
from databroker import Broker

RE = RunEngine({})
bec = BestEffortCallback()
data_db = Broker.named('temp')

RE.subscribe(bec)
RE.subscribe(data_db.insert)

# Pre-bind the fixed detector list to the count plan
my_count = partial(count, [example_detector])
# This is equivalent to RE(count([example_detector]))
RE(my_count())

# Any extra arguments passed to my_count are forwarded to the original count
RE(my_count(num=3, delay=1))

Chaining Multiple Plans

You can build a custom plan by chaining existing plans together using Python's yield from syntax. Below is a practical example:

from bluesky.plans import scan
from ophyd.sim import example_detector, sample_motor

detector_list = [example_detector]

def coarse_then_fine(detectors, motor, start, stop):
    "Scan from start to stop first at low resolution, then at high resolution"
    yield from scan(detectors, motor, start, stop, 5)
    yield from scan(detectors, motor, start, stop, 20)

RE(coarse_then_fine(detector_list, sample_motor, -1, 1))

All plans imported from bluesky.plans generate full datasets (runs). Plans in the bluesky.plan_stubs module implement smaller reusable operations that you can combine to build custom workflows.

The mv() plan moves one or more devices to their target positions and waits for all movements to complete:

from bluesky.plan_stubs import mv
from ophyd.sim import motor_one, motor_two

# Move both motors to their targets simultaneously and wait for completion
RE(mv(motor_one, 1, motor_two, 10))

You can combine mv() and count() into a single custom plan like this:

def move_then_count():
    "Move motors into position, then count the detector"
    yield from mv(motor_one, 1, motor_two, 10)
    yield from count(detector_list)

RE(move_then_count())

It is critical to remember yield from. The example below will not execute your plans at all, it only defines them without running:

# WRONG EXAMPLE!
def broken_plan():
    # Forgot the "yield from" statements!
    mv(motor_one, 1, motor_two, 10)
    count(detector_list)

RE(broken_plan())

Much more complex customizations are possible, which we will cover in later sections of this tutorial. You can also refer to the full list of plan stubs for more building blocks.

Important Warning: Do not place RE(...) inside loops or custom functions. You should always call RE directly, typically from user input at the terminal, exactly once per workflow.

It is common for new users to accidentally write a loop like this:

from bluesky.plans import scan
from ophyd.sim import sample_motor, example_detector

# DO NOT DO THIS!
for step_count in [1, 2, 3]:
    print(f"{step_count} steps")
    RE(scan([example_detector], sample_motor, 5, 10, step_count))

Or wrap it in a function like this:

# DO NOT DO THIS!
def bad_function():
    for step_count in [1, 2, 3]:
        print(f"{step_count} steps")
        RE(scan([example_detector], sample_motor, 5, 10, step_count))

The correct approach builds the loop into your custom plan, and calls RE once:

from bluesky.plans import scan
from ophyd.sim import sample_motor, example_detector

def good_multi_scan():
    for step_count in [1, 2, 3]:
        print(f"{step_count} steps")
        yield from scan([example_detector], sample_motor, 5, 10, step_count)

RE(good_multi_scan())

If you nest RE calls inside other functions, you can end up with multiple nested entries and exits from the RunEngine, which causes unexpected behavior especially when handling interruptions and errors. To use an analogy: plans are the musical score, hardware is the orchestra, and the RunEngine is the conductor. You should only have one conductor, who leads the entire performance from start to finish.

Baseline Readings and Supplemental Data

In addition to you're primary detectors and motors of interest, it is often useful to capture snapshot readings (called "baseline readings") from auxiliary hardware. These readings help you verify consistency over time (for example: "Is the sample chamber temperature the same as it was last week?"). Bluesky lets you automatically capture these readings at the start and end of every run, with no extra work required per experiment.

Configuration

If you are using a pre-configured Bluesky installation at a user facility, this configuration is already completed, and you can skip this section. You can verify this by checking the sd object. If you see output similar to this, you are ready to go:

>>> sd
Out[1]: SupplementalData(baseline=[], monitors=[], flyers=[])

To enable baseline capture, add the supplemental data preprocessor to your RunEngine as shown below:

from bluesky.preprocessors import SupplementalData

sd = SupplementalData()
RE.preprocessors.append(sd)

Select Baseline Devices

You specify which devices should be automatically read at the start and end of every run. If you are using a shared configuration, this list is already populated, so check the content of sd.baseline before modifying it.

As an example, suppose we want baseline readings from three detectors and two motors. We add these devices to a list and assign it to sd.baseline:

from ophyd.sim import det1, det2, det3, motor1, motor2

sd.baseline = [det1, det2, det3, motor1, motor2]

You can mix both detectors and movable motors in this list. Bluesky does not care if a device is movable or not — it only reads the current value of each device, and all Bluesky-compatible devices support reading.

Usage

Now when you run any scan with your primary devices, RunEngine will automatically capture baseline readings before and after the run:

from ophyd.sim import det, motor
from bluesky.plans import scan

RE(scan([det], motor, -1, 1, 5))

You can clear or update the list of baseline devices at any time:

sd.baseline = []

This demonstrates a core design benefit of Bluesky: by separating plan definitions from the RunEngine executor, it is easy to apply global configuration changes with out needing to update every individual plan.

Accessing Baseline Data

When you first retrieve a run, you may think your baseline data is missing:

run_header = data_db[-1]
run_header.table()

By default, header.table() only returns the "primary" data stream containing your main experimental data. You can access other streams, including baseline, by name. You can get a full list of streams for a given run from the stream_names attribute of the header. For more details on working with streams, refer to the Databroker documentation.

Other Types of Supplemental Data

We used sd.baseline for snapshot readings in this example. The SupplementalData object also supports sd.monitors for asynchronous signal monitoring during a run, and sd.flyers for devices that participate in fly scanning. Refer to the supplemental data documentation for more details on these features.