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

refactor: class refactoring, pending change to SlsDetectorCam

Browse Source
This commit is contained in:
appel_c 2023-08-21 21:52:09 +02:00
parent 287c667621
commit b1150c41fe
1 changed files with 107 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,17 +1,17 @@
import json
import os
from typing import List
import requests
import numpy as np
from typing import List
from ophyd.areadetector import ADComponent as ADCpt, PilatusDetectorCam, DetectorBase
from ophyd.areadetector.plugins import FileBase
from bec_lib.core import BECMessage, MessageEndpoints
from bec_lib.core import BECMessage, MessageEndpoints, RedisConnector
from bec_lib.core.file_utils import FileWriterMixin
from bec_lib.core import bec_logger
logger = bec_logger.logger
@ -19,14 +19,18 @@ class PilatusDetectorCamEx(PilatusDetectorCam, FileBase):
pass
# TODO refactor class -> move away from DetectorBase and PilatusDetectorCamEx class to Device. -> this will be cleaner
class PilatusCsaxs(DetectorBase):
"""Pilatus_2 300k detector for CSAXS
Parent class: DetectorBase
Device class: PilatusDetectorCamEx
Attributes:
name str: 'eiger'
prefix (str): PV prefix (X12SA-ES-PILATUS300K:)
"""
in device config, device_access needs to be set true to inject the device manager
"""
_html_docs = ["PilatusDoc.html"]
cam = ADCpt(PilatusDetectorCamEx, "cam1:")
def __init__(
@ -41,9 +45,6 @@ class PilatusCsaxs(DetectorBase):
device_manager=None,
**kwargs,
):
self.device_manager = device_manager
self.username = "e21206" # TODO get from config
# self.username = self.device_manager.producer.get(MessageEndpoints.account()).decode()
super().__init__(
prefix=prefix,
name=name,
@ -53,10 +54,15 @@ class PilatusCsaxs(DetectorBase):
parent=parent,
**kwargs,
)
# TODO how to get base_path
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.username}/Data10/pilatus_2/"}
self.device_manager = device_manager
self.name = name
self.username = "e21206"
# TODO once running from BEC
# self.username = self.device_manager.producer.get(MessageEndpoints.account()).decode()
self.service_cfg = {"base_path": f"/sls/X12SA/data/{self.username}/Data10/data/"}
self.filewriter = FileWriterMixin(self.service_cfg)
self.num_frames = 0
self._producer = RedisConnector(["localhost:6379"]).producer()
self.readout = 0.003 # 3 ms
self.triggermode = 0 # 0 : internal, scan must set this if hardware triggered
@ -64,24 +70,46 @@ class PilatusCsaxs(DetectorBase):
msg = self.device_manager.producer.get(MessageEndpoints.scan_status())
return BECMessage.ScanStatusMessage.loads(msg)
def stage(self) -> List[object]:
# TODO remove
# scan_msg = self._get_current_scan_msg()
# self.metadata = {
# "scanID": scan_msg.content["scanID"],
# "RID": scan_msg.content["info"]["RID"],
# "queueID": scan_msg.content["info"]["queueID"],
# }
self.scan_number = 10 # scan_msg.content["info"]["scan_number"]
self.exp_time = 0.5 # scan_msg.content["info"]["exp_time"]
self.num_frames = 3 # scan_msg.content["info"]["num_points"]
# TODO remove
# self.username = self.device_manager.producer.get(MessageEndpoints.account()).decode()
def _load_scan_metadata(self) -> None:
scan_msg = self._get_current_scan_msg()
self.metadata = {
"scanID": scan_msg.content["scanID"],
"RID": scan_msg.content["info"]["RID"],
"queueID": scan_msg.content["info"]["queueID"],
}
self.scanID = scan_msg.content["scanID"]
self.scan_number = scan_msg.content["info"]["scan_number"]
self.exp_time = scan_msg.content["info"]["exp_time"]
self.num_frames = scan_msg.content["info"]["num_points"]
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
)
# set pilatus threshol
self._set_threshold()
def _prep_det(self) -> None:
self._set_det_threshold()
self._set_acquisition_params()
# set Epic PVs for filewriting
def _set_det_threshold(self) -> None:
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)
def _set_acquisition_params(self) -> None:
"""set acquisition parameters on the detector"""
self.cam.acquire_time.set(self.exp_time)
self.cam.acquire_period.set(self.exp_time + self.readout)
self.cam.num_images.set(self.num_frames)
self.cam.num_exposures.set(1)
self.cam.trigger_mode.set(self.triggermode)
def _prep_file_writer(self) -> None:
"""Prepare the file writer for pilatus_2
a zmq service is running on xbl-daq-34 that is waiting
for a zmq message to start the writer for the pilatus_2 x12sa-pd-2
"""
self.cam.file_path.set(f"/dev/shm/zmq/")
self.cam.file_name.set(f"{self.username}_2_{self.scan_number:05d}")
self.cam.auto_increment.set(1) # auto increment
@ -89,13 +117,12 @@ class PilatusCsaxs(DetectorBase):
self.cam.file_format.set(0) # 0: TIFF
self.cam.file_template.set("%s%s_%5.5d.cbf")
# compile zmq stream for data transfer
scan_dir = self.filewriter._get_scan_directory(
scan_bundle=1000, scan_number=self.scan_number, leading_zeros=5
)
self.destination_path = os.path.join(
self.service_cfg["base_path"]
) # os.path.join(self.service_cfg["base_path"], scan_dir)
# TODO Filewriter Plugin to write cbfs to h5!
# Pilatus_2 writes cbf files -> where do we like to write those!
# scan_dir = self.filewriter._get_scan_directory(
# scan_bundle=1000, scan_number=self.scan_number, leading_zeros=5
# ) # os.path.join(self.service_cfg["base_path"], scan_dir)
self.destination_path = "/sls/X12SA/data/{self.username}/Data10/pilatus_2/"
data_msg = {
"source": [
@ -106,6 +133,7 @@ class PilatusCsaxs(DetectorBase):
}
]
}
logger.info(data_msg)
headers = {"Content-Type": "application/json", "Accept": "application/json"}
@ -141,16 +169,11 @@ class PilatusCsaxs(DetectorBase):
if not res.ok:
res.raise_for_status()
self._set_acquisition_params(
exp_time=self.exp_time,
readout=self.readout,
num_frames=self.num_frames,
triggermode=self.triggermode,
)
return super().stage()
def unstage(self) -> List[object]:
def _close_file_writer(self) -> None:
"""Close the file writer for pilatus_2
a zmq service is running on xbl-daq-34 that is waiting
for a zmq message to stop the writer for the pilatus_2 x12sa-pd-2
"""
headers = {"Content-Type": "application/json", "Accept": "application/json"}
data_msg = [
"zmqWriter",
@ -167,8 +190,6 @@ class PilatusCsaxs(DetectorBase):
data=json.dumps(data_msg),
headers=headers,
)
# Reset triggermode to internal
self.triggermode = 0
if not res.ok:
res.raise_for_status()
@ -176,46 +197,49 @@ class PilatusCsaxs(DetectorBase):
res = requests.delete(url="http://x12sa-pd-2:8080/stream/pilatus_2")
if not res.ok:
res.raise_for_status()
def stage(self) -> List[object]:
"""stage the detector and file writer"""
# TODO remove once running from BEC
# self._load_scan_metadata()
self.scan_number = 10
self.exp_time = 0.5
self.num_frames = 3
self.mokev = 12
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(),
)
return super().stage()
def unstage(self) -> List[object]:
"""unstage the detector and file writer"""
# Reset to software trigger
self.triggermode = 0
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(),
)
return super().unstage()
def _set_threshold(self) -> None:
# TODO readout mono, monitor threshold and set it if mokev is different
# mokev = self.device_manager.devices.mokev.obj.read()["mokev"]["value"]
# TODO remove
mokev = 16
# TODO refactor naming from name, pilatus_2
pil_threshold = self.cam.threshold_energy.read()["pilatus_2_cam_threshold_energy"]["value"]
if not np.isclose(mokev / 2, pil_threshold, rtol=0.05):
self.cam.threshold_energy.set(mokev / 2)
def _set_acquisition_params(
self, exp_time: float, readout: float, num_frames: int, triggermode: int
) -> None:
"""set acquisition parameters on the detector
Args:
exp_time (float): exposure time
readout (float): readout time
num_frames (int): images per scan
triggermode (int):
0 Internal
1 Ext. Enable
2 Ext. Trigger
3 Mult. Trigger
4 Alignment
Returns:
None
"""
self.cam.acquire_time.set(exp_time)
self.cam.acquire_period.set(exp_time + readout)
self.cam.num_images.set(num_frames)
self.cam.num_exposures.set(1)
self.cam.trigger_mode.set(triggermode)
def acquire(self) -> None:
"""Start acquisition in software trigger mode,
or arm the detector in hardware of the detector
"""
self.cam.acquire.set(1)
def stop(self, *, success=False) -> None:
"""Stop the scan, with camera and file writer"""
self.cam.acquire.set(0)
self.unstage()
super().stop(success=success)