customized per_step to collect light and dark frames at each motor point

customized_per_step.py

import bluesky.plans as bp
import bluesky.plan_stubs as bps
from bluesky.utils import Msg, short_uid as _short_uid
from bluesky.preprocessors import subs_wrapper
from bluesky.simulators import summarize_plan
from bluesky.callbacks import LiveTable


# shutter configuration, change wrt beamline!
shutter = sh
sh_open = 1
sh_close = 0


# vendor blusky nd_per_step
def light_dark_nd_step(detectors, step, pos_cache):
    """
    Inner loop of an N-dimensional step scan

    This is the default function for ``per_step`` param`` in ND plans.

    Parameters
    ----------
    detectors : iterable
        devices to read
    step : dict
        mapping motors to positions in this step
    pos_cache : dict
        mapping motors to their last-set positions
    """
    def move():
        yield Msg('checkpoint')
        grp = _short_uid('set')
        for motor, pos in step.items():
            if pos == pos_cache[motor]:
                # This step does not move this motor.
                continue
            yield Msg('set', motor, pos, group=grp)
            pos_cache[motor] = pos
        yield Msg('wait', None, group=grp)

    motors = step.keys()
    yield from move()
    yield from bps.abs_set(sh, sh_open, wait=True)
    yield from bps.trigger_and_read(list(detectors) + list(motors) +
                                    [shutter])
    yield from bps.abs_set(sh, sh_close, wait=True)
    yield from bps.trigger_and_read(list(detectors) + list(motors) +
                                    [shutter], name='dark')

print("""
Syntax example
--------------
# scan motor1 from -5 to 5 with 3 steps and motor2 from -2 to 2 with 3
# steps
plan = bp.grid_scan([area_det, shutter], motor1, -5, 5, 3,
                    motor2, -2, 2, 3, True,
                    per_step=light_dark_nd_step)

# dry-run to confirm experiment steps
summarize_plan(plan)

# execute
plan = bp.grid_scan([area_det, shutter], motor1, -5, 5, 3,
                    motor2, -2, 2, 3, True,
                    per_step=light_dark_nd_step, md=my_md)
plan = subs_wrapper(plan, LiveTable([area_det, shutter, motor1, motor2]))
xrun(<sample_ind>, plan)
""")