mx/AareDAQ
0
1 Fork 0
Code Issues 27 Pull Requests 1 Actions Packages Projects Releases 99 Wiki Activity

Restructure

Browse Source
This commit is contained in:
leonarski_f committed 2026-01-15 15:21:32 +01:00
1 parent b506542b01
commit 0044b782ae
5 files changed
+191 -906
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/aare/daq/auth.py
+1 -1
View File
@@ -9,7 +9,7 @@ from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from aare.aaredaq.config import BeamlineConfig
from aare.daq.config import BeamlineConfig
if os.environ.get("JWT_AAREDAQ_KEY") is None:
raise Exception("JWT_AAREDAQ_KEY environment variable not set, cannot guarantee safe authentication.")
src/aare/daq/daq.py
+44 -383
View File
@@ -1,6 +1,7 @@
import copy
import secrets
import time
from copy import deepcopy
from datetime import datetime
from math import ceil
from typing import List, Tuple, Optional
@@ -10,13 +11,13 @@ import numpy as np
import redis
import aare.common.face_detection as fd
from aare.aaredaq import workflows
from aare.aaredaq.aaredb import AareWrapper
from aare.aaredaq.autofocus import calculate_focus_measure
from aare.aaredaq.config import BeamlineConfig, ABR_POS_MOUNT
from aare.aaredaq.config import BeamlineStateEnum
from aare.aaredaq.devices import BeamlineDevices
from aare.aaredaq.mlbox import MlBox
from aare.daq import workflows
from aare.daq.aaredb import AareWrapper
from aare.daq.autofocus import calculate_focus_measure
from aare.daq.config import BeamlineConfig, aerotech_pos_MOUNT
from aare.daq.config import BeamlineStateEnum
from aare.daq.devices import BeamlineDevices
from aare.daq.mlbox import MlBox
from aare.common.beamline import MXBeamline
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.diffraction_geometry import DiffractionGeometry
@@ -195,13 +196,13 @@ class AareDAQ:
@property
def omega(self) -> float:
return self.__devs.aerotech.omega
return self.__devs.aerotech_omega
def __omega(self, val: float):
self.__saved_box = None
if -2000 < val < 2000:
try:
self.__devs.aerotech.move(val, wait=True, speed=180.0, direct=True)
self.__devs.aerotech_omega = val
self.__cfg.state_busy = False
except Exception as e:
logger.error(f"Omega error: {e}")
@@ -219,7 +220,7 @@ class AareDAQ:
def omega_rel(self, val: float):
self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
curr_omega = self.__devs.aerotech.omega
curr_omega = self.__devs.aerotech_omega
self.__omega(curr_omega + val)
@property
@@ -230,15 +231,14 @@ class AareDAQ:
def zoom(self, val: float):
self.__saved_box = None
self.__devs.zoom = val
print(f"{self.__cfg.state}")
if self.__cfg.state == BeamlineStateEnum.BeamLocation:
self.__cfg.zoom_mode = ZoomModeEnum.BeamLocation
else:
self.__cfg.zoom_mode = ZoomModeEnum.User
zoom_settings = self.__cfg.zoom_settings.get_camera_settings(val)
print(f"Setting zoom {val} {zoom_settings}")
self.samcam_settings = zoom_settings
#elf.__cfg.state_busy = False
@property
def light(self) -> float:
@@ -260,9 +260,7 @@ class AareDAQ:
def sample(self) -> SampleShortInfo | None:
if self.__cfg.current_sample is not None and self.__cfg.current_sample.location is None:
return self.__cfg.current_sample
if self.__devs.tell.get_mounted_sample() is None:
return None
return self.__cfg.current_sample
return None
@property
def samcam_settings(self) -> SampleCameraSettings:
@@ -272,34 +270,12 @@ class AareDAQ:
def samcam_settings(self, s: SampleCameraSettings):
self.__devs.samcam_settings = s
def autofocus(self, f: AutofocusSettings):
pos = self.__cfg.abr_meas_pos
z_min = pos.z - f.z_range_um / 2.0
z_max = pos.z + f.z_range_um / 2.0
measures = []
for i in range(f.z_steps):
pos.z = z_min + i * (z_max - z_min) / f.z_steps
self.__devs.abr_pos = pos
self.__devs.sample_cam.get_single_image()
image = self.__devs.sample_cam.get_image(gray=True)
val = calculate_focus_measure(image, f.center_x_pxl, f.center_y_pxl, f.radius_pxl)
measures.append((pos.z, val))
best_pos, _ = max(measures, key=lambda x: x[1])
pos.z = best_pos
self.__devs.abr_pos = pos
self.__cfg.abr_meas_pos = pos
self.__devs.sample_cam.collect_auto()
def tweak_abr_meas_pos(self, c: Coordinate):
self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
try:
new_meas_pos = self.__cfg.abr_meas_pos + c
self.__cfg.abr_meas_pos = new_meas_pos
self.__devs.abr_pos = new_meas_pos
self.__devs.aerotech_pos = new_meas_pos
self.__saved_box = None
self.__cfg.state_busy = False
except:
@@ -309,7 +285,7 @@ class AareDAQ:
def save_abr_meas_pos(self):
self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
try:
self.__cfg.abr_meas_pos = self.__devs.abr_pos
self.__cfg.abr_meas_pos = self.__devs.aerotech_pos
self.__cfg.state_busy = False
except:
self.__cfg.state_busy = False
@@ -318,7 +294,7 @@ class AareDAQ:
def goto_abr_meas_pos(self):
self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
try:
self.__devs.abr_pos = self.__cfg.abr_meas_pos
self.__devs.aerotech_pos = self.__cfg.abr_meas_pos
self.__cfg.state_busy = False
except:
self.__cfg.state_busy = False
@@ -364,110 +340,8 @@ class AareDAQ:
raise MountingFailed(f"{mount_error}")
def __mount(self, target: SampleShortInfo | None):
self.__set_state(BeamlineStateEnum.RobotSampleExchange)
self.__saved_box = None
logger.info("Moving smargon to home")
self.__devs.smargon.move_home(wait=True)
logger.info("moving aerotech to mount position")
self.__devs.abr_pos = ABR_POS_MOUNT
time.sleep(0.1)
logger.info("checking beamstop")
if self.__devs.bsz.value < 24.0:
raise Exception("Beamstop Z below 24.0 mm - potentially unsafe with mounting")
logger.info(f"Checking magnet postion sensor positon: {self.__devs.magnet_position_sensor_readout.value}")
logger.info(f"Checking smargon position: {self.__devs.smargon.readback}")
logger.info(f"Checking abr position: {self.__devs.abr_pos}")
if self.__devs.magnet_position_sensor.value != 0:
time.sleep(1)
logger.warning("!!!!!!!!!!!!!!!!!!!!!MAGNET CONTROLLER BROKE AGAIN!!!!!!!!!!!!!!!!!!")
logger.debug(f"Checking magnet position sensor positon: {self.__devs.magnet_position_sensor_readout.value}")
logger.debug(f"Checking smargon position: {self.__devs.smargon.readback}")
logger.debug(f"Checking abr position: {self.__devs.abr_pos}")
if self.__devs.magnet_position_sensor.value != 0:
start = time.time()
end = start + 360
while self.__devs.magnet_position_sensor.value != 0:
time.sleep(1)
logger.debug(f"Checking magnet position sensor positon: {self.__devs.magnet_position_sensor_readout.value}")
logger.debug(f"Checking smargon position: {self.__devs.smargon.readback}")
logger.debug(f"Checking abr position: {self.__devs.abr_pos}")
if time.time() > end:
raise Exception(f"Magnet position sensor is not in position: {self.__devs.magnet_position_sensor_readout.value}")
#Reenable TELL after doors locked
logger.info("enable tell motion after doors locked")
self.__devs.tell.check_enable_motion()
logger.info("Waiting for tell to be ready")
self.__devs.tell.wait_ready()
self.__devs.tell.wait_mount_complete()
logger.debug("Moving tell to mount position")
self.__devs.tell.set_in_mount_position(True)
curr_sample = self.__cfg.current_sample
if curr_sample is not None and curr_sample.location is None:
logger.info(f"Unmounting current sample")
# TODO: Smarter way to know manual sample has be removed from goniometer
self.__aare.sample_unmounted(curr_sample)
self.__cfg.current_sample = None
elif target is None:
logger.info(f"Moving tell to unmount with no target")
self.__devs.tell.unmount(wait=True, timeout=360)
self.__aare.sample_unmounted(curr_sample)
self.__cfg.current_sample = None
else:
# reset zoom
self.zoom = 1
# mount sample
self.__aare.sample_unmounted(curr_sample)
logger.info(f"Moving tell to mount")
value = self.__devs.tell.mount(
address=target.tell_address(),
force=True,
auto_unmount=True,
read_dm=False,
wait=True,
timeout=360,
)
logger.debug(f"Post tell mount, pre db input")
if value is not None:
if value == "No Pin in Gripper":
mount_error = "No sample was detected in gripper"
logger.error(mount_error)
self.__aare.sample_failed(target, failed_comment = mount_error)
self.__mount_failure_handler(mount_error=mount_error)
self.__cfg.current_sample = None
logger.debug(f"setting current sample to None as previous unmount succeeded but mount failed")
raise MountingFailed(f"{mount_error}: drying gripper {target.sample_name} {target.location} {target.pin}")
elif value == "dry":
logger.info("Robot is drying")
else:
logger.info(f"Event detected from robot: {value}")
mounted = self.__devs.tell.get_mounted_sample()
logger.info(f"response from tell {mounted}")
if not mounted:
mount_error = "Failed to mount target"
logger.error(f"{mount_error}: {target.db_id} {target.location} {target.pin}")
self.__aare.sample_failed(target, failed_comment = mount_error)
#self.__mount_failure_handler(mount_error=mount_error)
self.__cfg.current_sample = None
logger.debug(f"setting current sample to None but not sure if unmount succeeded")
raise MountingFailed(f"No Sample detected on smart magent: {target.sample_name} {target.location} {target.pin}")
if target is not None:
self.__aare.sample_mounted(target)
logger.info(f"Target mounted: {target.db_id} {target.location} {target.pin}")
self.__cfg.failed_mount_count = 0
self.save_screenshot_db(target.db_id, f"sample_mounted")
else:
self.__cfg.failed_mount_count = 0
logger.info(f"Target is None: {target}")
self.__cfg.current_sample = target
# TODO: Need to know if dry needs to happen
pass
@sample.setter
def sample(self, target: SampleShortInfo | None):
# This will set internally state to SampleExchange, but will return to SampleAlignment
@@ -524,16 +398,16 @@ class AareDAQ:
@property
def camera_image(self) -> np.ndarray | None:
image = self.__devs.sample_cam.get_image(gray=False)
image = self.__devs.samcam_get_image(gray=False)
return image
@property
def camera_image_gray(self) -> np.ndarray | None:
image = self.__devs.sample_cam.get_image(gray=True)
image = self.__devs.samcam_get_image(gray=True)
return image
def list_loaded_pucks(self) -> List[PuckLoadedInfo]:
return self.__devs.tell.get_detected_pucks()
return []
def __auto_center(self, grid: RasterGridRequest) -> CompletedRasterGrid | None:
@@ -547,7 +421,7 @@ class AareDAQ:
r = self.__ml_bounding_box(sample.db_id, f"ml_{geom.omega_deg:.2f}deg")
if r is None:
self.__devs.aerotech.move(geom.omega_deg + 90.0, wait=True)
self.__devs.aerotech_omega = geom.omega_deg + 90.0
time.sleep(0.2)
r = self.__ml_bounding_box(sample.db_id, f"ml_{geom.omega_deg + 90.0:.2f}deg")
@@ -561,7 +435,7 @@ class AareDAQ:
grid.omega_deg = geom.omega_deg
res1 = self.__raster(grid)
grid.omega_deg += 90
self.__devs.aerotech.move(grid.omega_deg, wait=True)
self.__devs.aerotech_omega = grid.omega_deg
grid.n_x = 1
#TODO generate y scan rather than had code for 1x50
@@ -580,118 +454,7 @@ class AareDAQ:
def __raster(self, r: RasterGridRequest) -> CompletedRasterGridElem:
#TODO more sensible max_time adding, for large gridscans may fail.
max_time = r.exp_time_s * r.n_y * r.n_x + 60
if r.dtz is not None:
self.__cfg.dtz = r.dtz
self.__set_state(BeamlineStateEnum.DataCollection)
save_smargon_position = self.__devs.smargon.readback
if r.smargon_top_left is not None:
delta_mm = self.sample_geometry.smargon_nudge(Coordinate(x=r.grid_size_mm.x / 2, y=r.grid_size_mm.y / 2))
self.__devs.smargon.target = SmargonCoordinate(sh_mm=r.smargon_top_left.sh_mm + delta_mm,
phi_deg=r.smargon_top_left.phi_deg,
chi_deg=r.smargon_top_left.chi_deg)
if r.transmission is not None:
self.__devs.transmission.set(r.transmission, wait=False)
self.__devs.aerotech.move(r.omega_deg, wait=True, speed=360.0)
self.__devs.transmission.wait()
self.__devs.smargon.wait()
status = self.status
logger.info(f"raster status: {status}")
logger.info(f"raster grid request: {r}")
self.__jfjoch.measure_raster(r, status)
self.__aare.create_gridscan_run(self.sample, r, status)
if self.sample is not None and self.sample.db_id is not None:
self.save_screenshot_db(self.sample.db_id, "before_raster")
if r.n_x == 1:
self.__devs.aerotech.measure_vertical_line(r.exp_time_s, r.grid_size_mm.y, r.n_y)
else:
self.__devs.aerotech.measure_raster_simple(
r.exp_time_s, r.grid_size_mm.x, r.grid_size_mm.y, r.n_x, r.n_y
)
self.__devs.aerotech.wait_scan_done(max_time)
self.__devs.aerotech.reset()
self.__devs.abr_pos = self.__cfg.abr_meas_pos
result = self.__jfjoch.wait_till_done(60)
images = result.images
result_array = create_quality_filtered_array(images,
'spots_low_res', min_spots=None,
min_efficiency=1.0, min_background=None,
min_low_res_spots=None)
#if not has_sufficient_low_res_spots(result_array, self.MIN_SPOTS_LOW_RES_THRESHOLD):
# logger.error("Auto X-ray centering failed: insufficient low-resolution spots "
# f"(threshold={self.MIN_SPOTS_LOW_RES_THRESHOLD})")
# self.__aare.ingest_gridscan(sample=self.sample, raster_result=result,
# raster_request=r, geom=self.sample_geometry,
# com=None, beam_mark_pxl=self.__cfg.get_beam_mark(self.zoom))
# raise AXCFailed("Auto X-ray centering failed: insufficient low-resolution spots")
self.__cfg.crystal_size = get_xtal_size(self.__cfg.crystal_size, result_array, r)
com = raster_centre_of_mass(result_array, images)
if com is None:
logger.debug(f"using old method as COM is disabled")
com = identify_crystal_raster(result, r)
if com:
com_mm = com.get_com_mm(r)
grid_mm_x = com_mm.x
grid_mm_y = com_mm.y
else:
grid_mm_x = None
grid_mm_y = None
if grid_mm_x or grid_mm_y:
new_delta_mm = self.sample_geometry.smargon_nudge(Coordinate(x=grid_mm_x, y=grid_mm_y))
if r.n_x != 1:
result_list = get_result_list_from_com(images, com)
self.__cfg.last_best_b_factor = get_best_b_factor(result_list)
self.__cfg.last_best_res = get_best_res(result_list)
logger.debug(f"b_factor: {self.__cfg.last_best_b_factor}, best_res: {self.__cfg.last_best_res}")
else:
new_delta_mm = None
if new_delta_mm is not None:
if r.smargon_top_left:
new_target = r.smargon_top_left
else:
new_target = save_smargon_position
logger.debug(f'moving SMARGON to target new delta mm {new_target.sh_mm + new_delta_mm} mm')
self.__devs.smargon.target = SmargonCoordinate(sh_mm=new_target.sh_mm + new_delta_mm,
phi_deg=new_target.phi_deg,
chi_deg=new_target.chi_deg)
else:
logger.error("Auto finding optimal image failed. Using previous position.")
self.__devs.smargon.target = save_smargon_position
if self.sample is not None and self.sample.db_id is not None and result is not None:
self.__devs.reflector_up = True
time.sleep(0.1)
self.save_screenshot_db(self.sample.db_id, f"post_raster_{r.omega_deg}deg")
if com and hasattr(com, "max_image"):
diffraction_image_filename = f"{self.sample.db_id}_best_diffraction_from_raster_image_{com.max_image}"
diffraction_image = self.__jfjoch.get_diffraction_image(com.max_image)
self.__aare.upload_jpg(self.sample.db_id, diffraction_image_filename, diffraction_image)
try:
self.__aare.ingest_gridscan(sample = self.sample, raster_result =result,
raster_request = r, geom = self.sample_geometry,
com = com, beam_mark_pxl=self.__cfg.get_beam_mark(self.zoom))
except Exception as e:
logger.error(f"Exception ingesting grid scan: {e}")
self.__devs.reflector_up = False
return CompletedRasterGridElem(request=copy.deepcopy(r), result=result, centre_of_mass=com)
return None
def measure_raster(self, r: RasterGridRequest, auto: bool) -> CompletedRasterGrid:
self.__cfg.try_set_busy(timeout=ceil(360))
@@ -714,76 +477,7 @@ class AareDAQ:
raise e
def __rotation(self, request: RotationScanRequest) -> CompletedRotationScan:
if request.dtz is not None:
logger.info(f'requesting dtz to move to {request.dtz}')
self.__cfg.dtz = request.dtz
if self.sample is not None and self.sample.db_id is not None:
self.save_screenshot_db(self.sample.db_id, f"pre_rotation{self.sample.db_id}_{self.__devs.zoom}")
self.__set_state(BeamlineStateEnum.DataCollection)
if request.transmission is not None:
logger.info(f'requesting transmission to move to {request.transmission}')
self.__devs.transmission.set(request.transmission, wait=False)
if request.start is not None:
self.__devs.smargon.target = request.start
omega_start = self.omega
self.__devs.transmission.wait()
if request.start is not None:
self.__devs.smargon.wait()
status = self.status
self.__jfjoch.measure_rotation(request, status, self.__cfg.xrf)
self.__aare.create_rotation_run(self.sample, request, status)
if self.sample is not None and self.sample.db_id is not None:
self.save_screenshot_db(self.sample.db_id, "before_dc")
if request.screening:
self.__devs.aerotech.measure_screening(request.start_omega_deg,
request.wedge_omega_deg,
request.exp_time_s,
request.incr_omega_deg,
request.steps)
self.__devs.aerotech.wait_scan_done(request.exp_time_s * request.steps + 60)
else:
total_omega = request.incr_omega_deg * request.steps
total_time = request.exp_time_s * request.steps
self.__devs.aerotech.measure_standard(
request.start_omega_deg, total_omega, total_time
)
if request.start is not None and request.end is not None:
smargon_time_step = request.time_sec / float(request.steps)
pos_step = (request.end.sh_mm - request.start.sh_mm) * (1.0 / float(request.steps))
for i in range(request.steps):
self.__devs.smargon.target = SmargonCoordinate(
sh_mm=request.start.sh_mm + pos_step * i
)
time.sleep(smargon_time_step)
self.__devs.aerotech.wait_scan_done(total_time + 60)
self.__devs.aerotech.reset()
self.__devs.aerotech.move(omega_start, wait=False, speed=180.0, direct=True)
result = self.__jfjoch.wait_till_done(60)
self.__aare.sample_collected(self.sample)
self.__devs.aerotech.move(omega_start, wait=True, speed=180.0, direct=True)
if self.sample is not None and self.sample.db_id is not None:
self.save_screenshot_db(self.sample.db_id, "after_dc")
try:
self.__aare.ingest_scan(sample=self.sample, result=result,
geom=self.sample_geometry, beam_mark_pxl=self.__cfg.get_beam_mark(self.zoom))
except Exception as e:
logger.error(f"Exception ingesting scan: {e}")
return CompletedRotationScan(request=copy.deepcopy(request), result=result)
return None
def measure_rotation(self, request: RotationScanRequest) -> CompletedRotationScan:
total_time = request.exp_time_s * request.steps
@@ -813,9 +507,9 @@ class AareDAQ:
exposure = zoom_settings[zoom_value].exposure
gain = zoom_settings[zoom_value].gain
self.__devs.samcam_settings = SampleCameraSettings(exposure=exposure, gain=gain)
self.__devs.zoom_sync(zoom_value)
self.__devs.set_zoom(zoom_value, wait=True)
time.sleep(5.0) # Wait for settings to stabilize
image = self.__devs.sample_cam.get_image(gray=False)
image = self.__devs.samcam_get_image(gray=False)
self.__cfg.put_alc_bkg(zoom_value, exposure, gain, image)
bgr_array = cv2.cvtColor(image, cv2.COLOR_RGB2BGR)
cv2.imwrite(f"bkg{zoom_value:.0f}_{exposure*1000:.0f}_{gain:.0f}.jpg", bgr_array)
@@ -846,21 +540,21 @@ class AareDAQ:
self.__cfg.state_busy = False
raise RuntimeError("Cannot set dtz during data collection")
elif state == BeamlineStateEnum.SampleAlignment:
self.__devs.dtz.move(val, wait=False)
self.__devs.set_dtz(val, wait=False)
self.__cfg.dtz = val
self.__cfg.state_busy = False
@property
def smargon(self) -> SmargonCoordinate:
return self.__devs.smargon.readback
return self.__devs.smargon_pos
@smargon.setter
def smargon(self, sc: SmargonCoordinate):
self.__cfg.set_busy(BeamlineStateEnum.SampleAlignment)
try:
self.__saved_box = None
self.__devs.smargon.target = sc
self.__devs.smargon.wait()
self.__devs.smargon_pos = sc
self.__devs.smargon_wait()
self.__cfg.state_busy = False
except Exception as e:
self.__cfg.state_busy = False
@@ -884,48 +578,18 @@ class AareDAQ:
@property
def sample_geometry(self) -> SampleGeometryModel:
zoom = self.__devs.zoom
abr_pos = self.__cfg.abr_meas_pos
try:
aerotech_coord = Coordinate(
x=self.__devs.gmx.value - abr_pos.x,
y=self.__devs.gmy.value - abr_pos.y,
z=self.__devs.gmz.value - abr_pos.z,
)
sample_geom = SampleGeometryModel(
beam_location_pxl=self.__cfg.beam_mark_coeff.apply(zoom),
pixel_in_mm=self.__cfg.pixel_to_mm(zoom),
omega_deg=self.__devs.aerotech.omega,
smargon=self.__devs.smargon.readback,
beam_size_mm=self.__cfg.beam_size_mm,
aerotech=aerotech_coord,
aerotech_meas=self.__devs.abr_pos
)
return sample_geom
except Exception as e:
try:
test_x = self.__devs.gmx.value - abr_pos.x
print(f"Error creating sample geometry model: {e}")
print(f"Error not ABR, check smargon then restart server")
raise Exception(f"Error getting sample geometry {e}")
except:
print(f"Error getting sample geometry: {e}")
print(f"!!!!! ABR MAY HAVE DISCONNECTED !!!!!")
print(f"!!!!! CHECK ABR EPICS PANEL FOR ERRORS!!!!!")
print("!!!!! IF NOT RESTART GUI !!!!!")
print("CLOSING EXPERIMENTAL HUCTH FOR SAFETY")
self.__devs.exp_shutter.close()
return SampleGeometryModel(
beam_location_pxl=self.__cfg.beam_mark_coeff.apply(zoom),
pixel_in_mm=self.__cfg.pixel_to_mm(zoom),
omega_deg=self.__devs.aerotech.omega,
smargon=self.__devs.smargon.readback,
beam_size_mm=self.__cfg.beam_size_mm,
aerotech=Coordinate(x=0.0, y=0.0, z=0.0),
aerotech_meas=self.__devs.abr_pos
)
aerotech_pos_ref = self.__cfg.abr_meas_pos
aerotech_coord = self.__devs.aerotech_pos - aerotech_pos_ref
sample_geom = SampleGeometryModel(
beam_location_pxl=self.__cfg.beam_mark_coeff.apply(zoom),
pixel_in_mm=self.__cfg.pixel_to_mm(zoom),
omega_deg=self.__devs.aerotech_omega,
smargon=self.__devs.smargon_pos,
beam_size_mm=self.__cfg.beam_size_mm,
aerotech=aerotech_coord,
aerotech_meas=self.__devs.aerotech_pos
)
return sample_geom
@property
def beam_center(self) -> Tuple[float, float]:
@@ -946,9 +610,6 @@ class AareDAQ:
def get_beam_mark(self):
return self.__cfg.get_beam_mark(self.__devs.zoom)
def listen_changes(self) -> redis.client.PubSub:
return self.__cfg.listen_changes()
def __ml_bounding_box(self, sample_id: int | None = None, filename: str | None = None) -> RasterGridRequest | None:
time.sleep(0.2) # Just to be sure image is stable
#curr_image = self.camera_image
src/aare/daq/devices.py
+120 -257
View File
@@ -1,351 +1,214 @@
# Abstractions of devices for beamline
# Each "standard" device needs three elements:
# - property to read device value
# - setter with option to do sync/async move
# - property setter, which assumes that sync move is done (excl. zoom, which is async by default)
import time
import numpy as np
from epics import PV
from aare.common.beamline import MXBeamline
from aare.common.coordinate import Coordinate
from aare.common.coordinate import Coordinate, SmargonCoordinate
from aare.common.models import SampleCameraSettings, StagePositionEnum
from aare.devices import smargon, aerotech
from aare.devices import smargon
from aare.devices.area_detector import epicsAD
from aare.devices.enum_pv import EnumPv
from aare.devices.filter_transmission import FilterTransmission
from aare.devices.fluorimeter import Fluorimeter
from aare.devices.experimental_hutch_shutter import ExperimentalHutchShutter
from aare.devices.my_motor import MyMotor
from aare.devices.non_standard import NonStandard
from aare.devices.tell_client import TellClient
from aare.devices.workflow_tools import wait_position
class BeamlineDevices:
def __init__(self, beamline: MXBeamline):
#service_config = ServiceConfig(redis={"host": "x06da-bec-001.psi.ch", "port": 6379})
#client = BECClient(service_config, name="Filips-Custom-Client")
#client.start()
#self.__bec_dev = client.device_manager.devices
#self.__bec_scans = client.scans
BEAMLINE = beamline.value.upper()
self.tell = TellClient(beamline)
self.smargon = smargon.Smargon(beamline)
self.__energy_pv = PV(f"{BEAMLINE}-OP-DCCM:ENERGY1")
self.__smargon = smargon.Smargon(beamline)
self.__ring_current_pv = PV(f"ARS07-DPCT-0100:CURR")
self.__cryojet_temp = PV(f"{BEAMLINE}-ES-CRSM:TEMP_RBV")
self.magnet_position_sensor = PV(f"{BEAMLINE}-ES-DF1:CBOX-CMP1")
self.magnet_position_sensor_readout = PV(f"{BEAMLINE}-ES-DF1:CBOX-USER1")
self.dtz = MyMotor(f"{BEAMLINE}-ES-DET:TRZ1")
# self.dty = MyMotor(f"{BEAMLINE}-MOCK-DET:TRY1")
self.bsz = MyMotor(f"{BEAMLINE}-ES-BS:TRZ1")
self.bsx = MyMotor(f"{BEAMLINE}-ES-BS:TRX1")
self.bsy = MyMotor(f"{BEAMLINE}-ES-BS:TRY1")
self.__bpm = PV(f"{BEAMLINE}-OP-XBPM1:SumAll:MeanValue_RBV")
self.__anneal = PV(f"{BEAMLINE}-ES-ANN:SET_POS")
self.aerotech = aerotech.Abr(beamline)
self.aerotech.reload_programs()
self.gmx = NonStandard(
name="GMX",
setpv=f"{BEAMLINE}-ES-DF1:GMX-VAL",
getpv=f"{BEAMLINE}-ES-DF1:GMX-RBV",
speed=(f"{BEAMLINE}-ES-DF1:GMX-SETV", f"{BEAMLINE}-ES-DF1:GMX-SETV"),
move_done_when=(f"{BEAMLINE}-ES-DF1:GMX-DONE", 1),
)
self.transmission = FilterTransmission(beamline)
self.gmy = NonStandard(
name="GMY",
setpv=f"{BEAMLINE}-ES-DF1:GMY-VAL",
getpv=f"{BEAMLINE}-ES-DF1:GMY-RBV",
speed=(f"{BEAMLINE}-ES-DF1:GMY-SETV", f"{BEAMLINE}-ES-DF1:GMY-SETV"),
move_done_when=(f"{BEAMLINE}-ES-DF1:GMY-DONE", 1),
)
self.gmz = NonStandard(
name="GMZ",
setpv=f"{BEAMLINE}-ES-DF1:GMZ-VAL",
getpv=f"{BEAMLINE}-ES-DF1:GMZ-RBV",
speed=(f"{BEAMLINE}-ES-DF1:GMZ-SETV", f"{BEAMLINE}-ES-DF1:GMZ-SETV"),
move_done_when=(f"{BEAMLINE}-ES-DF1:GMZ-DONE", 1),
)
self.__lamp_light = NonStandard(
name="FrontLight",
setpv=f"{BEAMLINE}-ES-FL:SET-BRGHT",
getpv=f"{BEAMLINE}-ES-FL:SET-BRGHT",
tolerance=0.01,
predefs={
"on": 2.5,
"half": 1.9,
"off": 1.0,
},
)
self.cryo = NonStandard(
name="CryoJet",
setpv=f"{BEAMLINE}-ES-CJ:TRX1",
getpv=f"{BEAMLINE}-ES-CJ:TRX1.RBV",
move_done_when=(f"{BEAMLINE}-ES-CJ:TRX1.DMOV", 1),
speed=(f"{BEAMLINE}-ES-CJ:TRX1.VELO", f"{BEAMLINE}-ES-CJ:TRX1.VELO"),
)
self.__zoom = NonStandard(
name="Zoom",
setpv=f"{BEAMLINE}-ES-SAMCAM:ZOOM.VAL",
getpv=f"{BEAMLINE}-ES-SAMCAM:ZOOM.RBV",
move_done_when=(f"{BEAMLINE}-ES-SAMCAM:ZOOM.DMOV", 1),
stoppv=(f"{BEAMLINE}-ES-SAMCAM:ZOOM.STOP", 1),
)
self.__shutter_rbv = PV(f"{BEAMLINE}-ES-PH1:GET")
self.__shutter = PV(f"{BEAMLINE}-ES-PH1:SET")
self.exp_shutter = ExperimentalHutchShutter(beamline)
# xrf_stage = EnumPv(
# {
# "name": "KetekStage",
# "pv": f"{BEAMLINE}-MOCK-FD:SET-POS",
# "readback": f"{BEAMLINE}-MOCK-FD:GET-POS",
# "states": {"park": "park", "measure": "measure"},
# }
# )
self.__reflector = EnumPv(
{
"name": "BackReflector",
"pv": f"{BEAMLINE}-ES-BL:SET-POS",
"readback": f"{BEAMLINE}-ES-BL:GET-POS",
"states": {
"park": "park",
"measure": "measure",
},
}
)
self.detector_cover = PV(f"{BEAMLINE}-ES-DETCOV:SET")
self.__beamstop_stage = EnumPv(
{
"name": "Beamstop_Stage",
"pv": f"{BEAMLINE}-ES-BS:SET-POS",
"readback": f"{BEAMLINE}-ES-BS:GET-POS",
"states": {
"park": "park",
"measure": "measure",
},
}
)
self.__collimator = EnumPv(
{
"name": "Collimator",
"pv": f"{BEAMLINE}-ES-COL:SET-POS",
"readback": f"{BEAMLINE}-ES-COL:GET-POS",
"states": {
"parking": "parking",
"measure": "measure",
"down": "down",
},
}
)
self.__scinti = EnumPv(
{
"name": "Scintillator",
"pv": f"{BEAMLINE}-ES-SCL:SET-POS",
"readback": f"{BEAMLINE}-ES-SCL:GET-POS",
"states": {
"parking": "parking",
"measure": "measure",
"down": "down",
},
}
)
self.sample_cam = epicsAD(f"{BEAMLINE}-SAMCAM:")
self.fluorimeter = Fluorimeter(beamline)
self.__dtz = MyMotor(f"{BEAMLINE}-ES-DET:TRZ1")
self.__sample_cam = epicsAD(f"{BEAMLINE}-SAMCAM:")
# Lamp light
@property
def lamp_light(self) -> float:
return self.__lamp_light.value
return 0
@lamp_light.setter
def lamp_light(self, v: float):
self.__lamp_light.move(v, wait=False)
pass
# Zoom
@property
def zoom(self) -> float:
return self.__zoom.value
return 0
@zoom.setter
def zoom(self, value: float):
self.__zoom.move(value, wait=False)
# status = self.__bec_dev.samzoom.move(value)
# status.wait()
self.set_zoom(value, wait=False)
def zoom_sync(self, value: float):
self.__zoom.move(value, wait=True)
def set_zoom(self, value: float, /, wait: bool = True):
pass
# Collimator
@property
def collimator(self) -> StagePositionEnum:
if self.__collimator.position_is("down"):
return StagePositionEnum.DOWN
elif self.__collimator.position_is("measure"):
return StagePositionEnum.MEASURE
elif self.__collimator.position_is("parking"):
return StagePositionEnum.PARK
else:
return StagePositionEnum.UNKNOWN
return StagePositionEnum.DOWN
@collimator.setter
def collimator(self, value: StagePositionEnum):
self.set_collimator(value, wait=True)
def set_collimator(self, value: StagePositionEnum, /, wait: bool = True):
if value == StagePositionEnum.DOWN:
self.__collimator.move("down", wait=wait)
elif value == StagePositionEnum.MEASURE:
self.__collimator.move("measure", wait=wait)
elif value == StagePositionEnum.PARK:
self.__collimator.move("parking", wait=wait)
pass
# Scintillator
@property
def scintillator(self) -> StagePositionEnum:
if self.__scinti.position_is("down"):
return StagePositionEnum.DOWN
elif self.__scinti.position_is("measure"):
return StagePositionEnum.MEASURE
elif self.__scinti.position_is("parking"):
return StagePositionEnum.PARK
else:
return StagePositionEnum.UNKNOWN
return StagePositionEnum.DOWN
@scintillator.setter
def scintillator(self, value: StagePositionEnum):
self.set_scintillator(value, wait=True)
def set_scintillator(self, value: StagePositionEnum, /, wait: bool = True):
if value == StagePositionEnum.DOWN:
self.__scinti.move("down", wait=wait)
elif value == StagePositionEnum.MEASURE:
self.__scinti.move("measure", wait=wait)
elif value == StagePositionEnum.PARK:
self.__scinti.move("parking", wait=wait)
pass
# Reflector (backlight?)
@property
def reflector_up(self) -> bool:
return bool(self.__reflector.position_is("measure"))
return False
@reflector_up.setter
def reflector_up(self, value: bool):
if value:
self.__reflector.move("measure", wait=True)
else:
self.__reflector.move("park", wait=True)
pass
# Beamstop
@property
def beamstop_stage_up(self) -> bool:
return bool(self.__beamstop_stage.position_is("measure"))
return False
@beamstop_stage_up.setter
def beamstop_stage_up(self, value: bool):
if value:
self.__beamstop_stage.move("measure", wait=True)
else:
self.__beamstop_stage.move("park", wait=True)
pass
@property
def abr_pos(self) -> Coordinate:
return Coordinate(x=self.gmx.readback, y=self.gmy.readback, z=self.gmz.readback)
def beamstop_z(self) -> float:
return 35.0
@abr_pos.setter
def abr_pos(self, pos: Coordinate):
self.gmx.move(pos.x, wait=True)
self.gmy.move(pos.y, wait=True)
self.gmz.move(pos.z, wait=True)
wait_position(self.gmx, pos.x, 0.005)
wait_position(self.gmy, pos.y, 0.005)
wait_position(self.gmz, pos.z, 0.005)
@beamstop_z.setter
def beamstop_z(self, value: float):
pass
# Optics
@property
def energy_kev(self) -> float:
return self.__energy_pv.value
return 12.4
@property
def ring_current(self) -> float:
return max(0.0, self.__ring_current_pv.value)
return max(0.0, 0.0)
@property
def flux(self) -> float:
return 0
@property
def full_flux(self) -> float:
return 0
# Cryojet
@property
def cryojet_temp(self) -> float:
return self.__cryojet_temp.value
return 100
def enable_motors(self):
self.aerotech.unlock()
self.dtz.refresh() # Why ??
self.bsz.refresh() # Why ??
def disable_motors(self):
self.aerotech.lock()
def anneal(self, time: float):
pass
# Shutter
@property
def shutter(self) -> bool:
return self.__shutter_rbv.value == 1
return False
@shutter.setter
def shutter(self, opened: bool):
if opened:
self.__shutter.put(1)
else:
self.__shutter.put(0)
pass
# Sample camera
@property
def samcam_settings(self) -> SampleCameraSettings:
return SampleCameraSettings(
gain=self.sample_cam.gain.value,
exposure=self.sample_cam.expo.value
gain=self.__sample_cam.gain.value,
exposure=self.__sample_cam.expo.value
)
@samcam_settings.setter
def samcam_settings(self, settings: SampleCameraSettings):
self.sample_cam.setup(settings.gain, settings.exposure)
self.__sample_cam.setup(settings.gain, settings.exposure)
@property
def flux(self) -> float:
# Very approximate value - to be changed
# 2 nA -> 4e11 ph/s
if self.transmission.get() is None:
return 0.0
return self.transmission.get() * self.full_flux
def samcam_get_image(self, /, gray: bool = False) -> np.ndarray:
return self.__sample_cam.get_image(gray=gray)
# Detector Z
@property
def full_flux(self) -> float:
# Very approximate value - to be changed
# 2 nA -> 4e11 ph/s
return self.__bpm.value / 2.0 * 4e11
def dtz(self) -> float:
return self.__dtz.readback
@dtz.setter
def dtz(self, value: float):
self.set_dtz(value, wait=True)
def set_dtz(self, value: float, /, wait: bool = True):
self.__dtz.move(value, wait=wait)
@property
def dtz_low(self) -> float:
return self.dtz.get("LLM")
return self.__dtz.get("LLM")
@property
def dtz_high(self) -> float:
return self.dtz.get("HLM")
return self.__dtz.get("HLM")
def anneal(self, value: float):
self.__anneal.put(1)
time.sleep(value)
self.__anneal.put(0)
# Aertoech Automation1
@property
def aerotech_pos(self) -> Coordinate:
return Coordinate(x=0, y=0, z=0)
@aerotech_pos.setter
def aerotech_pos(self, pos: Coordinate):
pass
@property
def aerotech_omega(self) -> float:
return 0.0
@aerotech_omega.setter
def aerotech_omega(self, val: float):
self.set_omega(val, wait=True)
def set_aerotech_omega(self, val: float, /, wait: bool = True):
pass
@property
def aerotech_lock(self) -> bool:
return False
@aerotech_lock.setter
def aerotech_lock(self, val: bool):
pass
# Smargon goniometer
@property
def smargon_pos(self) -> SmargonCoordinate:
return self.__smargon.readback
def set_smargon_pos(self, pos: SmargonCoordinate, /, wait: bool = True):
self.__smargon.target = pos
@smargon_pos.setter
def smargon_pos(self, pos: SmargonCoordinate):
self.set_smargon_pos(pos, wait=True)
def smargon_wait(self):
self.__smargon.wait()
def smargon_move_home(self):
self.__smargon.move_home(wait=True)
src/aare/daq/server.py
+3 -3
View File
@@ -21,10 +21,10 @@ from fastapi import status as api_status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from starlette.responses import StreamingResponse
from aare.aaredaq import auth
from aare.daq import auth
from aare.common.beamline import mx_beamline
from aare.aaredaq.config import BeamlineConfig
from aare.aaredaq.daq import (AareDAQ, LoopCenteringFailed, TransformationInvalidException,
from aare.daq.config import BeamlineConfig
from aare.daq.daq import (AareDAQ, LoopCenteringFailed, TransformationInvalidException,
MountingFailed, WarningTellException, CriticalTellException)
app = FastAPI()
src/aare/daq/workflows.py
+23 -262
View File
@@ -1,16 +1,5 @@
import time
from epics import poll
from aare.aaredaq.config import ABR_POS_MOUNT, ABR_OMEGA_MOUNT
from aare.aaredaq.devices import BeamlineDevices
from aare.aaredaq.config import BeamlineConfig
from aare.common.coordinate import Coordinate
from aare.common.logger_config import setup_logger
from aare.common.models import StagePositionEnum, ZoomModeEnum
from aare.devices.mx_lib import pv_wait
logger = setup_logger("aareDAQ")
from aare.daq.devices import BeamlineDevices
from aare.daq.config import BeamlineConfig
def move_bsz(devs: BeamlineDevices, target: float):
if abs(target - devs.bsz.position) > 0.1:
@@ -18,8 +7,7 @@ def move_bsz(devs: BeamlineDevices, target: float):
reflector_measure = devs.reflector_up
devs.reflector_up = False
devs.beamstop_stage_up = True
devs.bsz.move(target, wait=True)
devs.beamstop_z = target
if reflector_measure:
devs.reflector_up = True
@@ -29,174 +17,39 @@ def move_bsz(devs: BeamlineDevices, target: float):
def wait_for_dc_devices(devs: BeamlineDevices):
timeout = time.time() + 60
timeisup = False
# devs.detector_cover.wait()
while not timeisup and (
not devs.dtz.done_moving
):
poll(0.1)
timeisup = timeout < time.time()
if timeisup:
if not devs.dtz.done_moving:
logger.info("DTZ still moving.")
raise RuntimeError("Timeout moving devs. Dropping to maintenance.")
# Should wait for DTZ
pass
def wait_for_se_devices(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
_cryo = hub_cryo.cryojet_in_use
_cryopark = hub_cryo.cryojet_park_position
if _cryo and not devs.cryo.position_is(_cryopark):
raise RuntimeError("Cryojet did not reach far position")
# Should wait for Cryojet go far
pass
def common_2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
logger.info(" moving COLLIMATOR to PARK position")
devs.collimator = StagePositionEnum.PARK
logger.info(" moving SMARGON to HOME position")
devs.smargon.move_home(wait=True)
logger.info(" moving AEROTECH to MOUNT position")
devs.aerotech.move(ABR_OMEGA_MOUNT, wait=True, direct=True)
devs.abr_pos = ABR_POS_MOUNT
logger.info(" moving REFLECTOR to DOWN position")
devs.reflector_up = False
logger.info(" LOCKING AEROTERCH")
devs.aerotech.lock()
logger.info(f" Smargon position in RSE: {devs.smargon.readback} ABS: {devs.abr_pos.x} {devs.abr_pos.y} {devs.abr_pos.z}")
pass
def m2se(devs: BeamlineDevices, cfg: BeamlineConfig):
print("executing MAINTENANCE -> sample exchange")
devs.detector_cover.put(1) # Close detector cover
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
cfg.zoom_mode = ZoomModeEnum.User
devs.samcam_settings = cfg.zoom_settings.get_camera_settings(devs.zoom)
devs.collimator = StagePositionEnum.PARK
devs.scintillator = StagePositionEnum.DOWN
_dtzpark = hub.dtz_park
_cryopark = hub_cryo.cryojet_park_position
_cryo = hub_cryo.cryojet_in_use
# devs.detector_cover.move("closed")
if not devs.aerotech.is_ready_and_willing():
devs.aerotech.stop()
time.sleep(5.0)
raise RuntimeError("Aerotech device is not ready")
if devs.dtz.position < _dtzpark:
devs.dtz.move(_dtzpark)
devs.gmx.set_speed(100.0)
devs.abr_pos = ABR_POS_MOUNT
devs.aerotech.move(ABR_OMEGA_MOUNT)
if _cryo:
devs.cryo.move(_cryopark)
devs.reflector_up = False
devs.bsx.move(0.0)
devs.bsy.move(0.0)
move_bsz(devs, hub.bsz)
devs.beamstop_stage_up = False
pass
def sa2se(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
logger.info(" moving cryojet to park position")
_cryopark = hub_cryo.cryojet_park_position
_dtzpark = hub.dtz_park
devs.aerotech.move(ABR_OMEGA_MOUNT)
if devs.cryo.position < _cryopark:
devs.cryo.move(_cryopark)
logger.info(" moving COLLIMATOR to DOWN position")
devs.collimator = StagePositionEnum.DOWN
logger.info(" moving REFLECTOR to DOWN position")
devs.reflector_up = False
devs.beamstop_stage_up = False
logger.info(" moving SMARGON to HOME position")
devs.smargon.move_home(wait=True)
logger.info(" moving DETECTOR COVER to DOWN position")
devs.detector_cover.put(1) # Close detector cover
if devs.dtz.position < _dtzpark:
logger.info(" moving DETECTOR to PARK position")
devs.dtz.move(_dtzpark)
pass
def sa2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
common_2rse(devs, cfg)
def dc2rse(devs: BeamlineDevices, cfg: BeamlineConfig):
common_2rse(devs, cfg)
def se2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
cfg.zoom_mode = ZoomModeEnum.User
devs.detector_cover.put(1)
devs.samcam_settings = cfg.zoom_settings.get_camera_settings(devs.zoom)
hub_cryo = cfg.cryojet_settings
_cryo = hub_cryo.cryojet_in_use
cryo_meas = hub_cryo.cryojet_measurement_position
_dtz_safety = hub.dtz_bsz_safety_margin
devs.aerotech.unlock()
logger.info("aerotech unlocked")
devs.aerotech.set_direct_mode()
logger.info("aerotech in direct mode")
logger.info(" moving reflector to up position")
devs.reflector_up = True
logger.info("moving beamstop to up position")
devs.beamstop_stage_up = True
logger.info(" setting lamp to 2.5")
devs.lamp_light = 2.5
logger.info(" moving aerotech to measure position")
devs.abr_pos = cfg.abr_meas_pos
if _cryo:
logger.info(" moving cryojet to measure position")
devs.cryo.move(cryo_meas)
logger.info("moving detector to measure position")
devs.dtz.move(cfg.dtz, wait=False)
pass
def rse2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
se2sa(devs, cfg)
def sa2dc(devs: BeamlineDevices, cfg: BeamlineConfig):
devs.detector_cover.put(2) # Open detector cover
devs.reflector_up = False
#devs.reflector.wait("park")
devs.beamstop_stage_up = True
devs.dtz.move(cfg.dtz, wait=True)
devs.collimator = StagePositionEnum.MEASURE
pv_wait(devs.dtz, None, tolerance=1.0)
wait_for_dc_devices(devs)
pass
def dc2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
devs.detector_cover.put(1) # Close detector cover
devs.collimator = StagePositionEnum.PARK
devs.reflector_up = True
# devs.gmz.move(cfg.abr_meas_pos.z, wait=True)
pass
def sa2xrf(devs: BeamlineDevices, cfg: BeamlineConfig):
"""sample alignment to XrfCollection"""
@@ -209,127 +62,35 @@ def xrf2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
def sa2ws(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
_dtzwash = hub.dtz_wash_sample_distance
devs.cryo.move(hub.cryojet_park_position)
if devs.dtz.position < _dtzwash:
devs.dtz.move(_dtzwash)
devs.reflector_up = False
devs.beamstop_stage_up = False
pass
def ws2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
cryo_meas = hub_cryo.cryojet_measurement_position
_cryo = hub_cryo.cryojet_in_use
devs.dtz.move(cfg.dtz)
if _cryo:
devs.cryo.move(cryo_meas)
devs.reflector_up = True
devs.beamstop_stage_up = True
pass
def sa2ba(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
_cryopark = hub_cryo.cryojet_park_position
_cryo = hub_cryo.cryojet_in_use
if _cryo:
devs.cryo.move(_cryopark)
devs.gmx.move(-38.0)
devs.lamp_light = 2.5
devs.beamstop_stage_up = True
move_bsz(devs, 0.0)
pass
def ba2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
devs.lamp_light = 2.5
devs.beamstop_stage_up = True
move_bsz(devs, hub.bsz)
devs.reflector_up = True
devs.abr_pos = cfg.abr_meas_pos
pass
def sa2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
val=1000
devs.zoom = val
cfg.zoom_mode = ZoomModeEnum.BeamLocation
devs.detector_cover.put(1) # Close detector cover
devs.samcam_settings = cfg.zoom_settings.get_camera_settings(val)
devs.lamp_light = 2.5
devs.abr_pos = Coordinate(x=-26, y=0, z=0)
devs.cryo.move(80)
devs.set_scintillator(StagePositionEnum.MEASURE, wait=False)
devs.set_collimator(StagePositionEnum.MEASURE, wait=False)
devs.beamstop_stage_up = True
devs.reflector_up = False
devs.scintillator = StagePositionEnum.MEASURE
devs.collimator = StagePositionEnum.MEASURE
devs.shutter = True
pass
def bl2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
hub_cryo = cfg.cryojet_settings
cryo_meas = hub_cryo.cryojet_measurement_position
_cryo = hub_cryo.cryojet_in_use
devs.shutter = False
devs.set_scintillator(StagePositionEnum.DOWN, wait=False)
devs.set_collimator(StagePositionEnum.PARK, wait=False)
devs.reflector_up = True
devs.scintillator = StagePositionEnum.DOWN
devs.collimator = StagePositionEnum.PARK
val=1
devs.zoom = val
cfg.zoom_mode = ZoomModeEnum.User
devs.samcam_settings = cfg.zoom_settings.get_camera_settings(val)
if _cryo:
devs.cryo.move(cryo_meas)
devs.abr_pos = cfg.abr_meas_pos
pass
def bl2ba(devs: BeamlineDevices, cfg: BeamlineConfig):
"""beam location to beamstop alignment"""
devs.gmx.move(-38)
devs.reflector_up = True
# devs.shutter.move("closed")
time.sleep(0.5)
move_bsz(devs, 0)
pass
def ba2bl(devs: BeamlineDevices, cfg: BeamlineConfig):
hub = cfg.settings
move_bsz(devs, hub.bsz)
devs.abr_pos = cfg.abr_meas_pos + Coordinate(x=1)
devs.reflector_up = True
pass
def sa2dh(devs: BeamlineDevices, cfg: BeamlineConfig):
logger.info("moved to dewer transfer")
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)
pass
def dh2sa(devs: BeamlineDevices, cfg: BeamlineConfig):
#devs.tell.move_cold(wait=True)
logger.info("return to sample alignment")
pass