Protocols#

MovableHasName#

class blop.protocols.MovableHasName(*args, **kwargs)[source]#

A movable that has a name.

We use this instead of bluesky.protocols.NamedMovable since we do not want to require HasHints on the movable.

A Movable and HasName is sufficient. HasHints should be optional.

Actuator#

blop.protocols.Actuator#: alias of MovableHasName | Flyable

Sensor#

blop.protocols.Sensor#: alias of Readable | EventCollectable | EventPageCollectable

AcquisitionPlan#

class blop.protocols.AcquisitionPlan(*args, **kwargs)[source]#

Bases: Protocol

A protocol for custom data acquisition plans.

This protocol defines how to acquire data from the beamline. Most users will use the default blop.plans.default_acquire() plan, which performs a list scan over the suggested points. Custom implementations are only needed for specialized acquisition strategies (e.g., fly scans, complex detector configurations).

See also

blop.plans.default_acquire: Default acquisition plan implementation.
blop.ax.Agent: Accepts an optional acquisition plan during initialization.

Notes

The acquisition plan is a Bluesky plan that should move the actuators to each suggested position and acquire data from the sensors. It must return the UID of the Bluesky run so that the evaluation function can retrieve the data.

__call__(suggestions, actuators, sensors=None, md=None)[source]#

Acquire data for optimization.

This should be a Bluesky plan that moves the actuators to each of their suggested positions and acquires data from the sensors.

Parameters:

suggestions: list[dict]: A list of dictionaries, each containing the parameterization of a point to evaluate. The “_id” key is optional and can be used to identify each suggestion. It is suggested to add “_id” values to the run metadata for later identification of the acquired data.
actuators: Sequence[Actuator]: The actuators to move to their suggested positions.
sensors: Sequence[Sensor], optional: The sensors that produce data to evaluate.
mddict[str, Any] | None, optional: Metadata to attach to the start document

Returns:

str: The unique identifier of the Bluesky run.

EvaluationFunction#

class blop.protocols.EvaluationFunction(*args, **kwargs)[source]#

Bases: Protocol

A protocol for transforming acquired data into measurable outcomes.

This protocol defines how to extract and compute optimization objectives from acquired data. Custom implementations are needed to define how your beamline data translates into the outcomes you want to optimize.

See also

blop.ax.Agent: Accepts an evaluation function during initialization.

Notes

The evaluation function is called after data acquisition to compute outcomes from the acquired data. It should extract relevant data from the Bluesky run and compute objective values and metrics for each suggestion.

Examples

See the tutorial documentation for complete examples of evaluation functions: Your first Bayesian optimization with Blop

__call__(uid, suggestions)[source]#

Evaluate the data from a Bluesky run and produce outcomes.

Parameters:

uid: str: The unique identifier of the Bluesky run to evaluate.
suggestions: list[dict]: A list of dictionaries, each containing the parameterization of a point to evaluate. The “_id” key is optional and can be used to identify each suggestion.

Returns:

list[dict]: A list of dictionaries containing the outcomes of the run, one for each suggested parameterization. The “_id” key is optional and can be used to identify each outcome.

Optimizer#

class blop.protocols.Optimizer(*args, **kwargs)[source]#

Bases: Protocol

A minimal optimizer interface for optimization.

This protocol defines the interface for optimizers in Blop. Most users will use the built-in blop.ax.optimizer.AxOptimizer, which provides Bayesian optimization via Ax. Custom implementations are only needed for specialized optimization algorithms.

See also

blop.ax.optimizer.AxOptimizer: Built-in Ax-based optimizer implementation.
blop.ax.Agent: High-level interface that uses AxOptimizer internally.

suggest(num_points=None)[source]#

Suggest a set of points in the input space, to be evaulated next.

The “_id” key is optional and can be used to identify suggested trials for later evaluation and ingestion.

Parameters:

num_pointsint | None, optional: The number of points to suggest. If not provided, will default to 1.

Returns:

list[dict]: A list of dictionaries, each containing a parameterization of a point to evaluate next. Each dictionary must contain a unique “_id” key to identify each parameterization.

ingest(points)[source]#

Ingest a set of points into the experiment. Either from previously suggested points or from an external source.

The “_id” key is optional and can be used to identify points from previously suggested trials or to identify the point as a “baseline” trial.

Parameters:

pointslist[dict]: A list of dictionaries, each containing the outcomes of each suggested parameterization.

get_best_points()[source]#

Get a list of the optimal points found during optimization.

For single-objective optimization, returns a single best point. For multi-objective optimization, returns the Pareto-optimal set.

Returns:

list[tuple[Any, Mapping, Mapping]]

Each element in the list is a tuple of:

“_id” of the suggestion
suggested parameters
measured outcomes

BaseOptimizationProblem#

class blop.protocols.BaseOptimizationProblem(optimizer, actuators, sensors, evaluation_function, acquisition_plan=None)[source]#

Bases: Generic[TActuator, TSensor, TPlan]

Base class for optimization problem definitions.

Provides the common structure shared by all optimization problem types. Users should use the concrete subclasses OptimizationProblem or QueueserverOptimizationProblem instead of this class directly.

