336 lines
11 KiB
Python
336 lines
11 KiB
Python
import time
|
|
|
|
from epics import PV
|
|
|
|
from aaredaqlib.beamline import MXBeamline
|
|
from aaredaqlib.coordinate import Coordinate
|
|
from aaredaqlib.models import SampleCameraSettings, StagePositionEnum
|
|
from mxlibs3 import aerotech, smargon
|
|
from mxlibs3.area_detector import epicsAD
|
|
from mxlibs3.enum_pv import EnumPv
|
|
from mxlibs3.filter_transmission import FilterTransmission
|
|
from mxlibs3.my_motor import MyMotor
|
|
from mxlibs3.non_standard import NonStandard
|
|
from mxlibs3.tell_client import TellClient
|
|
from mxlibs3.workflow_tools import wait_position
|
|
|
|
|
|
#from bec_lib.client import BECClient
|
|
#from bec_lib.service_config import ServiceConfig
|
|
|
|
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.__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")
|
|
|
|
# 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:")
|
|
|
|
@property
|
|
def lamp_light(self) -> float:
|
|
return self.__lamp_light.value
|
|
|
|
@lamp_light.setter
|
|
def lamp_light(self, v: float):
|
|
self.__lamp_light.move(v, wait=False)
|
|
|
|
@property
|
|
def zoom(self) -> float:
|
|
return self.__zoom.value
|
|
|
|
@zoom.setter
|
|
def zoom(self, value: float):
|
|
self.__zoom.move(value, wait=False)
|
|
# status = self.__bec_dev.samzoom.move(value)
|
|
# status.wait()
|
|
|
|
def zoom_sync(self, value: float):
|
|
self.__zoom.move(value, wait=True)
|
|
|
|
@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
|
|
|
|
@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)
|
|
|
|
@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
|
|
|
|
@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)
|
|
|
|
@property
|
|
def reflector_up(self) -> bool:
|
|
return bool(self.__reflector.position_is("measure"))
|
|
|
|
@reflector_up.setter
|
|
def reflector_up(self, value: bool):
|
|
if value:
|
|
self.__reflector.move("measure", wait=True)
|
|
else:
|
|
self.__reflector.move("park", wait=True)
|
|
|
|
@property
|
|
def beamstop_stage_up(self) -> bool:
|
|
return bool(self.__beamstop_stage.position_is("measure"))
|
|
|
|
@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)
|
|
|
|
@property
|
|
def abr_pos(self) -> Coordinate:
|
|
return Coordinate(x=self.gmx.readback, y=self.gmy.readback, z=self.gmz.readback)
|
|
|
|
@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)
|
|
|
|
@property
|
|
def energy_kev(self) -> float:
|
|
return self.__energy_pv.value
|
|
|
|
@property
|
|
def ring_current(self) -> float:
|
|
return max(0.0, self.__ring_current_pv.value)
|
|
|
|
@property
|
|
def cryojet_temp(self) -> float:
|
|
return self.__cryojet_temp.value
|
|
|
|
def enable_motors(self):
|
|
self.aerotech.unlock()
|
|
self.dtz.refresh() # Why ??
|
|
self.bsz.refresh() # Why ??
|
|
|
|
def disable_motors(self):
|
|
self.aerotech.lock()
|
|
|
|
@property
|
|
def shutter(self) -> bool:
|
|
return self.__shutter_rbv.value == 1
|
|
|
|
@shutter.setter
|
|
def shutter(self, opened: bool):
|
|
if opened:
|
|
self.__shutter.put(1)
|
|
else:
|
|
self.__shutter.put(0)
|
|
|
|
@property
|
|
def samcam_settings(self) -> SampleCameraSettings:
|
|
return SampleCameraSettings(
|
|
gain=self.sample_cam.gain.value,
|
|
exposure=self.sample_cam.expo.value
|
|
)
|
|
|
|
@samcam_settings.setter
|
|
def samcam_settings(self, settings: SampleCameraSettings):
|
|
print(settings)
|
|
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.__bpm.value / 2.0 * 4e11
|
|
|
|
def anneal(self, value: float):
|
|
self.__anneal.put(1)
|
|
time.sleep(value)
|
|
self.__anneal.put(0)
|