diff --git a/debye_bec/bec_ipython_client/plugins/move_to_label.py b/debye_bec/bec_ipython_client/plugins/move_to_label.py new file mode 100644 index 0000000..dd14970 --- /dev/null +++ b/debye_bec/bec_ipython_client/plugins/move_to_label.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import builtins +from typing import TYPE_CHECKING + +from bec_lib import bec_logger +from debye_bec.devices.absorber import STATUS as ABS_STATUS + +logger = bec_logger.logger +# import builtins to avoid linter errors +dev = builtins.__dict__.get("dev") + +class MoveToLabelError(Exception): + """Exception for the MoveToLabel function""" + +def move_to_label(): + """ + Function to move several motors to a specific position defined in the label dict. + """ + + label = get_device_conditions(label="digitalTwin") + + # Get absorber status and close if open + logger.info("Check Frontend Absorber Status") + abs_was_open = dev.abs.status.get() == ABS_STATUS.OPEN + if abs_was_open: + logger.info(" Close Frontend Absorber") + status = dev.abs.close() + status.wait() + + # Move Frontend Slits + logger.info("Move Frontend Slits into position") + devices = ["sldi_centerx", "sldi_centery", "sldi_gapx", "sldi_gapy"] + matches = {key: label[key] for key in devices if key in label} + statuses = [] + for device in matches.values(): + statuses.append(device['device'].move(device['value'])) + for status in statuses: + status.wait(timeout=30) + + # Move Collimating mirror + logger.info("Move Collimating Mirror into position") + if "cm_rotx" in label: # pitch + logger.info(" Move pitch into position") + surveyed_movement( + axis=label['cm_rotx'], + surveyed_axes= [ + {'device': dev.cm_rotz, 'abs_tol': 0.1}, + ] + ) + + # Restore absorber position + logger.info("Restore Frontend Absorber Status") + if abs_was_open: + status = dev.abs.open() + status.wait() + + +def surveyed_movement(axis, surveyed_axes): + """ + Moves an axis while surverying a set of axes. + + Args: + axis (DeviceCondition): Device condition + surveyed_axes (list): List of dicts (same format as DeviceCondition) + + Raises: + If during movement of axis, one of the surveyed axes moves out of tolerance. + """ + + for surv_ax in surveyed_axes: + surv_ax['old_value'] = surv_ax['device'].read() + status = axis['device'].move(axis['value']) + while status.status == 'RUNNING': + for surv_ax in surveyed_axes: + if abs(surv_ax['device'].read() - surv_ax['old_value']) > surv_ax['abs_tol']: + axis['device'].stop() + raise MoveToLabelError( + f"During movement of {axis['device'].name}, {surv_ax['device'].name} " + + f"started to move unexpectedly (old pos: {surv_ax['old_value']}, " + + f"current pos: {surv_ax['device'].read()})" + )