See also

OptimizationProblem: Concrete problem type for standard usage.
QueueserverOptimizationProblem: Concrete problem type for queue server usage.

optimizer: Optimizer#

actuators: Sequence[TActuator]#

sensors: Sequence[TSensor]#

evaluation_function: EvaluationFunction#

acquisition_plan: TPlan | None = None#

OptimizationProblem#

class blop.protocols.OptimizationProblem(optimizer, actuators, sensors, evaluation_function, acquisition_plan=None)[source]#

Bases: BaseOptimizationProblem[MovableHasName | Flyable, Readable | EventCollectable | EventPageCollectable, AcquisitionPlan]

An optimization problem to solve. Immutable once initialized.

This dataclass encapsulates all components needed for optimization into a single immutable structure. It is typically created via blop.ax.Agent.to_optimization_problem() and used with optimization plans like blop.plans.optimize().

Attributes:

optimizer: Optimizer: Suggests points to evaluate and ingests outcomes to inform the optimization.
actuators: Sequence[Actuator]: Objects that can be moved to control the beamline using the Bluesky RunEngine. A subset of the actuators’ names must match the names of suggested parameterizations.
sensors: Sequence[Sensor]: Objects that can produce data to acquire data from the beamline using the Bluesky RunEngine.
evaluation_function: EvaluationFunction: A callable to evaluate data from a Bluesky run and produce outcomes.
acquisition_plan: AcquisitionPlan, optional: A Bluesky plan to acquire data from the beamline. If not provided, a default plan will be used.

See also

blop.ax.Agent.to_optimization_problem: Creates an OptimizationProblem from an Agent.
blop.plans.optimize: Bluesky plan that uses an OptimizationProblem.

acquisition_plan: TPlan | None = None#

optimizer: Optimizer#

actuators: Sequence[TActuator]#

sensors: Sequence[TSensor]#

evaluation_function: EvaluationFunction#

QueueserverOptimizationProblem#

class blop.protocols.QueueserverOptimizationProblem(optimizer, actuators, sensors, evaluation_function, acquisition_plan=None, acquisition_plan_kwargs=None)[source]#

Bases: BaseOptimizationProblem[str, str, str]

An optimization problem to solve. Immutable once initialized.

This dataclass encapsulates all components needed for optimization into a single immutable structure. It is typically created via blop.ax.queueserver_agent.QueueserverAgent.to_optimization_problem() and used with bluesky-queueserver-api. Actuators, sensors, and the acquisition plan are referenced by their names, since their instances live on a remote server.

Attributes:

optimizer: Optimizer: Suggests points to evaluate and ingests outcomes to inform the optimization.
actuators: Sequence[str]: Names of objects that can be moved to control the beamline using the Bluesky RunEngine. A subset of the actuators’ names must match the names of suggested parameterizations.
sensors: Sequence[str]: Names of objects that can produce data to acquire data from the beamline using the Bluesky RunEngine.
evaluation_function: EvaluationFunction: A callable to evaluate data from a Bluesky run and produce outcomes.
acquisition_plan: str, optional: The name of a Bluesky plan to acquire data. If not provided, a default plan name will be used. The plan must match the arguments of AcquisitionPlan.
acquisition_plan_kwargs: Mapping[str, Any], optional: Additional plan arguments to pass to the Bluesky plan.

See also

blop.ax.queueserver_agent.QueueserverAgent.to_optimization_problem: Creates a QueueserverOptimizationProblem from an agent.
blop.queueserver.QueueserverOptimizationRunner: Runs the optimization loop using the bluesky-queueserver-api.

acquisition_plan: TPlan | None = None#

optimizer: Optimizer#

actuators: Sequence[TActuator]#

sensors: Sequence[TSensor]#

evaluation_function: EvaluationFunction#

acquisition_plan_kwargs: Mapping[str, Any] | None = None#

CanRegisterSuggestions#

class blop.protocols.CanRegisterSuggestions(*args, **kwargs)[source]#

Bases: Protocol

A protocol for optimizers that can register suggestions.

This allows them to add an “_id” key to the suggestions dynamically and ensure that the suggestions are unique.

register_suggestions(suggestions)[source]#

Parameters:

suggestions: list[dict]: The suggestions to register. The “_id” key is optional and will be overwritten if present.

Returns:

list[dict]: The original suggestions with an “_id” key added.

TrialFaultAware#

class blop.protocols.TrialFaultAware(*args, **kwargs)[source]#

Bases: Protocol

A protocol to accept information about trial failures of the optimization loop.

Used to invalidate or register early stop on data for the optimizer, or do necesary cleanup of processes not directly tied to the run engine

register_failures(suggestions)[source]#

Parameters:

suggestions: list[dict]: The suggestions to fail. Ids must be present.

Checkpointable#

class blop.protocols.Checkpointable(*args, **kwargs)[source]#

Bases: Protocol

A protocol for objects that can can write state to persistent storage.

Implementers configure storage at construction time (e.g., a file path, databse URI). The checkpoint method then saves or updates to that pre-configured location.

checkpoint()[source]#

Write the object’s state to persistent storage.