diff --git a/pxiii_bec/scans/mx_measurements.py b/pxiii_bec/scans/mx_measurements.py deleted file mode 100644 index b805eac..0000000 --- a/pxiii_bec/scans/mx_measurements.py +++ /dev/null @@ -1,473 +0,0 @@ -"""MX measurements module - -Scan primitives for standard BEC scans at the PX beamlines at SLS. -Theese scans define the event model and can be called from higher levels. -""" - -import time -import numpy as np -from bec_lib import bec_logger -from bec_server.scan_server.scans import AsyncFlyScanBase - -logger = bec_logger.logger - - -class AbrCmd: - """Valid Aerotech ABR scan commands from the AeroBasic files""" - - NONE = 0 - RASTER_SCAN_SIMPLE = 1 - MEASURE_STANDARD = 2 - VERTICAL_LINE_SCAN = 3 - SCREENING = 4 - # SUPER_FAST_OMEGA = 5 # Some Japanese measured samples in capilaries at high RPMs - # STILL_WEDGE = 6 # NOTE: Unused Step scan - # STILLS = 7 # NOTE: Unused Just send triggers to detector - # REPEAT_SINGLE_OSCILLATION = 8 # NOTE: Unused - # SINGLE_OSCILLATION = 9 - # OLD_FASHIONED = 10 # NOTE: Unused - # RASTER_SCAN = 11 - # JET_ROTATION = 12 # NOTE: Unused - # X_HELICAL = 13 # NOTE: Unused - # X_RUNSEQ = 14 # NOTE: Unused - # JUNGFRAU = 15 - # MSOX = 16 # NOTE: Unused - # SLIT_SCAN = 17 # NOTE: Unused - # RASTER_SCAN_STILL = 18 - # SCAN_SASTT = 19 - # SCAN_SASTT_V2 = 20 - # SCAN_SASTT_V3 = 21 - - -class AerotechFlyscanBase(AsyncFlyScanBase): - """Base class for MX flyscans - - Low-level base class for standard scans at the PX beamlines at SLS. Theese - scans use the A3200 rotation stage and the actual experiment is performed - using an AeroBasic script controlled via global variables. The base class - has some basic safety features like checking status then sets globals and - fires off the scan. Implementations can choose to set the corresponding - configurations in child classes or pass it as command line parameters. - - IMPORTANT: The AeroBasic scripts take care of the PSO configuration. - - Parameters: - ----------- - abr_complete : bool - Wait for the launched ABR task to complete. - """ - - scan_type = "fly" - scan_report_hint = "table" - arg_input = {} - arg_bundle_size = {"bundle": len(arg_input), "min": None, "max": None} - - # Aerotech stage config - abr_raster_reset = False - abr_complete = False - abr_timeout = None - pointID = 0 - num_pos = 0 - - def __init__(self, *args, parameter: dict = None, **kwargs): - """Just set num_pos=0 to avoid hanging and override defaults if explicitly set from - parameters. - """ - super().__init__(parameter=parameter, **kwargs) - if "abr_raster_reset" in self.caller_kwargs: - self.abr_raster_reset = self.caller_kwargs.get("abr_raster_reset") - if "abr_complete" in self.caller_kwargs: - self.abr_complete = self.caller_kwargs.get("abr_complete") - if "abr_timeout" in self.caller_kwargs: - self.abr_timeout = self.caller_kwargs.get("abr_timeout") - - def pre_scan(self): - """Mostly just checking if ABR stage is ok...""" - # TODO: Move roughly to start position??? - - # ABR status checking - stat = yield from self.stubs.send_rpc_and_wait("abr", "status.get") - if stat not in (0, "OK"): - raise RuntimeError("Aerotech ABR seems to be in error state {stat}, please reset") - task = yield from self.stubs.send_rpc_and_wait("abr", "task1.get") - # From what I got values are copied to local vars at the start of scan, - # so only kickoff should be forbidden. - if task not in (1, "OK"): - raise RuntimeError("Aerotech ABR task #1 seems to busy") - - # Reset the raster scan engine - if self.abr_raster_reset: - yield from self.stubs.send_rpc_and_wait("abr", "raster_scan_done.set", 0) - - # Call super - yield from super().pre_scan() - - def scan_core(self): - """The actual scan logic comes here.""" - # Kick off the run - yield from self.stubs.send_rpc_and_wait("abr", "bluekickoff") - logger.info("Measurement launched on the ABR stage...") - - # Wait for grid scanner to finish - if self.abr_complete: - if self.abr_timeout is not None: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete", self.abr_timeout) - st.wait() - else: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete") - st.wait() - - def cleanup(self): - """Set scan progress to 1 to finish the scan""" - self.num_pos = 1 - return super().cleanup() - - -class MeasureStandardWedge(AerotechFlyscanBase): - """Standard wedge scan using the OMEGA motor - - Measure an absolute continous line scan from `start` to `start` + `range` - during `move_time` on the Omega axis with PSO output. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.standard_wedge(start=42, range=10, move_time=20) - - Parameters - ---------- - start : (float, float) - Scan start position of the axis. - range : (float, float) - Scan range of the axis. - move_time : (float) - Total travel time for the movement [s]. - ready_rate : float, optional - No clue what is this... (default=500) - """ - - scan_name = "standardscan" - required_kwargs = ["start", "range", "move_time"] - - -class MeasureVerticalLine(AerotechFlyscanBase): - """Vertical line scan using the GMY motor - - Simple relative continous line scan that records a single vertical line - with PSO output. There's no actual stepping, it's only used for velocity - calculation. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.measure_vline(range_y=12, steps_y=40, exp_time=0.1) - - Parameters - ---------- - range : float - Step size [mm]. - steps : int - Scan range of the axis. - exp_time : float - Eeffective exposure time per step [s] - """ - - scan_name = "vlinescan" - required_kwargs = ["exp_time", "range", "steps"] - - -class MeasureRasterSimple(AerotechFlyscanBase): - """Simple raster scan - - Measure a simplified relative zigzag raster scan in the X-Y plane. - The scan is relative assumes the goniometer is currently at the CENTER of - the first cell (i.e. TOP-LEFT). Each line is executed as a single continous - movement, i.e. there's no actual stepping in the X direction. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.raster_simple(exp_time=0.1, range_x=4, range_y=4, steps_x=80, steps_y=80) - - Parameters - ---------- - exp_time : float - Effective exposure time for each cell along the X axis [s]. - range_x : float - Scan step size [mm]. - range_y : float - Scan step size [mm]. - steps_x : int - Number of scan steps in X (fast). Only used for velocity calculation. - steps_y : int - Number of scan steps in Y (slow). - """ - - scan_name = "rasterscan" - required_kwargs = ["exp_time", "range_x", "range_y", "steps_x", "steps_y"] - - -class MeasureScreening(AerotechFlyscanBase): - """Sample screening scan - - Sample screening scan that scans intervals on the rotation axis taking - 1 image/interval. This makes sure that we hit diffraction peaks if there - are any crystals. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.measure_screening(start=42, range=180, steps=18, exp_time=0.1, oscrange=2.0) - - Parameters - ---------- - start : float - Absolute scan start position of the omega axis [deg]. - range : float - Total screened range of the omega axis relative to 'start' [deg]. - steps : int - Number of blurred intervals. - exp_time : float - Exposure time per blurred interval [s]. - oscrange : float - Motion blurring of each interval [deg] - delta : float - Safety margin for sub-range movements (default: 0.5) [deg]. - """ - - scan_name = "screeningscan" - required_kwargs = ["start", "range", "steps", "exp_time", "oscrange"] - - -class MeasureHelical(AerotechFlyscanBase): - """Helical scan using the OMEGA motor - - Measure an absolute continous line scan from `start` to `start` + `range` - during `move_time` on the Omega axis with PSO output. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.standard_wedge(start=42, range=10, move_time=20) - - Parameters - ---------- - start : float - Scan start position of the axis. - range : float - Scan range of the axis. - move_time : float - Total travel time for the movement [s]. - ready_rate : float, optional - No clue what is this... (default=500) - sg_start : (float, float, float, float, float) - Complete SmarGon coordinate in tuple form. - sg_end : (float, float, float, float, float) - Complete SmarGon coordinate in tuple form. - sg_steps : int - Number of steps with SmarGon. - """ - - scan_name = "helicalscan" - required_kwargs = ["start", "range", "move_time", "sg_start", "sg_end", "sg_steps"] - - def pre_scan(self): - """Mostly just checking if ABR stage is ok...""" - - # Smargon has no velocity control - self.smargon_start = np.array(self.caller_kwargs.get("sg_start")) - self.smargon_end = np.array(self.caller_kwargs.get("sg_end")) - self.smargon_steps = self.caller_kwargs.get("sg_steps") - self.smargon_range = self.smargon_end - self.smargon_start - self.smargon_step_size = self.smargon_range / self.smargon_steps - self.smargon_step_time = self.caller_kwargs.get("move_time") / self.smargon_steps - - logger.info(f"Start:\t{self.smargon_start}") - logger.info(f"End:\t{self.smargon_end}") - logger.info(f"Steps:\t{self.smargon_steps}") - logger.info(f"Range:\t{self.smargon_range}") - logger.info(f"StepSize:\t{self.smargon_step_size}") - logger.info(f"StepTime:\t{self.smargon_step_time}") - - # TODO: Move roughly to start position??? - st0 = yield from self.stubs.send_rpc("shx", "omove", self.smargon_start[0]) - st1 = yield from self.stubs.send_rpc("shy", "omove", self.smargon_start[1]) - st2 = yield from self.stubs.send_rpc("shz", "omove", self.smargon_start[2]) - st3 = yield from self.stubs.send_rpc("chi", "omove", self.smargon_start[3]) - st4 = yield from self.stubs.send_rpc("phi", "omove", self.smargon_start[4]) - st0.wait() - st1.wait() - st2.wait() - st3.wait() - st4.wait() - - # Call super - yield from super().pre_scan() - - def scan_core(self): - """The actual scan logic comes here.""" - # Kick off the run - yield from self.stubs.send_rpc_and_wait("abr", "kickoff") - logger.info("Measurement launched on the ABR stage...") - - logger.info("Performing SmarGon stepping...") - for ss in range(self.smargon_steps): - sg_pos = self.smargon_start + ss * self.smargon_step_size - # Move to position but don't care - st0 = yield from self.stubs.send_rpc("shx", "omove", sg_pos[0]) - st1 = yield from self.stubs.send_rpc("shy", "omove", sg_pos[1]) - st2 = yield from self.stubs.send_rpc("shz", "omove", sg_pos[2]) - st3 = yield from self.stubs.send_rpc("chi", "omove", sg_pos[3]) - st4 = yield from self.stubs.send_rpc("phi", "omove", sg_pos[4]) - t_start = time.time() - st0.wait() - st1.wait() - st2.wait() - st3.wait() - st4.wait() - t_end = time.time() - t_elapsed = t_end - t_start - time.sleep(max(self.smargon_step_time - t_elapsed, 0)) - - # Wait for scan task to finish - if self.abr_complete: - if self.abr_timeout is not None: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete", self.abr_timeout) - st.wait() - else: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete") - st.wait() - - -class MeasureHelical2(AerotechFlyscanBase): - """Helical scan using the OMEGA motor - - Measure an absolute continous line scan from `start` to `start` + `range` - during `move_time` on the Omega axis with PSO output. - - The scan itself is executed by the scan service running on the Aerotech - controller. Ophyd just configures, launches it and waits for completion. - - Example - ------- - >>> scans.standard_wedge(start=42, range=10, move_time=20) - - Parameters - ---------- - start : float - Scan start position of the axis. - range : float - Scan range of the axis. - move_time : float - Total travel time for the movement [s]. - ready_rate : float, optional - No clue what is this... (default=500) - sg_start : (float, float, float, float, float) - Complete SmarGon coordinate in tuple form. - sg_end : (float, float, float, float, float) - Complete SmarGon coordinate in tuple form. - sg_steps : int - Number of steps with SmarGon. - """ - - scan_name = "helicalscan2" - required_kwargs = ["start", "range", "move_time", "sg_start", "sg_end", "sg_steps"] - point_id = 0 - - # def __init__(self, *args, parameter: dict = None, **kwargs): - # """Just set num_pos=0 to avoid hanging and override defaults if explicitly set from - # parameters. - # """ - # self.num_pos = kwargs["sg_steps"] - # super().__init__(*args, parameter=parameter, **kwargs) - - def prepare_positions(self): - # Smargon has no velocity control - self.smargon_start = np.array(self.caller_kwargs.get("sg_start")) - self.smargon_end = np.array(self.caller_kwargs.get("sg_end")) - self.smargon_steps = self.caller_kwargs.get("sg_steps") - self.smargon_range = self.smargon_end - self.smargon_start - self.smargon_step_size = self.smargon_range / self.smargon_steps - self.smargon_step_time = self.caller_kwargs.get("move_time") / self.smargon_steps - - logger.info(f"Start:\t{self.smargon_start}") - logger.info(f"End:\t{self.smargon_end}") - logger.info(f"Steps:\t{self.smargon_steps}") - logger.info(f"Range:\t{self.smargon_range}") - logger.info(f"StepSize:\t{self.smargon_step_size}") - logger.info(f"StepTime:\t{self.smargon_step_time}") - - self.num_pos = self.smargon_steps - self.positions = np.linspace(self.smargon_start, self.smargon_end, self.smargon_steps) - self.start_pos = self.positions[0, :] - # Call super - yield from super().prepare_positions() - - # def update_scan_motors(self): - # """ Update step scan motors""" - # self.scan_motors = ['shx', 'shy', 'shz', 'chi', 'phi'] - - def pre_scan(self): - """Mostly just checking if ABR stage is ok...""" - # Move roughly to start position - st0 = yield from self.stubs.send_rpc("shx", "omove", self.start_pos[0]) - st1 = yield from self.stubs.send_rpc("shy", "omove", self.start_pos[1]) - st2 = yield from self.stubs.send_rpc("shz", "omove", self.start_pos[2]) - st3 = yield from self.stubs.send_rpc("chi", "omove", self.start_pos[3]) - st4 = yield from self.stubs.send_rpc("phi", "omove", self.start_pos[4]) - st0.wait() - st1.wait() - st2.wait() - st3.wait() - st4.wait() - - # print(f"\n\n{self.readout_priority}\n\n") - - # Call super - yield from super().pre_scan() - - def scan_core(self): - """The actual scan logic comes here.""" - # Kick off the run - yield from self.stubs.send_rpc_and_wait("abr", "kickoff") - logger.info("Measurement launched on the ABR stage...") - - logger.info("Performing SmarGon stepping...") - for _, sg_pos in enumerate(self.positions): - # sg_pos = self.smargon_start + ss * self.smargon_step_size - # Move to position but don't care - st0 = yield from self.stubs.send_rpc("shx", "omove", sg_pos[0]) - st1 = yield from self.stubs.send_rpc("shy", "omove", sg_pos[1]) - st2 = yield from self.stubs.send_rpc("shz", "omove", sg_pos[2]) - st3 = yield from self.stubs.send_rpc("chi", "omove", sg_pos[3]) - st4 = yield from self.stubs.send_rpc("phi", "omove", sg_pos[4]) - t_start = time.time() - st0.wait() - st1.wait() - st2.wait() - st3.wait() - st4.wait() - t_end = time.time() - t_elapsed = t_end - t_start - time.sleep(max(self.smargon_step_time - t_elapsed, 0)) - yield from self.stubs.read(group="monitored", point_id=self.point_id) - self.point_id += 1 - - # Wait for scan task to finish - if self.abr_complete: - if self.abr_timeout is not None: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete", self.abr_timeout) - st.wait() - else: - st = yield from self.stubs.send_rpc_and_wait("abr", "complete") - st.wait()