Plan Stubs

blop.plan_stubs.list_scan_in_run(suggestions, actuators, sensors=None, *, per_step=None, **kwargs)

Acquire suggestions inside an already-open Bluesky run.

This plan moves through the suggestions, optionally reordering them for efficient motion, and executes a Bluesky list scan without opening a child run. The list scan’s stage and unstage messages are preserved.

Warning

The single-run optimization API is experimental. This plan may change in future releases without a deprecation period.

Parameters:

suggestionsSequence[Mapping]

Suggested parameterizations to execute. Each suggestion must contain a hashable _id.

actuatorsSequence[Actuator]

Actuators to move to the suggested positions.

sensorsSequence[Sensor] | None, optional

Sensors that produce data to evaluate. Non-readable sensors are ignored.

per_stepbp.PerStep | None, optional

Bluesky list-scan step hook. Custom hooks may emit any number of events into any streams.

**kwargsAny

Additional keyword arguments to pass to bluesky.plans.list_scan().

Returns:

tuple[Hashable, …]

Suggestion IDs in the order the suggestions were executed.

This identifier intentionally does not encode stream names, event UIDs, event counts, or per-stream offsets. Custom per_step hooks may emit any number of events into any number of streams. The matching evaluation function is responsible for interpreting those documents and correlating them with these ordered suggestion IDs.

Yields:

Msg

Bluesky messages.

blop.plan_stubs.navigate_to_best(actuators, optimizer=None, parameterization=None)

Move actuators to the best point found during optimization.

If no explicit parameterization is provided, queries the optimizer for its best point(s). For multi-objective optimizers that return multiple Pareto-optimal points, an explicit parameterization must be provided.

Parameters:

actuatorsSequence[Actuator]

The actuators to move to the best parameterization.

optimizerOptimizer | None, optional

The optimizer to query for the best point.

parameterizationMapping | None, optional

Explicit parameterization to navigate to. If None, queries the optimizer’s best point. For multi-objective problems, call optimizer.get_best_points() to inspect the Pareto set and select one.

Raises:

TypeError

If both parameterization and optimizer arguments are None.

ValueError

If the optimizer returns multiple Pareto-optimal points and no explicit parameterization is provided.

blop.plan_stubs.read_step(uid, suggestions, outcomes, n_points, readable_cache, stream_name='primary')

Plan stub to read the suggestions and outcomes of a single optimization step.

If fewer suggestions are returned than n_points arrays are padded to n_points length with np.nan to ensure consistent shapes for event-model specification.

The emitted acquisition_uid field retains native array-like identifiers. Other hashable identifiers are represented by repr(uid).

Parameters:

uidHashable

The acquisition identifier returned by the acquisition plan.

suggestionsSequence[Mapping]

Sequence of suggestion mappings, each containing an ID_KEY.

outcomesSequence[Mapping]

Sequence of outcome mappings, each containing an ID_KEY matching suggestions.

n_pointsint

Expected number of suggestions. Arrays will be padded to this length if needed.

readable_cachedict[str, InferredReadable]

Cache of InferredReadable objects to reuse across iterations.

stream_namestr, optional

Event stream name for the optimization tracking event.