Plans#

blop.plans.optimize(optimization_problem, iterations=1, n_points=1, checkpoint_interval=None, readable_cache=None, **kwargs)[source]#

Solve the optimization problem.

Parameters:

optimization_problemOptimizationProblem: The optimization problem to solve.
iterationsint | None, optional: The maximum number of optimization iterations to run. If None, run until the optimizer’s stopping criterion is met. An optimizer implementing blop.protocols.SupportsStoppingCriteria is required when None.
n_pointsint, optional: The number of points to suggest per iteration.
checkpoint_intervalint | None, optional: The number of iterations between optimizer checkpoints. If None, checkpoints will not be saved. Optimizer must implement the blop.protocols.Checkpointable protocol.
readable_cache: dict[str, InferredReadable] | None = None: Cache of readable objects to store the suggestions and outcomes as events. If None, a new cache will be created.
**kwargsAny: Additional keyword arguments to pass to the optimize_step() plan.

Raises:

ValueError: If iterations is None and the optimizer does not implement blop.protocols.SupportsStoppingCriteria.

See also

blop.protocols.OptimizationProblem: The problem to solve.
blop.protocols.Checkpointable: The protocol for checkpointable objects.
optimize_step: The plan to execute a single step of the optimization.

blop.plans.optimize_in_run(optimization_problem, iterations=1, n_points=1, checkpoint_interval=None, readable_cache=None, **kwargs)[source]#

Solve an optimization problem by evaluating acquisition documents inside one run.

Warning

This plan is experimental. Its API is not yet stable and may change in future releases without a deprecation period.

Parameters:

optimization_problemOptimizationProblem: The optimization problem to solve.
iterationsint, optional: The number of optimization iterations to run.
n_pointsint, optional: The number of points to suggest per iteration.
checkpoint_intervalint | None, optional: The number of iterations between optimizer checkpoints. If None, checkpoints will not be saved. Optimizer must implement the blop.protocols.Checkpointable protocol.
readable_cache: dict[str, InferredReadable] | None = None: Cache of readable objects to store the suggestions and outcomes as optimization events. If None, a new cache will be created.
**kwargsAny: Additional keyword arguments to pass to the acquisition plan.

blop.plans.optimize_step(optimization_problem, n_points=1, *args, **kwargs)[source]#

Single step of the optimization loop.

Parameters:

optimization_problemOptimizationProblem: The optimization problem to solve.
n_pointsint, optional: The number of points to suggest.

Returns:

tuple[Hashable, Sequence[Mapping], Sequence[Mapping]]: The acquisition identifier, suggestions, and outcomes of the step.

blop.plans.default_acquire(suggestions, actuators, sensors=None, md=None, *, per_step=None, **kwargs)[source]#

Acquire data for optimization. Simply a list scan.

Includes "blop_suggestions" metadata containing the routed suggestions for backwards compatibility and "blop_acquisition_order" containing IDs in actual scan order. Use those IDs, rather than positions in suggestions, to associate acquired rows.

Parameters:

suggestions: Sequence[Mapping]: A sequence of mappings, each containing the parameterization of a point to evaluate. Each mapping must contain a unique "_id" key used to associate the acquired data with its suggestion.
actuators: Sequence[Actuator]: The actuators to move and the inputs to move them to.
sensors: Sequence[Sensor]: The sensors that produce data to evaluate.
mdMapping[str, Any] | None, optional: Metadata to attach to the start document.
per_step: bp.PerStep | None, optional: The plan to execute for each step of the scan.
**kwargs: Any: Additional keyword arguments to pass to the list_scan plan.

Returns:

str: The UID of the Bluesky run.

See also

bluesky.plans.list_scan: The Bluesky plan to acquire data.

blop.plans.acquire_baseline(optimization_problem, parameterization=None, **kwargs)[source]#

Acquire a baseline reading. Useful for relative outcome constraints.

Parameters:

optimization_problemOptimizationProblem: The optimization problem to solve.
parameterizationdict[str, Any] | None = None: Move the DOFs to the given parameterization, if provided.

See also

default_acquire: The default plan to acquire data.

blop.plans.sample_suggestions(optimization_problem, suggestions, readable_cache=None, **kwargs)[source]#

Evaluate specific parameter combinations.

This plan acquires data for given suggestions and ingests results into the optimizer. Supports both optimizer-generated suggestions (with “_id”) and manual points (without “_id”, if optimizer implements CanRegisterSuggestions).

Parameters:

optimization_problemOptimizationProblem

The optimization problem.

suggestionsSequence[Mapping]

Parameter combinations to evaluate. Can be:

Optimizer suggestions (with “_id” keys from suggest())
Manual points (without “_id”, requires CanRegisterSuggestions protocol)

readable_cachedict[str, InferredReadable] | None

Cache for storing suggestions/outcomes as events.

**kwargsAny

Additional arguments for acquisition plan.

Returns:

uidHashable: The acquisition identifier returned by the acquisition plan.
suggestionsSequence[Mapping]: Suggestions with “_id” keys.
outcomesSequence[Mapping]: Evaluated outcomes.

Raises:

ValueError: If suggestions lack “_id” and optimizer doesn’t implement CanRegisterSuggestions.

See also

optimize_step: Standard optimizer-driven step.
blop.protocols.CanRegisterSuggestions: Protocol for manual suggestions.