2019.08.08 Milinda Tramp time series plan

tramp_time_series.py

##### imports ######
import time
from functools import partial
import bluesky.plan_stubs as bps
import bluesky.preprocessors as bpp
from bluesky.simulators import summarize_plan
from bluesky.callbacks import LiveTable
from xpdacq.xpdacq import glbl
from xpdacq.beamtime import (
    close_shutter_stub,
    open_shutter_stub,
    _check_mini_expo
)

####### customized scan #########
def configure_area_det_expo(exposure):
    det = xpd_configuration["area_det"]

    yield from bps.abs_set(
        det.cam.acquire_time, glbl["frame_acq_time"], wait=True
    )
    acq_time = det.cam.acquire_time.get()
    _check_mini_expo(exposure, acq_time)
    # compute number of frames
    num_frame = np.ceil(exposure / acq_time)
    yield from bps.abs_set(det.images_per_set, num_frame, wait=True)
    computed_exposure = num_frame * acq_time

    # print exposure time
    print(
        "INFO: requested exposure time = {} - > computed exposure time"
        "= {}".format(exposure, computed_exposure)
    )
    return num_frame, acq_time, computed_exposure


def simple_dark_light_count(detectors, shutter):
    """
    simply collect dark-light frame pair

    Parameters
    ----------
    detectors : iterable
        devices to read
    shutter : device
        ophyd object of shutter
    """
    yield from close_shutter_stub()
    yield from bps.trigger_and_read(list(detectors) + [shutter],
                                    name="dark")
    yield from open_shutter_stub()
    yield from bps.trigger_and_read(list(detectors) + [shutter])
    # close shutter to protect det
    yield from close_shutter_stub()


def xrd_pdf_dark_light_plan(dets, shutter, motor, pdf_pos, xrd_pos,
                            verbose=False):
    """
    move detector between xrd and pdf positions,
     take dark-light pair at each positions

    Parameters
    ----------
    dets : iterable
        **ALL** devices to read and record
    shutter : device
        ophyd object of shutter
    motor : settable object from Ophyd
        motor used to move between xrd and pdf positions
    pdf_pos : float
        position of pdf measurement
    xrd_pos : float
        position of xrd measurement
    verbose : bool, optional
        argument to show entire
    """
    t0 = time.time()
    for pos in (pdf_pos, xrd_pos):
        yield from bps.abs_set(motor, pos, wait=True)
        # check point for revert
        yield from bps.checkpoint()
        yield from simple_dark_light_count(list(dets), shutter)
        # check point for revert
        yield from bps.checkpoint()
    t1 = time.time()
    if verbose:
        print("INFO: one xrd-pdf dark-light cycle is : %.2f s" % (t1-t0))


def customized_plan(dets, exposure_time, shutter, temp_mv, temp_list,
                    motor, pdf_pos, xrd_pos, num, delay=None):
    """
    per UC from Millinda's email

    Parameters
    ----------
    dets : iterable
        **ALL** ophyd devices to read and record
    exposure_time : float
        exposure time at each count
    temp_mv : device
        ophyd object of the temperature controller
    temp_list : list
        list of temperatures to drive to
    shutter : device
        ophyd object of the shutter
    motor : settable object from Ophyd
        motor used to move between xrd and pdf positions
    pdf_pos : float
        position of pdf measurement
    xrd_pos : float
        position of xrd measurement
    num : int
        repeating time of the xrd-pdf measurment
    delay : float, optional
        delay between each xrd-pdf measurment
        if None, no delay. default is None.

    Example
    -------
    # set alias of beamline objects name, please change accordingly
    rga_obj = rga
    area_det = pe1
    temp_mv = eurotherm
    shutter = fs
    det_mv = Det_1_Z  # motor for moving det between pdf and xrd positions
    z1, z2 = 1195, 1995  # pdf and xrd positions repectively
    exposure_time = 600  # exposure time for all scans
    delay = 15  # delay time between two dark-light cycle
    num = 50  # number of dark-light collection

    # set up detectors to trigger and run the plan
    dets = [area_det, rga_obj, det_mv, temp_mv, shutter]
    temp_list = [200, 300, 400]
    ugh = customized_plan(dets, exposure_time, shutter,
                          temp_mv, temp_list,
                          det_mv, z1, z2, num, delay)
    ughh = bpp.subs_wrapper(ugh, LiveTable(dets))
    glbl["shutter_control"] = False  # disable auto-dark
    xrun(1, ughh)   # measure smaple index 1
    glbl["shutter_control"] = True # enable auto-dark
    """
    # configure area det first, out side of plan to avoid any issue
    rv = yield from configure_area_det_expo(exposure_time)
    num_frame, acq_time, computed_exposure = rv
    # minimum md
    md = {"sp_time_per_frame": acq_time, "sp_num_frames": num_frame,
          "sp_requested_exposure": exposure_time,
          "sp_computed_exposure": computed_exposure,
          "sp_type": "tramp_tseries",
          "detectors": [det.name for det in dets],
          "hints": {}
          }
    try:
        dimensions = [(m.hints['fields'], 'primary') for m in [temp_mv, motor]]
    except (AttributeError, KeyError):
        pass
    else:
        md['hints'].setdefault('dimensions', dimensions)
        
    @bpp.stage_decorator(dets)
    @bpp.run_decorator(md=md)
    def repeat_plan():
        # !!!FIXME!!!: yield from bps.abs_set(temp_mv.rate, 10)
        for set_temp in temp_list:
            print("Drive temp_mv to ", set_temp)
            yield from bps.abs_set(temp_mv, set_temp, wait=True)
            print("collect data")
            yield from bps.repeat(partial(xrd_pdf_dark_light_plan,
                                              dets, shutter,
                                              motor, pdf_pos, xrd_pos,
                                              False), num, delay)
        # !!!FIXME!!!: yield from bps.abs_set(temp_mv.rate, 50)
        set_temp = 300
        print("Drive temp_mv to ", set_temp)
        yield from bps.abs_set(temp_mv, set_temp, wait=True)

    return (yield from repeat_plan())