Write Bluesky Plans for Blueapi

See also

The bluesky documentation for an introduction to bluesky plans and general forms/advice. Blueapi has some additional requirements, which are explained below.

Format

See also

Explanation of why blueapi treats plans in a special way

Plans in Python files look like this:

from typing import Any

from bluesky.protocols import Movable, Readable
from bluesky.utils import MsgGenerator


def my_plan(
    detector: Readable,
    motor: Movable,
    steps: int,
    sample_name: str,
    extra_metadata: dict[str, Any],
) -> MsgGenerator[None]:
    # logic goes here
    ...

Detection

The type annotations in the example above (e.g. : str, : int, -> MsgGenerator) are required as blueapi uses them to detect that this function is intended to be a plan and generate its runtime API. If there is an __all__ dunder present in the module, blueapi will read that and import anything within that qualifies as a plan, per its type annotations. If not it will read everything in the module that hasn’t been imported, for example it will ignore a plan imported from another module.

Input annotations should be as broad as possible, the least specific implementation that is sufficient to accomplish the requirements of the plan. For example, if a plan is written to drive a specific motor (MyMotor), but only uses the general methods on the Movable protocol, it should take Movable as a parameter annotation rather than MyMotor.

Injecting Devices

Some plans are created for specific sets of devices, or will almost always be used with the same devices, it is useful to be able to specify defaults. Dodal makes this easy with its factory functions.

Injecting Metadata

The bluesky event model allows for rich structured metadata to be attached to a scan. To enable this to be used consistently, blueapi encourages a standard form.

Plans (as opposed to stubs) should include metadata as their final parameter, if they do it must have the type dict[str, Any] | None, and a default of None. If the plan calls to a stub/plan which takes metadata, the plan must pass down its metadata, which may be a differently named parameter.

from typing import Any

import bluesky.plans as bp
from bluesky.protocols import Readable
from bluesky.utils import MsgGenerator


def pass_metadata(
    det: Readable,
    metadata: dict[str, Any] | None = None,
) -> MsgGenerator:
    yield from bp.count([det], md=metadata or {})

Docstrings

Blueapi exposes the docstrings of plans to clients, along with the parameter types. It is therefore worthwhile to make these detailed and descriptive. This may include units of arguments (e.g. seconds or microseconds), its purpose in the function, the purpose of the plan etc.

from typing import Any

import bluesky.plan_stubs as bps
import bluesky.plans as bp
from bluesky.protocols import Movable, Readable
from bluesky.utils import MsgGenerator
from dodal.common.coordination import inject

_DEFAULT_TEMPERATURE_CONTROLLER = inject("sample_temperature_controller")
_DEFAULT_PRESSURE_CONTROLLER = inject("sample_pressure_controller")


def temp_pressure_snapshot(
    detectors: list[Readable],
    temperature: Movable = _DEFAULT_TEMPERATURE_CONTROLLER,
    pressure: Movable = _DEFAULT_PRESSURE_CONTROLLER,
    target_temperature: float = 273.0,
    target_pressure: float = 10**5,
    metadata: dict[str, Any] | None = None,
) -> MsgGenerator:
    """
    Moves devices for pressure and temperature (defaults fetched from the context)
    and captures a single frame from a collection of devices
    Args:
        detectors: A list of devices to read while the sample is at STP
        temperature: A device controlling temperature of the sample,
            defaults to fetching a device name "sample_temperature" from the context
        pressure: A device controlling pressure on the sample,
            defaults to fetching a device name "sample_pressure" from the context
        target_pressure: target temperature in Kelvin. Default 273
        target_pressure: target pressure in Pa. Default 10**5
    Returns:
        MsgGenerator: Plan
    Yields:
        Iterator[MsgGenerator]: Bluesky messages
    """

    # Prepare sample environment
    yield from bps.abs_set(temperature, target_temperature, wait=True, group="init")
    yield from bps.abs_set(pressure, target_pressure, wait=True, group="init")
    yield from bps.wait(group="init")

    # Take data
    yield from bp.count(detectors, num=1, md=metadata or {})