bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-05-28 15:40:42 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix: pil300k device, pending readout

Browse Source
This commit is contained in:
e21206 2023-08-28 22:31:40 +02:00
parent 96a131d374
commit b91f8dbc68
1 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
ophyd_devices/epics/devices/pilatus_csaxs.py
View File

@ -84,16 +84,20 @@ class PilatusCsaxs(DetectorBase):
self.username = self.device_manager.producer.get(MessageEndpoints.account()).decode()
self.device_manager.devices.mokev.read()["mokev"]["value"]
# self.triggermode = scan_msg.content["info"]["trigger_mode"]
self.filename = self.filewriter.compile_full_filename(
self.scan_number, "pilatus_2", 1000, 5, True
)
# self.filename = self.filewriter.compile_full_filename(
# self.scan_number, "pilatus_2", 1000, 5, True
# )
# TODO fix with BEC running
# self.filename = '/sls/X12SA/Data10/e21206/data/test.h5'
def _prep_det(self) -> None:
# TODO slow reaction, seemed to have timeout.
self._set_det_threshold()
self._set_acquisition_params()
def _set_det_threshold(self) -> None:
threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name]["value"]
# threshold = self.cam.threshold_energy.read()[self.cam.threshold_energy.name]['value']
if not np.isclose(self.mokev / 2, threshold, rtol=0.05):
self.cam.threshold_energy.set(self.mokev / 2)
@ -210,11 +214,11 @@ class PilatusCsaxs(DetectorBase):
self._prep_det()
self._prep_file_writer()
msg = BECMessage.FileMessage(file_path=self.filename, done=False)
self._producer.set_and_publish(
MessageEndpoints.public_file(self.scanID, "pilatus_2"),
msg.dumps(),
)
# msg = BECMessage.FileMessage(file_path=self.filename, done=False)
# self._producer.set_and_publish(
# MessageEndpoints.public_file(self.scanID, "pilatus_2"),
# msg.dumps(),
# )
return super().stage()
@ -225,11 +229,11 @@ class PilatusCsaxs(DetectorBase):
self._close_file_writer()
# TODO check if acquisition is done and successful!
state = True
msg = BECMessage.FileMessage(file_path=self.filepath, done=True, successful=state)
self.producer.set_and_publish(
MessageEndpoints.public_file(self.metadata["scanID"], self.name),
msg.dumps(),
)
# msg = BECMessage.FileMessage(file_path=self.filepath, done=True, successful=state)
# self.producer.set_and_publish(
# MessageEndpoints.public_file(self.metadata["scanID"], self.name),
# msg.dumps(),
# )
return super().unstage()
def acquire(self) -> None: