Files
AareDAQ/src/aare/daq/workflows.py
T

246 lines
9.2 KiB
Python

from aare.common.models import SampleCameraSettings
from aare.common.logger_config import setup_logger
from aare.devices.area_detector import AutoEnum
from aare.devices.bec_worker import BeamlineState
from aare.daq.devices import BeamlineDevices
from aare.daq.config import BeamlineConfig, ABR_POS_MOUNT, ABR_OMEGA_MOUNT
import time
logger = setup_logger("aareDAQ")
SAFE_POSITION = 600
def _move_detector_to_safe_position_if_needed(devs: BeamlineDevices, cfg: BeamlineConfig):
safe_position = cfg.dtz_safe_position
current_dtz = cfg.dtz
if safe_position is None:
logger.debug("No dtz_safe_position configured; skipping detector move")
return
if current_dtz is None:
logger.debug("Current dtz is unknown; skipping detector safe-position move")
return
if current_dtz < safe_position:
logger.info(f"Moving detector to safe position {safe_position}")
return devs.bec_worker.det_z(value=safe_position)
logger.debug(
f"Detector already at or beyond safe position "
f"(current={current_dtz}, safe={safe_position}); skipping move"
)
return None
def common_2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
_move_detector_to_safe_position_if_needed(devs, cfg)
devs.bec_worker.move_to(BeamlineState.ROBOT_SAMPLE_EXCHANGE)
print('move smargon home')
devs.smargon_move_home()
print('try to move aerotech')
devs.aerotech_pos = ABR_POS_MOUNT
devs.aerotech_omega = ABR_OMEGA_MOUNT
def rse2se(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
common_2rse(devs, cfg)
devs.samcam_auto(AutoEnum.ONCE)
def m2se(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.MANUAL_SAMPLE_EXCHANGE)
logger.info(f"moved to sample exchange, now moving to {cfg.abr_meas_pos}")
devs.aerotech_pos = ABR_POS_MOUNT
devs.aerotech_omega = ABR_OMEGA_MOUNT
devs.smargon_move_home()
devs.samcam_auto(AutoEnum.ONCE)
def m2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def sa2se(devs: BeamlineDevices, cfg: BeamlineConfig):
try:
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.MANUAL_SAMPLE_EXCHANGE)
logger.info(f"moved to sample exchange, now moving to {cfg.abr_meas_pos}")
devs.aerotech_pos = ABR_POS_MOUNT
devs.aerotech_omega = ABR_OMEGA_MOUNT
devs.smargon_move_home()
devs.samcam_auto(AutoEnum.ONCE)
except Exception as e:
logger.error(f"Failed to move to sample exchange: {e}")
devs.aerotech_pos = ABR_POS_MOUNT
devs.aerotech_omega = ABR_OMEGA_MOUNT
devs.smargon_move_home()
raise
def sa2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
common_2rse(devs, cfg)
def sa2xtal_snapshot(devs:BeamlineDevices, cfg:BeamlineConfig):
logger.info("SA to Xtal Snapshot")
devs.samcam_settings = SampleCameraSettings(gain=0.0, exposure=0.0005)
logger.debug(f"new sam cam settings: {devs.samcam_settings}")
devs.bec_worker.move_to(BeamlineState.XTAL_SNAPSHOT)
def dc2xtal_snapshot(devs:BeamlineDevices, cfg:BeamlineConfig):
logger.debug("DC to Xtal Snapshot")
devs.samcam_settings = SampleCameraSettings(gain=0.0, exposure=0.0005)
logger.debug(f"new sam cam settings: {devs.samcam_settings}")
devs.bec_worker.move_to(BeamlineState.XTAL_SNAPSHOT)
def xtal_snapshot2dc(devs:BeamlineDevices, cfg:BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.DATA_COLLECTION)
devs.samcam_auto(AutoEnum.ONCE)
def xtal_snapshot2sa(devs:BeamlineDevices, cfg:BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def xtal_snapshot2rse(devs:BeamlineDevices, cfg:BeamlineConfig):
common_2rse(devs, cfg)
def xtal_snapshot2se(devs:BeamlineDevices, cfg:BeamlineConfig):
sa2se(devs, cfg)
def xtal_snapshot2xrf(devs:BeamlineDevices, cfg:BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
def xtal_snapshot2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAM_VISUALISATION)
def xtal_snapshot2dh(devs: BeamlineDevices, cfg: BeamlineConfig):
xtal_snapshot2sa(devs,cfg)
sa2dh(devs,cfg)
def dc2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
common_2rse(devs, cfg)
def se2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.aerotech_pos = cfg.abr_meas_pos
devs.samcam_auto(AutoEnum.ONCE)
def rse2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
se2sa(devs, cfg)
def sa2dc(devs: BeamlineDevices, cfg: BeamlineConfig):
start = time.perf_counter()
det_z_pos = cfg.dtz
status = devs.bec_worker.det_z(value=det_z_pos)
logger.info(f"start detector move to {cfg.dtz} at {time.perf_counter() - start:.2f}")
devs.samcam_auto(AutoEnum.AUTO)
logger.info(f"sam cam auto exposure in {time.perf_counter() - start:.2f}")
devs.bec_worker.move_to(BeamlineState.DATA_COLLECTION)
logger.info(f"changed to data collection in {time.perf_counter() - start:.2f}s")
status.wait(timeout=60)
logger.info(f"det z moved to {cfg.dtz} in {time.perf_counter() - start:.2f}s")
devs.samcam_auto(AutoEnum.ONCE)
logger.info(f"sam cam auto exposure in {time.perf_counter() - start:.2f}")
def dc2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
status = _move_detector_to_safe_position_if_needed(devs, cfg)
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.aerotech_pos = cfg.abr_meas_pos
if hasattr(status, "wait"):
status.wait(timeout=60)
devs.samcam_auto(AutoEnum.ONCE)
def sa2xrf(devs: BeamlineDevices, cfg: BeamlineConfig):
"""sample alignment to XrfCollection"""
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
devs.samcam_auto(AutoEnum.ONCE)
def xrf2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
"""XrfCollection to sample alignment"""
devs.samcam_auto(AutoEnum.AUTO)
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def common2flux_measurement(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
def sa2flux_measurement(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
def flux_measurement2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
def flux_measurement2ba(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAMSTOP_ALIGNMENT)
def flux_measurement2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAM_VISUALISATION)
def ba2flux_measurement(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
def bl2flux_measurement(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.FLUX_MEASUREMENT)
def sa2ba(devs: BeamlineDevices, cfg: BeamlineConfig):
"""Sample alignment to BeamstopAlignment"""
devs.bec_worker.move_to(BeamlineState.BEAMSTOP_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def ba2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def sa2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAM_VISUALISATION)
devs.samcam_settings.exposure = 0.001
#TODO SAMCAM SETTINGS SHOULD BE DOEN via zoom settings
def bl2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def bl2ba(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAMSTOP_ALIGNMENT)
devs.samcam_auto(AutoEnum.ONCE)
def ba2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.BEAM_VISUALISATION)
devs.samcam_settings.exposure = 0.001
def sa2dh(devs: BeamlineDevices, cfg: BeamlineConfig):
common2dh(devs, cfg)
def common2dh(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.samcam_auto(AutoEnum.AUTO)
common_2rse(devs, cfg)
devs.samcam_auto(AutoEnum.ONCE)
if devs.tell.get_mounted_sample() is not None:
try:
# Best effort try to unmount
devs.tell.unmount(wait=True)
except Exception as e:
print(f"Error for unmounting: {e}")
devs.tell.dry(wait_cold=-1, wait=False)
def dh2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.SAMPLE_ALIGNMENT)
def common2maintenance(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.bec_worker.move_to(BeamlineState.MAINTENANCE)
if __name__ == "__main__":
from aare.common.beamline import mx_beamline
beamline = mx_beamline()
config = BeamlineConfig(beamline)
devices = BeamlineDevices(beamline)
common_2rse(devices, config)
time.sleep(10.0)
rse2sa(devices, config)