374 lines
11 KiB
Python
374 lines
11 KiB
Python
# 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 numpy as np
|
|
from epics import PV
|
|
|
|
from aare.common.beamline import MXBeamline
|
|
from aare.common.coordinate import SmargonCoordinate, AerotechCoordinate
|
|
from aare.common.logger_config import setup_logger
|
|
from aare.common.models import SampleCameraSettings, StagePositionEnum
|
|
from aare.devices import smargon, aerotech
|
|
from aare.devices.area_detector import epicsAD, AutoEnum
|
|
from aare.devices.enum_pv import EnumPV
|
|
from aare.devices.my_motor import MyMotor
|
|
|
|
from aare.devices.set_get_pv import SetGetPV, PredefinedPV
|
|
from aare.devices.tell_client import make_tell_client
|
|
from aare.devices.bec_worker import BECClientWorker
|
|
|
|
logger = setup_logger("aareDAQ")
|
|
|
|
class BeamlineDevices:
|
|
def __init__(self, beamline: MXBeamline):
|
|
BEAMLINE = beamline.value.upper()
|
|
self.tell = make_tell_client(beamline)
|
|
self.aerotech = aerotech.AerotechController(beamline)
|
|
logger.debug("initialising BEC worker")
|
|
self.bec_worker = BECClientWorker(beamline)
|
|
logger.debug("initialising BEC worker done")
|
|
self.__smargon = smargon.Smargon(beamline)
|
|
|
|
self.__ring_current_pv = PV(f"ARS07-DPCT-0100:CURR")
|
|
|
|
self.__dtz = MyMotor(f"{BEAMLINE}-ES-DET:TRZ")
|
|
self.__dty = MyMotor(f"{BEAMLINE}-ES-DET:TRY")
|
|
|
|
self.__det_cov = EnumPV(name="det_cov",
|
|
setpv=f"{BEAMLINE}-ES-DETCOV:SET",
|
|
getpv=f"{BEAMLINE}-ES-DETCOV:GET")
|
|
|
|
self.__sample_cam = epicsAD(f"{BEAMLINE}-ES-MS:")
|
|
|
|
self.__front_light = PredefinedPV(name='front_light',
|
|
setpv=f"{BEAMLINE}-ES-FL:SET",
|
|
getpv=f"{BEAMLINE}-ES-FL:SET",
|
|
predefs={"off":1.49,
|
|
'half':2.0,
|
|
'max':3.0},
|
|
timeout=10.0
|
|
)
|
|
self.__back_light = PredefinedPV(name='back_light',
|
|
setpv =f"{BEAMLINE}-ES-BL:SET",
|
|
getpv=f"{BEAMLINE}-ES-BL:SET",
|
|
predefs={"off": 0,
|
|
'half': 0.98,
|
|
'max': 1.2},
|
|
timeout=10.0
|
|
)
|
|
|
|
self.__back_light_pos = EnumPV(name = "back_light_pos",
|
|
setpv = f"{BEAMLINE}-ES-BL:POS-SET",
|
|
getpv = f"{BEAMLINE}-ES-BL:POS-GET",
|
|
timeout = 10.0)
|
|
|
|
self.__collimator_pos = MyMotor(f"{BEAMLINE}-ES-COL:TRY")
|
|
self.__collimator_X = MyMotor(f"{BEAMLINE}-ES-COL:TRX") # HOW TO HANDLE!!!
|
|
|
|
self.__scintillator_pos = PV(f"{BEAMLINE}-ES-SCL:TRY") # how to handle!!!
|
|
self.__scintillator_z = PV(f"{BEAMLINE}-ES-SCL:TRZ")
|
|
|
|
self.__beamstop_pos = EnumPV(name = "beamstop_pos",
|
|
setpv = f"{BEAMLINE}-ES-BS:POS-SET",
|
|
getpv = f"{BEAMLINE}-ES-BS:POS-GET",
|
|
timeout = 10.0)
|
|
|
|
self.__beamstop_x = MyMotor(f"{BEAMLINE}-ES-BS:TRX")
|
|
self.__beamstop_y = MyMotor(f"{BEAMLINE}-ES-BS:TRY")
|
|
self.__beamstop_z = MyMotor(f"{BEAMLINE}-ES-BS:TRZ")
|
|
|
|
|
|
self.__zoom = SetGetPV(name = f"zoom",
|
|
setpv = f"{BEAMLINE}-ES-MS:ZOOM.VAL",
|
|
getpv = f"{BEAMLINE}-ES-MS:ZOOM.RBV")
|
|
|
|
self.__cryojet_pos = EnumPV(name='cryojet_pos',
|
|
setpv = f"{BEAMLINE}-ES-CS:POS-SET",
|
|
getpv = f"{BEAMLINE}-ES-CS:POS-GET",
|
|
timeout = 10.0)
|
|
|
|
self.__cryojet_x = MyMotor(f"{BEAMLINE}-ES-CS:TRX") #currently in is 5 out is 15?
|
|
|
|
self.magnet_position_sensor = PV(f"{BEAMLINE}-ES-DF1:CBOX-CMP1")
|
|
self.magnet_position_sensor_readout = PV(f"{BEAMLINE}-ES-DF1:CBOX-USER1")
|
|
|
|
|
|
|
|
|
|
# Transmission
|
|
@property
|
|
def transmission(self) -> float:
|
|
return 1.0
|
|
|
|
@transmission.setter
|
|
def transmission(self, value: float):
|
|
self.set_transmission(value, wait=False)
|
|
|
|
def set_transmission(self, value: float, /, wait: bool = True):
|
|
pass
|
|
|
|
# Lamp light
|
|
@property
|
|
def lamp_light(self) -> float:
|
|
return self.__front_light.value
|
|
|
|
@lamp_light.setter
|
|
def lamp_light(self, v: float):
|
|
self.set_front_light(v, wait=False)
|
|
|
|
def set_front_light(self, v: float, /, wait: bool = True):
|
|
self.__front_light.move(v, wait=wait)
|
|
|
|
# Back light
|
|
@property
|
|
def back_light(self) -> float:
|
|
return self.__back_light.value
|
|
|
|
@back_light.setter
|
|
def back_light(self, v: float):
|
|
self.set_back_light(v, wait=False)
|
|
|
|
def set_back_light(self, v: float, /, wait: bool = True):
|
|
self.__back_light.move(v, wait=wait)
|
|
|
|
# Zoom
|
|
@property
|
|
def zoom(self) -> float:
|
|
return self.__zoom.value
|
|
|
|
@zoom.setter
|
|
def zoom(self, value: float):
|
|
self.set_zoom(value, wait=True)
|
|
|
|
def set_zoom(self, value: float, /, wait: bool = True):
|
|
self.__zoom.move(value, wait=wait)
|
|
|
|
# Collimator
|
|
@property
|
|
def collimator(self) -> float:
|
|
return self.__collimator_pos.get()
|
|
|
|
@collimator.setter
|
|
def collimator(self, value: float):
|
|
self.set_collimator(value, wait=True)
|
|
|
|
def set_collimator(self, value: float, /, wait: bool = True):
|
|
self.__collimator_pos.move(value, wait=wait)
|
|
|
|
# Scintillator
|
|
@property
|
|
def scintillator(self) -> float:
|
|
return self.__scintillator_pos.get()
|
|
|
|
@scintillator.setter
|
|
def scintillator(self, value: float):
|
|
self.set_scintillator(value, wait=True)
|
|
|
|
def set_scintillator(self, value: float, /, wait: bool = True):
|
|
self.__scintillator_pos.put(value, wait=wait)
|
|
|
|
# Reflector (backlight?)
|
|
@property
|
|
def reflector_up(self) -> bool:
|
|
return self.__back_light_pos.position.upper() == StagePositionEnum.MEASURE.name
|
|
|
|
@reflector_up.setter
|
|
def reflector_up(self, value: StagePositionEnum):
|
|
self.set_reflector_up(value, wait=True)
|
|
|
|
def set_reflector_up(self, value: StagePositionEnum, /, wait: bool = True):
|
|
self.__back_light_pos.move(value, wait=wait)
|
|
|
|
# Beamstop
|
|
@property
|
|
def beamstop_stage_up(self) -> bool:
|
|
return False
|
|
|
|
@beamstop_stage_up.setter
|
|
def beamstop_stage_up(self, value: bool):
|
|
pass
|
|
|
|
@property
|
|
def beamstop_z(self) -> float:
|
|
return 35.0
|
|
|
|
@beamstop_z.setter
|
|
def beamstop_z(self, value: float):
|
|
pass
|
|
|
|
# Optics
|
|
@property
|
|
def energy_kev(self) -> float:
|
|
return 12.4
|
|
|
|
@property
|
|
def ring_current(self) -> float:
|
|
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 100
|
|
|
|
def anneal(self, time: float):
|
|
pass
|
|
|
|
@property
|
|
def cryojet_pos(self) -> StagePositionEnum:
|
|
return StagePositionEnum(self.__cryojet_pos.position.upper())
|
|
|
|
@cryojet_pos.setter
|
|
def cryojet_pos(self, value: StagePositionEnum):
|
|
self.cryojet_pos_setter(value, wait=True)
|
|
|
|
def cryojet_pos_setter(self, value: StagePositionEnum, wait:bool=False):
|
|
self.__cryojet_pos.move(value, wait=wait)
|
|
|
|
# Shutter
|
|
@property
|
|
def shutter(self) -> bool:
|
|
return False
|
|
|
|
@shutter.setter
|
|
def shutter(self, opened: bool):
|
|
pass
|
|
|
|
# Sample camera
|
|
@property
|
|
def samcam_settings(self) -> SampleCameraSettings:
|
|
return SampleCameraSettings(
|
|
gain=self.__sample_cam.gain_rbv.value,
|
|
exposure=self.__sample_cam.expo_rbv.value
|
|
)
|
|
|
|
@samcam_settings.setter
|
|
def samcam_settings(self, settings: SampleCameraSettings):
|
|
self.__sample_cam.setup(settings.gain, settings.exposure)
|
|
|
|
def samcam_get_image(self, /, gray: bool = False) -> np.ndarray:
|
|
return self.__sample_cam.get_image(gray=gray)
|
|
|
|
def samcam_auto(self, state: AutoEnum):
|
|
self.__sample_cam.set_auto(state)
|
|
|
|
def samcam_frame_id(self) -> int:
|
|
"""
|
|
Camera UniqueId for the last produced frame (monotonic counter from AreaDetector).
|
|
"""
|
|
return int(self.__sample_cam.uid.get())
|
|
|
|
# Detector Z
|
|
@property
|
|
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")
|
|
|
|
@property
|
|
def dtz_high(self) -> float:
|
|
return self.__dtz.get("HLM")
|
|
|
|
@property
|
|
def aerotech_pos(self) -> AerotechCoordinate:
|
|
#TODO add u/omega????
|
|
return self.aerotech.get_position()
|
|
|
|
@aerotech_pos.setter
|
|
def aerotech_pos(
|
|
self,
|
|
coord: AerotechCoordinate,
|
|
/,
|
|
wait: bool = True,
|
|
incremental: bool = False,
|
|
):
|
|
self.aerotech.position(
|
|
coord,
|
|
wait=wait,
|
|
incremental=incremental,
|
|
)
|
|
|
|
@property
|
|
def aerotech_omega(self) -> float:
|
|
return self.aerotech.status().u.pos
|
|
|
|
@aerotech_omega.setter
|
|
def aerotech_omega(self, val: float):
|
|
self.set_aerotech_omega(val, wait=True)
|
|
|
|
def set_aerotech_omega(
|
|
self,
|
|
val: float,
|
|
/,
|
|
wait: bool = True,
|
|
incremental: bool = False,
|
|
):
|
|
target = AerotechCoordinate(omega_deg=val)
|
|
return self.aerotech.position(
|
|
target,
|
|
wait=wait,
|
|
incremental=incremental,
|
|
)
|
|
|
|
@property
|
|
def aerotech_lock(self) -> bool:
|
|
return False
|
|
|
|
@aerotech_lock.setter
|
|
def aerotech_lock(self, val: bool):
|
|
pass
|
|
|
|
def aerotech_stop(self):
|
|
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, timeout: float = 10.0):
|
|
self.__smargon.wait(timeout=timeout)
|
|
|
|
def smargon_move_home(self):
|
|
self.__smargon.move_home(wait=True)
|
|
|
|
def smargon_aerotech_wait(self):
|
|
self.__smargon.wait_aerotech(timeout=10.0)
|
|
|
|
if __name__ == "__main__":
|
|
from aare.common.beamline import mx_beamline
|
|
beamline = mx_beamline()
|
|
devs = BeamlineDevices(beamline)
|
|
print(devs.aerotech_pos)
|
|
# devs.aerotech_pos = Coordinate(x=124.0, y=1.0, z=1.0)
|
|
# print(devs.aerotech_pos)
|
|
# devs.aerotech_omega = 0.0
|
|
print(devs.reflector_up)
|
|
devs.reflector_up = StagePositionEnum.MEASURE |