bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-07-12 19:51:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

online_changes

Browse Source
This commit is contained in:
e20588
2022-11-11 11:34:15 +01:00
parent bba6109330
commit 7b221ea979
2 changed files with 44 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
ophyd_devices/eiger1p5m_csaxs/eiger1p5m.py
View File

@ -4,7 +4,7 @@ from typing import List
from bec_utils import BECMessage, MessageEndpoints, bec_logger
from ophyd import Component as Cpt
from ophyd import Device, DeviceStatus, EpicsSignal, Signal
from ophyd import Device, DeviceStatus, EpicsSignal, Signal, EpicsSignalRO
logger = bec_logger.logger
@ -49,22 +49,21 @@ class _SLSDetectorConfigSignal(Signal):
class Eiger1p5MDetector(Device):
USER_ACCESS = []
file_path = Cpt(
EpicsSignal, name="file_path", read_pv="XOMNYI-DET-OUTDIR:0.DESC", kind="config"
)
file_path = Cpt(EpicsSignal, name="file_path", read_pv="XOMNYI-DET-OUTDIR:0.DESC")
detector_out_scan_dir = Cpt(
EpicsSignal, name="detector_out_scan_dir", read_pv="XOMNYI-DET-OUTDIR:1.DESC", kind="config"
EpicsSignal, name="detector_out_scan_dir", read_pv="XOMNYI-DET-OUTDIR:1.DESC"
)
frames = Cpt(EpicsSignal, name="frames", read_pv="XOMNYI-DET-CYCLES:0", kind="config")
exp_time = Cpt(EpicsSignal, name="exp_time", read_pv="XOMNYI-DET-EXPTIME:0", kind="config")
index = Cpt(EpicsSignal, name="index", read_pv="XOMNYI-DET-INDEX:0", kind="config")
frames = Cpt(EpicsSignal, name="frames", read_pv="XOMNYI-DET-CYCLES:0")
exp_time = Cpt(EpicsSignal, name="exp_time", read_pv="XOMNYI-DET-EXPTIME:0")
index = Cpt(EpicsSignal, name="index", read_pv="XOMNYI-DET-INDEX:0")
detector_control = Cpt(
EpicsSignal, name="detector_control", read_pv="XOMNYI-DET-CONTROL:0.DESC", kind="config"
EpicsSignal, name="detector_control", read_pv="XOMNYI-DET-CONTROL:0.DESC"
)
framescaught = Cpt(EpicsSignalRO, name="framescaught", read_pv="XOMNYI-DET-CONTROL:0.VAL")
file_pattern = Cpt(_SLSDetectorConfigSignal, name="file_pattern", value="", kind="config")
burst = Cpt(_SLSDetectorConfigSignal, name="burst", value=1, kind="config")
save_file = Cpt(_SLSDetectorConfigSignal, name="save_file", value=False, kind="config")
file_pattern = Cpt(_SLSDetectorConfigSignal, name="file_pattern", value="")
burst = Cpt(_SLSDetectorConfigSignal, name="burst", value=1)
save_file = Cpt(_SLSDetectorConfigSignal, name="save_file", value=False)
def __init__(self, *, name, kind=None, parent=None, device_manager=None, **kwargs):
self.device_manager = device_manager
@ -82,26 +81,6 @@ class Eiger1p5MDetector(Device):
self.metadata = {}
self.username = "e20588" # TODO get from config
def trigger(self):
status = DeviceStatus(self)
self.subscribe(status._finished, event_type=self.SUB_ACQ_DONE, run=False)
# def acquire():
# try:
# for _ in range(self.burst.get()):
# ttime.sleep(self.exp_time.get())
# if self._stopped:
# raise DeviceStop
# except DeviceStop:
# pass
# finally:
# self._stopped = False
# self._done_acquiring()
# threading.Thread(target=acquire, daemon=True).start()
return status
def _get_current_scan_msg(self) -> BECMessage.ScanStatusMessage:
msg = self.device_manager.producer.get(MessageEndpoints.scan_status())
return BECMessage.ScanStatusMessage.loads(msg)
@ -146,7 +125,7 @@ class Eiger1p5MDetector(Device):
# wait for detector to be "armed"
logger.info("Waiting for detector setup")
while True:
det_ctrl = self.detector_control.get()["detector_control"]["value"]
det_ctrl = self.detector_control.get()
if det_ctrl == "armed":
break
time.sleep(0.005)
@ -156,7 +135,21 @@ class Eiger1p5MDetector(Device):
return super().stage()
def unstage(self) -> List[object]:
signals = {"config": self.read_configuration(), "data": self.file_name}
time_waited = 0
sleep_time = 0.2
framesexpected = self.frames.get()
while True:
framescaught = self.framescaught.get()
if self.framescaught.get() < framesexpected:
logger.info(
f"Waiting for frames: Transferred {framescaught} out of {framesexpected}"
)
time_waited += sleep_time
time.sleep(sleep_time)
continue
break
self.detector_control.put("stop")
signals = {"config": self.read(), "data": self.file_name}
msg = BECMessage.DeviceMessage(signals=signals, metadata=self.metadata)
self.device_manager.producer.set_and_publish(
MessageEndpoints.device_read(self.name), msg.dumps()
@ -167,3 +160,8 @@ class Eiger1p5MDetector(Device):
self.detector_control.put("stop")
super().stop(success=success)
self._stopped = True
if __name__ == "__main__":
eiger = Eiger1p5MDetector(name="eiger1p5m", label="eiger1p5m")
breakpoint()

15
ophyd_devices/rt_lamni/rt_lamni_ophyd.py
View File

@ -450,9 +450,18 @@ class RtLamniController(Controller):
)
time.sleep(0.03)
# needs to be changed to configurable center psoition
self.get_device_manager().devices.lsamx.obj.move(8.866, wait=True)
self.get_device_manager().devices.lsamy.obj.move(10.18, wait=True)
lsamx_user_params = self.get_device_manager().devices.lsamx.user_parameter
if lsamx_user_params is None or lsamx_user_params.get("center") is None:
raise RuntimeError("lsamx center is not defined")
lsamy_user_params = self.get_device_manager().devices.lsamy.user_parameter
if lsamy_user_params is None or lsamy_user_params.get("center") is None:
raise RuntimeError("lsamy center is not defined")
lsamx_center = lsamx_user_params.get("center")
lsamy_center = lsamy_user_params.get("center")
self.get_device_manager().devices.lsamx.obj.move(lsamx_center, wait=True)
self.get_device_manager().devices.lsamy.obj.move(lsamy_center, wait=True)
self.socket_put("J1")
_waitforfeedbackctr = 0